diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 874a794bb..218383d3f 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -194,7 +194,10 @@ stable Link Code identity: each start/resume appends a *run* that the adapter's cold (stopped) sessions, `session.resume` wakes one under the same id, and `session.import` registers a provider-local history session as a cold record. Transcripts are not copied — they stay in provider-local history and are read back -through the history contract. +through the history contract. When a settled turn cannot be reconstructed, its complete +start-to-stop interval may still render from the bounded live journal, identified by turn, +run, and epoch. Once that interval is unavailable, the host retains the prompt and renders +an unavailable-output placeholder; journal replay never creates provider checkpoints. ### Engine runtime ownership and composition diff --git a/packages/client/workbench/src/mock/__tests__/dev-mock-conversation.test.ts b/packages/client/workbench/src/mock/__tests__/dev-mock-conversation.test.ts new file mode 100644 index 000000000..57edd48d6 --- /dev/null +++ b/packages/client/workbench/src/mock/__tests__/dev-mock-conversation.test.ts @@ -0,0 +1,137 @@ +import type { ValidatedWireMessage, WirePayload } from '@linkcode/schema'; +import { OperationIdSchema, SessionIdSchema } from '@linkcode/schema'; +import type { Transport } from '@linkcode/transport'; +import { createWireMessage } from '@linkcode/transport'; +import { noop } from 'foxts/noop'; +import { wait } from 'foxts/wait'; +import { describe, expect, it } from 'vitest'; +import { DevMockHost } from '../dev-mock-host'; + +function createHost() { + const sent: WirePayload[] = []; + let handler: ((msg: ValidatedWireMessage) => void) | null = null; + const transport: Transport = { + connect: () => Promise.resolve(), + send(msg: ValidatedWireMessage) { + sent.push(msg.payload); + }, + onMessage(cb) { + handler = cb; + return noop; + }, + onClose: () => noop, + close: noop, + }; + const host = new DevMockHost(transport); + host.start(); + + async function request(payload: WirePayload, replyTo: string): Promise { + if (!handler) throw new Error('mock host not subscribed'); + handler(createWireMessage(payload)); + for (let i = 0; i < 100; i++) { + // eslint-disable-next-line no-await-in-loop -- polling for the mock's latency-delayed reply. + await wait(50); + const reply = sent.find((p) => 'replyTo' in p && p.replyTo === replyTo); + if (reply) return reply; + } + throw new Error(`no reply for ${replyTo}`); + } + + return { sent, request }; +} + +describe('dev mock host conversation parity', () => { + it('answers turn.submit, graph.get, and read coherently', async () => { + const { request } = createHost(); + const started = await request( + { kind: 'session.start', clientReqId: 'r1', opts: { kind: 'claude-code', cwd: '/mock' } }, + 'r1', + ); + if (started.kind !== 'session.started') throw new Error('session did not start'); + const sessionId = started.sessionId; + + const submitted = await request( + { + kind: 'turn.submit', + clientReqId: 's1', + sessionId, + operationId: OperationIdSchema.parse('op-mock-1'), + input: { type: 'shell-command', command: 'ls' }, + }, + 's1', + ); + if (submitted.kind !== 'turn.submitted') throw new Error('turn was not submitted'); + + const graph = await request( + { kind: 'conversation.graph.get', clientReqId: 'g1', sessionId }, + 'g1', + ); + if (graph.kind !== 'conversation.graph.result') throw new Error('no graph result'); + expect(graph.turns).toHaveLength(1); + expect(graph.activeLeafTurnId).toBe(submitted.turnId); + expect(graph.turns[0]).toMatchObject({ + turnId: submitted.turnId, + parentTurnId: null, + siblingOrdinal: 1, + state: 'completed', + inputSummary: '$ ls', + }); + + const read = await request({ kind: 'conversation.read', clientReqId: 'c1', sessionId }, 'c1'); + if (read.kind !== 'conversation.read.result') throw new Error('no read result'); + expect(read.watermark).toBeDefined(); + expect(read.cursor).toBeUndefined(); + expect(read.events).toHaveLength(2); + const [userRow, placeholder] = read.events; + if (!('event' in userRow) || userRow.event.type !== 'user-message') { + throw new Error('expected a user row first'); + } + expect(userRow.event.content).toEqual([{ type: 'text', text: '$ ls' }]); + // Deterministic like the daemon: a re-read converges on the same row identity. + expect(userRow.event.messageId).toBe(`msg-${submitted.turnId}`); + expect(placeholder).toMatchObject({ + type: 'history-unavailable', + turnId: submitted.turnId, + }); + }, 15000); + + it('fails loudly on parameters it would otherwise ignore', async () => { + const { request } = createHost(); + const started = await request( + { kind: 'session.start', clientReqId: 'r1', opts: { kind: 'claude-code', cwd: '/mock' } }, + 'r1', + ); + if (started.kind !== 'session.started') throw new Error('session did not start'); + const sessionId = started.sessionId; + + const pagedRead = await request( + { kind: 'conversation.read', clientReqId: 'c-paged', sessionId, cursor: '1' }, + 'c-paged', + ); + expect(pagedRead.kind).toBe('request.failed'); + + const parentSubmit = await request( + { + kind: 'turn.submit', + clientReqId: 's-parent', + sessionId, + operationId: OperationIdSchema.parse('op-mock-parent'), + input: { type: 'shell-command', command: 'ls' }, + parentTurnId: null, + expectedGraphRevision: 0, + }, + 's-parent', + ); + expect(parentSubmit.kind).toBe('request.failed'); + }, 15000); + + it('fails loudly for conversation reads on unknown sessions', async () => { + const { request } = createHost(); + const unknown = SessionIdSchema.parse('mock-sess-missing'); + const reply = await request( + { kind: 'conversation.read', clientReqId: 'c-x', sessionId: unknown }, + 'c-x', + ); + expect(reply.kind).toBe('request.failed'); + }, 15000); +}); diff --git a/packages/client/workbench/src/mock/dev-mock-host.ts b/packages/client/workbench/src/mock/dev-mock-host.ts index 17e6c3176..6ba74995e 100644 --- a/packages/client/workbench/src/mock/dev-mock-host.ts +++ b/packages/client/workbench/src/mock/dev-mock-host.ts @@ -7,6 +7,8 @@ import type { AgentKind, AgentRuntimes, ContentBlock, + ConversationGraphTurn, + ConversationReadItem, CustomMcpServer, CustomMcpServerPatchOp, CustomMcpServerPublic, @@ -17,8 +19,10 @@ import type { MessageId, PermissionOutcome, Plugin, + PromptId, ProvidersConfig, QuestionOutcome, + RunId, SessionId, SessionInfo, SessionResource, @@ -28,6 +32,8 @@ import type { TerminalMetadata, TerminalReplayEvent, ToolCall, + TurnId, + TurnSubmitInput, WireMessage, WirePayload, WorkspaceId, @@ -125,6 +131,8 @@ interface MockSession extends SessionInfo { effort?: EffortLevel; /** Bumped by cancel/stop so an in-flight prompt turn knows to bail out. */ epoch: number; + /** Minimal turn tree: one lineage appended per `turn.submit` (showcase parity). */ + graphTurns: MockTurn[]; showcase?: boolean; showcaseSeeded?: boolean; longThread?: boolean; @@ -132,6 +140,11 @@ interface MockSession extends SessionInfo { terminalId?: string; } +interface MockTurn { + graph: ConversationGraphTurn; + content: ContentBlock[]; +} + interface PendingPermission { sessionId: SessionId; /** The pending snapshot the ask was raised for; the response re-emits it resolved. */ @@ -202,6 +215,7 @@ export class DevMockHost { private workspaceSeq = 0; private terminalSeq = 0; private resourceSeq = 0; + private turnSeq = 0; /** Assets a mock `asset.ensure` has "installed"; list/runtime replies reflect it afterwards. */ private readonly installedAssets = new Set(); private readonly cleanGitWorkspaces = new Set(); @@ -355,6 +369,68 @@ export class DevMockHost { case 'agent.input': await this.handleInput(p.clientReqId, p.sessionId, p.input); break; + case 'turn.submit': + await wait(CONTROL_LATENCY_MS); + await this.submitTurn(p); + break; + case 'conversation.graph.get': { + await wait(CONTROL_LATENCY_MS); + const session = this.sessions.get(p.sessionId); + if (!session) { + this.sendFailure(p.clientReqId, `Unknown session: ${p.sessionId}`); + break; + } + const leaf = session.graphTurns.at(-1); + this.send({ + kind: 'conversation.graph.result', + replyTo: p.clientReqId, + sessionId: p.sessionId, + graphRevision: session.graphTurns.length, + ...(leaf !== undefined && { activeLeafTurnId: leaf.graph.turnId }), + turns: session.graphTurns.map((turn) => structuredClone(turn.graph)), + }); + break; + } + case 'conversation.read': { + await wait(CONTROL_LATENCY_MS); + const session = this.sessions.get(p.sessionId); + if (!session) { + this.sendFailure(p.clientReqId, `Unknown session: ${p.sessionId}`); + break; + } + // Fail loudly on parameters the mock would silently ignore. + if (p.leafTurnId !== undefined || p.cursor !== undefined || p.limit !== undefined) { + this.sendFailure(p.clientReqId, 'Dev mock host does not support read paging yet.'); + break; + } + const leaf = session.graphTurns.at(-1); + // Minimal parity: host user rows + the no-history placeholder, one final page. The mock + // has no provider transcripts, so this mirrors the daemon's prompt-only fallback. + const events = session.graphTurns.flatMap(({ graph, content }): ConversationReadItem[] => [ + { + turnId: graph.turnId, + runId: graph.runId, + ts: graph.createdAt, + event: { + type: 'user-message', + // Deterministic like the daemon: re-reads must converge on one row per turn. + messageId: `msg-${graph.turnId}` as MessageId, + content: structuredClone(content), + }, + }, + { type: 'history-unavailable', turnId: graph.turnId, runId: graph.runId }, + ]); + this.send({ + kind: 'conversation.read.result', + replyTo: p.clientReqId, + sessionId: p.sessionId, + graphRevision: session.graphTurns.length, + ...(leaf !== undefined && { leafTurnId: leaf.graph.turnId }), + watermark: { epoch: 0, seq: 0 }, + events, + }); + break; + } case 'resource.list': await wait(CONTROL_LATENCY_MS); this.send({ @@ -757,7 +833,7 @@ export class DevMockHost { } private addSession( - init: Omit & { + init: Omit & { status: SessionStatus; origin?: SessionInfo['origin']; }, @@ -774,6 +850,7 @@ export class DevMockHost { sessionId: this.nextSessionId(), origin: origin ?? { type: 'created' }, epoch: 0, + graphTurns: [], }; this.sessions.set(session.sessionId, session); return session; @@ -1235,11 +1312,74 @@ export class DevMockHost { this.sendSuccess(replyTo); } + /** Appends a completed turn to the mock graph and streams the scripted reply — the daemon's + * submit saga reduced to showcase parity. */ + private async submitTurn(p: Extract): Promise { + const session = this.sessions.get(p.sessionId); + if (!session) { + this.sendFailure(p.clientReqId, `Unknown session: ${p.sessionId}`); + return; + } + if (session.status === 'stopped') { + this.sendFailure(p.clientReqId, `Session is stopped, resume it first: ${p.sessionId}`); + return; + } + if (session.status === 'running') { + this.sendFailure(p.clientReqId, `Session is busy: ${p.sessionId}`); + return; + } + // Fail loudly on parameters the mock would silently ignore (plain sends only). + if (p.parentTurnId !== undefined || p.expectedGraphRevision !== undefined) { + this.sendFailure(p.clientReqId, 'Dev mock host does not support explicit-parent submits.'); + return; + } + const content = turnSubmitContent(p.input); + this.turnSeq += 1; + const turnId = `turn-mock-${this.turnSeq.toString(36)}` as TurnId; + const parent = session.graphTurns.at(-1); + const graph: ConversationGraphTurn = { + turnId, + sessionId: p.sessionId, + parentTurnId: parent?.graph.turnId ?? null, + siblingOrdinal: 1, + input: + p.input.type === 'prompt' + ? { type: 'prompt', promptId: `prompt-mock-${this.turnSeq.toString(36)}` as PromptId } + : p.input, + runId: `run-mock-${this.turnSeq.toString(36)}` as RunId, + state: 'completed', + createdAt: Date.now(), + inputSummary: promptText(content).slice(0, 140), + }; + session.graphTurns.push({ graph, content }); + this.send({ kind: 'turn.submitted', replyTo: p.clientReqId, turnId }); + if (p.input.type === 'prompt') { + const result = await this.streamMockTurn(session, content); + if (!result.ok) graph.state = 'failed'; + return; + } + // Command/shell turns just echo — the mock has no directive execution behind turn.submit. + this.emit(p.sessionId, { + type: 'user-message', + messageId: this.nextMessageId('mock-user'), + content, + }); + } + private async prompt( replyTo: string, session: MockSession, content: ContentBlock[], ): Promise { + const result = await this.streamMockTurn(session, content); + if (result.ok) this.sendSuccess(replyTo); + else this.sendFailure(replyTo, result.message, { reportedInConversation: true }); + } + + private async streamMockTurn( + session: MockSession, + content: ContentBlock[], + ): Promise<{ ok: true } | { ok: false; message: string }> { const text = promptText(content); if (text && !session.title) session.title = text.slice(0, 80); session.status = 'running'; @@ -1258,10 +1398,7 @@ export class DevMockHost { return session.epoch !== epoch; }; - if (await cancelledAfter(200)) { - this.sendSuccess(replyTo); - return; - } + if (await cancelledAfter(200)) return { ok: true }; const thoughtId = this.nextMessageId('mock-thought'); this.emit(session.sessionId, { type: 'agent-thought-chunk', @@ -1270,10 +1407,7 @@ export class DevMockHost { }); if (text.toLowerCase() === FAIL_PROMPT) { - if (await cancelledAfter(200)) { - this.sendSuccess(replyTo); - return; - } + if (await cancelledAfter(200)) return { ok: true }; const message = `Mock failure requested via the "${FAIL_PROMPT}" prompt.`; this.emit(session.sessionId, { type: 'error', @@ -1283,8 +1417,7 @@ export class DevMockHost { }); session.status = 'idle'; this.emit(session.sessionId, { type: 'status', status: 'idle' }); - this.sendFailure(replyTo, message, { reportedInConversation: true }); - return; + return { ok: false, message }; } const messageId = this.nextMessageId('mock-message'); @@ -1293,10 +1426,7 @@ export class DevMockHost { if (chunks != null) { for (let i = 0, len = chunks.length; i < len; i++) { // eslint-disable-next-line no-await-in-loop -- word-by-word streaming: chunks are paced sequentially by design. - if (await cancelledAfter(CHUNK_LATENCY_MS)) { - this.sendSuccess(replyTo); - return; - } + if (await cancelledAfter(CHUNK_LATENCY_MS)) return { ok: true }; this.emit(session.sessionId, { type: 'agent-message-chunk', messageId, @@ -1314,7 +1444,7 @@ export class DevMockHost { this.emit(session.sessionId, { type: 'stop', stopReason: 'end_turn' }); session.status = 'idle'; this.emit(session.sessionId, { type: 'status', status: 'idle' }); - this.sendSuccess(replyTo); + return { ok: true }; } /** Emitted in one burst, not streamed: this transcript exists to be long, not to look live. */ @@ -1706,6 +1836,21 @@ export class DevMockHost { } } +function turnSubmitContent(input: TurnSubmitInput): ContentBlock[] { + switch (input.type) { + case 'prompt': + return input.blocks.flatMap((block) => + block.type === 'text' ? [textBlock(block.text)] : [], + ); + case 'command': + return [textBlock(`/${input.name}${input.arguments ? ` ${input.arguments}` : ''}`)]; + case 'shell-command': + return [textBlock(`$ ${input.command}`)]; + default: + return []; + } +} + function promptText(content: readonly ContentBlock[]): string { return content .reduce((text, block) => { diff --git a/packages/foundation/schema/src/wire/conversation.ts b/packages/foundation/schema/src/wire/conversation.ts index 937169026..a33430160 100644 --- a/packages/foundation/schema/src/wire/conversation.ts +++ b/packages/foundation/schema/src/wire/conversation.ts @@ -39,6 +39,29 @@ export const ConversationEventSchema = z.object({ }); export type ConversationEvent = z.infer; +/** Turn-scoped marker: this turn's provider output is unavailable (no-history harness, compacted + * or deleted transcript, migrated turn, failed provider read) — the host prompt row is all there + * is. The deliberate prompt-only fallback, never a broken graph. */ +export const ConversationPlaceholderSchema = z.object({ + type: z.literal('history-unavailable'), + turnId: TurnIdSchema, + runId: RunIdSchema.optional(), +}); +export type ConversationPlaceholder = z.infer; + +/** One item of a `conversation.read` page. */ +export const ConversationReadItemSchema = z.union([ + ConversationEventSchema, + ConversationPlaceholderSchema, +]); +export type ConversationReadItem = z.infer; + +/** A graph node plus the short input label that renders `← 1/N →` without reading content. */ +export const ConversationGraphTurnSchema = ConversationTurnSchema.extend({ + inputSummary: z.string().optional(), +}); +export type ConversationGraphTurn = z.infer; + /** Conversation-graph wire variants. `turn.submit`'s parent/revision contract: `parentTurnId` * ABSENT = plain send onto the active leaf; `null` = new root lineage; a turn id = edit/continue * under that turn. `expectedGraphRevision` is required iff `parentTurnId` is present (enforced by @@ -70,7 +93,7 @@ export const conversationWireVariants = [ sessionId: SessionIdSchema, graphRevision: z.number().int().nonnegative(), activeLeafTurnId: TurnIdSchema.optional(), - turns: z.array(ConversationTurnSchema), + turns: z.array(ConversationGraphTurnSchema), }), /** Session-scoped broadcast: the graph changed shape or moved its default leaf; clients holding * a stale snapshot revalidate via `conversation.graph.get`. */ @@ -96,9 +119,10 @@ export const conversationWireVariants = [ graphRevision: z.number().int().nonnegative(), /** The leaf the projection was read toward; absent while the session has no turns. */ leafTurnId: TurnIdSchema.optional(), - /** Merge cut for the live event plane; only the final page's watermark is authoritative. */ - watermark: ConversationWatermarkSchema, - events: z.array(ConversationEventSchema), + /** Merge cut for the live event plane. ONLY the final page carries it (and the live tail); + * every non-final page omits it — clients merge against the final page's watermark alone. */ + watermark: ConversationWatermarkSchema.optional(), + events: z.array(ConversationReadItemSchema), cursor: z.string().optional(), }), ] as const; diff --git a/packages/foundation/schema/src/wire/index.ts b/packages/foundation/schema/src/wire/index.ts index f40d9a134..fddcdcd2a 100644 --- a/packages/foundation/schema/src/wire/index.ts +++ b/packages/foundation/schema/src/wire/index.ts @@ -1,6 +1,12 @@ export { type ConversationEvent, ConversationEventSchema, + type ConversationGraphTurn, + ConversationGraphTurnSchema, + type ConversationPlaceholder, + ConversationPlaceholderSchema, + type ConversationReadItem, + ConversationReadItemSchema, type TurnSubmitInput, TurnSubmitInputSchema, } from './conversation'; diff --git a/packages/foundation/schema/tests/contract/wire/conversation.test.ts b/packages/foundation/schema/tests/contract/wire/conversation.test.ts index 242b62846..1cf8e26b7 100644 --- a/packages/foundation/schema/tests/contract/wire/conversation.test.ts +++ b/packages/foundation/schema/tests/contract/wire/conversation.test.ts @@ -105,6 +105,63 @@ describe('conversation read/graph frames', () => { ).toBe(true); }); + it('accepts an inputSummary on a graph turn', () => { + expect( + parses({ + kind: 'conversation.graph.result', + replyTo: 'request-1', + sessionId: 'session-1', + graphRevision: 1, + turns: [ + { + turnId: 'turn-1', + sessionId: 'session-1', + parentTurnId: null, + siblingOrdinal: 1, + input: { type: 'prompt', promptId: 'prompt-1' }, + runId: 'run-1', + state: 'completed', + createdAt: 1, + inputSummary: 'hello there', + }, + ], + }), + ).toBe(true); + }); + + it('accepts a non-final page: no watermark, cursor set, placeholder items allowed', () => { + expect( + parses({ + kind: 'conversation.read.result', + replyTo: 'request-1', + sessionId: 'session-1', + graphRevision: 2, + leafTurnId: 'turn-2', + events: [ + { + turnId: 'turn-1', + runId: 'run-1', + event: { type: 'user-message', messageId: 'm-1', content: [] }, + }, + { type: 'history-unavailable', turnId: 'turn-1', runId: 'run-1' }, + ], + cursor: '2', + }), + ).toBe(true); + }); + + it('rejects a read item that is neither an event nor a placeholder', () => { + expect( + parses({ + kind: 'conversation.read.result', + replyTo: 'request-1', + sessionId: 'session-1', + graphRevision: 0, + events: [{ turnId: 'turn-1' }], + }), + ).toBe(false); + }); + it('scopes conversation.graph.changed to its session', () => { const payload = WirePayloadSchema.parse({ kind: 'conversation.graph.changed', diff --git a/packages/host/agent-adapter/AGENTS.md b/packages/host/agent-adapter/AGENTS.md index a796bcb8f..fd8a853fe 100644 --- a/packages/host/agent-adapter/AGENTS.md +++ b/packages/host/agent-adapter/AGENTS.md @@ -23,6 +23,13 @@ retain the daemon environment. Codex config/history paths use that same environm app-server treats it as a complete replacement. Resolution fails loudly instead of falling back to launchd's PATH. +Claude history reads resolve the project environment and the recorded run's account configuration. +SDK history calls for another `CLAUDE_CONFIG_DIR` run in a disposable worker: never change the +daemon's process environment around an async SDK call. Raw transcript supplements use the same +root. A missing native session rejects a direct history read instead of returning an empty seed. +Changing or deleting the account configuration can still make older history inaccessible; this +does not relocate transcripts or create a second transcript store. + Pins as of 2026-07 (package.json ranges are caret; the lockfile is the real pin): | agent | JS package | version | diff --git a/packages/host/agent-adapter/src/__tests__/claude-code-effort.test.ts b/packages/host/agent-adapter/src/__tests__/claude-code-effort.test.ts index 3e8331601..0117a94ee 100644 --- a/packages/host/agent-adapter/src/__tests__/claude-code-effort.test.ts +++ b/packages/host/agent-adapter/src/__tests__/claude-code-effort.test.ts @@ -51,6 +51,7 @@ class FakeQuery { }); private readonly buffered: Array = []; private waiting: ((msg: WireMessage | null) => void) | null = null; + private failure: Error | undefined; constructor(input: QueryInput) { this.options = input.options; @@ -70,6 +71,11 @@ class FakeQuery { } } + fail(error: Error): void { + this.failure = error; + this.push(null); + } + async *[Symbol.asyncIterator](): AsyncGenerator { while (true) { const next = @@ -79,7 +85,10 @@ class FakeQuery { await new Promise((resolve) => { this.waiting = resolve; }); - if (next === null) return; + if (next === null) { + if (this.failure) throw this.failure; + return; + } yield next; } } @@ -955,5 +964,29 @@ describe('ClaudeCodeAdapter turn lifecycle', () => { 'Claude failed (error_during_execution, turn_setup_failed): MCP server failed; Connection refused', recoverable: true, }); + events.length = 0; + queries[0].fail( + new Error('Claude Code returned an error result: MCP server failed; Connection refused'), + ); + await waitIdle(events); + expect(events.filter((event) => event.type === 'error')).toEqual([]); + }); + + it('still reports a stream failure in a new turn after a failed result', async () => { + const { adapter, events } = await makeAdapter(); + await prompt(adapter); + queries[0].push({ type: 'result', subtype: 'error_during_execution', errors: ['failed turn'] }); + await waitIdle(events); + await prompt(adapter); + events.length = 0; + queries[0].fail(new Error('connection lost')); + await waitIdle(events); + expect(events.filter((event) => event.type === 'error')).toEqual([ + { + type: 'error', + message: 'claude-code: query failed (Error: connection lost)', + recoverable: true, + }, + ]); }); }); diff --git a/packages/host/agent-adapter/src/__tests__/claude-history-root.test.ts b/packages/host/agent-adapter/src/__tests__/claude-history-root.test.ts new file mode 100644 index 000000000..065213680 --- /dev/null +++ b/packages/host/agent-adapter/src/__tests__/claude-history-root.test.ts @@ -0,0 +1,90 @@ +import { randomUUID } from 'node:crypto'; +import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import * as sdk from '@anthropic-ai/claude-agent-sdk'; +import { afterEach, describe, expect, it } from 'vitest'; +import { asHistoryId } from '../history-util'; +import { ClaudeCodeAdapter } from '../native/claude-code'; +import { claudeHistorySdk } from '../native/claude-history-sdk'; + +const roots: string[] = []; +afterEach(async () => { + await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true }))); +}); + +async function transcript(sessionId: string, answer: string) { + const root = await mkdtemp(path.join(tmpdir(), 'linkcode-claude-root-')); + roots.push(root); + const project = path.join(root, 'projects', 'fixture'); + await mkdir(project, { recursive: true }); + const userId = randomUUID(); + const assistantId = randomUUID(); + const rows = [ + { + type: 'user', + uuid: userId, + parentUuid: null, + sessionId, + timestamp: '2026-09-01T00:00:00Z', + message: { role: 'user', content: 'hello' }, + }, + { + type: 'assistant', + uuid: assistantId, + parentUuid: userId, + sessionId, + timestamp: '2026-09-01T00:00:01Z', + message: { + id: assistantId, + role: 'assistant', + model: 'claude-sonnet-5', + content: [{ type: 'text', text: answer }], + }, + }, + ]; + await writeFile( + path.join(project, `${sessionId}.jsonl`), + rows.map((row) => JSON.stringify(row)).join('\n') + '\n', + ); + return { root, assistantId }; +} + +describe('Claude native history config roots', () => { + it('reads concurrent account roots without changing the daemon environment', async () => { + const historyId = asHistoryId(randomUUID()); + const before = process.env.CLAUDE_CONFIG_DIR; + const [a, b] = await Promise.all([ + transcript(historyId, 'account A'), + transcript(historyId, 'account B'), + ]); + const results = await Promise.all( + [a, b].map(({ root }) => + new ClaudeCodeAdapter().readHistory({ + historyId, + config: { extraEnv: { CLAUDE_CONFIG_DIR: root } }, + }), + ), + ); + expect( + results.map((result) => + result.events.flatMap(({ event }) => (event.type === 'agent-message' ? event.content : [])), + ), + ).toEqual([[{ type: 'text', text: 'account A' }], [{ type: 'text', text: 'account B' }]]); + expect(process.env.CLAUDE_CONFIG_DIR).toBe(before); + }); + + it('forks in the source config root and reports a missing history explicitly', async () => { + const historyId = asHistoryId(randomUUID()); + const { root, assistantId } = await transcript(historyId, 'retained answer'); + const scoped = claudeHistorySdk(sdk, root); + const child = await scoped.forkSession(historyId, { upToMessageId: assistantId }); + expect(await scoped.getSessionMessages(child.sessionId)).toHaveLength(2); + await expect( + new ClaudeCodeAdapter().readHistory({ + historyId: asHistoryId(randomUUID()), + config: { extraEnv: { CLAUDE_CONFIG_DIR: root } }, + }), + ).rejects.toThrow('native history'); + }); +}); diff --git a/packages/host/agent-adapter/src/adapter.ts b/packages/host/agent-adapter/src/adapter.ts index 746b4ed55..1e3c99c31 100644 --- a/packages/host/agent-adapter/src/adapter.ts +++ b/packages/host/agent-adapter/src/adapter.ts @@ -20,6 +20,8 @@ export type AgentStartCatalogOptions = Partial; type AssistantMessage = AssistantSDKMessage['message']; @@ -543,7 +549,14 @@ export class ClaudeCodeAdapter extends BaseAgentAdapter { '@anthropic-ai/claude-agent-sdk', () => import('@anthropic-ai/claude-agent-sdk'), ); - this.getSessionInfo = Object.hasOwn(sdk, 'getSessionInfo') ? sdk.getSessionInfo : undefined; + const root = claudeConfigDir( + claudeCodeEnv(this.processEnvironment, readAgentCredential(opts.config)) ?? + this.processEnvironment, + opts.cwd, + ); + this.getSessionInfo = Object.hasOwn(sdk, 'getSessionInfo') + ? claudeHistorySdk(sdk, root).getSessionInfo + : undefined; if (this.resumeFrom && this.getSessionInfo) { try { const info = await this.getSessionInfo(this.resumeFrom, { dir: opts.cwd }); @@ -668,7 +681,8 @@ export class ClaudeCodeAdapter extends BaseAgentAdapter { '@anthropic-ai/claude-agent-sdk', () => import('@anthropic-ai/claude-agent-sdk'), ); - const fork = await mod.forkSession(opts.historyId, { + const root = await this.historyConfigRoot(startOpts); + const fork = await claudeHistorySdk(mod, root).forkSession(opts.historyId, { upToMessageId: predecessor, dir: startOpts.cwd, }); @@ -684,7 +698,8 @@ export class ClaudeCodeAdapter extends BaseAgentAdapter { ); const offset = cursorOffset(opts?.cursor); const limit = boundedLimit(opts?.limit, 50, 200); - const sessions = await mod.listSessions({ + const root = await this.historyConfigRoot(opts ?? {}); + const sessions = await claudeHistorySdk(mod, root).listSessions({ dir: opts?.cwd, limit: limit + 1, offset, @@ -695,11 +710,13 @@ export class ClaudeCodeAdapter extends BaseAgentAdapter { }; } - override async readHistory(opts: AgentHistoryReadOptions): Promise { - const mod = await this.loadSdk( + override async readHistory(opts: AgentHistoryReadContext): Promise { + const sdk = await this.loadSdk( '@anthropic-ai/claude-agent-sdk', () => import('@anthropic-ai/claude-agent-sdk'), ); + const root = await this.historyConfigRoot(opts); + const mod = claudeHistorySdk(sdk, root); const offset = cursorOffset(opts.cursor); const limit = boundedLimit(opts.limit, 1000, 1000); const [info, messages, subagentEvents, supplement] = await Promise.all([ @@ -709,14 +726,22 @@ export class ClaudeCodeAdapter extends BaseAgentAdapter { offset, }), readSubagentTranscripts(mod, opts.historyId, (agentId) => - this.readSubagentPatches(opts.historyId, agentId), + this.readSubagentPatches(opts.historyId, agentId, root), ), // Every page needs the raw transcript: getSessionMessages strips each result row's // structured toolUseResult, so the mapper re-attaches envelopes from here. The compaction // splice below stays first-page-only (the swapped-in summary is the SDK chain's head row). - this.readTranscriptSupplement(opts.historyId), + this.readTranscriptSupplement(opts.historyId, root), ]); const historyId = opts.historyId; + if ( + offset === 0 && + info === undefined && + messages.length === 0 && + supplement.droppedRows.length === 0 + ) { + throw new Error(`claude-code: native history ${historyId} is unavailable in ${root}`); + } const mapper = createClaudeHistoryEventMapper( historyId, supplement.records, @@ -773,16 +798,32 @@ export class ClaudeCodeAdapter extends BaseAgentAdapter { } /** Test seam over the raw transcript probe (see `readClaudeTranscriptSupplement`). */ - protected readTranscriptSupplement(sessionId: string): Promise { - return readClaudeTranscriptSupplement(sessionId); + protected readTranscriptSupplement( + sessionId: string, + root = claudeConfigDir(process.env), + ): Promise { + return readClaudeTranscriptSupplement(sessionId, root); } /** Test seam over the per-subagent transcript probe (see `readSubagentPatches`). */ protected readSubagentPatches( sessionId: string, agentId: string, + root = claudeConfigDir(process.env), ): Promise> { - return readSubagentPatches(sessionId, agentId); + return readSubagentPatches(sessionId, agentId, root); + } + + private async historyConfigRoot(opts: { + cwd?: string; + config?: StartOptions['config']; + }): Promise { + const environment = + opts.cwd === undefined ? process.env : await resolveAgentShellEnvironment(opts.cwd); + return claudeConfigDir( + claudeCodeEnv(environment, readAgentCredential(opts.config)) ?? environment, + opts.cwd, + ); } protected async onPrompt(content: ContentBlock[]): Promise { @@ -928,9 +969,13 @@ export class ClaudeCodeAdapter extends BaseAgentAdapter { * emits. Returns only when the underlying process exits (crash, `close()`, or the CLI quitting). */ private async consume(q: Query): Promise { let streamError: unknown; + let failedResult = false; try { for await (const msg of q) { - if (this.q === q) this.handleMessage(msg); + if (this.q === q) { + this.handleMessage(msg); + if (msg.type === 'result') failedResult = msg.subtype !== 'success'; + } } } catch (err) { streamError = err; @@ -949,7 +994,7 @@ export class ClaudeCodeAdapter extends BaseAgentAdapter { // this once. Without re-arming it, an async spawn failure silently starts a new conversation. this.resumeFrom = this.lastSessionRef; if (!cancelling) { - if (streamError !== undefined) { + if (streamError !== undefined && (!failedResult || interruptedTurn)) { this.emitError( `claude-code: query failed (${extractErrorMessage(streamError) ?? 'unknown error'})`, ); @@ -1934,8 +1979,11 @@ function harvestToolUseResult( * `.jsonl` (the id is unique, so at most one probe succeeds). Any failure degrades to * an empty supplement: history still reads, just without compaction markers or result envelopes. */ -async function readClaudeProjectText(segments: readonly string[]): Promise { - const projectsDir = path.join(homedir(), '.claude', 'projects'); +async function readClaudeProjectText( + segments: readonly string[], + root: string, +): Promise { + const projectsDir = path.join(root, 'projects'); let dirs: string[]; try { dirs = await readdir(projectsDir); @@ -1950,10 +1998,11 @@ async function readClaudeProjectText(segments: readonly string[]): Promise { // The id becomes a filename — refuse anything that could traverse out of the projects dir. if (!SAFE_SESSION_ID.test(sessionId)) return EMPTY_SUPPLEMENT; - const text = await readClaudeProjectText([`${sessionId}.jsonl`]); + const text = await readClaudeProjectText([`${sessionId}.jsonl`], root); return text ? buildClaudeTranscriptSupplement(text.split('\n')) : EMPTY_SUPPLEMENT; } @@ -1971,10 +2020,14 @@ async function readClaudeTranscriptSupplement( async function readSubagentPatches( sessionId: string, agentId: string, + root: string, ): Promise> { // Both ids become path segments. if (!SAFE_SESSION_ID.test(sessionId) || !SAFE_SESSION_ID.test(agentId)) return new Map(); - const text = await readClaudeProjectText([sessionId, 'subagents', `agent-${agentId}.jsonl`]); + const text = await readClaudeProjectText( + [sessionId, 'subagents', `agent-${agentId}.jsonl`], + root, + ); return text ? buildClaudeTranscriptSupplement(text.split('\n')).toolUsePatches : new Map(); } @@ -2001,7 +2054,7 @@ function mapClaudeHistorySession(session: SDKSessionInfo): AgentHistorySession { * stream's parent-linked events. Keyed by that parent id for splicing after the spawn announce. */ async function readSubagentTranscripts( - mod: typeof import('@anthropic-ai/claude-agent-sdk'), + mod: ClaudeHistorySdk, sessionId: string, patchesFor: (agentId: string) => Promise>, ): Promise> { diff --git a/packages/host/agent-adapter/src/native/claude-history-sdk.ts b/packages/host/agent-adapter/src/native/claude-history-sdk.ts new file mode 100644 index 000000000..96e62a0e8 --- /dev/null +++ b/packages/host/agent-adapter/src/native/claude-history-sdk.ts @@ -0,0 +1,57 @@ +import { homedir } from 'node:os'; +import path from 'node:path'; +import { Worker } from 'node:worker_threads'; + +type SDK = typeof import('@anthropic-ai/claude-agent-sdk'); +export type ClaudeHistorySdk = Pick< + SDK, + | 'getSessionInfo' + | 'getSessionMessages' + | 'listSessions' + | 'listSubagents' + | 'getSubagentMessages' + | 'forkSession' +>; + +export function claudeConfigDir(environment: NodeJS.ProcessEnv, cwd = process.cwd()): string { + return path.resolve( + cwd, + environment.CLAUDE_CONFIG_DIR || path.join(environment.HOME || homedir(), '.claude'), + ); +} + +// The SDK captures its config root from process.env. A worker isolates concurrent account roots. +const HISTORY_WORKER = ` +const { parentPort, workerData } = require('node:worker_threads'); +import(workerData.sdk).then(async (sdk) => { + parentPort.postMessage(await sdk[workerData.method](...workerData.args)); +}); +`; + +function callHistory(root: string, method: keyof ClaudeHistorySdk, args: unknown[]): Promise { + const worker = new Worker(HISTORY_WORKER, { + eval: true, + execArgv: [], + env: { ...process.env, CLAUDE_CONFIG_DIR: root }, + workerData: { sdk: import.meta.resolve('@anthropic-ai/claude-agent-sdk'), method, args }, + }); + return new Promise((resolve, reject) => { + worker.once('message', resolve); + worker.once('error', reject); + worker.once('exit', (code) => + reject(new Error(`Claude history worker exited before replying (${code})`)), + ); + }).finally(() => worker.terminate()); +} + +export function claudeHistorySdk(sdk: ClaudeHistorySdk, root: string): ClaudeHistorySdk { + if (root === claudeConfigDir(process.env)) return sdk; + return { + getSessionInfo: (...args) => callHistory(root, 'getSessionInfo', args), + getSessionMessages: (...args) => callHistory(root, 'getSessionMessages', args), + listSessions: (...args) => callHistory(root, 'listSessions', args), + listSubagents: (...args) => callHistory(root, 'listSubagents', args), + getSubagentMessages: (...args) => callHistory(root, 'getSubagentMessages', args), + forkSession: (...args) => callHistory(root, 'forkSession', args), + }; +} diff --git a/packages/host/engine/src/__tests__/conversation-live-journal.test.ts b/packages/host/engine/src/__tests__/conversation-live-journal.test.ts index 775096363..4c3a1ab08 100644 --- a/packages/host/engine/src/__tests__/conversation-live-journal.test.ts +++ b/packages/host/engine/src/__tests__/conversation-live-journal.test.ts @@ -2,14 +2,18 @@ import type { AgentEvent, MessageId, RunId, SessionId } from '@linkcode/schema'; import { compareConversationWatermarks } from '@linkcode/schema'; import { describe, expect, it } from 'vitest'; import type { JournaledEvent } from '../conversation/live-journal'; -import { ConversationLiveJournal, ConversationLiveJournals } from '../conversation/live-journal'; +import { + ConversationLiveJournal, + ConversationLiveJournals, + inflightChunkKey, +} from '../conversation/live-journal'; const runId = 'run-journal' as RunId; -function chunk(text: string): AgentEvent { +function chunk(text: string, messageId = 'msg-1'): AgentEvent { return { type: 'agent-message-chunk', - messageId: 'msg-1' as MessageId, + messageId: messageId as MessageId, content: { type: 'text', text }, }; } @@ -135,6 +139,45 @@ describe('ConversationLiveJournal tailAfter', () => { }); }); +describe('ConversationLiveJournal evicted in-flight streams', () => { + function stampedChunk(seq: number, messageId: string): JournaledEvent { + return { epoch: 1, seq, runId, ts: 1, event: chunk(`c-${seq}`, messageId) }; + } + + it('marks a stream cleared once eviction removes any of its deltas', () => { + const journal = new ConversationLiveJournal(Number.MAX_SAFE_INTEGER, 3); + journal.append(stampedChunk(1, 'msg-a')); + journal.append(stampedChunk(2, 'msg-a')); + journal.append(stampedChunk(3, 'msg-b')); + expect(journal.isChunkCleared('msg-a')).toBe(false); + + // Evicts seq 1 (msg-a's head): its retained tail must never render headless. + journal.append(stampedChunk(4, 'msg-b')); + expect(journal.isChunkCleared('msg-a')).toBe(true); + expect(journal.isChunkCleared('msg-b')).toBe(false); + }); + + it('keys tool-call content chunks by toolCallId and full-state events not at all', () => { + expect( + inflightChunkKey({ + type: 'tool-call-content-chunk', + toolCallId: 'tool-9', + content: { type: 'content', content: { type: 'text', text: 'x' } }, + }), + ).toBe('tool-9'); + expect(inflightChunkKey({ type: 'status', status: 'running' })).toBeUndefined(); + }); + + it('treats every stream as cleared once the key cap overflows', () => { + const journal = new ConversationLiveJournal(Number.MAX_SAFE_INTEGER, 1); + for (let seq = 1; seq <= 1030; seq++) { + journal.append(stampedChunk(seq, `msg-${seq}`)); + } + // 1029 distinct keys evicted > the 1024 cap: even a never-evicted stream reads as cleared. + expect(journal.isChunkCleared('msg-never-appended')).toBe(true); + }); +}); + describe('ConversationLiveJournals registry', () => { it('creates one journal per session and drops it with the live session', () => { const journals = new ConversationLiveJournals(); diff --git a/packages/host/engine/src/__tests__/conversation-projection.test.ts b/packages/host/engine/src/__tests__/conversation-projection.test.ts new file mode 100644 index 000000000..d8f213fd8 --- /dev/null +++ b/packages/host/engine/src/__tests__/conversation-projection.test.ts @@ -0,0 +1,699 @@ +import { asHistoryId } from '@linkcode/agent-adapter'; +import type { + AgentEvent, + AgentHistoryEvent, + AgentHistoryReadOptions, + AgentHistoryReadResult, + ConversationReadItem, + ConversationTurnState, + MessageId, + RunId, + SessionId, + SessionRecord, + TurnId, +} from '@linkcode/schema'; +import type { Transport } from '@linkcode/transport'; +import { Cause, Effect, Exit } from 'effect'; +import { noop } from 'foxts/noop'; +import { describe, expect, it } from 'vitest'; +import { InMemoryConversationStore } from '../conversation/conversation-store'; +import type { JournaledEvent } from '../conversation/live-journal'; +import { ConversationLiveJournals } from '../conversation/live-journal'; +import { ConversationProjectionService, pageReadItems } from '../conversation/projection-service'; +import { ConversationTurnService } from '../conversation/turn-service'; +import { RequestError } from '../failure'; +import { HistoryService } from '../session/history-service'; +import { SessionRecordRegistry } from '../session/session-record-registry'; +import { InMemorySessionStore } from '../session/session-store'; +import { FakeAdapter } from './fixtures/session-harness'; + +const sessionId = 'sess-projection' as SessionId; +const runId = 'run-1' as RunId; + +const OPEN_ASK: AgentEvent = { + type: 'permission-request', + requestId: 'perm-open', + title: 'Run', + subject: { type: 'tool-call', toolCallId: 't1' }, + options: [{ optionId: 'ok', name: 'Allow', kind: 'allow_once' }], +}; + +function chunk(messageId: string, text: string): AgentEvent { + return { + type: 'agent-message-chunk', + messageId: messageId as MessageId, + content: { type: 'text', text }, + }; +} + +function stamped(seq: number, turnId: TurnId, event: AgentEvent): JournaledEvent { + return { epoch: 3, seq, runId, turnId, ts: 1000 + seq, event }; +} + +class CannedHistoryAdapter extends FakeAdapter { + constructor(private readonly events: AgentHistoryEvent[]) { + super(); + } + + override readHistory(opts: AgentHistoryReadOptions): Promise { + return Promise.resolve({ + session: { historyId: opts.historyId, kind: this.kind, cwd: '/repo', createdAt: 1 }, + events: [...this.events], + }); + } +} + +async function makeService(opts: { + journals: ConversationLiveJournals; + record: SessionRecord; + openRequests?: AgentEvent[]; + historyEvents?: AgentHistoryEvent[]; +}) { + const runTask = (effect: Effect.Effect) => { + void Effect.runPromise(effect); + }; + const transport: Transport = { + connect: () => Promise.resolve(), + send: noop, + onMessage: () => noop, + onClose: () => noop, + close: noop, + }; + const records = new SessionRecordRegistry(new InMemorySessionStore(), noop); + await Effect.runPromise(records.start(runTask)); + records.register(opts.record); + const store = new InMemoryConversationStore(); + const turns = new ConversationTurnService(store, records, transport, runTask); + const history = new HistoryService(() => + opts.historyEvents ? new CannedHistoryAdapter(opts.historyEvents) : new FakeAdapter(), + ); + const service = new ConversationProjectionService( + turns, + records, + history, + opts.journals, + () => opts.openRequests ?? [], + ); + return { service, store }; +} + +function makeRecord(activeLeafTurnId: TurnId, withHistory = false): SessionRecord { + return { + sessionId, + kind: 'claude-code', + cwd: '/repo', + origin: { type: 'created' }, + createdAt: 1, + updatedAt: 1, + runs: [{ runId, startedAt: 1, ...(withHistory && { historyId: asHistoryId('hist-1') }) }], + activeLeafTurnId, + graphRevision: 1, + eventEpoch: 3, + }; +} + +describe('conversation projection live tail (CODE-35)', () => { + it('clears a truncated in-flight stream and still delivers open asks', async () => { + const liveTurnId = 'turn-live' as TurnId; + const journals = new ConversationLiveJournals(Number.MAX_SAFE_INTEGER, 4); + const journal = journals.open(sessionId); + journal.append(stamped(1, liveTurnId, chunk('msg-a', 'head '))); + journal.append(stamped(2, liveTurnId, chunk('msg-a', 'mid '))); + journal.append( + stamped(3, liveTurnId, { + type: 'tool-call', + toolCall: { + toolCallId: 't1', + title: 'Run tests', + kind: 'execute', + status: 'completed', + content: [], + }, + }), + ); + journal.append(stamped(4, liveTurnId, chunk('msg-b', 'fresh '))); + journal.append(stamped(5, liveTurnId, chunk('msg-b', 'tail'))); + + const { service, store } = await makeService({ + journals, + record: makeRecord(liveTurnId), + openRequests: [OPEN_ASK], + }); + await store.saveTurn({ + turnId: liveTurnId, + sessionId, + parentTurnId: null, + siblingOrdinal: 1, + input: { type: 'shell-command', command: 'pnpm test' }, + runId, + state: 'running', + createdAt: 10, + }); + + const result = await Effect.runPromise(service.read({ sessionId })); + + expect(result.cursor).toBeUndefined(); + expect(result.watermark).toEqual({ epoch: 3, seq: 5 }); + const tailEvents = result.events.flatMap((item) => ('event' in item ? [item.event] : [])); + // msg-a lost its head (seq 1 evicted) but seq 2 survived: the RETAINED continuation must be + // dropped too — no headless splice, the message restarts. + expect(tailEvents.filter((e) => e.type === 'agent-message-chunk')).toEqual([ + chunk('msg-b', 'fresh '), + chunk('msg-b', 'tail'), + ]); + expect(tailEvents).toContainEqual( + expect.objectContaining({ type: 'tool-call', toolCall: expect.anything() }), + ); + // Eviction reached the live turn's region: the read must not claim its content is complete. + expect(result.events).toContainEqual({ + type: 'history-unavailable', + turnId: liveTurnId, + runId, + }); + // The open ask reaches the reader even though its request event never survived the journal. + const ask = result.events.find((item) => 'event' in item && item.event === OPEN_ASK); + expect(ask).toMatchObject({ turnId: liveTurnId, runId }); + }); + + it('surfaces truncation when a full-state event above the cut was evicted', async () => { + const doneTurnId = 'turn-done' as TurnId; + const liveTurnId = 'turn-live' as TurnId; + const journals = new ConversationLiveJournals(Number.MAX_SAFE_INTEGER, 2); + const journal = journals.open(sessionId); + journal.append(stamped(1, doneTurnId, { type: 'stop', stopReason: 'end_turn' })); + journal.append( + stamped(2, liveTurnId, { + type: 'tool-call', + toolCall: { + toolCallId: 't-lost', + title: 'Completed then lost', + kind: 'execute', + status: 'completed', + content: [], + }, + }), + ); + journal.append(stamped(3, liveTurnId, chunk('msg-live', 'one '))); + journal.append(stamped(4, liveTurnId, chunk('msg-live', 'two'))); + + const { service, store } = await makeService({ journals, record: makeRecord(liveTurnId) }); + await store.saveTurn({ + turnId: doneTurnId, + sessionId, + parentTurnId: null, + siblingOrdinal: 1, + input: { type: 'shell-command', command: 'ls' }, + runId, + state: 'completed', + createdAt: 5, + }); + await store.saveTurn({ + turnId: liveTurnId, + sessionId, + parentTurnId: doneTurnId, + siblingOrdinal: 1, + input: { type: 'shell-command', command: 'pwd' }, + runId, + state: 'running', + createdAt: 10, + }); + + const result = await Effect.runPromise(service.read({ sessionId })); + + // The completed tool snapshot was evicted and will never re-emit — the tail is silently + // short of it, so the in-flight turn must carry the incompleteness marker. + const tailEvents = result.events.flatMap((item) => ('event' in item ? [item.event] : [])); + expect(tailEvents).not.toContainEqual(expect.objectContaining({ type: 'tool-call' })); + expect(result.events).toContainEqual({ + type: 'history-unavailable', + turnId: liveTurnId, + runId, + }); + expect(result.watermark).toEqual({ epoch: 3, seq: 4 }); + }); + + it('orders the tail by stamp, never by journal append order', async () => { + const liveTurnId = 'turn-live' as TurnId; + const journals = new ConversationLiveJournals(); + const journal = journals.open(sessionId); + journal.append(stamped(1, liveTurnId, chunk('msg-live', 'one '))); + journal.append(stamped(2, liveTurnId, chunk('msg-live', 'two'))); + // An old-epoch straggler appended late sits after newer entries in the journal. + journal.append({ + epoch: 2, + seq: 9, + runId, + turnId: liveTurnId, + ts: 1, + event: chunk('msg-old', 'stale'), + }); + + const { service, store } = await makeService({ journals, record: makeRecord(liveTurnId) }); + await store.saveTurn({ + turnId: liveTurnId, + sessionId, + parentTurnId: null, + siblingOrdinal: 1, + input: { type: 'shell-command', command: 'pnpm test' }, + runId, + state: 'running', + createdAt: 10, + }); + + const result = await Effect.runPromise(service.read({ sessionId })); + + const stamps = result.events.flatMap((item) => + 'event' in item && item.epoch !== undefined ? [[item.epoch, item.seq]] : [], + ); + expect(stamps).toEqual([ + [2, 9], + [3, 1], + [3, 2], + ]); + }); + + it('cuts the tail at the last event of a settled path turn', async () => { + const doneTurnId = 'turn-done' as TurnId; + const liveTurnId = 'turn-live' as TurnId; + const journals = new ConversationLiveJournals(); + const journal = journals.open(sessionId); + journal.append( + stamped(1, doneTurnId, { + type: 'tool-call', + toolCall: { + toolCallId: 't0', + title: 'Old work', + kind: 'execute', + status: 'completed', + content: [], + }, + }), + ); + journal.append(stamped(2, doneTurnId, { type: 'stop', stopReason: 'end_turn' })); + journal.append(stamped(3, liveTurnId, chunk('msg-live', 'streaming'))); + + const { service, store } = await makeService({ journals, record: makeRecord(liveTurnId) }); + await store.saveTurn({ + turnId: doneTurnId, + sessionId, + parentTurnId: null, + siblingOrdinal: 1, + input: { type: 'shell-command', command: 'ls' }, + runId, + state: 'completed', + createdAt: 5, + }); + await store.saveTurn({ + turnId: liveTurnId, + sessionId, + parentTurnId: doneTurnId, + siblingOrdinal: 1, + input: { type: 'shell-command', command: 'pwd' }, + runId, + state: 'running', + createdAt: 10, + }); + + const result = await Effect.runPromise(service.read({ sessionId })); + + // Settled-turn journal events fall below the cut: durable sources own that turn (here the + // no-history placeholder), so only the live turn's stream rides the tail. + const stamps = result.events.flatMap((item) => + 'event' in item && item.seq !== undefined ? [item.seq] : [], + ); + expect(stamps).toEqual([3]); + expect(result.events).toContainEqual({ + type: 'history-unavailable', + turnId: doneTurnId, + runId, + }); + expect(result.watermark).toEqual({ epoch: 3, seq: 3 }); + }); +}); + +describe('conversation projection attribution gate', () => { + function shellTurn( + turnId: string, + parentTurnId: string | null, + command: string, + state: ConversationTurnState, + ordinal = 1, + ) { + return { + turnId: turnId as TurnId, + sessionId, + parentTurnId: parentTurnId as TurnId | null, + siblingOrdinal: ordinal, + input: { type: 'shell-command' as const, command }, + runId, + state, + createdAt: 10, + }; + } + + function providerUser(itemId: string, command: string): AgentHistoryEvent { + return { + historyId: asHistoryId('hist-1'), + itemId, + event: { + type: 'user-message', + messageId: itemId as MessageId, + content: [{ type: 'text', text: `$ ${command}` }], + }, + }; + } + + function providerAnswer(itemId: string, text: string): AgentHistoryEvent { + return { + historyId: asHistoryId('hist-1'), + itemId, + event: { + type: 'agent-message', + messageId: itemId as MessageId, + content: [{ type: 'text', text }], + }, + }; + } + + function answers(events: readonly ConversationReadItem[]): Array<[string, TurnId | undefined]> { + return events.flatMap((item) => + 'event' in item && item.event.type === 'agent-message' && item.event.messageId !== undefined + ? [[item.event.messageId as string, item.turnId] as [string, TurnId | undefined]] + : [], + ); + } + + function placeholderTurnIds(events: readonly ConversationReadItem[]): TurnId[] { + return events.flatMap((item) => ('event' in item ? [] : [item.turnId])); + } + + it.each(['complete', 'evicted', 'gap', 'wrong-run', 'wrong-epoch', 'no-stop'] as const)( + 'replays unavailable settled history only from a complete journal: %s', + async (condition) => { + const turnId = 'turn-retained' as TurnId; + const journals = new ConversationLiveJournals( + Number.MAX_SAFE_INTEGER, + condition === 'evicted' ? 2 : 100, + ); + const journal = journals.open(sessionId); + journal.append(stamped(1, turnId, providerUser('u-a', 'a').event)); + journal.append( + stamped(condition === 'gap' ? 3 : 2, turnId, providerAnswer('ans-a', 'answer a').event), + ); + if (condition !== 'no-stop') { + journal.append({ + ...stamped(3, turnId, { type: 'stop', stopReason: 'end_turn' }), + ...(condition === 'wrong-run' && { runId: 'other-run' as RunId }), + ...(condition === 'wrong-epoch' && { epoch: 4 }), + }); + } + const { service, store } = await makeService({ + journals, + record: makeRecord(turnId, true), + historyEvents: [], + }); + await store.saveTurn(shellTurn(turnId, null, 'a', 'completed')); + const read = await Effect.runPromise(service.read({ sessionId })); + expect(answers(read.events)).toEqual(condition === 'complete' ? [['ans-a', turnId]] : []); + expect(placeholderTurnIds(read.events)).toEqual(condition === 'complete' ? [] : [turnId]); + expect( + read.events.filter((item) => 'event' in item && item.event.type === 'user-message'), + ).toHaveLength(1); + }, + ); + + it('never attributes positionally on an inactive sibling lineage — even an identical retry', async () => { + const { service, store } = await makeService({ + journals: new ConversationLiveJournals(), + record: makeRecord('turn-b2' as TurnId, true), + historyEvents: [ + providerUser('u-a', 'a'), + providerAnswer('ans-a', 'answer a'), + providerUser('u-b', 'b'), + providerAnswer('ans-b', 'answer b'), + ], + }); + await store.saveTurn(shellTurn('turn-a', null, 'a', 'completed')); + await store.saveTurn(shellTurn('turn-b1', 'turn-a', 'b', 'completed', 1)); + await store.saveTurn(shellTurn('turn-b2', 'turn-a', 'b', 'completed', 2)); + + // The inactive sibling B1 carries IDENTICAL prompt text to the active B2: counts and + // fingerprints both pass, so only the active-lineage gate stops the mis-slice. + const inactive = await Effect.runPromise( + service.read({ sessionId, leafTurnId: 'turn-b1' as TurnId }), + ); + expect(answers(inactive.events)).toEqual([]); + expect(placeholderTurnIds(inactive.events)).toEqual(['turn-a', 'turn-b1']); + + // The active lineage attributes normally. + const active = await Effect.runPromise(service.read({ sessionId })); + expect(answers(active.events)).toEqual([ + ['ans-a', 'turn-a'], + ['ans-b', 'turn-b2'], + ]); + expect(placeholderTurnIds(active.events)).toEqual([]); + }); + + it('attributes nothing when the trailing extra partition is not the in-flight prompt', async () => { + const { service, store } = await makeService({ + journals: new ConversationLiveJournals(), + record: makeRecord('turn-c' as TurnId, true), + // Interior injected row: count passes only via the +1 tolerance, which must verify the + // TRAILING row as the live prompt — here the trailing row is B's, so nothing attributes. + historyEvents: [ + providerUser('u-a', 'a'), + providerAnswer('ans-a', 'answer a'), + providerUser('u-x', 'x'), + providerAnswer('ans-x', 'answer x'), + providerUser('u-b', 'b'), + providerAnswer('ans-b', 'answer b'), + ], + }); + await store.saveTurn(shellTurn('turn-a', null, 'a', 'completed')); + await store.saveTurn(shellTurn('turn-b', 'turn-a', 'b', 'completed')); + await store.saveTurn(shellTurn('turn-c', 'turn-b', 'c', 'running')); + + const result = await Effect.runPromise(service.read({ sessionId })); + + expect(answers(result.events)).toEqual([]); + expect(placeholderTurnIds(result.events)).toEqual(['turn-a', 'turn-b']); + }); + + it('tolerates exactly one trailing partition that verifies as the in-flight prompt', async () => { + const { service, store } = await makeService({ + journals: new ConversationLiveJournals(), + record: makeRecord('turn-c' as TurnId, true), + historyEvents: [ + providerUser('u-a', 'a'), + providerAnswer('ans-a', 'answer a'), + providerUser('u-b', 'b'), + providerAnswer('ans-b', 'answer b'), + providerUser('u-c', 'c'), + ], + }); + await store.saveTurn(shellTurn('turn-a', null, 'a', 'completed')); + await store.saveTurn(shellTurn('turn-b', 'turn-a', 'b', 'completed')); + await store.saveTurn(shellTurn('turn-c', 'turn-b', 'c', 'running')); + + const result = await Effect.runPromise(service.read({ sessionId })); + + expect(answers(result.events)).toEqual([ + ['ans-a', 'turn-a'], + ['ans-b', 'turn-b'], + ]); + expect(placeholderTurnIds(result.events)).toEqual([]); + }); + + it('rejects a count compensated by the live echo when a settled row is missing', async () => { + const { service, store } = await makeService({ + journals: new ConversationLiveJournals(), + record: makeRecord('turn-c' as TurnId, true), + // A's row vanished (the image-only-prompt lossiness class); the live echo makes the count + // pass, but position 0 no longer fingerprints as A — nothing may attribute shifted. + historyEvents: [ + providerUser('u-b', 'b'), + providerAnswer('ans-b', 'answer b'), + providerUser('u-c', 'c'), + ], + }); + await store.saveTurn(shellTurn('turn-a', null, 'a', 'completed')); + await store.saveTurn(shellTurn('turn-b', 'turn-a', 'b', 'completed')); + await store.saveTurn(shellTurn('turn-c', 'turn-b', 'c', 'running')); + + const result = await Effect.runPromise(service.read({ sessionId })); + + expect(answers(result.events)).toEqual([]); + expect(placeholderTurnIds(result.events)).toEqual(['turn-a', 'turn-b']); + }); + + it('attributes the matching prefix and degrades from the first mismatch onward', async () => { + const { service, store } = await makeService({ + journals: new ConversationLiveJournals(), + record: makeRecord('turn-b' as TurnId, true), + historyEvents: [ + providerUser('u-a', 'a'), + providerAnswer('ans-a', 'answer a'), + providerUser('u-z', 'z'), + providerAnswer('ans-z', 'answer z'), + ], + }); + await store.saveTurn(shellTurn('turn-a', null, 'a', 'completed')); + await store.saveTurn(shellTurn('turn-b', 'turn-a', 'b', 'completed')); + + const result = await Effect.runPromise(service.read({ sessionId })); + + // Position 0 verifies and renders; position 1 mismatches, so B (and everything after it) + // degrades — alignment is never resynced past a mismatch. + expect(answers(result.events)).toEqual([['ans-a', 'turn-a']]); + expect(placeholderTurnIds(result.events)).toEqual(['turn-b']); + }); +}); + +describe('pageReadItems byte budget', () => { + function textItem(turnId: string, text: string): ConversationReadItem { + return { + turnId: turnId as TurnId, + event: { + type: 'user-message', + messageId: `msg-${turnId}` as MessageId, + content: [{ type: 'text', text }], + }, + }; + } + + function bytes(item: ConversationReadItem): number { + return Buffer.byteLength(JSON.stringify(item), 'utf8'); + } + + it('splits durable items at the budget and never stalls on one oversized item', () => { + const items = [textItem('t1', 'x'.repeat(400)), textItem('t2', 'y'.repeat(400))]; + const budget = bytes(items[0]) + 10; + + const first = pageReadItems(items, [], 0, 1000, budget); + expect(first.events).toEqual([items[0]]); + expect(first.nextOffset).toBe(1); + + const second = pageReadItems(items, [], 1, 1000, budget); + expect(second.events).toEqual([items[1]]); + expect(second.nextOffset).toBeUndefined(); + + // An item alone above the budget still ships as its own page. + const oversized = pageReadItems([textItem('t3', 'z'.repeat(4000))], [], 0, 1000, budget); + expect(oversized.events).toHaveLength(1); + expect(oversized.nextOffset).toBeUndefined(); + }); + + it('keeps the live tail atomic to the final page', () => { + const durable = [textItem('t1', 'x'.repeat(100))]; + const tail = [textItem('live', 'w'.repeat(300))]; + const budget = bytes(durable[0]) + 10; + + // The tail does not fit next to the durable remainder: it gets its own final page. + const first = pageReadItems(durable, tail, 0, 1000, budget); + expect(first.events).toEqual(durable); + expect(first.nextOffset).toBe(1); + + const last = pageReadItems(durable, tail, 1, 1000, budget); + expect(last.events).toEqual(tail); + expect(last.nextOffset).toBeUndefined(); + }); + + it('trims an oversized tail from the front and clears the damaged stream', () => { + const bigChunk: ConversationReadItem = { + event: chunk('msg-big', 'x'.repeat(500)), + }; + const laterChunk: ConversationReadItem = { event: chunk('msg-big', 'tail piece') }; + const otherChunk: ConversationReadItem = { event: chunk('msg-ok', 'intact') }; + const budget = bytes(bigChunk) + bytes(otherChunk) + bytes(laterChunk) - 1; + + const page = pageReadItems([], [bigChunk, laterChunk, otherChunk], 0, 1000, budget); + // Dropping the stream head drops its retained continuation too — never a headless splice. + expect(page.events).toEqual([otherChunk]); + expect(page.nextOffset).toBeUndefined(); + }); +}); + +describe('conversation read cursor integrity', () => { + async function pagedService() { + const liveTurnId = 'turn-live' as TurnId; + const setup = await makeService({ + journals: new ConversationLiveJournals(), + record: makeRecord(liveTurnId), + }); + await setup.store.saveTurn({ + turnId: 'turn-done' as TurnId, + sessionId, + parentTurnId: null, + siblingOrdinal: 1, + input: { type: 'shell-command', command: 'ls' }, + runId, + state: 'completed', + createdAt: 5, + }); + await setup.store.saveTurn({ + turnId: liveTurnId, + sessionId, + parentTurnId: 'turn-done' as TurnId, + siblingOrdinal: 1, + input: { type: 'shell-command', command: 'pwd' }, + runId, + state: 'running', + createdAt: 10, + }); + return { ...setup, liveTurnId }; + } + + async function expectConflict( + effect: Effect.Effect, + ): Promise { + const exit = await Effect.runPromiseExit(effect); + if (!Exit.isFailure(exit)) throw new Error('expected a conflict failure'); + const error = Cause.squash(exit.cause); + if (!(error instanceof RequestError)) throw new Error('expected a RequestError'); + expect(error.code).toBe('conflict'); + } + + it('pages with a structured cursor and rejects it once a turn settles', async () => { + const { service, store, liveTurnId } = await pagedService(); + + const first = await Effect.runPromise(service.read({ sessionId, limit: 1 })); + expect(first.cursor).toBeDefined(); + expect(first.watermark).toBeUndefined(); + + const second = await Effect.runPromise( + service.read({ sessionId, cursor: first.cursor, limit: 1 }), + ); + expect(second.events).toHaveLength(1); + + // The live turn settles WITHOUT a graph-revision bump: the attribution gate's shape flipped, + // so the old cursor must conflict instead of splicing across the mutation. + await store.saveTurn({ + turnId: liveTurnId, + sessionId, + parentTurnId: 'turn-done' as TurnId, + siblingOrdinal: 1, + input: { type: 'shell-command', command: 'pwd' }, + runId, + state: 'completed', + createdAt: 10, + }); + await expectConflict(service.read({ sessionId, cursor: first.cursor, limit: 1 })); + }); + + it('rejects undecodable and tampered cursors instead of restarting silently', async () => { + const { service } = await pagedService(); + await expectConflict(service.read({ sessionId, cursor: 'garbage' })); + await expectConflict(service.read({ sessionId, cursor: '{}' })); + await expectConflict( + service.read({ + sessionId, + cursor: JSON.stringify({ + graphRevision: 999, + leafTurnId: 'turn-live', + settled: 1, + offset: 1, + }), + }), + ); + }); +}); diff --git a/packages/host/engine/src/__tests__/engine-conversation-read.test.ts b/packages/host/engine/src/__tests__/engine-conversation-read.test.ts new file mode 100644 index 000000000..ce196a75a --- /dev/null +++ b/packages/host/engine/src/__tests__/engine-conversation-read.test.ts @@ -0,0 +1,490 @@ +import { Buffer } from 'node:buffer'; +import { asHistoryId } from '@linkcode/agent-adapter'; +import type { + AgentEvent, + AgentHistoryEvent, + AgentHistoryReadOptions, + AgentHistoryReadResult, + ConversationReadItem, + MessageId, + WirePayload, +} from '@linkcode/schema'; +import { + compareConversationWatermarks, + MAX_ATTACHMENT_TOTAL_BASE64_LENGTH, + OperationIdSchema, + SessionIdSchema, +} from '@linkcode/schema'; +import { nullthrow } from 'foxts/guard'; +import { describe, expect, it } from 'vitest'; +import { + FakeAdapter, + createSessionHarness as harness, + settleEngineTasks, + startedSessionId as startedId, +} from './fixtures/session-harness'; + +const HISTORY_ID = asHistoryId('hist-1'); + +interface SharedHistory { + events: AgentHistoryEvent[]; + failRead: boolean; + lastReadOpts?: AgentHistoryReadOptions; +} + +/** Provider history double: canned corpus, optionally failing reads (the CODE-645 class). The + * corpus is shared across instances because cold reads construct a fresh adapter per call. */ +class HistoryFakeAdapter extends FakeAdapter { + constructor(private readonly shared: SharedHistory) { + super(); + } + + override readHistory(opts: AgentHistoryReadOptions): Promise { + this.shared.lastReadOpts = opts; + if (this.shared.failRead) { + return Promise.reject(new Error('history_mode paginated is unsupported')); + } + return Promise.resolve({ + session: { historyId: opts.historyId, kind: this.kind, cwd: '/repo', createdAt: 1 }, + events: [...this.shared.events], + }); + } +} + +function historyEvent(itemId: string, event: AgentEvent): AgentHistoryEvent { + return { historyId: HISTORY_ID, itemId, event }; +} + +function userRow(itemId: string, text: string): AgentHistoryEvent { + return historyEvent(itemId, { + type: 'user-message', + messageId: itemId as MessageId, + content: [{ type: 'text', text }], + }); +} + +function assistantRow(itemId: string, text: string): AgentHistoryEvent { + return historyEvent(itemId, { + type: 'agent-message', + messageId: itemId as MessageId, + content: [{ type: 'text', text }], + }); +} + +async function startedHarness(makeAdapter: () => FakeAdapter = () => new FakeAdapter()) { + const h = harness(undefined, makeAdapter); + await h.engine.start(); + await h.inject({ + kind: 'session.start', + clientReqId: 'r1', + opts: { kind: 'claude-code', cwd: '/repo' }, + }); + const sessionId = startedId(h.sent, 'r1'); + return { ...h, sessionId, adapter: nullthrow(h.adapters[0]) }; +} + +type Harness = Awaited>; + +async function completeTurn(h: Harness, clientReqId: string, text: string): Promise { + await h.inject({ + kind: 'turn.submit', + clientReqId, + sessionId: h.sessionId, + operationId: OperationIdSchema.parse(`op-${clientReqId}`), + input: { type: 'prompt', blocks: [{ type: 'text', text }] }, + }); + h.adapter.emit({ type: 'stop', stopReason: 'end_turn' }); + h.adapter.emit({ type: 'status', status: 'idle' }); + await settleEngineTasks(); +} + +function readResult(sent: WirePayload[], replyTo: string) { + const reply = sent.find( + (payload) => payload.kind === 'conversation.read.result' && payload.replyTo === replyTo, + ); + if (reply?.kind !== 'conversation.read.result') { + throw new Error(`no conversation.read.result for ${replyTo}`); + } + return reply; +} + +function graphResult(sent: WirePayload[], replyTo: string) { + const reply = sent.find( + (payload) => payload.kind === 'conversation.graph.result' && payload.replyTo === replyTo, + ); + if (reply?.kind !== 'conversation.graph.result') { + throw new Error(`no conversation.graph.result for ${replyTo}`); + } + return reply; +} + +function userTexts(events: readonly ConversationReadItem[]): string[] { + return events.flatMap((item) => { + if (!('event' in item) || item.event.type !== 'user-message') return []; + return item.event.content.flatMap((block) => (block.type === 'text' ? [block.text] : [])); + }); +} + +describe('conversation.graph.get', () => { + it('serves the turn tree with states, ordinals, and input summaries', async () => { + const h = await startedHarness(); + await completeTurn(h, 's1', 'first prompt'); + await completeTurn(h, 's2', 'second prompt'); + + await h.inject({ kind: 'conversation.graph.get', clientReqId: 'g1', sessionId: h.sessionId }); + + const graph = graphResult(h.sent, 'g1'); + expect(graph.graphRevision).toBe(2); + expect(graph.turns).toHaveLength(2); + const [first, second] = graph.turns; + expect(first).toMatchObject({ + parentTurnId: null, + siblingOrdinal: 1, + state: 'completed', + inputSummary: 'first prompt', + }); + expect(second).toMatchObject({ + parentTurnId: first.turnId, + siblingOrdinal: 1, + state: 'completed', + inputSummary: 'second prompt', + }); + expect(graph.activeLeafTurnId).toBe(second.turnId); + }); + + it('fails loudly for an unknown session instead of dropping the request', async () => { + const h = await startedHarness(); + + const unknown = SessionIdSchema.parse('session-x'); + await h.inject({ kind: 'conversation.graph.get', clientReqId: 'g-x', sessionId: unknown }); + await h.inject({ kind: 'conversation.read', clientReqId: 'r-x', sessionId: unknown }); + + expect(h.sent).toContainEqual( + expect.objectContaining({ kind: 'request.failed', replyTo: 'g-x', code: 'not_found' }), + ); + expect(h.sent).toContainEqual( + expect.objectContaining({ kind: 'request.failed', replyTo: 'r-x', code: 'not_found' }), + ); + }); +}); + +describe('conversation.read', () => { + it('retains a completed live reply when history fails after a real prompt dispatch', async () => { + const shared: SharedHistory = { events: [], failRead: true }; + const h = await startedHarness(() => new HistoryFakeAdapter(shared)); + h.adapter.emit({ type: 'session-ref', historyId: HISTORY_ID }); + await h.inject({ + kind: 'turn.submit', + clientReqId: 's-live', + sessionId: h.sessionId, + operationId: OperationIdSchema.parse('op-live'), + input: { type: 'prompt', blocks: [{ type: 'text', text: 'keep this reply' }] }, + }); + h.adapter.emit(assistantRow('a-live', 'retained reply').event); + h.adapter.emit({ type: 'stop', stopReason: 'end_turn' }); + h.adapter.emit({ type: 'status', status: 'idle' }); + await settleEngineTasks(); + + await h.inject({ kind: 'conversation.read', clientReqId: 'rr-live', sessionId: h.sessionId }); + + const result = readResult(h.sent, 'rr-live'); + expect(userTexts(result.events)).toEqual(['keep this reply']); + expect(result.events).toContainEqual( + expect.objectContaining({ event: assistantRow('a-live', 'retained reply').event }), + ); + expect(result.events).not.toContainEqual( + expect.objectContaining({ type: 'history-unavailable' }), + ); + }); + + it('renders prompts and placeholders when the harness has no history', async () => { + const h = await startedHarness(); + await completeTurn(h, 's1', 'hello one'); + await completeTurn(h, 's2', 'hello two'); + await h.inject({ kind: 'session.stop', clientReqId: 'stop', sessionId: h.sessionId }); + + await h.inject({ kind: 'conversation.read', clientReqId: 'rr', sessionId: h.sessionId }); + + const result = readResult(h.sent, 'rr'); + expect(result.cursor).toBeUndefined(); + expect(result.watermark).toBeDefined(); + expect(userTexts(result.events)).toEqual(['hello one', 'hello two']); + const placeholders = result.events.filter((item) => !('event' in item)); + expect(placeholders).toHaveLength(2); + // Placeholders sit under their own turns, interleaved with the prompts. + expect(result.events.map((item) => ('event' in item ? item.event.type : item.type))).toEqual([ + 'user-message', + 'history-unavailable', + 'user-message', + 'history-unavailable', + ]); + }); + + it('degrades to the prompt-only fallback when the provider read fails', async () => { + const shared: SharedHistory = { events: [], failRead: true }; + const h = await startedHarness(() => new HistoryFakeAdapter(shared)); + h.adapter.emit({ type: 'session-ref', historyId: HISTORY_ID }); + await completeTurn(h, 's1', 'still readable'); + await h.inject({ kind: 'session.stop', clientReqId: 'stop', sessionId: h.sessionId }); + + await h.inject({ kind: 'conversation.read', clientReqId: 'rr', sessionId: h.sessionId }); + + const result = readResult(h.sent, 'rr'); + expect(userTexts(result.events)).toEqual(['still readable']); + expect(result.events).toContainEqual(expect.objectContaining({ type: 'history-unavailable' })); + expect(h.sent).not.toContainEqual( + expect.objectContaining({ kind: 'request.failed', replyTo: 'rr' }), + ); + }); + + it('merges provider assistant output under host user rows', async () => { + const shared: SharedHistory = { events: [], failRead: false }; + const h = await startedHarness(() => new HistoryFakeAdapter(shared)); + h.adapter.emit({ type: 'session-ref', historyId: HISTORY_ID }); + await completeTurn(h, 's1', 'real prompt'); + shared.events = [userRow('u1', 'real prompt'), assistantRow('a1', 'provider answer')]; + await h.inject({ kind: 'session.stop', clientReqId: 'stop', sessionId: h.sessionId }); + + await h.inject({ kind: 'conversation.read', clientReqId: 'rr', sessionId: h.sessionId }); + + const result = readResult(h.sent, 'rr'); + // The user row is host truth, rendered exactly once — the provider's own row never doubles it. + expect(userTexts(result.events)).toEqual(['real prompt']); + const assistant = result.events.find( + (item) => 'event' in item && item.event.type === 'agent-message', + ); + expect(assistant).toBeDefined(); + if (assistant === undefined || !('event' in assistant)) throw new Error('unreachable'); + expect(assistant.turnId).toBeDefined(); + expect(assistant.runId).toBeDefined(); + expect(result.events).not.toContainEqual( + expect.objectContaining({ type: 'history-unavailable' }), + ); + // codex resolves its rollout home through the project env, so the read must carry the cwd. + expect(shared.lastReadOpts?.cwd).toBe('/repo'); + }); + + it('refreshes a stale corpus captured before the newest settle', async () => { + const shared: SharedHistory = { events: [], failRead: false }; + const h = await startedHarness(() => new HistoryFakeAdapter(shared)); + h.adapter.emit({ type: 'session-ref', historyId: HISTORY_ID }); + shared.events = [userRow('u1', 'one prompt'), assistantRow('a1', 'answer one')]; + await completeTurn(h, 's1', 'one prompt'); + // Warm the TTL cache with the one-turn corpus. + await h.inject({ kind: 'conversation.read', clientReqId: 'rr-1', sessionId: h.sessionId }); + expect(userTexts(readResult(h.sent, 'rr-1').events)).toEqual(['one prompt']); + + shared.events = [ + userRow('u1', 'one prompt'), + assistantRow('a1', 'answer one'), + userRow('u2', 'two prompt'), + assistantRow('a2', 'answer two'), + ]; + await completeTurn(h, 's2', 'two prompt'); + await h.inject({ kind: 'conversation.read', clientReqId: 'rr-2', sessionId: h.sessionId }); + + // A cached one-turn corpus would degrade both turns to placeholders; the settle forces a + // cache-bypassing refresh, so the new turn's answer attributes. + const result = readResult(h.sent, 'rr-2'); + expect(result.events).not.toContainEqual( + expect.objectContaining({ type: 'history-unavailable' }), + ); + expect(result.events).toContainEqual( + expect.objectContaining({ + event: expect.objectContaining({ type: 'agent-message', messageId: 'a2' }), + }), + ); + }); + + it('refreshes a mid-turn capture whose settled turn had no answer rows yet', async () => { + const shared: SharedHistory = { events: [], failRead: false }; + const h = await startedHarness(() => new HistoryFakeAdapter(shared)); + h.adapter.emit({ type: 'session-ref', historyId: HISTORY_ID }); + shared.events = [userRow('u1', 'one prompt'), assistantRow('a1', 'answer one')]; + await completeTurn(h, 's1', 'one prompt'); + // Turn two is dispatched; the provider has written its echo but no answer yet. + await h.inject({ + kind: 'turn.submit', + clientReqId: 's2', + sessionId: h.sessionId, + operationId: OperationIdSchema.parse('op-s2-live'), + input: { type: 'prompt', blocks: [{ type: 'text', text: 'two prompt' }] }, + }); + h.adapter.emit({ type: 'status', status: 'running' }); + await settleEngineTasks(); + shared.events = [ + userRow('u1', 'one prompt'), + assistantRow('a1', 'answer one'), + userRow('u2', 'two prompt'), + ]; + // Warm the cache mid-turn: the trailing echo verifies as the in-flight turn's own row. + await h.inject({ kind: 'conversation.read', clientReqId: 'rr-mid', sessionId: h.sessionId }); + expect(userTexts(readResult(h.sent, 'rr-mid').events)).toEqual(['one prompt', 'two prompt']); + + h.adapter.emit({ type: 'stop', stopReason: 'end_turn' }); + h.adapter.emit({ type: 'status', status: 'idle' }); + await settleEngineTasks(); + shared.events = [...shared.events, assistantRow('a2', 'answer two')]; + await h.inject({ kind: 'conversation.read', clientReqId: 'rr-2', sessionId: h.sessionId }); + + // Without the settle-driven refresh, the cached capture passes every gate with an empty + // partition and the settled turn renders answer-less forever. + expect(readResult(h.sent, 'rr-2').events).toContainEqual( + expect.objectContaining({ + event: expect.objectContaining({ type: 'agent-message', messageId: 'a2' }), + }), + ); + }); + + it('keeps a launch-window read below the run’s later events', async () => { + const h = await startedHarness(); + // No adapter event has flowed yet, so the session has no live journal. + await h.inject({ kind: 'conversation.read', clientReqId: 'rr-launch', sessionId: h.sessionId }); + const early = readResult(h.sent, 'rr-launch'); + expect(early.watermark).toBeDefined(); + if (early.watermark === undefined) throw new Error('unreachable'); + + h.adapter.emit({ type: 'status', status: 'running' }); + await settleEngineTasks(); + const stampedEvent = h.sent.find( + (payload) => + payload.kind === 'agent.event' && payload.epoch !== undefined && payload.seq !== undefined, + ); + if (stampedEvent?.kind !== 'agent.event') throw new Error('no stamped agent.event'); + // A client that adopted the launch-window watermark must not drop the run's own stream. + expect( + compareConversationWatermarks(early.watermark, { + epoch: stampedEvent.epoch ?? 0, + seq: stampedEvent.seq ?? 0, + }), + ).toBeLessThan(0); + }); + + it('serves the live tail with stamps, open asks, and no duplicated user echo', async () => { + const h = await startedHarness(); + await h.inject({ + kind: 'turn.submit', + clientReqId: 's1', + sessionId: h.sessionId, + operationId: OperationIdSchema.parse('op-live'), + input: { type: 'prompt', blocks: [{ type: 'text', text: 'live prompt' }] }, + }); + h.adapter.emit({ type: 'status', status: 'running' }); + h.adapter.emit({ + type: 'agent-message-chunk', + messageId: 'msg-live' as MessageId, + content: { type: 'text', text: 'streaming…' }, + }); + h.adapter.emit({ + type: 'permission-request', + requestId: 'perm-live', + title: 'Run', + subject: { type: 'tool-call', toolCallId: 't1' }, + options: [{ optionId: 'ok', name: 'Allow', kind: 'allow_once' }], + }); + await settleEngineTasks(); + + await h.inject({ kind: 'conversation.read', clientReqId: 'rr', sessionId: h.sessionId }); + + const result = readResult(h.sent, 'rr'); + expect(result.cursor).toBeUndefined(); + expect(result.watermark).toBeDefined(); + // The in-flight prompt renders once, from host truth — the live echo never doubles it. + expect(userTexts(result.events)).toEqual(['live prompt']); + const chunkItem = result.events.find( + (item) => 'event' in item && item.event.type === 'agent-message-chunk', + ); + expect(chunkItem).toBeDefined(); + if (chunkItem === undefined || !('event' in chunkItem)) throw new Error('unreachable'); + expect(chunkItem.epoch).toBeDefined(); + expect(chunkItem.seq).toBeDefined(); + expect(result.events).toContainEqual( + expect.objectContaining({ + event: expect.objectContaining({ type: 'permission-request', requestId: 'perm-live' }), + }), + ); + }); + + it('rejects stale and undecodable cursors with a typed conflict', async () => { + const h = await startedHarness(); + await completeTurn(h, 's1', 'one'); + await completeTurn(h, 's2', 'two'); + await h.inject({ + kind: 'conversation.read', + clientReqId: 'rr-p1', + sessionId: h.sessionId, + limit: 1, + }); + const page1 = readResult(h.sent, 'rr-p1'); + expect(page1.cursor).toBeDefined(); + + // The graph moves between pages: the pinned cursor must conflict, never splice. + await completeTurn(h, 's3', 'three'); + await h.inject({ + kind: 'conversation.read', + clientReqId: 'rr-stale', + sessionId: h.sessionId, + cursor: page1.cursor, + }); + expect(h.sent).toContainEqual( + expect.objectContaining({ kind: 'request.failed', replyTo: 'rr-stale', code: 'conflict' }), + ); + + // Garbage never maps to offset 0 (a silent restart). + await h.inject({ + kind: 'conversation.read', + clientReqId: 'rr-garbage', + sessionId: h.sessionId, + cursor: 'garbage', + }); + expect(h.sent).toContainEqual( + expect.objectContaining({ kind: 'request.failed', replyTo: 'rr-garbage', code: 'conflict' }), + ); + }); + + it('pages by the logical-message byte budget with the watermark on the final page only', async () => { + const h = await startedHarness(); + await completeTurn(h, 's1', big('a')); + await completeTurn(h, 's2', big('b')); + await completeTurn(h, 's3', big('c')); + await h.inject({ kind: 'session.stop', clientReqId: 'stop', sessionId: h.sessionId }); + + const pages = []; + let cursor: string | undefined; + for (let page = 0; page < 5; page++) { + const clientReqId = `rr-${page}`; + // eslint-disable-next-line no-await-in-loop -- cursor paging is sequential by nature. + await h.inject({ + kind: 'conversation.read', + clientReqId, + sessionId: h.sessionId, + ...(cursor !== undefined && { cursor }), + }); + const result = readResult(h.sent, clientReqId); + pages.push(result); + cursor = result.cursor; + if (cursor === undefined) break; + } + + expect(pages.length).toBeGreaterThan(1); + for (let i = 0, len = pages.length; i < len; i++) { + const page = pages[i]; + const final = i === pages.length - 1; + // Only the final page carries the merge watermark; earlier pages carry none. + expect(page.watermark === undefined).toBe(!final); + expect(page.cursor === undefined).toBe(final); + const pageBytes = page.events.reduce( + (sum, item) => sum + Buffer.byteLength(JSON.stringify(item), 'utf8'), + 0, + ); + expect(pageBytes).toBeLessThanOrEqual(MAX_ATTACHMENT_TOTAL_BASE64_LENGTH); + } + // Nothing is lost across the page split. + expect(pages.flatMap((page) => userTexts(page.events))).toEqual([big('a'), big('b'), big('c')]); + }, 30000); +}); + +/** Big enough that two prompts cannot share one page under the ~16 MiB budget. */ +function big(fill: string): string { + return fill.repeat(9 * 1024 * 1024); +} diff --git a/packages/host/engine/src/__tests__/engine-conversation-stubs.test.ts b/packages/host/engine/src/__tests__/engine-conversation-stubs.test.ts deleted file mode 100644 index cd798fa37..000000000 --- a/packages/host/engine/src/__tests__/engine-conversation-stubs.test.ts +++ /dev/null @@ -1,28 +0,0 @@ -import type { WirePayload } from '@linkcode/schema'; -import { describe, expect, it } from 'vitest'; -import { createSessionHarness } from './fixtures/session-harness'; - -const requests: WirePayload[] = [ - { kind: 'conversation.graph.get', clientReqId: 'r-graph', sessionId: 'session-1' }, - { kind: 'conversation.read', clientReqId: 'r-read', sessionId: 'session-1' }, -] as WirePayload[]; - -describe('conversation request stubs', () => { - it('refuses unimplemented conversation reads loudly instead of dropping them', async () => { - const h = createSessionHarness(); - await h.engine.start(); - - await Promise.all(requests.map((request) => h.inject(request))); - - const replyIds = ['r-graph', 'r-read']; - for (let i = 0, len = replyIds.length; i < len; i++) { - expect(h.sent).toContainEqual( - expect.objectContaining({ - kind: 'request.failed', - replyTo: replyIds[i], - code: 'unsupported', - }), - ); - } - }); -}); diff --git a/packages/host/engine/src/__tests__/history-service.test.ts b/packages/host/engine/src/__tests__/history-service.test.ts index 83e7ba4be..2ccea340b 100644 --- a/packages/host/engine/src/__tests__/history-service.test.ts +++ b/packages/host/engine/src/__tests__/history-service.test.ts @@ -54,6 +54,23 @@ describe('HistoryService', () => { expect(state.lastReadOptions?.mcpServerNames).toBeUndefined(); }); + it('invalidates converted history when its account configuration changes', async () => { + const state: FakeHistoryState = { listCalls: 0, readCalls: 0, resumeCalls: 0 }; + let root = '/claude-a'; + const service = new HistoryService(fakeHistoryFactory(state), { + ttlMs: 60000, + historyConfig: () => ({ extraEnv: { CLAUDE_CONFIG_DIR: root } }), + }); + await Effect.runPromise(service.read('claude-code', { historyId })); + await Effect.runPromise(service.read('claude-code', { historyId })); + expect(state.readCalls).toBe(1); + expect(state.lastReadOptions?.config).toEqual({ extraEnv: { CLAUDE_CONFIG_DIR: '/claude-a' } }); + root = '/claude-b'; + await Effect.runPromise(service.read('claude-code', { historyId })); + expect(state.readCalls).toBe(2); + expect(state.lastReadOptions?.config).toEqual({ extraEnv: { CLAUDE_CONFIG_DIR: '/claude-b' } }); + }); + it('evicts expired cache entries instead of keeping dead transcripts', async () => { const state: FakeHistoryState = { listCalls: 0, readCalls: 0, resumeCalls: 0 }; let now = 0; diff --git a/packages/host/engine/src/conversation/live-journal.ts b/packages/host/engine/src/conversation/live-journal.ts index e4aa96fbc..abf646b8c 100644 --- a/packages/host/engine/src/conversation/live-journal.ts +++ b/packages/host/engine/src/conversation/live-journal.ts @@ -18,12 +18,29 @@ interface JournalEntry { } const DEFAULT_JOURNAL_BYTE_CAP = 10 * 1024 * 1024; -const DEFAULT_JOURNAL_EVENT_CAP = 10_000; +const DEFAULT_JOURNAL_EVENT_CAP = 10000; +/** Distinct evicted in-flight streams remembered per journal; past it every retained chunk is + * treated as headless (an extreme-storm degradation, never unbounded growth). */ +const EVICTED_CHUNK_KEY_CAP = 1024; function stampOf(event: JournaledEvent): ConversationWatermark { return { epoch: event.epoch, seq: event.seq }; } +/** The stream identity of a delta-carrying event: rendering its tail without its head splices + * garbage, unlike full-snapshot events which replace by id. */ +export function inflightChunkKey(event: AgentEvent): string | undefined { + switch (event.type) { + case 'agent-message-chunk': + case 'agent-thought-chunk': + return event.messageId; + case 'tool-call-content-chunk': + return event.toolCallId; + default: + return undefined; + } +} + /** * Byte/event-bounded live tail of one session's stamped `agent.event` stream (the * `TerminalReplayJournal` pattern). Bounded is non-negotiable: an unthrottled chunk storm must @@ -38,6 +55,9 @@ export class ConversationLiveJournal { /** Highest position evicted by the caps; a watermark below it lost events it never saw. */ private evictedThrough: ConversationWatermark | undefined; private last: ConversationWatermark | undefined; + /** In-flight streams that lost their head to eviction ({@link inflightChunkKey}). */ + private readonly evictedChunkKeys = new Set(); + private evictedChunkKeysOverflowed = false; constructor( private readonly maxBytes = DEFAULT_JOURNAL_BYTE_CAP, @@ -65,6 +85,27 @@ export class ConversationLiveJournal { return this.entries.map(({ event }) => event); } + /** A settled turn is replayable only while its entire start-to-stop interval is retained. */ + completedTurn(turnId: TurnId, runId: RunId): JournaledEvent[] | undefined { + const start = this.entries.findIndex( + ({ event }) => + event.turnId === turnId && event.runId === runId && event.event.type === 'user-message', + ); + if (start < 0) return; + const first = this.entries[start].event; + const events: JournaledEvent[] = []; + let seq = first.seq; + for (let i = start, len = this.entries.length; i < len; i++) { + const entry = this.entries[i].event; + if (entry.epoch !== first.epoch || entry.seq !== seq) return; + seq += 1; + if (entry.turnId !== turnId) continue; + if (entry.runId !== runId) return; + if (entry.event.type !== 'user-message') events.push(entry); + if (entry.event.type === 'stop') return events; + } + } + append(event: JournaledEvent): void { const stamp = stampOf(event); this.first ??= stamp; @@ -81,6 +122,14 @@ export class ConversationLiveJournal { const removed = this.entries.shift(); if (!removed) break; this.byteCount -= removed.bytes; + const chunkKey = inflightChunkKey(removed.event.event); + if (chunkKey !== undefined) { + if (this.evictedChunkKeys.size >= EVICTED_CHUNK_KEY_CAP) { + this.evictedChunkKeysOverflowed = true; + } else { + this.evictedChunkKeys.add(chunkKey); + } + } const evicted = stampOf(removed.event); if ( this.evictedThrough === undefined || @@ -91,6 +140,12 @@ export class ConversationLiveJournal { } } + /** Whether a retained chunk of this in-flight stream lost earlier deltas to eviction — a reader + * must clear and restart that message, never splice a headless tail (CODE-35 class). */ + isChunkCleared(key: string): boolean { + return this.evictedChunkKeysOverflowed || this.evictedChunkKeys.has(key); + } + /** * Retained events above `watermark`, and whether that set is provably complete. `gap` means * events past the watermark were evicted or never reached this journal (an older epoch's tail): diff --git a/packages/host/engine/src/conversation/projection-service.ts b/packages/host/engine/src/conversation/projection-service.ts new file mode 100644 index 000000000..c58bfacd5 --- /dev/null +++ b/packages/host/engine/src/conversation/projection-service.ts @@ -0,0 +1,679 @@ +import { Buffer } from 'node:buffer'; +import { boundedLimit } from '@linkcode/agent-adapter'; +import type { + AgentEvent, + AgentHistoryEvent, + AgentHistoryId, + ContentBlock, + ConversationGraphTurn, + ConversationReadItem, + ConversationTurn, + ConversationWatermark, + SessionId, + SessionRecord, + TurnId, +} from '@linkcode/schema'; +import { + compareConversationWatermarks, + MAX_ATTACHMENT_TOTAL_BASE64_LENGTH, + MessageIdSchema, + TurnIdSchema, +} from '@linkcode/schema'; +import { Effect } from 'effect'; +import { appendArrayInPlace } from 'foxts/append-array-in-place'; +import { OperationError, RequestError } from '../failure'; +import type { HistoryService } from '../session/history-service'; +import { promptContentFingerprint } from '../session/live-session'; +import type { SessionRecordRegistry } from '../session/session-record-registry'; +import type { ConversationLiveJournals } from './live-journal'; +import { inflightChunkKey } from './live-journal'; +import type { ConversationTurnService } from './turn-service'; +import { TERMINAL_TURN_STATES } from './turn-service'; + +export interface ConversationGraphResult { + readonly sessionId: SessionId; + readonly graphRevision: number; + readonly activeLeafTurnId?: TurnId; + readonly turns: ConversationGraphTurn[]; +} + +export interface ConversationReadRequest { + readonly sessionId: SessionId; + readonly leafTurnId?: TurnId | undefined; + readonly cursor?: string | undefined; + readonly limit?: number | undefined; +} + +export interface ConversationReadResult { + readonly sessionId: SessionId; + readonly graphRevision: number; + readonly leafTurnId?: TurnId; + /** Present ONLY on the final page, together with the live tail. */ + readonly watermark?: ConversationWatermark; + readonly events: ConversationReadItem[]; + readonly cursor?: string; +} + +const INPUT_SUMMARY_MAX_LENGTH = 140; +const WHITESPACE_RUN_RE = /\s+/g; +/** One page = one logical tunnel message; oversized reassembly is silently dropped by the tunnel + * (the history-util.ts byte-budget rationale applies verbatim). */ +const READ_PAGE_BYTE_BUDGET = MAX_ATTACHMENT_TOTAL_BASE64_LENGTH; + +interface ProviderPartition { + readonly userRow: AgentHistoryEvent; + readonly rest: AgentHistoryEvent[]; +} + +/** + * Composes the root→leaf conversation projection: user rows from the durable ConversationStore + * (host truth — never provider history, never the journal), assistant/tool events from provider + * history, the live tail from the bounded live journal merged by `(epoch, seq)` stamp. Provider + * lossiness uses a complete retained live turn, or a prompt-only placeholder after eviction. + */ +export class ConversationProjectionService { + constructor( + private readonly turns: ConversationTurnService, + private readonly records: SessionRecordRegistry, + private readonly history: HistoryService, + private readonly journals: ConversationLiveJournals, + /** Authoritative open interactive requests of the live session (the CODE-35 backstop). */ + private readonly openRequests: (sessionId: SessionId) => AgentEvent[], + ) {} + + graph( + sessionId: SessionId, + ): Effect.Effect { + const { records, turns } = this; + const inputSummary = this.inputSummary.bind(this); + return Effect.gen(function* () { + const record = records.get(sessionId); + if (!record) { + return yield* Effect.fail( + new RequestError({ code: 'not_found', message: `Unknown session: ${sessionId}` }), + ); + } + const sessionTurns = yield* turns.listTurns(sessionId); + sessionTurns.sort(byCreation); + const graphTurns: ConversationGraphTurn[] = []; + for (let i = 0, len = sessionTurns.length; i < len; i++) { + const summary = yield* inputSummary(sessionTurns[i]); + graphTurns.push( + summary === undefined ? sessionTurns[i] : { ...sessionTurns[i], inputSummary: summary }, + ); + } + return { + sessionId, + graphRevision: record.graphRevision, + ...(record.activeLeafTurnId !== undefined && { + activeLeafTurnId: record.activeLeafTurnId, + }), + turns: graphTurns, + }; + }); + } + + read( + request: ConversationReadRequest, + ): Effect.Effect { + const { records, turns } = this; + const composeDurable = this.composeDurable.bind(this); + const composeTail = this.composeTail.bind(this); + return Effect.gen(function* () { + const record = records.get(request.sessionId); + if (!record) { + return yield* Effect.fail( + new RequestError({ + code: 'not_found', + message: `Unknown session: ${request.sessionId}`, + }), + ); + } + const sessionTurns = yield* turns.listTurns(request.sessionId); + const byId = new Map(sessionTurns.map((turn) => [turn.turnId, turn])); + if (request.leafTurnId !== undefined && !byId.has(request.leafTurnId)) { + return yield* Effect.fail( + new RequestError({ code: 'not_found', message: `Unknown turn: ${request.leafTurnId}` }), + ); + } + const leafTurnId = request.leafTurnId ?? record.activeLeafTurnId; + const path = pathToLeaf(byId, leafTurnId); + // A bare offset would splice across mutations (a settle flips the attribution gate, the + // leaf moves, garbage restarts silently): the cursor pins the exact projection shape it + // paged, and any drift or undecodable cursor is a typed conflict — never a silent splice. + const settled = path.filter((turn) => TERMINAL_TURN_STATES.has(turn.state)).length; + let offset = 0; + if (request.cursor !== undefined) { + const decoded = decodeReadCursor(request.cursor); + if ( + decoded?.graphRevision !== record.graphRevision || + decoded.leafTurnId !== leafTurnId || + decoded.settled !== settled + ) { + return yield* Effect.fail( + new RequestError({ + code: 'conflict', + message: 'The conversation changed while paging; restart the read', + }), + ); + } + offset = decoded.offset; + } + // Positional attribution is sound only on the active lineage: a sibling lineage has the + // same path length by construction (and can carry identical prompt text on a retry), so an + // inactive-leaf read renders host rows + placeholders until per-turn bindings (CODE-632). + const isActiveLineage = leafTurnId !== undefined && leafTurnId === record.activeLeafTurnId; + const durable = yield* composeDurable(record, path, isActiveLineage); + const { tail, watermark } = composeTail(request.sessionId, record.eventEpoch, path); + const { events, nextOffset } = pageReadItems( + durable, + tail, + offset, + boundedLimit(request.limit, 1000, 1000), + ); + const cursor = + nextOffset !== undefined && leafTurnId !== undefined + ? JSON.stringify({ + graphRevision: record.graphRevision, + leafTurnId, + settled, + offset: nextOffset, + }) + : undefined; + return { + sessionId: request.sessionId, + graphRevision: record.graphRevision, + ...(leafTurnId !== undefined && { leafTurnId }), + // ONLY the final page carries the live tail and the watermark the client merges against. + ...(cursor === undefined ? { watermark } : { cursor }), + events, + }; + }); + } + + /** Host user rows, attributed provider output, then complete retained turns or placeholders. */ + private composeDurable( + record: SessionRecord, + path: ConversationTurn[], + isActiveLineage: boolean, + ): Effect.Effect { + const { records, journals } = this; + const readProviderEvents = this.readProviderEvents.bind(this); + const hostUserContent = this.hostUserContent.bind(this); + return Effect.gen(function* () { + const items: ConversationReadItem[] = []; + const contents: Array = []; + for (let i = 0, len = path.length; i < len; i++) { + contents.push(yield* hostUserContent(path[i])); + } + const cold = path.filter((turn) => TERMINAL_TURN_STATES.has(turn.state)); + // A failed turn expects no provider rows (nothing durable ran) and gets no placeholder. + const expectsProvider = cold.filter((turn) => turn.state !== 'failed'); + const liveIndex = path.findIndex((turn) => !TERMINAL_TURN_STATES.has(turn.state)); + const historyId = records.historyId(record.sessionId); + let attributed: ProviderPartition[] = []; + let leading: AgentHistoryEvent[] = []; + if (isActiveLineage && historyId !== undefined && expectsProvider.length > 0) { + const corpus = yield* readProviderEvents(record, historyId); + if (corpus !== undefined) { + const hostFingerprints: Array = []; + for (let i = 0, len = path.length; i < len; i++) { + const turn = path[i]; + if (!TERMINAL_TURN_STATES.has(turn.state) || turn.state === 'failed') continue; + const content = contents[i]; + hostFingerprints.push(content && promptContentFingerprint(content)); + } + let liveFingerprint: string | undefined; + if (liveIndex >= 0) { + const liveContent = contents[liveIndex]; + if (liveContent) liveFingerprint = promptContentFingerprint(liveContent); + } + const result = attributeCorpus(corpus, hostFingerprints, liveFingerprint); + attributed = result.attributed; + leading = result.leading; + } + } + for (let i = 0, len = leading.length; i < len; i++) { + items.push(projectedItem(undefined, leading[i])); + } + let partitionIndex = 0; + for (let i = 0, len = path.length; i < len; i++) { + const turn = path[i]; + const content = contents[i]; + if (content !== undefined) items.push(projectedUserRow(turn, content)); + if (!TERMINAL_TURN_STATES.has(turn.state)) continue; // in-flight output rides the live tail + if (turn.state === 'failed') continue; // nothing durable ran; the state badge is the story + const partition = attributed[partitionIndex]; + partitionIndex += 1; + if (partition === undefined) { + const retained = journals.get(record.sessionId)?.completedTurn(turn.turnId, turn.runId); + if (retained === undefined) { + items.push({ type: 'history-unavailable', turnId: turn.turnId, runId: turn.runId }); + } else { + appendArrayInPlace(items, retained); + } + } else { + for (let j = 0, restLen = partition.rest.length; j < restLen; j++) { + items.push(projectedItem(turn, partition.rest[j])); + } + } + } + return items; + }); + } + + /** The live tail: retained journal events above the last event attributed to a settled path + * turn, minus user echoes (host rows own user display) and headless chunk streams, plus the + * authoritative open interactive requests. */ + private composeTail( + sessionId: SessionId, + eventEpoch: number, + path: ConversationTurn[], + ): { tail: ConversationReadItem[]; watermark: ConversationWatermark } { + const journal = this.journals.get(sessionId); + const liveTurn = path.find((turn) => !TERMINAL_TURN_STATES.has(turn.state)); + const tail: ConversationReadItem[] = []; + const seenRequestIds = new Set(); + // Journal-less sessions (cold, or a launch whose first event hasn't flowed) cut every prior + // epoch and NOTHING in the current one: seqs start at 1, so the run's own events all compare + // above {epoch, 0} — a client adopting this during the launch window drops nothing. + let watermark: ConversationWatermark = { epoch: eventEpoch, seq: 0 }; + if (journal) { + const snapshot = journal.snapshot(); + const terminalIds = new Set(); + for (let i = 0, len = path.length; i < len; i++) { + if (TERMINAL_TURN_STATES.has(path[i].state)) terminalIds.add(path[i].turnId); + } + // The durable/live boundary: everything at or below the last event of a settled path turn + // is already covered by durable sources (or superseded by their placeholders). + let cut: ConversationWatermark | undefined; + for (let i = 0, len = snapshot.length; i < len; i++) { + const entry = snapshot[i]; + if (entry.turnId === undefined || !terminalIds.has(entry.turnId)) continue; + const stamp = { epoch: entry.epoch, seq: entry.seq }; + if (cut === undefined || compareConversationWatermarks(stamp, cut) > 0) cut = stamp; + } + // tailAfter's gap semantics own the eviction question: eviction reaching ABOVE the cut may + // have destroyed full-snapshot events (a settled tool call, a resolution) for good. + const { events: aboveCut, gap } = + cut === undefined ? { events: snapshot, gap: journal.truncated } : journal.tailAfter(cut); + for (let i = 0, len = aboveCut.length; i < len; i++) { + const entry = aboveCut[i]; + const event = entry.event; + // User rows are host truth — a live echo must not double the durable row. + if (event.type === 'user-message') continue; + const chunkKey = inflightChunkKey(event); + if (chunkKey !== undefined && journal.isChunkCleared(chunkKey)) continue; + if (event.type === 'permission-request' || event.type === 'question-request') { + seenRequestIds.add(event.requestId); + } + tail.push({ + ...(entry.turnId !== undefined && { turnId: entry.turnId }), + runId: entry.runId, + epoch: entry.epoch, + seq: entry.seq, + ts: entry.ts, + event, + }); + } + // The journal returns append order — a stale old-epoch straggler can sit after newer + // entries; the projection merges by stamp, never array order. + tail.sort(byStamp); + // The retained tail is provably incomplete: what remains still renders, but the in-flight + // turn carries the placeholder so the read never claims completeness for it. + if (gap && liveTurn !== undefined) { + tail.push({ type: 'history-unavailable', turnId: liveTurn.turnId, runId: liveTurn.runId }); + } + if (journal.watermark !== undefined) watermark = journal.watermark; + } + // CODE-35 backstop: open interactive requests reach the reader even when their original + // events were evicted or fell below the durable cut. + const openRequests = this.openRequests(sessionId); + for (let i = 0, len = openRequests.length; i < len; i++) { + const request = openRequests[i]; + if (request.type !== 'permission-request' && request.type !== 'question-request') continue; + if (seenRequestIds.has(request.requestId)) continue; + tail.push({ + ...(liveTurn !== undefined && { turnId: liveTurn.turnId, runId: liveTurn.runId }), + event: request, + }); + } + return { tail, watermark }; + } + + /** The full provider corpus behind the TTL cache, or undefined when unreadable — unsupported + * harness, failed read (CODE-645), deleted transcript — so the caller degrades to prompt-only. */ + private readProviderEvents( + record: SessionRecord, + historyId: AgentHistoryId, + ): Effect.Effect { + const { history } = this; + const { cwd, kind, sessionId } = record; + // A cached corpus captured before the newest settle can miss that turn's rows (or hold its + // partial answer) — bypass it so a post-settle read never attributes a stale slice. + const freshAfter = this.turns.lastSettledAt(sessionId); + return Effect.gen(function* () { + const events: AgentHistoryEvent[] = []; + const seenCursors = new Set(); + let cursor: string | undefined; + do { + // cwd is load-bearing for codex: its rollout home resolves through the project env. + const result = yield* history.read(kind, { + historyId, + cwd, + cursor, + freshAfter, + limit: 1000, + }); + for (let i = 0, len = result.events.length; i < len; i++) events.push(result.events[i]); + cursor = result.cursor; + if (cursor !== undefined) { + if (seenCursors.has(cursor)) { + return yield* Effect.fail( + new OperationError({ + subsystem: 'agent', + operation: 'conversation.read.history', + publicMessage: 'Provider history read returned a repeated cursor', + cause: undefined, + }), + ); + } + seenCursors.add(cursor); + } + } while (cursor !== undefined); + return events; + }).pipe( + Effect.catch((error) => + (error instanceof OperationError + ? Effect.logWarning( + 'Provider history unavailable for conversation read', + { sessionId, operation: error.operation }, + error.cause, + ) + : Effect.void + ).pipe(Effect.as(undefined)), + ), + ); + } + + /** The turn's user-row content from host truth; undefined for migrated null-prompt turns + * (which render as placeholders until per-turn bindings land, CODE-632). */ + private hostUserContent( + turn: ConversationTurn, + ): Effect.Effect { + const input = turn.input; + if (input.type === 'command' || input.type === 'shell-command') { + return Effect.succeed([{ type: 'text' as const, text: inputText(input) }]); + } + if (input.promptId === null) return Effect.undefined; + return this.turns.getPrompt(input.promptId).pipe( + Effect.map((prompt) => { + if (!prompt) return; + // attachment_ref blocks join the projection when the attachment store lands. + return prompt.blocks.flatMap((block) => + block.type === 'text' ? [{ type: 'text' as const, text: block.text }] : [], + ); + }), + ); + } + + private inputSummary(turn: ConversationTurn): Effect.Effect { + const input = turn.input; + if (input.type === 'command' || input.type === 'shell-command') { + return Effect.succeed(truncateSummary(inputText(input))); + } + if (input.promptId === null) return Effect.undefined; + return this.turns.getPrompt(input.promptId).pipe( + Effect.map((prompt) => { + if (!prompt) return; + const text = prompt.blocks + .flatMap((block) => (block.type === 'text' ? [block.text] : [])) + .join(' ') + .replaceAll(WHITESPACE_RUN_RE, ' ') + .trim(); + return text.length === 0 ? undefined : truncateSummary(text); + }), + ); + } +} + +/** + * Slices the composed projection into pages under the shared byte budget: durable items page by + * budget and `limit`; the live tail is atomic to the final page (trimmed from the front when it + * alone exceeds the budget, clearing any in-flight stream whose deltas were dropped). The first + * item of a page always ships, so pagination cannot stall on one oversized record. + */ +export function pageReadItems( + durable: readonly ConversationReadItem[], + tail: readonly ConversationReadItem[], + offset: number, + limit: number, + budget = READ_PAGE_BYTE_BUDGET, +): { events: ConversationReadItem[]; nextOffset?: number } { + const page: ConversationReadItem[] = []; + let pageBytes = 0; + let index = Math.min(offset, durable.length); + for (const len = durable.length; index < len; index += 1) { + if (page.length >= limit) break; + const size = itemBytes(durable[index]); + if (page.length > 0 && pageBytes + size > budget) break; + pageBytes += size; + page.push(durable[index]); + } + if (index < durable.length) return { events: page, nextOffset: index }; + const trimmedTail = trimTailToBudget(tail, budget); + const tailBytes = trimmedTail.reduce((sum, item) => sum + itemBytes(item), 0); + if (page.length > 0 && trimmedTail.length > 0 && pageBytes + tailBytes > budget) { + return { events: page, nextOffset: index }; + } + return { events: [...page, ...trimmedTail] }; +} + +interface ReadCursor { + readonly graphRevision: number; + readonly leafTurnId: TurnId; + readonly settled: number; + readonly offset: number; +} + +function decodeReadCursor(raw: string): ReadCursor | undefined { + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch { + return undefined; + } + if ( + typeof parsed !== 'object' || + parsed === null || + !('graphRevision' in parsed) || + typeof parsed.graphRevision !== 'number' || + !('settled' in parsed) || + typeof parsed.settled !== 'number' || + !('offset' in parsed) || + typeof parsed.offset !== 'number' || + !Number.isSafeInteger(parsed.offset) || + parsed.offset < 0 || + !('leafTurnId' in parsed) + ) { + return undefined; + } + const leaf = TurnIdSchema.safeParse(parsed.leafTurnId); + if (!leaf.success) return undefined; + return { + graphRevision: parsed.graphRevision, + leafTurnId: leaf.data, + settled: parsed.settled, + offset: parsed.offset, + }; +} + +function trimTailToBudget( + tail: readonly ConversationReadItem[], + budget: number, +): ConversationReadItem[] { + let total = tail.reduce((sum, item) => sum + itemBytes(item), 0); + if (total <= budget) return [...tail]; + const clearedKeys = new Set(); + let start = 0; + while (total > budget && start < tail.length - 1) { + const removed = tail[start]; + start += 1; + total -= itemBytes(removed); + if ('event' in removed) { + const key = inflightChunkKey(removed.event); + if (key !== undefined) clearedKeys.add(key); + } + } + const items = tail.slice(start); + if (clearedKeys.size === 0) return items; + // Same rule as journal eviction: a stream that lost deltas restarts, never splices headless. + return items.filter((item) => { + if (!('event' in item)) return true; + const key = inflightChunkKey(item.event); + return key === undefined || !clearedKeys.has(key); + }); +} + +function itemBytes(item: ConversationReadItem): number { + return Buffer.byteLength(JSON.stringify(item), 'utf8'); +} + +/** Root→leaf path through `parentTurnId`; a broken chain fails loud rather than rendering wrong. */ +function pathToLeaf( + byId: Map, + leafTurnId: TurnId | undefined, +): ConversationTurn[] { + if (leafTurnId === undefined) return []; + const path: ConversationTurn[] = []; + const seen = new Set(); + let currentId: TurnId | null = leafTurnId; + while (currentId !== null) { + if (seen.has(currentId)) { + throw new RequestError({ code: 'conflict', message: 'The turn graph contains a cycle' }); + } + seen.add(currentId); + const turn = byId.get(currentId); + if (!turn) { + throw new RequestError({ code: 'conflict', message: `Missing turn in path: ${currentId}` }); + } + path.push(turn); + currentId = turn.parentTurnId; + } + return path.reverse(); +} + +/** + * The attribution gate (§9 discipline): positions attribute only while each partition's user row + * fingerprint-matches the host prompt at that position; the FIRST mismatch degrades that turn and + * every later one to placeholders — alignment is lost past a mismatch, never resynced positionally. + * One trailing extra partition is tolerated only when it fingerprint-verifies as the in-flight + * turn's own row (the live tail owns it); any other count anomaly attributes nothing. + */ +function attributeCorpus( + corpus: readonly AgentHistoryEvent[], + hostFingerprints: ReadonlyArray, + liveFingerprint: string | undefined, +): { attributed: ProviderPartition[]; leading: AgentHistoryEvent[] } { + const none = { attributed: [], leading: [] }; + const split = partitionAtUserRows(corpus); + let candidates = split.partitions; + const trailing = candidates.at(-1); + if (trailing !== undefined && candidates.length === hostFingerprints.length + 1) { + if (liveFingerprint === undefined || userRowFingerprint(trailing.userRow) !== liveFingerprint) { + return none; + } + candidates = candidates.slice(0, -1); + } + if (candidates.length !== hostFingerprints.length) return none; + const attributed: ProviderPartition[] = []; + for (let i = 0, len = candidates.length; i < len; i++) { + const hostFingerprint = hostFingerprints[i]; + if ( + hostFingerprint === undefined || + userRowFingerprint(candidates[i].userRow) !== hostFingerprint + ) { + break; + } + attributed.push(candidates[i]); + } + return { attributed, leading: attributed.length > 0 ? split.leading : [] }; +} + +function userRowFingerprint(entry: AgentHistoryEvent): string | undefined { + return entry.event.type === 'user-message' + ? promptContentFingerprint(entry.event.content) + : undefined; +} + +/** Splits a provider corpus at its user rows: partition i is user row i plus what follows it. */ +function partitionAtUserRows(corpus: readonly AgentHistoryEvent[]): { + leading: AgentHistoryEvent[]; + partitions: ProviderPartition[]; +} { + const leading: AgentHistoryEvent[] = []; + const partitions: ProviderPartition[] = []; + for (let i = 0, len = corpus.length; i < len; i++) { + const entry = corpus[i]; + if (entry.event.type === 'user-message') { + partitions.push({ userRow: entry, rest: [] }); + } else { + const current = partitions.at(-1); + if (current === undefined) leading.push(entry); + else current.rest.push(entry); + } + } + return { leading, partitions }; +} + +function projectedItem( + turn: ConversationTurn | undefined, + entry: AgentHistoryEvent, +): ConversationReadItem { + return { + ...(turn !== undefined && { turnId: turn.turnId, runId: turn.runId }), + ...(entry.ts !== undefined && { ts: entry.ts }), + event: entry.event, + }; +} + +function projectedUserRow(turn: ConversationTurn, content: ContentBlock[]): ConversationReadItem { + return { + turnId: turn.turnId, + runId: turn.runId, + ts: turn.createdAt, + event: { + type: 'user-message', + // Deterministic identity: re-reads and page overlaps converge on one row per turn. + messageId: MessageIdSchema.parse(`msg-${turn.turnId}`), + content, + }, + }; +} + +function inputText( + input: Extract, +): string { + return input.type === 'command' + ? `/${input.name}${input.arguments === undefined ? '' : ` ${input.arguments}`}` + : `$ ${input.command}`; +} + +function truncateSummary(text: string): string { + return text.length > INPUT_SUMMARY_MAX_LENGTH + ? `${text.slice(0, INPUT_SUMMARY_MAX_LENGTH - 1)}…` + : text; +} + +function byCreation(a: ConversationTurn, b: ConversationTurn): number { + return a.createdAt - b.createdAt || a.turnId.localeCompare(b.turnId); +} + +/** Lexicographic stamp order for journal-derived tail items (all stamped at sort time). */ +function byStamp(a: ConversationReadItem, b: ConversationReadItem): number { + if (!('event' in a) || !('event' in b)) return 0; + return compareConversationWatermarks( + { epoch: a.epoch ?? 0, seq: a.seq ?? 0 }, + { epoch: b.epoch ?? 0, seq: b.seq ?? 0 }, + ); +} diff --git a/packages/host/engine/src/conversation/request-handler.ts b/packages/host/engine/src/conversation/request-handler.ts index 02b45b496..36afebd56 100644 --- a/packages/host/engine/src/conversation/request-handler.ts +++ b/packages/host/engine/src/conversation/request-handler.ts @@ -2,64 +2,100 @@ import type { WirePayload } from '@linkcode/schema'; import type { Transport } from '@linkcode/transport'; import { createWireMessage } from '@linkcode/transport'; import { Effect } from 'effect'; -import { RequestError } from '../failure'; import type { SessionLifecycleService } from '../session/lifecycle-service'; import type { WireResponder } from '../wire/responder'; +import type { ConversationProjectionService } from './projection-service'; type ConversationRequest = Extract< WirePayload, { kind: 'turn.submit' | 'conversation.graph.get' | 'conversation.read' } >; -/** Wire surface for the turn tree. `turn.submit` runs the submit saga; the read/projection kinds - * keep failing loudly — never silently ignored — until the projection lands. */ +/** Wire surface for the turn tree: `turn.submit` runs the submit saga; `conversation.graph.get` + * and `conversation.read` serve the host-composed projection. */ export class ConversationRequestHandler { constructor( private readonly transport: Transport, private readonly lifecycle: SessionLifecycleService, + private readonly projection: ConversationProjectionService, private readonly responder: WireResponder, ) {} handle(payload: ConversationRequest): Effect.Effect { - if (payload.kind !== 'turn.submit') { - return this.responder.reply( - payload.clientReqId, - Effect.fail( - new RequestError({ - code: 'unsupported', - message: `${payload.kind} is not implemented yet`, - }), - ), - ); - } - return this.responder.reply( - payload.clientReqId, - this.lifecycle.submitTurn(payload).pipe( - Effect.flatMap((operation) => - Effect.sync(() => { - // A stored failure replays verbatim: its code/message ARE the terminal result. - this.transport.send( - createWireMessage( - operation.state === 'succeeded' - ? { - kind: 'turn.submitted', - replyTo: payload.clientReqId, - turnId: operation.turnId, - } - : { - kind: 'request.failed', + switch (payload.kind) { + case 'conversation.graph.get': + return this.responder.reply( + payload.clientReqId, + this.projection.graph(payload.sessionId).pipe( + Effect.flatMap((result) => + Effect.sync(() => { + this.transport.send( + createWireMessage({ + kind: 'conversation.graph.result', + replyTo: payload.clientReqId, + ...result, + }), + ); + }), + ), + ), + ); + case 'conversation.read': + return this.responder.reply( + payload.clientReqId, + this.projection + .read({ + sessionId: payload.sessionId, + leafTurnId: payload.leafTurnId, + cursor: payload.cursor, + limit: payload.limit, + }) + .pipe( + Effect.flatMap((result) => + Effect.sync(() => { + this.transport.send( + createWireMessage({ + kind: 'conversation.read.result', replyTo: payload.clientReqId, - code: operation.error.code, - message: operation.error.message, - ...(operation.error.reportedInConversation && { - reportedInConversation: true, - }), - }, + ...result, + }), + ); + }), ), - ); - }), - ), - ), - ); + ), + ); + case 'turn.submit': + return this.responder.reply( + payload.clientReqId, + this.lifecycle.submitTurn(payload).pipe( + Effect.flatMap((operation) => + Effect.sync(() => { + // A stored failure replays verbatim: its code/message ARE the terminal result. + this.transport.send( + createWireMessage( + operation.state === 'succeeded' + ? { + kind: 'turn.submitted', + replyTo: payload.clientReqId, + turnId: operation.turnId, + } + : { + kind: 'request.failed', + replyTo: payload.clientReqId, + code: operation.error.code, + message: operation.error.message, + ...(operation.error.reportedInConversation && { + reportedInConversation: true, + }), + }, + ), + ); + }), + ), + ), + ); + default: + return Effect.void; + } } } diff --git a/packages/host/engine/src/conversation/turn-service.ts b/packages/host/engine/src/conversation/turn-service.ts index 074cc68ab..f3a966cde 100644 --- a/packages/host/engine/src/conversation/turn-service.ts +++ b/packages/host/engine/src/conversation/turn-service.ts @@ -75,7 +75,11 @@ export type TerminalOperation = readonly error: TurnFailure; }); -const TERMINAL_TURN_STATES = new Set(['completed', 'failed', 'cancelled']); +export const TERMINAL_TURN_STATES = new Set([ + 'completed', + 'failed', + 'cancelled', +]); interface RunningTurn { readonly turn: ConversationTurn; @@ -91,6 +95,9 @@ interface RunningTurn { export class ConversationTurnService { /** The running turn per session; settles are addressed by the turn's own runId. */ private readonly running = new Map(); + /** When a turn last flipped terminal — the projection's cache-freshness bound: a provider + * corpus captured before the newest settle may be missing that turn's rows. */ + private readonly settledAt = new Map(); constructor( private readonly store: ConversationStore, @@ -115,6 +122,14 @@ export class ConversationTurnService { return storeOperation('conversation.turns.list', () => this.store.listTurns(sessionId)); } + getPrompt(promptId: PromptId): Effect.Effect { + return storeOperation('conversation.prompt.get', () => this.store.getPrompt(promptId)); + } + + lastSettledAt(sessionId: SessionId): number | undefined { + return this.settledAt.get(sessionId); + } + listBindings(turnId: TurnId): Effect.Effect { return storeOperation('conversation.bindings.list', () => this.store.listBindings(turnId)); } @@ -126,6 +141,7 @@ export class ConversationTurnService { Effect.tap(() => Effect.sync(() => { this.running.delete(sessionId); + this.settledAt.delete(sessionId); }), ), ); @@ -373,6 +389,7 @@ export class ConversationTurnService { /** Settles run off synchronous adapter callbacks, so persistence is enqueued best-effort. */ private persistTurnState(turn: ConversationTurn, state: ConversationTurnState): void { if (TERMINAL_TURN_STATES.has(turn.state)) return; + this.settledAt.set(turn.sessionId, Date.now()); this.runTask( storeOperation('conversation.turn.save', () => this.store.saveTurn({ ...turn, state })).pipe( Effect.catch((error) => diff --git a/packages/host/engine/src/engine.ts b/packages/host/engine/src/engine.ts index 92cd1e595..ad2155134 100644 --- a/packages/host/engine/src/engine.ts +++ b/packages/host/engine/src/engine.ts @@ -8,7 +8,7 @@ import { Cause, Effect, FiberSet } from 'effect'; import { CustomMcpServerService } from './agent/custom-mcp-service'; import { adoptDetectedLogins } from './agent/detected-logins'; import { AgentLoginService } from './agent/login-service'; -import { InMemoryProviderConfigStore } from './agent/provider-config'; +import { applyProviderDefaults, InMemoryProviderConfigStore } from './agent/provider-config'; import { AgentRequestHandler } from './agent/request-handler'; import { AgentRuntimeService } from './agent/runtime-service'; import { ManagedAssetService } from './asset/service'; @@ -24,6 +24,7 @@ import { BrowserReplHost } from './browser/repl-host'; import { BrowserRequestHandler } from './browser/request-handler'; import { InMemoryConversationStore } from './conversation/conversation-store'; import { ConversationLiveJournals } from './conversation/live-journal'; +import { ConversationProjectionService } from './conversation/projection-service'; import { ConversationRequestHandler } from './conversation/request-handler'; import { ConversationTurnService } from './conversation/turn-service'; import type { EngineDeps } from './deps'; @@ -116,6 +117,19 @@ export const createEngineRuntime = Effect.fn('Engine.create')(function* ( ); const history = new HistoryService(factory, { injectedMcpServerNames: (kind) => startOptions.injectedMcpServerNames(kind), + historyConfig(kind, historyId) { + for (const record of records.values()) { + if (record.kind !== kind) continue; + const run = record.runs.findLast((candidate) => candidate.historyId === historyId); + if (run) { + return applyProviderDefaults( + { kind, cwd: record.cwd, accountId: run.accountId }, + providerStore.get(), + providerStore.getAccounts(), + ).options.config; + } + } + }, }); const runtimes = yield* AgentRuntimeService.make( { @@ -231,9 +245,17 @@ export const createEngineRuntime = Effect.fn('Engine.create')(function* ( sessionLifecycle, responder, ); + const conversationProjection = new ConversationProjectionService( + conversationTurns, + records, + history, + conversationJournals, + (sessionId) => sessions.openInteractiveRequests(sessionId), + ); const conversationRequests = new ConversationRequestHandler( transport, sessionLifecycle, + conversationProjection, responder, ); const scheduler = new ScheduleService( diff --git a/packages/host/engine/src/session/history-service.ts b/packages/host/engine/src/session/history-service.ts index 1babaa14f..5e4e73a14 100644 --- a/packages/host/engine/src/session/history-service.ts +++ b/packages/host/engine/src/session/history-service.ts @@ -26,6 +26,9 @@ export type HistoryListOptions = AgentHistoryListOptions & { export type HistoryReadOptions = AgentHistoryReadOptions & { forceRefresh?: boolean; + /** Bypass a cache entry built at or before this timestamp — the caller knows the corpus moved + * (e.g. a turn settled) and a same-or-older capture may be missing rows. */ + freshAfter?: number; }; export interface HistoryServiceOptions { @@ -34,6 +37,7 @@ export interface HistoryServiceOptions { /** MCP server names the engine injects at session start (start-options-resolver) — passed to * cold reads so replayed calls to injected servers resolve like config-declared ones. */ injectedMcpServerNames?: (kind: AgentKind) => readonly string[]; + historyConfig?: (kind: AgentKind, historyId: AgentHistoryId) => StartOptions['config']; } interface ListCacheEntry { @@ -42,7 +46,9 @@ interface ListCacheEntry { } interface EventCacheEntry { + configFingerprint: string | undefined; expiresAt: number; + builtAt: number; version: number; session: AgentHistorySession; events: AgentHistoryEvent[]; @@ -57,6 +63,7 @@ export class HistoryService { private readonly ttlMs: number; private readonly now: () => number; private readonly injectedMcpServerNames?: (kind: AgentKind) => readonly string[]; + private readonly historyConfig?: HistoryServiceOptions['historyConfig']; constructor( private readonly factory: AdapterFactory, @@ -65,6 +72,7 @@ export class HistoryService { this.ttlMs = opts.ttlMs ?? 30000; this.now = opts.now ?? Date.now; this.injectedMcpServerNames = opts.injectedMcpServerNames; + this.historyConfig = opts.historyConfig; } list( @@ -117,6 +125,8 @@ export class HistoryService { const limit = boundedLimit(opts.limit, 1000, 1000); const key = eventCacheKey(kind, opts.historyId); const cwd = opts.cwd ?? this.historyCwdById.get(key); + const config = this.historyConfig?.(kind, opts.historyId); + const configFingerprint = JSON.stringify(config); const now = this.now(); this.sweepExpired(now); const cached = this.eventCache.get(key); @@ -125,6 +135,9 @@ export class HistoryService { cached && !opts.forceRefresh && cached.expiresAt > now && + cached.configFingerprint === configFingerprint && + // Same-millisecond builds count as stale: the settle/build order is unknowable then. + (opts.freshAfter === undefined || cached.builtAt > opts.freshAfter) && cached.version === HISTORY_CONVERSION_CACHE_VERSION && (!cached.partialCursor || offset < cached.events.length) ) { @@ -142,6 +155,7 @@ export class HistoryService { } const mcpServerNames = this.injectedMcpServerNames?.(kind); const readContext = { + ...(config && { config }), ...(cwd && { cwd }), ...(mcpServerNames?.length && { mcpServerNames }), }; @@ -151,7 +165,9 @@ export class HistoryService { Effect.map(sanitizeHistoryResult), Effect.flatMap((fullResult) => { const entry: EventCacheEntry = { + configFingerprint, expiresAt: now + this.ttlMs, + builtAt: now, version: HISTORY_CONVERSION_CACHE_VERSION, session: fullResult.session, events: [...fullResult.events], @@ -330,8 +346,10 @@ function agentHistoryOperation( }); } -function stripForceRefresh(opts: T): Omit { - const { forceRefresh: _forceRefresh, ...rest } = opts; +function stripForceRefresh( + opts: T, +): Omit { + const { forceRefresh: _forceRefresh, freshAfter: _freshAfter, ...rest } = opts; return rest; } diff --git a/packages/host/engine/src/session/orchestrator.ts b/packages/host/engine/src/session/orchestrator.ts index 070723792..1593975fd 100644 --- a/packages/host/engine/src/session/orchestrator.ts +++ b/packages/host/engine/src/session/orchestrator.ts @@ -110,6 +110,16 @@ export class SessionOrchestrator { if (session) this.events.broadcast(sessionId, session, session.replay()); } + /** Authoritative open/responding interactive requests — the CODE-35 backstop: a conversation + * read must carry them even when the journal evicted or cut their original events. */ + openInteractiveRequests(sessionId: SessionId): AgentEvent[] { + const session = this.sessions.get(sessionId); + if (!session) return []; + return session.interactions + .replay() + .filter((event) => event.type === 'permission-request' || event.type === 'question-request'); + } + sendInput( sessionId: SessionId, input: AgentInput, diff --git a/packages/host/engine/src/session/session-event-processor.ts b/packages/host/engine/src/session/session-event-processor.ts index bc23996fa..158c14668 100644 --- a/packages/host/engine/src/session/session-event-processor.ts +++ b/packages/host/engine/src/session/session-event-processor.ts @@ -43,8 +43,12 @@ export class SessionEventProcessor { private readonly journals: ConversationLiveJournals, ) {} - broadcast(sessionId: SessionId, session: LiveSession, events: Iterable): void { - const turnId = this.turns.runningTurnId(sessionId, session.runId); + broadcast( + sessionId: SessionId, + session: LiveSession, + events: Iterable, + turnId = this.turns.runningTurnId(sessionId, session.runId), + ): void { for (const event of events) this.send(sessionId, session, event, turnId); } diff --git a/packages/host/engine/src/session/session-input-dispatcher.ts b/packages/host/engine/src/session/session-input-dispatcher.ts index ff79d2295..88584ec9a 100644 --- a/packages/host/engine/src/session/session-input-dispatcher.ts +++ b/packages/host/engine/src/session/session-input-dispatcher.ts @@ -118,22 +118,33 @@ export class SessionInputDispatcher { } const persisted = intent; const dispatch = Effect.gen(function* () { - // Echo before awaiting send: provider events can outrun the dispatch acknowledgement. + // Echo before awaiting send: provider events can outrun the acknowledgement. The + // persisted intent supplies its turn ID before the adapter signals running. if (promptMessageId !== undefined && input.type === 'prompt') { - events.broadcast(sessionId, session, session.trackPrompt(promptMessageId, input.content)); + events.broadcast( + sessionId, + session, + session.trackPrompt(promptMessageId, input.content), + persisted?.turn.turnId, + ); records.setTitleFromContent(sessionId, input.content); } else if (input.type === 'command' || input.type === 'shell-command') { const text = input.type === 'command' ? `/${input.name}${input.arguments ? ` ${input.arguments}` : ''}` : `$ ${input.command}`; - events.broadcast(sessionId, session, [ - { - type: 'user-message', - messageId: nextMessageId(), - content: [{ type: 'text', text }], - }, - ]); + events.broadcast( + sessionId, + session, + [ + { + type: 'user-message', + messageId: nextMessageId(), + content: [{ type: 'text', text }], + }, + ], + persisted?.turn.turnId, + ); } const responseInput = input.type === 'permission-response' || input.type === 'question-response'