fix(chat reliability): execution authority, atomic admission, failure classification, durable runtime tools

src/index.ts

@@ -77,6 +77,8 @@ export type { ClaudeCliEvent } from './provider/claude-cli.ts';
 export { decide, fold, initialState, TURN_EVENT_SCHEMA_VERSION } from './orchestrate/turn-engine.ts';
 export type { TurnCommand, TurnEvent as DurableTurnEvent, TurnEventKind as DurableTurnEventKind, OrchestratorState, QueueState, UsageMeters } from './orchestrate/turn-engine.ts';
 export { EventSink } from './orchestrate/event-sink.ts';
+export { createSqliteExecutionAuthority, ExecutionAuthority, ExecutionAuthorityBusyError } from './orchestrate/execution-authority.ts';
+export type { ExecutionAuthorityDeps, ExecutionAttempt, ExecutionRead, RecoveryResult } from './orchestrate/execution-authority.ts';
 export type { Effect, EventRow, OutboxRow } from './orchestrate/event-sink.ts';
 export { EffectWorker } from './orchestrate/effect-worker.ts';
 export type { EffectWorkerDeps } from './orchestrate/effect-worker.ts';

src/loop.ts

@@ -5,9 +5,34 @@
  *  via the persistence hook. */
 import type { Provider, ProviderMessage, ProviderTool } from './provider/index.ts';
 import { ProviderError } from './provider/index.ts';
+import type { TextBlockActivity } from './provider/activity.ts';
+import { createBlockTracker } from './provider/activity.ts';
+
+/** Loop failure codes. Budget/abort/invalid are loop-owned outcomes;
+ *  PROVIDER_* codes classify provider failures (chat-reliability F3: the loop
+ *  must never relabel a permanent provider rejection as exhausted runway).
+ *  The `string & {}` member passes provider-owned codes through unchanged
+ *  (e.g. PROVIDER_BAD_REQUEST, PROVIDER_UNAVAILABLE, MCP_PENDING). */
+export type LoopErrorCode =
+  | 'BUDGET_EXCEEDED'
+  | 'ABORTED'
+  | 'TOOL_CALL_INVALID'
+  | 'PROVIDER_TIMEOUT'
+  | 'PROVIDER_FAILED'
+  | (string & {});
+
+/** What actually terminated the loop. Consumers key continuation eligibility
+ *  (budget stash vs honest failure) off the cause, not the message string. */
+export type LoopErrorCause =
+  | 'budget-hop'
+  | 'budget-wall'
+  | 'budget-token'
+  | 'caller-abort'
+  | 'provider'
+  | 'tool-invalid';
 
 export class LoopError extends Error {
-  constructor(message: string, readonly code: 'BUDGET_EXCEEDED' | 'ABORTED' | 'TOOL_CALL_INVALID', readonly partial: ProviderMessage[]) {
+  constructor(message: string, readonly code: LoopErrorCode, readonly partial: ProviderMessage[], override readonly cause?: LoopErrorCause, readonly status?: number) {
     super(message);
     this.name = 'LoopError';
   }
@@ -62,10 +87,18 @@ export interface LoopDeps {
   onDelta?: (delta: string) => void;
   /** R2-F4: provider-scoped session key (claude-cli --resume isolation) */
   sessionKey?: string;
-  /** structured tool events (claude-cli harness + registry loop) */
-  onTool?: (e: { phase: 'call' | 'result'; name: string; arg?: string; result?: string; isError?: boolean }) => void;
+  /** structured tool events (claude-cli harness + registry loop).
+   *  invocationId: the provider-NATIVE call id when one exists (runtime
+   *  registry tools carry it from the assistant tool_calls); durability keys
+   *  per INVOCATION, never per tool name. */
+  onTool?: (e: { phase: 'call' | 'result'; name: string; invocationId?: string; arg?: string; result?: string; isError?: boolean }) => void;
   /** R3 decision 4: opt INTO the hard token cap (fan-out children); default advisory */
   hardTokenCap?: boolean;
+  /** chat-reliability F2 (text block lifecycle): per-block activity for the
+   *  durable chunk rail — start/delta/end per assistant message (hop), so a
+   *  replayed log can separate consecutive hops' narration instead of gluing
+   *  it. Purely additive: absent → behavior identical to before. */
+  onActivity?: (a: TextBlockActivity) => void;
   signal?: AbortSignal;
 }
 
@@ -85,38 +118,83 @@ export async function runLoop(systemPrompt: string, userMessage: string, deps: L
   ];
   const tokens = { in: 0, out: 0 };
   let repairs = 0;
+  // chat-reliability F2: the block tracker runs only when a consumer asked
+  // for block activity; block ids are hop-scoped and stable per run.
+  const tracker = deps.onActivity ? createBlockTracker(deps.onActivity, (h) => `blk_h${h}`) : undefined;
 
   for (let hop = 1; hop <= budget.maxHops; hop++) {
-    if (deps.signal?.aborted) throw new LoopError('aborted by caller', 'ABORTED', messages);
+    if (deps.signal?.aborted) throw new LoopError('aborted by caller', 'ABORTED', messages, 'caller-abort');
     const remainingMs = budget.wallMs - (Date.now() - started);
-    if (remainingMs <= 0) throw new LoopError(`wall budget exhausted after ${Date.now() - started}ms`, 'BUDGET_EXCEEDED', messages);
+    if (remainingMs <= 0) throw new LoopError(`wall budget exhausted after ${Date.now() - started}ms`, 'BUDGET_EXCEEDED', messages, 'budget-wall');
 
     deps.onEvent?.({ type: 'turn.start', hop });
     // the LOOP owns the wall: even a provider that ignores remainingMs gets
-    // aborted at the deadline (mock-hang proof, review F5)
-    const wallSignal = AbortSignal.any([AbortSignal.timeout(remainingMs), ...(deps.signal ? [deps.signal] : [])]);
+    // aborted at the deadline (mock-hang proof, review F5). F3: the deadline
+    // stays a SEPARATE signal so the catch can attribute the failure from
+    // observed signal state instead of the exception name; composition with
+    // the caller signal happens only here, at the provider seam.
+    const wallDeadline = AbortSignal.timeout(remainingMs);
+    const wallSignal = AbortSignal.any(deps.signal ? [wallDeadline, deps.signal] : [wallDeadline]);
+    // chat-reliability identity contract: harness tool events without a
+    // provider-native id get one ALLOCATED at first observation — never key
+    // durability by tool name (repeated same-name invocations must stay
+    // distinct). Pairing call→result is FIFO per name WITHIN this provider
+    // call only (a deterministic window); an unpaired result stays uncorrelated.
+    const harnessPending = new Map<string, string[]>();
+    let harnessSeq = 0;
     let res;
+    // F2: the LOOP accumulates the streamed text itself (sum of deltas) — the
+    // completed message's content may differ from the stream (claude-cli
+    // snapshot vs partial stream), so text.end must not trust content.
+    let streamedText = '';
     try {
       // streaming: the EVENT always flows (WS UIs); the direct callback stays
       // opt-in for hosts that want it without the event bus
-      res = await deps.provider.complete(messages, toolDefs(deps.tools), { remainingMs, signal: wallSignal, ...(deps.sessionKey !== undefined ? { sessionKey: deps.sessionKey } : {}), onDelta: (d: string) => { deps.onDelta?.(d); deps.onEvent?.({ type: 'turn.delta', hop, detail: { delta: d } }); }, onTool: (e: { phase: 'call' | 'result'; name: string; arg?: string; result?: string; isError?: boolean }) => { deps.onTool?.(e); deps.onEvent?.({ type: 'turn.tool', hop, detail: e }); } });
+      res = await deps.provider.complete(messages, toolDefs(deps.tools), { remainingMs, signal: wallSignal, ...(deps.sessionKey !== undefined ? { sessionKey: deps.sessionKey } : {}), onDelta: (d: string) => { streamedText += d; const bid = tracker?.onDelta(hop, d); deps.onDelta?.(d); deps.onEvent?.({ type: 'turn.delta', hop, detail: { delta: d, ...(bid !== undefined ? { blockId: bid } : {}) } }); }, onTool: (e: { phase: 'call' | 'result'; name: string; invocationId?: string; arg?: string; result?: string; isError?: boolean }) => {
+          let out = e;
+          if (e.invocationId === undefined) {
+            const pending = harnessPending.get(e.name) ?? [];
+            if (e.phase === 'call') {
+              const id = `alloc_${hop}_${harnessSeq++}`;
+              pending.push(id);
+              harnessPending.set(e.name, pending);
+              out = { ...e, invocationId: id };
+            } else {
+              const id = pending.shift();
+              harnessPending.set(e.name, pending);
+              if (id !== undefined) out = { ...e, invocationId: id };
+            }
+          }
+          deps.onTool?.(out);
+          deps.onEvent?.({ type: 'turn.tool', hop, detail: out });
+        } });
     } catch (err) {
-      // caller abort vs wall timeout: AbortSignal.timeout raises TimeoutError;
-      // a caller signal abort propagates AbortError (or a custom reason)
-      const isWallTimeout = err instanceof Error && err.name === 'TimeoutError';
-      if (deps.signal?.aborted && !isWallTimeout) throw new LoopError('aborted by caller', 'ABORTED', messages);
-      // provider exhausted its own retries/timeouts — surface with partial transcript
-      const pe = err instanceof ProviderError ? err : null;
-      const msg = pe ? `${pe.code}: ${pe.message}`
-        : err instanceof Error && (err.name === 'TimeoutError' || err.name === 'AbortError') ? 'PROVIDER_TIMEOUT: wall deadline reached during provider call'
-        : `provider failed: ${err instanceof Error ? err.message : String(err)}`;
-      throw new LoopError(msg, 'BUDGET_EXCEEDED', messages);
+      // F3 classification, by precedence: an observed caller abort wins
+      // (explicit Stop beats everything, even a concurrent wall expiry), then
+      // the loop-owned wall, then the provider's own typed failure, then
+      // unattributed timeouts, then unknown. A permanent provider rejection
+      // (auth, bad request) is NEVER relabeled as exhausted runway.
+      if (deps.signal?.aborted) throw new LoopError('aborted by caller', 'ABORTED', messages, 'caller-abort');
+      if (wallDeadline.aborted) throw new LoopError(`wall budget exhausted after ${Date.now() - started}ms`, 'BUDGET_EXCEEDED', messages, 'budget-wall');
+      if (err instanceof ProviderError) throw new LoopError(`${err.code}: ${err.message}`, err.code, messages, 'provider', err.status);
+      if (err instanceof Error && (err.name === 'TimeoutError' || err.name === 'AbortError')) {
+        throw new LoopError(`PROVIDER_TIMEOUT: ${err.message}`, 'PROVIDER_TIMEOUT', messages, 'provider');
+      }
+      throw new LoopError(`provider failed: ${err instanceof Error ? err.message : String(err)}`, 'PROVIDER_FAILED', messages, 'provider');
     }
 
     // butler#8 (defensive twin): a provider that RESOLVES after the caller
     // aborted (claude-cli exits 0 with an empty result on interrupt) is still
     // an abort, not an empty answer — classify before touching the result.
-    if (deps.signal?.aborted) throw new LoopError('aborted by caller', 'ABORTED', messages);
+    // F3: the guard fires only when the caller signal is actually observed
+    // aborted; ordinary success passes untouched.
+    if (deps.signal?.aborted) throw new LoopError('aborted by caller', 'ABORTED', messages, 'caller-abort');
+
+    // F2: close the hop's text block BEFORE the message lands. text.end prefers
+    // the STREAMED accumulation when any deltas flowed (byte-exact to what the
+    // UI saw); content is the fallback for snapshot-only hops (which usually
+    // have no block at all — then the tracker emits nothing).
+    if (tracker) tracker.onMessageComplete(hop, streamedText !== '' ? streamedText : (res.message.content ?? ''));
 
     if (res.usage) { tokens.in += res.usage.promptTokens; tokens.out += res.usage.completionTokens; }
     messages.push(res.message);
@@ -131,7 +209,7 @@ export async function runLoop(systemPrompt: string, userMessage: string, deps: L
     // via hardTokenCap=true; the runaway guard remains the universal backstop.
     if (tokens.in + tokens.out > budget.maxTokens) {
       if (deps.hardTokenCap) {
-        throw new LoopError(`token budget exceeded (${tokens.in + tokens.out} > ${budget.maxTokens})`, 'BUDGET_EXCEEDED', messages);
+        throw new LoopError(`token budget exceeded (${tokens.in + tokens.out} > ${budget.maxTokens})`, 'BUDGET_EXCEEDED', messages, 'budget-token');
       }
       deps.onEvent?.({ type: 'turn.delta', hop, detail: { advisory: `cost ${tokens.in + tokens.out} > ${budget.maxTokens} (advisory)` } });
     }
@@ -166,21 +244,29 @@ export async function runLoop(systemPrompt: string, userMessage: string, deps: L
           deps.onMessage?.(msg, { hop, tokensIn: 0, tokensOut: 0 });
           continue;
         }
-        throw new LoopError(`tool ${call.function.name} sent invalid JSON twice`, 'TOOL_CALL_INVALID', messages);
+        throw new LoopError(`tool ${call.function.name} sent invalid JSON twice`, 'TOOL_CALL_INVALID', messages, 'tool-invalid');
       }
       deps.onEvent?.({ type: 'turn.tool', hop, detail: { phase: 'call', name: call.function.name } });
+      // chat-reliability F1 (tool durability): ordinary registry tools must
+      // ride the SAME durable seam as streamed harness tools — the loop emits
+      // per-invocation call/result callbacks so the transport's durable path
+      // records them (previously runtime tools produced ZERO durable events).
+      deps.onTool?.({ phase: 'call', name: call.function.name, invocationId: call.id, arg: call.function.arguments });
       let obs: string;
+      let toolFailed = false;
       try {
         obs = await tool.execute(args, { signal: deps.signal ?? undefined });
       } catch (err) {
         // handler throws become observations (ReAct) — the loop continues
         obs = `error: ${err instanceof Error ? err.message : String(err)}`;
+        toolFailed = true;
       }
+      deps.onTool?.({ phase: 'result', name: call.function.name, invocationId: call.id, result: obs, isError: toolFailed });
       const msg: ProviderMessage = { role: 'tool', content: obs, tool_call_id: call.id };
       messages.push(msg);
       deps.onMessage?.(msg, { hop, tokensIn: 0, tokensOut: 0 });
       deps.onEvent?.({ type: 'turn.tool', hop, detail: { phase: 'result', name: call.function.name, result: `${obs.length} bytes` } });
     }
   }
-  throw new LoopError(`hop budget exhausted (${budget.maxHops})`, 'BUDGET_EXCEEDED', messages);
+  throw new LoopError(`hop budget exhausted (${budget.maxHops})`, 'BUDGET_EXCEEDED', messages, 'budget-hop');
 }

src/memory/contract.ts

@@ -116,6 +116,20 @@ export interface Storage {
   // read — to seed the loop so the model continues instead of redoing hops.
   stashPartial(sessionId: string, turnId: string, json: string): Promise<void>;
   takePartial(sessionId: string): Promise<{ turnId: string; json: string } | null>;
+  /** chat-reliability F1/F2 (optional capability, authority-driven hosts):
+   *  non-destructive peek + exact compare-and-delete of the continuation
+   *  stash. The authority peeks during preparation and consumes ONLY at the
+   *  committed execution boundary, so a failed preparation never destroys a
+   *  continuation. */
+  peekPartial?(sessionId: string): Promise<{ turnId: string; json: string; created_at: number } | null>;
+  consumePartialExact?(sessionId: string, turnId: string, json: string): Promise<boolean>;
+  /** chat-reliability finalizeOwned: SYNCHRONOUS variants for use INSIDE the
+   *  atomic finalization transaction (better-sqlite3 forbids await in a
+   *  transaction; the async methods above cannot join the commit). Optional
+   *  capability — hosts requiring atomic finalization must use a driver that
+   *  provides them and verify it at startup. */
+  applyAccountingSync?(sessionId: string, patch: { status?: 'active' | 'ended' | 'error'; tokensInDelta?: number; tokensOutDelta?: number }): void;
+  stashPartialSync?(sessionId: string, turnId: string, json: string): void;
 
   // notes (explicit-write only; supersession, never delete)
   insertNote(n: Omit<NoteRow, 'id' | 'created_at' | 'superseded_by' | 'superseded_at' | 'superseded_reason'>): Promise<NoteRow>;

src/memory/sqlite.ts

@@ -189,6 +189,31 @@ export class SqliteStorage implements Storage {
     return { turnId: row.turn_id, json: row.json };
   }
 
+  /** chat-reliability F1/F2: non-destructive peek — preparation validates without consuming. */
+  async peekPartial(sessionId: string): Promise<{ turnId: string; json: string; created_at: number } | null> {
+    const row = this.db.prepare('SELECT turn_id, json, created_at FROM butler_turn_partials WHERE session_id = ?').get(sessionId) as { turn_id: string; json: string; created_at: number } | undefined;
+    return row ? { turnId: row.turn_id, json: row.json, created_at: row.created_at } : null;
+  }
+
+  /** chat-reliability F1/F2: exact CAS delete — consumes the stash only if it
+   *  is still the exact (turnId, json) we snapshotted at the execution boundary. */
+  async consumePartialExact(sessionId: string, turnId: string, json: string): Promise<boolean> {
+    return this.db.prepare('DELETE FROM butler_turn_partials WHERE session_id = ? AND turn_id = ? AND json = ?').run(sessionId, turnId, json).changes > 0;
+  }
+
+  /** chat-reliability finalizeOwned: sync, in-transaction session accounting. */
+  applyAccountingSync(sessionId: string, patch: { status?: 'active' | 'ended' | 'error'; tokensInDelta?: number; tokensOutDelta?: number }): void {
+    const cur = this.db.prepare('SELECT status, tokens_in, tokens_out FROM butler_sessions WHERE id = ?').get(sessionId) as { status: string; tokens_in: number; tokens_out: number } | undefined;
+    if (!cur) throw new Error(`session ${sessionId} not found`);
+    this.db.prepare('UPDATE butler_sessions SET status = ?, tokens_in = ?, tokens_out = ?, updated_at = ? WHERE id = ?')
+      .run(patch.status ?? cur.status, cur.tokens_in + (patch.tokensInDelta ?? 0), cur.tokens_out + (patch.tokensOutDelta ?? 0), Date.now(), sessionId);
+  }
+
+  /** chat-reliability finalizeOwned: sync, in-transaction budget-stash write. */
+  stashPartialSync(sessionId: string, turnId: string, json: string): void {
+    this.db.prepare('INSERT OR REPLACE INTO butler_turn_partials(session_id, turn_id, json, created_at) VALUES (?,?,?,?)').run(sessionId, turnId, json, Date.now());
+  }
+
   async sessionMessages(sessionId: string): Promise<MessageRow[]> {
     return this.db.prepare('SELECT * FROM butler_messages WHERE session_id = ? ORDER BY rowid').all(sessionId) as MessageRow[];
   }