From 19f90dff277820b23482b4d7997dfd8a5f67de31 Mon Sep 17 00:00:00 2001 From: "haozhe.yang" Date: Wed, 8 Jul 2026 15:00:27 +0800 Subject: [PATCH] feat(agent-core): track agent live phase in a single runtime model field - add an Agent-scope runtime domain that holds the whole live phase as one discriminated-union field (idle, running, streaming, tool_call, retrying, awaiting_approval, interrupted, ended) - drive it from existing turn, step, delta, tool, retry, interrupt and approval events, edge-triggered with reference-equality dedup so delta bursts do not flood the wire log - carry the phase on the existing agent.status.updated channel and add the AgentPhase type/schema to the protocol (backward compatible) --- .changeset/agent-runtime-phase.md | 6 + .../src/agent/runtime/runtime.ts | 86 +++++++ .../src/agent/runtime/runtimeOps.ts | 85 +++++++ .../src/agent/runtime/runtimeService.ts | 218 +++++++++++++++++ .../agent-core-v2/src/agent/usage/usageOps.ts | 2 + packages/agent-core-v2/src/index.ts | 3 + .../test/runtime/runtime.test.ts | 228 ++++++++++++++++++ packages/protocol/src/events.ts | 129 ++++++++++ 8 files changed, 757 insertions(+) create mode 100644 .changeset/agent-runtime-phase.md create mode 100644 packages/agent-core-v2/src/agent/runtime/runtime.ts create mode 100644 packages/agent-core-v2/src/agent/runtime/runtimeOps.ts create mode 100644 packages/agent-core-v2/src/agent/runtime/runtimeService.ts create mode 100644 packages/agent-core-v2/test/runtime/runtime.test.ts diff --git a/.changeset/agent-runtime-phase.md b/.changeset/agent-runtime-phase.md new file mode 100644 index 000000000..7c9b360da --- /dev/null +++ b/.changeset/agent-runtime-phase.md @@ -0,0 +1,6 @@ +--- +"@moonshot-ai/agent-core-v2": minor +"@moonshot-ai/protocol": minor +--- + +Track the agent's live phase (idle, running, streaming, tool call, retrying, awaiting approval, interrupted, ended) as a single model field driven by the existing turn events, and carry it on the status update channel for downstream consumers. diff --git a/packages/agent-core-v2/src/agent/runtime/runtime.ts b/packages/agent-core-v2/src/agent/runtime/runtime.ts new file mode 100644 index 000000000..fd661a6a1 --- /dev/null +++ b/packages/agent-core-v2/src/agent/runtime/runtime.ts @@ -0,0 +1,86 @@ +/** + * `runtime` domain (L5) — Agent-scope live phase contract. + * + * Defines the public contract of the agent's whole live phase: the `AgentPhase` + * discriminated union (each variant carries its own ancillary fields) and the + * `IAgentRuntimeService` used to read the current phase via `phase()`. The + * phase is the agent-level, fine-grained counterpart of the session-level + * `sessionActivity` status: it splits `running` into waiting / streaming / + * tool_call / retrying and adds `interrupted` / `ended`. Agent-scoped — one + * instance per agent. + */ + +import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation'; +import type { TurnEndReason } from '@moonshot-ai/protocol'; + +export type AgentPhase = + | { readonly kind: 'idle' } + | { + readonly kind: 'running'; + readonly turnId: number; + readonly step: number; + readonly stepId: string; + readonly since: number; + } + | { + readonly kind: 'streaming'; + readonly turnId: number; + readonly step: number; + readonly stepId: string; + readonly stream: 'assistant' | 'thinking' | 'tool_call'; + readonly toolCallId?: string; + readonly toolName?: string; + readonly since: number; + } + | { + readonly kind: 'tool_call'; + readonly turnId: number; + readonly step: number; + readonly toolCallId: string; + readonly name: string; + readonly since: number; + } + | { + readonly kind: 'retrying'; + readonly turnId: number; + readonly step: number; + readonly stepId: string; + readonly failedAttempt: number; + readonly nextAttempt: number; + readonly maxAttempts: number; + readonly delayMs: number; + readonly errorName?: string; + readonly statusCode?: number; + readonly since: number; + } + | { + readonly kind: 'awaiting_approval'; + readonly turnId: number; + readonly step?: number; + readonly approval: unknown; + readonly since: number; + } + | { + readonly kind: 'interrupted'; + readonly turnId: number; + readonly step?: number; + readonly reason: 'aborted' | 'max_steps' | 'error'; + readonly message?: string; + readonly at: number; + } + | { + readonly kind: 'ended'; + readonly turnId: number; + readonly reason: TurnEndReason; + readonly durationMs?: number; + readonly at: number; + }; + +export interface IAgentRuntimeService { + readonly _serviceBrand: undefined; + + phase(): AgentPhase; +} + +export const IAgentRuntimeService: ServiceIdentifier = + createDecorator('agentRuntimeService'); diff --git a/packages/agent-core-v2/src/agent/runtime/runtimeOps.ts b/packages/agent-core-v2/src/agent/runtime/runtimeOps.ts new file mode 100644 index 000000000..a7ce3473b --- /dev/null +++ b/packages/agent-core-v2/src/agent/runtime/runtimeOps.ts @@ -0,0 +1,85 @@ +/** + * `runtime` domain (L5) — wire Model (`RuntimeModel`) and the `runtime.set_phase` + * Op (`setRuntimePhase`) that holds the agent's whole live phase. + * + * Declares the phase as a single-field wire Model (`{ phase }`, initial + * `{ kind: 'idle' }`) plus one Op whose `apply` is a pure, edge-triggered + * replacement: it returns the SAME reference when the incoming phase is + * unchanged under `phaseEqual` (which ignores `since` / `at` timestamps), so the + * wire's reference-equality gate stays quiet and high-frequency deltas do not + * flood the append log — only genuine phase transitions persist. The + * `agent.status.updated` `phase` slice is derived from the Op's `toEvent` + * (published on `dispatch`, never on `replay`). Consumed by the Agent-scope + * `runtimeService`. + */ + +import { defineModel } from '#/wire/model'; +import { defineOp } from '#/wire/op'; + +import type { AgentPhase } from './runtime'; + +export interface RuntimeModelState { + readonly phase: AgentPhase; +} + +export const RuntimeModel = defineModel('runtime', () => ({ + phase: { kind: 'idle' }, +})); + +export const setRuntimePhase = defineOp(RuntimeModel, 'runtime.set_phase', { + apply: (s, p: { phase: AgentPhase }): RuntimeModelState => + phaseEqual(s.phase, p.phase) ? s : { phase: p.phase }, + toEvent: (p) => ({ type: 'agent.status.updated' as const, phase: p.phase }), +}); + +/** + * Structural equality for phase transitions, ignoring the `since` / `at` + * timestamps so that re-entering the same logical phase (e.g. a burst of + * same-stream deltas) is treated as a no-op. + */ +export function phaseEqual(a: AgentPhase, b: AgentPhase): boolean { + if (a.kind !== b.kind) return false; + switch (a.kind) { + case 'idle': + return true; + case 'running': { + const c = b as typeof a; + return a.turnId === c.turnId && a.step === c.step && a.stepId === c.stepId; + } + case 'streaming': { + const c = b as typeof a; + return ( + a.turnId === c.turnId && + a.step === c.step && + a.stepId === c.stepId && + a.stream === c.stream && + a.toolCallId === c.toolCallId + ); + } + case 'tool_call': { + const c = b as typeof a; + return a.turnId === c.turnId && a.toolCallId === c.toolCallId; + } + case 'retrying': { + const c = b as typeof a; + return ( + a.turnId === c.turnId && + a.step === c.step && + a.failedAttempt === c.failedAttempt && + a.nextAttempt === c.nextAttempt + ); + } + case 'awaiting_approval': { + const c = b as typeof a; + return a.turnId === c.turnId; + } + case 'interrupted': { + const c = b as typeof a; + return a.turnId === c.turnId && a.reason === c.reason; + } + case 'ended': { + const c = b as typeof a; + return a.turnId === c.turnId && a.reason === c.reason; + } + } +} diff --git a/packages/agent-core-v2/src/agent/runtime/runtimeService.ts b/packages/agent-core-v2/src/agent/runtime/runtimeService.ts new file mode 100644 index 000000000..aa3247173 --- /dev/null +++ b/packages/agent-core-v2/src/agent/runtime/runtimeService.ts @@ -0,0 +1,218 @@ +/** + * `runtime` domain (L5) — `IAgentRuntimeService` implementation. + * + * Derives the agent's live phase from the existing `IEventBus` facts and folds + * it into the `wire` `RuntimeModel` (mutated only through the + * `runtime.set_phase` Op, read through `wire.getModel`). Subscriptions are + * edge-triggered: a handler builds the candidate phase and `setPhase` only + * dispatches when `phaseEqual` says it changed, so high-frequency delta streams + * collapse into a single `streaming` record. The current phase and the + * approval-resume target are kept as live-only fields (never in the Model) so + * `wire.replay` stays silent and resumes into `idle`. Bound at Agent scope. + */ + +import { Disposable } from '#/_base/di/lifecycle'; +import { InstantiationType } from '#/_base/di/extensions'; +import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; +import { IEventBus } from '#/app/event/eventBus'; +import type { PermissionApprovalRequestContext } from '#/agent/permissionGate/permissionGateService'; +import type { TurnEndReason } from '@moonshot-ai/protocol'; +import { IAgentWireService } from '#/wire/tokens'; +import type { IWireService } from '#/wire/wireService'; + +import { type AgentPhase, IAgentRuntimeService } from './runtime'; +import { phaseEqual, RuntimeModel, setRuntimePhase } from './runtimeOps'; + +interface TurnCursor { + readonly turnId: number; + readonly step: number; + readonly stepId: string; +} + +export class AgentRuntimeService extends Disposable implements IAgentRuntimeService { + declare readonly _serviceBrand: undefined; + + private cursor: TurnCursor = { turnId: -1, step: 0, stepId: '' }; + private current: AgentPhase = { kind: 'idle' }; + private priorForApproval: AgentPhase | undefined; + + constructor( + @IAgentWireService private readonly wire: IWireService, + @IEventBus private readonly eventBus: IEventBus, + ) { + super(); + this._register(this.eventBus.subscribe('turn.started', (e) => this.onTurnStarted(e.turnId))); + this._register( + this.eventBus.subscribe('turn.step.started', (e) => + this.onStepStarted(e.turnId, e.step, e.stepId ?? ''), + ), + ); + this._register( + this.eventBus.subscribe('assistant.delta', () => this.onDelta('assistant')), + ); + this._register( + this.eventBus.subscribe('thinking.delta', () => this.onDelta('thinking')), + ); + this._register( + this.eventBus.subscribe('tool.call.delta', (e) => + this.onToolCallDelta(e.toolCallId, e.name), + ), + ); + this._register( + this.eventBus.subscribe('tool.call.started', (e) => + this.onToolCallStarted(e.toolCallId, e.name), + ), + ); + this._register(this.eventBus.subscribe('tool.result', () => this.onToolResult())); + this._register( + this.eventBus.subscribe('turn.step.retrying', (e) => + this.setPhase({ + kind: 'retrying', + turnId: e.turnId, + step: e.step, + stepId: e.stepId ?? '', + failedAttempt: e.failedAttempt, + nextAttempt: e.nextAttempt, + maxAttempts: e.maxAttempts, + delayMs: e.delayMs, + errorName: e.errorName, + statusCode: e.statusCode, + since: Date.now(), + }), + ), + ); + this._register( + this.eventBus.subscribe('turn.step.interrupted', (e) => + this.setPhase({ + kind: 'interrupted', + turnId: e.turnId, + step: e.step, + reason: e.reason as 'aborted' | 'max_steps' | 'error', + message: e.message, + at: Date.now(), + }), + ), + ); + this._register( + this.eventBus.subscribe('turn.step.completed', () => this.setPhase(this.running())), + ); + this._register( + this.eventBus.subscribe('turn.ended', (e) => + this.onTurnEnded(e.turnId, e.reason, e.durationMs), + ), + ); + this._register( + this.eventBus.subscribe('permission.approval.requested', (e) => + this.onApprovalRequested(e), + ), + ); + this._register( + this.eventBus.subscribe('permission.approval.resolved', () => this.onApprovalResolved()), + ); + } + + phase(): AgentPhase { + return this.wire.getModel(RuntimeModel).phase; + } + + private onTurnStarted(turnId: number): void { + this.cursor = { turnId, step: 0, stepId: '' }; + this.priorForApproval = undefined; + this.setPhase(this.running()); + } + + private onStepStarted(turnId: number, step: number, stepId: string): void { + this.cursor = { turnId, step, stepId }; + this.setPhase(this.running()); + } + + private onDelta(stream: 'assistant' | 'thinking'): void { + this.setPhase({ + kind: 'streaming', + turnId: this.cursor.turnId, + step: this.cursor.step, + stepId: this.cursor.stepId, + stream, + since: Date.now(), + }); + } + + private onToolCallDelta(toolCallId: string, name: string | undefined): void { + this.setPhase({ + kind: 'streaming', + turnId: this.cursor.turnId, + step: this.cursor.step, + stepId: this.cursor.stepId, + stream: 'tool_call', + toolCallId, + toolName: name, + since: Date.now(), + }); + } + + private onToolCallStarted(toolCallId: string, name: string): void { + this.setPhase({ + kind: 'tool_call', + turnId: this.cursor.turnId, + step: this.cursor.step, + toolCallId, + name, + since: Date.now(), + }); + } + + private onToolResult(): void { + this.setPhase(this.running()); + } + + private onTurnEnded(turnId: number, reason: TurnEndReason, durationMs: number | undefined): void { + this.setPhase({ kind: 'ended', turnId, reason, durationMs, at: Date.now() }); + this.cursor = { turnId: -1, step: 0, stepId: '' }; + this.priorForApproval = undefined; + } + + private onApprovalRequested(approval: PermissionApprovalRequestContext): void { + this.priorForApproval = this.current; + this.setPhase({ + kind: 'awaiting_approval', + turnId: approval.turnId, + step: this.cursor.step || undefined, + approval, + since: Date.now(), + }); + } + + private onApprovalResolved(): void { + const resume = this.priorForApproval; + this.priorForApproval = undefined; + if (resume !== undefined && resume.kind !== 'idle' && resume.kind !== 'ended') { + this.setPhase(resume); + } else { + this.setPhase(this.running()); + } + } + + private running(): AgentPhase { + return { + kind: 'running', + turnId: this.cursor.turnId, + step: this.cursor.step, + stepId: this.cursor.stepId, + since: Date.now(), + }; + } + + private setPhase(phase: AgentPhase): void { + if (phaseEqual(this.current, phase)) return; + this.current = phase; + this.wire.dispatch(setRuntimePhase({ phase })); + } +} + +registerScopedService( + LifecycleScope.Agent, + IAgentRuntimeService, + AgentRuntimeService, + InstantiationType.Delayed, + 'runtime', +); diff --git a/packages/agent-core-v2/src/agent/usage/usageOps.ts b/packages/agent-core-v2/src/agent/usage/usageOps.ts index 1003e0f73..1f921f0c6 100644 --- a/packages/agent-core-v2/src/agent/usage/usageOps.ts +++ b/packages/agent-core-v2/src/agent/usage/usageOps.ts @@ -14,6 +14,7 @@ import { addUsage, type TokenUsage } from '#/app/llmProtocol/usage'; import type { LLMRequestSource } from '#/agent/llmRequester/llmRequester'; +import type { AgentPhase } from '#/agent/runtime/runtime'; import { defineModel } from '#/wire/model'; import { defineOp } from '#/wire/op'; @@ -32,6 +33,7 @@ declare module '#/app/event/eventBus' { model?: string; maxContextTokens?: number; contextTokens?: number; + phase?: AgentPhase; }; } } diff --git a/packages/agent-core-v2/src/index.ts b/packages/agent-core-v2/src/index.ts index 000106746..6b2824a6d 100644 --- a/packages/agent-core-v2/src/index.ts +++ b/packages/agent-core-v2/src/index.ts @@ -176,6 +176,9 @@ export * from '#/agent/swarm/swarm'; export * from '#/agent/swarm/swarmService'; export * from '#/agent/usage/usage'; export * from '#/agent/usage/usageService'; +export * from '#/agent/runtime/runtime'; +export * from '#/agent/runtime/runtimeOps'; +export * from '#/agent/runtime/runtimeService'; export * from '#/agent/toolDedupe/toolDedupe'; export * from '#/agent/toolDedupe/toolDedupeService'; diff --git a/packages/agent-core-v2/test/runtime/runtime.test.ts b/packages/agent-core-v2/test/runtime/runtime.test.ts new file mode 100644 index 000000000..65237cf88 --- /dev/null +++ b/packages/agent-core-v2/test/runtime/runtime.test.ts @@ -0,0 +1,228 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; + +import { SyncDescriptor } from '#/_base/di/descriptors'; +import { DisposableStore } from '#/_base/di/lifecycle'; +import { TestInstantiationService } from '#/_base/di/test'; +import { USER_PROMPT_ORIGIN } from '#/agent/contextMemory/types'; +import type { PermissionApprovalRequestContext } from '#/agent/permissionGate/permissionGateService'; +import { type AgentPhase, IAgentRuntimeService } from '#/agent/runtime/runtime'; +import { AgentRuntimeService } from '#/agent/runtime/runtimeService'; +import { RuntimeModel } from '#/agent/runtime/runtimeOps'; +import { type DomainEvent, IEventBus } from '#/app/event/eventBus'; +import { EventBusService } from '#/app/event/eventBusService'; +import { AppendLogStore } from '#/persistence/backends/node-fs/appendLogStore'; +import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService'; +import { IAppendLogStore } from '#/persistence/interface/appendLogStore'; +import { IFileSystemStorageService } from '#/persistence/interface/storage'; +import { IAgentWireService } from '#/wire/tokens'; +import type { PersistedRecord } from '#/wire/wireService'; +import { WireService } from '#/wire/wireServiceImpl'; + +const SCOPE = 'wire'; +const KEY = 'runtime-test'; + +let disposables: DisposableStore; +let ix: TestInstantiationService; +let log: IAppendLogStore; +let eventBus: IEventBus; +let svc: IAgentRuntimeService; + +beforeEach(() => { + disposables = new DisposableStore(); + ix = disposables.add(new TestInstantiationService()); + ix.stub(IFileSystemStorageService, new InMemoryStorageService()); + ix.set(IAppendLogStore, new SyncDescriptor(AppendLogStore)); + ix.set(IAgentWireService, new SyncDescriptor(WireService, [{ logScope: SCOPE, logKey: KEY }])); + ix.set(IEventBus, new SyncDescriptor(EventBusService)); + ix.set(IAgentRuntimeService, new SyncDescriptor(AgentRuntimeService)); + log = ix.get(IAppendLogStore); + eventBus = ix.get(IEventBus); + svc = ix.get(IAgentRuntimeService); +}); + +afterEach(() => disposables.dispose()); + +function collect(): AgentPhase[] { + const phases: AgentPhase[] = []; + disposables.add( + eventBus.subscribe('agent.status.updated', (e: DomainEvent<'agent.status.updated'>) => { + if (e.phase !== undefined) phases.push(e.phase); + }), + ); + return phases; +} + +async function readRecords(): Promise { + const out: PersistedRecord[] = []; + for await (const record of log.read(SCOPE, KEY)) { + out.push(record); + } + return out; +} + +function startTurn(turnId = 1, step = 1, stepId = 's1'): void { + eventBus.publish({ type: 'turn.started', turnId, origin: USER_PROMPT_ORIGIN }); + eventBus.publish({ type: 'turn.step.started', turnId, step, stepId }); +} + +const approval = { + toolCallId: 'c1', + toolName: 'Read', + action: 'read', + display: {}, + turnId: 1, + toolInput: { path: '/tmp/x' }, +} as unknown as PermissionApprovalRequestContext; + +describe('AgentRuntimeService', () => { + it('starts idle', () => { + expect(svc.phase()).toEqual({ kind: 'idle' }); + }); + + it('turn.started then turn.step.started → running with step cursor', () => { + const phases = collect(); + startTurn(); + + expect(svc.phase()).toMatchObject({ kind: 'running', turnId: 1, step: 1, stepId: 's1' }); + expect(phases.map((p) => p.kind)).toEqual(['running', 'running']); + }); + + it('first assistant.delta enters streaming(assistant); subsequent deltas are debounced', () => { + const phases = collect(); + startTurn(); + const baseline = phases.length; + + eventBus.publish({ type: 'assistant.delta', turnId: 1, delta: 'he' }); + eventBus.publish({ type: 'assistant.delta', turnId: 1, delta: 'llo' }); + + expect(svc.phase()).toMatchObject({ kind: 'streaming', stream: 'assistant' }); + expect(phases.length).toBe(baseline + 1); + }); + + it('thinking.delta switches the stream variant', () => { + startTurn(); + eventBus.publish({ type: 'assistant.delta', turnId: 1, delta: 'x' }); + eventBus.publish({ type: 'thinking.delta', turnId: 1, delta: 'hmm' }); + + expect(svc.phase()).toMatchObject({ kind: 'streaming', stream: 'thinking' }); + }); + + it('tool.call.delta → streaming(tool_call); tool.call.started → tool_call; tool.result → running', () => { + const phases = collect(); + startTurn(); + + eventBus.publish({ + type: 'tool.call.delta', + turnId: 1, + toolCallId: 'c1', + name: 'Read', + argumentsPart: '{', + }); + expect(svc.phase()).toMatchObject({ + kind: 'streaming', + stream: 'tool_call', + toolCallId: 'c1', + toolName: 'Read', + }); + + eventBus.publish({ type: 'tool.call.started', turnId: 1, toolCallId: 'c1', name: 'Read', args: {} }); + expect(svc.phase()).toMatchObject({ kind: 'tool_call', toolCallId: 'c1', name: 'Read' }); + + eventBus.publish({ type: 'tool.result', turnId: 1, toolCallId: 'c1', output: 'ok', isError: false }); + expect(svc.phase()).toMatchObject({ kind: 'running', turnId: 1, step: 1 }); + expect(phases.map((p) => p.kind)).toEqual([ + 'running', + 'running', + 'streaming', + 'tool_call', + 'running', + ]); + }); + + it('turn.step.retrying → retrying with the backoff fields', () => { + startTurn(); + eventBus.publish({ + type: 'turn.step.retrying', + turnId: 1, + step: 1, + stepId: 's1', + failedAttempt: 1, + nextAttempt: 2, + maxAttempts: 3, + delayMs: 500, + errorName: 'RateLimitError', + errorMessage: 'slow down', + statusCode: 429, + }); + + expect(svc.phase()).toMatchObject({ + kind: 'retrying', + failedAttempt: 1, + nextAttempt: 2, + maxAttempts: 3, + delayMs: 500, + errorName: 'RateLimitError', + statusCode: 429, + }); + }); + + it('turn.step.interrupted → interrupted(reason)', () => { + startTurn(); + eventBus.publish({ type: 'turn.step.interrupted', turnId: 1, step: 1, reason: 'aborted' }); + + expect(svc.phase()).toMatchObject({ kind: 'interrupted', reason: 'aborted' }); + }); + + it('turn.ended → ended(reason)', () => { + startTurn(); + eventBus.publish({ type: 'turn.ended', turnId: 1, reason: 'completed', durationMs: 123 }); + + expect(svc.phase()).toMatchObject({ kind: 'ended', turnId: 1, reason: 'completed', durationMs: 123 }); + }); + + it('permission approval requests pause into awaiting_approval and resolve back to the prior phase', () => { + startTurn(); + eventBus.publish({ type: 'assistant.delta', turnId: 1, delta: 'hi' }); + expect(svc.phase().kind).toBe('streaming'); + + eventBus.publish({ type: 'permission.approval.requested', ...approval }); + expect(svc.phase().kind).toBe('awaiting_approval'); + + eventBus.publish({ type: 'permission.approval.resolved', ...approval, decision: 'approved' }); + expect(svc.phase()).toMatchObject({ kind: 'streaming', stream: 'assistant' }); + }); + + it('only persists genuine phase transitions (sparse wire log)', async () => { + startTurn(); + eventBus.publish({ type: 'assistant.delta', turnId: 1, delta: 'a' }); + eventBus.publish({ type: 'assistant.delta', turnId: 1, delta: 'b' }); + eventBus.publish({ type: 'assistant.delta', turnId: 1, delta: 'c' }); + eventBus.publish({ type: 'turn.ended', turnId: 1, reason: 'completed', durationMs: 1 }); + + const types = (await readRecords()).map((r) => r.type); + expect(types.filter((t) => t === 'runtime.set_phase')).toHaveLength(4); + }); + + it('replay rebuilds the last phase silently (no agent.status.updated emitted)', async () => { + startTurn(); + eventBus.publish({ type: 'assistant.delta', turnId: 1, delta: 'hi' }); + eventBus.publish({ type: 'turn.ended', turnId: 1, reason: 'completed', durationMs: 5 }); + const records = await readRecords(); + + const ix2 = disposables.add(new TestInstantiationService()); + ix2.stub(IFileSystemStorageService, new InMemoryStorageService()); + ix2.set(IAppendLogStore, new SyncDescriptor(AppendLogStore)); + ix2.set(IAgentWireService, new SyncDescriptor(WireService, [{ logScope: SCOPE, logKey: 'runtime-replay' }])); + ix2.set(IEventBus, new SyncDescriptor(EventBusService)); + const fresh = ix2.get(IAgentWireService); + const bus2 = ix2.get(IEventBus); + + const emitted: DomainEvent[] = []; + disposables.add(bus2.subscribe((e) => emitted.push(e))); + + await fresh.replay(...records); + + expect(fresh.getModel(RuntimeModel).phase).toMatchObject({ kind: 'ended', reason: 'completed' }); + expect(emitted.filter((e) => e.type === 'agent.status.updated')).toHaveLength(0); + }); +}); diff --git a/packages/protocol/src/events.ts b/packages/protocol/src/events.ts index 2c73c44b1..9b24bf7f5 100644 --- a/packages/protocol/src/events.ts +++ b/packages/protocol/src/events.ts @@ -384,6 +384,69 @@ export interface McpOAuthAuthorizationUrlUpdateData { export type TurnEndReason = 'completed' | 'cancelled' | 'failed' | 'blocked'; +export type AgentPhase = + | { readonly kind: 'idle' } + | { + readonly kind: 'running'; + readonly turnId: number; + readonly step: number; + readonly stepId: string; + readonly since: number; + } + | { + readonly kind: 'streaming'; + readonly turnId: number; + readonly step: number; + readonly stepId: string; + readonly stream: 'assistant' | 'thinking' | 'tool_call'; + readonly toolCallId?: string; + readonly toolName?: string; + readonly since: number; + } + | { + readonly kind: 'tool_call'; + readonly turnId: number; + readonly step: number; + readonly toolCallId: string; + readonly name: string; + readonly since: number; + } + | { + readonly kind: 'retrying'; + readonly turnId: number; + readonly step: number; + readonly stepId: string; + readonly failedAttempt: number; + readonly nextAttempt: number; + readonly maxAttempts: number; + readonly delayMs: number; + readonly errorName?: string; + readonly statusCode?: number; + readonly since: number; + } + | { + readonly kind: 'awaiting_approval'; + readonly turnId: number; + readonly step?: number; + readonly approval?: unknown; + readonly since: number; + } + | { + readonly kind: 'interrupted'; + readonly turnId: number; + readonly step?: number; + readonly reason: 'aborted' | 'max_steps' | 'error'; + readonly message?: string; + readonly at: number; + } + | { + readonly kind: 'ended'; + readonly turnId: number; + readonly reason: TurnEndReason; + readonly durationMs?: number; + readonly at: number; + }; + export interface AgentStatusUpdatedEvent { readonly type: 'agent.status.updated'; readonly model?: string; @@ -394,6 +457,7 @@ export interface AgentStatusUpdatedEvent { readonly swarmMode?: boolean; readonly permission?: PermissionMode; readonly usage?: UsageStatus; + readonly phase?: AgentPhase; } export interface SessionMetaUpdatedEvent { @@ -1147,6 +1211,70 @@ export const mcpOAuthAuthorizationUrlUpdateDataSchema = z.object({ export const turnEndReasonSchema = z.enum(['completed', 'cancelled', 'failed', 'blocked']) satisfies z.ZodType; +export const agentPhaseSchema = z.discriminatedUnion('kind', [ + z.object({ kind: z.literal('idle') }), + z.object({ + kind: z.literal('running'), + turnId: z.number(), + step: z.number(), + stepId: z.string(), + since: z.number(), + }), + z.object({ + kind: z.literal('streaming'), + turnId: z.number(), + step: z.number(), + stepId: z.string(), + stream: z.enum(['assistant', 'thinking', 'tool_call']), + toolCallId: z.string().optional(), + toolName: z.string().optional(), + since: z.number(), + }), + z.object({ + kind: z.literal('tool_call'), + turnId: z.number(), + step: z.number(), + toolCallId: z.string(), + name: z.string(), + since: z.number(), + }), + z.object({ + kind: z.literal('retrying'), + turnId: z.number(), + step: z.number(), + stepId: z.string(), + failedAttempt: z.number(), + nextAttempt: z.number(), + maxAttempts: z.number(), + delayMs: z.number(), + errorName: z.string().optional(), + statusCode: z.number().optional(), + since: z.number(), + }), + z.object({ + kind: z.literal('awaiting_approval'), + turnId: z.number(), + step: z.number().optional(), + approval: z.unknown().optional(), + since: z.number(), + }), + z.object({ + kind: z.literal('interrupted'), + turnId: z.number(), + step: z.number().optional(), + reason: z.enum(['aborted', 'max_steps', 'error']), + message: z.string().optional(), + at: z.number(), + }), + z.object({ + kind: z.literal('ended'), + turnId: z.number(), + reason: turnEndReasonSchema, + durationMs: z.number().optional(), + at: z.number(), + }), +]) satisfies z.ZodType; + export const agentStatusUpdatedEventSchema = z.object({ type: z.literal('agent.status.updated'), model: z.string().optional(), @@ -1157,6 +1285,7 @@ export const agentStatusUpdatedEventSchema = z.object({ swarmMode: z.boolean().optional(), permission: permissionModeSchema.optional(), usage: usageStatusSchema.optional(), + phase: agentPhaseSchema.optional(), }) satisfies z.ZodType; export const sessionMetaUpdatedEventSchema = z.object({