diff --git a/src/index.ts b/src/index.ts index 9fa458e..1889cd1 100644 --- a/src/index.ts +++ b/src/index.ts @@ -77,6 +77,8 @@ export type { ClaudeCliEvent } from './provider/claude-cli.ts'; export { decide, fold, initialState, TURN_EVENT_SCHEMA_VERSION } from './orchestrate/turn-engine.ts'; export type { TurnCommand, TurnEvent as DurableTurnEvent, TurnEventKind as DurableTurnEventKind, OrchestratorState, QueueState, UsageMeters } from './orchestrate/turn-engine.ts'; export { EventSink } from './orchestrate/event-sink.ts'; +export { createSqliteExecutionAuthority, ExecutionAuthority, ExecutionAuthorityBusyError } from './orchestrate/execution-authority.ts'; +export type { ExecutionAuthorityDeps, ExecutionAttempt, ExecutionRead, RecoveryResult } from './orchestrate/execution-authority.ts'; export type { Effect, EventRow, OutboxRow } from './orchestrate/event-sink.ts'; export { EffectWorker } from './orchestrate/effect-worker.ts'; export type { EffectWorkerDeps } from './orchestrate/effect-worker.ts'; diff --git a/src/loop.ts b/src/loop.ts index fb4fce8..427955d 100644 --- a/src/loop.ts +++ b/src/loop.ts @@ -5,9 +5,34 @@ * via the persistence hook. */ import type { Provider, ProviderMessage, ProviderTool } from './provider/index.ts'; import { ProviderError } from './provider/index.ts'; +import type { TextBlockActivity } from './provider/activity.ts'; +import { createBlockTracker } from './provider/activity.ts'; + +/** Loop failure codes. Budget/abort/invalid are loop-owned outcomes; + * PROVIDER_* codes classify provider failures (chat-reliability F3: the loop + * must never relabel a permanent provider rejection as exhausted runway). + * The `string & {}` member passes provider-owned codes through unchanged + * (e.g. PROVIDER_BAD_REQUEST, PROVIDER_UNAVAILABLE, MCP_PENDING). */ +export type LoopErrorCode = + | 'BUDGET_EXCEEDED' + | 'ABORTED' + | 'TOOL_CALL_INVALID' + | 'PROVIDER_TIMEOUT' + | 'PROVIDER_FAILED' + | (string & {}); + +/** What actually terminated the loop. Consumers key continuation eligibility + * (budget stash vs honest failure) off the cause, not the message string. */ +export type LoopErrorCause = + | 'budget-hop' + | 'budget-wall' + | 'budget-token' + | 'caller-abort' + | 'provider' + | 'tool-invalid'; export class LoopError extends Error { - constructor(message: string, readonly code: 'BUDGET_EXCEEDED' | 'ABORTED' | 'TOOL_CALL_INVALID', readonly partial: ProviderMessage[]) { + constructor(message: string, readonly code: LoopErrorCode, readonly partial: ProviderMessage[], override readonly cause?: LoopErrorCause, readonly status?: number) { super(message); this.name = 'LoopError'; } @@ -62,10 +87,18 @@ export interface LoopDeps { onDelta?: (delta: string) => void; /** R2-F4: provider-scoped session key (claude-cli --resume isolation) */ sessionKey?: string; - /** structured tool events (claude-cli harness + registry loop) */ - onTool?: (e: { phase: 'call' | 'result'; name: string; arg?: string; result?: string; isError?: boolean }) => void; + /** structured tool events (claude-cli harness + registry loop). + * invocationId: the provider-NATIVE call id when one exists (runtime + * registry tools carry it from the assistant tool_calls); durability keys + * per INVOCATION, never per tool name. */ + onTool?: (e: { phase: 'call' | 'result'; name: string; invocationId?: string; arg?: string; result?: string; isError?: boolean }) => void; /** R3 decision 4: opt INTO the hard token cap (fan-out children); default advisory */ hardTokenCap?: boolean; + /** chat-reliability F2 (text block lifecycle): per-block activity for the + * durable chunk rail — start/delta/end per assistant message (hop), so a + * replayed log can separate consecutive hops' narration instead of gluing + * it. Purely additive: absent → behavior identical to before. */ + onActivity?: (a: TextBlockActivity) => void; signal?: AbortSignal; } @@ -85,38 +118,83 @@ export async function runLoop(systemPrompt: string, userMessage: string, deps: L ]; const tokens = { in: 0, out: 0 }; let repairs = 0; + // chat-reliability F2: the block tracker runs only when a consumer asked + // for block activity; block ids are hop-scoped and stable per run. + const tracker = deps.onActivity ? createBlockTracker(deps.onActivity, (h) => `blk_h${h}`) : undefined; for (let hop = 1; hop <= budget.maxHops; hop++) { - if (deps.signal?.aborted) throw new LoopError('aborted by caller', 'ABORTED', messages); + if (deps.signal?.aborted) throw new LoopError('aborted by caller', 'ABORTED', messages, 'caller-abort'); const remainingMs = budget.wallMs - (Date.now() - started); - if (remainingMs <= 0) throw new LoopError(`wall budget exhausted after ${Date.now() - started}ms`, 'BUDGET_EXCEEDED', messages); + if (remainingMs <= 0) throw new LoopError(`wall budget exhausted after ${Date.now() - started}ms`, 'BUDGET_EXCEEDED', messages, 'budget-wall'); deps.onEvent?.({ type: 'turn.start', hop }); // the LOOP owns the wall: even a provider that ignores remainingMs gets - // aborted at the deadline (mock-hang proof, review F5) - const wallSignal = AbortSignal.any([AbortSignal.timeout(remainingMs), ...(deps.signal ? [deps.signal] : [])]); + // aborted at the deadline (mock-hang proof, review F5). F3: the deadline + // stays a SEPARATE signal so the catch can attribute the failure from + // observed signal state instead of the exception name; composition with + // the caller signal happens only here, at the provider seam. + const wallDeadline = AbortSignal.timeout(remainingMs); + const wallSignal = AbortSignal.any(deps.signal ? [wallDeadline, deps.signal] : [wallDeadline]); + // chat-reliability identity contract: harness tool events without a + // provider-native id get one ALLOCATED at first observation — never key + // durability by tool name (repeated same-name invocations must stay + // distinct). Pairing call→result is FIFO per name WITHIN this provider + // call only (a deterministic window); an unpaired result stays uncorrelated. + const harnessPending = new Map(); + let harnessSeq = 0; let res; + // F2: the LOOP accumulates the streamed text itself (sum of deltas) — the + // completed message's content may differ from the stream (claude-cli + // snapshot vs partial stream), so text.end must not trust content. + let streamedText = ''; try { // streaming: the EVENT always flows (WS UIs); the direct callback stays // opt-in for hosts that want it without the event bus - res = await deps.provider.complete(messages, toolDefs(deps.tools), { remainingMs, signal: wallSignal, ...(deps.sessionKey !== undefined ? { sessionKey: deps.sessionKey } : {}), onDelta: (d: string) => { deps.onDelta?.(d); deps.onEvent?.({ type: 'turn.delta', hop, detail: { delta: d } }); }, onTool: (e: { phase: 'call' | 'result'; name: string; arg?: string; result?: string; isError?: boolean }) => { deps.onTool?.(e); deps.onEvent?.({ type: 'turn.tool', hop, detail: e }); } }); + res = await deps.provider.complete(messages, toolDefs(deps.tools), { remainingMs, signal: wallSignal, ...(deps.sessionKey !== undefined ? { sessionKey: deps.sessionKey } : {}), onDelta: (d: string) => { streamedText += d; const bid = tracker?.onDelta(hop, d); deps.onDelta?.(d); deps.onEvent?.({ type: 'turn.delta', hop, detail: { delta: d, ...(bid !== undefined ? { blockId: bid } : {}) } }); }, onTool: (e: { phase: 'call' | 'result'; name: string; invocationId?: string; arg?: string; result?: string; isError?: boolean }) => { + let out = e; + if (e.invocationId === undefined) { + const pending = harnessPending.get(e.name) ?? []; + if (e.phase === 'call') { + const id = `alloc_${hop}_${harnessSeq++}`; + pending.push(id); + harnessPending.set(e.name, pending); + out = { ...e, invocationId: id }; + } else { + const id = pending.shift(); + harnessPending.set(e.name, pending); + if (id !== undefined) out = { ...e, invocationId: id }; + } + } + deps.onTool?.(out); + deps.onEvent?.({ type: 'turn.tool', hop, detail: out }); + } }); } catch (err) { - // caller abort vs wall timeout: AbortSignal.timeout raises TimeoutError; - // a caller signal abort propagates AbortError (or a custom reason) - const isWallTimeout = err instanceof Error && err.name === 'TimeoutError'; - if (deps.signal?.aborted && !isWallTimeout) throw new LoopError('aborted by caller', 'ABORTED', messages); - // provider exhausted its own retries/timeouts — surface with partial transcript - const pe = err instanceof ProviderError ? err : null; - const msg = pe ? `${pe.code}: ${pe.message}` - : err instanceof Error && (err.name === 'TimeoutError' || err.name === 'AbortError') ? 'PROVIDER_TIMEOUT: wall deadline reached during provider call' - : `provider failed: ${err instanceof Error ? err.message : String(err)}`; - throw new LoopError(msg, 'BUDGET_EXCEEDED', messages); + // F3 classification, by precedence: an observed caller abort wins + // (explicit Stop beats everything, even a concurrent wall expiry), then + // the loop-owned wall, then the provider's own typed failure, then + // unattributed timeouts, then unknown. A permanent provider rejection + // (auth, bad request) is NEVER relabeled as exhausted runway. + if (deps.signal?.aborted) throw new LoopError('aborted by caller', 'ABORTED', messages, 'caller-abort'); + if (wallDeadline.aborted) throw new LoopError(`wall budget exhausted after ${Date.now() - started}ms`, 'BUDGET_EXCEEDED', messages, 'budget-wall'); + if (err instanceof ProviderError) throw new LoopError(`${err.code}: ${err.message}`, err.code, messages, 'provider', err.status); + if (err instanceof Error && (err.name === 'TimeoutError' || err.name === 'AbortError')) { + throw new LoopError(`PROVIDER_TIMEOUT: ${err.message}`, 'PROVIDER_TIMEOUT', messages, 'provider'); + } + throw new LoopError(`provider failed: ${err instanceof Error ? err.message : String(err)}`, 'PROVIDER_FAILED', messages, 'provider'); } // butler#8 (defensive twin): a provider that RESOLVES after the caller // aborted (claude-cli exits 0 with an empty result on interrupt) is still // an abort, not an empty answer — classify before touching the result. - if (deps.signal?.aborted) throw new LoopError('aborted by caller', 'ABORTED', messages); + // F3: the guard fires only when the caller signal is actually observed + // aborted; ordinary success passes untouched. + if (deps.signal?.aborted) throw new LoopError('aborted by caller', 'ABORTED', messages, 'caller-abort'); + + // F2: close the hop's text block BEFORE the message lands. text.end prefers + // the STREAMED accumulation when any deltas flowed (byte-exact to what the + // UI saw); content is the fallback for snapshot-only hops (which usually + // have no block at all — then the tracker emits nothing). + if (tracker) tracker.onMessageComplete(hop, streamedText !== '' ? streamedText : (res.message.content ?? '')); if (res.usage) { tokens.in += res.usage.promptTokens; tokens.out += res.usage.completionTokens; } messages.push(res.message); @@ -131,7 +209,7 @@ export async function runLoop(systemPrompt: string, userMessage: string, deps: L // via hardTokenCap=true; the runaway guard remains the universal backstop. if (tokens.in + tokens.out > budget.maxTokens) { if (deps.hardTokenCap) { - throw new LoopError(`token budget exceeded (${tokens.in + tokens.out} > ${budget.maxTokens})`, 'BUDGET_EXCEEDED', messages); + throw new LoopError(`token budget exceeded (${tokens.in + tokens.out} > ${budget.maxTokens})`, 'BUDGET_EXCEEDED', messages, 'budget-token'); } deps.onEvent?.({ type: 'turn.delta', hop, detail: { advisory: `cost ${tokens.in + tokens.out} > ${budget.maxTokens} (advisory)` } }); } @@ -166,21 +244,29 @@ export async function runLoop(systemPrompt: string, userMessage: string, deps: L deps.onMessage?.(msg, { hop, tokensIn: 0, tokensOut: 0 }); continue; } - throw new LoopError(`tool ${call.function.name} sent invalid JSON twice`, 'TOOL_CALL_INVALID', messages); + throw new LoopError(`tool ${call.function.name} sent invalid JSON twice`, 'TOOL_CALL_INVALID', messages, 'tool-invalid'); } deps.onEvent?.({ type: 'turn.tool', hop, detail: { phase: 'call', name: call.function.name } }); + // chat-reliability F1 (tool durability): ordinary registry tools must + // ride the SAME durable seam as streamed harness tools — the loop emits + // per-invocation call/result callbacks so the transport's durable path + // records them (previously runtime tools produced ZERO durable events). + deps.onTool?.({ phase: 'call', name: call.function.name, invocationId: call.id, arg: call.function.arguments }); let obs: string; + let toolFailed = false; try { obs = await tool.execute(args, { signal: deps.signal ?? undefined }); } catch (err) { // handler throws become observations (ReAct) — the loop continues obs = `error: ${err instanceof Error ? err.message : String(err)}`; + toolFailed = true; } + deps.onTool?.({ phase: 'result', name: call.function.name, invocationId: call.id, result: obs, isError: toolFailed }); const msg: ProviderMessage = { role: 'tool', content: obs, tool_call_id: call.id }; messages.push(msg); deps.onMessage?.(msg, { hop, tokensIn: 0, tokensOut: 0 }); deps.onEvent?.({ type: 'turn.tool', hop, detail: { phase: 'result', name: call.function.name, result: `${obs.length} bytes` } }); } } - throw new LoopError(`hop budget exhausted (${budget.maxHops})`, 'BUDGET_EXCEEDED', messages); + throw new LoopError(`hop budget exhausted (${budget.maxHops})`, 'BUDGET_EXCEEDED', messages, 'budget-hop'); } diff --git a/src/memory/contract.ts b/src/memory/contract.ts index 0b93279..d36ec5a 100644 --- a/src/memory/contract.ts +++ b/src/memory/contract.ts @@ -116,6 +116,20 @@ export interface Storage { // read — to seed the loop so the model continues instead of redoing hops. stashPartial(sessionId: string, turnId: string, json: string): Promise; takePartial(sessionId: string): Promise<{ turnId: string; json: string } | null>; + /** chat-reliability F1/F2 (optional capability, authority-driven hosts): + * non-destructive peek + exact compare-and-delete of the continuation + * stash. The authority peeks during preparation and consumes ONLY at the + * committed execution boundary, so a failed preparation never destroys a + * continuation. */ + peekPartial?(sessionId: string): Promise<{ turnId: string; json: string; created_at: number } | null>; + consumePartialExact?(sessionId: string, turnId: string, json: string): Promise; + /** chat-reliability finalizeOwned: SYNCHRONOUS variants for use INSIDE the + * atomic finalization transaction (better-sqlite3 forbids await in a + * transaction; the async methods above cannot join the commit). Optional + * capability — hosts requiring atomic finalization must use a driver that + * provides them and verify it at startup. */ + applyAccountingSync?(sessionId: string, patch: { status?: 'active' | 'ended' | 'error'; tokensInDelta?: number; tokensOutDelta?: number }): void; + stashPartialSync?(sessionId: string, turnId: string, json: string): void; // notes (explicit-write only; supersession, never delete) insertNote(n: Omit): Promise; diff --git a/src/memory/sqlite.ts b/src/memory/sqlite.ts index a2ef542..007be0b 100644 --- a/src/memory/sqlite.ts +++ b/src/memory/sqlite.ts @@ -189,6 +189,31 @@ export class SqliteStorage implements Storage { return { turnId: row.turn_id, json: row.json }; } + /** chat-reliability F1/F2: non-destructive peek — preparation validates without consuming. */ + async peekPartial(sessionId: string): Promise<{ turnId: string; json: string; created_at: number } | null> { + const row = this.db.prepare('SELECT turn_id, json, created_at FROM butler_turn_partials WHERE session_id = ?').get(sessionId) as { turn_id: string; json: string; created_at: number } | undefined; + return row ? { turnId: row.turn_id, json: row.json, created_at: row.created_at } : null; + } + + /** chat-reliability F1/F2: exact CAS delete — consumes the stash only if it + * is still the exact (turnId, json) we snapshotted at the execution boundary. */ + async consumePartialExact(sessionId: string, turnId: string, json: string): Promise { + return this.db.prepare('DELETE FROM butler_turn_partials WHERE session_id = ? AND turn_id = ? AND json = ?').run(sessionId, turnId, json).changes > 0; + } + + /** chat-reliability finalizeOwned: sync, in-transaction session accounting. */ + applyAccountingSync(sessionId: string, patch: { status?: 'active' | 'ended' | 'error'; tokensInDelta?: number; tokensOutDelta?: number }): void { + const cur = this.db.prepare('SELECT status, tokens_in, tokens_out FROM butler_sessions WHERE id = ?').get(sessionId) as { status: string; tokens_in: number; tokens_out: number } | undefined; + if (!cur) throw new Error(`session ${sessionId} not found`); + this.db.prepare('UPDATE butler_sessions SET status = ?, tokens_in = ?, tokens_out = ?, updated_at = ? WHERE id = ?') + .run(patch.status ?? cur.status, cur.tokens_in + (patch.tokensInDelta ?? 0), cur.tokens_out + (patch.tokensOutDelta ?? 0), Date.now(), sessionId); + } + + /** chat-reliability finalizeOwned: sync, in-transaction budget-stash write. */ + stashPartialSync(sessionId: string, turnId: string, json: string): void { + this.db.prepare('INSERT OR REPLACE INTO butler_turn_partials(session_id, turn_id, json, created_at) VALUES (?,?,?,?)').run(sessionId, turnId, json, Date.now()); + } + async sessionMessages(sessionId: string): Promise { return this.db.prepare('SELECT * FROM butler_messages WHERE session_id = ? ORDER BY rowid').all(sessionId) as MessageRow[]; } diff --git a/src/orchestrate/event-sink.ts b/src/orchestrate/event-sink.ts index 107b307..6aed3e6 100644 --- a/src/orchestrate/event-sink.ts +++ b/src/orchestrate/event-sink.ts @@ -79,13 +79,18 @@ export type Effect = export class EventSink { private states = new Map(); - constructor(private readonly db: Database.Database, opts?: { enqueueRunTurn?: boolean }) { + constructor(private readonly db: Database.Database, opts?: { enqueueRunTurn?: boolean; passive?: boolean }) { db.exec(DDL); // v0.5.6: hosts that run turns INLINE (the conversation transport's own // runLoop) never consume runTurn effects — enqueueing them grows the outbox // forever (the 63-row backlog). Default stays true (worker-based hosts). this.enqueueRunTurn = opts?.enqueueRunTurn !== false; - this.recoverOrphans(); + // chat-reliability F2: constructors are passive by choice now. The + // boot-time orphan sweep is a RECOVERY effect only the elected execution + // authority may run; a second sink (MCP context, read-only tool process) + // must never fail a live owner's turn on construction. Legacy callers + // keep the sweep unless they opt into passive mode. + if (opts?.passive !== true) this.recoverOrphans(); } private readonly enqueueRunTurn: boolean; @@ -139,6 +144,49 @@ export class EventSink { * receipt + outbox). Returns the appended events and the enqueued effects. * Idempotent per commandKey: a replayed command returns [] and no effects. */ handle(sessionId: string, cmd: TurnCommand, commandKey: string): { events: TurnEvent[]; effects: Effect[] } { + return this.apply(sessionId, cmd, commandKey); + } + + /** chat-reliability F1: handle a command with EXTRA durable writes (e.g. the + * execution-ledger row) inside the SAME commit — admission is atomic across + * events, projections, receipts and the caller's ledger insert. On any + * throw the SQLite transaction rolls back and the in-memory state cache for + * the session is DISCARDED (never a post-rollback poisoned cache). */ + handleInTransaction( + sessionId: string, + cmd: TurnCommand, + commandKey: string, + alsoRun: ((db: Database.Database) => void) | undefined, + ): { events: TurnEvent[]; effects: Effect[] } { + return this.apply(sessionId, cmd, commandKey, alsoRun); + } + + /** Drop the cached fold for a session (rollback/recovery discipline). */ + invalidate(sessionId: string): void { + this.states.delete(sessionId); + } + + /** chat-reliability F1/F2: run caller writes on the sink's OWN connection in + * one transaction (nested sink calls become savepoints). The execution + * authority composes its ledger rows with event writes here — two + * connections to one file must never nest transactions. */ + withTransaction(fn: (db: Database.Database) => T): T { + const run = this.db.transaction(() => fn(this.db)); + return run(); + } + + /** One consistent snapshot: folded state + events + lastSeq from the same + * read boundary (chat-reliability common-read contract). */ + readSnapshot(sessionId: string): { state: OrchestratorState; events: TurnEvent[]; lastSeq: number } { + return this.db.transaction(() => { + const events = this.events(sessionId); + let st = initialState(sessionId); + for (const e of events) st = fold(st, e); + return { state: st, events, lastSeq: events.length ? events[events.length - 1]!.seq : 0 }; + })(); + } + + private apply(sessionId: string, cmd: TurnCommand, commandKey: string, alsoRun?: (db: Database.Database) => void): { events: TurnEvent[]; effects: Effect[] } { if (this.db.prepare('SELECT 1 FROM butler_command_receipts WHERE command_key = ?').get(commandKey)) { return { events: [], effects: [] }; } @@ -172,9 +220,18 @@ export class EventSink { const anyTurn = appended.some((e) => e.kind === 'turn.started'); if (anyTurn) effects.push({ type: 'runTurn', sessionId, turnId: cmd.turnId }); } + // caller's additional durable writes join THIS commit (F1 atomic admission) + alsoRun?.(this.db); this.db.prepare('INSERT OR IGNORE INTO butler_command_receipts(command_key, session_id, decided_at) VALUES (?,?,?)').run(commandKey, sessionId, now); }); - txn(); + try { + txn(); + } catch (err) { + // rollback discipline: the cached fold may contain pre-rollback appends — + // discard it so the next read refolds from the durable log. + this.states.delete(sessionId); + throw err; + } // outbox rows AFTER the txn commits (the effect worker owns retirement) if (this.enqueueRunTurn) { for (const eff of effects) { diff --git a/src/orchestrate/execution-authority.ts b/src/orchestrate/execution-authority.ts new file mode 100644 index 0000000..018da14 --- /dev/null +++ b/src/orchestrate/execution-authority.ts @@ -0,0 +1,494 @@ +/** chat-reliability F1/F2: execution authority — one compliant executor per + * canonical local agent SQLite DB. + * + * Election: a dedicated tiny control DB (`.execution-control.db`) + * holds a `BEGIN IMMEDIATE` transaction for the authority's entire lifetime. + * A competing executor fails election with SQLITE_BUSY before serving; the OS + * releases the lock when the process dies. This is NOT a distributed lease. + * + * The bare lock is explicitly NOT sufficient (proved by an orphan-child + * probe): worker rows are tracked durably, unsettled previous-epoch work is + * recovered truthfully (one EXECUTION_INTERRUPTED terminal), and unproven + * survivors put the authority in global quarantine — fresh execution is + * blocked until an operator resolves the evidence. No scheduler, no timers, + * no PID-only death guesses. + * + * Admission: tryReserve installs an in-process attempt synchronously (no + * await between check and acquire); admit() commits the turn.request events, + * projections, command receipt AND the execution-ledger row in ONE agent-DB + * transaction (EventSink.handleInTransaction) and requires the intended + * turn.started event to have landed. Losers and queue decisions roll back + * entirely. The continuation stash is PEEKED during preparation and consumed + * only at beginExecution()'s committed boundary via exact CAS delete. */ + +import Database from 'better-sqlite3'; +import { createHash, randomUUID } from 'node:crypto'; +import { lstatSync, realpathSync } from 'node:fs'; +import type { EventSink } from './event-sink.ts'; +import type { Storage } from '../memory/contract.ts'; +import type { TurnCommand, TurnEvent } from './turn-engine.ts'; + +/** Thrown when another executor already holds the control lock for this + * agent DB. The caller must fail startup — never serve alongside it. */ +export class ExecutionAuthorityBusyError extends Error { + constructor(readonly controlPath: string) { + super(`another executor holds the authority lock: ${controlPath}`); + this.name = 'ExecutionAuthorityBusyError'; + } +} + +export interface ExecutionAttempt { + readonly sessionId: string; + readonly turnId: string; + readonly epoch: string; + readonly attemptId: string; + readonly controller: AbortController; +} + +export type ReserveResult = + | { status: 'reserved'; attempt: ExecutionAttempt } + | { status: 'busy'; turnId?: string } + | { status: 'unavailable'; reason: 'quarantined' | 'recovering' | 'draining' | 'closed' }; + +export type AdmissionResult = + | { status: 'admitted'; events: TurnEvent[] } + | { status: 'rejected'; code: RejectCode; turnId?: string; message?: string }; + +type RejectCode = 'SESSION_ACTIVE' | 'REQUEST_ALREADY_ADMITTED' | 'REQUEST_ID_REUSE' | 'QUEUE_DECISION' | 'STALE_ATTEMPT' | 'STOP_BEFORE_START' | 'ROLLBACK'; + +/** Common execution read (session GET + timeline GET must agree). */ +export type ExecutionRead = + | { state: 'idle'; admission: 'ready' | 'blocked'; blockedReason?: string } + | { state: 'preparing' | 'running' | 'stopping'; turnId: string; startedAt: number; admission: 'busy' } + | { state: 'uncertain'; turnId?: string; startedAt: number | null; reason: 'legacy_owner' | 'owner_not_local' | 'recovery_in_progress' | 'authority_unavailable' | 'worker_quarantine'; admission: 'blocked' }; + +export type StopResult = + | { status: 'aborting'; turnId: string } + | { status: 'no_active' } + | { status: 'stale' } + | { status: 'nonlocal' }; + +export type RecoveryResult = { recovered: number; quarantined: boolean; uncertainWorkers: number }; + +export interface ResolveWorkerInput { + expectedEpoch: string; + operatorIdentity: string; + evidenceRef: string; +} + +const EXECUTION_DDL = ` +CREATE TABLE IF NOT EXISTS butler_execution_authority( + singleton INTEGER PRIMARY KEY CHECK(singleton=1), epoch TEXT NOT NULL, + db_identity TEXT NOT NULL, pid INTEGER NOT NULL, acquired_at INTEGER NOT NULL, + phase TEXT NOT NULL CHECK(phase IN ('recovering','ready','draining','quarantined','released'))); +CREATE TABLE IF NOT EXISTS butler_executions( + turn_id TEXT PRIMARY KEY, session_id TEXT NOT NULL, epoch TEXT NOT NULL, + principal_id TEXT NOT NULL, request_id TEXT, started_at INTEGER NOT NULL, + phase TEXT NOT NULL CHECK(phase IN ('preparing','executing','settled')), + stop_requested INTEGER NOT NULL DEFAULT 0, terminal_code TEXT, + resume_turn_id TEXT, resume_json TEXT, settled_at INTEGER, request_hash TEXT, + UNIQUE(principal_id,session_id,request_id)); +CREATE UNIQUE INDEX IF NOT EXISTS idx_execution_active ON butler_executions(session_id) WHERE phase != 'settled'; +CREATE TABLE IF NOT EXISTS butler_execution_workers( + worker_id TEXT PRIMARY KEY, epoch TEXT NOT NULL, turn_id TEXT, kind TEXT NOT NULL, + phase TEXT NOT NULL CHECK(phase IN ('starting','running','settled','uncertain')), + pid INTEGER, identity TEXT, created_at INTEGER NOT NULL, settled_at INTEGER); +`; + +const CONTROL_DDL = ` +CREATE TABLE IF NOT EXISTS authority_lock(singleton INTEGER PRIMARY KEY CHECK(singleton=1),format INTEGER NOT NULL); +INSERT OR IGNORE INTO authority_lock VALUES(1,1); +`; + +/** Internal control-flow signal — mapped to an AdmissionResult, never escapes. */ +class Reject extends Error { + constructor(readonly code: RejectCode, readonly turnId?: string, message?: string) { + super(message ?? code); + } +} + +const requestFingerprint = (content: string): string => + createHash('sha256').update(JSON.stringify({ v: 1, content }), 'utf8').digest('hex'); + +export interface ExecutionAuthorityDeps { + /** canonical path of the AGENT event DB (the DB the sink writes to) */ + dbPath: string; + sink: EventSink; + storage: Storage; + /** override for the derived control path (tests) */ + controlPath?: string; +} + +export async function createSqliteExecutionAuthority(deps: ExecutionAuthorityDeps): Promise { + const canonical = realpathSync(deps.dbPath); + const controlPath = deps.controlPath ?? `${canonical}.execution-control.db`; + // identity safety: never follow a symlinked/replaced control target + // (a not-yet-existing control file is fine — SQLite creates it) + let controlStat: ReturnType | undefined; + try { controlStat = lstatSync(controlPath); } catch { controlStat = undefined; } + if (controlStat?.isSymbolicLink()) throw new Error(`refusing symlinked control DB: ${controlPath}`); + + const control = new Database(controlPath); + control.pragma('busy_timeout = 0'); + try { + // both statements need the write lock — either failing with SQLITE_BUSY + // means another executor owns this control DB + control.exec(CONTROL_DDL); + control.exec('BEGIN IMMEDIATE'); // held for the authority's lifetime + } catch { + control.close(); + throw new ExecutionAuthorityBusyError(controlPath); + } + + const agent = new Database(canonical); + agent.pragma('busy_timeout = 5000'); + agent.exec(EXECUTION_DDL); + + const authority = new ExecutionAuthority(canonical, control, agent, deps.sink, deps.storage, randomUUID()); + authority.recoverPreviousEpochs(); // synchronous SQLite — complete before readiness + return authority; +} + +export class ExecutionAuthority { + private phase: 'recovering' | 'ready' | 'draining' | 'quarantined' | 'released' = 'recovering'; + private readonly active = new Map(); + private readonly tracked = new Map>(); + private released = false; + + constructor( + readonly dbPath: string, + private readonly control: Database.Database, + private readonly agent: Database.Database, + private readonly sink: EventSink, + private readonly storage: Storage, + readonly epoch: string, + ) { + this.agent + .prepare('INSERT OR REPLACE INTO butler_execution_authority(singleton,epoch,db_identity,pid,acquired_at,phase) VALUES (1,?,?,?,?,?)') + .run(epoch, this.dbPath, process.pid, Date.now(), 'recovering'); + } + + // ── admission ── + + tryReserve(sessionId: string): ReserveResult { + if (this.released) return { status: 'unavailable', reason: 'closed' }; + if (this.phase === 'quarantined') return { status: 'unavailable', reason: 'quarantined' }; + if (this.phase === 'draining') return { status: 'unavailable', reason: 'draining' }; + if (this.phase !== 'ready') return { status: 'unavailable', reason: 'recovering' }; + const local = this.active.get(sessionId); + if (local) return { status: 'busy', turnId: local.turnId }; + const durable = this.unsettledRow(sessionId); + if (durable) return { status: 'busy', turnId: durable.turn_id }; + const attempt: ExecutionAttempt = { + sessionId, + turnId: `t_${randomUUID().slice(0, 12)}`, + epoch: this.epoch, + attemptId: randomUUID(), + controller: new AbortController(), + }; + this.active.set(sessionId, attempt); // atomic in-process acquire + return { status: 'reserved', attempt }; + } + + /** Durable atomic admission: turn.request events + projections + receipt + + * execution row in one commit; the intended turn.started must have landed. + * Releases the reservation on every non-admitted path. */ + admit(attempt: ExecutionAttempt, input: { content: string; principalId: string; requestId?: string }): AdmissionResult { + if (!this.owns(attempt)) return { status: 'rejected', code: 'STALE_ATTEMPT' }; + const fp = requestFingerprint(input.content); + const release = () => this.releaseAttempt(attempt); + try { + // ONE connection (the sink's), ONE commit: epoch check, ledger check, + // request idempotence, turn.request events + projections + receipt and + // the execution row land together or not at all. + this.sink.withTransaction((db) => { + const auth = db.prepare('SELECT epoch FROM butler_execution_authority WHERE singleton = 1').get() as { epoch: string } | undefined; + if (!auth || auth.epoch !== this.epoch) throw new Reject('STALE_ATTEMPT'); + const unsettled = db.prepare("SELECT turn_id FROM butler_executions WHERE session_id = ? AND phase != 'settled'").get(attempt.sessionId) as { turn_id: string } | undefined; + if (unsettled && unsettled.turn_id !== attempt.turnId) throw new Reject('SESSION_ACTIVE', unsettled.turn_id); + if (input.requestId) { + const prior = db + .prepare('SELECT turn_id, request_hash FROM butler_executions WHERE principal_id = ? AND session_id = ? AND request_id = ?') + .get(input.principalId, attempt.sessionId, input.requestId) as { turn_id: string; request_hash: string } | undefined; + if (prior) { + if (prior.request_hash === fp) throw new Reject('REQUEST_ALREADY_ADMITTED', prior.turn_id); + throw new Reject('REQUEST_ID_REUSE', prior.turn_id); + } + } + const cmd: TurnCommand = { kind: 'turn.request', sessionId: attempt.sessionId, turnId: attempt.turnId, prompt: input.content, refIds: [], at: Date.now() } as TurnCommand; + const { events } = this.sink.handleInTransaction(attempt.sessionId, cmd, `chat:${attempt.sessionId}:${attempt.turnId}`, (txnDb) => { + txnDb.prepare('INSERT INTO butler_executions(turn_id,session_id,epoch,principal_id,request_id,started_at,phase,stop_requested,request_hash) VALUES (?,?,?,?,?,?,?,0,?)') + .run(attempt.turnId, attempt.sessionId, this.epoch, input.principalId, input.requestId ?? null, Date.now(), 'preparing', fp); + }); + // the intended turn must have ACTUALLY started in this commit — a + // queue decision or a foreign active turn rolls everything back. + if (!events.some((e) => e.kind === 'turn.started' && e.turn_id === attempt.turnId)) { + throw new Reject('QUEUE_DECISION'); + } + }); + return { status: 'admitted', events: [] }; + } catch (err) { + release(); + this.sink.invalidate(attempt.sessionId); + if (err instanceof Reject) return { status: 'rejected', code: err.code, ...(err.turnId !== undefined ? { turnId: err.turnId } : {}), message: err.message }; + return { status: 'rejected', code: 'ROLLBACK', message: err instanceof Error ? err.message : String(err) }; + } + } + + /** Execution boundary: snapshot + CAS-consume the continuation stash, mark + * the execution `executing`. Returns the seeded partial messages (validated) + * or null. A Stop requested before start aborts without consuming anything. + * NOTE: the stash lives in the MAIN storage DB while the ledger lives in + * the agent DB — the snapshot is committed to the ledger first, then the + * exact stash bytes are CAS-deleted; a crash between the two leaves an + * inspectable snapshot and a recoverable (interrupted) execution. */ + async beginExecution(attempt: ExecutionAttempt): Promise<{ seeded: unknown[] | null; stashTurnId: string | null } | { status: 'rejected'; code: 'STALE_ATTEMPT' | 'STOP_BEFORE_START' | 'STASH_CHANGED' }> { + if (!this.owns(attempt)) return { status: 'rejected', code: 'STALE_ATTEMPT' }; + const row = this.agent.prepare('SELECT phase, stop_requested FROM butler_executions WHERE turn_id = ?').get(attempt.turnId) as { phase: string; stop_requested: number } | undefined; + if (!row || row.phase !== 'preparing') return { status: 'rejected', code: 'STALE_ATTEMPT' }; + if (row.stop_requested) { + this.releaseAttempt(attempt); + return { status: 'rejected', code: 'STOP_BEFORE_START' }; + } + const stash = (await this.storage.peekPartial?.(attempt.sessionId)) ?? null; + let seeded: unknown[] | null = null; + if (stash) { + try { seeded = JSON.parse(stash.json) as unknown[]; } catch { seeded = null; } // corrupt stash never kills the turn; bytes stay + } + const mark = this.agent.transaction(() => { + const auth = this.agent.prepare('SELECT epoch FROM butler_execution_authority WHERE singleton = 1').get() as { epoch: string } | undefined; + if (!auth || auth.epoch !== this.epoch) throw new Reject('STALE_ATTEMPT'); + if (stash) { + this.agent.prepare('UPDATE butler_executions SET phase = ?, resume_turn_id = ?, resume_json = ? WHERE turn_id = ? AND phase = ?') + .run('executing', stash.turnId, stash.json, attempt.turnId, 'preparing'); + } else { + this.agent.prepare('UPDATE butler_executions SET phase = ? WHERE turn_id = ? AND phase = ?').run('executing', attempt.turnId, 'preparing'); + } + }); + try { + mark(); + } catch { + this.releaseAttempt(attempt); + return { status: 'rejected', code: 'STALE_ATTEMPT' }; + } + if (stash) { + const consumed = (await this.storage.consumePartialExact?.(attempt.sessionId, stash.turnId, stash.json)) ?? false; + if (!consumed) { + // the stash changed under us — abort before any provider work; the + // newer stash stays intact for its own owner. + await this.settle(attempt, 'STASH_CHANGED'); + return { status: 'rejected', code: 'STASH_CHANGED' }; + } + } + return { seeded, stashTurnId: stash?.turnId ?? null }; + } + + /** Durable durable-write seam for the owning transport: every command is + * fenced by epoch + attempt identity. */ + handleOwned(attempt: ExecutionAttempt, sessionId: string, cmd: TurnCommand, commandKey: string): { events: TurnEvent[] } { + if (!this.owns(attempt)) throw new Error('stale attempt'); + const result = this.sink.handle(sessionId, cmd, commandKey); + return { events: result.events }; + } + + /** Terminal settlement of an execution (idempotent; owner-only). */ + async settle(attempt: ExecutionAttempt, terminalCode: string | null): Promise { + if (!this.owns(attempt)) return; + this.agent + .prepare("UPDATE butler_executions SET phase = 'settled', terminal_code = ?, settled_at = COALESCE(settled_at, ?) WHERE turn_id = ? AND epoch = ?") + .run(terminalCode, Date.now(), attempt.turnId, this.epoch); + this.releaseAttempt(attempt); + } + + /** chat-reliability §3.5: OWNED ATOMIC FINALIZATION — terminal event, + * projections, session accounting, budget stash and the execution-ledger + * settle commit in ONE sink-connection transaction. Exact duplicate + * finalize is a no-op (engine terminality guard returns no events → + * already-terminal; the ledger still reconciles). A storage failure rolls + * everything back and leaves the attempt reserved for an idempotent retry + * — success is never reported for an uncommitted terminal. */ + async finalizeOwned( + attempt: ExecutionAttempt, + input: { + kind: 'turn.complete' | 'turn.fail'; + answer?: string; + error?: string; + commandKey: string; + settleCode: string; + accounting?: { status?: 'active' | 'ended' | 'error'; tokensInDelta?: number; tokensOutDelta?: number }; + stash?: { json: string }; + }, + ): Promise<{ status: 'finalized' | 'already-terminal' | 'stale'; accounted: boolean }> { + if (!this.owns(attempt)) return { status: 'stale', accounted: false }; + let accounted = false; + let wroteTerminal = false; + try { + this.sink.withTransaction((db) => { + const auth = db.prepare('SELECT epoch FROM butler_execution_authority WHERE singleton = 1').get() as { epoch: string } | undefined; + if (!auth || auth.epoch !== this.epoch) throw new Error('stale authority epoch'); + const cmd: TurnCommand = (input.kind === 'turn.complete' + ? { kind: 'turn.complete', sessionId: attempt.sessionId, turnId: attempt.turnId, answer: input.answer ?? '' } + : { kind: 'turn.fail', sessionId: attempt.sessionId, turnId: attempt.turnId, error: input.error ?? 'turn failed' }) as TurnCommand; + const { events } = this.sink.handleInTransaction(attempt.sessionId, cmd, input.commandKey, undefined); + wroteTerminal = events.some((e) => e.kind === 'turn.completed' || e.kind === 'turn.failed'); + if (input.stash) this.storage.stashPartialSync?.(attempt.sessionId, attempt.turnId, input.stash.json); + if (input.accounting) { + this.storage.applyAccountingSync?.(attempt.sessionId, input.accounting); + accounted = true; + } + db.prepare("UPDATE butler_executions SET phase = 'settled', terminal_code = ?, settled_at = ? WHERE turn_id = ? AND epoch = ? AND phase != 'settled'") + .run(input.settleCode, Date.now(), attempt.turnId, this.epoch); + }); + } catch (err) { + this.sink.invalidate(attempt.sessionId); + throw err; // attempt stays reserved: retry this idempotent finalization only + } + this.sink.invalidate(attempt.sessionId); + this.releaseAttempt(attempt); + return { status: wroteTerminal ? 'finalized' : 'already-terminal', accounted }; + } + + /** Release the in-process reservation ONLY if the identity still matches. */ + releaseAttempt(attempt: ExecutionAttempt): void { + if (this.active.get(attempt.sessionId) === attempt) this.active.delete(attempt.sessionId); + } + + // ── stop / read / workers ── + + requestStop(sessionId: string, expectedTurnId?: string): StopResult { + const local = this.active.get(sessionId); + const durable = this.unsettledRow(sessionId); + if (!local && !durable) return { status: 'no_active' }; + const turnId = local?.turnId ?? durable!.turn_id; + if (expectedTurnId !== undefined && turnId !== expectedTurnId) return { status: 'stale' }; + if (durable) this.agent.prepare('UPDATE butler_executions SET stop_requested = 1 WHERE turn_id = ?').run(durable.turn_id); + if (local) local.controller.abort(); + return { status: 'aborting', turnId }; + } + + /** Single-snapshot execution read; performs NO recovery and NO dispatch. */ + readSession(sessionId: string): ExecutionRead { + if (this.phase === 'quarantined') { + const row = this.unsettledRow(sessionId); + return { state: 'uncertain', ...(row?.turn_id !== undefined ? { turnId: row.turn_id } : {}), startedAt: row?.started_at ?? null, reason: 'worker_quarantine', admission: 'blocked' }; + } + const local = this.active.get(sessionId); + const row = this.unsettledRow(sessionId); + if (local && row && row.turn_id === local.turnId) { + const stopping = row.stop_requested === 1; + return { state: stopping ? 'stopping' : row.phase === 'preparing' ? 'preparing' : 'running', turnId: row.turn_id, startedAt: row.started_at, admission: 'busy' }; + } + if (row) { + // durable non-settled work this process does not own locally + return { state: 'uncertain', turnId: row.turn_id, startedAt: row.started_at, reason: 'owner_not_local', admission: 'blocked' }; + } + // legacy: an active fold with no execution row is uncertain, never idle + const fold = this.sink.readSnapshot(sessionId).state; + if (fold.activeTurnId !== null) { + return { state: 'uncertain', turnId: fold.activeTurnId, startedAt: null, reason: 'legacy_owner', admission: 'blocked' }; + } + return { state: 'idle', admission: this.phase === 'ready' ? 'ready' : 'blocked', ...(this.phase === 'ready' ? {} : { blockedReason: `authority ${this.phase}` }) }; + } + + /** Record `starting` BEFORE any spawn/tail; settled only on actual promise + * settlement. The returned promise mirrors `start()`'s outcome. */ + trackWorker(scope: { turnId?: string; identity?: Record }, kind: string, start: () => Promise): Promise { + const workerId = `w_${randomUUID().slice(0, 12)}`; + this.agent + .prepare('INSERT INTO butler_execution_workers(worker_id,epoch,turn_id,kind,phase,pid,identity,created_at) VALUES (?,?,?,?,?,?,?,?)') + .run(workerId, this.epoch, scope.turnId ?? null, kind, 'starting', process.pid, JSON.stringify(scope.identity ?? {}), Date.now()); + const tracked = Promise.resolve() + .then(start) + .finally(() => { + this.agent.prepare('UPDATE butler_execution_workers SET phase = ?, settled_at = ? WHERE worker_id = ?').run('settled', Date.now(), workerId); + this.tracked.delete(workerId); + }); + this.tracked.set(workerId, tracked); + return tracked; + } + + /** Operator-only: resolve one uncertain worker with recorded evidence. + * All uncertain workers must be resolved before readiness returns. */ + resolveUncertainWorker(workerId: string, input: ResolveWorkerInput): boolean { + const row = this.agent.prepare('SELECT epoch, phase, identity FROM butler_execution_workers WHERE worker_id = ?').get(workerId) as { epoch: string; phase: string; identity: string | null } | undefined; + if (!row || row.phase !== 'uncertain') return false; + if (row.epoch !== input.expectedEpoch) return false; + const evidence = { ...(row.identity ? JSON.parse(row.identity) as Record : {}), resolvedBy: input.operatorIdentity, evidenceRef: input.evidenceRef }; + const updated = this.agent + .prepare("UPDATE butler_execution_workers SET phase = 'settled', settled_at = ?, identity = ? WHERE worker_id = ? AND phase = 'uncertain'") + .run(Date.now(), JSON.stringify(evidence), workerId); + if (updated.changes === 0) return false; + const remaining = this.agent.prepare("SELECT COUNT(*) AS n FROM butler_execution_workers WHERE phase = 'uncertain'").get() as { n: number }; + if (remaining.n === 0 && this.phase === 'quarantined') this.setPhase('ready'); + return true; + } + + // ── recovery / lifecycle ── + + /** Only the lock-holding authority runs this, before accepting messages. + * Known previous-epoch executions get ONE truthful interruption terminal + * (or ledger-only reconcile when already terminal). Nothing is replayed. */ + recoverPreviousEpochs(): RecoveryResult { + const stale = this.agent.prepare("SELECT turn_id, session_id FROM butler_executions WHERE epoch != ? AND phase != 'settled'").all(this.epoch) as Array<{ turn_id: string; session_id: string }>; + let recovered = 0; + for (const row of stale) { + try { + this.sink.withTransaction((db) => { + const fold = this.sink.readSnapshot(row.session_id).state; + if (fold.activeTurnId === row.turn_id) { + this.sink.handleInTransaction( + row.session_id, + { kind: 'turn.fail', sessionId: row.session_id, turnId: row.turn_id, error: 'execution interrupted: the owning executor went away; partial content is preserved in prior events — resend to continue' } as TurnCommand, + `recover:${row.turn_id}`, + undefined, + ); + } + db.prepare("UPDATE butler_executions SET phase = 'settled', terminal_code = 'EXECUTION_INTERRUPTED', settled_at = ? WHERE turn_id = ? AND phase != 'settled'").run(Date.now(), row.turn_id); + }); + recovered += 1; + } finally { + this.sink.invalidate(row.session_id); + } + } + // unfinished workers from any earlier epoch are unproven survivors + this.agent.prepare("UPDATE butler_execution_workers SET phase = 'uncertain' WHERE epoch != ? AND phase IN ('starting','running')").run(this.epoch); + const uncertain = (this.agent.prepare("SELECT COUNT(*) AS n FROM butler_execution_workers WHERE phase = 'uncertain'").get() as { n: number }).n; + this.setPhase(uncertain > 0 ? 'quarantined' : 'ready'); + return { recovered, quarantined: uncertain > 0, uncertainWorkers: uncertain }; + } + + /** Close admission, await tracked workers, then (and only then) release the + * control lock. A drain problem keeps the lock held — never close anyway. */ + async drain(): Promise { + if (this.released) return; + this.setPhase('draining'); + await Promise.allSettled([...this.tracked.values()]); + this.setPhase('released'); + this.released = true; + try { + this.control.exec('ROLLBACK'); // releases BEGIN IMMEDIATE + } finally { + this.control.close(); + this.agent.close(); + } + } + + currentPhase(): 'recovering' | 'ready' | 'draining' | 'quarantined' | 'released' { + return this.phase; + } + + // ── internals ── + + private owns(attempt: ExecutionAttempt): boolean { + return attempt.epoch === this.epoch && this.active.get(attempt.sessionId) === attempt; + } + + private unsettledRow(sessionId: string): { turn_id: string; started_at: number; phase: string; stop_requested: number } | undefined { + return this.agent.prepare("SELECT turn_id, started_at, phase, stop_requested FROM butler_executions WHERE session_id = ? AND phase != 'settled'").get(sessionId) as { turn_id: string; started_at: number; phase: string; stop_requested: number } | undefined; + } + + private setPhase(phase: 'recovering' | 'ready' | 'draining' | 'quarantined' | 'released'): void { + this.phase = phase; + this.agent.prepare('UPDATE butler_execution_authority SET phase = ? WHERE singleton = 1').run(phase); + } +} diff --git a/src/orchestrate/turn-engine.ts b/src/orchestrate/turn-engine.ts index a6ca965..b46260c 100644 --- a/src/orchestrate/turn-engine.ts +++ b/src/orchestrate/turn-engine.ts @@ -54,7 +54,7 @@ export type TurnEvent = | (TurnEventBase & { kind: 'user_message.recorded'; detail: { role: 'user' | 'assistant'; content: string } }) | (TurnEventBase & { kind: 'message.queued'; detail: { content: string; disposition: 'steering' | 'followup' } }) | (TurnEventBase & { kind: 'queue.updated'; detail: { steering: string[]; followUp: string[] } }) - | (TurnEventBase & { kind: 'chunk'; detail: { delta: string } }) + | (TurnEventBase & { kind: 'chunk'; detail: { delta: string; block?: TurnChunkBlock } }) | (TurnEventBase & { kind: 'tool_call'; detail: { toolCallId: string; name: string; arg?: string } }) | (TurnEventBase & { kind: 'tool_call_update'; detail: { toolCallId: string; result?: string; isError?: boolean } }) | (TurnEventBase & { kind: 'usage'; detail: UsageMeters }) @@ -65,13 +65,25 @@ export type TurnEvent = // ── Commands (the ONLY way state changes) ── +/** chat-reliability F2 (text block lifecycle): durable block metadata riding + * chunk commands. Additive and optional — legacy readers concatenate `delta` + * exactly as before (start/end chunks carry an EMPTY delta). */ +export interface TurnChunkBlock { + id: string; + phase: 'start' | 'delta' | 'end'; + /** provider-native block identity when one exists */ + native?: string; + /** FULL block text — present on phase:'end' */ + text?: string; +} + export type TurnCommand = | { kind: 'session.import'; sessionId: string; note: string } | { kind: 'turn.request'; sessionId: string; turnId: string; prompt: string; refIds: string[]; at: number } | { kind: 'message.append'; sessionId: string; role: 'user' | 'assistant'; content: string; narrative?: boolean } | { kind: 'message.queue'; sessionId: string; content: string; disposition: 'steering' | 'followup' } | { kind: 'turn.start'; sessionId: string; turnId: string } - | { kind: 'turn.chunk'; sessionId: string; turnId: string; delta: string } + | { kind: 'turn.chunk'; sessionId: string; turnId: string; delta: string; block?: TurnChunkBlock } | { kind: 'turn.tool'; sessionId: string; turnId: string; phase: 'call' | 'update'; toolCallId: string; name?: string; arg?: string; result?: string; isError?: boolean } | { kind: 'turn.usage'; sessionId: string; turnId: string; meters: UsageMeters } | { kind: 'turn.compaction'; sessionId: string; turnId: string; summary: string; preservedEventIds: string[] } @@ -148,7 +160,9 @@ export function decide(state: OrchestratorState, cmd: TurnCommand): TurnEvent[] case 'turn.chunk': { const t = state.turns.find((x) => x.id === cmd.turnId); if (!t || t.status !== 'running') return []; - return [{ ...S(cmd.turnId), kind: 'chunk', detail: { delta: cmd.delta } }]; + // F2: `block` rides along additively (absent on legacy chunks); the fold + // reads only the fields it knows and ignores any other unknown field. + return [{ ...S(cmd.turnId), kind: 'chunk', detail: { delta: cmd.delta, ...(cmd.block !== undefined ? { block: cmd.block } : {}) } }]; } case 'turn.tool': { diff --git a/src/provider/activity.ts b/src/provider/activity.ts new file mode 100644 index 0000000..e8d0567 --- /dev/null +++ b/src/provider/activity.ts @@ -0,0 +1,47 @@ +/** chat-reliability F2 — text BLOCK lifecycle (the glued-narration fix). + * Streams of per-token deltas concatenate into ONE block per assistant + * message (hop); without explicit start/end boundaries a replayed log glues + * consecutive hops' narration together ("…checked.develop…"). This module is + * the minimal typed contract plus a stateful tracker the loop feeds; the + * transport derives durable block metadata from it (chunk-rail enrichment). + * + * Byte-exactness contract: the tracker forwards ONLY the delta bytes it is + * given — it never joins, pads, trims, or injects whitespace. Chunk splits + * inside a block (de + velop) concatenate byte-exactly. */ + +/** One activity per block boundary or delta. `native` carries the provider's + * own block identity (e.g. an Anthropic content-block id) when one exists; + * blockId is the loop-allocated durable identity. `text` is the delta text + * for start/delta and the FULL block text for end. */ +export type TextBlockActivity = { type: 'text.start' | 'text.delta' | 'text.end'; hop: number; blockId: string; native?: string; text: string }; + +/** Stateful per-run block tracker. One assistant message per hop = ONE block: + * - first delta of a hop → text.start (with the allocated blockId), then + * text.delta for that same chunk + * - every delta (including the first) → text.delta + * - onMessageComplete(hop, fullText) → text.end with the FULL block text + * - a hop with no deltas emits NOTHING (no phantom blocks) */ +export function createBlockTracker( + emit: (a: TextBlockActivity) => void, + allocBlockId: (hop: number) => string, +): { onDelta(hop: number, delta: string): string | undefined; onMessageComplete(hop: number, fullText: string): void } { + let open: { hop: number; blockId: string } | null = null; + return { + onDelta(hop: number, delta: string): string | undefined { + if (!open || open.hop !== hop) { + // a fresh hop opens a fresh block; an unclosed prior block is simply + // left unterminated — its deltas are already durable downstream. + open = { hop, blockId: allocBlockId(hop) }; + emit({ type: 'text.start', hop, blockId: open.blockId, text: delta }); + return open.blockId; + } + emit({ type: 'text.delta', hop, blockId: open.blockId, text: delta }); + return open.blockId; + }, + onMessageComplete(hop: number, fullText: string): void { + if (!open || open.hop !== hop) return; // no deltas streamed → no block → nothing to end + emit({ type: 'text.end', hop, blockId: open.blockId, text: fullText }); + open = null; + }, + }; +} diff --git a/src/provider/claude-cli.ts b/src/provider/claude-cli.ts index c3490e0..0a25c7a 100644 --- a/src/provider/claude-cli.ts +++ b/src/provider/claude-cli.ts @@ -25,7 +25,7 @@ export interface ClaudeCliEvent { result?: string; usage?: { input_tokens?: number; output_tokens?: number; cache_read_input_tokens?: number; cache_creation_input_tokens?: number } | null; tool_use_id?: string; - message?: { content?: Array<{ type: string; text?: string; name?: string; input?: unknown; is_error?: boolean; content?: unknown }> }; + message?: { content?: Array<{ type: string; text?: string; name?: string; id?: string; tool_use_id?: string; input?: unknown; is_error?: boolean; content?: unknown }> }; error?: string; /** v0.5.3 --include-partial-messages: Anthropic SSE events wrapped by the CLI * as {type:'stream_event', event:{...}}. content_block_delta carries @@ -100,7 +100,7 @@ export class ClaudeCliProvider implements Provider { async complete( messages: ProviderMessage[], _tools: ProviderTool[], - opts: { remainingMs: number; signal?: AbortSignal | undefined; onDelta?: (delta: string) => void; onTool?: (e: { phase: 'call' | 'result'; name: string; arg?: string; result?: string; isError?: boolean }) => void; sessionKey?: string }, + opts: { remainingMs: number; signal?: AbortSignal | undefined; onDelta?: (delta: string) => void; onTool?: (e: { phase: 'call' | 'result'; name: string; invocationId?: string; arg?: string; result?: string; isError?: boolean }) => void; sessionKey?: string }, ): Promise { // v0.5.5: the CLI cold-start race leaves MCP servers 'pending' at init — // the model would see zero tools (sess_88de6757). Serialization prevents @@ -127,7 +127,7 @@ export class ClaudeCliProvider implements Provider { private async completeOnce( messages: ProviderMessage[], _tools: ProviderTool[], - opts: { remainingMs: number; signal?: AbortSignal | undefined; onDelta?: (delta: string) => void; onTool?: (e: { phase: 'call' | 'result'; name: string; arg?: string; result?: string; isError?: boolean }) => void; sessionKey?: string }, + opts: { remainingMs: number; signal?: AbortSignal | undefined; onDelta?: (delta: string) => void; onTool?: (e: { phase: 'call' | 'result'; name: string; invocationId?: string; arg?: string; result?: string; isError?: boolean }) => void; sessionKey?: string }, ): Promise { const bin = (this.cfg as { binPath?: string }).binPath ?? 'claude'; // R2-F4: isolation — the caller's key addresses its OWN claude session @@ -192,8 +192,14 @@ export class ClaudeCliProvider implements Provider { let stderr = ''; const events: ClaudeCliEvent[] = []; let settled = false; - // correlate tool_results to calls by ORDER (result blocks carry no name) - const pendingCalls: string[] = []; + // correlate tool_results to calls by ORDER (result blocks carry no name); + // chat-reliability F2(b): entries keep the provider-NATIVE tool_use id so + // results correlate to the same invocation the durable rail keyed. + const pendingCalls: Array<{ name: string; id?: string }> = []; + // F2(a): text_delta bytes streamed since the last assistant snapshot — the + // completed snapshot must REPLACE/complete the streamed block, never + // re-emit bytes the stream already delivered (duplicated narration). + let streamedSinceSnapshot = ''; const onLine = (line: string): void => { const t = line.trim(); if (!t) return; @@ -228,21 +234,35 @@ export class ClaudeCliProvider implements Provider { if (ev.type === 'stream_event') { const se = ev.event; if (se?.type === 'content_block_delta' && se.delta?.type === 'text_delta' && se.delta.text) { + streamedSinceSnapshot += se.delta.text; opts.onDelta?.(se.delta.text); // the answer tokens, live } return; // thinking_delta & infrastructure events; the full assistant message still arrives } if (ev.type === 'assistant' && ev.message?.content) { + // F2(a): the completed assistant message is the SNAPSHOT of the block + // the stream (if any) already delivered. Never re-emit streamed bytes: + // - nothing streamed → snapshot-only mode: the blocks ARE the stream + // - snapshot extends the stream (cut mid-block) → emit only the tail + // - snapshot already streamed → emit nothing; its completion lands + // as the loop's text.end, not as duplicated deltas + const textBlocks = (ev.message.content ?? []) + .filter((b) => b.type === 'text' && b.text && !INTERNAL_PROGRESS.has(b.text.trim())) + .map((b) => b.text as string); + const snapshot = textBlocks.join('\n\n'); + if (streamedSinceSnapshot === '') { + for (const t of textBlocks) opts.onDelta?.(t); + } else if (snapshot.startsWith(streamedSinceSnapshot) && snapshot.length > streamedSinceSnapshot.length) { + opts.onDelta?.(snapshot.slice(streamedSinceSnapshot.length)); + } + streamedSinceSnapshot = ''; for (const blk of ev.message.content) { - if (blk.type === 'text' && blk.text) { - if (!INTERNAL_PROGRESS.has(blk.text.trim())) opts.onDelta?.(blk.text); - } if (blk.type === 'tool_use') { const sum = inputSummary(blk.input); const nm = blk.name ?? 'tool'; - pendingCalls.push(nm); - if (sum) opts.onTool?.({ phase: 'call', name: nm, arg: sum }); - else opts.onTool?.({ phase: 'call', name: nm }); + pendingCalls.push({ name: nm, ...(blk.id !== undefined ? { id: blk.id } : {}) }); + if (sum) opts.onTool?.({ phase: 'call', name: nm, arg: sum, ...(blk.id !== undefined ? { invocationId: blk.id } : {}) }); + else opts.onTool?.({ phase: 'call', name: nm, ...(blk.id !== undefined ? { invocationId: blk.id } : {}) }); } } } @@ -251,12 +271,18 @@ export class ClaudeCliProvider implements Provider { if (blk.type !== 'tool_result') continue; const text = typeof blk.content === 'string' ? blk.content : Array.isArray(blk.content) ? (blk.content as Array<{ type: string; text?: string }>).filter((b) => b.type === 'text').map((b) => b.text ?? '').join(' ') : ''; - const head = text.trim().split('\n')[0] ?? ''; - if (head) { - const nm = pendingCalls.shift() ?? 'tool'; - if (blk.is_error) opts.onTool?.({ phase: 'result', name: nm, result: head.slice(0, 160), isError: true }); - else opts.onTool?.({ phase: 'result', name: nm, result: head.slice(0, 160) }); - } + // F2(c): FULL results — no first-line reduction, no 160-char clip, + // and EMPTY results are recorded honestly (result:'', isError:false) + // instead of being dropped from the trace. + const pending = pendingCalls.shift() ?? { name: 'tool' }; + const invocationId = blk.tool_use_id ?? pending.id; + opts.onTool?.({ + phase: 'result', + name: pending.name, + ...(invocationId !== undefined ? { invocationId } : {}), + result: text, + isError: Boolean(blk.is_error), + }); } } }; diff --git a/src/provider/types.ts b/src/provider/types.ts index 755ee62..5c600a0 100644 --- a/src/provider/types.ts +++ b/src/provider/types.ts @@ -63,7 +63,7 @@ export interface Provider { complete( messages: ProviderMessage[], tools: ProviderTool[], - opts: { remainingMs: number; signal?: AbortSignal | undefined; onDelta?: ((delta: string) => void) | undefined; onTool?: ((e: { phase: 'call' | 'result'; name: string; arg?: string; result?: string; isError?: boolean }) => void) | undefined; sessionKey?: string | undefined }, + opts: { remainingMs: number; signal?: AbortSignal | undefined; onDelta?: ((delta: string) => void) | undefined; onTool?: ((e: { phase: 'call' | 'result'; name: string; invocationId?: string; arg?: string; result?: string; isError?: boolean }) => void) | undefined; sessionKey?: string | undefined }, ): Promise; readonly model: string; } diff --git a/src/transport/conversation.ts b/src/transport/conversation.ts index 705e248..28fe56b 100644 --- a/src/transport/conversation.ts +++ b/src/transport/conversation.ts @@ -6,6 +6,9 @@ import type { FastifyInstance } from 'fastify'; import type { Storage } from '../memory/contract.ts'; import { runLoop, LoopError } from '../loop.ts'; import type { EventSink } from '../orchestrate/event-sink.ts'; +import type { ExecutionAuthority, ExecutionAttempt } from '../orchestrate/execution-authority.ts'; +import type { TurnCommand } from '../orchestrate/turn-engine.ts'; +import type { TextBlockActivity } from '../provider/activity.ts'; import type { Provider, ProviderMessage, ProviderResponse, ProviderTool } from '../provider/index.ts'; import type { Policy, ToolRegistry } from '../tools.ts'; import type { TurnEvent } from '../loop.ts'; @@ -55,8 +58,17 @@ export interface ConversationDeps { * semantics of when this fires — butler stays workflow-agnostic. */ prepareTurn?: (sessionId: string, body: { content: string; commands?: string[] }) => { promptBlock?: string; provider?: Provider } | null | Promise<{ promptBlock?: string; provider?: Provider } | null>; /** v0.6.2: called after the turn settles (success OR failure) — the host's - * post-edit window (commit/push/PR/cleanup around an editor turn). */ - afterTurn?: (sessionId: string, outcome: { ok: boolean; turnId: string; answer: string }) => void; + * post-edit window (commit/push/PR/cleanup around an editor turn). + * chat-reliability: may return a Promise — the POST NEVER awaits it, but + * under an execution authority the tail is registered as a tracked worker + * so normal shutdown drains it before releasing ownership. */ + afterTurn?: (sessionId: string, outcome: { ok: boolean; turnId: string; answer: string }) => void | Promise; + /** chat-reliability F1/F2: durable execution authority. When set, message + * admission is cross-process atomic (control-DB election + one-transaction + * durable admission), Stop targets the owned attempt, and the session read + * reports the common execution snapshot. When unset, the legacy + * process-local reservation applies (single-instance deployments only). */ + execution?: ExecutionAuthority; } export interface TurnOutcome { @@ -96,6 +108,14 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver // abort the active turn (review F1): clean ABORTED outcome, partial transcript persisted app.post('/api/agent/sessions/:id/abort', async (req, reply) => { const { id } = req.params as { id: string }; + if (deps.execution) { + const body = (req.body ?? {}) as { expectedTurnId?: string }; + const r = deps.execution.requestStop(id, body.expectedTurnId); + if (r.status === 'no_active') return reply.code(409).send({ error: 'no active turn on this session' }); + if (r.status === 'stale') return reply.code(409).send({ error: 'stale turn target' }); + if (r.status === 'nonlocal') return reply.code(503).send({ error: 'active turn is not locally owned' }); + return { aborting: true, turnId: r.turnId }; + } const ac = activeTurns.get(id); if (!ac) return reply.code(409).send({ error: 'no active turn on this session' }); ac.abort(); @@ -117,6 +137,13 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver // live-turn fact for reloads: a user message with no answer yet is either // ABORTED (dead turn) or IN FLIGHT (still running here or in a sibling // process) — the UI must not guess (sess_e2e40228 lesson) + if (deps.execution) { + // chat-reliability common read: positively-known local ownership is the + // ONLY thing reported as liveTurn; uncertainty is explicit, never idle. + const ex = deps.execution.readSession(id); + const liveTurn = ex.state === 'preparing' || ex.state === 'running' || ex.state === 'stopping' ? { turnId: ex.turnId, startedAt: ex.startedAt } : null; + return { session: row, messages: await storage.sessionMessages(id), execution: ex, ...(liveTurn ? { liveTurn } : {}) }; + } let liveTurn: { turnId: string; startedAt: number } | null = null; if (deps.sink) { try { @@ -142,62 +169,159 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver app.post('/api/agent/sessions/:id/messages', async (req, reply) => { const { id } = req.params as { id: string }; - const b = (req.body ?? {}) as { content?: string; commands?: string[]; model?: string }; + const b = (req.body ?? {}) as { content?: string; commands?: string[]; model?: string; clientRequestId?: string }; if (!b.content?.trim()) return reply.code(400).send({ error: 'content required' }); const session = await storage.getSession(id); if (!session) return reply.code(404).send({ error: 'session not found' }); - if (activeTurns.has(id)) return reply.code(409).send({ error: 'turn already in flight on this session' }); - + const content = b.content; // narrowed string (validated above) — safe in closures + const exec = deps.execution ?? null; + // F1 (chat-reliability): check-and-acquire is atomic — no await between + // check and acquire — and the reservation precedes every durable write, + // history/stash preparation and host hook. Every exit path below releases + // it; the admission race loser gets 409 BEFORE any side effect. + let ac: AbortController; + let durableTurnId: string; + let ownedAttempt: ExecutionAttempt | null = null; + if (exec) { + // authority-driven admission: cross-process atomic via the control-DB + // election + one-transaction durable admission (events + receipt + + // ledger row land together; the intended turn.started is required). + const res = exec.tryReserve(id); + if (res.status === 'busy') return reply.code(409).send({ error: 'turn already in flight on this session', ...(res.turnId !== undefined ? { turnId: res.turnId } : {}) }); + if (res.status === 'unavailable') return reply.code(503).send({ error: `chat execution unavailable (${res.reason}) — try again later`, code: res.reason }); + const attempt = res.attempt; + const adm = exec.admit(attempt, { content, principalId: session.user_id ?? 'anonymous', ...(b.clientRequestId !== undefined ? { requestId: b.clientRequestId } : {}) }); + if (adm.status !== 'admitted') { + if (adm.code === 'REQUEST_ALREADY_ADMITTED') return reply.code(409).send({ error: 'request already admitted', code: adm.code, ...(adm.turnId !== undefined ? { turnId: adm.turnId } : {}) }); + if (adm.code === 'REQUEST_ID_REUSE') return reply.code(409).send({ error: 'client request id was already used with different content', code: adm.code }); + if (adm.code === 'SESSION_ACTIVE') return reply.code(409).send({ error: 'turn already in flight on this session', code: adm.code }); + if (adm.code === 'QUEUE_DECISION') return reply.code(409).send({ error: 'session is resolving prior work — resend to run as the next turn', code: adm.code }); + return reply.code(500).send({ error: `admission failed: ${adm.message ?? adm.code}`, code: adm.code }); + } + ownedAttempt = attempt; + ac = attempt.controller; + durableTurnId = attempt.turnId; + } else { + if (activeTurns.has(id)) return reply.code(409).send({ error: 'turn already in flight on this session' }); + ac = new AbortController(); + activeTurns.set(id, ac); + durableTurnId = `t_${crypto.randomUUID().slice(0, 12)}`; + } + // durable-write seam: authority path fences every command by the owned + // attempt; local path writes through the sink directly. + const durable = (cmd: TurnCommand, key: string): void => { + if (exec && ownedAttempt) exec.handleOwned(ownedAttempt, id, cmd, key); + else if (deps.sink) deps.sink.handle(id, cmd, key); + }; + // chat-reliability F2 (text block lifecycle): durable block metadata for + // the chunk rail, derived from the loop's block tracker (onActivity). + // start/end persist as their own chunk commands with an EMPTY scalar + // delta — the delta text still rides its own delta-phase chunk, so the + // concatenated scalar stream stays byte-exact and legacy readers are + // untouched. Per-delta chunks gain block identity via the per-request + // hop→block map (text.delta needs no command of its own). + const blockRail = () => { + const byHop = new Map(); + return { + onActivity: (a: TextBlockActivity): void => { + if (!deps.sink && !exec) return; + if (a.type === 'text.start') { + byHop.set(a.hop, { id: a.blockId, ...(a.native !== undefined ? { native: a.native } : {}) }); + durable({ kind: 'turn.chunk', sessionId: id, turnId: durableTurnId, delta: '', block: { id: a.blockId, phase: 'start', ...(a.native !== undefined ? { native: a.native } : {}) } } as TurnCommand, `chat:${durableTurnId}:d:${Math.random()}`); + } else if (a.type === 'text.end') { + durable({ kind: 'turn.chunk', sessionId: id, turnId: durableTurnId, delta: '', block: { id: a.blockId, phase: 'end', ...(a.native !== undefined ? { native: a.native } : {}), text: a.text } } as TurnCommand, `chat:${durableTurnId}:d:${Math.random()}`); + } + }, + blockFor: (hop: number): { id: string; phase: 'delta'; native?: string } | undefined => { + const b = byHop.get(hop); + return b ? { id: b.id, phase: 'delta' as const, ...(b.native !== undefined ? { native: b.native } : {}) } : undefined; + }, + }; + }; + // These survive the preparation window — declared OUTSIDE the prep try so + // the execution section can read them (block scoping). + let resumeNote = ''; + let seeded: ProviderMessage[] = []; + let history: Array<{ role: 'user' | 'assistant'; content: string }> = []; + try { // R3-S5 single-writer rule: user message rides the durable log when a sink // is wired; the projection writes butler_messages. Legacy path (no sink) - // keeps the direct append so existing hosts don't break. - const durableTurnId = `t_${crypto.randomUUID().slice(0, 12)}`; - if (deps.sink) { + // keeps the direct append so existing hosts don't break. The AUTHORITY + // path already committed turn.requested/user_message/started + the ledger + // row inside admit(). + if (exec) { + // committed atomically by admit() — nothing to do here + } else if (deps.sink) { // R3 queue-drain: a followUp message queued against an earlier (now-dead) // turn starts ITS OWN turn — the queue must never strand content silently. const st = deps.sink.state(id); - const queuedIdx = st.queue.followUp.indexOf(b.content); + const queuedIdx = st.queue.followUp.indexOf(content); const noActive = st.activeTurnId === null; if (queuedIdx !== -1 && noActive) { // drain: this request IS the queued message → run as the next turn deps.sink.handle(id, { kind: 'message.queue', sessionId: id, content: b.content, disposition: 'steering' }, `chat:${id}:${durableTurnId}:dequeue`); } - deps.sink.handle(id, { kind: 'turn.request', sessionId: id, turnId: durableTurnId, prompt: b.content, refIds: [], at: Date.now() }, `chat:${id}:${durableTurnId}`); + deps.sink.handle(id, { kind: 'turn.request', sessionId: id, turnId: durableTurnId, prompt: content, refIds: [], at: Date.now() }, `chat:${id}:${durableTurnId}`); } else { await storage.appendMessage({ session_id: id, role: 'user', content: b.content, tool_calls_json: null, tool_results_json: null, tokens_in: 0, tokens_out: 0 }); } // resume context from rows alone (condition 4) - const history = (await storage.sessionMessages(id)) + history = (await storage.sessionMessages(id)) .filter((m) => m.role === 'user' || (m.role === 'assistant' && m.content)) .map((m) => ({ role: m.role as 'user' | 'assistant', content: m.content ?? '' })); // no-hassle resume (2026-10-21 plan): if the previous turn died on the // budget guard, its transcript (tool results included) was stashed — // seed it back so the model CONTINUES instead of redoing (and re-dying - // on) the same hops. One-shot: takePartial consumes the stash. - let resumeNote = ''; - let seeded: ProviderMessage[] = []; - try { - const stash = await storage.takePartial(id); - if (stash) { - seeded = JSON.parse(stash.json) as ProviderMessage[]; - resumeNote = '\n\n[resume] The previous turn on this session ran out of runway mid-task. Its full transcript, including tool results, is prepended to the history below. Continue that work from where it stopped — do not repeat completed steps.'; - } - } catch { /* a corrupt stash must never kill the next turn */ } + // on) the same hops. AUTHORITY path: the stash is NOT consumed during + // preparation — beginExecution() snapshot-CAS-consumes + // it at the committed execution boundary, so a preparation failure can + // never destroy a continuation. + if (!exec) { + try { + const stash = await storage.takePartial(id); + if (stash) { + seeded = JSON.parse(stash.json) as ProviderMessage[]; + resumeNote = '\n\n[resume] The previous turn on this session ran out of runway mid-task. Its full transcript, including tool results, is prepended to the history below. Continue that work from where it stopped — do not repeat completed steps.'; + } + } catch { /* a corrupt stash must never kill the next turn */ } + } + } catch (prepErr) { + // F1: a preparation failure (durable write, history read, stash parse) + // must never strand the session reservation. + if (exec && ownedAttempt) exec.releaseAttempt(ownedAttempt); + else activeTurns.delete(id); + return reply.code(500).send({ error: `turn preparation failed: ${prepErr instanceof Error ? prepErr.message : String(prepErr)}` }); + } const outcome: TurnOutcome = { sessionId: id, ok: false, tokens: { in: 0, out: 0 }, hops: 0 }; - const ac = new AbortController(); - activeTurns.set(id, ac); + // ownership liveness flag: the disconnect hook may abort only while THIS + // request still owns the turn (identity guard for both admission paths) + let ownsLiveTurn = true; + const releaseOwnership = async (code: string): Promise => { + ownsLiveTurn = false; + if (exec && ownedAttempt) await exec.settle(ownedAttempt, code); + else activeTurns.delete(id); + }; + // host tail: fire-and-forget for the POST, but DRAINABLE under an + // authority (tracked worker) — a returned promise never blocks the envelope + const invokeAfterTurn = (ok: boolean, answer: string): void => { + let tail: void | Promise; + try { tail = deps.afterTurn?.(id, { ok, turnId: durableTurnId, answer }); } catch { return; /* host hook best-effort */ } + if (tail instanceof Promise) { + if (exec) void exec.trackWorker({ turnId: durableTurnId, identity: { seam: 'afterTurn' } }, 'afterTurn', () => tail); + else void tail.catch(() => undefined); + } + }; // #7: client disconnect mid-flight aborts the turn. The seam is the // RESPONSE stream: req.raw 'close' fires when the request body completes // (not socket death); reply.raw 'close' with !writableEnded = the client - // vanished before the answer was flushed. The map-identity guard makes - // the normal post-response close (which fires after activeTurns.delete) - // a no-op. + // vanished before the answer was flushed. The liveness guard makes + // the normal post-response close (which fires after release) a no-op. if (deps.abortOnDisconnect) { reply.raw.on('close', () => { - if (!reply.raw.writableEnded && activeTurns.get(id) === ac) ac.abort(); + if (!reply.raw.writableEnded && ownsLiveTurn) ac.abort(); }); } try { @@ -211,7 +335,7 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver try { const stored = await storage.noteEmbeddings(deps.embedder.model); if (stored.length) { - const qv = (await deps.embedder.embed([b.content]))[0]!; + const qv = (await deps.embedder.embed([content]))[0]!; const byId = new Map(notes.map((n) => [n.id, n])); const ranked = stored .map((se) => ({ note: byId.get(se.noteId), score: cosine(qv, Array.from(se.vector)) })) @@ -246,43 +370,110 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver try { prep = await deps.prepareTurn?.(id, b as { content: string; commands?: string[] }) ?? null; } catch { prep = null; } // decoration failure must never kill the turn const turnProvider = prep?.provider ?? (routedComplete ? { complete: (m, t, o) => routedComplete(m, t, o), model: routedModel ?? deps.provider.model } : deps.provider); - const result = await runLoop(deps.systemPrompt + cmdBlock + notesBlock + (prep?.promptBlock ?? '') + resumeNote, b.content, { - sessionKey: id, // R2-F4: per-Lionidas-session claude --resume isolation - // prior turns only — the current message rides as userMessage; a - // stashed partial (if any) rides FIRST: it is the most recent context - history: [...seeded, ...(history.length > 1 ? history.slice(-40, -1) : [])], - provider: turnProvider, - tools: toolSet.map((t) => toolFromRegistry(t.name, registry, policy, { sessionId: id })), - ...(deps.budget ? { budget: deps.budget } : {}), - signal: ac.signal, - // rev10: live token stream → transport events (WS bridge streams UI) - ...(deps.onDelta ? { onDelta: deps.onDelta(id) } : {}), - onMessage: (msg, meta) => { - if (!sink) void storage.appendMessage({ - session_id: id, role: msg.role, content: msg.content, - tool_calls_json: msg.tool_calls ? JSON.stringify(msg.tool_calls) : null, - tool_results_json: null, tokens_in: meta.tokensIn, tokens_out: meta.tokensOut, - }); - }, - onTool: (t) => { - // R3 fix: tool calls are durable entities too — the rail correlates by - // toolCallId, and the replay log must show HOW an answer was derived. - if (!sink) return; - const stable = `tc_${durableTurnId}_${t.name}`; - if (t.phase === 'call') sink.handle(id, { kind: 'turn.tool', sessionId: id, turnId: durableTurnId, phase: 'call', toolCallId: stable, name: t.name, ...(t.arg !== undefined ? { arg: t.arg } : {}) }, `chat:${durableTurnId}:tc:${stable}`); - else sink.handle(id, { kind: 'turn.tool', sessionId: id, turnId: durableTurnId, phase: 'update', toolCallId: stable, ...(t.result !== undefined ? { result: t.result } : {}), ...(t.isError !== undefined ? { isError: t.isError } : {}) }, `chat:${durableTurnId}:tu:${stable}:${Math.random()}`); - }, - onEvent: (e) => { - deps.emit({ ...e, sessionId: id }); - if (sink) { - if (e.type === 'turn.delta' && e.detail && typeof e.detail === 'object' && 'delta' in e.detail) sink.handle(id, { kind: 'turn.chunk', sessionId: id, turnId: durableTurnId, delta: String((e.detail as { delta: unknown }).delta) }, `chat:${durableTurnId}:d:${Math.random()}`); - if (e.type === 'provider.response' && e.detail && typeof e.detail === 'object' && 'tokens' in e.detail) { - const t = (e.detail as { tokens?: { promptTokens?: number; completionTokens?: number } }).tokens; - if (t) sink.handle(id, { kind: 'turn.usage', sessionId: id, turnId: durableTurnId, meters: { cost: (t.promptTokens ?? 0) + (t.completionTokens ?? 0), contextUsed: t.promptTokens ?? 0, contextWindow: deps.contextWindow ?? 200_000 } }, `chat:${durableTurnId}:u:${Math.random()}`); - } - } - }, - }); + // AUTHORITY execution boundary: snapshot + CAS-consume the continuation + // stash and mark the execution `executing` BEFORE any provider call. + // Stop-before-start is an honest aborted outcome (no provider ran). + if (exec && ownedAttempt) { + const begun = await exec.beginExecution(ownedAttempt); + if ('status' in begun && begun.status === 'rejected') { + durable({ kind: 'turn.fail', sessionId: id, turnId: durableTurnId, error: 'aborted before the provider started — no work was lost; resend to retry' } as TurnCommand, `chat:${durableTurnId}:fail`); + await releaseOwnership(begun.code); + outcome.ok = false; + outcome.error = 'aborted before start — no provider ran'; + outcome.code = 'ABORTED'; + await storage.updateSession(id, { status: 'active' }); + return { turn: outcome }; + } + const begunSeeded = (begun as { seeded: unknown[] | null }).seeded; + if (begunSeeded && begunSeeded.length) { + seeded = begunSeeded as ProviderMessage[]; + resumeNote = '\n\n[resume] The previous turn on this session ran out of runway mid-task. Its full transcript, including tool results, is prepended to the history below. Continue that work from where it stopped — do not repeat completed steps.'; + } + } + const blocks = blockRail(); + const result = await (exec && ownedAttempt + ? exec.trackWorker({ turnId: durableTurnId, identity: { seam: 'provider' } }, 'provider', () => runLoop(deps.systemPrompt + cmdBlock + notesBlock + (prep?.promptBlock ?? '') + resumeNote, content, { + sessionKey: id, // R2-F4: per-Lionidas-session claude --resume isolation + // prior turns only — the current message rides as userMessage; a + // stashed partial (if any) rides FIRST: it is the most recent context + history: [...seeded, ...(history.length > 1 ? history.slice(-40, -1) : [])], + provider: turnProvider, + tools: toolSet.map((t) => toolFromRegistry(t.name, registry, policy, { sessionId: id })), + ...(deps.budget ? { budget: deps.budget } : {}), + signal: ac.signal, + // rev10: live token stream → transport events (WS bridge streams UI) + ...(deps.onDelta ? { onDelta: deps.onDelta(id) } : {}), + onMessage: (msg, meta) => { + if (!sink) void storage.appendMessage({ + session_id: id, role: msg.role, content: msg.content, + tool_calls_json: msg.tool_calls ? JSON.stringify(msg.tool_calls) : null, + tool_results_json: null, tokens_in: meta.tokensIn, tokens_out: meta.tokensOut, + }); + }, + onTool: (t) => { + // R3 fix: tool calls are durable entities too — the rail correlates by + // toolCallId, and the replay log must show HOW an answer was derived. + // chat-reliability F1: identity is per INVOCATION (provider-native + // call id) — repeated same-name invocations are distinct durable + // calls; the name-derived id remains only for legacy harness events + // that carry no native id. + if (!sink && !exec) return; + const stable = t.invocationId !== undefined ? `tc_${durableTurnId}_i_${t.invocationId}` : `tc_${durableTurnId}_${t.name}`; + if (t.phase === 'call') durable({ kind: 'turn.tool', sessionId: id, turnId: durableTurnId, phase: 'call', toolCallId: stable, name: t.name, ...(t.arg !== undefined ? { arg: t.arg } : {}) } as TurnCommand, `chat:${durableTurnId}:tc:${stable}`); + else durable({ kind: 'turn.tool', sessionId: id, turnId: durableTurnId, phase: 'update', toolCallId: stable, ...(t.result !== undefined ? { result: t.result } : {}), ...(t.isError !== undefined ? { isError: t.isError } : {}) } as TurnCommand, `chat:${durableTurnId}:tu:${stable}:${Math.random()}`); + }, + // F2: block start/end ride their own chunk commands; per-delta + // chunks pick up block identity in the onEvent chunk handler. + onActivity: blocks.onActivity, + onEvent: (e) => { + deps.emit({ ...e, sessionId: id }); + if (sink || exec) { + if (e.type === 'turn.delta' && e.detail && typeof e.detail === 'object' && 'delta' in e.detail) durable({ kind: 'turn.chunk', sessionId: id, turnId: durableTurnId, delta: String((e.detail as { delta: unknown }).delta), ...(blocks.blockFor(e.hop) ? { block: blocks.blockFor(e.hop) } : {}) } as TurnCommand, `chat:${durableTurnId}:d:${Math.random()}`); + if (e.type === 'provider.response' && e.detail && typeof e.detail === 'object' && 'tokens' in e.detail) { + const t = (e.detail as { tokens?: { promptTokens?: number; completionTokens?: number } }).tokens; + if (t) durable({ kind: 'turn.usage', sessionId: id, turnId: durableTurnId, meters: { cost: (t.promptTokens ?? 0) + (t.completionTokens ?? 0), contextUsed: t.promptTokens ?? 0, contextWindow: deps.contextWindow ?? 200_000 } } as TurnCommand, `chat:${durableTurnId}:u:${Math.random()}`); + } + } + }, + })) as unknown as Awaited> + : runLoop(deps.systemPrompt + cmdBlock + notesBlock + (prep?.promptBlock ?? '') + resumeNote, content, { + sessionKey: id, // R2-F4: per-Lionidas-session claude --resume isolation + // prior turns only — the current message rides as userMessage; a + // stashed partial (if any) rides FIRST: it is the most recent context + history: [...seeded, ...(history.length > 1 ? history.slice(-40, -1) : [])], + provider: turnProvider, + tools: toolSet.map((t) => toolFromRegistry(t.name, registry, policy, { sessionId: id })), + ...(deps.budget ? { budget: deps.budget } : {}), + signal: ac.signal, + // rev10: live token stream → transport events (WS bridge streams UI) + ...(deps.onDelta ? { onDelta: deps.onDelta(id) } : {}), + onMessage: (msg, meta) => { + if (!sink) void storage.appendMessage({ + session_id: id, role: msg.role, content: msg.content, + tool_calls_json: msg.tool_calls ? JSON.stringify(msg.tool_calls) : null, + tool_results_json: null, tokens_in: meta.tokensIn, tokens_out: meta.tokensOut, + }); + }, + onTool: (t) => { + if (!sink && !exec) return; + const stable = t.invocationId !== undefined ? `tc_${durableTurnId}_i_${t.invocationId}` : `tc_${durableTurnId}_${t.name}`; + if (t.phase === 'call') durable({ kind: 'turn.tool', sessionId: id, turnId: durableTurnId, phase: 'call', toolCallId: stable, name: t.name, ...(t.arg !== undefined ? { arg: t.arg } : {}) } as TurnCommand, `chat:${durableTurnId}:tc:${stable}`); + else durable({ kind: 'turn.tool', sessionId: id, turnId: durableTurnId, phase: 'update', toolCallId: stable, ...(t.result !== undefined ? { result: t.result } : {}), ...(t.isError !== undefined ? { isError: t.isError } : {}) } as TurnCommand, `chat:${durableTurnId}:tu:${stable}:${Math.random()}`); + }, + // F2: block start/end ride their own chunk commands; per-delta + // chunks pick up block identity in the onEvent chunk handler. + onActivity: blocks.onActivity, + onEvent: (e) => { + deps.emit({ ...e, sessionId: id }); + if (sink || exec) { + if (e.type === 'turn.delta' && e.detail && typeof e.detail === 'object' && 'delta' in e.detail) durable({ kind: 'turn.chunk', sessionId: id, turnId: durableTurnId, delta: String((e.detail as { delta: unknown }).delta), ...(blocks.blockFor(e.hop) ? { block: blocks.blockFor(e.hop) } : {}) } as TurnCommand, `chat:${durableTurnId}:d:${Math.random()}`); + if (e.type === 'provider.response' && e.detail && typeof e.detail === 'object' && 'tokens' in e.detail) { + const t = (e.detail as { tokens?: { promptTokens?: number; completionTokens?: number } }).tokens; + if (t) durable({ kind: 'turn.usage', sessionId: id, turnId: durableTurnId, meters: { cost: (t.promptTokens ?? 0) + (t.completionTokens ?? 0), contextUsed: t.promptTokens ?? 0, contextWindow: deps.contextWindow ?? 200_000 } } as TurnCommand, `chat:${durableTurnId}:u:${Math.random()}`); + } + } + }, + })); // v0.5.4: an EMPTY answer after real work is not success. A turn that // streamed narration or called tools and still ends with "" burned its // context (the 05:46 incident: 1M tokens, 40 tool calls, answer "") — @@ -292,21 +483,47 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver const turnEvents = sink ? sink.events(id) : []; const didWork = turnEvents.some((e) => e.turn_id === durableTurnId && (e.kind === 'chunk' || e.kind === 'tool_call' || e.kind === 'tool_call_update')); if (!result.answer.trim() && didWork) { - if (sink) sink.handle(id, { kind: 'turn.fail', sessionId: id, turnId: durableTurnId, error: 'turn ended with an empty answer after processing (likely context exhaustion mid-run) — partial transcript preserved in the event log; resend to continue' }, `chat:${durableTurnId}:emptyfail`); outcome.ok = false; outcome.error = 'empty answer after work — turn failed honestly (see event log)'; - await storage.updateSession(id, { status: 'error' }); - activeTurns.delete(id); + if (exec && ownedAttempt) { + await exec.finalizeOwned(ownedAttempt, { + kind: 'turn.fail', + error: 'turn ended with an empty answer after processing (likely context exhaustion mid-run) — partial transcript preserved in the event log; resend to continue', + commandKey: `chat:${durableTurnId}:emptyfail`, + settleCode: 'EMPTY_AFTER_WORK', + accounting: { status: 'error' }, + }); + ownsLiveTurn = false; + } else { + durable({ kind: 'turn.fail', sessionId: id, turnId: durableTurnId, error: 'turn ended with an empty answer after processing (likely context exhaustion mid-run) — partial transcript preserved in the event log; resend to continue' } as TurnCommand, `chat:${durableTurnId}:emptyfail`); + await storage.updateSession(id, { status: 'error' }); + await releaseOwnership('EMPTY_AFTER_WORK'); + } return { turn: outcome }; } - if (sink) sink.handle(id, { kind: 'turn.complete', sessionId: id, turnId: durableTurnId, answer: result.answer }, `chat:${durableTurnId}:done`); outcome.ok = true; outcome.answer = result.answer; outcome.tokens = result.tokens; outcome.hops = result.hops; - await storage.updateSession(id, { status: 'active', tokens_in: session.tokens_in + result.tokens.in, tokens_out: session.tokens_out + result.tokens.out }); - activeTurns.delete(id); - try { deps.afterTurn?.(id, { ok: true, turnId: durableTurnId, answer: result.answer }); } catch { /* host hook best-effort */ } + if (exec && ownedAttempt) { + // §3.5 owned ATOMIC finalization: terminal event + projections + token + // accounting + ledger settle commit together; the envelope reports + // success only for a committed terminal. + const fin = await exec.finalizeOwned(ownedAttempt, { + kind: 'turn.complete', + answer: result.answer, + commandKey: `chat:${durableTurnId}:done`, + settleCode: 'COMPLETED', + accounting: { status: 'active', tokensInDelta: result.tokens.in, tokensOutDelta: result.tokens.out }, + }); + ownsLiveTurn = false; + if (!fin.accounted) await storage.updateSession(id, { status: 'active', tokens_in: session.tokens_in + result.tokens.in, tokens_out: session.tokens_out + result.tokens.out }); + } else { + durable({ kind: 'turn.complete', sessionId: id, turnId: durableTurnId, answer: result.answer } as TurnCommand, `chat:${durableTurnId}:done`); + await storage.updateSession(id, { status: 'active', tokens_in: session.tokens_in + result.tokens.in, tokens_out: session.tokens_out + result.tokens.out }); + await releaseOwnership('COMPLETED'); + } + invokeAfterTurn(true, result.answer); return { turn: outcome }; } catch (err) { const e = err as { code?: string; message?: string }; @@ -317,16 +534,39 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver // provider, unknown) keeps today's honest session-poisoning. const budgetExhausted = err instanceof LoopError && err.code === 'BUDGET_EXCEEDED'; if (budgetExhausted) { - try { await storage.stashPartial(id, durableTurnId, JSON.stringify(err.partial ?? [])); } catch { /* stash is best-effort */ } + // AUTHORITY path: the budget stash joins the atomic finalization + // transaction below (finalizeOwned), never a separate best-effort write + if (!exec) { try { await storage.stashPartial(id, durableTurnId, JSON.stringify(err.partial ?? [])); } catch { /* stash is best-effort */ } } outcome.error = 'ran out of runway — the work so far is kept; send a message to continue from where it stopped'; } else { outcome.error = e.message ?? String(err); } - if (deps.sink) deps.sink.handle(id, { kind: 'turn.fail', sessionId: id, turnId: durableTurnId, error: outcome.error }, `chat:${durableTurnId}:fail`); outcome.code = e.code ?? undefined; - await storage.updateSession(id, { status: budgetExhausted ? 'active' : 'error' }); - activeTurns.delete(id); - try { deps.afterTurn?.(id, { ok: false, turnId: durableTurnId, answer: '' }); } catch { /* host hook best-effort */ } + if (exec && ownedAttempt) { + // §3.5 owned ATOMIC finalization: terminal event + budget stash + + // session accounting + ledger settle in ONE commit. A storage failure + // leaves the attempt reserved for an idempotent retry and reports + // outcome-pending — never a false success. + try { + const fin = await exec.finalizeOwned(ownedAttempt, { + kind: 'turn.fail', + error: outcome.error, + commandKey: `chat:${durableTurnId}:fail`, + settleCode: budgetExhausted ? 'BUDGET_EXCEEDED' : (e.code ?? 'FAILED'), + ...(budgetExhausted && err instanceof LoopError ? { stash: { json: JSON.stringify(err.partial ?? []) } } : {}), + accounting: { status: budgetExhausted ? 'active' : 'error' }, + }); + ownsLiveTurn = false; + if (!fin.accounted) await storage.updateSession(id, { status: budgetExhausted ? 'active' : 'error' }); + } catch (finErr) { + return reply.code(503).send({ error: `finalization storage failure — outcome pending, the turn stays reserved for idempotent retry: ${finErr instanceof Error ? finErr.message : String(finErr)}`, code: 'FINALIZATION_PENDING' }); + } + } else { + if (deps.sink || exec) durable({ kind: 'turn.fail', sessionId: id, turnId: durableTurnId, error: outcome.error } as TurnCommand, `chat:${durableTurnId}:fail`); + await storage.updateSession(id, { status: budgetExhausted ? 'active' : 'error' }); + await releaseOwnership(budgetExhausted ? 'BUDGET_EXCEEDED' : (e.code ?? 'FAILED')); + } + invokeAfterTurn(false, ''); return { turn: outcome }; // the turn failed; the REQUEST succeeded — outcome carries the verdict } }); diff --git a/test/activity-blocks.test.ts b/test/activity-blocks.test.ts new file mode 100644 index 0000000..4a146a9 --- /dev/null +++ b/test/activity-blocks.test.ts @@ -0,0 +1,185 @@ +/** chat-reliability F2 (text block lifecycle): a streamed turn's durable + * chunk rail must carry BLOCK identity — one block per assistant message + * (hop) — so replayed narration separates cleanly instead of gluing + * ("…check.develop…"). Byte-exactness: chunk splits inside a block + * concatenate with zero whitespace injection; the legacy scalar `delta` + * field stays on EVERY chunk. */ +import { describe, expect, it, afterAll } from 'vitest'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import Fastify from 'fastify'; +import type { FastifyInstance } from 'fastify'; +import Database from 'better-sqlite3'; +import type { Provider, ProviderMessage, ProviderResponse, ProviderTool } from '../src/provider/types.ts'; +import { SqliteStorage } from '../src/memory/sqlite.ts'; +import { EventSink } from '../src/orchestrate/event-sink.ts'; +import { ToolRegistry } from '../src/tools.ts'; +import { registerConversationApi } from '../src/transport/conversation.ts'; + +type Opts = Parameters[2]; +const dirs: string[] = []; +afterAll(() => { for (const d of dirs) rmSync(d, { recursive: true, force: true }); }); + +async function boot(provider: Provider): Promise<{ app: FastifyInstance; url: string; sink: EventSink; storage: SqliteStorage }> { + const dir = mkdtempSync(join(tmpdir(), 'butler-blkdur-')); + dirs.push(dir); + const storage = new SqliteStorage(join(dir, 'agent.db')); + const sink = new EventSink(storage['db'] as InstanceType, { passive: true, enqueueRunTurn: false }); + const registry = new ToolRegistry().register({ + name: 'probe', description: 'p', permission: 'read', parameters: { type: 'object', properties: { q: { type: 'string' } } }, + execute: async (args) => `observed:${String(args['q'])}`, + }); + const app = Fastify(); + await registerConversationApi(app, { + storage, provider, registry, policy: { name: 'draft', allows: 'draft' }, + systemPrompt: 'sys', emit: () => undefined, sink, + }); + await app.listen({ port: 0, host: '127.0.0.1' }); + return { app, url: `http://127.0.0.1:${(app.server.address() as { port: number }).port}`, sink, storage }; +} + +/** streams `deltas` live via opts.onDelta, then completes with content + * (which may deliberately DIFFER from the stream — snapshot vs stream). */ +class StreamingProvider implements Provider { + readonly model = 'streaming'; + calls = 0; + constructor(private readonly turns: Array<{ deltas?: string[]; content?: string; toolCalls?: ProviderMessage['tool_calls'] }>) {} + async complete(_m: ProviderMessage[], _t: ProviderTool[], opts: Opts): Promise { + const turn = this.turns[Math.min(this.calls, this.turns.length - 1)]!; + this.calls += 1; + for (const d of turn.deltas ?? []) opts.onDelta?.(d); + return { + message: { role: 'assistant', content: turn.content ?? null, ...(turn.toolCalls ? { tool_calls: turn.toolCalls } : {}) }, + finishReason: turn.toolCalls ? 'tool_calls' : 'stop', + usage: { promptTokens: 5, completionTokens: 2 }, + }; + } +} + +type ChunkDetail = { delta: string; block?: { id: string; phase: 'start' | 'delta' | 'end'; native?: string; text?: string } }; + +async function runTurn(rig: { url: string }, content: string): Promise { + const created = await (await fetch(`${rig.url}/api/agent/sessions`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' })).json() as { session: { id: string } }; + const res = await fetch(`${rig.url}/api/agent/sessions/${created.session.id}/messages`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ content }) }); + expect(res.status).toBe(200); + expect(((await res.json()) as { turn: { ok: boolean } }).turn.ok).toBe(true); + return created.session.id; +} + +function chunkEvents(sink: EventSink, sessionId: string): Array<{ detail: ChunkDetail }> { + return sink.events(sessionId).filter((e) => e.kind === 'chunk') as unknown as Array<{ detail: ChunkDetail }>; +} + +describe('text block lifecycle durability (F2)', () => { + it('a two-hop streamed turn: distinct block ids, byte-exact concatenation, scalar delta on every chunk', async () => { + // hop 2's completed content DIFFERS from its stream — proves text.end + // below prefers the streamed accumulation + const provider = new StreamingProvider([ + { deltas: ['Let me ', 'check.'], content: 'Let me check.', toolCalls: [{ id: 'c1', type: 'function', function: { name: 'probe', arguments: '{"q":"x"}' } }] }, + { deltas: ['de', 'velop'], content: 'develop-final' }, + ]); + const rig = await boot(provider); + try { + const sid = await runTurn(rig, 'run probe'); + const chunks = chunkEvents(rig.sink, sid); + expect(chunks.length).toBeGreaterThan(0); + // legacy contract: the scalar delta field is present on EVERY chunk + for (const c of chunks) expect(typeof c.detail.delta).toBe('string'); + const deltaChunks = chunks.filter((c) => c.detail.block?.phase === 'delta'); + expect(deltaChunks).toHaveLength(4); // 2 deltas per hop + for (const c of deltaChunks) expect(c.detail.delta.length).toBeGreaterThan(0); // real text rides the delta chunks + // one block per hop → distinct ids + const ids = [...new Set(deltaChunks.map((c) => c.detail.block!.id))]; + expect(ids).toHaveLength(2); + // byte-exact per block: chunk splits concatenate with no injection + const byBlock = new Map(); + for (const c of deltaChunks) byBlock.set(c.detail.block!.id, (byBlock.get(c.detail.block!.id) ?? '') + c.detail.delta); + expect([...byBlock.values()].sort()).toEqual(['Let me check.', 'develop'].sort()); + // de +velop = develop (the canonical split proof) + // start boundaries: one per block, EMPTY scalar delta (no text duplication) + const starts = chunks.filter((c) => c.detail.block?.phase === 'start'); + expect(starts).toHaveLength(2); + for (const s of starts) expect(s.detail.delta).toBe(''); + } finally { + await rig.app.close(); + rig.storage.close(); + } + }); + + it('text.end carries the full STREAMED text — not the completed snapshot', async () => { + const provider = new StreamingProvider([ + { deltas: ['Let me ', 'check.'], content: 'Let me check.', toolCalls: [{ id: 'c1', type: 'function', function: { name: 'probe', arguments: '{"q":"x"}' } }] }, + { deltas: ['de', 'velop'], content: 'develop-final' }, // snapshot ≠ stream + ]); + const rig = await boot(provider); + try { + const sid = await runTurn(rig, 'run probe'); + const ends = chunkEvents(rig.sink, sid).filter((c) => c.detail.block?.phase === 'end'); + expect(ends).toHaveLength(2); // one end per block + const endTexts = ends.map((c) => c.detail.block!.text); + expect(endTexts).toContain('develop'); // streamed accumulation wins + expect(endTexts).toContain('Let me check.'); + expect(endTexts).not.toContain('develop-final'); // content was NOT trusted + // end chunks keep the scalar field (empty — the text rode delta chunks) + for (const e of ends) expect(e.detail.delta).toBe(''); + } finally { + await rig.app.close(); + rig.storage.close(); + } + }); + + it('turn.delta bus events carry the live blockId; absent without a block', async () => { + const events: Array<{ type: string; hop: number; detail?: { delta?: string; blockId?: string } }> = []; + const dir = mkdtempSync(join(tmpdir(), 'butler-blkdur-')); + dirs.push(dir); + const storage = new SqliteStorage(join(dir, 'agent.db')); + const sink = new EventSink(storage['db'] as InstanceType, { passive: true, enqueueRunTurn: false }); + const registry = new ToolRegistry().register({ + name: 'probe', description: 'p', permission: 'read', parameters: { type: 'object', properties: { q: { type: 'string' } } }, + execute: async (args) => `observed:${String(args['q'])}`, + }); + const provider = new StreamingProvider([ + { deltas: ['a', 'b'], toolCalls: [{ id: 'c1', type: 'function', function: { name: 'probe', arguments: '{"q":"x"}' } }] }, + { deltas: ['c'], content: 'abc' }, + ]); + const app = Fastify(); + await registerConversationApi(app, { + storage, provider, registry, policy: { name: 'draft', allows: 'draft' }, + systemPrompt: 'sys', + emit: (e) => { events.push(e as unknown as { type: string; hop: number; detail?: { delta?: string; blockId?: string } }); }, + sink, + }); + await app.listen({ port: 0, host: '127.0.0.1' }); + try { + const created = await (await fetch(`http://127.0.0.1:${(app.server.address() as { port: number }).port}/api/agent/sessions`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' })).json() as { session: { id: string } }; + const res = await fetch(`http://127.0.0.1:${(app.server.address() as { port: number }).port}/api/agent/sessions/${created.session.id}/messages`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ content: 'stream two hops' }) }); + expect(res.status).toBe(200); + const deltas = events.filter((e) => e.type === 'turn.delta'); + expect(deltas.map((e) => e.detail?.delta)).toEqual(['a', 'b', 'c']); + expect(deltas[0]!.detail?.blockId).toBe('blk_h1'); // hops are 1-indexed + expect(deltas[1]!.detail?.blockId).toBe('blk_h1'); + expect(deltas[2]!.detail?.blockId).toBe('blk_h2'); // tool hop → NEW block id + } finally { + await app.close(); + storage.close(); + } + }); + + it('a hop with no deltas emits no block', async () => { + const provider = new StreamingProvider([ + { toolCalls: [{ id: 'c1', type: 'function', function: { name: 'probe', arguments: '{"q":"silent"}' } }] }, // no deltas + { content: 'done' }, // no deltas + ]); + const rig = await boot(provider); + try { + const sid = await runTurn(rig, 'run probe quietly'); + const chunks = chunkEvents(rig.sink, sid); + expect(chunks.filter((c) => c.detail.block !== undefined)).toHaveLength(0); // no phantom blocks + expect(chunks).toHaveLength(0); // nothing streamed at all + } finally { + await rig.app.close(); + rig.storage.close(); + } + }); +}); diff --git a/test/chat-tool-durability.test.ts b/test/chat-tool-durability.test.ts new file mode 100644 index 0000000..d9d0fab --- /dev/null +++ b/test/chat-tool-durability.test.ts @@ -0,0 +1,159 @@ +import { describe, expect, it, afterAll } from 'vitest'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import Fastify from 'fastify'; +import type { FastifyInstance } from 'fastify'; +import Database from 'better-sqlite3'; +import type { Provider, ProviderMessage, ProviderResponse, ProviderTool } from '../src/provider/types.ts'; +import { MockProvider } from '../src/provider/mock.ts'; +import { SqliteStorage } from '../src/memory/sqlite.ts'; +import { EventSink } from '../src/orchestrate/event-sink.ts'; +import { ToolRegistry } from '../src/tools.ts'; +import { registerConversationApi } from '../src/transport/conversation.ts'; + +/** chat-reliability F1 (tool durability): a provider-requested REGISTRY tool + * must land in the durable event log (tool_call + tool_call_update) — + * previously runtime tools emitted no onTool callbacks at all, so replayed + * history showed zero tool activity for real work. */ + +type Opts = Parameters[2]; +const dirs: string[] = []; +afterAll(() => { for (const d of dirs) rmSync(d, { recursive: true, force: true }); }); + +async function boot(provider: Provider, execute: (args: Record) => Promise = async (args) => `observed:${String(args['q'])}`): Promise<{ app: FastifyInstance; url: string; sink: EventSink; storage: SqliteStorage }> { + const dir = mkdtempSync(join(tmpdir(), 'butler-tooldur-')); + dirs.push(dir); + const agent = join(dir, 'agent.db'); + const storage = new SqliteStorage(agent); + const sink = new EventSink(storage['db'] as InstanceType, { passive: true, enqueueRunTurn: false }); + const registry = new ToolRegistry().register({ + name: 'probe', description: 'p', permission: 'read', parameters: { type: 'object', properties: { q: { type: 'string' } } }, + execute, + }); + const app = Fastify(); + await registerConversationApi(app, { + storage, provider, registry, policy: { name: 'draft', allows: 'draft' }, + systemPrompt: 'sys', emit: () => undefined, sink, + }); + await app.listen({ port: 0, host: '127.0.0.1' }); + return { app, url: `http://127.0.0.1:${(app.server.address() as { port: number }).port}`, sink, storage }; +} + +class AnsweringProvider implements Provider { + readonly model = 'answering'; + calls = 0; + constructor(private readonly turns: Array<{ toolCalls?: ProviderMessage['tool_calls']; content?: string }>) {} + async complete(_m: ProviderMessage[], _t: ProviderTool[], _o: Opts): Promise { + const turn = this.turns[Math.min(this.calls, this.turns.length - 1)]!; + this.calls += 1; + return { + message: { role: 'assistant', content: turn.content ?? null, ...(turn.toolCalls ? { tool_calls: turn.toolCalls } : {}) }, + finishReason: turn.toolCalls ? 'tool_calls' : 'stop', + usage: { promptTokens: 5, completionTokens: 2 }, + }; + } +} + +describe('runtime tool durability', () => { + it('a provider-requested registry tool lands durable tool_call + tool_call_update events', async () => { + const provider = new AnsweringProvider([ + { toolCalls: [{ id: 'c1', type: 'function', function: { name: 'probe', arguments: '{"q":"tree"}' } }] }, + { content: 'checked the tree' }, + ]); + const rig = await boot(provider); + try { + const created = await (await fetch(`${rig.url}/api/agent/sessions`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' })).json() as { session: { id: string } }; + const res = await fetch(`${rig.url}/api/agent/sessions/${created.session.id}/messages`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ content: 'run probe' }) }); + expect(res.status).toBe(200); + expect(((await res.json()) as { turn: { ok: boolean } }).turn.ok).toBe(true); + const events = rig.sink.events(created.session.id); + const calls = events.filter((e) => e.kind === 'tool_call'); + const updates = events.filter((e) => e.kind === 'tool_call_update'); + expect(calls.length).toBeGreaterThanOrEqual(1); // PREVIOUSLY 0 — the defect lock + expect(updates.length).toBeGreaterThanOrEqual(1); + expect((calls[0]!.detail as { name?: string }).name).toBe('probe'); + } finally { + await rig.app.close(); + rig.storage.close(); + } + }); + + it('two same-name invocations in one turn stay DISTINCT durable calls (per-invocation identity)', async () => { + const provider = new AnsweringProvider([ + { toolCalls: [ + { id: 'call_one', type: 'function', function: { name: 'probe', arguments: '{"q":"one"}' } }, + { id: 'call_two', type: 'function', function: { name: 'probe', arguments: '{"q":"two"}' } }, + ] }, + { content: 'ran it twice' }, + ]); + const rig = await boot(provider); + try { + const created = await (await fetch(`${rig.url}/api/agent/sessions`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' })).json() as { session: { id: string } }; + const res = await fetch(`${rig.url}/api/agent/sessions/${created.session.id}/messages`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ content: 'run probe twice' }) }); + expect(res.status).toBe(200); + const events = rig.sink.events(created.session.id); + const calls = events.filter((e) => e.kind === 'tool_call'); + const updates = events.filter((e) => e.kind === 'tool_call_update'); + expect(calls).toHaveLength(2); // PREVIOUSLY 1 — name-derived identity collapsed invocations + expect(new Set(calls.map((c) => (c.detail as { toolCallId?: string }).toolCallId)).size).toBe(2); + expect(updates).toHaveLength(2); + const args = calls.map((c) => (c.detail as { arg?: string }).arg); + expect(args).toContain('{"q":"one"}'); + expect(args).toContain('{"q":"two"}'); + } finally { + await rig.app.close(); + rig.storage.close(); + } + }); + + it('harness onTool events WITHOUT native ids get allocated identity — same-name stays distinct', async () => { + const provider: Provider = { + model: 'harness-same-name', + async complete(_m, _t, o) { + for (const value of ['one', 'two']) { + o.onTool?.({ phase: 'call', name: 'probe', arg: value }); + o.onTool?.({ phase: 'result', name: 'probe', result: value }); + } + return { message: { role: 'assistant', content: 'done' }, finishReason: 'stop', usage: null }; + }, + }; + const rig = await boot(provider); + try { + const created = await (await fetch(`${rig.url}/api/agent/sessions`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' })).json() as { session: { id: string } }; + const res = await fetch(`${rig.url}/api/agent/sessions/${created.session.id}/messages`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ content: 'run twice' }) }); + expect(res.status).toBe(200); + const events = rig.sink.events(created.session.id); + const calls = events.filter((e) => e.kind === 'tool_call'); + const ids = calls.map((c) => String((c.detail as { toolCallId?: string }).toolCallId)); + expect(calls).toHaveLength(2); // PREVIOUSLY 1 — name fallback collapsed the invocations + expect(new Set(ids).size).toBe(2); + expect(ids.every((id) => id.includes('_i_alloc_'))).toBe(true); // allocated identity, not name-derived + const args = calls.map((c) => String((c.detail as { arg?: string }).arg)); + expect(args).toEqual(['one', 'two']); + } finally { + await rig.app.close(); + rig.storage.close(); + } + }); + + it('a failed tool handler records an isError update, never a silent success', async () => { + const provider = new AnsweringProvider([ + { toolCalls: [{ id: 'c1', type: 'function', function: { name: 'probe', arguments: '{"q":"boom"}' } }] }, + { content: 'handled the failure' }, + ]); + const rig = await boot(provider, async () => { throw new Error('tool exploded'); }); + try { + const created = await (await fetch(`${rig.url}/api/agent/sessions`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' })).json() as { session: { id: string } }; + const res = await fetch(`${rig.url}/api/agent/sessions/${created.session.id}/messages`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ content: 'run probe' }) }); + expect(res.status).toBe(200); + const events = rig.sink.events(created.session.id); + const updates = events.filter((e) => e.kind === 'tool_call_update'); + expect(updates.length).toBeGreaterThanOrEqual(1); + expect((updates[0]!.detail as { isError?: boolean }).isError).toBe(true); // honest failure, recorded + } finally { + await rig.app.close(); + rig.storage.close(); + } + }); +}); diff --git a/test/claude-cli-activity.test.ts b/test/claude-cli-activity.test.ts new file mode 100644 index 0000000..c46d5e3 --- /dev/null +++ b/test/claude-cli-activity.test.ts @@ -0,0 +1,111 @@ +/** chat-reliability F2 — claude-cli parser fixes over a FAKE binary: + * (a) the completed assistant snapshot must REPLACE/complete the streamed + * block — never re-emit duplicated deltas; + * (b) native tool_use ids flow into onTool invocationId; + * (c) results are FULL (no 160-char clip) and EMPTY results are recorded. */ +import { describe, it, expect } from 'vitest'; +import { mkdtempSync, rmSync, writeFileSync, chmodSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { claudeCliProvider } from '../src/provider/claude-cli.ts'; + +type ToolEv = { phase: 'call' | 'result'; name: string; invocationId?: string; arg?: string; result?: string; isError?: boolean }; + +function fakeClaude(events: unknown[]): string { + const dir = mkdtempSync(join(tmpdir(), 'butler-cc-')); + const path = join(dir, 'claude'); + const script = `#!/usr/bin/env node +const evts = ${JSON.stringify(events)}; +process.stdout.write(evts.map((e) => JSON.stringify(e)).join('\\n') + '\\n'); +`; + writeFileSync(path, script); + chmodSync(path, 0o755); + return path; +} + +const MSGS = [ + { role: 'system' as const, content: 'sys prompt' }, + { role: 'user' as const, content: 'do the thing' }, +]; + +describe('claude-cli parser: snapshot/delta dedup + native tool ids + full results', () => { + it('streamed deltas + completed snapshot → NO duplicated text', async () => { + const bin = fakeClaude([ + { type: 'system', subtype: 'init', session_id: 's_f2' }, + { type: 'stream_event', event: { type: 'content_block_delta', delta: { type: 'text_delta', text: 'de' } } }, + { type: 'stream_event', event: { type: 'content_block_delta', delta: { type: 'text_delta', text: 'velop' } } }, + { type: 'assistant', message: { content: [{ type: 'text', text: 'develop' }] } }, // the snapshot — must NOT re-emit + { type: 'result', subtype: 'success', result: 'develop', session_id: 's_f2' }, + ]); + const deltas: string[] = []; + const p = claudeCliProvider({ kind: 'claude-cli', baseUrl: 'x', apiKey: 'k', model: 'cc', binPath: bin }); + const r = await p.complete(MSGS, [], { remainingMs: 10_000, onDelta: (d) => { deltas.push(d); } }); + expect(deltas).toEqual(['de', 'velop']); // exactly the stream — no 'develop' duplicate + expect(deltas.join('')).toBe('develop'); // byte-exact + expect(r.message.content).toBe('develop'); + }); + + it('a snapshot that EXTENDS a cut stream completes the block with the missing tail only', async () => { + const bin = fakeClaude([ + { type: 'system', subtype: 'init', session_id: 's_tail' }, + { type: 'stream_event', event: { type: 'content_block_delta', delta: { type: 'text_delta', text: 'de' } } }, // stream cut mid-block + { type: 'assistant', message: { content: [{ type: 'text', text: 'develop' }] } }, + { type: 'result', subtype: 'success', result: 'develop', session_id: 's_tail' }, + ]); + const deltas: string[] = []; + const p = claudeCliProvider({ kind: 'claude-cli', baseUrl: 'x', apiKey: 'k', model: 'cc', binPath: bin }); + await p.complete(MSGS, [], { remainingMs: 10_000, onDelta: (d) => { deltas.push(d); } }); + expect(deltas.join('')).toBe('develop'); // completed… + expect(deltas).toEqual(['de', 'velop']); // …without re-emitting 'de' + }); + + it('snapshot-only mode (no stream events) still emits the text exactly once', async () => { + const bin = fakeClaude([ + { type: 'system', subtype: 'init', session_id: 's_snap' }, + { type: 'assistant', message: { content: [{ type: 'text', text: 'Hello' }] } }, + { type: 'result', subtype: 'success', result: 'Hello', session_id: 's_snap' }, + ]); + const deltas: string[] = []; + const p = claudeCliProvider({ kind: 'claude-cli', baseUrl: 'x', apiKey: 'k', model: 'cc', binPath: bin }); + await p.complete(MSGS, [], { remainingMs: 10_000, onDelta: (d) => { deltas.push(d); } }); + expect(deltas).toEqual(['Hello']); + }); + + it('native tool_use ids are preserved into onTool invocationId (call AND result)', async () => { + const bin = fakeClaude([ + { type: 'system', subtype: 'init', session_id: 's_tool' }, + { type: 'assistant', message: { content: [{ type: 'tool_use', id: 'toolu_NATIVE_42', name: 'screens_registry', input: { command: 'list active screens' } }] } }, + { type: 'user', message: { content: [{ type: 'tool_result', tool_use_id: 'toolu_NATIVE_42', content: 'ok' }] } }, + { type: 'result', subtype: 'success', result: 'used the tool', session_id: 's_tool' }, + ]); + const tools: ToolEv[] = []; + const p = claudeCliProvider({ kind: 'claude-cli', baseUrl: 'x', apiKey: 'k', model: 'cc', binPath: bin }); + await p.complete(MSGS, [], { remainingMs: 10_000, onTool: (e) => { tools.push(e); } }); + const call = tools.find((t) => t.phase === 'call'); + expect(call?.name).toBe('screens_registry'); + expect(call?.invocationId).toBe('toolu_NATIVE_42'); // PREVIOUSLY dropped + const result = tools.find((t) => t.phase === 'result'); + expect(result?.invocationId).toBe('toolu_NATIVE_42'); // correlated via native id + expect(result?.isError).toBe(false); + }); + + it('empty results are recorded (result:"", isError:false) — and long results are NOT truncated at 160 chars', async () => { + const long = 'X'.repeat(300); + const bin = fakeClaude([ + { type: 'system', subtype: 'init', session_id: 's_res' }, + { type: 'assistant', message: { content: [{ type: 'tool_use', id: 'toolu_NATIVE_42', name: 'screens_registry', input: { command: 'probe' } }] } }, + { type: 'user', message: { content: [{ type: 'tool_result', tool_use_id: 'toolu_NATIVE_42', content: '' }] } }, // empty — was OMITTED + { type: 'user', message: { content: [{ type: 'tool_result', content: long }] } }, // order-correlated, 300 chars + { type: 'result', subtype: 'success', result: 'probed', session_id: 's_res' }, + ]); + const tools: ToolEv[] = []; + const p = claudeCliProvider({ kind: 'claude-cli', baseUrl: 'x', apiKey: 'k', model: 'cc', binPath: bin }); + await p.complete(MSGS, [], { remainingMs: 10_000, onTool: (e) => { tools.push(e); } }); + const results = tools.filter((t) => t.phase === 'result'); + expect(results).toHaveLength(2); + const empty = results.find((t) => t.result === ''); + expect(empty).toMatchObject({ isError: false, invocationId: 'toolu_NATIVE_42' }); // PREVIOUSLY dropped entirely + const full = results.find((t) => t.result === long); + expect(full).toBeDefined(); // PREVIOUSLY clipped to 160 chars + }); +}); diff --git a/test/conversation-admission.test.ts b/test/conversation-admission.test.ts new file mode 100644 index 0000000..bfe2fdf --- /dev/null +++ b/test/conversation-admission.test.ts @@ -0,0 +1,123 @@ +import { describe, expect, it } from 'vitest'; +import Fastify from 'fastify'; +import type { Provider, ProviderMessage, ProviderResponse, ProviderTool } from '../src/provider/types.ts'; +import { SqliteStorage } from '../src/memory/sqlite.ts'; +import { ToolRegistry } from '../src/tools.ts'; +import { registerConversationApi } from '../src/transport/conversation.ts'; + +/** chat-reliability F1: admission check-and-acquire must be atomic across the + * async preparation window (durable writes, history reads, stash). Two + * same-session sends racing through that window: exactly ONE accepted turn, + * ONE 409, no double provider execution, no phantom user message. A failed + * preparation must release the reservation (no stranded session). */ + +type Opts = Parameters[2]; + +class SlowProvider implements Provider { + readonly model = 'slow'; + calls = 0; + constructor(private readonly ms: number) {} + async complete(_m: ProviderMessage[], _t: ProviderTool[], _o: Opts): Promise { + this.calls += 1; + await new Promise((r) => setTimeout(r, this.ms)); + return { message: { role: 'assistant', content: 'done' }, finishReason: 'stop', usage: { promptTokens: 5, completionTokens: 2 } }; + } +} + +/** fails sessionMessages ONCE (simulates a preparation-window storage error), + * then delegates normally. */ +class FlakyPrepStorage extends SqliteStorage { + private failed = false; + constructor() { super(':memory:'); } + override async sessionMessages(id: string) { + if (!this.failed) { this.failed = true; throw new Error('transient history read failure'); } + return super.sessionMessages(id); + } +} + +async function boot(provider: Provider, storage: SqliteStorage) { + const app = Fastify(); + const registry = new ToolRegistry().register({ + name: 'noop', description: 'n', permission: 'read', parameters: { type: 'object', properties: {} }, + execute: async () => 'observation', + }); + await registerConversationApi(app, { + storage, provider, registry, policy: { name: 'draft', allows: 'draft' }, + systemPrompt: 'sys', emit: () => undefined, + }); + await app.listen({ port: 0, host: '127.0.0.1' }); + return { app, baseUrl: `http://127.0.0.1:${(app.server.address() as { port: number }).port}` }; +} + +const post = (b: string, sid: string, content: string) => + fetch(`${b}/api/agent/sessions/${sid}/messages`, { + method: 'POST', headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ content }), + }); + +describe('F1 — atomic same-session admission across the async prep window', () => { + it('two concurrent sends: one 200, one 409, provider runs exactly once', async () => { + const storage = new SqliteStorage(':memory:'); + const provider = new SlowProvider(120); + const { app, baseUrl } = await boot(provider, storage); + try { + const created = await (await fetch(`${baseUrl}/api/agent/sessions`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' })).json() as { session: { id: string } }; + const sid = created.session.id; + + const [a, b] = await Promise.all([post(baseUrl, sid, 'first'), post(baseUrl, sid, 'second')]); + const statuses = [a.status, b.status].sort(); + expect(statuses).toEqual([200, 409]); + + // the loser executed nothing and left no phantom user row + expect(provider.calls).toBe(1); + const messages = await storage.sessionMessages(sid); + expect(messages.map((m) => m.role)).toEqual(['user', 'assistant']); + const loser = a.status === 409 ? a : b; + expect(await loser.json()).toEqual({ error: 'turn already in flight on this session' }); + } finally { + await app.close(); + storage.close(); + } + }); + + it('the 409 does not poison the session: a follow-up send after completion succeeds', async () => { + const storage = new SqliteStorage(':memory:'); + const provider = new SlowProvider(40); + const { app, baseUrl } = await boot(provider, storage); + try { + const created = await (await fetch(`${baseUrl}/api/agent/sessions`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' })).json() as { session: { id: string } }; + const sid = created.session.id; + + const [a, b] = await Promise.all([post(baseUrl, sid, 'first'), post(baseUrl, sid, 'second')]); + expect([a.status, b.status].sort()).toEqual([200, 409]); + const followUp = await post(baseUrl, sid, 'third'); + expect(followUp.status).toBe(200); + expect(provider.calls).toBe(2); + } finally { + await app.close(); + storage.close(); + } + }); + + it('a preparation failure releases the reservation instead of stranding the session', async () => { + const storage = new FlakyPrepStorage(); + const provider = new SlowProvider(10); + const { app, baseUrl } = await boot(provider, storage); + try { + const created = await (await fetch(`${baseUrl}/api/agent/sessions`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' })).json() as { session: { id: string } }; + const sid = created.session.id; + + const first = await post(baseUrl, sid, 'first'); + expect(first.status).toBe(500); + expect(((await first.json()) as { error: string }).error).toContain('turn preparation failed'); + + // reservation was released — the next send must be admitted, not 409 + const second = await post(baseUrl, sid, 'retry'); + expect(second.status).toBe(200); + expect(provider.calls).toBe(1); + } finally { + await app.close(); + storage.close(); + } + }); +}); diff --git a/test/conversation-authority.test.ts b/test/conversation-authority.test.ts new file mode 100644 index 0000000..7837d80 --- /dev/null +++ b/test/conversation-authority.test.ts @@ -0,0 +1,209 @@ +import { describe, expect, it, afterAll } from 'vitest'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import Fastify from 'fastify'; +import Database from 'better-sqlite3'; +import type { Provider, ProviderMessage, ProviderResponse, ProviderTool } from '../src/provider/types.ts'; +import { ProviderError } from '../src/provider/types.ts'; +import { SqliteStorage } from '../src/memory/sqlite.ts'; +import { EventSink } from '../src/orchestrate/event-sink.ts'; +import { ToolRegistry } from '../src/tools.ts'; +import { registerConversationApi } from '../src/transport/conversation.ts'; +import { createSqliteExecutionAuthority, ExecutionAuthorityBusyError, type ExecutionAuthority } from '../src/orchestrate/execution-authority.ts'; + +/** chat-reliability F1/F2/F3 end-to-end at the transport seam: the message + * POST driven by the execution authority over real HTTP — atomic admission, + * preserved provider classification in the envelope, authority-fenced Stop, + * honest budget pause, single-executor election across transport instances. */ + +type Opts = Parameters[2]; +const dirs: string[] = []; +afterAll(() => { for (const d of dirs) rmSync(d, { recursive: true, force: true }); }); + +class SlowProvider implements Provider { + readonly model = 'slow'; + calls = 0; + constructor(private readonly ms: number, private readonly failWith?: () => never) {} + async complete(_m: ProviderMessage[], _t: ProviderTool[], _o: Opts): Promise { + this.calls += 1; + if (this.failWith) this.failWith(); + await new Promise((r) => setTimeout(r, this.ms)); + return { message: { role: 'assistant', content: 'done' }, finishReason: 'stop', usage: { promptTokens: 5, completionTokens: 2 } }; + } +} + +interface Rig { + dir: string; agent: string; storage: SqliteStorage; sink: EventSink; auth: ExecutionAuthority; + baseUrl: (app: Fastify.FastifyInstance) => string; +} + +describe('authority-driven transport — drainable host tails', () => { + it('an afterTurn promise is a tracked worker: the POST returns first, drain awaits the tail', async () => { + let tailDone = false; + const provider = new SlowProvider(10); + const dir = mkdtempSync(join(tmpdir(), 'butler-conv-auth-tail-')); + dirs.push(dir); + const agent = join(dir, 'agent.db'); + const storage = new SqliteStorage(agent); + const sink = new EventSink(storage['db'] as InstanceType, { passive: true, enqueueRunTurn: false }); + const auth = await createSqliteExecutionAuthority({ dbPath: agent, sink, storage }); + const app = Fastify(); + const registry = new ToolRegistry().register({ + name: 'nop', description: 'n', permission: 'read', parameters: { type: 'object', properties: {} }, + execute: async () => 'observation', + }); + await registerConversationApi(app, { + storage, provider, registry, policy: { name: 'draft', allows: 'draft' }, + systemPrompt: 'sys', emit: () => undefined, execution: auth, + afterTurn: async () => { await new Promise((res) => setTimeout(res, 120)); tailDone = true; }, + }); + await app.listen({ port: 0, host: '127.0.0.1' }); + const url = `http://127.0.0.1:${(app.server.address() as { port: number }).port}`; + try { + const sid = (await (await fetch(`${url}/api/agent/sessions`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' })).json() as { session: { id: string } }).session.id; + const res = await fetch(`${url}/api/agent/sessions/${sid}/messages`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ content: 'work then tail' }) }); + expect(res.status).toBe(200); + expect(((await res.json()) as { turn: { ok: boolean } }).turn.ok).toBe(true); + expect(tailDone).toBe(false); // the envelope did NOT wait for the tail + await auth.drain(); // normal shutdown drains the tail before releasing the lock + expect(tailDone).toBe(true); + } finally { + await app.close(); + await auth.drain(); + storage.close(); + } + }); +}); + +async function bootRig(name: string, provider: Provider): Promise { + const dir = mkdtempSync(join(tmpdir(), `butler-conv-auth-${name}-`)); + dirs.push(dir); + const agent = join(dir, 'agent.db'); + const storage = new SqliteStorage(agent); + const sink = new EventSink(storage['db'] as InstanceType, { passive: true, enqueueRunTurn: false }); + const auth = await createSqliteExecutionAuthority({ dbPath: agent, sink, storage }); + const app = Fastify(); + const registry = new ToolRegistry().register({ + name: 'nop', description: 'n', permission: 'read', parameters: { type: 'object', properties: {} }, + execute: async () => 'observation', + }); + await registerConversationApi(app, { + storage, provider, registry, policy: { name: 'draft', allows: 'draft' }, + systemPrompt: 'sys', emit: () => undefined, execution: auth, + }); + await app.listen({ port: 0, host: '127.0.0.1' }); + const url = `http://127.0.0.1:${(app.server.address() as { port: number }).port}`; + return { dir, agent, storage, sink, auth, app, url, baseUrl: () => url }; +} + +const post = (url: string, sid: string, body: Record) => + fetch(`${url}/api/agent/sessions/${sid}/messages`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify(body) }); + +const newSession = async (url: string): Promise => { + const r = await fetch(`${url}/api/agent/sessions`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' }); + return ((await r.json()) as { session: { id: string } }).session.id; +}; + +describe('authority-driven transport', () => { + it('concurrent same-session sends: one 200, one 409 — provider and hooks run exactly once', async () => { + let hooks = 0; + const provider = new SlowProvider(120); + const rig = await bootRig('race', provider); + try { + const sid = await newSession(rig.url); + const [a, b] = await Promise.all([post(rig.url, sid, { content: 'first' }), post(rig.url, sid, { content: 'second' })]); + expect([a.status, b.status].sort()).toEqual([200, 409]); + expect(provider.calls).toBe(1); + expect(hooks).toBe(0); + const messages = await rig.storage.sessionMessages(sid); + expect(messages.filter((m) => m.role === 'user')).toHaveLength(1); // no phantom user row + // follow-up send works — ownership was released + const again = await post(rig.url, sid, { content: 'third' }); + expect(again.status).toBe(200); + expect(provider.calls).toBe(2); + } finally { + await rig.app.close(); + await rig.auth.drain(); + rig.storage.close(); + } + }); + + it('F3 consumer wiring: provider 400 surfaces PROVIDER_BAD_REQUEST — no budget stash, session errors honestly', async () => { + const provider = new SlowProvider(1, () => { throw new ProviderError('bad request', 'PROVIDER_BAD_REQUEST', 400); }); + const rig = await bootRig('p400', provider); + try { + const sid = await newSession(rig.url); + const res = await post(rig.url, sid, { content: 'hello' }); + expect(res.status).toBe(200); + const body = (await res.json()) as { turn: { ok: boolean; code?: string; error?: string } }; + expect(body.turn.ok).toBe(false); + expect(body.turn.code).toBe('PROVIDER_BAD_REQUEST'); + expect(body.turn.error).toContain('PROVIDER_BAD_REQUEST'); + expect(await rig.storage.peekPartial?.(sid)).toBeNull(); // no budget stash for a provider failure + const sess = await rig.storage.getSession(sid); + expect(sess?.status).toBe('error'); + expect(rig.auth.readSession(sid)).toMatchObject({ state: 'idle', admission: 'ready' }); // ownership released + } finally { + await rig.app.close(); + await rig.auth.drain(); + rig.storage.close(); + } + }); + + it('duplicate clientRequestId returns REQUEST_ALREADY_ADMITTED with the original turn, without re-executing', async () => { + const provider = new SlowProvider(10); + const rig = await bootRig('reqid', provider); + try { + const sid = await newSession(rig.url); + const rid = '22222222-2224-4222-8222-222222222222'; + const first = await post(rig.url, sid, { content: 'same prompt', clientRequestId: rid }); + expect(first.status).toBe(200); + expect(((await first.json()) as { turn: { ok: boolean } }).turn.ok).toBe(true); + const dup = await post(rig.url, sid, { content: 'same prompt', clientRequestId: rid }); + expect(dup.status).toBe(409); + const dupBody = (await dup.json()) as { code?: string; turnId?: string }; + expect(dupBody.code).toBe('REQUEST_ALREADY_ADMITTED'); + expect(typeof dupBody.turnId).toBe('string'); + expect(provider.calls).toBe(1); // never re-executed + } finally { + await rig.app.close(); + await rig.auth.drain(); + rig.storage.close(); + } + }); + + it('explicit Stop mid-turn → ABORTED envelope; the session returns to ready', async () => { + const provider = new SlowProvider(10_000); + const rig = await bootRig('stop', provider); + try { + const sid = await newSession(rig.url); + const inflight = post(rig.url, sid, { content: 'long work' }); + await new Promise((r) => setTimeout(r, 150)); // let it admit + begin + const stop = await fetch(`${rig.url}/api/agent/sessions/${sid}/abort`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' }); + expect(stop.status).toBe(200); + const res = await inflight; + const body = (await res.json()) as { turn: { ok: boolean; code?: string } }; + expect(body.turn.ok).toBe(false); + expect(body.turn.code).toBe('ABORTED'); + expect(rig.auth.readSession(sid)).toMatchObject({ state: 'idle', admission: 'ready' }); + } finally { + await rig.app.close(); + await rig.auth.drain(); + rig.storage.close(); + } + }); + + it('a second transport instance on the same agent DB cannot elect its own authority', async () => { + const rig = await bootRig('elect', new SlowProvider(1)); + try { + const sink2 = new EventSink(new Database(rig.agent), { passive: true, enqueueRunTurn: false }); + await expect(createSqliteExecutionAuthority({ dbPath: rig.agent, sink: sink2, storage: rig.storage })) + .rejects.toBeInstanceOf(ExecutionAuthorityBusyError); + } finally { + await rig.app.close(); + await rig.auth.drain(); + rig.storage.close(); + } + }); +}); diff --git a/test/execution-authority.test.ts b/test/execution-authority.test.ts new file mode 100644 index 0000000..8f3f0e6 --- /dev/null +++ b/test/execution-authority.test.ts @@ -0,0 +1,296 @@ +import { describe, expect, it, afterAll } from 'vitest'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import Database from 'better-sqlite3'; +import { SqliteStorage } from '../src/memory/sqlite.ts'; +import { EventSink } from '../src/orchestrate/event-sink.ts'; +import { createSqliteExecutionAuthority, ExecutionAuthority, ExecutionAuthorityBusyError, type ExecutionAttempt } from '../src/orchestrate/execution-authority.ts'; + +/** chat-reliability F1/F2: the execution authority — election, atomic durable + * admission, stash CAS boundary, truthful dead-owner recovery, worker + * quarantine — exercised on real disk SQLite (two files: storage + agent). */ + +const dirs: string[] = []; +afterAll(() => { for (const d of dirs) rmSync(d, { recursive: true, force: true }); }); + +function rig(name: string): { dir: string; main: string; agent: string; storage: SqliteStorage; sink: EventSink } { + const dir = mkdtempSync(join(tmpdir(), `butler-auth-${name}-`)); + dirs.push(dir); + // production shape: storage (sessions/messages/stash) and the event sink + // share the SAME agent DB file; the authority opens a second connection to it + const main = join(dir, 'main.db'); + const agent = join(dir, 'agent.db'); + const storage = new SqliteStorage(agent); + const sink = new EventSink(storage['db'] as InstanceType, { passive: true, enqueueRunTurn: false }); + return { dir, main, agent, storage, sink }; +} + +async function elect(r: ReturnType): Promise { + return createSqliteExecutionAuthority({ dbPath: r.agent, sink: r.sink, storage: r.storage }); +} + +async function ensureSession(r: ReturnType, sessionId: string): Promise { + if (!(await r.storage.getSession(sessionId))) await r.storage.createSession({ id: sessionId }); +} + +async function admittedTurn(r: ReturnType, auth: ExecutionAuthority, sessionId: string, content = 'do the thing', requestId?: string): Promise { + await ensureSession(r, sessionId); + const res = auth.tryReserve(sessionId); + expect(res.status).toBe('reserved'); + const attempt = (res as { attempt: ExecutionAttempt }).attempt; + const adm = auth.admit(attempt, { content, principalId: 'u1', ...(requestId ? { requestId } : {}) }); + expect(adm.status).toBe('admitted'); + return attempt; +} + +describe('F2 — election: one authority per agent DB', () => { + it('a second executor fails election with SQLITE_BUSY, not silent co-ownership', async () => { + const r = rig('elect'); + const a = await elect(r); + await expect(elect(r)).rejects.toBeInstanceOf(ExecutionAuthorityBusyError); + await a.drain(); + // after a clean release the next executor CAN take over + const b = await elect(r); + expect(b.currentPhase()).toBe('ready'); + await b.drain(); + }); + + it('a passive sink constructor never fails stale live work (no sweep)', async () => { + const r = rig('passive'); + await ensureSession(r, 'sess_legacy'); + // fabricate a legacy active fold directly + r.sink.handle('sess_legacy', { kind: 'turn.request', sessionId: 'sess_legacy', turnId: 't_legacy', prompt: 'p', refIds: [], at: Date.now() } as never, 'chat:sess_legacy:t_legacy'); + const before = r.sink.events('sess_legacy').length; + const second = new EventSink(new Database(r.agent), { passive: true, enqueueRunTurn: false }); + expect(second.events('sess_legacy').length).toBe(before); // nothing failed + expect(second.readSnapshot('sess_legacy').state.activeTurnId).toBe('t_legacy'); + }); +}); + +describe('F1 — atomic durable admission', () => { + it('admit lands events + ledger row in one commit; the session is busy; other sessions are free', async () => { + const r = rig('admit'); + const auth = await elect(r); + const attempt = await admittedTurn(r, auth, 'sess_a'); + + const fold = r.sink.readSnapshot('sess_a').state; + expect(fold.activeTurnId).toBe(attempt.turnId); + const row = new Database(r.agent, { readonly: true }).prepare('SELECT phase, epoch, principal_id FROM butler_executions WHERE turn_id = ?').get(attempt.turnId) as { phase: string; epoch: string; principal_id: string }; + expect(row.phase).toBe('preparing'); + expect(row.epoch).toBe(auth.epoch); + expect(row.principal_id).toBe('u1'); + expect(auth.tryReserve('sess_a').status).toBe('busy'); + expect(auth.tryReserve('sess_b').status).toBe('reserved'); + await auth.drain(); + }); + + it('duplicate requestId: same content → REQUEST_ALREADY_ADMITTED with the original turn; different content → REQUEST_ID_REUSE', async () => { + const r = rig('reqid'); + const auth = await elect(r); + const first = await admittedTurn(r, auth, 'sess_a', 'same prompt', '11111111-1114-4111-8111-111111111111'); + await auth.settle(first, 'COMPLETED'); + + const again = auth.tryReserve('sess_a'); + expect(again.status).toBe('reserved'); + const dup = auth.admit((again as { attempt: ExecutionAttempt }).attempt, { content: 'same prompt', principalId: 'u1', requestId: '11111111-1114-4111-8111-111111111111' }); + expect(dup).toMatchObject({ status: 'rejected', code: 'REQUEST_ALREADY_ADMITTED' }); + expect((dup as { turnId?: string }).turnId).toBe(first.turnId); + + const other = auth.tryReserve('sess_a'); + const reuse = auth.admit((other as { attempt: ExecutionAttempt }).attempt, { content: 'DIFFERENT prompt', principalId: 'u1', requestId: '11111111-1114-4111-8111-111111111111' }); + expect(reuse).toMatchObject({ status: 'rejected', code: 'REQUEST_ID_REUSE' }); + await auth.drain(); + }); + + it('a queue decision rolls back entirely: no execution row, no receipt, fold unchanged', async () => { + const r = rig('queue'); + const auth = await elect(r); + await ensureSession(r, 'sess_q'); + // legacy active fold with NO execution row + r.sink.handle('sess_q', { kind: 'turn.request', sessionId: 'sess_q', turnId: 't_old', prompt: 'p', refIds: [], at: Date.now() } as never, 'chat:sess_q:t_old'); + const res = auth.tryReserve('sess_q'); + expect(res.status).toBe('reserved'); + const adm = auth.admit((res as { attempt: ExecutionAttempt }).attempt, { content: 'new', principalId: 'u1' }); + expect(adm).toMatchObject({ status: 'rejected', code: 'QUEUE_DECISION' }); + const check = new Database(r.agent, { readonly: true }); + expect((check.prepare('SELECT COUNT(*) AS n FROM butler_executions').get() as { n: number }).n).toBe(0); + expect(r.sink.readSnapshot('sess_q').state.activeTurnId).toBe('t_old'); // untouched + await auth.drain(); + }); +}); + +describe('F1 — execution boundary: stash peek + CAS consume', () => { + it('beginExecution snapshots then CAS-consumes the exact stash; Stop-before-start leaves it intact', async () => { + const r = rig('stash'); + const auth = await elect(r); + await r.storage.stashPartial('sess_s', 't_prev', JSON.stringify([{ role: 'tool', content: 'earlier work' }])); + + const attempt = await admittedTurn(r, auth, 'sess_s'); + // preparation failed/never ran → stash still there + expect((await r.storage.peekPartial?.('sess_s'))?.turnId).toBe('t_prev'); + + const begun = await auth.beginExecution(attempt); + expect(beguun(begun)).toBe(true); + const { seeded } = begun as { seeded: unknown[] | null }; + expect(seeded).toEqual([{ role: 'tool', content: 'earlier work' }]); + expect(await r.storage.peekPartial?.('sess_s')).toBeNull(); // consumed exactly once + const row = new Database(r.agent, { readonly: true }).prepare('SELECT phase, resume_turn_id, resume_json FROM butler_executions WHERE turn_id = ?').get(attempt.turnId) as { phase: string; resume_turn_id: string; resume_json: string }; + expect(row.phase).toBe('executing'); + expect(row.resume_turn_id).toBe('t_prev'); + expect(JSON.parse(row.resume_json)).toEqual([{ role: 'tool', content: 'earlier work' }]); + await auth.drain(); + }); + + it('Stop requested during preparation aborts before start without touching the stash', async () => { + const r = rig('stop'); + const auth = await elect(r); + await r.storage.stashPartial('sess_st', 't_prev', '[{"role":"user","content":"x"}]'); + const attempt = await admittedTurn(r, auth, 'sess_st'); + const stop = auth.requestStop('sess_st'); + expect(stop).toMatchObject({ status: 'aborting', turnId: attempt.turnId }); + const begun = await auth.beginExecution(attempt); + expect(begun).toMatchObject({ status: 'rejected', code: 'STOP_BEFORE_START' }); + expect((await r.storage.peekPartial?.('sess_st'))?.turnId).toBe('t_prev'); + expect(attempt.controller.signal.aborted).toBe(true); + await auth.drain(); + }); +}); + +function beguun(x: unknown): boolean { + return typeof x === 'object' && x !== null && 'seeded' in x; +} + +describe('F2 — truthful recovery and quarantine', () => { + it('a died executor\'s unsettled turn gets ONE interruption terminal; re-recovery is a no-op; the session returns to ready', async () => { + const r = rig('recover'); + const a = await elect(r); + const attempt = await admittedTurn(r, a, 'sess_r'); + await a.drain(); // owner released WITHOUT settling (simulated death) + + const b = await elect(r); + expect(b.currentPhase()).toBe('ready'); + const evs = r.sink.events('sess_r'); + const failed = evs.filter((e) => e.kind === 'turn.failed'); + expect(failed).toHaveLength(1); + expect((failed[0]!.detail as { error: string }).error).toContain('interrupted'); + expect(r.sink.readSnapshot('sess_r').state.activeTurnId).toBeNull(); // terminal + const row = new Database(r.agent, { readonly: true }).prepare('SELECT phase, terminal_code FROM butler_executions WHERE turn_id = ?').get(attempt.turnId) as { phase: string; terminal_code: string }; + expect(row.phase).toBe('settled'); + expect(row.terminal_code).toBe('EXECUTION_INTERRUPTED'); + expect(b.readSession('sess_r')).toMatchObject({ state: 'idle', admission: 'ready' }); + + await b.drain(); + const c = await elect(r); + expect(r.sink.events('sess_r').filter((e) => e.kind === 'turn.failed')).toHaveLength(1); // still exactly one + expect(c.readSession('sess_r')).toMatchObject({ state: 'idle', admission: 'ready' }); + await c.drain(); + }); + + it('unproven previous-epoch workers quarantine ALL fresh execution until operator resolution', async () => { + const r = rig('quarantine'); + const a = await elect(r); + // simulate a worker the dead executor never settled: insert directly + const raw = new Database(r.agent); + raw.prepare("INSERT INTO butler_execution_workers(worker_id,epoch,turn_id,kind,phase,pid,identity,created_at) VALUES ('w_orphan','" + a.epoch + "',NULL,'cli-child','running'," + process.pid + ",'{}',1)").run(); + raw.close(); + await a.drain(); + + const b = await elect(r); + expect(b.currentPhase()).toBe('quarantined'); + expect(b.tryReserve('sess_any')).toMatchObject({ status: 'unavailable', reason: 'quarantined' }); + expect(b.readSession('sess_any')).toMatchObject({ state: 'uncertain', reason: 'worker_quarantine', admission: 'blocked' }); + + // wrong epoch evidence is refused + expect(b.resolveUncertainWorker('w_orphan', { expectedEpoch: 'not-the-epoch', operatorIdentity: 'op', evidenceRef: 'ps: verified' })).toBe(false); + expect(b.resolveUncertainWorker('w_orphan', { expectedEpoch: a.epoch, operatorIdentity: 'op', evidenceRef: 'ps: tree verified dead' })).toBe(true); + expect(b.currentPhase()).toBe('ready'); + expect(b.tryReserve('sess_any').status).toBe('reserved'); + await b.drain(); + }); +}); + +describe('F2 — reads and lifecycle', () => { + it('readSession walks idle → preparing → running → stopping → idle through a real turn', async () => { + const r = rig('reads'); + const auth = await elect(r); + expect(auth.readSession('sess_rw')).toMatchObject({ state: 'idle', admission: 'ready' }); + const attempt = await admittedTurn(r, auth, 'sess_rw'); + expect(auth.readSession('sess_rw')).toMatchObject({ state: 'preparing', turnId: attempt.turnId, admission: 'busy' }); + await auth.beginExecution(attempt); + expect(auth.readSession('sess_rw')).toMatchObject({ state: 'running', turnId: attempt.turnId }); + auth.requestStop('sess_rw'); + expect(auth.readSession('sess_rw')).toMatchObject({ state: 'stopping', turnId: attempt.turnId }); + // close the turn the way the owning transport does (terminal event first), + // then settle the ledger — reads must converge to idle + auth.handleOwned(attempt, 'sess_rw', { kind: 'turn.fail', sessionId: 'sess_rw', turnId: attempt.turnId, error: 'aborted by caller' } as never, `chat:${attempt.turnId}:fail`); + await auth.settle(attempt, 'ABORTED'); + expect(auth.readSession('sess_rw')).toMatchObject({ state: 'idle', admission: 'ready' }); + await auth.drain(); + }); + + it('requestStop with a stale expectedTurnId is rejected; unknown sessions are no_active', async () => { + const r = rig('stop2'); + const auth = await elect(r); + expect(auth.requestStop('sess_none')).toMatchObject({ status: 'no_active' }); + const attempt = await admittedTurn(r, auth, 'sess_st2'); + expect(auth.requestStop('sess_st2', 't_someone_else')).toMatchObject({ status: 'stale' }); + expect(auth.requestStop('sess_st2', attempt.turnId)).toMatchObject({ status: 'aborting', turnId: attempt.turnId }); + await auth.drain(); + }); + + it('finalizeOwned commits terminal + accounting + ledger settle atomically; duplicates are no-ops', async () => { + const r = rig('finalize'); + const auth = await elect(r); + const attempt = await admittedTurn(r, auth, 'sess_fin'); + await auth.beginExecution(attempt); + const fin = await auth.finalizeOwned(attempt, { + kind: 'turn.complete', answer: 'all done', commandKey: `chat:${attempt.turnId}:done`, settleCode: 'COMPLETED', + accounting: { status: 'active', tokensInDelta: 7, tokensOutDelta: 3 }, + }); + expect(fin).toMatchObject({ status: 'finalized', accounted: true }); + // one terminal event; fold terminal; accounting applied; ledger settled + expect(r.sink.events('sess_fin').filter((e) => e.kind === 'turn.completed')).toHaveLength(1); + expect(r.sink.readSnapshot('sess_fin').state.activeTurnId).toBeNull(); + const sess = await r.storage.getSession('sess_fin'); + expect(sess?.tokens_in).toBe(7); + expect(sess?.tokens_out).toBe(3); + expect(auth.readSession('sess_fin')).toMatchObject({ state: 'idle', admission: 'ready' }); + // duplicate finalize (already released) is stale — never a second terminal + const dup = await auth.finalizeOwned(attempt, { kind: 'turn.fail', error: 'x', commandKey: `chat:${attempt.turnId}:done`, settleCode: 'X' }); + expect(dup.status).toBe('stale'); + expect(r.sink.events('sess_fin').filter((e) => e.kind === 'turn.failed')).toHaveLength(0); + await auth.drain(); + }); + + it('finalizeOwned budget failure stashes the continuation INSIDE the same commit', async () => { + const r = rig('finbudget'); + const auth = await elect(r); + const attempt = await admittedTurn(r, auth, 'sess_fb'); + await auth.beginExecution(attempt); + await auth.finalizeOwned(attempt, { + kind: 'turn.fail', error: 'ran out of runway', commandKey: `chat:${attempt.turnId}:fail`, settleCode: 'BUDGET_EXCEEDED', + stash: { json: '[{"role":"tool","content":"partial"}]' }, + accounting: { status: 'active' }, + }); + const stash = await r.storage.peekPartial?.('sess_fb'); + expect(stash?.turnId).toBe(attempt.turnId); // stash + terminal + settle committed together + expect(r.sink.events('sess_fb').filter((e) => e.kind === 'turn.failed')).toHaveLength(1); + await auth.drain(); + }); + + it('drain awaits tracked workers before releasing the lock', async () => { + const r = rig('drain'); + const auth = await elect(r); + let finished = false; + const worker = auth.trackWorker({ identity: { probe: true } }, 'tail', async () => { + await new Promise((resolve) => setTimeout(resolve, 80)); + finished = true; + }); + const drained = auth.drain(); + await worker; + expect(finished).toBe(true); + await drained; + expect(auth.currentPhase()).toBe('released'); + }); +}); diff --git a/test/loop-classification.test.ts b/test/loop-classification.test.ts new file mode 100644 index 0000000..0724be0 --- /dev/null +++ b/test/loop-classification.test.ts @@ -0,0 +1,151 @@ +import { describe, expect, it } from 'vitest'; +import { MockProvider, type ScriptedTurn } from '../src/provider/mock.ts'; +import { ProviderError, type Provider, type ProviderMessage, type ProviderResponse, type ProviderTool } from '../src/provider/types.ts'; +import { runLoop, LoopError } from '../src/loop.ts'; + +/** chat-reliability F3: the loop classifies failures by actual cause/signal — + * a permanent provider rejection (400/auth) must NEVER surface as + * BUDGET_EXCEEDED (which the transport treats as a resumable pause). + * Matrix mirrored from the isolated classifier probe, here on the REAL loop. */ + +type Opts = Parameters[2]; + +const toolCall = (i: number): ScriptedTurn => ({ + kind: 'assistant', + content: null, + toolCalls: [{ id: `c${i}`, type: 'function', function: { name: 'noop', arguments: '{}' } }], +}); + +const noopTool = { + name: 'noop', + description: 'n', + parameters: { type: 'object', properties: {} }, + execute: async () => 'observation', +}; + +class ThrowingProvider implements Provider { + readonly model = 'throwing'; + calls = 0; + constructor(private readonly make: () => unknown) {} + async complete(_m: ProviderMessage[], _t: ProviderTool[], _o: Opts): Promise { + this.calls += 1; + throw this.make(); + } +} + +/** sleeps, then answers; optionally ignores abort to model a provider that + * resolves normally AFTER the caller aborted (post-return guard twin). */ +class SleepyProvider implements Provider { + readonly model = 'sleepy'; + calls = 0; + constructor(private readonly ms: number, private readonly ignoreAbort = false) {} + async complete(_m: ProviderMessage[], _t: ProviderTool[], o: Opts): Promise { + this.calls += 1; + await new Promise((resolve, reject) => { + const t = setTimeout(resolve, this.ms); + if (!this.ignoreAbort) o.signal?.addEventListener('abort', () => { clearTimeout(t); reject(o.signal!.reason); }, { once: true }); + }); + return { message: { role: 'assistant', content: 'ok' }, finishReason: 'stop', usage: { promptTokens: 1, completionTokens: 1 } }; + } +} + +const expectLoopError = async (fn: () => Promise) => { + try { await fn(); expect.unreachable('runLoop should have thrown'); } + catch (err) { expect(err).toBeInstanceOf(LoopError); return err as LoopError; } +}; + +describe('F3 — provider failures keep their own classification', () => { + it('permanent 400 surfaces PROVIDER_BAD_REQUEST, not BUDGET_EXCEEDED', async () => { + const p = new ThrowingProvider(() => new ProviderError('bad request', 'PROVIDER_BAD_REQUEST', 400)); + const err = await expectLoopError(() => runLoop('s', 'u', { provider: p, tools: [] })); + expect(err.code).toBe('PROVIDER_BAD_REQUEST'); + expect(err.cause).toBe('provider'); + expect(err.status).toBe(400); + expect(err.message).toContain('PROVIDER_BAD_REQUEST'); + }); + + it('PROVIDER_UNAVAILABLE passes through unchanged', async () => { + const p = new ThrowingProvider(() => new ProviderError('upstream down', 'PROVIDER_UNAVAILABLE', 503, true)); + const err = await expectLoopError(() => runLoop('s', 'u', { provider: p, tools: [] })); + expect(err.code).toBe('PROVIDER_UNAVAILABLE'); + expect(err.cause).toBe('provider'); + }); + + it('unattributed TimeoutError (wall NOT expired, caller NOT aborted) → PROVIDER_TIMEOUT', async () => { + const raw = new Error('socket stale'); + raw.name = 'TimeoutError'; + const p = new ThrowingProvider(() => raw); + const err = await expectLoopError(() => runLoop('s', 'u', { provider: p, tools: [], budget: { wallMs: 60_000 } })); + expect(err.code).toBe('PROVIDER_TIMEOUT'); + expect(err.cause).toBe('provider'); + }); + + it('unknown provider exception → PROVIDER_FAILED', async () => { + const p = new ThrowingProvider(() => new Error('socket exploded')); + const err = await expectLoopError(() => runLoop('s', 'u', { provider: p, tools: [] })); + expect(err.code).toBe('PROVIDER_FAILED'); + expect(err.cause).toBe('provider'); + }); +}); + +describe('F3 — genuine loop limits stay BUDGET_EXCEEDED with an explicit cause', () => { + it('hop exhaustion → budget-hop', async () => { + const p = new MockProvider([toolCall(0)]); + const err = await expectLoopError(() => runLoop('s', 'u', { provider: p, tools: [noopTool], budget: { maxHops: 2 } })); + expect(err.code).toBe('BUDGET_EXCEEDED'); + expect(err.cause).toBe('budget-hop'); + }); + + it('wall expiry during a hung provider call → budget-wall (loop-owned deadline)', async () => { + const p = new MockProvider([{ kind: 'hang', ms: 30_000 }]); + const err = await expectLoopError(() => runLoop('s', 'u', { provider: p, tools: [], budget: { wallMs: 50 } })); + expect(err.code).toBe('BUDGET_EXCEEDED'); + expect(err.cause).toBe('budget-wall'); + }); + + it('hard token cap → budget-token', async () => { + const p = new MockProvider([{ kind: 'assistant', content: null, toolCalls: [{ id: 'a', type: 'function', function: { name: 'noop', arguments: '{}' } }], usage: { promptTokens: 500, completionTokens: 500 } }]); + const err = await expectLoopError(() => runLoop('s', 'u', { provider: p, tools: [noopTool], budget: { maxTokens: 100 }, hardTokenCap: true })); + expect(err.code).toBe('BUDGET_EXCEEDED'); + expect(err.cause).toBe('budget-token'); + }); +}); + +describe('F3 — explicit Stop beats everything', () => { + it('caller abort before the loop starts → ABORTED/caller-abort', async () => { + const ac = new AbortController(); + ac.abort(); + const p = new SleepyProvider(1); + const err = await expectLoopError(() => runLoop('s', 'u', { provider: p, tools: [], signal: ac.signal })); + expect(err.code).toBe('ABORTED'); + expect(err.cause).toBe('caller-abort'); + expect(p.calls).toBe(0); + }); + + it('caller abort mid-provider-call → ABORTED (not a provider timeout)', async () => { + const ac = new AbortController(); + setTimeout(() => ac.abort(), 30); + const p = new SleepyProvider(10_000); + const err = await expectLoopError(() => runLoop('s', 'u', { provider: p, tools: [], signal: ac.signal, budget: { wallMs: 60_000 } })); + expect(err.code).toBe('ABORTED'); + expect(err.cause).toBe('caller-abort'); + }); + + it('provider resolving NORMALLY after caller abort is still ABORTED (post-return guard)', async () => { + const ac = new AbortController(); + setTimeout(() => ac.abort(), 20); + const p = new SleepyProvider(60, true); // ignores the signal, resolves anyway + const err = await expectLoopError(() => runLoop('s', 'u', { provider: p, tools: [], signal: ac.signal, budget: { wallMs: 60_000 } })); + expect(err.code).toBe('ABORTED'); + expect(err.cause).toBe('caller-abort'); + }); +}); + +describe('F3 — ordinary success is untouched', () => { + it('a normal provider call bypasses classification entirely', async () => { + const p = new SleepyProvider(5); + const result = await runLoop('s', 'u', { provider: p, tools: [], budget: { wallMs: 60_000 } }); + expect(result.answer).toBe('ok'); + expect(p.calls).toBe(1); + }); +}); diff --git a/test/loop.test.ts b/test/loop.test.ts index a5f0a40..313d6a6 100644 --- a/test/loop.test.ts +++ b/test/loop.test.ts @@ -141,7 +141,7 @@ describe('loop — budgets (condition 2)', () => { expect((err as LoopError).code).toBe('ABORTED'); }); - it('never-responding provider → PROVIDER_TIMEOUT inside wall budget; session resumable (review F5)', async () => { + it('never-responding provider → wall budget kills it; loop-owned budget cause, resumable (review F5 + chat-reliability F3)', async () => { const provider = new MockProvider([{ kind: 'hang', ms: 10_000 }]); const t0 = Date.now(); const err = await runLoop('sys', 'go', { @@ -151,8 +151,11 @@ describe('loop — budgets (condition 2)', () => { }).catch((e: unknown) => e); const elapsed = Date.now() - t0; expect(err).toBeInstanceOf(LoopError); + // F3: the LOOP's wall expired (observed deadline signal), so this is a + // genuine budget outcome with an explicit cause — not a provider timeout. expect((err as LoopError).code).toBe('BUDGET_EXCEEDED'); - expect((err as LoopError).message).toContain('PROVIDER_TIMEOUT'); + expect((err as LoopError).cause).toBe('budget-wall'); + expect((err as LoopError).message).toContain('wall budget exhausted'); // the wall budget held: aborted well inside it (not 10s) expect(elapsed).toBeLessThan(3_000); // partial transcript persisted = resumable shape