Skip to content

Commit ed1eecd

Browse files
ralphstodomingoclaudeanandgupta42
authored
fix: write session traces in headless serve mode (#886)
* fix: write session traces in headless serve mode Session tracing was implemented only at the terminal entrypoints — the TUI worker and `run` construct the tracer and feed it events, but `serve` had no trace wiring at all. Sessions driven over HTTP (e.g. the VS Code chat panel, which spawns `altimate serve` and POSTs to `/session/{id}/prompt_async`) never wrote `ses_*.json` files to `~/.local/share/altimate-code/traces/`: the only hook in the shared session pipeline is `Tracer.active?.logSpan`, and `Tracer.active` was never set in serve mode. - Extract the TUI worker's inline event-stream→trace logic into a shared `TraceConsumer` (`altimate/observability/trace-consumer.ts`): config loading, per-session trace map, MAX_TRACES eviction, event feeding, reset/flush - Rewire `cli/cmd/tui/worker.ts` to use the shared consumer (behavior unchanged) - Add `subscribeTraceConsumer()` — an in-process `/event` subscription mirroring the worker's loop — and start it in `ServeCommand`, so serve sessions produce the same trace files as the terminal - Tests: full event-sequence → completed trace file, interleaved sessions, malformed events, disabled config, flush/reset semantics Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix: don't clobber trace files with post-idle straggler events Real serve sessions emit trailing events after `session.status: idle` — the user message is re-published with its generated `summary` once title generation completes. Finalizing immediately on idle deleted the trace from the map, and the straggler then lazily re-created an EMPTY trace for the same sessionID whose finalization overwrote the rich `<sessionId>.json` file (observed end-to-end in docker code-server: 42KB trace clobbered down to 749 bytes). Fix: schedule finalization 2s after idle instead of finalizing inline; any event for the session during the grace window pushes the deadline back and is absorbed into the live trace. As a bonus the generated session title now lands in trace metadata. flush()/reset()/eviction clear pending timers. Verified: chat session in docker code-server (glob tool call + 2 generations) produces a complete 42KB trace with title, model, tool span, and token totals. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * test: update event-literal invariant for trace-consumer extraction The worker's inline event handling moved into the shared altimate/observability/trace-consumer.ts — point the cross-module event-type literal invariant at the new consumer (and cover the full set it consumes) instead of cli/cmd/tui/worker.ts. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * chore: wrap startEventStream trace reset in altimate_change markers worker.ts is an upstream-shared file; the traceConsumer.reset() call (which replaced main's inline sessionTraces clear) must be wrapped so the upstream merge tooling protects it. Caught by script/upstream/analyze.ts --markers --strict (the Marker Guard CI check). * chore: wrap traceConsumer.loadConfig call in altimate_change markers * fix: finalize serve traces on shutdown + harden trace consumer (#886) Addresses the multi-model consensus review on PR #886. The `TraceConsumer` extraction was sound, but the `serve` integration had lifecycle gaps. `cli/cmd/serve.ts`: - Capture the `subscribeTraceConsumer` handle and wire `SIGINT`/`SIGTERM`/ `beforeExit` to `stop()` so traces are finalized on shutdown instead of left un-`completed`. Mirrors the existing pattern in `cli/cmd/run.ts`. `altimate/observability/trace-consumer.ts`: - `stop()` now drains the event loop (bounded by a 1s timeout) before `flush()`, so finalization can't race an in-flight `handleEvent`. - The `for await` stream loop catches mid-stream errors and reconnects with exponential backoff (250ms -> 30s) instead of dying permanently; backoff sleeps are abortable so shutdown isn't delayed. - Finalize and evict per-session state on `session.deleted` (a real end-of-session signal) so long-lived serve processes don't accumulate `sessionUserMsgIds` / cached traces. - `loadConfig` failure now fails safe (`enabled = false`) instead of silently tracing to the default dir, and logs at debug. - `getOrCreateTrace` and `handleEvent` catch blocks log at debug (still never throw) for headless diagnosability. - Auth header uses `Buffer.from(...).toString("base64")` (UTF-8 safe) instead of `btoa`. - Added a test seam: `subscribeTraceConsumer` accepts an optional injected `consumer` and `subscribe` source (production behaviour unchanged via defaults). Tests: - New `subscribeTraceConsumer` coverage: `stop()` drains + finalizes; a mid-stream throw reconnects rather than killing the loop. - New `TraceConsumer` coverage: incremental snapshots land on disk BEFORE any flush (the real serve path), and `session.deleted` finalizes + releases state. - De-flaked the `reset()` test (poll instead of fixed `setTimeout`). - Registered the `session.deleted` literal in `bridge-merge-invariants`. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix: address cubic review on serve trace lifecycle (#886) Follow-up to the consensus-review fixes, addressing cubic-dev-ai comments: - serve.ts: SIGINT/SIGTERM shutdown now exits with signal-conventional codes (130 / 143) instead of 0, so a signalled stop isn't masked as a successful run. beforeExit still exits 0. Matches cli/cmd/run.ts. - trace-consumer.ts (loadConfig): a config-load error no longer disables tracing — it keeps `enabled` true and falls back to Trace.create()'s default exporter (the original "must not prevent the host from working" intent), but now logs a warning so the fallback isn't silent. - trace-consumer.ts (getOrCreateTrace): drop the dead `Trace.setActive(trace)` call. It sets a process-global active trace that nothing reads and that is a footgun in multi-session serve (last-event-wins); per-session routing is via the sessionTraces map. Verified: marker guard, typecheck, and trace-consumer / tracing-followups / worker-trace-clearing / bridge-merge-invariants suites all green. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Co-authored-by: anandgupta42 <anand@altimate.ai>
1 parent 5ffeadb commit ed1eecd

7 files changed

Lines changed: 946 additions & 282 deletions

File tree

‎packages/opencode/src/altimate/observability/trace-consumer.ts‎

Lines changed: 443 additions & 0 deletions
Large diffs are not rendered by default.

‎packages/opencode/src/cli/cmd/serve.ts‎

Lines changed: 32 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,9 @@ import { Flag } from "../../flag/flag"
55
import { Workspace } from "../../control-plane/workspace"
66
import { Project } from "../../project/project"
77
import { Installation } from "../../installation"
8+
// altimate_change start — trace: session tracing in headless serve
9+
import { subscribeTraceConsumer } from "../../altimate/observability/trace-consumer"
10+
// altimate_change end
811

912
export const ServeCommand = cmd({
1013
command: "serve",
@@ -20,7 +23,35 @@ export const ServeCommand = cmd({
2023
console.log(`altimate-code server listening on http://${server.hostname}:${server.port}`)
2124
// altimate_change end
2225

26+
// altimate_change start — trace: session tracing in headless serve
27+
// Sessions driven over HTTP (e.g. the VS Code chat panel) have no TUI
28+
// worker observing the event stream, so traces were never written in
29+
// serve mode. Subscribe the shared trace consumer to the in-process
30+
// event stream so serve sessions produce the same trace files as the
31+
// terminal entrypoints.
32+
const traceSub = subscribeTraceConsumer({ directory: process.cwd() })
33+
34+
// Finalize traces on shutdown. `serve` blocks forever on the promise below
35+
// and otherwise dies abruptly on signal, so without these handlers the
36+
// consumer's stop()/flush()/endTrace() never runs and serve traces are
37+
// left un-finalized (status never "completed", no summary/narrative).
38+
// Mirrors the SIGINT/SIGTERM/beforeExit pattern in cli/cmd/run.ts.
39+
let isShuttingDown = false
40+
const shutdown = async (code: number) => {
41+
if (isShuttingDown) return
42+
isShuttingDown = true
43+
await traceSub.stop()
44+
await server.stop()
45+
process.exit(code)
46+
}
47+
// Exit with signal-conventional codes (128 + signal number) so a
48+
// SIGINT/SIGTERM isn't masked as a successful (0) run. beforeExit is a
49+
// normal drain, so it exits 0. Matches cli/cmd/run.ts.
50+
process.once("SIGINT", () => void shutdown(130))
51+
process.once("SIGTERM", () => void shutdown(143))
52+
process.once("beforeExit", () => void shutdown(0))
53+
// altimate_change end
54+
2355
await new Promise(() => {})
24-
await server.stop()
2556
},
2657
})

‎packages/opencode/src/cli/cmd/tui/worker.ts‎

Lines changed: 17 additions & 239 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ import type { BunWebSocketData } from "hono/bun"
1212
import { Flag } from "@/flag/flag"
1313
import { setTimeout as sleep } from "node:timers/promises"
1414
// altimate_change start — trace: session tracing in TUI
15-
import { Trace, FileExporter, HttpExporter, type TraceExporter } from "@/altimate/observability/tracing"
15+
import { TraceConsumer } from "@/altimate/observability/trace-consumer"
1616
// altimate_change end
1717

1818
await Log.init({
@@ -47,113 +47,22 @@ const eventStream = {
4747
abort: undefined as AbortController | undefined,
4848
}
4949

50-
// altimate_change start — trace: monotonic stream generation. Bumped on every
51-
// startEventStream() so an in-flight getOrCreateTrace() can detect that its
52-
// owning stream was torn down while it was suspended at an await. Keyed on a
53-
// counter rather than the AbortController's object identity so the guard does
54-
// not silently depend on startEventStream always allocating a fresh controller.
55-
let streamGeneration = 0
56-
// altimate_change end
57-
58-
// altimate_change start — trace: per-session traces
59-
const sessionTraces = new Map<string, Trace>()
60-
const sessionUserMsgIds = new Map<string, Set<string>>() // Per-session user message IDs (cleaned up on session end)
61-
const MAX_TRACES = 100
62-
63-
// Cached tracing config — loaded once at first use
64-
let tracingConfigLoaded = false
65-
let tracingEnabled = true
66-
let tracingExporters: TraceExporter[] | undefined
67-
let tracingMaxFiles: number | undefined
68-
69-
async function loadTracingConfig() {
70-
if (tracingConfigLoaded) return
71-
tracingConfigLoaded = true
72-
try {
73-
const cfg = await Config.get()
74-
const tc = cfg.tracing
75-
if (tc?.enabled === false) { tracingEnabled = false; return }
76-
const exporters: TraceExporter[] = [new FileExporter(tc?.dir)]
77-
if (tc?.exporters) {
78-
for (const exp of tc.exporters) {
79-
exporters.push(new HttpExporter(exp.name, exp.endpoint, exp.headers))
80-
}
81-
}
82-
tracingExporters = exporters
83-
tracingMaxFiles = tc?.maxFiles
84-
} catch {
85-
// Config failure should not prevent TUI from working
86-
}
87-
}
88-
// altimate_change end
89-
90-
// altimate_change start — trace: get or create per-session trace
91-
async function getOrCreateTrace(sessionID: string): Promise<Trace | null> {
92-
if (!sessionID || !tracingEnabled) return null
93-
if (sessionTraces.has(sessionID)) return sessionTraces.get(sessionID)!
94-
// altimate_change start — capture the stream generation that owns this call so
95-
// we can detect a concurrent startEventStream() (e.g. setWorkspace) that
96-
// aborted us and cleared the cache while we were suspended at the rehydrate
97-
// await below. A counter (not AbortController identity) so we don't depend on
98-
// startEventStream's allocation strategy.
99-
const generationAtEntry = streamGeneration
100-
// altimate_change end
101-
try {
102-
if (sessionTraces.size >= MAX_TRACES) {
103-
const oldest = sessionTraces.keys().next().value
104-
if (oldest) {
105-
Log.Default.warn(`[tracing] Evicting trace for session ${oldest} — ${MAX_TRACES} concurrent sessions reached`)
106-
sessionTraces.get(oldest)?.endTrace().catch(() => {})
107-
sessionTraces.delete(oldest)
108-
sessionUserMsgIds.delete(oldest)
109-
}
110-
}
111-
const trace = tracingExporters
112-
? Trace.withExporters([...tracingExporters], { maxFiles: tracingMaxFiles })
113-
: Trace.create()
114-
// altimate_change start — prefer disk-rehydration on cache miss for an
115-
// existing session (worker restart, MAX_TRACES eviction). startTrace would
116-
// push a fresh root span into empty `this.spans` and the immediate
117-
// snapshot would clobber the rich on-disk file. Defense in depth in
118-
// addition to keeping the cache alive across turns.
119-
// Async to keep the event-stream loop unblocked on large existing traces.
120-
if (!(await trace.rehydrateFromFile(sessionID))) {
121-
trace.startTrace(sessionID, {})
122-
}
123-
// altimate_change end
124-
// altimate_change start — if a new stream replaced ours while we were
125-
// awaiting rehydrate, this Trace belongs to a stream that's already been
126-
// aborted and its cache cleared. Inserting it now would resurrect an orphan
127-
// writer into the freshly-cleared map. Discard it and defer to whatever the
128-
// live stream has. The check and the set below run in the same synchronous
129-
// turn (no await between them), so the insert can't race a later
130-
// startEventStream — this closes the suspend-at-await hole specifically.
131-
if (streamGeneration !== generationAtEntry) {
132-
void trace.endTrace().catch(() => {})
133-
return sessionTraces.get(sessionID) ?? null
134-
}
135-
Trace.setActive(trace)
136-
sessionTraces.set(sessionID, trace)
137-
return trace
138-
// altimate_change end
139-
} catch {
140-
return null
141-
}
142-
}
50+
// altimate_change start — trace: per-session traces (shared consumer)
51+
// All per-session trace state + event handling lives in TraceConsumer so the
52+
// headless `serve` entrypoint (VS Code chat panel) gets identical behaviour.
53+
// reset() bumps the consumer's stream generation (the equivalent of the old
54+
// inline cache-clear) to invalidate any in-flight rehydrate.
55+
const traceConsumer = new TraceConsumer()
14356
// altimate_change end
14457

14558
const startEventStream = (input: { directory: string; workspaceID?: string }) => {
14659
if (eventStream.abort) eventStream.abort.abort()
147-
// altimate_change start — new stream generation; invalidates any in-flight
148-
// getOrCreateTrace() suspended at its rehydrate await (see generationAtEntry).
149-
streamGeneration++
60+
// altimate_change start — trace: clear stale per-stream trace state before
61+
// starting a new stream instance. reset() also bumps the consumer's stream
62+
// generation, invalidating any in-flight getOrCreateTrace() suspended at its
63+
// rehydrate await.
64+
traceConsumer.reset()
15065
// altimate_change end
151-
// Clear stale per-stream trace state before starting a new stream instance
152-
for (const [, trace] of sessionTraces) {
153-
void trace.endTrace().catch(() => {})
154-
}
155-
sessionTraces.clear()
156-
sessionUserMsgIds.clear()
15766

15867
const abort = new AbortController()
15968
eventStream.abort = abort
@@ -175,8 +84,9 @@ const startEventStream = (input: { directory: string; workspaceID?: string }) =>
17584
})
17685

17786
;(async () => {
178-
// Load tracing config once before processing events
179-
await loadTracingConfig()
87+
// altimate_change start — trace: load tracing config once before processing events
88+
await traceConsumer.loadConfig()
89+
// altimate_change end
18090
while (!signal.aborted) {
18191
const events = await Promise.resolve(
18292
sdk.event.subscribe(
@@ -194,135 +104,7 @@ const startEventStream = (input: { directory: string; workspaceID?: string }) =>
194104

195105
for await (const event of events.stream) {
196106
// altimate_change start — trace: feed events to per-session trace
197-
try {
198-
if (event.type === "message.updated") {
199-
const info = (event as any).properties?.info
200-
// Resolve sessionID: use info.sessionID directly, or fall back to
201-
// finding the session via info.parentID (assistant messages may only
202-
// carry the parent message ID, not the session ID).
203-
let resolvedSessionID = info?.sessionID as string | undefined
204-
if (!resolvedSessionID && info?.parentID) {
205-
for (const [sid, msgIds] of sessionUserMsgIds) {
206-
if (msgIds.has(info.parentID)) {
207-
resolvedSessionID = sid
208-
break
209-
}
210-
}
211-
}
212-
if (resolvedSessionID) {
213-
// Create trace eagerly on user message (arrives before part events)
214-
const trace =
215-
sessionTraces.get(resolvedSessionID) ??
216-
(info.role === "user" ? await getOrCreateTrace(resolvedSessionID) : null)
217-
if (info.role === "user") {
218-
if (info.id) {
219-
if (!sessionUserMsgIds.has(resolvedSessionID)) sessionUserMsgIds.set(resolvedSessionID, new Set())
220-
sessionUserMsgIds.get(resolvedSessionID)!.add(info.id)
221-
}
222-
if (trace) {
223-
const title = (info as any).summary?.title || (info as any).summary?.body
224-
if (title) trace.setTitle(String(title).slice(0, 80), String(title))
225-
}
226-
}
227-
if (info.role === "assistant") {
228-
const r = trace ?? (await getOrCreateTrace(resolvedSessionID))
229-
r?.enrichFromAssistant({
230-
modelID: info.modelID,
231-
providerID: info.providerID,
232-
agent: info.agent,
233-
variant: info.variant,
234-
})
235-
}
236-
}
237-
}
238-
// altimate_change end
239-
// altimate_change start — trace: part events
240-
if (event.type === "message.part.updated") {
241-
const part = (event as any).properties?.part
242-
if (part) {
243-
// Create trace on first event for this session (lazy creation)
244-
const trace = sessionTraces.get(part.sessionID) ?? (await getOrCreateTrace(part.sessionID))
245-
if (trace) {
246-
if (part.type === "step-start") trace.logStepStart(part)
247-
if (part.type === "step-finish") trace.logStepFinish(part)
248-
// altimate_change start — split the user-vs-assistant text routes.
249-
// User text parts arrive without `time.end` set (it's a meaningful
250-
// concept only for processing-end of assistant chunks), so the old
251-
// `&& part.time?.end` gate dropped the prompt entirely. We trust
252-
// `sessionUserMsgIds.has(messageID)` as the user-text signal and
253-
// call `setPrompt(text)` only — never `setTitle` — to avoid racing
254-
// the auto-generated title from `session.updated` (Path C).
255-
if (part.type === "text") {
256-
// altimate_change start — skip synthetic / ignored text parts.
257-
// `Session.createUserMessage` (prompt.ts) attaches many `synthetic: true`
258-
// text parts to the user message — MCP resource banners, decoded file
259-
// contents, retry/reminder text, plan-mode reminders, agent-handoff
260-
// tags. They all share the user's `messageID` so they would otherwise
261-
// pass the `sessionUserMsgIds` check below and override `metadata.prompt`
262-
// with the LAST synthetic blob (typically file content) and render one
263-
// fake "▶ You" bubble per synthetic part in the chat tab. The synthetic
264-
// and ignored flags exist precisely to mark non-authored content; this
265-
// is exactly the place to consult them. We skip silently rather than
266-
// `continue`-ing the event-loop iteration because the outer loop still
267-
// needs to forward the event downstream via `Rpc.emit`.
268-
const isAuthoredText = !part.synthetic && !part.ignored
269-
// altimate_change end
270-
if (
271-
isAuthoredText &&
272-
part.messageID &&
273-
sessionUserMsgIds.get(part.sessionID)?.has(part.messageID)
274-
) {
275-
const text = String(part.text || "")
276-
if (text) {
277-
trace.setPrompt(text)
278-
// altimate_change start — record each user message as a span
279-
// so the chat tab can render multi-turn conversations.
280-
// Without a span, the viewer can only display `metadata.prompt`
281-
// (singular) and every subsequent user message is silently
282-
// dropped from the conversation rendering.
283-
trace.logUserMessage(text)
284-
// altimate_change end
285-
}
286-
} else if (isAuthoredText && part.time?.end) {
287-
// Assistant response text (only counts when processing-end fires)
288-
trace.logText(part)
289-
}
290-
}
291-
// altimate_change end
292-
if (part.type === "tool" && (part.state?.status === "completed" || part.state?.status === "error")) {
293-
trace.logToolCall(part)
294-
}
295-
}
296-
}
297-
}
298-
// altimate_change end
299-
// altimate_change start — trace: session title capture and finalization
300-
// Capture session title from session.updated events
301-
if (event.type === "session.updated") {
302-
const info = (event as any).properties?.info
303-
if (info?.id && info?.title) {
304-
const trace = sessionTraces.get(info.id)
305-
if (trace) trace.setTitle(String(info.title))
306-
}
307-
}
308-
// altimate_change start — DO NOT finalize the trace on session.status=idle.
309-
// `idle` fires after every turn (busy → idle transition), not at session end.
310-
// Calling `endTrace` + `sessionTraces.delete` here treats each turn as the
311-
// end of the session: the next event for the same session in a later turn
312-
// hits a cache miss in getOrCreateTrace, constructs a fresh Trace.create()
313-
// with empty `this.spans`, and the immediate `snapshot()` clobbers the
314-
// rich on-disk `ses_<id>.json` with a single root-span file. Symptoms:
315-
// - waterfall view collapses to the system-prompt span after every turn
316-
// - "What was asked / No prompt recorded" because metadata.prompt was
317-
// captured on the destroyed instance, never on the replacement
318-
// Sessions in altimate-code are long-lived across many turns; the Trace
319-
// should live as long as the worker has the session in cache. Finalization
320-
// happens on `shutdown` (worker.ts:312) and on MAX_TRACES eviction
321-
// (worker.ts:87). No per-turn finalization is correct.
322-
// altimate_change end
323-
} catch {
324-
// Trace must never interrupt event forwarding
325-
}
107+
await traceConsumer.handleEvent(event)
326108
// altimate_change end
327109

328110
Rpc.emit("event", event as Event)
@@ -404,11 +186,7 @@ export const rpc = {
404186
Log.Default.info("worker shutting down")
405187
if (eventStream.abort) eventStream.abort.abort()
406188
// altimate_change start — trace: flush all active traces on shutdown
407-
for (const [sid, trace] of sessionTraces) {
408-
await trace.endTrace().catch(() => {})
409-
}
410-
sessionTraces.clear()
411-
sessionUserMsgIds.clear()
189+
await traceConsumer.flush()
412190
// altimate_change end
413191
await Instance.disposeAll()
414192
if (server) server.stop(true)

0 commit comments

Comments
 (0)