simplify(session): fold trace-only usage/error events into load-bearing events

The session event vocabulary carried two standalone trace-only events that
were not load-bearing as separate records. Fold their facts into nearby
load-bearing events and delete the standalone variants.

- Token usage now rides on `assistant/message` as an optional `usage` field —
  the assembled model output and its accounting travel together. The loop folds
  `assembler.usage` onto the append instead of emitting a separate `usage`
  event.
- The max-tokens path is the no-data-loss host: a step cut off with usage but
  EMPTY content (e.g. only a dropped tool call) previously emitted a standalone
  `usage`; it now records an empty-content `assistant/message { content: [],
  usage }`. `deriveMessages()` skips empty-content assistant messages, so the
  usage host never injects a spurious content-less assistant turn into the
  provider transcript. A step with neither content nor usage appends nothing.
- An operational error's step number now rides on `turn/end.reason` for
  `kind: 'error'` (`{ kind: 'error', step, message, code? }`) — the durable
  turn outcome ACP and resume already consume. `failTurn` sets the reason
  directly (no separate session `error` event). `agent/error` + logging are
  unchanged for live diagnostics.
- No format-version bump: pre-release, no persisted data, so per the format
  policy there is nothing to migrate or reject (the RFC's "refresh the format
  version" criterion over-reached). `version` stays 1.
- ACP fixtures + goldens re-recorded (keyless replay): dropped standalone
  usage/error lines, usage folded onto assistant/message, error step on
  turn/end.reason.

RFC moved proposed -> implemented with an implementation note recording the two
scope refinements.
This commit is contained in:
Tianyi Cui
2026-06-21 10:00:06 +08:00
parent c6ed980d6f
commit 2be60b9a22
30 changed files with 480 additions and 470 deletions

View File

@@ -154,7 +154,7 @@ export interface LoopHandle {
* stream ctx.llm.stream(req) ⟵ waterfall llm/stream (raw chunks)
* session('assistant/chunk'); emit agent/stream-chunk
* msg = waterfall agent/step-result ⟵ BEFORE the log append, so the
* session('assistant/message','usage') session records what actually ran
* session('assistant/message' {content, usage?}) session records what actually ran
* each tool-call in msg (sequential, abort-checked):
* session('tool/call'); ctx.tools.execute() ⟵ waterfall tools/execute
* session('tool/result')
@@ -313,41 +313,24 @@ async function runTurn(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle,
return false
}
// Record a step/turn failure exactly once: append the single `error` event
// (only while the turn is still open — see below), set the error reason, and
// emit agent/error (contained — trap: a throwing agent/error listener must not
// re-escape and strand the turn). Disposal and abort set `reason` directly
// without calling this (no `error` event for those — they are not failures).
// Record a step/turn failure exactly once: set the error reason (carrying the
// failing `step` — the durable failure lives entirely on turn/end.reason, there
// is no separate session error event) and emit agent/error (contained — trap: a
// throwing agent/error listener must not re-escape and strand the turn).
// Disposal and abort set `reason` directly without calling this (they are not
// failures).
const failTurn = (err: CodedError): void => {
if (errorReported) return
errorReported = true
// Only append the session `error` INSIDE the turn (before turn/end). If the
// turn has already ended — the only way here is a throwing agent/turn-end
// listener after closeTurn(true) already appended turn/end — appending now
// would land the error AFTER the last turn/end, where the persistence
// backend treats it as a crash tail and drops it on resume (the turn-enclosure RFC). In
// that case report via agent/error + the logger only; the turn is balanced.
if (!turnEnded) {
// Set `reason` BEFORE the append: Session.append pushes the error event
// before notifying session/event listeners, so a throwing listener would
// otherwise leave `reason` unset (and closeTurn would record the wrong
// reason / the outer catch would skip closeTurn). The append is contained
// — the error event is already in the log either way; a throwing listener
// must not abort finalization.
reason = { kind: 'error', ...errorData(err) }
try {
session.append('error', { turn, step, ...errorData(err) })
} catch (appendError: unknown) {
ctx.logger.warn(`agent "${agent.id}": session/event listener threw on the error event at turn ${turn}: ${toError(appendError).message}`)
}
} else {
ctx.logger.warn(`agent "${agent.id}": agent/turn-end listener threw after turn ${turn} closed: ${err.message}`)
}
// Set `reason` here so the durable failure is captured before closeTurn
// appends turn/end. The step number rides along so the operational error's
// location survives in the durable log.
reason = { kind: 'error', step, ...errorData(err) }
try {
ctx.emit('agent/error', agent, turn, step, err)
} catch {
// contained: the error is already logged; a throwing agent/error
// listener must not prevent the turn from closing.
// contained: the error is already captured on `reason`; a throwing
// agent/error listener must not prevent the turn from closing.
}
}
@@ -608,11 +591,14 @@ async function runStep(
if (assembler.finish.kind === 'max-tokens') {
let message: Message = withoutToolCalls(assembler.message())
message = withoutToolCalls(await ctx.waterfall('agent/step-result', agent, turn, step, message, () => Promise.resolve(message)))
if (message.content.length > 0) {
session.append('assistant/message', { turn, step, content: message.content })
}
if (assembler.usage) {
session.append('usage', { turn, step, usage: assembler.usage })
// Fire the assistant/message when there is content OR usage: a max-tokens
// step can be cut off with empty content but still carry token accounting,
// and assistant/message is the only host for usage (there is no standalone
// usage event). An empty-content assistant/message is skipped by
// deriveMessages(), so hosting usage on it never injects a spurious assistant
// turn into derived history.
if (message.content.length > 0 || assembler.usage) {
session.append('assistant/message', { turn, step, content: message.content, ...(assembler.usage ? { usage: assembler.usage } : {}) })
}
return { hadToolCalls: false, finish: assembler.finish }
}
@@ -623,10 +609,7 @@ async function runStep(
let message: Message = assembler.message()
message = await ctx.waterfall('agent/step-result', agent, turn, step, message, () => Promise.resolve(message))
session.append('assistant/message', { turn, step, content: message.content })
if (assembler.usage) {
session.append('usage', { turn, step, usage: assembler.usage })
}
session.append('assistant/message', { turn, step, content: message.content, ...(assembler.usage ? { usage: assembler.usage } : {}) })
// --- Tool execution (sequential; parallel execution is a TODO) ---
// ToolRegistry.execute converts tool failures (including aborts) into