diff --git a/apps/desktop/src/main/__tests__/localized-main-shell-contract.test.ts b/apps/desktop/src/main/__tests__/localized-main-shell-contract.test.ts index 01af232cd9..980e63ad8f 100644 --- a/apps/desktop/src/main/__tests__/localized-main-shell-contract.test.ts +++ b/apps/desktop/src/main/__tests__/localized-main-shell-contract.test.ts @@ -423,7 +423,7 @@ describe('localized main shell contract', () => { it('surfaces permission denial in Chinese instead of raw English backend text', async () => { const components = await readFile(resolve(process.cwd(), '..', '..', 'packages', 'ui', 'src', 'components.tsx'), 'utf8'); - const aiSdk = await readFile(resolve(process.cwd(), '..', '..', 'packages', 'runtime', 'src', 'ai-sdk-backend.ts'), 'utf8'); + const toolRuntime = await readFile(resolve(process.cwd(), '..', '..', 'packages', 'runtime', 'src', 'tool-runtime.ts'), 'utf8'); const piAgent = await readFile(resolve(process.cwd(), '..', '..', 'packages', 'runtime', 'src', 'pi-agent-backend.ts'), 'utf8'); assert.match(components, /formatUserVisibleToolText\(text: string\)[\s\S]*User denied permission[\s\S]*用户已拒绝权限请求/); @@ -433,8 +433,8 @@ describe('localized main shell contract', () => { assert.match(components, /item\.result && !permissionDenied/); assert.match(components, /formatUserVisibleToolText\(redactSecrets\(extractErrorText\(props\.result\)\)\)/); assert.match(components, /capLines\(formatUserVisibleToolText\(redactSecrets\(content\.text\)\)\)/); - assert.match(aiSdk, /const reason = '用户已拒绝权限请求';/); + assert.match(toolRuntime, /const reason = '用户已拒绝权限请求';/); assert.match(piAgent, /text: '用户已拒绝权限请求'/); - assert.doesNotMatch(`${aiSdk}\n${piAgent}`, /User denied permission/); + assert.doesNotMatch(`${toolRuntime}\n${piAgent}`, /User denied permission/); }); }); diff --git a/apps/desktop/src/main/__tests__/web-search-boundary.test.ts b/apps/desktop/src/main/__tests__/web-search-boundary.test.ts index 1e4094bddf..9dea1c5c16 100644 --- a/apps/desktop/src/main/__tests__/web-search-boundary.test.ts +++ b/apps/desktop/src/main/__tests__/web-search-boundary.test.ts @@ -360,7 +360,7 @@ describe('web-search renderer boundary (PR-WEB-SEARCH-TAVILY-0)', () => { it('WebSearch agent errors render as repair-oriented cards, not raw JSON', async () => { const ui = await readFile(join(REPO_ROOT, 'packages/ui/src/components.tsx'), 'utf8'); - const runtime = await readFile(join(REPO_ROOT, 'packages/runtime/src/ai-sdk-backend.ts'), 'utf8'); + const runtime = await readFile(join(REPO_ROOT, 'packages/runtime/src/tool-runtime.ts'), 'utf8'); const agentTool = await readFile(join(REPO_ROOT, 'apps/desktop/src/main/web-search/agent-tool.ts'), 'utf8'); const coreEvents = await readFile(join(REPO_ROOT, 'packages/core/src/events.ts'), 'utf8'); const overlay = ui.match(/function OverlayPreview[\s\S]*?if \(content\.kind === 'json'\)/); diff --git a/packages/runtime/src/__tests__/runtime-runner.test.ts b/packages/runtime/src/__tests__/runtime-runner.test.ts index 6815f1384f..60d71677e8 100644 --- a/packages/runtime/src/__tests__/runtime-runner.test.ts +++ b/packages/runtime/src/__tests__/runtime-runner.test.ts @@ -177,7 +177,39 @@ describe('RuntimeRunner', () => { expect(result.events[1]!.author).toBe('agent'); }); - test('a terminal event ends the result and stops collecting flow events', async () => { + test('caller-provided invocationId and runId are used across result, user event, and flow', async () => { + const providers = makeProviders(); + const flow = new ScriptFlow((ctx) => [ + flowTextEvent(ctx, 'flow-uses-caller-ids'), + flowTerminalEvent(ctx, 'completed'), + ]); + const runner = new RuntimeRunner({ flow, providers }); + + const result = await runner.run( + makeRequest({ + invocationId: 'inv-production-1', + runId: 'run-production-1', + }), + ); + + expect(result.invocationId).toBe('inv-production-1'); + expect(result.runId).toBe('run-production-1'); + expect(flow.seen).toHaveLength(1); + expect(flow.seen[0]!.invocationId).toBe('inv-production-1'); + expect(flow.seen[0]!.runId).toBe('run-production-1'); + + const userEvent = result.events[0]!; + expect(userEvent.author).toBe('user'); + expect(userEvent.invocationId).toBe('inv-production-1'); + expect(userEvent.runId).toBe('run-production-1'); + + for (const ev of result.events) { + expect(ev.invocationId).toBe('inv-production-1'); + expect(ev.runId).toBe('run-production-1'); + } + }); + + test('default behavior stops collecting at the first terminal flow event', async () => { const providers = makeProviders(); const flow = new ScriptFlow((ctx) => [ flowTextEvent(ctx, 'partial'), @@ -203,6 +235,34 @@ describe('RuntimeRunner', () => { ).toBe(false); }); + test('stopOnTerminal false keeps draining and fails on any non-completed terminal event', async () => { + const providers = makeProviders(); + const flow = new ScriptFlow((ctx) => [ + flowTerminalEvent(ctx, 'completed'), + flowTextEvent(ctx, 'cleanup-after-completed'), + flowTerminalEvent(ctx, 'aborted'), + flowTextEvent(ctx, 'cleanup-after-aborted'), + ]); + const runner = new RuntimeRunner({ flow, providers, stopOnTerminal: false }); + + const result = await runner.run(makeRequest()); + + expect(result.status).toBe('failed'); + expect(result.failure?.class).toBe('aborted'); + expect(result.failure?.terminalStatus).toBe('aborted'); + expect(result.events).toHaveLength(5); + expect( + result.events.some( + (ev) => ev.content?.kind === 'text' && ev.content.text === 'cleanup-after-completed', + ), + ).toBe(true); + expect( + result.events.some( + (ev) => ev.content?.kind === 'text' && ev.content.text === 'cleanup-after-aborted', + ), + ).toBe(true); + }); + test('a flow that throws maps to a failed result (user event retained)', async () => { const providers = makeProviders(); const flow = new ThrowingFlow(new Error('boom')); diff --git a/packages/runtime/src/__tests__/session-manager.test.ts b/packages/runtime/src/__tests__/session-manager.test.ts index 989d6ffd2d..a8f88ce3b1 100644 --- a/packages/runtime/src/__tests__/session-manager.test.ts +++ b/packages/runtime/src/__tests__/session-manager.test.ts @@ -23,6 +23,7 @@ import { type SessionStore, } from '../session-manager.js'; import type { AgentBackend } from '../ai-sdk-backend.js'; +import type { InvocationResult } from '../invocation-context.js'; describe('SessionManager permission mode updates', () => { test('updates header, rebuilds active backend, and writes an audit note', async () => { @@ -207,6 +208,50 @@ describe('SessionManager permission mode updates', () => { expect(built).toEqual(['Before']); }); + test('sendMessage is driven through RuntimeRunner while preserving the SessionEvent stream', async () => { + const store = new MemorySessionStore(); + const runStore = new MemoryAgentRunStore(); + const backends = new BackendRegistry(); + const observed: InvocationResult[] = []; + backends.register('fake', (ctx) => new TestBackend(ctx)); + const manager = new SessionManager({ + store, + runStore, + backends, + newId: nextId(), + now: nextNow(6_500), + runtimeSource: 'test', + runtimeInvocationObserver: (result) => { + observed.push(result); + }, + }); + const session = await manager.createSession(makeInput()); + + const sessionEvents = await collectSessionEvents( + manager.sendMessage(session.id, { turnId: 'turn-1', text: 'hello' }), + ); + + expect(sessionEvents.map((event) => event.type)).toEqual(['text_delta', 'complete']); + expect(sessionEvents.map((event) => event.id)).toEqual(['turn-1-delta', 'turn-1-complete']); + expect(observed.length).toBe(1); + + const [run] = await runStore.listSessionRuns(session.id); + if (!run) throw new Error('AgentRunStore run was not created'); + const result = observed[0]!; + expect(result.runId).toBe(run.runId); + expect(result.sessionId).toBe(session.id); + expect(result.turnId).toBe('turn-1'); + expect(result.status).toBe('completed'); + expect(result.events.map((event) => event.runId)).toEqual([run.runId, run.runId, run.runId]); + expect(result.events.map((event) => event.sessionId)).toEqual([session.id, session.id, session.id]); + expect(result.events.map((event) => event.turnId)).toEqual(['turn-1', 'turn-1', 'turn-1']); + expect(result.events.map((event) => event.role)).toEqual(['user', 'model', 'system']); + expect(result.events.map((event) => event.id)).toEqual(['id-3', 'turn-1-delta', 'turn-1-complete']); + expect(result.events[0]?.content).toEqual({ kind: 'text', text: 'hello' }); + expect(result.events[1]?.content).toEqual({ kind: 'text', text: 'ok' }); + expect(result.events[2]?.status).toBe('completed'); + }); + test('rejects backend configuration updates while a turn is actively streaming', async () => { const store = new MemorySessionStore(); const backends = new BackendRegistry(); @@ -1168,6 +1213,14 @@ async function drain(iterable: AsyncIterable): Promise { } } +async function collectSessionEvents(iterable: AsyncIterable): Promise { + const events: SessionEvent[] = []; + for await (const event of iterable) { + events.push(event); + } + return events; +} + async function expectRejects(promise: Promise, pattern: RegExp): Promise { try { await promise; diff --git a/packages/runtime/src/invocation-context.ts b/packages/runtime/src/invocation-context.ts index 10ecde4e9c..4cd2fa6ed1 100644 --- a/packages/runtime/src/invocation-context.ts +++ b/packages/runtime/src/invocation-context.ts @@ -4,11 +4,9 @@ * Source: docs/runtime-v2-architecture-evolution.md §Target Architecture, * §Proposed Module Shape, and Phase 2 (RuntimeRunner Shell). * - * Phase 2 scope (this node): types + injectable providers only. The - * RuntimeRunner consumes these to build a testable invocation shell driven - * by fake services. It is deliberately NOT wired to SessionStore / - * SessionManager yet — that delegation lands in a later phase, after the - * AgentFlow / projection nodes exist. The value here is the seam and tests. + * Phase 2 scope: types + injectable providers. RuntimeRunner consumes these + * to build a testable invocation shell and can also be handed production ids + * from an already-created AgentRun while migration wiring is in progress. * * Identity hierarchy carried on every context: sessionId ⊃ invocationId ⊃ * runId ⊃ turnId. These mirror the canonical RuntimeEvent fields so events @@ -58,15 +56,14 @@ export interface InvocationLineage { /** * Request to run one agent invocation. The runner owns preflight, context - * creation, the initial user RuntimeEvent, and flow dispatch. It does not - * read or write SessionStore in this skeleton. - * - * `invocationId` / `runId` are generated by the runner through the injected - * providers (see InvocationContext); they are intentionally not on the - * request so callers cannot assert a fake spine identity. + * creation, the initial user RuntimeEvent, and flow dispatch. Callers may + * provide existing production spine ids (for example from AgentRun); when + * omitted, the runner generates them through the injected providers. */ export interface InvocationRequest { sessionId: string; + invocationId?: string; + runId?: string; turnId: string; text: string; /** Optional attachments bound to this user turn. */ diff --git a/packages/runtime/src/runtime-runner.ts b/packages/runtime/src/runtime-runner.ts index c0d5f8a6d8..06b97361e4 100644 --- a/packages/runtime/src/runtime-runner.ts +++ b/packages/runtime/src/runtime-runner.ts @@ -4,10 +4,10 @@ * Source: docs/runtime-v2-architecture-evolution.md §Target Architecture and * Phase 2 (RuntimeRunner Shell). * - * This is an internal seam, not the production hot path. It is intentionally - * decoupled from SessionManager / SessionStore so it can be exercised in - * tests with fake services, and so SessionManager.sendMessage can delegate to - * it incrementally in a later phase without a big-bang rewrite. + * RuntimeRunner is the invocation shell. It remains decoupled from + * SessionManager / SessionStore so it can be exercised with fake services, + * while still being able to wrap production AgentRun streams during the + * Runtime v2 migration. * * Responsibilities (per the node spec): * 1. Run an injectable preflight gate. @@ -17,8 +17,9 @@ * 5. Return a structured result with the collected events and a terminal * status. * - * Out-of-scope (deliberately): SessionStore writes, projection driving, - * AgentRunStore ledger writes, and replacing SessionManager.sendMessage. + * Out-of-scope (deliberately): direct SessionStore writes, projection + * driving, and AgentRunStore ledger writes. Those remain owned by AgentRun + * while SessionManager delegates invocation execution through this shell. */ import { @@ -98,6 +99,12 @@ export interface RuntimeRunnerDeps { gate?: RuntimeGate; /** Injectable id/time providers. Defaults to crypto.randomUUID / Date.now. */ providers?: InvocationProviders; + /** + * Whether to stop collecting at the first terminal RuntimeEvent. Defaults + * to true for standalone runner callers; production bridges can set false + * to keep draining cleanup/trailing events from wrapped streams. + */ + stopOnTerminal?: boolean; } // ============================================================================ @@ -108,24 +115,27 @@ export class RuntimeRunner { private readonly flow: AgentFlowLike; private readonly gate: RuntimeGate | undefined; private readonly providers: InvocationProviders; + private readonly stopOnTerminal: boolean; constructor(deps: RuntimeRunnerDeps) { this.flow = deps.flow; this.gate = deps.gate; this.providers = deps.providers ?? createDefaultInvocationProviders(); + this.stopOnTerminal = deps.stopOnTerminal ?? true; } /** * Run one invocation end-to-end and return a structured result. * * Event order is guaranteed: the initial user RuntimeEvent is always - * collected before any flow event. Collection stops at the first terminal - * RuntimeEvent; a terminal event is what ends the result. + * collected before any flow event. By default collection stops at the first + * terminal RuntimeEvent; callers that wrap streams with cleanup/trailing + * events can opt into full draining through RuntimeRunnerDeps. */ async run(request: InvocationRequest): Promise { const startedAt = this.providers.now(); - const invocationId = this.providers.newId(); - const runId = this.providers.newId(); + const invocationId = request.invocationId ?? this.providers.newId(); + const runId = request.runId ?? this.providers.newId(); // 1. Preflight (injectable gate). On failure we admit no invocation: no // context, no user event, no flow dispatch. @@ -187,10 +197,11 @@ export class RuntimeRunner { events.push(buildUserEvent(ctx, request)); const flowInput = buildFlowInput(request); - // 5. Dispatch to the flow and collect canonical events. The first - // terminal event ends the result; events emitted after it are not - // collected. A thrown error or a non-completed terminal status maps - // the result to 'failed'. + // 5. Dispatch to the flow and collect canonical events. By default the + // first terminal event ends collection; when stopOnTerminal is false, + // keep draining while remembering any non-completed terminal status. + // A thrown error or a non-completed terminal status maps the result + // to 'failed'. let failure: InvocationFailure | undefined; let terminalSeen = false; try { @@ -198,8 +209,10 @@ export class RuntimeRunner { events.push(ev); if (isTerminalRuntimeEvent(ev)) { terminalSeen = true; - failure = failureFromTerminalEvent(ev); - break; + failure ??= failureFromTerminalEvent(ev); + if (this.stopOnTerminal) { + break; + } } } } catch (error) { diff --git a/packages/runtime/src/session-manager.ts b/packages/runtime/src/session-manager.ts index 9deae6258d..54713d05de 100644 --- a/packages/runtime/src/session-manager.ts +++ b/packages/runtime/src/session-manager.ts @@ -52,6 +52,15 @@ import type { AgentBackend } from './ai-sdk-backend.js'; import type { RunTraceRecorder } from './run-trace.js'; import { AgentRun, type AgentRunActiveSession, type AgentRunLineage } from './agent-run.js'; import { classifyAgentRunRecovery, type AgentRunRecoveryDecision } from './agent-run-recovery.js'; +import { + createSessionEventMapMemory, + mapSessionEventToRuntimeEvent, +} from './ai-sdk-flow.js'; +import type { + InvocationResult, + InvocationSource, +} from './invocation-context.js'; +import { RuntimeRunner } from './runtime-runner.js'; export interface StopSessionInput { source?: 'stop_button'; @@ -119,6 +128,8 @@ export interface SessionManagerDeps { backends: BackendRegistry; newId: () => string; now: () => number; + runtimeSource?: InvocationSource; + runtimeInvocationObserver?: (result: InvocationResult) => void | Promise; } interface ActiveSession extends AgentRunActiveSession { @@ -313,12 +324,11 @@ export class SessionManager { * (desktop main) is expected to forward the events to the renderer over * the IPC bridge. * - * Phase 1 vertical (§9): - * 1. Append UserMessage to JSONL + flush. - * 2. Lock connection (set connectionLocked=true) if not already. - * 3. Lookup or build the AgentBackend for this session. - * 4. backend.send(input) → forward events. - * 5. Update lastMessageAt + hasUnread when complete. + * Runtime v2 bridge: + * 1. Create one AgentRun, which remains the persistence/ledger owner. + * 2. Run that AgentRun through RuntimeRunner for canonical RuntimeEvents. + * 3. Forward the original SessionEvents to callers unchanged. + * 4. Drain the AgentRun stream fully so stop/abort cleanup semantics stay intact. */ async *sendMessage( sessionId: string, @@ -344,7 +354,70 @@ export class SessionManager { this.appendTurnState(targetSessionId, turnId, status, lineage, options), }, }); - yield* run.execute(); + + const sessionEvents = new AsyncEventQueue(); + const abortController = new AbortController(); + let agentRunIterator: AsyncIterator | undefined; + let flowDone = false; + const runner = new RuntimeRunner({ + providers: { newId: this.deps.newId, now: this.deps.now }, + stopOnTerminal: false, + flow: { + run: async function* (ctx, request) { + const memory = createSessionEventMapMemory(); + agentRunIterator = run.execute()[Symbol.asyncIterator](); + try { + while (!request.abortSignal?.aborted) { + const next = await agentRunIterator.next(); + if (next.done) break; + if (request.abortSignal?.aborted) break; + await sessionEvents.push(next.value); + yield mapSessionEventToRuntimeEvent(next.value, ctx, memory); + } + } catch (error) { + if (!isAsyncEventQueueClosed(error)) { + sessionEvents.fail(error); + } + throw error; + } finally { + flowDone = true; + if (request.abortSignal?.aborted) { + await agentRunIterator.return?.().catch(() => undefined); + } + sessionEvents.close(); + } + }, + }, + }); + const runnerResult = runner.run({ + sessionId, + runId: run.runId, + turnId: run.turnId, + text: input.text, + source: this.deps.runtimeSource ?? 'desktop', + lineage: run.lineage, + abortSignal: abortController.signal, + }).then(async (result) => { + await this.deps.runtimeInvocationObserver?.(result); + return result; + }, (error) => { + sessionEvents.fail(error); + throw error; + }); + + try { + for await (const event of sessionEvents) { + yield event; + } + await runnerResult; + } finally { + if (!flowDone) { + abortController.abort(); + sessionEvents.close(); + await agentRunIterator?.return?.().catch(() => undefined); + } + await runnerResult.catch(() => undefined); + } } async stopSession(sessionId: string, input: StopSessionInput = {}): Promise { @@ -813,6 +886,90 @@ function normalizeStopSessionSource(source: StopSessionInput['source'] | undefin } } +class AsyncEventQueueClosed extends Error { + constructor() { + super('Async event queue closed'); + this.name = 'AsyncEventQueueClosed'; + } +} + +function isAsyncEventQueueClosed(error: unknown): boolean { + return error instanceof AsyncEventQueueClosed; +} + +interface AsyncEventQueueEntry { + value: T; + delivered: () => void; + rejected: (error: unknown) => void; +} + +class AsyncEventQueue implements AsyncIterable { + private readonly values: Array> = []; + private readonly waiters: Array<{ + resolve: (entry: AsyncEventQueueEntry | undefined) => void; + reject: (error: unknown) => void; + }> = []; + private closed = false; + private failure: unknown; + + [Symbol.asyncIterator](): AsyncIterator { + return this.consume()[Symbol.asyncIterator](); + } + + push(value: T): Promise { + if (this.failure) return Promise.reject(this.failure); + if (this.closed) return Promise.reject(new AsyncEventQueueClosed()); + return new Promise((resolve, reject) => { + const entry = { value, delivered: resolve, rejected: reject }; + const waiter = this.waiters.shift(); + if (waiter) { + waiter.resolve(entry); + return; + } + this.values.push(entry); + }); + } + + fail(error: unknown): void { + if (this.failure) return; + this.failure = error; + for (const value of this.values.splice(0)) value.rejected(error); + for (const waiter of this.waiters.splice(0)) waiter.reject(error); + } + + close(): void { + if (this.closed) return; + this.closed = true; + const closed = new AsyncEventQueueClosed(); + for (const value of this.values.splice(0)) value.rejected(closed); + for (const waiter of this.waiters.splice(0)) waiter.resolve(undefined); + } + + private async *consume(): AsyncIterable { + while (true) { + const entry = await this.nextEntry(); + if (!entry) return; + try { + yield entry.value; + } finally { + entry.delivered(); + } + } + } + + private nextEntry(): Promise | undefined> { + if (this.values.length > 0) { + const next = this.values.shift()!; + return Promise.resolve(next); + } + if (this.failure) return Promise.reject(this.failure); + if (this.closed) return Promise.resolve(undefined); + return new Promise | undefined>((resolve, reject) => { + this.waiters.push({ resolve, reject }); + }); + } +} + // Re-export the suppressed-unused types so this file is the canonical home // for them. (Avoids TS "imported but unused" warnings.) export type {