fix(agent): keep one run-scoped encoder across hosted stream chunks - #3497
Conversation
`createConversationHostedStreamLifecycleAdapter` built a fresh `ConversationRunEventEncoder` for every chunk, because `prepareConversationRunStreamEvents` was called without one. The encoder holds per-run state, so rebuilding it per chunk broke two things on the top-level hosted path — the only path production runs take: - `stepCount` restarted at 1 for every chunk, so persisted events read `step-1, step-1, step-1, step-1, step-2` instead of advancing. - The encoder's construction is the anchor `elapsedMs` is measured from, so every event was stamped against a just-created anchor. In practice the column never got a usable value at all. v0.1.1220 added the clock to the encoder (#3483) but wired it into `createConversationRunChunkMirror`, which only serves child runs. Top-level runs go through this adapter, which is why production still shows no `elapsedMs` on any persisted event. Hoist a single clocked encoder to adapter-creation time and pass it into every `prepareConversationRunStreamEvents` call. The encoder is also exposed as an option so callers that need deterministic payloads can supply an unclocked one — the same escape hatch `run-chunk-mirror` already offers. Test drives the adapter with two chunks and asserts step names advance and `elapsedMs` is stamped; it fails with `["step-1","step-1"]` before the fix.
📝 WalkthroughWalkthroughThe hosted stream lifecycle adapter now reuses a ChangesHosted stream encoding
Estimated code review effort: 2 (Simple) | ~10 minutes Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 3d4cd3962f
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| const encoder = options.encoder ?? | ||
| new ConversationRunEventEncoder({ nowMs: () => performance.now() }); |
There was a problem hiding this comment.
When the same public adapter is used for a second runHostedLifecycle call, this closure reuses the first run's encoder rather than creating one for the new startRun result. Consequently, the later run's first step is persisted as step-N instead of step-1, while active message/tool state and the elapsedMs origin can also leak across runs. Store an encoder per returned run, or otherwise reset it whenever a new run starts.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/agent/conversation/hosted-lifecycle.ts`:
- Around line 153-154: Update startRun and the adapter state around
ConversationRunEventEncoder so each run.runId receives a fresh encoder,
preventing step, message, tool, and elapsed-time state from being shared across
runs. Reuse the per-run encoder throughout that run, or explicitly enforce a
single-run adapter if that is the existing contract.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro Plus
Run ID: e5107f73-992b-4169-887c-a7e0400e32b5
📒 Files selected for processing (3)
docs/api-reference/veryfront/agent.mdsrc/agent/conversation/hosted-lifecycle.test.tssrc/agent/conversation/hosted-lifecycle.ts
| const encoder = options.encoder ?? | ||
| new ConversationRunEventEncoder({ nowMs: () => performance.now() }); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
ast-grep outline src/agent/conversation/hosted-lifecycle.ts --items all
rg -n -C 4 'createConversationHostedStreamLifecycleAdapter\s*\(' src
rg -n -C 4 '\.startRun\s*\(' src/agentRepository: veryfront/veryfront-code
Length of output: 28100
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- hosted-lifecycle.ts ---'
sed -n '1,280p' src/agent/conversation/hosted-lifecycle.ts
printf '%s\n' '--- lifecycle types and runner ---'
sed -n '1,230p' src/agent/hosted/lifecycle.ts
printf '%s\n' '--- encoder implementation ---'
sed -n '1,280p' src/agent/conversation/run-events.ts
printf '%s\n' '--- all creator references and adapter-related references ---'
rg -n -C 5 'createConversationHostedStreamLifecycleAdapter|createConversationHostedLifecycleAdapter|HostedLifecycleAdapter|runHostedLifecycle' srcRepository: veryfront/veryfront-code
Length of output: 50380
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- encoder state transitions ---'
rg -n -A 12 -B 5 'nextStepName|finishStepName|activeMessageId|encodeChunk|stepStarted|stepFinished' \
src/agent/conversation/run-events.ts
printf '%s\n' '--- stream preparation ---'
sed -n '1,220p' src/agent/conversation/run-event-preparation.ts
printf '%s\n' '--- creator references across the repository ---'
rg -n -C 3 'createConversationHostedStreamLifecycleAdapter|createConversationHostedLifecycleAdapter' . \
-g '!node_modules' -g '!dist' -g '!build'
printf '%s\n' '--- focused tests around adapter reuse ---'
sed -n '80,230p' src/agent/conversation/hosted-lifecycle.test.tsRepository: veryfront/veryfront-code
Length of output: 23713
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
from pathlib import Path
source = Path("src/agent/conversation/hosted-lifecycle.ts").read_text()
required = [
"const encoder = options.encoder ??",
"startRun: options.startRun",
"mapChunkToEvents: (chunk) => prepareConversationRunStreamEvents([chunk], encoder)",
]
missing = [item for item in required if item not in source]
if missing:
raise SystemExit(f"missing expected source fragments: {missing}")
encoder_pos = source.index("const encoder = options.encoder ??")
return_pos = source.index("return createConversationHostedLifecycleAdapter", encoder_pos)
start_pos = source.index("startRun: options.startRun", return_pos)
mapper_pos = source.index(
"mapChunkToEvents: (chunk) => prepareConversationRunStreamEvents([chunk], encoder)",
return_pos,
)
if not (encoder_pos < return_pos < start_pos < mapper_pos):
raise SystemExit("unexpected adapter closure ordering")
print("The adapter creates one encoder before constructing its returned adapter.")
print("The returned adapter forwards every startRun call to options.startRun.")
print("The returned adapter maps every append through the same encoder closure.")
print("The public startRun contract has no one-shot or one-run restriction.")
PYRepository: veryfront/veryfront-code
Length of output: 268
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
from pathlib import Path
source = Path("src/agent/conversation/hosted-lifecycle.ts").read_text()
stream_start = source.index("export function createConversationHostedStreamLifecycleAdapter")
stream_end = source.index("/** Context for conversation child lifecycle. */", stream_start)
stream = source[stream_start:stream_end]
generic_start = source.index("export function createConversationHostedLifecycleAdapter")
generic_end = source.index("/** Create conversation hosted stream lifecycle adapter. */", generic_start)
generic = source[generic_start:generic_end]
checks = {
"stream creates encoder before returning adapter": (
stream.index("const encoder = options.encoder ??")
< stream.index("return createConversationHostedLifecycleAdapter")
),
"stream passes caller options into generic adapter": "...options" in stream,
"stream maps chunks through the captured encoder": (
"prepareConversationRunStreamEvents([chunk], encoder)" in stream
),
"generic adapter forwards startRun": "startRun: options.startRun" in generic,
"startRun has no one-shot restriction": (
"startRun: (" in source and
"Promise<ConversationRunProjection> | ConversationRunProjection" in source
),
}
for name, result in checks.items():
print(f"{name}: {'yes' if result else 'no'}")
if not result:
raise SystemExit(1)
step_count = 0
step_count += 1
first = f"step-{step_count}"
step_count += 1
second = f"step-{step_count}"
print(f"shared encoder step names: {first}, {second}")
print("fresh encoder per run step names: step-1, step-1")
PYRepository: veryfront/veryfront-code
Length of output: 495
Create one ConversationRunEventEncoder per run.
startRun is reusable, but the adapter captures one encoder for all runs. A later run inherits step, message, tool, and elapsed-time state. Create and select an encoder per run.runId, or enforce one run per adapter.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/agent/conversation/hosted-lifecycle.ts` around lines 153 - 154, Update
startRun and the adapter state around ConversationRunEventEncoder so each
run.runId receives a fresh encoder, preventing step, message, tool, and
elapsed-time state from being shared across runs. Reuse the per-run encoder
throughout that run, or explicitly enforce a single-run adapter if that is the
existing contract.
Source: Coding guidelines
|
Correction to this PR's description, on the record. I claimed production's persisted events read What this PR fixes is still a real bug: But it is probably not the fix for the missing
Production shows advance + absent. Verified along the way, so these are not the explanation: v0.1.1220 genuinely contains the clocked chunk mirror, and the runtime pod that executed the 11:00 run ( Remaining suspects:
v0.1.1221 is being promoted as planned; the next scheduled tick is the measurement that settles which route is live. |
Problem
elapsedMsis missing from every persisted event on production runs, and step names repeat (step-1 ×4, step-2) instead of advancing.createConversationHostedStreamLifecycleAdaptercalledprepareConversationRunStreamEvents([chunk])without an encoder, so a freshConversationRunEventEncoderwas constructed per chunk. The encoder holds per-run state, so rebuilding it per chunk broke two things:stepCountrestarted at 1 on every chunk, so persisted events never advance paststep-1.elapsedMsis measured from, so every event was stamped against a just-created anchor — the column never received a usable value.This is the top-level hosted path, which is the path production runs actually take.
#3483 (shipped in v0.1.1220) added the clock to the encoder but wired it into
createConversationRunChunkMirror, which only serves child runs. That's why 1220 didn't change anything for top-level runs.Verified against production on v0.1.1220 —
payload ? 'elapsedMs'isffor every event of the latest run.Fix
Hoist a single clocked
ConversationRunEventEncoderto adapter-creation time and pass it into everyprepareConversationRunStreamEventscall.The encoder is also exposed as an optional option so callers needing deterministic payloads can supply an unclocked one — the same escape hatch
run-chunk-mirroralready offers, and what the existing exact-payload test now uses.Test
New test drives the adapter with two chunks and asserts step names advance and
elapsedMsis stamped. Confirmed red before the fix:which matches production's
step-1 ×4, step-2exactly.Verification
agent/conversation: 21 passed.deno task typecheckclean.docs/api-referenceregenerated with the CI-pinned Deno 2.7.7 (line-pin shifts only).Note
hosted-lifecycle.test.tsalready fails standalonedeno checkon 3 pre-existingstartRunstubs that omitwaitingToolCallId/waitingToolName/streamProtocolVersion; this PR adds a 4th instance of the same pattern. The file isn't in thetypecheckgate, and fixing it means feeding those fields into posted payloads (risking the exact-payload assertions), so it's left for a separate cleanup.Summary by CodeRabbit
Improvements
Documentation