perf: harness telemetry/span-volume reductions - #205
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: ⛔ Files ignored due to path filters (1)
📒 Files selected for processing (13)
✅ Files skipped from review due to trivial changes (3)
🚧 Files skipped from review as they are similar to previous changes (8)
📝 WalkthroughWalkthroughUpdates harness SDK dependency from ^0.12.0 to ^0.16.1, consolidates telemetry imports to ChangesHarness Backend Refinement
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Suggested reviewers
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
skill-check — worker0 verified, 13 skipped (no docs/).
Note 17 stale rendered artifact(s) detected on main, unrelated to this PR. This PR is fine; the drift was already there. A maintainer should open a chore PR to re-render these.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 9
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
console/web/src/pages/Traces/index.tsx (1)
112-116:⚠️ Potential issue | 🟠 Major | ⚡ Quick winRoute grouped-detail teardown through
closeDetail().This direct
setSelectedGroup(null)bypasses the auto-pause bookkeeping. If a user opens a group panel and then changes Group by back tonone, the panel closes butisPausedstaystrue, so live refresh remains frozen with no detail panel open. It also leaves stale group detail in place when switching between grouped attributes. UsecloseDetail()here so the selection reset andautoResume()stay in sync.Suggested fix
useEffect(() => { - if (!filterState.groupBy || filterState.groupBy === 'none') { - setSelectedGroup(null) - } - }, [filterState.groupBy]) + if (selectedGroup) { + closeDetail() + } + }, [filterState.groupBy, selectedGroup, closeDetail])🤖 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 `@console/web/src/pages/Traces/index.tsx` around lines 112 - 116, Replace the direct state reset setSelectedGroup(null) in the useEffect with a call to closeDetail() so route teardown uses the existing detail-close logic; specifically, inside the effect that watches filterState.groupBy (in Traces/index.tsx) remove setSelectedGroup(null) and invoke closeDetail() instead so selection, isPaused/autoResume() bookkeeping, and any grouped-detail cleanup stay in sync with the existing closeDetail implementation.console/web/src/pages/Traces/components/FlameGraph.tsx (1)
112-150:⚠️ Potential issue | 🟠 Major | ⚡ Quick winUse the same cycle-safe tree construction here.
buildSpanTree()now keeps self-/mutually-cyclic spans renderable, butbuildFlameNodes()still drops those spans because it only emits natural roots. In that casefilteredRowscontains rows frombuildSpanTree()whileflameMapis empty, so the flame graph goes blank for malformed traces that the waterfall now survives. Reusing the same parent-linking logic here would keep both views consistent.Also applies to: 152-156, 232-259
🤖 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 `@console/web/src/pages/Traces/components/FlameGraph.tsx` around lines 112 - 150, buildFlameNodes currently only emits "natural" roots and therefore drops spans involved in self/mutual cycles, causing the flame graph to go blank for traces buildSpanTree handles; update buildFlameNodes to reuse the same cycle-safe parent-linking logic used by buildSpanTree: when linking children to parents (using span.span_id and span.parent_span_id on the spanMap), detect and avoid creating cycles (e.g., track ancestors/visited when following parent links or break parent links that would introduce a cycle) and ensure every node that cannot be safely attached to a parent is added to roots so malformed traces remain renderable; keep the subsequent selfTime computation (using node.span.duration_ms and node.children) unchanged so timings stay correct.
🧹 Nitpick comments (1)
harness/tests/turn-orchestrator/coalesce-deltas.test.ts (1)
151-209: ⚡ Quick winAdd a rejection-path test for
runStreamTurn.The happy-path final flush is covered, but the risky case here is
streamTurnthrowing after one or more deltas have already been buffered. A small regression test for that would lock in thetry/finallybehavior once you fix it.🤖 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 `@harness/tests/turn-orchestrator/coalesce-deltas.test.ts` around lines 151 - 209, Add a rejection-path test for runStreamTurn that ensures buffered deltas are flushed when streamTurn throws: update the existing describe block by adding an it that uses mkPorts where the drive callback calls onDelta(P, td('a')) then throws (e.g., throw new Error('boom')); call runStreamTurn(ports, 'sid', {} as StreamContext) and assert the call does not throw, that emits contains the coalesced text delta ('a'), that outcome.error is non-null (or contains the error indicator used by runStreamTurn), and that outcome.body_streamed is true; reference runStreamTurn, mkPorts, streamTurn, and emitMessageUpdate when locating where to add the test.
🤖 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 `@console/web/src/lib/devtools-stream.ts`:
- Around line 98-104: The subscribe helper currently swallows errors from
client.call('ui::subscribe') causing the browser to remain unsubscribed; change
subscribe (and its callers) to retry on failure with exponential backoff until
success (or expose the error to higher-level connection state so the page can
recover). Specifically, update the subscribe function that calls
client.call('ui::subscribe', { browser_id: client.browserId, session_id: null })
to implement a retry loop/backoff (e.g., initial delay, exponential increase,
jitter) that re-invokes client.call on rejection and only logs when attempts
fail but continues retrying, and ensure callers observe success (resolve) so
live refresh is re-enabled once subscribe succeeds; keep using dlog for status
but do not swallow the rejection permanently.
In `@console/web/src/pages/Traces/hooks/useTraceData.ts`:
- Around line 120-126: The else branch that clears visible state
(setTraceListItems([]), fingerprintRef.current = '', prevTraceIdsRef.current =
new Set()) misses clearing the hover/pending buffer so flushPendingTraces() can
later reintroduce stale items; add pendingTracesRef.current = [] (or an empty
array/object matching its shape) in that same branch to reset the hover buffer
and prevent old traces from being replayed, ensuring flushPendingTraces() has
nothing to replay after a clear/filter change.
In `@console/web/src/pages/Traces/lib/spanTree.ts`:
- Around line 123-159: The recursive functions markCriticalPath,
unmarkCriticalPath, and traverse can blow the JS call stack for deep
parent/child chains; replace their recursion with explicit iterative stack-based
implementations that mirror current logic and respect the existing
cycle/visiting guards (use chainMark/SAFE/CYCLIC or the same visiting checks
used in calculateDepths) so behavior and ordering/side-effects remain identical.
For each function, traverse using a manual stack frame that stores node id,
iterator/index/state (e.g., entering vs leaving) to emulate recursion, push
children from the existing spanMap/children collection, and perform the same
marking/unmarking or visitor calls when popping or during enter as the original
recursive versions did; ensure mutual/self-parent cycles are skipped and
previously marked nodes are handled consistently. Also ensure traversal
preserves the original child order and updates any depth/critical flags in the
same points the recursive code did. Finally, add unit tests or a synthetic
deep-span trace to verify no stack overflow and identical outcomes compared to
the prior recursive behavior.
In `@harness/src/harness/fanout/traces-changed.ts`:
- Around line 58-61: The current turn_end handler clears and re-arms timer each
event which creates a trailing debounce that can never fire under steady
high-frequency events; change the logic in the async turn_end callback so you
only schedule flush once per coalescing window (e.g., if (!timer) timer =
setTimeout(flush, COALESCE_MS)) or implement a max-wait by capturing the
timestamp of the first event and also scheduling a secondary timeout to force
flush after MAX_WAIT; reference the symbols timer, flush, COALESCE_MS and the
async turn_end handler to locate and update the code.
In `@harness/src/runtime/otel.ts`:
- Around line 280-287: The current code replaces any existing baggage by
creating a new Baggage and calling propagation.setBaggage(...), so merge the new
iii.* entries into the active baggage instead: call
propagation.getBaggage(otelContext.active()) to obtain the current baggage (or
start with an empty baggage), then for each entry ('iii.function.id',
'iii.session.id', 'iii.message.id') call baggage = baggage.setEntry(key, { value
}) to produce a merged Baggage, and finally call
propagation.setBaggage(otelContext.active(), baggage) so existing baggage
entries are preserved; reference propagation.getBaggage,
propagation.createBaggage, baggage.setEntry, and propagation.setBaggage in
otel.ts.
In `@harness/src/turn-orchestrator/assistant-streaming/coalesce-deltas.ts`:
- Around line 50-66: The timer-driven flush currently nulls buf before awaiting
emit, causing lost/ordered deltas if emit rejects; change flush/arm to serialize
flushes by introducing an in-flight promise (e.g., inFlightFlush) that the timer
path awaits instead of calling void flush(), and only clear or replace buf after
emit resolves successfully; ensure clearTimer still prevents multiple timers,
have flush capture currentBuf locally, await emit(currentBuf.partial, merged)
and on success then set buf = null and resolve inFlightFlush, and have
subsequent flush callers wait on inFlightFlush to preserve ordering.
In `@harness/src/turn-orchestrator/assistant-streaming/run.ts`:
- Around line 67-74: The call to ports.streamTurn(...) may reject and skip the
subsequent await coalescer.flush(), losing buffered deltas; wrap the stream call
in a try/finally so coalescer.flush() is always awaited: call
ports.streamTurn(ctx, async (partial, event) => { ... }) inside a try, capture
its result into { final, error } as before, and then in a finally block await
coalescer.flush() (so coalescer.onEvent and body_streamed handling remain
unchanged); preserve propagation of any thrown error after flushing if needed.
In `@harness/src/turn-orchestrator/events.ts`:
- Around line 31-49: The seqBySession Map currently grows unbounded; change it
to track last-seen timestamps and add a reclamation policy: replace
seqBySession: Map<string, number> with Map<string, {n:number, last:number}>
(keep PROCESS_EPOCH and nextSeq names), update nextSeq(session_id) to bump both
counter and last timestamp, and introduce constants (e.g. MAX_SESSIONS and
SESSION_TTL_MS) plus a cleanup function that evicts entries older than
SESSION_TTL_MS or prunes the oldest entries when size > MAX_SESSIONS; call the
cleanup from nextSeq (and run it periodically with setInterval) so inactive
session entries are reclaimed, and ensure _resetSeqForTests() still clears the
map and stops/resets any timer used for cleanup.
In `@harness/src/turn-orchestrator/state-runtime/store.ts`:
- Around line 96-109: The current change-detection uses prev (possibly the
caller-supplied previous) to compute prevView and viewChanged, but the review
asks to use the persisted overwrite result's old_value as the source of truth;
update the logic that computes prevView and viewChanged to derive prevView from
result.old_value (convert via toView) instead of prev, and pass that derived
prevView into emitTurnStateChanged (and use it for the 'state:created' vs
'state:updated' decision) so the emitted old_value and dedup compare use the
actual persisted prior value rather than the optional previous param; update
references around nextView, toView, viewChanged, and emitTurnStateChanged
accordingly.
---
Outside diff comments:
In `@console/web/src/pages/Traces/components/FlameGraph.tsx`:
- Around line 112-150: buildFlameNodes currently only emits "natural" roots and
therefore drops spans involved in self/mutual cycles, causing the flame graph to
go blank for traces buildSpanTree handles; update buildFlameNodes to reuse the
same cycle-safe parent-linking logic used by buildSpanTree: when linking
children to parents (using span.span_id and span.parent_span_id on the spanMap),
detect and avoid creating cycles (e.g., track ancestors/visited when following
parent links or break parent links that would introduce a cycle) and ensure
every node that cannot be safely attached to a parent is added to roots so
malformed traces remain renderable; keep the subsequent selfTime computation
(using node.span.duration_ms and node.children) unchanged so timings stay
correct.
In `@console/web/src/pages/Traces/index.tsx`:
- Around line 112-116: Replace the direct state reset setSelectedGroup(null) in
the useEffect with a call to closeDetail() so route teardown uses the existing
detail-close logic; specifically, inside the effect that watches
filterState.groupBy (in Traces/index.tsx) remove setSelectedGroup(null) and
invoke closeDetail() instead so selection, isPaused/autoResume() bookkeeping,
and any grouped-detail cleanup stay in sync with the existing closeDetail
implementation.
---
Nitpick comments:
In `@harness/tests/turn-orchestrator/coalesce-deltas.test.ts`:
- Around line 151-209: Add a rejection-path test for runStreamTurn that ensures
buffered deltas are flushed when streamTurn throws: update the existing describe
block by adding an it that uses mkPorts where the drive callback calls
onDelta(P, td('a')) then throws (e.g., throw new Error('boom')); call
runStreamTurn(ports, 'sid', {} as StreamContext) and assert the call does not
throw, that emits contains the coalesced text delta ('a'), that outcome.error is
non-null (or contains the error indicator used by runStreamTurn), and that
outcome.body_streamed is true; reference runStreamTurn, mkPorts, streamTurn, and
emitMessageUpdate when locating where to add the test.
🪄 Autofix (Beta)
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: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: f845bc9d-2b3c-4ded-8bd7-f81e699d63a5
⛔ Files ignored due to path filters (1)
harness/pnpm-lock.yamlis excluded by!**/pnpm-lock.yaml
📒 Files selected for processing (45)
console/web/src/lib/devtools-stream.test.tsconsole/web/src/lib/devtools-stream.tsconsole/web/src/pages/Traces/api/traces.test.tsconsole/web/src/pages/Traces/api/traces.tsconsole/web/src/pages/Traces/components/FlameGraph.tsxconsole/web/src/pages/Traces/components/ServiceBreakdown.tsxconsole/web/src/pages/Traces/components/SessionDetailPanel.tsxconsole/web/src/pages/Traces/components/SpanErrorsTab.tsxconsole/web/src/pages/Traces/components/SpanOtelLogsTab.tsxconsole/web/src/pages/Traces/components/TraceFilters.tsxconsole/web/src/pages/Traces/components/TraceGroupsView.tsxconsole/web/src/pages/Traces/components/WaterfallChart.tsxconsole/web/src/pages/Traces/hooks/useTraceData.tsconsole/web/src/pages/Traces/hooks/useTraceGroups.tsconsole/web/src/pages/Traces/index.tsxconsole/web/src/pages/Traces/lib/attributeText.test.tsconsole/web/src/pages/Traces/lib/attributeText.tsconsole/web/src/pages/Traces/lib/minimapMarkers.test.tsconsole/web/src/pages/Traces/lib/minimapMarkers.tsconsole/web/src/pages/Traces/lib/percent.test.tsconsole/web/src/pages/Traces/lib/percent.tsconsole/web/src/pages/Traces/lib/spanTree.test.tsconsole/web/src/pages/Traces/lib/spanTree.tsconsole/web/src/pages/Traces/lib/traceListItem.test.tsconsole/web/src/pages/Traces/lib/traceListItem.tsconsole/web/src/pages/Traces/lib/traceTransform.test.tsconsole/web/src/pages/Traces/lib/traceTransform.tsconsole/web/src/pages/Traces/lib/treeFlatten.test.tsconsole/web/src/pages/Traces/lib/treeFlatten.tsharness/package.jsonharness/src/context-compaction/handler-async.tsharness/src/context-compaction/handler-sync.tsharness/src/context-compaction/prune.tsharness/src/harness/fanout/index.tsharness/src/harness/fanout/traces-changed.tsharness/src/runtime/otel.tsharness/src/turn-orchestrator/assistant-streaming/coalesce-deltas.tsharness/src/turn-orchestrator/assistant-streaming/run.tsharness/src/turn-orchestrator/events.tsharness/src/turn-orchestrator/state-runtime/store.tsharness/tests/harness/fanout/traces-changed.test.tsharness/tests/runtime/log-bridge.test.tsharness/tests/turn-orchestrator/coalesce-deltas.test.tsharness/tests/turn-orchestrator/events.test.tsharness/tests/turn-orchestrator/store.test.ts
| const subscribe = () => | ||
| client | ||
| .call('ui::subscribe', { browser_id: client.browserId, session_id: null }) | ||
| .then(() => dlog('ui::subscribe ok (all sessions)')) | ||
| .catch((err) => dlog('ui::subscribe failed', err)) | ||
|
|
||
| subscribe() |
There was a problem hiding this comment.
Retry failed ui::subscribe calls instead of only logging them.
If this RPC rejects once on mount or reconnect, this browser stays unsubscribed until the next connection-state transition because both call sites reuse this helper and the error is swallowed. With polling removed, that silently disables live refresh while the socket remains connected. Please schedule a retry/backoff here or surface the failure so the page can recover.
Suggested direction
- const subscribe = () =>
- client
- .call('ui::subscribe', { browser_id: client.browserId, session_id: null })
- .then(() => dlog('ui::subscribe ok (all sessions)'))
- .catch((err) => dlog('ui::subscribe failed', err))
+ let subscribeRetry: ReturnType<typeof setTimeout> | undefined
+ const subscribe = () =>
+ client
+ .call('ui::subscribe', { browser_id: client.browserId, session_id: null })
+ .then(() => dlog('ui::subscribe ok (all sessions)'))
+ .catch((err) => {
+ dlog('ui::subscribe failed', err)
+ subscribeRetry = setTimeout(() => {
+ void subscribe()
+ }, 1000)
+ })
...
return () => {
+ if (subscribeRetry) clearTimeout(subscribeRetry)
off()🤖 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 `@console/web/src/lib/devtools-stream.ts` around lines 98 - 104, The subscribe
helper currently swallows errors from client.call('ui::subscribe') causing the
browser to remain unsubscribed; change subscribe (and its callers) to retry on
failure with exponential backoff until success (or expose the error to
higher-level connection state so the page can recover). Specifically, update the
subscribe function that calls client.call('ui::subscribe', { browser_id:
client.browserId, session_id: null }) to implement a retry loop/backoff (e.g.,
initial delay, exponential increase, jitter) that re-invokes client.call on
rejection and only logs when attempts fail but continues retrying, and ensure
callers observe success (resolve) so live refresh is re-enabled once subscribe
succeeds; keep using dlog for status but do not swallow the rejection
permanently.
| } else { | ||
| setTraceListItems([]) | ||
| setHasOtelConfigured(false) | ||
| // Reset the dedup state so a later non-empty fetch is detected as | ||
| // fresh (otherwise the fingerprint/new-trace diff would be stale). | ||
| fingerprintRef.current = '' | ||
| prevTraceIdsRef.current = new Set() | ||
| } |
There was a problem hiding this comment.
Clear the hover buffer when the query result becomes empty.
pendingTracesRef.current can still hold the previous non-empty list if an update arrived while hovered. This branch clears the visible list, but flushPendingTraces() will later replay that stale buffer and bring old traces back after a clear/filter change.
Suggested fix
} else {
+ pendingTracesRef.current = null
setTraceListItems([])
// Reset the dedup state so a later non-empty fetch is detected as
// fresh (otherwise the fingerprint/new-trace diff would be stale).
fingerprintRef.current = ''
prevTraceIdsRef.current = new Set()
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| } else { | |
| setTraceListItems([]) | |
| setHasOtelConfigured(false) | |
| // Reset the dedup state so a later non-empty fetch is detected as | |
| // fresh (otherwise the fingerprint/new-trace diff would be stale). | |
| fingerprintRef.current = '' | |
| prevTraceIdsRef.current = new Set() | |
| } | |
| } else { | |
| pendingTracesRef.current = null | |
| setTraceListItems([]) | |
| // Reset the dedup state so a later non-empty fetch is detected as | |
| // fresh (otherwise the fingerprint/new-trace diff would be stale). | |
| fingerprintRef.current = '' | |
| prevTraceIdsRef.current = new Set() | |
| } |
🤖 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 `@console/web/src/pages/Traces/hooks/useTraceData.ts` around lines 120 - 126,
The else branch that clears visible state (setTraceListItems([]),
fingerprintRef.current = '', prevTraceIdsRef.current = new Set()) misses
clearing the hover/pending buffer so flushPendingTraces() can later reintroduce
stale items; add pendingTracesRef.current = [] (or an empty array/object
matching its shape) in that same branch to reset the hover buffer and prevent
old traces from being replayed, ensuring flushPendingTraces() has nothing to
replay after a clear/filter change.
| // Classify each span's parent chain as either acyclic-to-a-root or | ||
| // cyclic. A span is only linked under its parent when its whole ancestor | ||
| // chain terminates at a real root without revisiting a node; otherwise it | ||
| // is promoted to a root. This keeps a self-parent (`parent === self`) or a | ||
| // mutual cycle (a↔b) from (a) dropping the span out of `roots` entirely, | ||
| // and (b) building a child cycle that would infinite-loop the | ||
| // critical-path DFS below and `flattenTree` downstream. Mirrors the | ||
| // `visiting` guard already used by `calculateDepths`. | ||
| const SAFE = 1 | ||
| const CYCLIC = 2 | ||
| const chainMark = new Map<string, 1 | 2>() | ||
|
|
||
| function chainReachesRoot(startId: string): boolean { | ||
| const path: string[] = [] | ||
| let cur: string | undefined = startId | ||
| let safe = true | ||
| while (cur !== undefined) { | ||
| const cached = chainMark.get(cur) | ||
| if (cached !== undefined) { | ||
| safe = cached === SAFE | ||
| break | ||
| } | ||
| if (path.includes(cur)) { | ||
| safe = false | ||
| break | ||
| } | ||
| path.push(cur) | ||
| const parentId: string | undefined = spanMap.get(cur)?.parent_span_id | ||
| if (!parentId || parentId === cur || !spanMap.has(parentId)) { | ||
| safe = true | ||
| break | ||
| } | ||
| cur = parentId | ||
| } | ||
| for (const id of path) chainMark.set(id, safe ? SAFE : CYCLIC) | ||
| return safe | ||
| } |
There was a problem hiding this comment.
Finish the stack-safety work in this module.
The new parent-chain guard fixes cycles, but markCriticalPath, unmarkCriticalPath, and traverse are still recursive. A trace with a few thousand nested spans can still blow the call stack here, which means the Traces view can still blank on exactly the deep workflows this PR is trying to harden.
Also applies to: 177-209, 234-273
🤖 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 `@console/web/src/pages/Traces/lib/spanTree.ts` around lines 123 - 159, The
recursive functions markCriticalPath, unmarkCriticalPath, and traverse can blow
the JS call stack for deep parent/child chains; replace their recursion with
explicit iterative stack-based implementations that mirror current logic and
respect the existing cycle/visiting guards (use chainMark/SAFE/CYCLIC or the
same visiting checks used in calculateDepths) so behavior and
ordering/side-effects remain identical. For each function, traverse using a
manual stack frame that stores node id, iterator/index/state (e.g., entering vs
leaving) to emulate recursion, push children from the existing spanMap/children
collection, and perform the same marking/unmarking or visitor calls when popping
or during enter as the original recursive versions did; ensure
mutual/self-parent cycles are skipped and previously marked nodes are handled
consistently. Also ensure traversal preserves the original child order and
updates any depth/critical flags in the same points the recursive code did.
Finally, add unit tests or a synthetic deep-span trace to verify no stack
overflow and identical outcomes compared to the prior recursive behavior.
| async () => { | ||
| logger.debug('traces-changed: agent::turn_end received'); | ||
| if (timer) clearTimeout(timer); | ||
| timer = setTimeout(flush, COALESCE_MS); |
There was a problem hiding this comment.
Don't re-arm the timer on every turn_end.
This is a pure trailing debounce. If turns keep landing faster than every 400ms, clearTimeout(timer) keeps pushing flush() out and traces live-refresh can stall indefinitely under steady load. Schedule once per window instead of resetting the timer on every frame, or add a maxWait.
Suggested change
async () => {
logger.debug('traces-changed: agent::turn_end received');
- if (timer) clearTimeout(timer);
+ if (timer) return null;
timer = setTimeout(flush, COALESCE_MS);
return null;
},📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| async () => { | |
| logger.debug('traces-changed: agent::turn_end received'); | |
| if (timer) clearTimeout(timer); | |
| timer = setTimeout(flush, COALESCE_MS); | |
| async () => { | |
| logger.debug('traces-changed: agent::turn_end received'); | |
| if (timer) return null; | |
| timer = setTimeout(flush, COALESCE_MS); | |
| } |
🤖 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 `@harness/src/harness/fanout/traces-changed.ts` around lines 58 - 61, The
current turn_end handler clears and re-arms timer each event which creates a
trailing debounce that can never fire under steady high-frequency events; change
the logic in the async turn_end callback so you only schedule flush once per
coalescing window (e.g., if (!timer) timer = setTimeout(flush, COALESCE_MS)) or
implement a max-wait by capturing the timestamp of the first event and also
scheduling a secondary timeout to force flush after MAX_WAIT; reference the
symbols timer, flush, COALESCE_MS and the async turn_end handler to locate and
update the code.
| const entries: Record<string, { value: string }> = { | ||
| 'iii.function.id': { value: functionId }, | ||
| }; | ||
| if (ids.sessionId !== undefined) entries['iii.session.id'] = { value: ids.sessionId }; | ||
| if (ids.messageId !== undefined) entries['iii.message.id'] = { value: ids.messageId }; | ||
|
|
||
| const baggage = propagation.createBaggage(entries); | ||
| const ctxWithBaggage = propagation.setBaggage(otelContext.active(), baggage); |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the relevant region in harness/src/runtime/otel.ts
FILE="harness/src/runtime/otel.ts"
echo "== File: $FILE =="
nl -ba "$FILE" | sed -n '230,330p'
# Find other baggage-related calls in this file for broader context
echo "== setBaggage / getBaggage / createBaggage usages =="
rg -n "setBaggage|getBaggage|createBaggage|setEntry" "$FILE" || true
# Find where the ctxWithBaggage is used and what other baggage wrappers do
echo "== Context.with around ctxWithBaggage =="
rg -n "ctxWithBaggage|propagation\.setBaggage" "$FILE" || trueRepository: iii-hq/workers
Length of output: 159
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Confirm file exists
FILE="harness/src/runtime/otel.ts"
echo "== Confirm file =="
ls -la "$FILE" || (echo "Missing: $FILE" && exit 1)
echo "== Show lines ~260-310 (with line numbers) =="
cat -n "$FILE" | sed -n '250,320p'
echo "== Search for baggage-related APIs in this file =="
rg -n "propagation\.setBaggage|propagation\.getBaggage|propagation\.createBaggage|setEntry" "$FILE" || true
echo "== Search for baggage usage/wrappers in this file =="
rg -n "baggage|Baggage" "$FILE" || trueRepository: iii-hq/workers
Length of output: 5463
🌐 Web query:
OpenTelemetry JavaScript propagation.setBaggage replaces existing baggage createBaggage setEntry semantics
💡 Result:
In OpenTelemetry JavaScript, the Baggage API follows immutable design patterns, meaning neither setBaggage nor setEntry modifies existing objects in place [1][2][3]. Instead, they return new instances, which you must manually manage to update the context [1][3]. The semantics for managing baggage are as follows: 1. Baggage Immutability: Both Baggage instances and Context instances are immutable [1][3]. Methods like setEntry and setBaggage do not update the original object but return a new, updated copy [1][4][3]. 2. setEntry Semantics: The Baggage.setEntry(key, entry) method returns a new Baggage instance containing the existing entries plus the newly set or updated entry [4]. It does not overwrite the original Baggage object [3]. If you wish to add multiple entries, you should chain the setEntry calls (or iterate over them) to produce the final Baggage object before applying it to a context [3]. 3. setBaggage Semantics: The propagation.setBaggage(context, baggage) method returns a new Context object that includes the provided Baggage [5][6]. It does not modify the context passed as an argument [1][2]. To update baggage in the current active context, you must follow this pattern [3]: 1. Retrieve the existing baggage using propagation.getBaggage(context.active()) [1][7][3]. 2. Create a new Baggage object (or use an empty one if none exists) and add/update entries using .setEntry() [1][6][3]. 3. Create a new context using propagation.setBaggage(context.active(), newBaggage) [6][3]. 4. Use context.with(newContext, callback) to execute your code within the context containing the updated baggage [1][2][3]. Common mistakes include assuming that calling these methods updates the global or current state directly without using the returned context in subsequent operations [1][2].
Citations:
- 1: How to properly set baggage in active context and propagate it to all open-telemetry/opentelemetry-js#4002
- 2: Setting a baggage value through the sdk api open-telemetry/opentelemetry-js#1787
- 3: https://oneuptime.com/blog/post/2026-01-07-opentelemetry-baggage-propagation/view
- 4: https://open-telemetry.github.io/opentelemetry-js/interfaces/_opentelemetry_sdk-node._opentelemetry_api.Baggage.html
- 5: https://open-telemetry.github.io/opentelemetry-js/classes/_opentelemetry_sdk-node._opentelemetry_api.PropagationAPI.html
- 6: https://oneuptime.com/blog/post/2026-01-24-baggage-propagation-opentelemetry/view
- 7: https://www.scoutapm.com/blog/opentelemetry-in-javascript
🌐 Web query:
@opentelemetry/api PropagationAPI setBaggage getBaggage createBaggage Baggage.setEntry documentation
💡 Result:
In the OpenTelemetry JavaScript API, Baggage is a collection of key-value pairs used to propagate contextual information across distributed systems [1]. Because Baggage objects are immutable, they are typically managed by creating new instances or modifying the context rather than editing an existing Baggage object in place [2][3]. Key components of the API include: 1. createBaggage: This method is used to initialize a new, empty Baggage instance [2][3]. It is the starting point for building a new set of entries. 2. Baggage.setEntry: This method adds or updates a key-value entry in the Baggage [2]. Crucially, because Baggage is immutable, setEntry does not modify the original instance; instead, it returns a new Baggage object containing the new entry [2][3]. The method signature is setEntry(key: string, entry: BaggageEntry): Baggage [2]. 3. Propagation and Context: To make Baggage available to downstream operations, you must associate it with a Context [1][3]. You use the propagation API (often via @opentelemetry/api) to inject the Baggage into a context [3]. - propagation.setBaggage(context, baggage) returns a new Context that includes the provided Baggage [3]. - propagation.getBaggage(context) extracts the Baggage from a given context [3]. To use these together, you typically: 1. Obtain the current (or empty) baggage [3]. 2. Use setEntry to add your data, receiving a new Baggage instance [2][3]. 3. Create a new Context using propagation.setBaggage with your active context and the new Baggage instance [3]. 4. Use context.with(newContext,...) to activate this context for the duration of your operation [3]. Note that while some language SDKs (like Python) provide helper functions like set_baggage directly on the module level to simplify this process [4], the JavaScript API follows the immutable pattern which necessitates managing the Context object explicitly [3]. Always ensure you are working with the returned objects, as the originals remain unchanged [2][3].
Citations:
- 1: https://opentelemetry.io/docs/concepts/signals/baggage/
- 2: https://open-telemetry.github.io/opentelemetry-js/interfaces/_opentelemetry_sdk-node._opentelemetry_api.Baggage.html
- 3: How to properly set baggage in active context and propagate it to all open-telemetry/opentelemetry-js#4002
- 4: https://opentelemetry-python.readthedocs.io/en/latest/api/baggage.html
🌐 Web query:
OpenTelemetry JS Baggage setEntry returns new instance immutable
💡 Result:
In OpenTelemetry JS, Baggage#setEntry returns a new Baggage instance (i.e., it does not mutate the existing one). [1][2] This behavior is required because the Baggage container MUST be immutable, so that the containing Context remains immutable. [3] So if you call setEntry on a baggage object, you must use the returned baggage and (when using Context) attach it to a new Context via propagation.setBaggage / context.with. Contexts (and their attached baggage) are immutable as well. [4][3]
Citations:
- 1: https://open-telemetry.github.io/opentelemetry-js/interfaces/_opentelemetry_sdk-node._opentelemetry_api.Baggage.html
- 2: https://github.com/open-telemetry/opentelemetry-js/blob/main/api/src/baggage/types.ts
- 3: https://opentelemetry.io/docs/specs/otel/baggage/api/
- 4: How to properly set baggage in active context and propagate it to all open-telemetry/opentelemetry-js#4002
🌐 Web query:
@opentelemetry/api PropagationAPI createBaggage signature arguments required optional default empty
💡 Result:
The createBaggage function in the @opentelemetry/api package (JavaScript) has the following signature: createBaggage(entries?: Record<string, BaggageEntry>): Baggage [1][2] Arguments: entries (optional): A record (object) where the keys are strings and the values are BaggageEntry objects [1][2]. This parameter is used to initialize the baggage with specific entries [1][2]. Default: If no arguments are provided, the entries parameter defaults to an empty object ({}) [2]. In practice, when calling this function to create baggage, you provide an object mapping your desired keys to their corresponding BaggageEntry objects (which typically contain the value and optional metadata) [2][3]: const baggage = propagation.createBaggage({ 'my-key': { value: 'my-value' } }); [3] The resulting Baggage object can then be associated with a context using propagation.setBaggage [4][3].
Citations:
- 1: https://open-telemetry.github.io/opentelemetry-js/classes/_opentelemetry_sdk-node._opentelemetry_api.PropagationAPI.html
- 2: https://github.com/open-telemetry/opentelemetry-js-api/blob/main/src/baggage/utils.ts
- 3: https://docs.honeycomb.io/send-data/javascript-nodejs/opentelemetry-sdk
- 4: https://open-telemetry.github.io/opentelemetry-js/classes/_opentelemetry_api._opentelemetry_api.PropagationAPI.html
🌐 Web query:
propagation.setBaggage merge existing baggage JavaScript
💡 Result:
In OpenTelemetry JavaScript, Baggage is immutable [1]. You cannot modify an existing Baggage object or context directly; instead, you must create a new version of the Baggage that includes your changes and then set that new Baggage into a context [1][2]. To merge new entries with existing baggage while preserving the old ones, follow this pattern [2]: 1. Retrieve the existing baggage from the current context using propagation.getActiveBaggage [1][2]. 2. If baggage exists, use the setEntry method on the existing baggage instance to create a new, updated Baggage object [2][3]. If no baggage exists, use propagation.createBaggage to initialize a new one [2]. 3. Apply the new context using context.with [1][4]. Example implementation: // 1. Get current baggage or create an empty one if none exists const currentBaggage = propagation.getActiveBaggage || propagation.createBaggage; // 2. Create new baggage containing the existing entries plus the new one // setEntry returns a new, immutable Baggage instance const updatedBaggage = currentBaggage.setEntry('my-key', { value: 'my-value' }); // 3. Create a new context with the updated baggage const newContext = propagation.setBaggage(context.active, updatedBaggage); // 4. Run your operations within this new context context.with(newContext, => { // Now 'my-key' is available in the active context }); Note that setBaggage does not modify the original context; it returns a new one [1][4]. Always use context.with or similar mechanisms to ensure your subsequent operations use the updated context [1].
Citations:
- 1: How to properly set baggage in active context and propagate it to all open-telemetry/opentelemetry-js#4002
- 2: https://oneuptime.com/blog/post/2026-01-24-baggage-propagation-opentelemetry/view
- 3: https://open-telemetry.github.io/opentelemetry-js/interfaces/_opentelemetry_sdk-node._opentelemetry_api.Baggage.html
- 4: Setting a baggage value through the sdk api open-telemetry/opentelemetry-js#1787
Merge the new iii.* baggage into the existing active baggage instead of replacing it
instrumentHandler creates a brand-new Baggage containing only iii.function.id/iii.session.id/iii.message.id and then calls propagation.setBaggage(otelContext.active(), baggage), which attaches that fresh instance to the returned context—dropping any unrelated baggage already present on the active context for downstream work inside handler.
Suggested fix
- const baggage = propagation.createBaggage(entries);
- const ctxWithBaggage = propagation.setBaggage(otelContext.active(), baggage);
+ let baggage = propagation.getBaggage(otelContext.active()) ?? propagation.createBaggage();
+ for (const [key, entry] of Object.entries(entries)) {
+ baggage = baggage.setEntry(key, entry);
+ }
+ const ctxWithBaggage = propagation.setBaggage(otelContext.active(), baggage);
return otelContext.with(ctxWithBaggage, () => handler(input));📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| const entries: Record<string, { value: string }> = { | |
| 'iii.function.id': { value: functionId }, | |
| }; | |
| if (ids.sessionId !== undefined) entries['iii.session.id'] = { value: ids.sessionId }; | |
| if (ids.messageId !== undefined) entries['iii.message.id'] = { value: ids.messageId }; | |
| const baggage = propagation.createBaggage(entries); | |
| const ctxWithBaggage = propagation.setBaggage(otelContext.active(), baggage); | |
| const entries: Record<string, { value: string }> = { | |
| 'iii.function.id': { value: functionId }, | |
| }; | |
| if (ids.sessionId !== undefined) entries['iii.session.id'] = { value: ids.sessionId }; | |
| if (ids.messageId !== undefined) entries['iii.message.id'] = { value: ids.messageId }; | |
| let baggage = propagation.getBaggage(otelContext.active()) ?? propagation.createBaggage(); | |
| for (const [key, entry] of Object.entries(entries)) { | |
| baggage = baggage.setEntry(key, entry); | |
| } | |
| const ctxWithBaggage = propagation.setBaggage(otelContext.active(), baggage); |
🤖 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 `@harness/src/runtime/otel.ts` around lines 280 - 287, The current code
replaces any existing baggage by creating a new Baggage and calling
propagation.setBaggage(...), so merge the new iii.* entries into the active
baggage instead: call propagation.getBaggage(otelContext.active()) to obtain the
current baggage (or start with an empty baggage), then for each entry
('iii.function.id', 'iii.session.id', 'iii.message.id') call baggage =
baggage.setEntry(key, { value }) to produce a merged Baggage, and finally call
propagation.setBaggage(otelContext.active(), baggage) so existing baggage
entries are preserved; reference propagation.getBaggage,
propagation.createBaggage, baggage.setEntry, and propagation.setBaggage in
otel.ts.
| async function flush(): Promise<void> { | ||
| clearTimer(); | ||
| if (!buf) return; | ||
| const merged = { ...buf.last, delta: buf.delta } as AssistantMessageEvent; | ||
| const { partial } = buf; | ||
| buf = null; | ||
| await emit(partial, merged); | ||
| } | ||
|
|
||
| // Periodic throttle, NOT a trailing debounce: a steady ~flushMs cadence keeps | ||
| // text appearing smoothly while continuous deltas arrive. A trailing debounce | ||
| // would withhold all text until the stream paused. | ||
| function arm(): void { | ||
| if (timer) return; | ||
| timer = setTimeout(() => { | ||
| void flush(); | ||
| }, flushMs); |
There was a problem hiding this comment.
Serialize timer flushes and keep the buffer until emit succeeds.
flush() nulls buf before await emit(...), and the timer path calls it via void flush(). If emit rejects on that path, the buffered delta is already lost; because that timer flush is also untracked, later discrete events can overtake the older buffered emit and break ordering. Please route timer/manual flushes through one in-flight promise and only clear or replace the buffer after the emit has completed successfully.
🤖 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 `@harness/src/turn-orchestrator/assistant-streaming/coalesce-deltas.ts` around
lines 50 - 66, The timer-driven flush currently nulls buf before awaiting emit,
causing lost/ordered deltas if emit rejects; change flush/arm to serialize
flushes by introducing an in-flight promise (e.g., inFlightFlush) that the timer
path awaits instead of calling void flush(), and only clear or replace buf after
emit resolves successfully; ensure clearTimer still prevents multiple timers,
have flush capture currentBuf locally, await emit(currentBuf.partial, merged)
and on success then set buf = null and resolve inFlightFlush, and have
subsequent flush callers wait on inFlightFlush to preserve ordering.
| const { final, error } = await ports.streamTurn(ctx, async (partial, event) => { | ||
| await ports.emitMessageUpdate(session_id, partial, event); | ||
| if (event.type === 'text_delta' || event.type === 'thinking_delta') { | ||
| body_streamed = true; | ||
| } | ||
| await coalescer.onEvent(partial, event); | ||
| }); | ||
| await coalescer.flush(); | ||
|
|
There was a problem hiding this comment.
Flush the coalescer in a finally block.
If ports.streamTurn(...) rejects after buffering some deltas, control never reaches Line 73 and the tail is dropped. Wrap the stream call in try/finally so buffered text is flushed even when the provider transport fails.
Suggested fix
- const { final, error } = await ports.streamTurn(ctx, async (partial, event) => {
- if (event.type === 'text_delta' || event.type === 'thinking_delta') {
- body_streamed = true;
- }
- await coalescer.onEvent(partial, event);
- });
- await coalescer.flush();
+ const { final, error } = await (async () => {
+ try {
+ return await ports.streamTurn(ctx, async (partial, event) => {
+ if (event.type === 'text_delta' || event.type === 'thinking_delta') {
+ body_streamed = true;
+ }
+ await coalescer.onEvent(partial, event);
+ });
+ } finally {
+ await coalescer.flush();
+ }
+ })();
return { final, error, body_streamed };📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| const { final, error } = await ports.streamTurn(ctx, async (partial, event) => { | |
| await ports.emitMessageUpdate(session_id, partial, event); | |
| if (event.type === 'text_delta' || event.type === 'thinking_delta') { | |
| body_streamed = true; | |
| } | |
| await coalescer.onEvent(partial, event); | |
| }); | |
| await coalescer.flush(); | |
| const { final, error } = await (async () => { | |
| try { | |
| return await ports.streamTurn(ctx, async (partial, event) => { | |
| if (event.type === 'text_delta' || event.type === 'thinking_delta') { | |
| body_streamed = true; | |
| } | |
| await coalescer.onEvent(partial, event); | |
| }); | |
| } finally { | |
| await coalescer.flush(); | |
| } | |
| })(); | |
🤖 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 `@harness/src/turn-orchestrator/assistant-streaming/run.ts` around lines 67 -
74, The call to ports.streamTurn(...) may reject and skip the subsequent await
coalescer.flush(), losing buffered deltas; wrap the stream call in a try/finally
so coalescer.flush() is always awaited: call ports.streamTurn(ctx, async
(partial, event) => { ... }) inside a try, capture its result into { final,
error } as before, and then in a finally block await coalescer.flush() (so
coalescer.onEvent and body_streamed handling remain unchanged); preserve
propagation of any thrown error after flushing if needed.
| /** Unique per process run; prefixes every item_id so a restart can't collide. */ | ||
| const PROCESS_EPOCH = uuidLike(); | ||
| /** | ||
| * Per-session monotonic counter. One small integer per session seen by this | ||
| * process; never reset across turns (matching the old persisted counter's | ||
| * monotonic-per-session semantics), so item_ids stay unique within the process. | ||
| */ | ||
| const seqBySession = new Map<string, number>(); | ||
|
|
||
| function nextSeq(session_id: string): number { | ||
| const n = seqBySession.get(session_id) ?? 0; | ||
| seqBySession.set(session_id, n + 1); | ||
| return n; | ||
| } | ||
|
|
||
| /** Test seam: clear the in-process counters. Do not call in production. */ | ||
| export function _resetSeqForTests(): void { | ||
| seqBySession.clear(); | ||
| } |
There was a problem hiding this comment.
Bound seqBySession or reclaim inactive sessions.
This map now grows forever with every distinct session_id the process ever sees. On a long-lived harness with session churn, event sequencing becomes unbounded in-memory state. Since the counter only feeds opaque item_ids, please add a reclamation strategy so inactive sessions can be released safely.
🤖 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 `@harness/src/turn-orchestrator/events.ts` around lines 31 - 49, The
seqBySession Map currently grows unbounded; change it to track last-seen
timestamps and add a reclamation policy: replace seqBySession: Map<string,
number> with Map<string, {n:number, last:number}> (keep PROCESS_EPOCH and
nextSeq names), update nextSeq(session_id) to bump both counter and last
timestamp, and introduce constants (e.g. MAX_SESSIONS and SESSION_TTL_MS) plus a
cleanup function that evicts entries older than SESSION_TTL_MS or prunes the
oldest entries when size > MAX_SESSIONS; call the cleanup from nextSeq (and run
it periodically with setInterval) so inactive session entries are reclaimed, and
ensure _resetSeqForTests() still clears the map and stops/resets any timer used
for cleanup.
| const nextView = toView(rec); | ||
| const prevView = prev != null ? toView(prev) : undefined; | ||
| const viewChanged = | ||
| prevView === undefined || JSON.stringify(prevView) !== JSON.stringify(nextView); | ||
|
|
||
| if (viewChanged) { | ||
| await emitTurnStateChanged( | ||
| iii, | ||
| rec.session_id, | ||
| prev == null ? 'state:created' : 'state:updated', | ||
| nextView, | ||
| prevView, | ||
| ); | ||
| } |
There was a problem hiding this comment.
Use the persisted old_value as the dedup source of truth.
This block compares against prev, but prev may come from the optional previous argument instead of the value actually overwritten by state::set. If that hint is stale, we can suppress a real turn_state_changed, emit the wrong old_value, and return the wrong prior state to saveRecord(). Please base viewChanged and the emitted old view on result.old_value instead.
🤖 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 `@harness/src/turn-orchestrator/state-runtime/store.ts` around lines 96 - 109,
The current change-detection uses prev (possibly the caller-supplied previous)
to compute prevView and viewChanged, but the review asks to use the persisted
overwrite result's old_value as the source of truth; update the logic that
computes prevView and viewChanged to derive prevView from result.old_value
(convert via toView) instead of prev, and pass that derived prevView into
emitTurnStateChanged (and use it for the 'state:created' vs 'state:updated'
decision) so the emitted old_value and dedup compare use the actual persisted
prior value rather than the optional previous param; update references around
nextView, toView, viewChanged, and emitTurnStateChanged accordingly.
- Added `@iii-dev/observability` as a new dependency at version 0.16.1. - Updated `iii-sdk` dependency to version 0.16.1. - Refactored telemetry imports in several files to use `@iii-dev/observability` instead of `iii-sdk/telemetry`. - Updated `pnpm-lock.yaml` to reflect the new dependency versions and ensure consistency across the project.
…pdate Each provider stream delta previously emitted its own message_update, and every emit fired two instrumented functions (state::update + stream::set), so a turn's span/RPC count scaled with output tokens (~388 spans / 2.5MB for a ~3s turn). Add a delta coalescer that merges consecutive same-type deltas (text / thinking / functioncall) into a single message_update on a ~60ms periodic flush, flushing through discrete events 1:1 and on the final boundary. The console renders streaming text by appending llm_event.delta, so concatenating deltas is wire-identical — no console change required.
Replace nextSeq()'s persisted state::update increment (one per agent event, ~66/turn) with an in-process per-session counter. The seq only feeds the opaque stream item_id — the console never reads it, the fanout strips it, and the engine delivers frames in insertion order — so a process-local counter is behavior-preserving. item_ids are prefixed with a random per-process epoch (uuidLike) so a restart can't collide with frames the previous process wrote for the same session. Each removed state::update also removes one engine state-trigger evaluation pass; on a streaming-heavy turn this takes ~66 writes (and ~66 state_triggers) to zero.
persistRecord emitted turn_state_changed on every turn_state write, even when the consumer-visible view was unchanged (transitionTo bumps updated_at_ms on every call, which toView drops). Diff the view and skip the emit when it's identical — removing a redundant stream::set (and its engine trigger-evaluation pass) per stale-skip / idempotent save. Always emit state:created so a fresh session still seeds the console mirror; the console's turn_state_changed handler is diff-based, so a suppressed no-op is observably identical.
060d415 to
8901015
Compare
ytallo
changed the title
|
Split out of the original combined PR: the Traces push live-refresh + page improvements now live in #206. This PR is harness telemetry/span-volume only. |
2 participants
Summary
Cut per-turn telemetry/span volume on the harness (streaming-delta coalescing, in-process event sequencing, no-op event suppression) plus a telemetry-import/deps housekeeping pass. All changes are observability-quality / cost reductions — no agent behavior change, no wire-contract or rendering change.
Changes
turn-orchestrator/assistant-streaming/coalesce-deltas.ts): merge consecutive same-type provider deltas (text/thinking/functioncall) into a singlemessage_updateon a ~60ms periodic flush. The console renders streaming text by appendingllm_event.delta, so concatenating deltas is wire-identical — no console change. Cuts the per-token emits (and the spans/RPCs each generates) on streaming-heavy turns.turn-orchestrator/events.ts):nextSeqno longer issues a persistedstate::updateper agent event; it uses an in-process per-session counter, with a random per-process epoch in the streamitem_idso a restart can't collide with prior frames (item_idis opaque — never read by the console, stripped by the fanout, delivered in insertion order). Removes one engine state write (and its trigger evaluation) per event.turn_state_changedemit on no-op saves (turn-orchestrator/state-runtime/store.ts):persistRecordonly emits when the consumer-visible view (toView) changes —transitionTobumpsupdated_at_mson every write, whichtoViewdrops — removing a redundantstream::set(and its engine trigger-evaluation pass) per stale-skip / idempotent save.state:createdis always emitted so a fresh session still seeds the console mirror.iii-sdk/telemetryto@iii-dev/observability(0.16.1),runtime/otel.tssimplification, and the dependency bump.Behavior / compatibility
llm_event.type).Test plan
pnpm test, 1051 tests) +pnpm typecheck.engine::traces::tree) — per-eventstate::updateeliminated; coalesced streamingmessage_updates.Summary by CodeRabbit
Release Notes
New Features
Improvements
Chores