diff --git a/packages/core/src/__tests__/runtime-event.test.ts b/packages/core/src/__tests__/runtime-event.test.ts index e0a5bbad10..281116577c 100644 --- a/packages/core/src/__tests__/runtime-event.test.ts +++ b/packages/core/src/__tests__/runtime-event.test.ts @@ -89,6 +89,28 @@ describe('RuntimeEvent content variants', () => { expect(content.text).toBe('hello'); }); + test('text content can carry attachment refs without changing its kind', () => { + const content: RuntimeEventContent = { + kind: 'text', + text: 'see attached', + attachments: [ + { + kind: 'image', + name: 'chart.png', + mimeType: 'image/png', + bytes: 123, + ref: { + kind: 'session_file', + sessionId: 'sess-1', + relativePath: 'attachments/chart.png', + }, + }, + ], + }; + if (content.kind !== 'text') throw new Error('unreachable'); + expect(content.attachments?.[0]?.name).toBe('chart.png'); + }); + test('thinking content may carry a replay signature', () => { const content: RuntimeEventContent = { kind: 'thinking', diff --git a/packages/core/src/runtime-event.ts b/packages/core/src/runtime-event.ts index e122c685ec..9fd3076b22 100644 --- a/packages/core/src/runtime-event.ts +++ b/packages/core/src/runtime-event.ts @@ -13,6 +13,7 @@ * projection, or ledger logic lives here. Those arrive in later nodes. */ +import type { AttachmentRef } from './events.js'; import type { PermissionRequest, PermissionResponse } from './permission.js'; // ============================================================================ @@ -88,6 +89,13 @@ export function isTerminalRuntimeEventStatus(value: unknown): boolean { export interface RuntimeEventTextContent { kind: 'text'; text: string; + /** + * Optional user-bound attachments carried with the text turn. Adapters + * MUST preserve these when converting legacy UserMessage rows so + * RuntimeEvent history does not silently degrade multimodal/file turns + * into plain text. + */ + attachments?: AttachmentRef[]; } export interface RuntimeEventThinkingContent { diff --git a/packages/runtime/src/__tests__/ai-sdk-flow.test.ts b/packages/runtime/src/__tests__/ai-sdk-flow.test.ts index 51d1773c03..f6d145ae5b 100644 --- a/packages/runtime/src/__tests__/ai-sdk-flow.test.ts +++ b/packages/runtime/src/__tests__/ai-sdk-flow.test.ts @@ -3,7 +3,7 @@ import { describe, test } from 'node:test'; import type { BackendKind } from '@maka/core/session'; import type { SessionEvent } from '@maka/core/events'; -import type { PermissionDecision } from '@maka/core/backend-types'; +import type { BackendSendInput, PermissionDecision } from '@maka/core/backend-types'; import type { RuntimeEvent } from '@maka/core/runtime-event'; import { isTerminalRuntimeEvent, @@ -20,6 +20,7 @@ import { flowSupportsControl, } from '../agent-flow.js'; import type { AgentBackend } from '../ai-sdk-backend.js'; +import { RuntimeRunner } from '../runtime-runner.js'; // ============================================================================ // Fake backend — scripted SessionEvent stream + recorded control calls @@ -38,6 +39,7 @@ class ScriptedBackend implements AgentBackend { readonly sessionId: string; readonly stopCalls: Array<'user_stop' | 'redirect'> = []; readonly permissionCalls: PermissionDecision[] = []; + readonly sendInputs: BackendSendInput[] = []; disposeCalls = 0; sendCalls = 0; private readonly events: SessionEvent[]; @@ -50,8 +52,9 @@ class ScriptedBackend implements AgentBackend { this.gate = c.gate; } - async *send(): AsyncIterable { + async *send(input: BackendSendInput): AsyncIterable { this.sendCalls += 1; + this.sendInputs.push(input); for (const e of this.events) { yield e; if (this.gate) await this.gate(); @@ -158,6 +161,55 @@ describe('AiSdkFlow seam', () => { assert.equal(backend.sendCalls, 1); }); + test('RuntimeRunner dispatches AiSdkFlow with defined context and preserved attachments', async () => { + const attachment = { + kind: 'image' as const, + name: 'chart.png', + mimeType: 'image/png', + bytes: 123, + ref: { kind: 'session_file' as const, sessionId: 'session-1', relativePath: 'attachments/chart.png' }, + }; + const history = [ + { + type: 'user' as const, + id: 'u-prev', + turnId: 'turn-prev', + ts: 1, + text: 'previous', + }, + ]; + const backend = new ScriptedBackend({ + events: [ev({ type: 'complete', stopReason: 'end_turn' })], + }); + const flow = new AiSdkFlow({ backend }); + let idSeq = 0; + const runner = new RuntimeRunner({ + flow, + providers: { + newId: () => `rt-${(idSeq += 1)}`, + now: () => 1000, + }, + }); + + const result = await runner.run({ + sessionId: 'session-1', + turnId: 'turn-1', + text: 'hi', + attachments: [attachment], + context: history, + source: 'test', + }); + + assert.equal(result.status, 'completed'); + assert.equal(backend.sendInputs.length, 1); + assert.deepEqual(backend.sendInputs[0], { + turnId: 'turn-1', + text: 'hi', + attachments: [attachment], + context: history, + }); + }); + test('maps thinking deltas/signature onto model thinking content', async () => { const backend = new ScriptedBackend({ events: [ @@ -281,7 +333,7 @@ describe('AiSdkFlow seam', () => { assert.equal(isTerminalRuntimeEvent(out[1]), true); }); - test('maps the abort path preserving order (faithful, no coalescing)', async () => { + test('maps the abort path to exactly one terminal event', async () => { const backend = new ScriptedBackend({ events: [ ev({ type: 'text_delta', messageId: 'm1', text: 'par' }), @@ -292,15 +344,59 @@ describe('AiSdkFlow seam', () => { const flow = new AiSdkFlow({ backend }); const out = await collect(flow.run(ctx, { text: 'hi', context: [] })); - // The adapter is faithful to the backend stream: both abort and the - // trailing complete are emitted (coalescing is a projection concern). - assert.equal(out.length, 3); + // AgentFlow guarantees exactly one terminal event, so the trailing + // complete(user_stop) from the legacy backend is coalesced away. + assert.equal(out.length, 2); assert.equal(out[1].status, 'aborted'); assert.equal(out[1].actions?.endInvocation, true); assert.equal(isTerminalRuntimeEvent(out[1]), true); - // Stream closes with the trailing terminal complete. - assert.equal(isTerminalRuntimeEvent(out[2]), true); - assert.equal(out[2].status, 'aborted'); + assert.equal(out.filter(isTerminalRuntimeEvent).length, 1); + }); + + test('stops yielding after the first terminal event', async () => { + const backend = new ScriptedBackend({ + events: [ + ev({ type: 'abort', reason: 'user_stop' }), + ev({ type: 'text_delta', messageId: 'm1', text: 'after-terminal' }), + ev({ type: 'complete', stopReason: 'user_stop' }), + ], + }); + const flow = new AiSdkFlow({ backend }); + const out = await collect(flow.run(ctx, { text: 'hi', context: [] })); + + assert.equal(out.length, 1); + assert.equal(out[0]?.status, 'aborted'); + assert.equal(isTerminalRuntimeEvent(out[0]), true); + }); + + test('RuntimeRunner consumes AiSdkFlow abort as one coherent failed outcome', async () => { + const backend = new ScriptedBackend({ + events: [ + ev({ type: 'text_delta', messageId: 'm1', text: 'par' }), + ev({ type: 'abort', reason: 'user_stop' }), + ev({ type: 'complete', stopReason: 'user_stop' }), + ], + }); + const flow = new AiSdkFlow({ backend }); + let idSeq = 0; + const runner = new RuntimeRunner({ + flow, + providers: { + newId: () => `id-${(idSeq += 1)}`, + now: () => 1000, + }, + }); + + const result = await runner.run({ + sessionId: 'session-1', + turnId: 'turn-1', + text: 'hi', + source: 'test', + }); + + assert.equal(result.status, 'failed'); + assert.equal(result.failure?.class, 'aborted'); + assert.equal(result.events.filter(isTerminalRuntimeEvent).length, 1); }); test('delegates stop / respondToPermission / dispose to the wrapped backend', async () => { diff --git a/packages/runtime/src/__tests__/runtime-event-adapters.test.ts b/packages/runtime/src/__tests__/runtime-event-adapters.test.ts index ae914a0c7c..2b2d2aa2ef 100644 --- a/packages/runtime/src/__tests__/runtime-event-adapters.test.ts +++ b/packages/runtime/src/__tests__/runtime-event-adapters.test.ts @@ -12,6 +12,7 @@ import { describe, test } from 'node:test'; import { expect } from '../test-helpers.js'; +import type { AttachmentRef } from '@maka/core/events'; import type { UserMessage, AssistantMessage, @@ -42,6 +43,14 @@ import { const ts = 1_700_000_000_000; const turnId = 't1'; +const attachment: AttachmentRef = { + kind: 'pdf', + name: 'brief.pdf', + mimeType: 'application/pdf', + bytes: 2048, + ref: { kind: 'session_file', sessionId: 'sess-1', relativePath: 'attachments/brief.pdf' }, +}; + const user = (id: string, text: string): UserMessage => ({ type: 'user', id, @@ -166,6 +175,20 @@ describe('storedMessageToRuntimeEvent', () => { expect(e.ts).toBe(ts + 1); }); + test('user message with attachments preserves attachment refs in text content', () => { + const e = storedMessageToRuntimeEvent( + { ...user('u-attach', 'see attached'), attachments: [attachment] }, + ctx, + ); + expect(e).not.toBeNull(); + if (!e) return; + expect(e.content).toEqual({ + kind: 'text', + text: 'see attached', + attachments: [attachment], + }); + }); + test('assistant message (text only) → role model, text content; thinking dropped', () => { const e = storedMessageToRuntimeEvent(assistant('a1', 'hi'), ctx); if (!e) throw new Error('expected event'); @@ -266,6 +289,19 @@ describe('storedMessageToRuntimeEvents', () => { expect(out[0]?.content).toEqual({ kind: 'text', text: 'hello' }); }); + test('user message with attachments → single attachment-preserving event', () => { + const out = storedMessageToRuntimeEvents( + { ...user('u-attach', 'see attached'), attachments: [attachment] }, + ctx, + ); + expect(out).toHaveLength(1); + expect(out[0]?.content).toEqual({ + kind: 'text', + text: 'see attached', + attachments: [attachment], + }); + }); + test('tool_call → empty array', () => { expect(storedMessageToRuntimeEvents(toolCall('tc', 'Read'), ctx)).toEqual([]); }); @@ -303,6 +339,19 @@ describe('runtimeEventToStoredMessageDraft', () => { expect(draft.ts).toBe(event.ts); }); + test('user text event with attachments → UserMessage with attachments', () => { + const event = ev({ + role: 'user', + author: 'user', + content: { kind: 'text', text: 'see attached', attachments: [attachment] }, + refs: { storedMessageId: 'u-attach' }, + }); + const draft = runtimeEventToStoredMessageDraft(event); + expect(draft).not.toBeNull(); + if (!draft || draft.type !== 'user') return; + expect(draft.attachments).toEqual([attachment]); + }); + test('model text event with modelId → AssistantMessage', () => { const event = ev({ role: 'model', @@ -329,6 +378,24 @@ describe('runtimeEventToStoredMessageDraft', () => { expect(runtimeEventToStoredMessageDraft(event)).toBeNull(); }); + test('partial user and model text events → null', () => { + const partialUser = ev({ + partial: true, + role: 'user', + author: 'user', + content: { kind: 'text', text: 'typing...' }, + }); + const partialModel = ev({ + partial: true, + role: 'model', + author: 'agent', + content: { kind: 'text', text: 'streaming...' }, + }); + + expect(runtimeEventToStoredMessageDraft(partialUser)).toBeNull(); + expect(runtimeEventToStoredMessageDraft(partialModel, { modelId: 'gpt-4o' })).toBeNull(); + }); + test('thinking event → null', () => { const event = ev({ role: 'model', diff --git a/packages/runtime/src/__tests__/runtime-runner.test.ts b/packages/runtime/src/__tests__/runtime-runner.test.ts index 91e4cefbd5..6815f1384f 100644 --- a/packages/runtime/src/__tests__/runtime-runner.test.ts +++ b/packages/runtime/src/__tests__/runtime-runner.test.ts @@ -6,6 +6,7 @@ import { type AgentFlowLike, type RuntimeGate, } from '../runtime-runner.js'; +import type { AttachmentRef } from '@maka/core/events'; import type { InvocationContext, InvocationProviders, @@ -40,6 +41,14 @@ function makeRequest(overrides: Partial = {}): InvocationRequ }; } +const attachment: AttachmentRef = { + kind: 'image', + name: 'chart.png', + mimeType: 'image/png', + bytes: 123, + ref: { kind: 'session_file', sessionId: 'sess-1', relativePath: 'attachments/chart.png' }, +}; + /** * Fake flow that runs a script to produce its events. The script receives * the InvocationContext so events can line up with the invocation spine. @@ -130,13 +139,16 @@ describe('RuntimeRunner', () => { test('initial user RuntimeEvent is emitted before any flow event', async () => { const providers = makeProviders(); - const flow = new ScriptFlow((ctx) => [flowTextEvent(ctx, 'hello')]); + const flow = new ScriptFlow((ctx) => [ + flowTextEvent(ctx, 'hello'), + flowTerminalEvent(ctx, 'completed'), + ]); const runner = new RuntimeRunner({ flow, providers }); const result = await runner.run(makeRequest({ text: 'ping' })); expect(result.status).toBe('completed'); - expect(result.events).toHaveLength(2); + expect(result.events).toHaveLength(3); const userEvent = result.events[0]!; expect(userEvent.role).toBe('user'); @@ -151,6 +163,20 @@ describe('RuntimeRunner', () => { expect(result.events[1]!.author).toBe('agent'); }); + test('a flow that exhausts without a terminal event maps to a failed result', async () => { + const providers = makeProviders(); + const flow = new ScriptFlow((ctx) => [flowTextEvent(ctx, 'hello')]); + const runner = new RuntimeRunner({ flow, providers }); + + const result = await runner.run(makeRequest()); + + expect(result.status).toBe('failed'); + expect(result.failure?.class).toBe('missing_terminal_event'); + expect(result.events).toHaveLength(2); + expect(result.events[0]!.author).toBe('user'); + expect(result.events[1]!.author).toBe('agent'); + }); + test('a terminal event ends the result and stops collecting flow events', async () => { const providers = makeProviders(); const flow = new ScriptFlow((ctx) => [ @@ -312,4 +338,57 @@ describe('RuntimeRunner', () => { // Providers are shared, so a fresh id from ctx is unique against runId. expect(ctx.newId() !== ctx.runId).toBe(true); }); + + test('flow receives normalized FlowInput with context default and attachments preserved', async () => { + const providers = makeProviders(); + const context = [ + { + type: 'user' as const, + id: 'u-prev', + turnId: 'prev-turn', + ts: 1, + text: 'previous', + }, + ]; + let seenInput: Parameters[1] | undefined; + const flow: AgentFlowLike = { + async *run(ctx, input) { + seenInput = input; + yield flowTerminalEvent(ctx, 'completed'); + }, + }; + const runner = new RuntimeRunner({ flow, providers }); + + const result = await runner.run( + makeRequest({ text: 'with file', context, attachments: [attachment] }), + ); + + expect(result.status).toBe('completed'); + expect(seenInput).toEqual({ + text: 'with file', + context, + attachments: [attachment], + }); + expect(result.events[0]!.content).toEqual({ + kind: 'text', + text: 'with file', + attachments: [attachment], + }); + }); + + test('flow input context defaults to an empty array', async () => { + const providers = makeProviders(); + let seenInput: Parameters[1] | undefined; + const flow: AgentFlowLike = { + async *run(ctx, input) { + seenInput = input; + yield flowTerminalEvent(ctx, 'completed'); + }, + }; + const runner = new RuntimeRunner({ flow, providers }); + + await runner.run(makeRequest()); + + expect(seenInput?.context).toEqual([]); + }); }); diff --git a/packages/runtime/src/ai-sdk-flow.ts b/packages/runtime/src/ai-sdk-flow.ts index a8f8b547a7..bf18444275 100644 --- a/packages/runtime/src/ai-sdk-flow.ts +++ b/packages/runtime/src/ai-sdk-flow.ts @@ -20,20 +20,20 @@ * - `run(ctx, input)`: drive the wrapped backend and emit `RuntimeEvent`s. * - `mapSessionEventToRuntimeEvent`: a documented, testable placeholder * mapping from the existing `SessionEvent` union onto `RuntimeEvent`. + * - coalesce duplicate terminal backend facts (e.g. `abort` followed by + * trailing `complete(user_stop)`) so the AgentFlow contract stays at + * exactly one terminal RuntimeEvent. * - control surface (`stop` / `respondToPermission` / `dispose`): delegate * to the wrapped backend so current control semantics are preserved. * * What this adapter deliberately does NOT do: * - rewrite or fork `AiSdkBackend.send()`; - * - coalesce the backend's `abort` + trailing `complete` into one event - * (the adapter is faithful to the source stream; coalescing is a - * runner/projection concern); * - own model-history projection (Phase 7) or tool-event actions (Phase 5). */ import type { CompleteEvent, SessionEvent } from '@maka/core/events'; import type { PermissionDecision } from '@maka/core/backend-types'; -import type { RuntimeEvent, RuntimeEventStatus } from '@maka/core/runtime-event'; +import { isTerminalRuntimeEvent, type RuntimeEvent, type RuntimeEventStatus } from '@maka/core/runtime-event'; import type { AgentBackend } from './ai-sdk-backend.js'; import { @@ -428,6 +428,7 @@ export class AiSdkFlow implements AgentFlow, AgentFlowControl { } const memory = createSessionEventMapMemory(); + let terminalEmitted = false; try { for await (const sessionEvent of this.backend.send({ turnId: ctx.turnId, @@ -435,7 +436,14 @@ export class AiSdkFlow implements AgentFlow, AgentFlowControl { ...(input.attachments !== undefined ? { attachments: input.attachments } : {}), context: input.context, })) { - yield mapSessionEventToRuntimeEvent(sessionEvent, ctx, memory); + const runtimeEvent = mapSessionEventToRuntimeEvent(sessionEvent, ctx, memory); + if (isTerminalRuntimeEvent(runtimeEvent)) { + if (terminalEmitted) continue; + terminalEmitted = true; + yield runtimeEvent; + break; + } + yield runtimeEvent; } } finally { if (abortSignal && onAbort) { diff --git a/packages/runtime/src/invocation-context.ts b/packages/runtime/src/invocation-context.ts index a1fe6b6972..10ecde4e9c 100644 --- a/packages/runtime/src/invocation-context.ts +++ b/packages/runtime/src/invocation-context.ts @@ -15,7 +15,9 @@ * minted inside a flow stay 1:1 with the invocation that produced them. */ +import type { AttachmentRef } from '@maka/core/events'; import type { RuntimeEvent, RuntimeEventStatus } from '@maka/core/runtime-event'; +import type { StoredMessage } from '@maka/core/session'; // ============================================================================ // InvocationSource @@ -67,6 +69,14 @@ export interface InvocationRequest { sessionId: string; turnId: string; text: string; + /** Optional attachments bound to this user turn. */ + attachments?: AttachmentRef[]; + /** + * Prior conversation history resolved by the caller/gate. RuntimeRunner + * passes this to AgentFlow as `context`, defaulting to [] so flows never + * receive an undefined model-history input. + */ + context?: StoredMessage[]; source: InvocationSource; /** Optional branch/agent lane; forwarded onto every emitted event. */ branch?: string; diff --git a/packages/runtime/src/runtime-event-adapters.ts b/packages/runtime/src/runtime-event-adapters.ts index c481e64a35..3b9f0b8c01 100644 --- a/packages/runtime/src/runtime-event-adapters.ts +++ b/packages/runtime/src/runtime-event-adapters.ts @@ -104,7 +104,13 @@ export function storedMessageToRuntimeEvent( partial: false, role: 'user', author: 'user', - content: { kind: 'text', text: message.text }, + content: { + kind: 'text', + text: message.text, + ...(message.attachments !== undefined && message.attachments.length > 0 + ? { attachments: message.attachments } + : {}), + }, refs: { storedMessageId: message.id }, }; @@ -231,6 +237,7 @@ export function runtimeEventToStoredMessageDraft( event: RuntimeEvent, options: RuntimeEventToDraftOptions = {}, ): StoredMessage | null { + if (event.partial) return null; const newId = options.newId ?? (() => createRuntimeEventId('msg')); const content = event.content; if (!content) return null; @@ -242,6 +249,9 @@ export function runtimeEventToStoredMessageDraft( turnId: event.turnId, ts: event.ts, text: content.text, + ...(content.attachments !== undefined && content.attachments.length > 0 + ? { attachments: content.attachments } + : {}), }; return draft; } diff --git a/packages/runtime/src/runtime-runner.ts b/packages/runtime/src/runtime-runner.ts index e5b00fbef0..c0d5f8a6d8 100644 --- a/packages/runtime/src/runtime-runner.ts +++ b/packages/runtime/src/runtime-runner.ts @@ -35,6 +35,7 @@ import type { InvocationResultStatus, } from './invocation-context.js'; import { createDefaultInvocationProviders } from './invocation-context.js'; +import type { FlowInput } from './agent-flow.js'; // ============================================================================ // RuntimeGate — narrow preflight seam @@ -84,7 +85,7 @@ export function runtimeGateFromCallback( * node's public surface. */ export interface AgentFlowLike { - run(ctx: InvocationContext, request: InvocationRequest): AsyncIterable; + run(ctx: InvocationContext, input: FlowInput): AsyncIterable; } // ============================================================================ @@ -184,16 +185,19 @@ export class RuntimeRunner { // 4. Emit the initial user RuntimeEvent before any flow event. 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'. let failure: InvocationFailure | undefined; + let terminalSeen = false; try { - for await (const ev of this.flow.run(ctx, request)) { + for await (const ev of this.flow.run(ctx, flowInput)) { events.push(ev); if (isTerminalRuntimeEvent(ev)) { + terminalSeen = true; failure = failureFromTerminalEvent(ev); break; } @@ -204,6 +208,12 @@ export class RuntimeRunner { ...(error instanceof Error && error.message ? { message: error.message } : {}), }; } + if (!failure && !terminalSeen) { + failure = { + class: 'missing_terminal_event', + message: 'flow exhausted without a terminal RuntimeEvent', + }; + } const status: InvocationResultStatus = failure ? 'failed' : 'completed'; return this.buildResult({ @@ -258,7 +268,22 @@ function buildUserEvent(ctx: InvocationContext, request: InvocationRequest): Run partial: false, role: 'user', author: 'user', - content: { kind: 'text', text: request.text }, + content: { + kind: 'text', + text: request.text, + ...(request.attachments !== undefined && request.attachments.length > 0 + ? { attachments: request.attachments } + : {}), + }, + }; +} + +function buildFlowInput(request: InvocationRequest): FlowInput { + return { + text: request.text, + context: request.context ?? [], + ...(request.attachments !== undefined ? { attachments: request.attachments } : {}), + ...(request.abortSignal ? { abortSignal: request.abortSignal } : {}), }; }