diff --git a/CHANGELOG.md b/CHANGELOG.md index 2dc4f45..f3df032 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,38 @@ +## [Unreleased] (detached send, server-owned queue, steering, push-on-append) + +### Added +- **Detached send (opt-in)** — `POST /api/agent/sessions/:id/messages` with `detach: true` + (requires an execution authority; else `400 DETACH_UNSUPPORTED`) answers at once: + `202 {startedTurnId}` when idle, `202 {queued: {queueId, disposition}}` when busy + (`mode: 'steer' | 'queue'`). Without `detach` the route is unchanged (held response). +- **Server-owned queue** — identity-bearing `message.queued` (`queueId`, `body {commands, model}`, + request id + principal + fingerprint), one `queue.removed {queueId, reason: claimed | deleted | + delivered}` row per removal, `queue.updated` snapshots gain parallel `ids`. + `DELETE /api/agent/sessions/:id/queue/:queueId` removes a pending item. +- **Claim-all drain** — when any admitted turn (held or detached) settles, or the authority + becomes ready at boot, every pending item runs as ONE next turn. The claim, the user row + (`content` = the items' own text, `items[]`) and `turn.started` commit in one transaction; + the composed prompt (`[N queued messages]` + items) goes only to the model. Legacy queue rows + without ids are never auto-run. A runaway guard refuses to claim an id twice in one process. +- **Steering** — `LoopDeps.takeSteering` is called at each tool boundary; the transport's + `steer.deliver` command records `queue.removed (delivered)`, `steer.delivered` and a + `user_message.recorded {via: 'steer'}` row (so later turns keep it in model history) in one commit. +- **Duplicate-submit guard** — before reservation, a detached send's `clientRequestId` is looked + up across admitted turns and every queued item (pending or gone): same content → the existing + `startedTurnId` / `queueId`, different content → `409 REQUEST_ID_REUSE`. +- **`EventSink.onAppend(listener)`** — committed rows pushed per session, only after the + outermost transaction commits; rollbacks are never pushed; listener faults are isolated. +- **`ExecutionAuthority.beginShutdown()`** — closing gate: nothing new is reserved after it, so a + turn settling during shutdown cannot start queued work. Hosts call it before aborting turns. +- **`session.notice {level, text}`** row kind; written when a detached turn's finalization keeps + failing after bounded retries (`ConversationDeps.finalizeRetryMs`, default 1s/5s/30s; the retry + stops at shutdown and leaves the attempt reserved for recovery). + +### Fixed +- Tool events without an invocation id get a per-turn counter id (`tc__n`, paired FIFO + by name) instead of the bare tool name, in the conversation transport and workflow runs. +- `TurnEventBase.seq` comment: per-session, not per-turn (the schema is `UNIQUE(session_id, seq)`). + ## [0.8.0] — 2026-10-21 (Stage 3: tree-sitter symbols + watchRoots + structured search) ### Added diff --git a/src/index.ts b/src/index.ts index 1889cd1..3254881 100644 --- a/src/index.ts +++ b/src/index.ts @@ -75,10 +75,10 @@ export { claudeCliProvider, ClaudeCliProvider } from './provider/claude-cli.ts'; 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 type { TurnCommand, TurnEvent as DurableTurnEvent, TurnEventKind as DurableTurnEventKind, OrchestratorState, QueueState, QueueEntry, QueueItemRef, QueueRemovalReason, QueuedBody, QueuedIdentity, NoticeLevel, 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 type { ExecutionAuthorityDeps, ExecutionAttempt, ExecutionRead, RecoveryResult, RequestLookup, EnqueueInput } from './orchestrate/execution-authority.ts'; +export type { AppendListener, 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 427955d..dd6971a 100644 --- a/src/loop.ts +++ b/src/loop.ts @@ -99,6 +99,12 @@ export interface LoopDeps { * replayed log can separate consecutive hops' narration instead of gluing * it. Purely additive: absent → behavior identical to before. */ onActivity?: (a: TextBlockActivity) => void; + /** steering seam: called once per hop AFTER the hop's tool results are + * appended and BEFORE the next provider call. Each returned item rides + * into the conversation as a user message. The caller owns durability + * (it records the delivery before returning). A throw is ignored — the + * items stay pending for the next turn. */ + takeSteering?: () => Array<{ queueId: string; content: string }>; signal?: AbortSignal; } @@ -267,6 +273,16 @@ export async function runLoop(systemPrompt: string, userMessage: string, deps: L 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` } }); } + // tool boundary: steers sent while this hop ran join before the next call + if (deps.takeSteering && !deps.signal?.aborted) { + let steers: Array<{ queueId: string; content: string }> = []; + try { steers = deps.takeSteering(); } catch { steers = []; } + for (const s of steers) { + const msg: ProviderMessage = { role: 'user', content: s.content }; + messages.push(msg); + deps.onMessage?.(msg, { hop, tokensIn: 0, tokensOut: 0 }); + } + } } throw new LoopError(`hop budget exhausted (${budget.maxHops})`, 'BUDGET_EXCEEDED', messages, 'budget-hop'); } diff --git a/src/orchestrate/event-sink.ts b/src/orchestrate/event-sink.ts index 6aed3e6..0a2511c 100644 --- a/src/orchestrate/event-sink.ts +++ b/src/orchestrate/event-sink.ts @@ -76,8 +76,43 @@ export type Effect = | { type: 'runTurn'; sessionId: string; turnId: string } | { type: 'emit'; event: TurnEvent }; +/** Push-on-append listener: rows of ONE session, in seq order, delivered only + * after the outermost transaction that wrote them has committed. */ +export type AppendListener = (sessionId: string, rows: TurnEvent[]) => void; + export class EventSink { private states = new Map(); + private readonly listeners = new Set(); + /** rows committed by an inner write whose OUTER transaction is still open */ + private pending: TurnEvent[] = []; + + /** Subscribe to committed appends. Process-local: rows written by another + * process (or by direct inserts outside the sink, e.g. recoverOrphans at + * construction) are never pushed — readers keep a catch-up read for those. + * A throwing listener is isolated (never affects the write or others). */ + onAppend(listener: AppendListener): () => void { + this.listeners.add(listener); + return () => { this.listeners.delete(listener); }; + } + + /** Deliver buffered rows once no transaction is open on the connection. */ + private flush(): void { + if (this.db.inTransaction || this.pending.length === 0) return; + const rows = this.pending; + this.pending = []; + if (this.listeners.size === 0) return; + const bySession = new Map(); + for (const r of rows) { + const list = bySession.get(r.session_id); + if (list) list.push(r); + else bySession.set(r.session_id, [r]); + } + for (const [sessionId, list] of bySession) { + for (const l of this.listeners) { + try { l(sessionId, list); } catch { /* listener faults never reach the writer */ } + } + } + } constructor(private readonly db: Database.Database, opts?: { enqueueRunTurn?: boolean; passive?: boolean }) { db.exec(DDL); @@ -172,7 +207,17 @@ export class EventSink { * 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(); + const mark = this.pending.length; + let out: T; + try { + out = run(); + } catch (err) { + // rolled back (outer txn or savepoint): its rows were never committed + this.pending.length = mark; + throw err; + } + this.flush(); + return out; } /** One consistent snapshot: folded state + events + lastSeq from the same @@ -232,6 +277,10 @@ export class EventSink { this.states.delete(sessionId); throw err; } + // push-on-append: buffered, flushed now when this was the outermost + // transaction, else when the enclosing withTransaction commits + this.pending.push(...appended); + this.flush(); // 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 index 018da14..e0af6aa 100644 --- a/src/orchestrate/execution-authority.ts +++ b/src/orchestrate/execution-authority.ts @@ -26,7 +26,7 @@ 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'; +import type { NoticeLevel, QueueEntry, QueueItemRef, QueueRemovalReason, QueuedBody, 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. */ @@ -70,6 +70,20 @@ export type StopResult = export type RecoveryResult = { recovered: number; quarantined: boolean; uncertainWorkers: number }; +/** Prior use of a client request id on a session (duplicate-submit guard). */ +export type RequestLookup = + | { kind: 'turn'; turnId: string; sameContent: boolean } + | { kind: 'queued'; queueId: string; sameContent: boolean } + | null; + +export interface EnqueueInput { + content: string; + disposition: 'steering' | 'followup'; + principalId: string; + requestId?: string; + body?: QueuedBody; +} + export interface ResolveWorkerInput { expectedEpoch: string; operatorIdentity: string; @@ -154,6 +168,10 @@ export class ExecutionAuthority { private readonly active = new Map(); private readonly tracked = new Map>(); private released = false; + /** closing gate: set by beginShutdown() BEFORE the host aborts running + * turns, so a turn settling during shutdown can never start new work */ + private closing = false; + private readonly readyWaiters: Array<() => void> = []; constructor( readonly dbPath: string, @@ -173,7 +191,7 @@ export class ExecutionAuthority { 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 === 'draining' || this.closing) 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 }; @@ -193,7 +211,7 @@ export class ExecutionAuthority { /** 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 { + admit(attempt: ExecutionAttempt, input: { content: string; principalId: string; requestId?: string; display?: string; items?: QueueItemRef[]; claim?: string[] }): AdmissionResult { if (!this.owns(attempt)) return { status: 'rejected', code: 'STALE_ATTEMPT' }; const fp = requestFingerprint(input.content); const release = () => this.releaseAttempt(attempt); @@ -215,7 +233,12 @@ export class ExecutionAuthority { 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 cmd: TurnCommand = { + kind: 'turn.request', sessionId: attempt.sessionId, turnId: attempt.turnId, prompt: input.content, refIds: [], at: Date.now(), + ...(input.display !== undefined ? { display: input.display } : {}), + ...(input.items !== undefined ? { items: input.items } : {}), + ...(input.claim !== undefined ? { claim: input.claim } : {}), + } 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); @@ -348,6 +371,95 @@ export class ExecutionAuthority { return { status: wroteTerminal ? 'finalized' : 'already-terminal', accounted }; } + // ── queue (detached send / steer / claim-all drain) ── + + /** Duplicate-submit guard, read BEFORE tryReserve: a client request id is + * looked up in BOTH stores — admitted turns (ledger) and every queued + * item of the session, pending or already claimed/deleted/delivered. */ + lookupRequest(sessionId: string, principalId: string, requestId: string, content: string): RequestLookup { + const fp = requestFingerprint(content); + const turn = this.agent + .prepare('SELECT turn_id, request_hash FROM butler_executions WHERE principal_id = ? AND session_id = ? AND request_id = ?') + .get(principalId, sessionId, requestId) as { turn_id: string; request_hash: string | null } | undefined; + if (turn) return { kind: 'turn', turnId: turn.turn_id, sameContent: turn.request_hash === fp }; + const queued = this.agent + .prepare("SELECT json_extract(detail_json,'$.queueId') AS qid, json_extract(detail_json,'$.fingerprint') AS fp FROM butler_turn_events WHERE session_id = ? AND kind = 'message.queued' AND json_extract(detail_json,'$.requestId') = ? AND json_extract(detail_json,'$.principalId') = ? ORDER BY seq LIMIT 1") + .get(sessionId, requestId, principalId) as { qid: string | null; fp: string | null } | undefined; + if (queued?.qid) return { kind: 'queued', queueId: queued.qid, sameContent: queued.fp === fp }; + return null; + } + + /** Durable enqueue with identity. Returns the item id (the existing one + * when the engine refused a still-pending duplicate). */ + enqueue(sessionId: string, input: EnqueueInput): { queueId: string; disposition: 'steering' | 'followup' } { + const queueId = `q_${randomUUID().slice(0, 12)}`; + this.sink.handle(sessionId, { + kind: 'message.queue', sessionId, content: input.content, disposition: input.disposition, + queueId, principalId: input.principalId, fingerprint: requestFingerprint(input.content), + ...(input.requestId !== undefined ? { requestId: input.requestId } : {}), + ...(input.body !== undefined ? { body: input.body } : {}), + } as TurnCommand, `queue:${sessionId}:${queueId}`); + const entries = this.sink.state(sessionId).entries; + const mine = entries.find((e) => e.queueId === queueId) + ?? (input.requestId !== undefined ? entries.find((e) => e.requestId === input.requestId && e.principalId === input.principalId) : undefined); + return { queueId: mine?.queueId ?? queueId, disposition: mine?.disposition ?? input.disposition }; + } + + /** Remove one pending item; false when it is unknown or already gone. */ + removeQueued(sessionId: string, queueId: string, reason: QueueRemovalReason = 'deleted'): boolean { + const { events } = this.sink.handle(sessionId, { kind: 'queue.remove', sessionId, queueId, reason } as TurnCommand, `queue:${sessionId}:${queueId}:remove`); + return events.length > 0; + } + + /** Pending items that may run (legacy rows without identity never auto-run). */ + pendingQueue(sessionId: string): QueueEntry[] { + return this.sink.state(sessionId).entries.filter((e) => !e.legacy); + } + + /** Sessions holding identity-bearing pending items (boot drain input). */ + sessionsWithPending(): string[] { + const rows = this.agent + .prepare("SELECT DISTINCT session_id FROM butler_turn_events WHERE kind = 'message.queued' AND json_extract(detail_json,'$.queueId') IS NOT NULL") + .all() as Array<{ session_id: string }>; + return rows.map((r) => r.session_id).filter((sid) => this.pendingQueue(sid).length > 0); + } + + /** Steer delivery at a tool boundary, owner-fenced: every pending steering + * item leaves the queue and is recorded as delivered (plus a user row for + * model memory) in ONE commit; the items are returned for the loop. */ + deliverSteering(attempt: ExecutionAttempt): QueueItemRef[] { + if (!this.owns(attempt)) return []; + const steers = this.pendingQueue(attempt.sessionId).filter((e) => e.disposition === 'steering'); + if (steers.length === 0) return []; + const { events } = this.handleOwned(attempt, attempt.sessionId, { + kind: 'steer.deliver', sessionId: attempt.sessionId, turnId: attempt.turnId, queueIds: steers.map((e) => e.queueId), + } as TurnCommand, `steer:${attempt.turnId}:${steers.map((e) => e.queueId).join(',')}`); + const delivered = events.find((e) => e.kind === 'steer.delivered'); + return delivered && delivered.kind === 'steer.delivered' ? delivered.detail.items : []; + } + + /** Generic durable notice row (no host vocabulary). */ + notice(sessionId: string, level: NoticeLevel, text: string, turnId?: string): void { + this.sink.handle(sessionId, { kind: 'session.notice', sessionId, level, text, ...(turnId !== undefined ? { turnId } : {}) } as TurnCommand, `notice:${sessionId}:${randomUUID()}`); + } + + /** Closing gate. Call BEFORE aborting running turns at shutdown: from here + * on nothing new is reserved (sends, drains), so a turn that settles + * during shutdown cannot start the next queued work. Idempotent. */ + beginShutdown(): void { + this.closing = true; + } + + isClosing(): boolean { + return this.closing || this.released || this.phase === 'draining'; + } + + /** Run `cb` once the authority is ready (now, or when quarantine resolves). */ + whenReady(cb: () => void): void { + if (this.phase === 'ready' && !this.closing) { cb(); return; } + this.readyWaiters.push(cb); + } + /** 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); @@ -461,6 +573,7 @@ export class ExecutionAuthority { * control lock. A drain problem keeps the lock held — never close anyway. */ async drain(): Promise { if (this.released) return; + this.closing = true; this.setPhase('draining'); await Promise.allSettled([...this.tracked.values()]); this.setPhase('released'); @@ -490,5 +603,10 @@ export class ExecutionAuthority { 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); + if (phase === 'ready' && !this.closing) { + for (const cb of this.readyWaiters.splice(0)) { + try { cb(); } catch { /* a waiter fault never blocks readiness */ } + } + } } } diff --git a/src/orchestrate/turn-engine.ts b/src/orchestrate/turn-engine.ts index b46260c..647e3ff 100644 --- a/src/orchestrate/turn-engine.ts +++ b/src/orchestrate/turn-engine.ts @@ -19,7 +19,7 @@ export interface TurnEventBase { session_id: string; parent_id: string | null; // pi: tree-ready from day one (nullable column) schema_version: number; // frozen-migration discipline (dsh) - seq: number; // per-turn monotonic, assigned by the sink + seq: number; // per-SESSION monotonic, assigned by the sink (UNIQUE(session_id, seq)) at: number; // epoch ms, assigned by the sink } @@ -30,6 +30,9 @@ export type TurnEventKind = | 'user_message.recorded' | 'message.queued' // pi: follow-up queue entry (decided, visible) | 'queue.updated' // pi: steering/follow-up queue state changes + | 'queue.removed' // one item left the queue (claimed | deleted | delivered) + | 'steer.delivered' // steering items handed to the running turn + | 'session.notice' // generic host-visible notice (info | warn | error) | 'chunk' // assistant text delta | 'tool_call' // ACP entity: first sighting fixes timeline position | 'tool_call_update' // ACP patch by toolCallId (never order-correlation) @@ -51,9 +54,12 @@ export type TurnEvent = | (TurnEventBase & { kind: 'session.imported'; detail: { note: string } }) | (TurnEventBase & { kind: 'turn.requested'; detail: { prompt: string; refIds: string[] } }) | (TurnEventBase & { kind: 'turn.started'; detail: {} }) - | (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: 'user_message.recorded'; detail: { role: 'user' | 'assistant'; content: string; items?: QueueItemRef[]; via?: 'steer' } }) + | (TurnEventBase & { kind: 'message.queued'; detail: { content: string; disposition: 'steering' | 'followup' } & QueuedIdentity }) + | (TurnEventBase & { kind: 'queue.updated'; detail: { steering: string[]; followUp: string[]; ids?: { steering: string[]; followUp: string[] } } }) + | (TurnEventBase & { kind: 'queue.removed'; detail: { queueId: string; reason: QueueRemovalReason } }) + | (TurnEventBase & { kind: 'steer.delivered'; detail: { items: QueueItemRef[] } }) + | (TurnEventBase & { kind: 'session.notice'; detail: { level: NoticeLevel; text: 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 } }) @@ -63,6 +69,24 @@ export type TurnEvent = | (TurnEventBase & { kind: 'turn.failed'; detail: { error: string; partial?: string } }) | (TurnEventBase & { kind: 'turn.aborted'; detail: { reason: string } }); +// ── Queue item identity (detached send / steer / claim-all drain) ── + +/** A queue item as carried on user rows and steer deliveries. */ +export interface QueueItemRef { queueId: string; content: string } +export type QueueRemovalReason = 'claimed' | 'deleted' | 'delivered'; +export type NoticeLevel = 'info' | 'warn' | 'error'; +/** Host-supplied request options a queued item keeps until it runs. */ +export interface QueuedBody { commands?: string[]; model?: string } +/** Additive identity on `message.queued` (absent on legacy rows). */ +export interface QueuedIdentity { + queueId?: string; + body?: QueuedBody; + /** duplicate-submit guard: the host's client request id + principal + content fingerprint */ + requestId?: string; + principalId?: string; + fingerprint?: string; +} + // ── Commands (the ONLY way state changes) ── /** chat-reliability F2 (text block lifecycle): durable block metadata riding @@ -79,9 +103,17 @@ export interface TurnChunkBlock { export type TurnCommand = | { kind: 'session.import'; sessionId: string; note: string } - | { kind: 'turn.request'; sessionId: string; turnId: string; prompt: string; refIds: string[]; at: number } + /** `display`/`items`: the text recorded as the user message (and kept in + * model history) when it differs from the composed `prompt`; `claim`: + * queue ids removed (reason `claimed`) in the SAME decision that starts + * the turn — a crash can never both start the turn and keep the items. */ + | { kind: 'turn.request'; sessionId: string; turnId: string; prompt: string; refIds: string[]; at: number; display?: string; items?: QueueItemRef[]; claim?: string[] } | { kind: 'message.append'; sessionId: string; role: 'user' | 'assistant'; content: string; narrative?: boolean } - | { kind: 'message.queue'; sessionId: string; content: string; disposition: 'steering' | 'followup' } + | ({ kind: 'message.queue'; sessionId: string; content: string; disposition: 'steering' | 'followup' } & QueuedIdentity) + | { kind: 'queue.remove'; sessionId: string; queueId: string; reason: QueueRemovalReason } + /** hand every pending steering item in `queueIds` to the running turn */ + | { kind: 'steer.deliver'; sessionId: string; turnId: string; queueIds: string[] } + | { kind: 'session.notice'; sessionId: string; turnId?: string; level: NoticeLevel; text: string } | { kind: 'turn.start'; sessionId: string; turnId: 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 } @@ -98,16 +130,35 @@ export interface QueueState { followUp: string[]; } +/** One pending queue item with its identity. Legacy rows (no queueId) get a + * synthetic `q_legacy_` id and `legacy: true` — they are visible but + * never auto-drained (they predate the claim-all contract). */ +export interface QueueEntry extends QueuedIdentity { + queueId: string; + content: string; + disposition: 'steering' | 'followup'; + legacy?: boolean; +} + export interface OrchestratorState { sessionId: string; turns: Array<{ id: string; status: 'requested' | 'running' | 'completed' | 'failed' | 'aborted'; prompt: string }>; activeTurnId: string | null; queue: QueueState; + /** pending items in queue order (steering and follow-up interleaved as queued) */ + entries: QueueEntry[]; imported: boolean; } export function initialState(sessionId: string): OrchestratorState { - return { sessionId, turns: [], activeTurnId: null, queue: { steering: [], followUp: [] }, imported: false }; + return { sessionId, turns: [], activeTurnId: null, queue: { steering: [], followUp: [] }, entries: [], imported: false }; +} + +/** queue.updated detail computed from entries (strings kept for legacy readers). */ +function snapshotOf(entries: QueueEntry[]): { steering: string[]; followUp: string[]; ids: { steering: string[]; followUp: string[] } } { + const st = entries.filter((e) => e.disposition === 'steering'); + const fu = entries.filter((e) => e.disposition === 'followup'); + return { steering: st.map((e) => e.content), followUp: fu.map((e) => e.content), ids: { steering: st.map((e) => e.queueId), followUp: fu.map((e) => e.queueId) } }; } // ── The pure orchestrator: decide(state, command) → events ── @@ -132,9 +183,19 @@ export function decide(state: OrchestratorState, cmd: TurnCommand): TurnEvent[] { ...S(cmd.turnId), kind: 'queue.updated', detail: { ...state.queue, followUp: [...state.queue.followUp, cmd.prompt] } }, ]; } + // claim-all: the claimed items leave the queue in THIS decision. A claim + // that matches no pending item decides nothing (the drain lost a race — + // never start a turn for items someone else already took). + const claimed = cmd.claim ? state.entries.filter((e) => cmd.claim!.includes(e.queueId)) : []; + if (cmd.claim && cmd.claim.length > 0 && claimed.length === 0) return []; + const claimEvents: TurnEvent[] = claimed.length === 0 ? [] : [ + ...claimed.map((e) => ({ ...S(cmd.turnId), kind: 'queue.removed' as const, detail: { queueId: e.queueId, reason: 'claimed' as const } })), + { ...S(cmd.turnId), kind: 'queue.updated', detail: snapshotOf(state.entries.filter((e) => !claimed.includes(e))) }, + ]; return [ + ...claimEvents, { ...S(cmd.turnId), kind: 'turn.requested', detail: { prompt: cmd.prompt, refIds: cmd.refIds } }, - { ...S(cmd.turnId), kind: 'user_message.recorded', detail: { role: 'user', content: cmd.prompt } }, + { ...S(cmd.turnId), kind: 'user_message.recorded', detail: { role: 'user', content: cmd.display ?? cmd.prompt, ...(cmd.items !== undefined ? { items: cmd.items } : {}) } }, { ...S(cmd.turnId), kind: 'turn.started', detail: {} }, ]; } @@ -142,13 +203,55 @@ export function decide(state: OrchestratorState, cmd: TurnCommand): TurnEvent[] case 'message.append': return [{ ...S(state.activeTurnId ?? ''), kind: 'user_message.recorded', detail: { role: cmd.role, content: cmd.content } }]; - case 'message.queue': + case 'message.queue': { + const { kind: _k, sessionId: _s, content, disposition, ...identity } = cmd; + // duplicate-submit guard (second line of defence; the host route checks + // removed items too): a still-pending item with the same request id + // from the same principal decides nothing. + if (identity.requestId !== undefined && state.entries.some((e) => e.requestId === identity.requestId && e.principalId === identity.principalId)) return []; + if (identity.queueId === undefined) { + // legacy shape (no identity): unchanged string-snapshot behaviour + return [ + { ...S(state.activeTurnId ?? ''), kind: 'message.queued', detail: { content, disposition } }, + { ...S(state.activeTurnId ?? ''), kind: 'queue.updated', detail: disposition === 'steering' + ? { steering: [...state.queue.steering, content], followUp: state.queue.followUp } + : { steering: state.queue.steering, followUp: [...state.queue.followUp, content] } }, + ]; + } + const entry: QueueEntry = { ...identity, queueId: identity.queueId, content, disposition }; return [ - { ...S(state.activeTurnId ?? ''), kind: 'message.queued', detail: { content: cmd.content, disposition: cmd.disposition } }, - { ...S(state.activeTurnId ?? ''), kind: 'queue.updated', detail: cmd.disposition === 'steering' - ? { steering: [...state.queue.steering, cmd.content], followUp: state.queue.followUp } - : { steering: state.queue.steering, followUp: [...state.queue.followUp, cmd.content] } }, + { ...S(state.activeTurnId ?? ''), kind: 'message.queued', detail: { content, disposition, ...identity } }, + { ...S(state.activeTurnId ?? ''), kind: 'queue.updated', detail: snapshotOf([...state.entries, entry]) }, ]; + } + + case 'queue.remove': { + const hit = state.entries.find((e) => e.queueId === cmd.queueId); + if (!hit) return []; // unknown or already gone: idempotent no-op + return [ + { ...S(state.activeTurnId ?? ''), kind: 'queue.removed', detail: { queueId: cmd.queueId, reason: cmd.reason } }, + { ...S(state.activeTurnId ?? ''), kind: 'queue.updated', detail: snapshotOf(state.entries.filter((e) => e !== hit)) }, + ]; + } + + case 'steer.deliver': { + const t = state.turns.find((x) => x.id === cmd.turnId); + if (!t || t.status !== 'running') return []; + const taken = state.entries.filter((e) => e.disposition === 'steering' && !e.legacy && cmd.queueIds.includes(e.queueId)); + if (taken.length === 0) return []; + const items = taken.map((e) => ({ queueId: e.queueId, content: e.content })); + return [ + ...taken.map((e) => ({ ...S(cmd.turnId), kind: 'queue.removed' as const, detail: { queueId: e.queueId, reason: 'delivered' as const } })), + { ...S(cmd.turnId), kind: 'queue.updated', detail: snapshotOf(state.entries.filter((e) => !taken.includes(e))) }, + { ...S(cmd.turnId), kind: 'steer.delivered', detail: { items } }, + // model memory: later turns rebuild history from user rows only — a + // delivered steer must be one, ordered before this turn's answer. + ...items.map((it) => ({ ...S(cmd.turnId), kind: 'user_message.recorded' as const, detail: { role: 'user' as const, content: it.content, via: 'steer' as const } })), + ]; + } + + case 'session.notice': + return [{ ...S(cmd.turnId ?? state.activeTurnId ?? ''), kind: 'session.notice', detail: { level: cmd.level, text: cmd.text } }]; case 'turn.start': { const t = state.turns.find((x) => x.id === cmd.turnId); @@ -215,7 +318,7 @@ export function decide(state: OrchestratorState, cmd: TurnCommand): TurnEvent[] // ── Fold: events → next state (replay IS the fold; projections read the same fold) ── export function fold(state: OrchestratorState, e: TurnEvent): OrchestratorState { - const next: OrchestratorState = { ...state, turns: [...state.turns], queue: { ...state.queue } }; + const next: OrchestratorState = { ...state, turns: [...state.turns], queue: { ...state.queue }, entries: [...(state.entries ?? [])] }; switch (e.kind) { case 'session.imported': next.imported = true; @@ -243,14 +346,40 @@ export function fold(state: OrchestratorState, e: TurnEvent): OrchestratorState return next; } case 'message.queued': { - const d = e.detail as { content: string; disposition: 'steering' | 'followup' }; + const d = e.detail as { content: string; disposition: 'steering' | 'followup' } & QueuedIdentity; + next.queue = { steering: [...next.queue.steering], followUp: [...next.queue.followUp] }; if (d.disposition === 'steering') next.queue.steering.push(d.content); else next.queue.followUp.push(d.content); + next.entries.push(d.queueId !== undefined + ? { ...d, queueId: d.queueId } + : { queueId: `q_legacy_${e.seq}`, content: d.content, disposition: d.disposition, legacy: true }); + return next; + } + case 'queue.removed': { + const d = e.detail as { queueId: string }; + next.entries = next.entries.filter((x) => x.queueId !== d.queueId); return next; } case 'queue.updated': { - const d = e.detail as { steering: string[]; followUp: string[] }; + const d = e.detail as { steering: string[]; followUp: string[]; ids?: { steering: string[]; followUp: string[] } }; next.queue = { steering: [...d.steering], followUp: [...d.followUp] }; + if (d.ids) { + const keep = new Set([...d.ids.steering, ...d.ids.followUp]); + next.entries = next.entries.filter((x) => keep.has(x.queueId)); + } else { + // legacy snapshot: entries follow the strings; an existing entry with + // the same content/disposition keeps its identity, the rest are legacy + const pool = [...next.entries]; + const take = (content: string, disposition: 'steering' | 'followup', i: number): QueueEntry => { + const at = pool.findIndex((x) => x.content === content && x.disposition === disposition); + if (at !== -1) return pool.splice(at, 1)[0]!; + return { queueId: `q_legacy_${e.seq}_${disposition}_${i}`, content, disposition, legacy: true }; + }; + next.entries = [ + ...d.steering.map((c, i) => take(c, 'steering', i)), + ...d.followUp.map((c, i) => take(c, 'followup', i)), + ]; + } return next; } default: diff --git a/src/transport/conversation.ts b/src/transport/conversation.ts index 28fe56b..68b80c7 100644 --- a/src/transport/conversation.ts +++ b/src/transport/conversation.ts @@ -69,6 +69,9 @@ export interface ConversationDeps { * reports the common execution snapshot. When unset, the legacy * process-local reservation applies (single-instance deployments only). */ execution?: ExecutionAuthority; + /** detached turns: waits (ms) between finalization retries when the + * terminal write fails (default 1s, 5s, 30s). Tests shorten it. */ + finalizeRetryMs?: number[]; } export interface TurnOutcome { @@ -167,52 +170,53 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver return { session: row, messages: await storage.sessionMessages(id), ...(liveTurn ? { liveTurn } : {}) }; }); - 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; 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' }); - const content = b.content; // narrowed string (validated above) — safe in closures + /** One admitted (or legacy-reserved) turn, from preparation to settlement. + * Shared by the held POST (awaits it), the detached POST and the queue + * drain (both run it as a tracked worker). Returns the HTTP envelope the + * held route sends; detached callers only log non-2xx results. */ + interface AdmittedRun { + id: string; + b: { content?: string; commands?: string[]; model?: string }; + content: string; + session: NonNullable>>; + ownedAttempt: ExecutionAttempt | null; + ac: AbortController; + durableTurnId: string; + /** no client waits on this turn: finalization retries itself */ + detached: boolean; + /** held route only: client-disconnect hook (#7) */ + reply?: { raw: { on(ev: 'close', cb: () => void): unknown; writableEnded: boolean } }; + } + type RunResult = { code: number; payload: unknown }; + + const runAdmittedTurn = async (run: AdmittedRun): Promise => { + const { id, b, content, session, ownedAttempt, ac, durableTurnId, detached } = run; 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); }; + // detached finalization: no client will retry a FINALIZATION_PENDING, so + // the worker retries the idempotent finalize itself (bounded backoff), + // stops at shutdown (the attempt stays reserved for boot recovery), and + // leaves a durable notice when every try failed. + const finalize = async (attempt: ExecutionAttempt, input: Parameters[1]) => { + if (!exec) throw new Error('no execution authority'); + if (!detached) return exec.finalizeOwned(attempt, input); + const waits = deps.finalizeRetryMs ?? [1_000, 5_000, 30_000]; + let lastErr: unknown; + for (let i = 0; ; i++) { + try { return await exec.finalizeOwned(attempt, input); } catch (err) { lastErr = err; } + if (i >= waits.length || exec.isClosing()) break; + const until = Date.now() + waits[i]!; + while (Date.now() < until && !exec.isClosing()) await new Promise((r) => setTimeout(r, Math.min(50, until - Date.now()))); + if (exec.isClosing()) break; + } + try { exec.notice(id, 'error', `the turn could not be saved as finished (${lastErr instanceof Error ? lastErr.message : String(lastErr)}) — it stays reserved until recovery`, durableTurnId); } catch { /* storage is what failed */ } + throw lastErr; + }; // 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 @@ -233,8 +237,8 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver } }, 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; + const blk = byHop.get(hop); + return blk ? { id: blk.id, phase: 'delta' as const, ...(blk.native !== undefined ? { native: blk.native } : {}) } : undefined; }, }; }; @@ -259,11 +263,11 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver 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: 'message.queue', sessionId: id, content, disposition: 'steering' }, `chat:${id}:${durableTurnId}:dequeue`); } 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 }); + await storage.appendMessage({ session_id: id, role: 'user', content, tool_calls_json: null, tool_results_json: null, tokens_in: 0, tokens_out: 0 }); } // resume context from rows alone (condition 4) @@ -293,7 +297,7 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver // 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)}` }); + return { code: 500, payload: { 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 }; // ownership liveness flag: the disconnect hook may abort only while THIS @@ -319,9 +323,11 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver // (not socket death); reply.raw 'close' with !writableEnded = the client // 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 && ownsLiveTurn) ac.abort(); + // Detached turns have no waiting client — never wired. + if (deps.abortOnDisconnect && run.reply) { + const raw = run.reply.raw; + raw.on('close', () => { + if (!raw.writableEnded && ownsLiveTurn) ac.abort(); }); } try { @@ -352,8 +358,8 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver const sink = deps.sink; // durable path: chunks/tools/usage/completion ride commands // commands-as-chat-turns: overlay composes AFTER the base (host's block // already carries command bodies only — no double base, review CR1) - const overlay = Array.isArray((b as { commands?: string[] }).commands) && deps.commandOverlay - ? deps.commandOverlay(((b as { commands?: string[] }).commands ?? []).filter((c): c is string => typeof c === 'string')) + const overlay = Array.isArray(b.commands) && deps.commandOverlay + ? deps.commandOverlay((b.commands ?? []).filter((c): c is string => typeof c === 'string')) : null; const cmdBlock = overlay ? overlay.promptBlock : ''; const allowedTools = registry.list().filter((t) => permissionRank(policy.allows) >= permissionRank(t.permission)); @@ -362,12 +368,12 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver : allowedTools; // model routing (PR2): the host resolved a model/combo for THIS body — // run the loop against the router's route-aware complete - const routedModel = deps.resolveModel?.(b as { content?: string; commands?: string[]; model?: string }) ?? undefined; + const routedModel = deps.resolveModel?.(b) ?? undefined; const routedComplete = routedModel !== undefined && deps.routeComplete ? deps.routeComplete(routedModel) : undefined; // v0.6.2: host decoration last — an editor-turn provider (cwd=worktree) // overrides routing; the block rides AFTER the command overlay let prep: { promptBlock?: string; provider?: Provider } | null = null; - try { prep = await deps.prepareTurn?.(id, b as { content: string; commands?: string[] }) ?? null; } + try { prep = await deps.prepareTurn?.(id, { ...b, content }) ?? 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); // AUTHORITY execution boundary: snapshot + CAS-consume the continuation @@ -382,7 +388,7 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver outcome.error = 'aborted before start — no provider ran'; outcome.code = 'ABORTED'; await storage.updateSession(id, { status: 'active' }); - return { turn: outcome }; + return { code: 200, payload: { turn: outcome } }; } const begunSeeded = (begun as { seeded: unknown[] | null }).seeded; if (begunSeeded && begunSeeded.length) { @@ -391,89 +397,73 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver } } const blocks = blockRail(); + // B5: a tool event with no invocation id gets a per-turn COUNTER id — + // never the bare tool name (repeated same-name calls stay distinct); + // a result pairs FIFO with the oldest open call of the same name. + let fallbackSeq = 0; + const openByName = new Map(); + const toolCallIdFor = (t: { phase: 'call' | 'result'; name: string; invocationId?: string }): string => { + if (t.invocationId !== undefined) return `tc_${durableTurnId}_i_${t.invocationId}`; + const open = openByName.get(t.name) ?? []; + if (t.phase === 'result') { + const paired = open.shift(); + if (paired !== undefined) return paired; + } + const fresh = `tc_${durableTurnId}_n${fallbackSeq++}`; + if (t.phase === 'call') { open.push(fresh); openByName.set(t.name, open); } + return fresh; + }; + const loopDeps: Parameters[2] = { + 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) } : {}), + // steering seam: queued steers join at each tool boundary; the + // authority records the delivery (owned, one commit) before handing + // the items over + ...(exec && ownedAttempt ? { takeSteering: () => exec.deliverSteering(ownedAttempt) } : {}), + 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; events with no native id get a per-turn counter id (B5). + if (!sink && !exec) return; + const stable = toolCallIdFor(t); + 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()}`); + } + } + }, + }; + const systemPrompt = deps.systemPrompt + cmdBlock + notesBlock + (prep?.promptBlock ?? '') + resumeNote; 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()}`); - } - } - }, - })); + ? exec.trackWorker({ turnId: durableTurnId, identity: { seam: 'provider' } }, 'provider', () => runLoop(systemPrompt, content, loopDeps)) as unknown as Awaited> + : runLoop(systemPrompt, content, loopDeps)); // 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 "") — @@ -486,7 +476,7 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver outcome.ok = false; outcome.error = 'empty answer after work — turn failed honestly (see event log)'; if (exec && ownedAttempt) { - await exec.finalizeOwned(ownedAttempt, { + await finalize(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`, @@ -499,7 +489,7 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver await storage.updateSession(id, { status: 'error' }); await releaseOwnership('EMPTY_AFTER_WORK'); } - return { turn: outcome }; + return { code: 200, payload: { turn: outcome } }; } outcome.ok = true; outcome.answer = result.answer; @@ -509,13 +499,22 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver // §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 }, - }); + let fin: Awaited>; + try { + fin = await finalize(ownedAttempt, { + kind: 'turn.complete', + answer: result.answer, + commandKey: `chat:${durableTurnId}:done`, + settleCode: 'COMPLETED', + accounting: { status: 'active', tokensInDelta: result.tokens.in, tokensOutDelta: result.tokens.out }, + }); + } catch (finErr) { + // held: today's path (the catch below records a failure terminal); + // detached: the retries are spent and a notice is recorded — do + // not run a second retry series for the failure terminal + if (!detached) throw finErr; + return { code: 503, payload: { error: `finalization storage failure — outcome pending: ${finErr instanceof Error ? finErr.message : String(finErr)}`, code: 'FINALIZATION_PENDING' } }; + } 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 { @@ -524,7 +523,7 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver await releaseOwnership('COMPLETED'); } invokeAfterTurn(true, result.answer); - return { turn: outcome }; + return { code: 200, payload: { turn: outcome } }; } catch (err) { const e = err as { code?: string; message?: string }; // no-hassle resume (2026-10-21 plan): budget exhaustion is a PAUSE, not @@ -548,7 +547,7 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver // leaves the attempt reserved for an idempotent retry and reports // outcome-pending — never a false success. try { - const fin = await exec.finalizeOwned(ownedAttempt, { + const fin = await finalize(ownedAttempt, { kind: 'turn.fail', error: outcome.error, commandKey: `chat:${durableTurnId}:fail`, @@ -559,7 +558,7 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver 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' }); + return { code: 503, payload: { 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`); @@ -567,8 +566,171 @@ export async function registerConversationApi(app: FastifyInstance, deps: Conver await releaseOwnership(budgetExhausted ? 'BUDGET_EXCEEDED' : (e.code ?? 'FAILED')); } invokeAfterTurn(false, ''); - return { turn: outcome }; // the turn failed; the REQUEST succeeded — outcome carries the verdict + return { code: 200, payload: { turn: outcome } }; // the turn failed; the REQUEST succeeded — outcome carries the verdict + } + }; + + // ── detached turns + the claim-all queue drain (authority path only) ── + + /** queue ids claimed by a turn this process admitted (runaway guard) */ + const claimedOnce = new Set(); + + /** Claim EVERY pending item (steers first, then follow-ups) into one turn + * with the reserved attempt. The claim, the user row (display text + + * items) and turn.started commit together inside admit(). Returns null + * (attempt released) when nothing could be claimed. Synchronous: no await + * between reservation and admission. */ + const admitClaimAll = (attempt: ExecutionAttempt, sid: string, fallbackPrincipal: string): { content: string; body: { content: string; commands?: string[]; model?: string } } | null => { + const exec = deps.execution!; + const pending = exec.pendingQueue(sid); + const ordered = [...pending.filter((e) => e.disposition === 'steering'), ...pending.filter((e) => e.disposition === 'followup')]; + if (ordered.length === 0) { exec.releaseAttempt(attempt); return null; } + const content = ordered.length === 1 + ? ordered[0]!.content + : `[${ordered.length} queued messages]\n\n${ordered.map((e) => e.content.trim()).join('\n\n')}`; + const display = ordered.map((e) => e.content).join('\n\n'); + const carrier = ordered.find((e) => e.body?.commands?.length || e.body?.model !== undefined)?.body; + // runaway guard: an item this process already claimed must never be + // claimed again (a claim that failed to remove its items would otherwise + // re-run them after every turn — an endless chain of model calls) + const repeat = ordered.find((e) => claimedOnce.has(e.queueId)); + if (repeat) { + exec.releaseAttempt(attempt); + app.log?.warn?.(`queue drain refused on ${sid}: item ${repeat.queueId} was already claimed by an earlier turn`); + return null; + } + const adm = exec.admit(attempt, { + content, principalId: ordered[0]!.principalId ?? fallbackPrincipal, display, + items: ordered.map((e) => ({ queueId: e.queueId, content: e.content })), + claim: ordered.map((e) => e.queueId), + }); + if (adm.status !== 'admitted') return null; // admit released the attempt + for (const e of ordered) claimedOnce.add(e.queueId); + return { content, body: { content, ...(carrier?.commands ? { commands: carrier.commands } : {}), ...(carrier?.model !== undefined ? { model: carrier.model } : {}) } }; + }; + + /** Start an admitted turn as a tracked worker; when it settles the session + * drains again (claim-all), so queued work never strands. */ + const launchDetached = (run: Omit): void => { + const exec = deps.execution!; + void exec.trackWorker({ turnId: run.durableTurnId, identity: { seam: 'detached' } }, 'detached', async () => { + const r = await runAdmittedTurn({ ...run, detached: true }); + if (r.code >= 400) app.log?.warn?.(`detached turn ${run.durableTurnId} on ${run.id} ended with ${r.code}: ${JSON.stringify(r.payload)}`); + await drainSession(run.id); + }).catch(() => undefined); + }; + + /** Post-turn and boot drain: when items are pending and the session is + * free, claim them all into one detached turn. No-op when closing, busy or + * empty. Held and detached turns both end here (review U3). */ + const drainSession = async (sid: string): Promise => { + const exec = deps.execution; + if (!exec || exec.isClosing()) return; + if (exec.pendingQueue(sid).length === 0) return; + const session = await storage.getSession(sid); + if (!session) return; + // await-free from reservation to admission + const res = exec.tryReserve(sid); + if (res.status !== 'reserved') return; + const claimed = admitClaimAll(res.attempt, sid, session.user_id ?? 'anonymous'); + if (!claimed) return; + launchDetached({ id: sid, b: claimed.body, content: claimed.content, session, ownedAttempt: res.attempt, ac: res.attempt.controller, durableTurnId: res.attempt.turnId }); + }; + + // boot drain: items left pending by a previous process run once the + // authority is ready (never during recovery/quarantine or after shutdown) + if (deps.execution) { + const exec = deps.execution; + exec.whenReady(() => { for (const sid of exec.sessionsWithPending()) void drainSession(sid); }); + } + + // remove one pending queue item (the composer's chip ×) + app.delete('/api/agent/sessions/:id/queue/:queueId', async (req, reply) => { + const { id, queueId } = req.params as { id: string; queueId: string }; + if (!deps.execution) return reply.code(400).send({ error: 'queue management requires an execution authority', code: 'DETACH_UNSUPPORTED' }); + if (!deps.execution.removeQueued(id, queueId, 'deleted')) return reply.code(404).send({ error: 'queue item not found' }); + return { removed: queueId }; + }); + + 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; clientRequestId?: string; detach?: boolean; mode?: 'steer' | 'queue' }; + if (!b.content?.trim()) return reply.code(400).send({ error: 'content required' }); + if (b.mode !== undefined && b.mode !== 'steer' && b.mode !== 'queue') return reply.code(400).send({ error: "mode must be 'steer' or 'queue'" }); + const detach = b.detach === true; + if (detach && !deps.execution) return reply.code(400).send({ error: 'detached send requires an execution authority', code: 'DETACH_UNSUPPORTED' }); + const session = await storage.getSession(id); + if (!session) return reply.code(404).send({ error: 'session not found' }); + const content = b.content; // narrowed string (validated above) — safe in closures + const exec = deps.execution ?? null; + const principalId = session.user_id ?? 'anonymous'; + const queuedBody = { ...(b.commands?.length ? { commands: b.commands } : {}), ...(b.model !== undefined ? { model: b.model } : {}) }; + // 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. + if (exec) { + // detached duplicate-submit guard (reviews T3/U2/V1), inside the + // await-free window: one lookup across admitted turns AND every queued + // item (pending or gone). A hit writes nothing. + if (detach && b.clientRequestId !== undefined) { + const prior = exec.lookupRequest(id, principalId, b.clientRequestId, content); + if (prior && !prior.sameContent) return reply.code(409).send({ error: 'client request id was already used with different content', code: 'REQUEST_ID_REUSE' }); + if (prior?.kind === 'turn') return reply.code(202).send({ startedTurnId: prior.turnId }); + if (prior?.kind === 'queued') return reply.code(202).send({ queued: { queueId: prior.queueId } }); + } + // 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 === 'unavailable') return reply.code(503).send({ error: `chat execution unavailable (${res.reason}) — try again later`, code: res.reason }); + if (res.status === 'busy') { + if (!detach) return reply.code(409).send({ error: 'turn already in flight on this session', ...(res.turnId !== undefined ? { turnId: res.turnId } : {}) }); + // detached + busy: the server owns the queue (steer = next tool + // boundary, queue = next turn) + const q = exec.enqueue(id, { + content, disposition: b.mode === 'steer' ? 'steering' : 'followup', principalId, + ...(b.clientRequestId !== undefined ? { requestId: b.clientRequestId } : {}), + ...(Object.keys(queuedBody).length ? { body: queuedBody } : {}), + }); + return reply.code(202).send({ queued: q }); + } + const attempt = res.attempt; + if (detach && exec.pendingQueue(id).length > 0) { + // idle send while items are pending: it joins the queue and the + // claim-all turn runs everything in order + const q = exec.enqueue(id, { + content, disposition: 'followup', principalId, + ...(b.clientRequestId !== undefined ? { requestId: b.clientRequestId } : {}), + ...(Object.keys(queuedBody).length ? { body: queuedBody } : {}), + }); + const claimed = admitClaimAll(attempt, id, principalId); + if (!claimed) return reply.code(202).send({ queued: q }); + launchDetached({ id, b: claimed.body, content: claimed.content, session, ownedAttempt: attempt, ac: attempt.controller, durableTurnId: attempt.turnId }); + return reply.code(202).send({ startedTurnId: attempt.turnId, queued: q }); + } + const adm = exec.admit(attempt, { content, principalId, ...(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 }); + } + if (detach) { + launchDetached({ id, b, content, session, ownedAttempt: attempt, ac: attempt.controller, durableTurnId: attempt.turnId }); + return reply.code(202).send({ startedTurnId: attempt.turnId }); + } + const r = await runAdmittedTurn({ id, b, content, session, ownedAttempt: attempt, ac: attempt.controller, durableTurnId: attempt.turnId, detached: false, reply }); + // U3: a held turn also hands off to the queue — after the envelope is built + setImmediate(() => { void drainSession(id); }); + return reply.code(r.code).send(r.payload); } + if (activeTurns.has(id)) return reply.code(409).send({ error: 'turn already in flight on this session' }); + const ac = new AbortController(); + activeTurns.set(id, ac); + const r = await runAdmittedTurn({ id, b, content, session, ownedAttempt: null, ac, durableTurnId: `t_${crypto.randomUUID().slice(0, 12)}`, detached: false, reply }); + return reply.code(r.code).send(r.payload); }); } diff --git a/src/workflow/run.ts b/src/workflow/run.ts index fdea10c..847966e 100644 --- a/src/workflow/run.ts +++ b/src/workflow/run.ts @@ -155,6 +155,21 @@ export async function runWorkflow( if (deps.sink && durableTurnId) { deps.sink.handle(run.id, { kind: 'turn.request', sessionId: run.id, turnId: durableTurnId, prompt: JSON.stringify(vin.value), refIds: [], at: Date.now() }, `wf:${durableTurnId}:req`); } + // B5: tool ids per INVOCATION — the loop's id when it has one, else a + // per-turn counter paired FIFO by name (never the bare tool name) + let fallbackSeq = 0; + const openByName = new Map(); + const toolCallIdFor = (t: { phase: 'call' | 'result'; name: string; invocationId?: string }): string => { + if (t.invocationId !== undefined) return `tc_${durableTurnId}_i_${t.invocationId}`; + const open = openByName.get(t.name) ?? []; + if (t.phase === 'result') { + const paired = open.shift(); + if (paired !== undefined) return paired; + } + const fresh = `tc_${durableTurnId}_n${fallbackSeq++}`; + if (t.phase === 'call') { open.push(fresh); openByName.set(t.name, open); } + return fresh; + }; const result = await runLoop(resolved.systemPrompt, JSON.stringify(vin.value), { sessionKey: run.id, // R2-F4: each workflow run is its own claude session provider: resolved.provider, @@ -162,7 +177,7 @@ export async function runWorkflow( ...(resolved.budgets ? { budget: resolved.budgets } : {}), onMessage: (msg) => { events.push(msg); }, onDelta: (d) => { if (deps.sink && durableTurnId) deps.sink.handle(run.id, { kind: 'turn.chunk', sessionId: run.id, turnId: durableTurnId, delta: d }, `wf:${durableTurnId}:d:${d.length}:${Math.random()}`); }, - onTool: (t) => { if (deps.sink && durableTurnId) deps.sink.handle(run.id, { kind: 'turn.tool', sessionId: run.id, turnId: durableTurnId, phase: t.phase === 'call' ? 'call' : 'update', toolCallId: `tc_${durableTurnId}_${t.name}`, ...(t.phase === 'call' ? { name: t.name, ...(t.arg !== undefined ? { arg: t.arg } : {}) } : { ...(t.result !== undefined ? { result: t.result } : {}), ...(t.isError !== undefined ? { isError: t.isError } : {}) }) }, `wf:${durableTurnId}:t:${t.name}:${t.phase}:${Math.random()}`); }, + onTool: (t) => { if (deps.sink && durableTurnId) deps.sink.handle(run.id, { kind: 'turn.tool', sessionId: run.id, turnId: durableTurnId, phase: t.phase === 'call' ? 'call' : 'update', toolCallId: toolCallIdFor(t), ...(t.phase === 'call' ? { name: t.name, ...(t.arg !== undefined ? { arg: t.arg } : {}) } : { ...(t.result !== undefined ? { result: t.result } : {}), ...(t.isError !== undefined ? { isError: t.isError } : {}) }) }, `wf:${durableTurnId}:t:${t.name}:${t.phase}:${Math.random()}`); }, // stream the stage turn: forward loop events (turn.delta/turn.tool) as workflow events onEvent: (e) => { deps.emit?.({ type: `workflow.${e.type}` as 'workflow.turn.delta', runId: run.id, stageId: stage.id, detail: e.detail }); }, }); diff --git a/test/conversation-detach.test.ts b/test/conversation-detach.test.ts new file mode 100644 index 0000000..468c82e --- /dev/null +++ b/test/conversation-detach.test.ts @@ -0,0 +1,437 @@ +import { afterAll, describe, expect, it } from 'vitest'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import Fastify from 'fastify'; +import type 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, type ConversationDeps } from '../src/transport/conversation.ts'; +import { createSqliteExecutionAuthority, type ExecutionAuthority } from '../src/orchestrate/execution-authority.ts'; +import type { TurnEvent } from '../src/orchestrate/turn-engine.ts'; + +/** Detached send (pentools parity, plan 20261022T171041Z B1/B5): instant + * 202, server-owned queue with steer/queue, claim-all drain into ONE turn, + * steer at the tool boundary (and in later turns' memory), queue removal, + * the closing gate, boot drain, duplicate-submit guard, finalize retry. */ + +type Opts = Parameters[2]; +const dirs: string[] = []; +afterAll(() => { for (const d of dirs) rmSync(d, { recursive: true, force: true }); }); + +const deferred = () => { + let resolve!: () => void; + const promise = new Promise((r) => { resolve = r; }); + return { promise, resolve }; +}; +const answer = (content: string): ProviderResponse => ({ message: { role: 'assistant', content }, finishReason: 'stop', usage: { promptTokens: 3, completionTokens: 1 } }); +const toolCall = (id: string): ProviderResponse => ({ message: { role: 'assistant', content: '', tool_calls: [{ id, type: 'function', function: { name: 'nop', arguments: '{}' } }] }, finishReason: 'tool_calls', usage: { promptTokens: 3, completionTokens: 1 } }); + +/** Scripted provider: `step(i, messages, opts)` answers call i; it records + * every call's messages so tests can read the prompt and history. */ +class Script implements Provider { + readonly model = 'script'; + readonly seen: ProviderMessage[][] = []; + constructor(private readonly step: (i: number, m: ProviderMessage[], o: Opts) => Promise) {} + async complete(m: ProviderMessage[], _t: ProviderTool[], o: Opts): Promise { + const i = this.seen.length; + this.seen.push(m.map((x) => ({ ...x }))); + return this.step(i, m, o); + } +} + +interface Rig { dir: string; storage: SqliteStorage; sink: EventSink; auth: ExecutionAuthority; app: Fastify.FastifyInstance; url: string; close(): Promise } + +async function boot(provider: Provider, extra: Partial = {}, dir?: string): Promise { + const d = dir ?? mkdtempSync(join(tmpdir(), 'butler-detach-')); + if (!dir) dirs.push(d); + const agent = join(d, 'agent.db'); + const storage = new SqliteStorage(agent); + const sink = new EventSink(storage.handle() as Database.Database, { 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, sink, + ...extra, + }); + 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: d, storage, sink, auth, app, url, + close: async () => { auth.beginShutdown(); await app.close(); await auth.drain(); storage.close(); }, + }; +} + +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; +}; +const kinds = (rig: Rig, sid: string, kind: TurnEvent['kind']) => rig.sink.events(sid).filter((e) => e.kind === kind); +async function until(pred: () => boolean, ms = 4000): Promise { + const end = Date.now() + ms; + while (!pred()) { + if (Date.now() > end) throw new Error('timed out waiting for condition'); + await new Promise((r) => setTimeout(r, 10)); + } +} +const terminalCount = (rig: Rig, sid: string) => rig.sink.events(sid).filter((e) => e.kind === 'turn.completed' || e.kind === 'turn.failed' || e.kind === 'turn.aborted').length; + +describe('detached send', () => { + it('idle: 202 {startedTurnId} at once while the provider is still working', async () => { + const gate = deferred(); + const rig = await boot(new Script(async () => { await gate.promise; return answer('done'); })); + try { + const sid = await newSession(rig.url); + const t0 = Date.now(); + const res = await post(rig.url, sid, { content: 'hello', detach: true }); + expect(res.status).toBe(202); + expect(Date.now() - t0).toBeLessThan(1000); + const body = await res.json() as { startedTurnId: string }; + expect(body.startedTurnId).toMatch(/^t_/); + expect(kinds(rig, sid, 'turn.completed')).toHaveLength(0); + gate.resolve(); + await until(() => kinds(rig, sid, 'turn.completed').length === 1); + } finally { gate.resolve(); await rig.close(); } + }); + + it('without an authority, detach is refused (never silently held)', async () => { + const dir = mkdtempSync(join(tmpdir(), 'butler-detach-noauth-')); + dirs.push(dir); + const storage = new SqliteStorage(join(dir, 'agent.db')); + const app = Fastify(); + await registerConversationApi(app, { + storage, provider: new Script(async () => answer('x')), registry: new ToolRegistry(), policy: { name: 'draft', allows: 'draft' }, + systemPrompt: 'sys', emit: () => undefined, + }); + 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 newSession(url); + const res = await post(url, sid, { content: 'x', detach: true }); + expect(res.status).toBe(400); + expect(((await res.json()) as { code: string }).code).toBe('DETACH_UNSUPPORTED'); + } finally { await app.close(); storage.close(); } + }); + + it('busy: 202 {queued} with an id; DELETE removes it once (then 404)', async () => { + const gate = deferred(); + const rig = await boot(new Script(async (i) => { if (i === 0) await gate.promise; return answer('done'); })); + try { + const sid = await newSession(rig.url); + await post(rig.url, sid, { content: 'first', detach: true }); + const q = await post(rig.url, sid, { content: 'second', detach: true, mode: 'queue' }); + expect(q.status).toBe(202); + const { queued } = await q.json() as { queued: { queueId: string; disposition: string } }; + expect(queued.queueId).toMatch(/^q_/); + expect(queued.disposition).toBe('followup'); + expect(kinds(rig, sid, 'message.queued')[0]!.detail).toMatchObject({ queueId: queued.queueId, content: 'second' }); + const del = await fetch(`${rig.url}/api/agent/sessions/${sid}/queue/${queued.queueId}`, { method: 'DELETE' }); + expect(del.status).toBe(200); + expect(kinds(rig, sid, 'queue.removed').map((e) => e.detail)).toEqual([{ queueId: queued.queueId, reason: 'deleted' }]); + expect((await fetch(`${rig.url}/api/agent/sessions/${sid}/queue/${queued.queueId}`, { method: 'DELETE' })).status).toBe(404); + gate.resolve(); + await until(() => terminalCount(rig, sid) === 1); + await new Promise((r) => setTimeout(r, 50)); + expect(kinds(rig, sid, 'turn.started')).toHaveLength(1); // the deleted item never ran + } finally { gate.resolve(); await rig.close(); } + }); +}); + +describe('claim-all drain (D-11)', () => { + it('3 queued + 1 leftover steer run as ONE next turn: 4 claimed removals, steers first, display + items on the user row', async () => { + const gate = deferred(); + const provider = new Script(async (i) => { if (i === 0) await gate.promise; return answer(`a${i}`); }); + const rig = await boot(provider); + try { + const sid = await newSession(rig.url); + await post(rig.url, sid, { content: 'first', detach: true }); + for (const c of ['q one', 'q two', 'q three']) expect((await post(rig.url, sid, { content: c, detach: true, mode: 'queue' })).status).toBe(202); + expect((await post(rig.url, sid, { content: 'steer me', detach: true, mode: 'steer' })).status).toBe(202); + gate.resolve(); // turn 1 has no tool boundary → the steer is left over + await until(() => terminalCount(rig, sid) === 2); + expect(kinds(rig, sid, 'turn.started')).toHaveLength(2); + const removed = kinds(rig, sid, 'queue.removed'); + expect(removed).toHaveLength(4); + expect(removed.every((e) => (e.detail as { reason: string }).reason === 'claimed')).toBe(true); + const prompt = provider.seen[1]!.at(-1)!.content!; + expect(prompt).toBe('[4 queued messages]\n\nsteer me\n\nq one\n\nq two\n\nq three'); + const users = kinds(rig, sid, 'user_message.recorded'); + const claimRow = users.at(-1)!.detail as { content: string; items: Array<{ content: string }> }; + expect(claimRow.content).toBe('steer me\n\nq one\n\nq two\n\nq three'); + expect(claimRow.content).not.toContain('queued messages'); + expect(claimRow.items.map((x) => x.content)).toEqual(['steer me', 'q one', 'q two', 'q three']); + expect(rig.auth.pendingQueue(sid)).toEqual([]); + } finally { gate.resolve(); await rig.close(); } + }); + + it('an idle send while items are pending joins them and runs as one claim-all turn', async () => { + const rig = await boot(new Script(async (i) => answer(`a${i}`))); + try { + const sid = await newSession(rig.url); + // pending item without a running turn (e.g. left by a shutdown) + rig.auth.enqueue(sid, { content: 'left over', disposition: 'followup', principalId: 'anonymous' }); + const res = await post(rig.url, sid, { content: 'new one', detach: true }); + expect(res.status).toBe(202); + const body = await res.json() as { startedTurnId: string; queued: { queueId: string } }; + expect(body.startedTurnId).toMatch(/^t_/); + await until(() => terminalCount(rig, sid) === 1); + const user = kinds(rig, sid, 'user_message.recorded').at(-1)!.detail as { content: string }; + expect(user.content).toBe('left over\n\nnew one'); + } finally { await rig.close(); } + }); + + it('the claim is atomic: a failed admission keeps every item pending and starts nothing', async () => { + const rig = await boot(new Script(async () => answer('x'))); + try { + const sid = await newSession(rig.url); + rig.auth.enqueue(sid, { content: 'one', disposition: 'followup', principalId: 'anonymous' }); + rig.auth.enqueue(sid, { content: 'two', disposition: 'followup', principalId: 'anonymous' }); + const db = rig.storage.handle() as Database.Database; + db.exec("CREATE TRIGGER boom BEFORE INSERT ON butler_executions BEGIN SELECT RAISE(ABORT, 'injected crash'); END"); + const res = rig.auth.tryReserve(sid); + if (res.status !== 'reserved') throw new Error('expected a reservation'); + const adm = rig.auth.admit(res.attempt, { content: 'x', principalId: 'anonymous', claim: rig.auth.pendingQueue(sid).map((e) => e.queueId) }); + expect(adm.status).toBe('rejected'); + expect(rig.auth.pendingQueue(sid).map((e) => e.content)).toEqual(['one', 'two']); + expect(kinds(rig, sid, 'queue.removed')).toHaveLength(0); + expect(kinds(rig, sid, 'turn.started')).toHaveLength(0); + db.exec('DROP TRIGGER boom'); + } finally { await rig.close(); } + }); + + it('a held turn ending also drains what a detached send queued (U3)', async () => { + const gate = deferred(); + const rig = await boot(new Script(async (i) => { if (i === 0) await gate.promise; return answer(`a${i}`); })); + try { + const sid = await newSession(rig.url); + const held = post(rig.url, sid, { content: 'held turn' }); + await until(() => kinds(rig, sid, 'turn.started').length === 1); + expect((await post(rig.url, sid, { content: 'queued behind', detach: true, mode: 'queue' })).status).toBe(202); + gate.resolve(); + expect((await held).status).toBe(200); + await until(() => terminalCount(rig, sid) === 2); + expect((kinds(rig, sid, 'user_message.recorded').at(-1)!.detail as { content: string }).content).toBe('queued behind'); + } finally { gate.resolve(); await rig.close(); } + }); +}); + +describe('steer at the tool boundary', () => { + it('a steer joins between hops, writes its rows in one commit, and stays in later turns\' memory (T1)', async () => { + const hop1 = deferred(); + const provider = new Script(async (i) => { + if (i === 0) { await hop1.promise; return toolCall('call_1'); } + if (i === 1) return answer('answer one'); + return answer('answer two'); + }); + const rig = await boot(provider); + const batches: TurnEvent[][] = []; + rig.sink.onAppend((_s, rows) => { batches.push(rows); }); + try { + const sid = await newSession(rig.url); + await post(rig.url, sid, { content: 'turn one', detach: true }); + await until(() => provider.seen.length === 1); + expect((await post(rig.url, sid, { content: 'also mention pineapple', detach: true, mode: 'steer' })).status).toBe(202); + hop1.resolve(); + await until(() => terminalCount(rig, sid) === 1); + // hop 2 saw the steer AFTER the tool result + const hop2 = provider.seen[1]!; + const toolAt = hop2.findIndex((m) => m.role === 'tool'); + const steerAt = hop2.findIndex((m) => m.role === 'user' && m.content === 'also mention pineapple'); + expect(toolAt).toBeGreaterThan(-1); + expect(steerAt).toBeGreaterThan(toolAt); + // durable rows: removal(delivered) + snapshot + steer.delivered + via:steer + // user row land in ONE commit — exactly one pushed batch holds them all + const group = batches.find((rows) => rows.some((e) => e.kind === 'steer.delivered'))!; + expect(group.map((e) => e.kind)).toEqual(['queue.removed', 'queue.updated', 'steer.delivered', 'user_message.recorded']); + expect(group[3]!.detail).toEqual({ role: 'user', content: 'also mention pineapple', via: 'steer' }); + // turn 2's history holds the steer between turn 1's prompt and answer + expect((await post(rig.url, sid, { content: 'turn two' })).status).toBe(200); + const hist = provider.seen[2]!.map((m) => `${m.role}:${m.content}`); + const p = hist.indexOf('user:turn one'); + const s = hist.indexOf('user:also mention pineapple'); + const a = hist.indexOf('assistant:answer one'); + expect(p).toBeGreaterThan(-1); + expect(s).toBeGreaterThan(p); + expect(a).toBeGreaterThan(s); + } finally { hop1.resolve(); await rig.close(); } + }); +}); + +describe('duplicate-submit guard (T3 / U2 / V1)', () => { + it('the same clientRequestId while busy → one queued item, same queueId; again after the drain claimed it → nothing new', async () => { + const gate = deferred(); + const rig = await boot(new Script(async (i) => { if (i === 0) await gate.promise; return answer(`a${i}`); })); + try { + const sid = await newSession(rig.url); + await post(rig.url, sid, { content: 'first', detach: true }); + const a = await (await post(rig.url, sid, { content: 'dup', detach: true, clientRequestId: 'r-1' })).json() as { queued: { queueId: string } }; + const b = await (await post(rig.url, sid, { content: 'dup', detach: true, clientRequestId: 'r-1' })).json() as { queued: { queueId: string } }; + expect(b.queued.queueId).toBe(a.queued.queueId); + expect(kinds(rig, sid, 'message.queued')).toHaveLength(1); + gate.resolve(); + await until(() => terminalCount(rig, sid) === 2); + const late = await post(rig.url, sid, { content: 'dup', detach: true, clientRequestId: 'r-1' }); + expect(late.status).toBe(202); + expect(((await late.json()) as { queued: { queueId: string } }).queued.queueId).toBe(a.queued.queueId); + await new Promise((r) => setTimeout(r, 50)); + expect(kinds(rig, sid, 'turn.started')).toHaveLength(2); + } finally { gate.resolve(); await rig.close(); } + }); + + it('a resend of a send that STARTED a turn returns that turn and never queues; different content → 409', async () => { + const gate = deferred(); + const rig = await boot(new Script(async () => { await gate.promise; return answer('x'); })); + try { + const sid = await newSession(rig.url); + const first = await (await post(rig.url, sid, { content: 'go', detach: true, clientRequestId: 'r-9' })).json() as { startedTurnId: string }; + const again = await post(rig.url, sid, { content: 'go', detach: true, clientRequestId: 'r-9' }); + expect(again.status).toBe(202); + expect(((await again.json()) as { startedTurnId: string }).startedTurnId).toBe(first.startedTurnId); + expect(kinds(rig, sid, 'message.queued')).toHaveLength(0); + const reuse = await post(rig.url, sid, { content: 'other', detach: true, clientRequestId: 'r-9' }); + expect(reuse.status).toBe(409); + expect(((await reuse.json()) as { code: string }).code).toBe('REQUEST_ID_REUSE'); + } finally { gate.resolve(); await rig.close(); } + }); +}); + +describe('shutdown, restart and the boot drain (F2 / F8 / T6)', () => { + it('closing gate: a stopped turn with a queued item starts nothing; the next boot drains it once through the mount hooks', async () => { + const gate = deferred(); + const first = await boot(new Script(async (_i, _m, o) => { + await Promise.race([gate.promise, new Promise((r) => o.signal?.addEventListener('abort', r))]); + return answer('x'); + })); + const dir = first.dir; + const sid = await newSession(first.url); + await post(first.url, sid, { content: 'long turn', detach: true }); + expect((await post(first.url, sid, { content: 'waiting item', detach: true, mode: 'queue' })).status).toBe(202); + // host shutdown order: closing gate FIRST, then abort the running turn + first.auth.beginShutdown(); + first.auth.requestStop(sid); + await until(() => terminalCount(first, sid) === 1); + await new Promise((r) => setTimeout(r, 100)); + expect(kinds(first, sid, 'turn.started')).toHaveLength(1); + expect(first.auth.pendingQueue(sid).map((e) => e.content)).toEqual(['waiting item']); + await first.close(); + + const hooks: string[] = []; + const second = await boot(new Script(async () => answer('resumed')), { + prepareTurn: (s) => { hooks.push(`prepare:${s}`); return null; }, + afterTurn: (s) => { hooks.push(`after:${s}`); }, + }, dir); + try { + await until(() => terminalCount(second, sid) === 2); + expect(kinds(second, sid, 'turn.started')).toHaveLength(2); + expect((kinds(second, sid, 'queue.removed').at(-1)!.detail as { reason: string }).reason).toBe('claimed'); + expect(hooks).toEqual([`prepare:${sid}`, `after:${sid}`]); + expect(second.auth.pendingQueue(sid)).toEqual([]); + } finally { gate.resolve(); await second.close(); } + }); + + it('legacy queue rows (no id) are never run by the boot drain', async () => { + const first = await boot(new Script(async () => answer('x'))); + const sid = await newSession(first.url); + first.sink.handle(sid, { kind: 'message.queue', sessionId: sid, content: 'Reply with exactly: OK', disposition: 'followup' }, 'legacy-1'); + const dir = first.dir; + await first.close(); + const second = await boot(new Script(async () => answer('should not run')), {}, dir); + try { + await new Promise((r) => setTimeout(r, 150)); + expect(kinds(second, sid, 'turn.started')).toHaveLength(0); + expect(second.auth.sessionsWithPending()).toEqual([]); + } finally { await second.close(); } + }); + + it('legacy queue rows are never pulled into a claim-all turn by a new send', async () => { + const rig = await boot(new Script(async (i) => answer(`a${i}`))); + try { + const sid = await newSession(rig.url); + rig.sink.handle(sid, { kind: 'message.queue', sessionId: sid, content: 'Reply with exactly: OK', disposition: 'followup' }, 'legacy-2'); + const res = await post(rig.url, sid, { content: 'fresh', detach: true }); + expect(res.status).toBe(202); + await until(() => terminalCount(rig, sid) === 1); + await new Promise((r) => setTimeout(r, 50)); + expect(kinds(rig, sid, 'turn.started')).toHaveLength(1); + expect((kinds(rig, sid, 'user_message.recorded').at(-1)!.detail as { content: string }).content).toBe('fresh'); + expect(kinds(rig, sid, 'queue.removed')).toHaveLength(0); + } finally { await rig.close(); } + }); +}); + +describe('detached finalization retry (R2 / T4 / T5)', () => { + it('finalize throws twice then succeeds → one terminal row, session free', async () => { + const rig = await boot(new Script(async () => answer('done')), { finalizeRetryMs: [10, 10, 10] }); + try { + const real = rig.auth.finalizeOwned.bind(rig.auth); + let calls = 0; + rig.auth.finalizeOwned = (async (...args: Parameters) => { + calls += 1; + if (calls <= 2) throw new Error('disk busy'); + return real(...args); + }) as ExecutionAuthority['finalizeOwned']; + const sid = await newSession(rig.url); + await post(rig.url, sid, { content: 'go', detach: true }); + await until(() => terminalCount(rig, sid) === 1); + expect(calls).toBe(3); + expect(kinds(rig, sid, 'turn.completed')).toHaveLength(1); + expect(kinds(rig, sid, 'session.notice')).toHaveLength(0); + await until(() => rig.auth.readSession(sid).state === 'idle'); + } finally { await rig.close(); } + }); + + it('finalize always fails → a session.notice error row; the attempt stays reserved for recovery', async () => { + const rig = await boot(new Script(async () => answer('done')), { finalizeRetryMs: [5, 5, 5] }); + try { + rig.auth.finalizeOwned = (async () => { throw new Error('disk gone'); }) as ExecutionAuthority['finalizeOwned']; + const sid = await newSession(rig.url); + await post(rig.url, sid, { content: 'go', detach: true }); + await until(() => kinds(rig, sid, 'session.notice').length === 1); + expect(kinds(rig, sid, 'session.notice')[0]!.detail).toMatchObject({ level: 'error' }); + expect(rig.auth.tryReserve(sid).status).toBe('busy'); + } finally { await rig.close(); } + }); + + it('shutdown during the retry backoff settles within the deadline', async () => { + const rig = await boot(new Script(async () => answer('done')), { finalizeRetryMs: [30_000, 30_000, 30_000] }); + rig.auth.finalizeOwned = (async () => { throw new Error('disk gone'); }) as ExecutionAuthority['finalizeOwned']; + const sid = await newSession(rig.url); + let failedOnce = false; + const real = rig.auth.finalizeOwned; + rig.auth.finalizeOwned = (async (...a: Parameters) => { failedOnce = true; return real(...a); }) as ExecutionAuthority['finalizeOwned']; + await post(rig.url, sid, { content: 'go', detach: true }); + await until(() => failedOnce); + const t0 = Date.now(); + await rig.close(); // beginShutdown + drain awaits the tracked detached worker + expect(Date.now() - t0).toBeLessThan(2000); + }); +}); + +describe('tool ids (B5)', () => { + it('results with no invocation id and no open call get distinct counter ids, never the bare tool name', async () => { + const provider = new Script(async (_i, _m, o) => { + const onTool = (o as { onTool?: (e: { phase: 'call' | 'result'; name: string; result?: string }) => void }).onTool; + onTool?.({ phase: 'result', name: 'Read', result: 'one' }); + onTool?.({ phase: 'result', name: 'Read', result: 'two' }); + return answer('done'); + }); + const rig = await boot(provider); + try { + const sid = await newSession(rig.url); + expect((await post(rig.url, sid, { content: 'go' })).status).toBe(200); + const ids = kinds(rig, sid, 'tool_call_update').map((e) => (e.detail as { toolCallId: string }).toolCallId); + expect(ids).toHaveLength(2); + expect(new Set(ids).size).toBe(2); + expect(ids.every((x) => !x.endsWith('_Read'))).toBe(true); + } finally { await rig.close(); } + }); +}); diff --git a/test/event-sink-onappend.test.ts b/test/event-sink-onappend.test.ts new file mode 100644 index 0000000..faa8a3e --- /dev/null +++ b/test/event-sink-onappend.test.ts @@ -0,0 +1,109 @@ +import { afterAll, describe, expect, it } from 'vitest'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import type Database from 'better-sqlite3'; +import { SqliteStorage } from '../src/memory/sqlite.ts'; +import { EventSink } from '../src/orchestrate/event-sink.ts'; +import type { TurnCommand, TurnEvent } from '../src/orchestrate/turn-engine.ts'; + +/** B2 push-on-append on the REAL EventSink (the scratch prototype modelled + * it): listeners fire only after the OUTERMOST commit, never for rows a + * rollback discarded, in order, and a throwing listener is isolated. */ + +const dirs: string[] = []; +afterAll(() => { for (const d of dirs) rmSync(d, { recursive: true, force: true }); }); + +const rig = () => { + const dir = mkdtempSync(join(tmpdir(), 'butler-onappend-')); + dirs.push(dir); + const storage = new SqliteStorage(join(dir, 'agent.db')); + const db = storage.handle() as Database.Database; + const sink = new EventSink(db, { passive: true, enqueueRunTurn: false }); + const pushed: Array<{ sid: string; rows: TurnEvent[] }> = []; + sink.onAppend((sid, rows) => { pushed.push({ sid, rows }); }); + return { storage, db, sink, pushed }; +}; +const note = (sid: string, text: string): TurnCommand => ({ kind: 'session.notice', sessionId: sid, level: 'info', text }); +const texts = (pushed: Array<{ rows: TurnEvent[] }>) => pushed.flatMap((p) => p.rows.map((r) => (r.detail as { text?: string }).text)); + +describe('EventSink.onAppend', () => { + it('a top-level write is pushed once, after commit, with its committed seq', () => { + const { sink, pushed, storage } = rig(); + sink.handle('s1', note('s1', 'a'), 'k1'); + expect(pushed).toHaveLength(1); + expect(pushed[0]!.sid).toBe('s1'); + expect(pushed[0]!.rows.map((r) => r.seq)).toEqual([1]); + storage.close(); + }); + + it('writes inside withTransaction are NOT pushed before the outer commit, then pushed in order', () => { + const { sink, pushed, storage } = rig(); + sink.withTransaction(() => { + sink.handleInTransaction('s1', note('s1', 'a'), 'k1', undefined); + sink.handleInTransaction('s1', note('s1', 'b'), 'k2', undefined); + expect(pushed).toHaveLength(0); + }); + expect(texts(pushed)).toEqual(['a', 'b']); + storage.close(); + }); + + it('an outer rollback is never pushed — not now, not with the next commit', () => { + const { sink, pushed, storage } = rig(); + expect(() => sink.withTransaction(() => { + sink.handleInTransaction('s1', note('s1', 'doomed'), 'k1', undefined); + throw new Error('boom'); + })).toThrow('boom'); + expect(pushed).toHaveLength(0); + sink.handle('s1', note('s1', 'next'), 'k2'); + expect(texts(pushed)).toEqual(['next']); + storage.close(); + }); + + it('an inner savepoint rollback discards only its own rows', () => { + const { sink, pushed, storage } = rig(); + sink.withTransaction(() => { + sink.handleInTransaction('s1', note('s1', 'kept'), 'k1', undefined); + try { + sink.withTransaction(() => { + sink.handleInTransaction('s1', note('s1', 'inner'), 'k2', undefined); + throw new Error('inner'); + }); + } catch { /* inner savepoint rolled back */ } + }); + expect(texts(pushed)).toEqual(['kept']); + storage.close(); + }); + + it('a failed top-level apply pushes nothing', () => { + const { sink, pushed, storage } = rig(); + expect(() => sink.handleInTransaction('s1', note('s1', 'x'), 'k1', () => { throw new Error('alsoRun failed'); })).toThrow('alsoRun failed'); + expect(pushed).toHaveLength(0); + storage.close(); + }); + + it('a throwing listener never affects the write or the other listeners; unsubscribe stops pushes', () => { + const { sink, pushed, storage } = rig(); + const off = sink.onAppend(() => { throw new Error('listener fault'); }); + expect(() => sink.handle('s1', note('s1', 'a'), 'k1')).not.toThrow(); + expect(texts(pushed)).toEqual(['a']); + expect(sink.events('s1')).toHaveLength(1); + off(); + const seen: string[] = []; + const off2 = sink.onAppend(() => { seen.push('x'); }); + off2(); + sink.handle('s1', note('s1', 'b'), 'k2'); + expect(seen).toEqual([]); + storage.close(); + }); + + it('rows of different sessions in one commit are pushed per session', () => { + const { sink, pushed, storage } = rig(); + sink.withTransaction(() => { + sink.handleInTransaction('s1', note('s1', 'one'), 'k1', undefined); + sink.handleInTransaction('s2', note('s2', 'two'), 'k2', undefined); + }); + expect(pushed.map((p) => p.sid)).toEqual(['s1', 's2']); + storage.close(); + }); +}); diff --git a/test/turn-engine-queue.test.ts b/test/turn-engine-queue.test.ts new file mode 100644 index 0000000..f7f4fab --- /dev/null +++ b/test/turn-engine-queue.test.ts @@ -0,0 +1,115 @@ +import { describe, expect, it } from 'vitest'; +import { decide, fold, initialState, type OrchestratorState, type TurnCommand, type TurnEvent } from '../src/orchestrate/turn-engine.ts'; + +/** Pure engine contract for detached send: identity-bearing queue items, + * per-item removal rows, claim-all admission (display/items), steer + * delivery with a model-memory user row, notices, and legacy folding. */ + +const SID = 's1'; +let seq = 0; +const apply = (st: OrchestratorState, cmd: TurnCommand): { st: OrchestratorState; events: TurnEvent[] } => { + const events = decide(st, cmd).map((e) => ({ ...e, seq: ++seq })); + let next = st; + for (const e of events) next = fold(next, e); + return { st: next, events }; +}; +const queue = (st: OrchestratorState, queueId: string, content: string, disposition: 'steering' | 'followup' = 'followup', extra: Record = {}) => + apply(st, { kind: 'message.queue', sessionId: SID, content, disposition, queueId, principalId: 'u1', ...extra } as TurnCommand); +const running = (): OrchestratorState => + apply(initialState(SID), { kind: 'turn.request', sessionId: SID, turnId: 't1', prompt: 'go', refIds: [], at: 0 }).st; + +describe('turn engine — queue identity', () => { + it('message.queue with an id records the identity and an id-bearing snapshot', () => { + const { st, events } = queue(running(), 'q_a', 'first', 'followup', { body: { model: 'm' }, requestId: 'r1' }); + expect(events.map((e) => e.kind)).toEqual(['message.queued', 'queue.updated']); + expect(events[0]!.detail).toMatchObject({ queueId: 'q_a', requestId: 'r1', principalId: 'u1', body: { model: 'm' } }); + expect(events[1]!.detail).toEqual({ steering: [], followUp: ['first'], ids: { steering: [], followUp: ['q_a'] } }); + expect(st.entries.map((e) => e.queueId)).toEqual(['q_a']); + expect(st.queue.followUp).toEqual(['first']); + }); + + it('a still-pending duplicate request id from the same principal decides nothing', () => { + const one = queue(running(), 'q_a', 'first', 'followup', { requestId: 'r1' }); + expect(queue(one.st, 'q_b', 'first', 'followup', { requestId: 'r1' }).events).toEqual([]); + // another principal's request id never collides + expect(queue(one.st, 'q_c', 'first', 'followup', { requestId: 'r1', principalId: 'u2' }).events).toHaveLength(2); + }); + + it('queue.remove writes one queue.removed + snapshot; unknown ids are a no-op', () => { + const a = queue(running(), 'q_a', 'first'); + const b = queue(a.st, 'q_b', 'second'); + const r = apply(b.st, { kind: 'queue.remove', sessionId: SID, queueId: 'q_a', reason: 'deleted' }); + expect(r.events.map((e) => e.kind)).toEqual(['queue.removed', 'queue.updated']); + expect(r.events[0]!.detail).toEqual({ queueId: 'q_a', reason: 'deleted' }); + expect(r.st.entries.map((e) => e.queueId)).toEqual(['q_b']); + expect(apply(r.st, { kind: 'queue.remove', sessionId: SID, queueId: 'q_a', reason: 'deleted' }).events).toEqual([]); + }); +}); + +describe('turn engine — claim-all admission (display vs prompt)', () => { + it('claims the items in the same decision and records the display text with items', () => { + let st = running(); + st = queue(st, 'q_a', 'alpha').st; + st = queue(st, 'q_b', 'beta').st; + st = apply(st, { kind: 'turn.complete', sessionId: SID, turnId: 't1', answer: 'ok' }).st; + const r = apply(st, { + kind: 'turn.request', sessionId: SID, turnId: 't2', prompt: '[2 queued messages]\n\nalpha\n\nbeta', refIds: [], at: 0, + display: 'alpha\n\nbeta', items: [{ queueId: 'q_a', content: 'alpha' }, { queueId: 'q_b', content: 'beta' }], claim: ['q_a', 'q_b'], + }); + expect(r.events.map((e) => e.kind)).toEqual(['queue.removed', 'queue.removed', 'queue.updated', 'turn.requested', 'user_message.recorded', 'turn.started']); + expect(r.events.filter((e) => e.kind === 'queue.removed').map((e) => e.detail)).toEqual([{ queueId: 'q_a', reason: 'claimed' }, { queueId: 'q_b', reason: 'claimed' }]); + const user = r.events.find((e) => e.kind === 'user_message.recorded')!; + expect(user.detail).toEqual({ role: 'user', content: 'alpha\n\nbeta', items: [{ queueId: 'q_a', content: 'alpha' }, { queueId: 'q_b', content: 'beta' }] }); + expect(r.events.find((e) => e.kind === 'turn.requested')!.detail).toMatchObject({ prompt: '[2 queued messages]\n\nalpha\n\nbeta' }); + expect(r.st.entries).toEqual([]); + expect(r.st.activeTurnId).toBe('t2'); + }); + + it('a claim that matches no pending item decides nothing (lost race)', () => { + const r = apply(initialState(SID), { kind: 'turn.request', sessionId: SID, turnId: 't9', prompt: 'x', refIds: [], at: 0, claim: ['q_gone'] }); + expect(r.events).toEqual([]); + }); +}); + +describe('turn engine — steer delivery', () => { + it('removes the steers, records the delivery AND one via:steer user row per item', () => { + let st = running(); + st = queue(st, 's_1', 'also check X', 'steering').st; + st = queue(st, 'q_f', 'later', 'followup').st; + const r = apply(st, { kind: 'steer.deliver', sessionId: SID, turnId: 't1', queueIds: ['s_1'] }); + expect(r.events.map((e) => e.kind)).toEqual(['queue.removed', 'queue.updated', 'steer.delivered', 'user_message.recorded']); + expect(r.events[0]!.detail).toEqual({ queueId: 's_1', reason: 'delivered' }); + expect(r.events[2]!.detail).toEqual({ items: [{ queueId: 's_1', content: 'also check X' }] }); + expect(r.events[3]!.detail).toEqual({ role: 'user', content: 'also check X', via: 'steer' }); + expect(r.st.entries.map((e) => e.queueId)).toEqual(['q_f']); // the follow-up stays queued + }); + + it('never delivers to a turn that is not running, nor follow-up items', () => { + let st = running(); + st = queue(st, 'q_f', 'later', 'followup').st; + expect(apply(st, { kind: 'steer.deliver', sessionId: SID, turnId: 't1', queueIds: ['q_f'] }).events).toEqual([]); + st = queue(st, 's_1', 'steer', 'steering').st; + const done = apply(st, { kind: 'turn.complete', sessionId: SID, turnId: 't1', answer: 'ok' }).st; + expect(apply(done, { kind: 'steer.deliver', sessionId: SID, turnId: 't1', queueIds: ['s_1'] }).events).toEqual([]); + }); +}); + +describe('turn engine — notices and legacy rows', () => { + it('session.notice is a plain row against the given turn', () => { + const r = apply(running(), { kind: 'session.notice', sessionId: SID, turnId: 't1', level: 'error', text: 'boom' }); + expect(r.events).toHaveLength(1); + expect(r.events[0]).toMatchObject({ kind: 'session.notice', turn_id: 't1', detail: { level: 'error', text: 'boom' } }); + }); + + it('legacy queue rows (no id) fold into legacy entries that keep their identity across snapshots', () => { + let st = running(); + st = apply(st, { kind: 'message.queue', sessionId: SID, content: 'old', disposition: 'followup' }).st; + expect(st.entries).toHaveLength(1); + expect(st.entries[0]).toMatchObject({ content: 'old', legacy: true }); + expect(st.entries[0]!.queueId.startsWith('q_legacy_')).toBe(true); + // a mixed snapshot: legacy strings keep identity, id-bearing entries survive + st = queue(st, 'q_new', 'new').st; + expect(st.entries.map((e) => e.content)).toEqual(['old', 'new']); + expect(st.entries[0]!.legacy).toBe(true); + }); +});