From 76f2754a1c8d5f175f0739b42e25cd6cb1680651 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 3 Jul 2026 00:16:19 +0800 Subject: [PATCH 1/2] fix(runtime): preserve stop semantics for synthetic terminals --- .../session-manager-terminal-ledger.test.ts | 52 ++++++++++++++ .../src/__tests__/session-manager.test.ts | 7 +- packages/runtime/src/agent-run.ts | 10 +-- packages/runtime/src/runtime-runner.ts | 9 --- packages/runtime/src/terminal-run-commit.ts | 68 ++++++++++++++++--- 5 files changed, 122 insertions(+), 24 deletions(-) diff --git a/packages/runtime/src/__tests__/session-manager-terminal-ledger.test.ts b/packages/runtime/src/__tests__/session-manager-terminal-ledger.test.ts index fdb0938238..8b5fde41ef 100644 --- a/packages/runtime/src/__tests__/session-manager-terminal-ledger.test.ts +++ b/packages/runtime/src/__tests__/session-manager-terminal-ledger.test.ts @@ -465,7 +465,59 @@ describe('SessionManager terminal ledger invariants', () => { const terminalEvents = (await runStore.readRuntimeEvents(session.id, run.runId)).filter(isTerminalRuntimeEvent); expect(terminalEvents).toHaveLength(1); expect(terminalEvents[0]?.status).toBe('failed'); + expect(terminalEvents[0]?.invocationId).toBe(run.runId); expect(terminalEvents[0]?.actions?.stateDelta?.failureClass).toBe('missing_terminal_event'); + expect(terminalEvents[0]?.actions?.stateDelta?.recovered).toBeUndefined(); + await new RuntimeReadModel({ runStore, runtimeEventStore: runStore }).getSessionView(session.id); + }); + + test('direct AgentRun stop synthesizes a cancelled terminal fact when no terminal event was recorded', async () => { + const store = new TinySessionStore(); + const runStore = new TinyAgentRunStore(); + const session = await store.create(makeInput()); + const backend = new ScriptBackend({ sessionId: session.id } as BackendFactoryContext, []); + const activeRuns = new Map(); + const turnToRunId = new Map(); + const run = new AgentRun({ + sessionId: session.id, + header: session, + userInput: { turnId: 'turn-1', text: 'hello' }, + store, + runStore, + runtimeEventStore: runStore, + newId: nextId(), + now: nextNow(41_250), + hooks: { + ensureActive: async () => ({ sessionId: session.id, backend, cachedHeader: session, activeRuns, turnToRunId }), + registerRun: (_active, activeRun) => { + activeRuns.set(activeRun.runId, activeRun); + turnToRunId.set(activeRun.turnId, activeRun.runId); + }, + unregisterRun: (_active, activeRun) => { + activeRuns.delete(activeRun.runId); + turnToRunId.delete(activeRun.turnId); + }, + updateHeader: (sessionId, patch) => store.updateHeader(sessionId, patch), + updateStatus: async () => {}, + appendTurnState: async () => {}, + }, + }); + + const begin = await run.begin(); + run.stop('stop_button'); + await run.finalize(); + + const header = await runStore.readRun(session.id, run.runId); + expect(header.status).toBe('cancelled'); + expect(header.failureClass).toBeUndefined(); + expect(header.abortSource).toBe('renderer.stop_button'); + const terminalEvents = (await runStore.readRuntimeEvents(session.id, run.runId)).filter(isTerminalRuntimeEvent); + expect(terminalEvents).toHaveLength(1); + expect(terminalEvents[0]?.status).toBe('aborted'); + expect(terminalEvents[0]?.invocationId).toBe(begin.initialRuntimeEvent.invocationId); + expect(terminalEvents[0]?.actions?.stateDelta?.abortSource).toBe('renderer.stop_button'); + expect(terminalEvents[0]?.actions?.stateDelta?.failureClass).toBeUndefined(); + expect(terminalEvents[0]?.actions?.stateDelta?.recovered).toBeUndefined(); await new RuntimeReadModel({ runStore, runtimeEventStore: runStore }).getSessionView(session.id); }); diff --git a/packages/runtime/src/__tests__/session-manager.test.ts b/packages/runtime/src/__tests__/session-manager.test.ts index 07c8b88762..c2e557fef8 100644 --- a/packages/runtime/src/__tests__/session-manager.test.ts +++ b/packages/runtime/src/__tests__/session-manager.test.ts @@ -526,7 +526,9 @@ describe('SessionManager permission mode updates', () => { expect(runtimeEvents[2]?.id).toBe(terminalEvents[0]?.id); expect(terminalEvents).toHaveLength(1); expect(terminalEvents[0]?.status).toBe('failed'); + expect(terminalEvents[0]?.invocationId).toBe(run.runId); expect(terminalEvents[0]?.actions?.stateDelta?.failureClass).toBe('missing_terminal_event'); + expect(terminalEvents[0]?.actions?.stateDelta?.recovered).toBeUndefined(); const view = await new RuntimeReadModel({ runStore, @@ -618,8 +620,9 @@ describe('SessionManager permission mode updates', () => { expect(run.failureClass).toBe('missing_terminal_event'); expect(terminalEvents).toHaveLength(1); expect(terminalEvents[0]?.status).toBe('failed'); - expect(terminalEvents[0]?.actions?.stateDelta?.recovered).toBe(true); - expect(terminalEvents[0]?.actions?.stateDelta?.recoveryReason).toBe('missing_terminal_event'); + expect(terminalEvents[0]?.invocationId).toBe(run.runId); + expect(terminalEvents[0]?.actions?.stateDelta?.recovered).toBeUndefined(); + expect(terminalEvents[0]?.actions?.stateDelta?.recoveryReason).toBeUndefined(); expect(terminalEvents[0]?.actions?.stateDelta?.failureClass).toBe('missing_terminal_event'); const view = await new RuntimeReadModel({ diff --git a/packages/runtime/src/agent-run.ts b/packages/runtime/src/agent-run.ts index 136f10ced8..6e489ea6cb 100644 --- a/packages/runtime/src/agent-run.ts +++ b/packages/runtime/src/agent-run.ts @@ -786,6 +786,7 @@ export class AgentRun { const runStore = this.input.runStore; const runtimeEventStore = this.input.runtimeEventStore; if (!runStore || !this.runStoreAvailable || !runtimeEventStore || !this.runtimeEventStoreAvailable) return; + const fallbackStatus = this.stopped || finalStatus?.status === 'aborted' ? 'cancelled' : 'failed'; const fallbackFailureClass = 'missing_terminal_event'; const fallbackFailureMessage = this.failureMessage ?? 'run finalized without a terminal RuntimeEvent'; try { @@ -808,12 +809,13 @@ export class AgentRun { : {}), ...(this.failureMessage ? { failureMessage: this.failureMessage } : {}), ...(this.abortSource ? { abortSource: this.abortSource } : {}), - fallbackFailureClass, - fallbackRecoveryReason: fallbackFailureClass, - fallbackFailureMessage, + fallbackStatus, + fallbackInvocationId: this.runId, + ...(fallbackStatus === 'failed' ? { fallbackFailureClass, fallbackFailureMessage } : {}), + ...(fallbackStatus === 'cancelled' ? { fallbackAbortSource: this.abortSource ?? 'user_stop' } : {}), allowHeaderCommitFailure: true, }); - if (result.createdTerminalEvent) { + if (result.createdTerminalEvent && result.status === 'failed') { this.failureClass = result.failureClass ?? fallbackFailureClass; this.failureMessage = fallbackFailureMessage; } diff --git a/packages/runtime/src/runtime-runner.ts b/packages/runtime/src/runtime-runner.ts index eb89a07451..962c9ad75a 100644 --- a/packages/runtime/src/runtime-runner.ts +++ b/packages/runtime/src/runtime-runner.ts @@ -95,12 +95,6 @@ export interface RuntimeRunnerDeps { gate?: RuntimeGate; /** Injectable id/time providers. Defaults to crypto.randomUUID / Date.now. */ providers?: InvocationProviders; - /** - * Called after the initial user RuntimeEvent is built and before the flow is - * dispatched. RuntimeRunner still does not own storage; orchestration layers - * can use this to keep durable ledgers ahead of renderer-visible events. - */ - onInitialRuntimeEvent?: (event: RuntimeEvent) => Promise | void; /** * Whether to stop collecting at the first terminal RuntimeEvent. Defaults * to true for standalone runner callers; production bridges can set false @@ -129,14 +123,12 @@ export class RuntimeRunner { private readonly flow: RunnableAgentFlow; private readonly gate: RuntimeGate | undefined; private readonly providers: InvocationProviders; - private readonly onInitialRuntimeEvent: RuntimeRunnerDeps['onInitialRuntimeEvent']; private readonly stopOnTerminal: boolean; constructor(deps: RuntimeRunnerDeps) { this.flow = deps.flow; this.gate = deps.gate; this.providers = deps.providers ?? createDefaultInvocationProviders(); - this.onInitialRuntimeEvent = deps.onInitialRuntimeEvent; this.stopOnTerminal = deps.stopOnTerminal ?? true; } @@ -229,7 +221,6 @@ export class RuntimeRunner { text: request.text, ...(request.attachments !== undefined ? { attachments: request.attachments } : {}), }); - await this.onInitialRuntimeEvent?.(userEvent); events.push(userEvent); const flowInput = buildFlowInput(request); diff --git a/packages/runtime/src/terminal-run-commit.ts b/packages/runtime/src/terminal-run-commit.ts index fdb4914739..7872572825 100644 --- a/packages/runtime/src/terminal-run-commit.ts +++ b/packages/runtime/src/terminal-run-commit.ts @@ -127,10 +127,11 @@ export interface CommitOrCreateTerminalRunFactInput extends Omit< > { terminalEvent?: RuntimeEvent; allowHeaderCommitFailure?: boolean; - fallbackFailureClass: string; + fallbackStatus: TerminalAgentRunStatus; + fallbackInvocationId: string; + fallbackFailureClass?: string; fallbackFailureMessage?: string; - fallbackRecoveryReason?: string; - fallbackDiagnostic?: Record; + fallbackAbortSource?: string; } export interface CommitOrCreateTerminalRunFactResult { @@ -146,19 +147,21 @@ export async function commitOrCreateTerminalRunFact( input: CommitOrCreateTerminalRunFactInput, ): Promise { const createdTerminalEvent = !input.terminalEvent; - const terminalEvent = input.terminalEvent ?? buildRecoveredTerminalRuntimeEvent({ + const terminalEvent = input.terminalEvent ?? buildLiveSyntheticTerminalRuntimeEvent({ id: input.newId(), + invocationId: input.fallbackInvocationId, run: { sessionId: input.sessionId, runId: input.runId, turnId: input.turnId, }, - status: 'failed', + status: input.fallbackStatus, ts: input.ts, - failureClass: input.fallbackFailureClass, - recoveryReason: input.fallbackRecoveryReason ?? input.fallbackFailureClass, - ...(input.fallbackDiagnostic ? { diagnostic: input.fallbackDiagnostic } : {}), - message: input.fallbackFailureMessage ?? input.failureMessage ?? input.fallbackFailureClass, + ...(input.fallbackFailureClass ? { failureClass: input.fallbackFailureClass } : {}), + ...(input.fallbackAbortSource ?? input.abortSource ? { abortSource: input.fallbackAbortSource ?? input.abortSource } : {}), + ...(input.fallbackFailureMessage ?? input.failureMessage + ? { message: input.fallbackFailureMessage ?? input.failureMessage } + : {}), }); const status = terminalRunStatusFromRuntimeEvent(terminalEvent); if (!status) { @@ -210,6 +213,53 @@ export async function commitOrCreateTerminalRunFact( }; } +export interface BuildLiveSyntheticTerminalRuntimeEventInput { + id: string; + invocationId: string; + run: Pick; + status: TerminalAgentRunStatus; + ts: number; + failureClass?: string; + abortSource?: string; + message?: string; +} + +export function buildLiveSyntheticTerminalRuntimeEvent( + input: BuildLiveSyntheticTerminalRuntimeEventInput, +): RuntimeEvent { + const failureClass = input.status === 'failed' ? input.failureClass ?? 'unknown' : undefined; + const abortSource = input.status === 'cancelled' ? input.abortSource ?? 'user_stop' : undefined; + return { + id: input.id, + invocationId: input.invocationId, + runId: input.run.runId, + sessionId: input.run.sessionId, + turnId: input.run.turnId, + ts: input.ts, + partial: false, + role: 'system', + author: 'system', + status: input.status === 'cancelled' ? 'aborted' : input.status, + ...(failureClass + ? { + content: { + kind: 'error', + code: failureClass, + reason: failureClass, + message: input.message ?? failureClass, + }, + } + : {}), + actions: { + endInvocation: true, + stateDelta: { + ...(failureClass ? { failureClass } : {}), + ...(abortSource ? { abortSource } : {}), + }, + }, + }; +} + export interface BuildRecoveredTerminalRuntimeEventInput { id: string; run: Pick; From 4d32997819435a904d2a08e1548d41b6ee412c21 Mon Sep 17 00:00:00 2001 From: AstroHan Date: Fri, 3 Jul 2026 00:56:08 +0800 Subject: [PATCH 2/2] fix(runtime): keep synthetic cancel metadata consistent --- .../session-manager-terminal-ledger.test.ts | 64 ++++++++++++++++++- packages/runtime/src/agent-run.ts | 5 +- packages/runtime/src/terminal-run-commit.ts | 61 +++++++----------- 3 files changed, 89 insertions(+), 41 deletions(-) diff --git a/packages/runtime/src/__tests__/session-manager-terminal-ledger.test.ts b/packages/runtime/src/__tests__/session-manager-terminal-ledger.test.ts index 8b5fde41ef..970e537b13 100644 --- a/packages/runtime/src/__tests__/session-manager-terminal-ledger.test.ts +++ b/packages/runtime/src/__tests__/session-manager-terminal-ledger.test.ts @@ -25,7 +25,13 @@ import { type SessionStore, } from '../session-manager.js'; import type { AgentBackend } from '../ai-sdk-backend.js'; -import { classifyTerminalRuntimeLedger, commitTerminalRunWithRuntimeFact } from '../terminal-run-commit.js'; +import { + buildRecoveredTerminalRuntimeEvent, + buildSyntheticTerminalRuntimeEvent, + classifyTerminalRuntimeLedger, + commitOrCreateTerminalRunFact, + commitTerminalRunWithRuntimeFact, +} from '../terminal-run-commit.js'; import { RuntimeReadModel } from '../runtime-read-model.js'; import { RuntimeKernel } from '../runtime-kernel.js'; @@ -185,6 +191,62 @@ describe('SessionManager terminal ledger invariants', () => { expect((await runStore.readRun(run.sessionId, run.runId)).status).toBe('running'); }); + test('synthetic cancelled terminal commits the fallback abortSource to the run header', async () => { + const runStore = new TinyAgentRunStore(); + const run = makeRunHeader({ status: 'running' }); + await runStore.createRun(run); + + await commitOrCreateTerminalRunFact({ + runStore, + runtimeEventStore: runStore, + newId: nextId(), + sessionId: run.sessionId, + runId: run.runId, + turnId: run.turnId, + ts: 3, + fallbackStatus: 'cancelled', + fallbackInvocationId: run.runId, + }); + + const header = await runStore.readRun(run.sessionId, run.runId); + expect(header.status).toBe('cancelled'); + expect(header.abortSource).toBe('user_stop'); + const terminalEvents = (await runStore.readRuntimeEvents(run.sessionId, run.runId)).filter(isTerminalRuntimeEvent); + expect(terminalEvents).toHaveLength(1); + expect(terminalEvents[0]?.status).toBe('aborted'); + expect(terminalEvents[0]?.actions?.stateDelta?.abortSource).toBe('user_stop'); + expect(terminalEvents[0]?.actions?.stateDelta?.recovered).toBeUndefined(); + }); + + test('synthetic terminal builder keeps live and recovered metadata distinct', () => { + const run = makeRunHeader({ status: 'running' }); + const live = buildSyntheticTerminalRuntimeEvent({ + id: 'live-terminal', + invocationId: run.runId, + run, + status: 'failed', + ts: 3, + failureClass: 'missing_terminal_event', + }); + expect(live.invocationId).toBe(run.runId); + expect(live.actions?.stateDelta?.failureClass).toBe('missing_terminal_event'); + expect(live.actions?.stateDelta?.recovered).toBeUndefined(); + expect(live.actions?.stateDelta?.recoveryReason).toBeUndefined(); + + const recovered = buildRecoveredTerminalRuntimeEvent({ + id: 'recovered-terminal', + run, + status: 'failed', + ts: 4, + failureClass: 'missing_terminal_event', + recoveryReason: 'run_interrupted', + }); + expect(recovered.invocationId).toBe(`recovery-${run.runId}`); + expect(recovered.actions?.stateDelta?.failureClass).toBe('missing_terminal_event'); + expect(recovered.actions?.stateDelta?.recovered).toBe(true); + expect(recovered.actions?.stateDelta?.recoveryReason).toBe('run_interrupted'); + }); + test('terminal ledger classification rejects multiple terminal RuntimeEvent signals', () => { const run = makeRunHeader({ status: 'running' }); diff --git a/packages/runtime/src/agent-run.ts b/packages/runtime/src/agent-run.ts index 6e489ea6cb..112b252c37 100644 --- a/packages/runtime/src/agent-run.ts +++ b/packages/runtime/src/agent-run.ts @@ -808,11 +808,12 @@ export class AgentRun { ? { failureClass: this.failureClass ?? finalStatus?.blockedReason } : {}), ...(this.failureMessage ? { failureMessage: this.failureMessage } : {}), - ...(this.abortSource ? { abortSource: this.abortSource } : {}), + ...(this.abortSource || fallbackStatus === 'cancelled' + ? { abortSource: this.abortSource ?? 'user_stop' } + : {}), fallbackStatus, fallbackInvocationId: this.runId, ...(fallbackStatus === 'failed' ? { fallbackFailureClass, fallbackFailureMessage } : {}), - ...(fallbackStatus === 'cancelled' ? { fallbackAbortSource: this.abortSource ?? 'user_stop' } : {}), allowHeaderCommitFailure: true, }); if (result.createdTerminalEvent && result.status === 'failed') { diff --git a/packages/runtime/src/terminal-run-commit.ts b/packages/runtime/src/terminal-run-commit.ts index 7872572825..c473093b07 100644 --- a/packages/runtime/src/terminal-run-commit.ts +++ b/packages/runtime/src/terminal-run-commit.ts @@ -131,7 +131,6 @@ export interface CommitOrCreateTerminalRunFactInput extends Omit< fallbackInvocationId: string; fallbackFailureClass?: string; fallbackFailureMessage?: string; - fallbackAbortSource?: string; } export interface CommitOrCreateTerminalRunFactResult { @@ -147,7 +146,10 @@ export async function commitOrCreateTerminalRunFact( input: CommitOrCreateTerminalRunFactInput, ): Promise { const createdTerminalEvent = !input.terminalEvent; - const terminalEvent = input.terminalEvent ?? buildLiveSyntheticTerminalRuntimeEvent({ + const effectiveAbortSource = input.fallbackStatus === 'cancelled' + ? input.abortSource ?? 'user_stop' + : input.abortSource; + const terminalEvent = input.terminalEvent ?? buildSyntheticTerminalRuntimeEvent({ id: input.newId(), invocationId: input.fallbackInvocationId, run: { @@ -158,7 +160,7 @@ export async function commitOrCreateTerminalRunFact( status: input.fallbackStatus, ts: input.ts, ...(input.fallbackFailureClass ? { failureClass: input.fallbackFailureClass } : {}), - ...(input.fallbackAbortSource ?? input.abortSource ? { abortSource: input.fallbackAbortSource ?? input.abortSource } : {}), + ...(effectiveAbortSource ? { abortSource: effectiveAbortSource } : {}), ...(input.fallbackFailureMessage ?? input.failureMessage ? { message: input.fallbackFailureMessage ?? input.failureMessage } : {}), @@ -196,6 +198,7 @@ export async function commitOrCreateTerminalRunFact( terminalEvent, status, ...(failureClass ? { failureClass } : {}), + ...(effectiveAbortSource ? { abortSource: effectiveAbortSource } : {}), terminalEventAlreadyPersisted, }); headerCommitted = true; @@ -213,7 +216,7 @@ export async function commitOrCreateTerminalRunFact( }; } -export interface BuildLiveSyntheticTerminalRuntimeEventInput { +export interface BuildSyntheticTerminalRuntimeEventInput { id: string; invocationId: string; run: Pick; @@ -221,14 +224,16 @@ export interface BuildLiveSyntheticTerminalRuntimeEventInput { ts: number; failureClass?: string; abortSource?: string; + recoveryReason?: string; + diagnostic?: Record; message?: string; } -export function buildLiveSyntheticTerminalRuntimeEvent( - input: BuildLiveSyntheticTerminalRuntimeEventInput, +export function buildSyntheticTerminalRuntimeEvent( + input: BuildSyntheticTerminalRuntimeEventInput, ): RuntimeEvent { const failureClass = input.status === 'failed' ? input.failureClass ?? 'unknown' : undefined; - const abortSource = input.status === 'cancelled' ? input.abortSource ?? 'user_stop' : undefined; + const abortSource = input.status === 'cancelled' ? input.abortSource : undefined; return { id: input.id, invocationId: input.invocationId, @@ -253,6 +258,8 @@ export function buildLiveSyntheticTerminalRuntimeEvent( actions: { endInvocation: true, stateDelta: { + ...(input.recoveryReason ? { recovered: true, recoveryReason: input.recoveryReason } : {}), + ...(input.diagnostic ?? {}), ...(failureClass ? { failureClass } : {}), ...(abortSource ? { abortSource } : {}), }, @@ -276,40 +283,18 @@ export interface BuildRecoveredTerminalRuntimeEventInput { export function buildRecoveredTerminalRuntimeEvent( input: BuildRecoveredTerminalRuntimeEventInput, ): RuntimeEvent { - const failureClass = input.status === 'failed' ? input.failureClass ?? 'unknown' : undefined; - const abortSource = input.status === 'cancelled' ? input.abortSource ?? 'unknown' : undefined; - return { + return buildSyntheticTerminalRuntimeEvent({ id: input.id, invocationId: input.invocationId ?? `recovery-${input.run.runId}`, - runId: input.run.runId, - sessionId: input.run.sessionId, - turnId: input.run.turnId, + run: input.run, ts: input.ts, - partial: false, - role: 'system', - author: 'system', - status: input.status === 'cancelled' ? 'aborted' : input.status, - ...(failureClass - ? { - content: { - kind: 'error', - code: failureClass, - reason: failureClass, - message: input.message ?? failureClass, - }, - } - : {}), - actions: { - endInvocation: true, - stateDelta: { - recovered: true, - recoveryReason: input.recoveryReason, - ...(input.diagnostic ?? {}), - ...(failureClass ? { failureClass } : {}), - ...(abortSource ? { abortSource } : {}), - }, - }, - }; + status: input.status, + ...(input.failureClass ? { failureClass: input.failureClass } : {}), + ...(input.status === 'cancelled' ? { abortSource: input.abortSource ?? 'unknown' } : {}), + recoveryReason: input.recoveryReason, + ...(input.diagnostic ? { diagnostic: input.diagnostic } : {}), + ...(input.message ? { message: input.message } : {}), + }); } export function hasTerminalAgentRunEvent(events: readonly Pick[]): boolean {