Wire RuntimeRunner through SessionManager

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/);
  });
 });

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'\)/);

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'));

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<unknown>): Promise<void> {
  }
 }
 
+async function collectSessionEvents(iterable: AsyncIterable<SessionEvent>): Promise<SessionEvent[]> {
+ const events: SessionEvent[] = [];
+ for await (const event of iterable) {
+ events.push(event);
+ }
+ return events;
+}
+
 async function expectRejects(promise: Promise<unknown>, pattern: RegExp): Promise<void> {
  try {
  await promise;

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. */

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<InvocationResult> {
  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,19 +197,22 @@ 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 {
  for await (const ev of this.flow.run(ctx, flowInput)) {
  events.push(ev);
  if (isTerminalRuntimeEvent(ev)) {
  terminalSeen = true;
- failure = failureFromTerminalEvent(ev);
- break;
+ failure ??= failureFromTerminalEvent(ev);
+ if (this.stopOnTerminal) {
+ break;
+ }
  }
  }
  } catch (error) {