From 9db9689546d48922e7569b605f29d9fb4c4c3f5e Mon Sep 17 00:00:00 2001 From: Aaron Gabriel Date: Fri, 19 Jun 2026 07:57:03 -0600 Subject: [PATCH 1/2] fix(robustness): harden turn-delivery + claim + def-lifecycle error paths (PR #3) Six post-audit point-fixes in the load-bearing reply-delivery, claim, and def-lifecycle paths. The audit flagged silent reply-loss as the scariest failure mode for an agent module; each fix has a regression test. FIX 1 (registry.ts) - vault 5xx/network during the outbound write no longer silently loses the reply. The outbound write now gets a BOUNDED retry (2 retries, linear backoff) on a TRANSIENT error (5xx or no-status network), but NOT on a 4xx (a real rejection). On a persistent failure the live view resolves to error (not done) AND the #agent/thread note is RE-RECORDED as status:error carrying the undelivered reply text - so the durable record never falsely claims a clean "ok". The claude -p turn is never re-run (no fork/quota burn); only the idempotent outbound WRITE retries. Backoff base is injectable for fast tests. FIX 2 (programmatic.ts) - mid-turn vault-token expiry: ASSESSED + DEFERRED (documented, no code). The token is minted FRESH per turn at the hub default ~90d TTL, so it cannot expire during a minutes-long turn; and the vault writes happen inside the opaque claude -p subprocess via the token baked into its .mcp.json, so the backend has no in-process seam to observe a 401 and re-mint. A re-mint-on-401 is infeasible at this layer + unnecessary; the real fix (if a long/short-TTL turn ever makes it real) is MCP-client refresh-on-401, flagged as a follow-up. FIX 3 (channel-queue.ts, vault.ts) - claimNext double-claim race. The claim PATCH is now a COMPARE-AND-SWAP: it carries if_updated_at (the note's last-seen updated_at); the vault returns 409 (stale precondition) / 428 (precondition required) when the race is lost, surfaced as a typed InboundClaimConflictError, on which claimNext re-lists and tries the next pending message instead of double-claiming. updated_at is threaded through InboundQueueNote. release/handled/sweep keep last-write-wins (force). FIX 4 (agent-defs.ts) - deleteDef now deletes the vault note FIRST, then deregisters the in-memory agent (mirrors the agent-vaults removal ordering): a vault-delete 502 throws before any teardown, leaving the def REGISTERED (it re-converges on the next poll) instead of orphaned in the confusing gone-from-memory-but-still-in-vault half-state. FIX 5 (agent-defs.ts, daemon.ts) - a grant-reconcile failure on def-delete is no longer silently swallowed. It is still best-effort (does not block the delete), but now warns loudly AND returns a grantsReconciled:false partial- success signal so the delete path/route reports that approved hub grants may be orphaned rather than claiming a clean full success. FIX 6 (vault.ts) - listInboundQueue no longer lets accumulated handled notes crowd pending out of the query cap. handled notes are excluded client-side (only pending + in-flight are the actionable queue) and the vault query is requested newest-first so a hard cap drops the oldest handled notes, never a recent pending. status is not indexable per-vault, so a server-side filter is a future scale optimization. Gates: typecheck 0 errors; bun test 970 pass / 0 fail; vitest 87 pass / 0 fail. Closes #101 Closes #96 Closes #103 Co-Authored-By: Claude Opus 4.8 (1M context) --- src/agent-defs.test.ts | 120 ++++++++++++++++++++++++ src/agent-defs.ts | 70 +++++++++++--- src/backends/channel-queue.test.ts | 76 ++++++++++++++- src/backends/channel-queue.ts | 88 +++++++++++++----- src/backends/programmatic.ts | 13 +++ src/backends/registry.test.ts | 131 ++++++++++++++++++++++++-- src/backends/registry.ts | 124 ++++++++++++++++++++++--- src/daemon.ts | 10 ++ src/transports/vault.test.ts | 144 ++++++++++++++++++++++++++++- src/transports/vault.ts | 105 ++++++++++++++++++--- 10 files changed, 809 insertions(+), 72 deletions(-) diff --git a/src/agent-defs.test.ts b/src/agent-defs.test.ts index 280bcf6..1648c2d 100644 --- a/src/agent-defs.test.ts +++ b/src/agent-defs.test.ts @@ -1187,3 +1187,123 @@ describe("AgentDefRegistry — grant-GC reconcile (#96)", () => { expect(calls.registered.map((s) => s.name)).toEqual(["uni"]); }); }); + +// --------------------------------------------------------------------------- +// FIX 4 (delete ordering: vault-delete first, then deregister) + FIX 5 (grant-GC +// failure on delete is surfaced, not swallowed) — PR #3. +// --------------------------------------------------------------------------- + +/** + * A fetch that serves the def list + by-id GET (so an agent instantiates) and routes + * a DELETE to a configurable outcome (`deleteStatus`). Records DELETEs so a test can + * assert the note-delete was attempted. Reconcile/PATCH succeed by default. + */ +function vaultFetchWithDelete(opts: { + defs: Array<{ id: string; content?: string; metadata?: Record }>; + deleteStatus?: number; // the status the DELETE returns (default 204 = success) + deletes?: string[]; // record each DELETEd note id +}): typeof fetch { + return (async (url: string | URL | Request, init?: RequestInit) => { + const u = String(url); + const method = init?.method ?? "GET"; + if (method === "DELETE") { + const id = decodeURIComponent(u.split("/api/notes/")[1]!); + opts.deletes?.push(id); + const status = opts.deleteStatus ?? 204; + return new Response(status >= 400 ? "delete failed" : null, { status }); + } + if (method === "PATCH") return new Response(null, { status: 200 }); + if (u.includes("/api/notes?") && u.includes("tag=%23agent%2Fdefinition")) { + return new Response(JSON.stringify(opts.defs), { + status: 200, + headers: { "content-type": "application/json" }, + }); + } + return new Response("[]", { status: 200 }); + }) as typeof fetch; +} + +describe("AgentDefRegistry — deleteDef ordering + grant-GC surfacing (FIX 4/5, PR #3)", () => { + test("FIX 4: a vault-delete failure leaves the def REGISTERED (not orphaned)", async () => { + const { deps, calls } = recorderDeps(); + const deletes: string[] = []; + const fetchFn = vaultFetchWithDelete({ + defs: [{ id: "Agents/uni", content: "role", metadata: { name: "uni" } }], + deleteStatus: 502, // the vault note delete 502s + deletes, + }); + const reg = new AgentDefRegistry(deps, { bindings: [binding], fetchFn }); + await reg.loadAll(); + expect(reg.findLiveByNote("Agents/uni")).not.toBeNull(); // live before delete. + + // The delete throws (the vault note delete failed) BEFORE any deregister. + await expect(reg.deleteDef("Agents/uni")).rejects.toThrow(/delete def Agents\/uni failed \(502\)/); + + // FIX 4 invariant: the agent is STILL registered (the in-memory def was NOT torn down + // on a failed vault delete) — it re-converges on the next poll rather than orphaning. + expect(reg.findLiveByNote("Agents/uni")).not.toBeNull(); + expect(calls.deregistered).toEqual([]); // nothing was deregistered. + expect(deletes).toEqual(["Agents/uni"]); // the delete WAS attempted (and failed). + }); + + test("FIX 4: a successful vault-delete deregisters cleanly", async () => { + const { deps, calls } = recorderDeps(); + const deletes: string[] = []; + const fetchFn = vaultFetchWithDelete({ + defs: [{ id: "Agents/uni", content: "role", metadata: { name: "uni" } }], + deleteStatus: 204, + deletes, + }); + const reg = new AgentDefRegistry(deps, { bindings: [binding], fetchFn }); + await reg.loadAll(); + + const removed = await reg.deleteDef("Agents/uni"); + expect(removed.name).toBe("uni"); + expect(removed.grantsReconciled).toBe(true); // no grants client → nothing to reconcile = ok. + // Now deregistered + removed from the live set. + expect(reg.findLiveByNote("Agents/uni")).toBeNull(); + expect(calls.deregistered).toEqual(["uni"]); + expect(deletes).toEqual(["Agents/uni"]); + }); + + test("FIX 5: a grant-reconcile failure on delete is SURFACED (grantsReconciled:false) — the note-delete still completes", async () => { + const { deps, calls } = recorderDeps(); + const reconciled: Array<{ agent: string; liveConnections: ConnectionSpec[] }> = []; + const grants = fakeGrantsClient({ reconciled, reconcileFails: true }); // reconcile POST 500s + const deletes: string[] = []; + const fetchFn = vaultFetchWithDelete({ + defs: [{ id: "Agents/uni", content: "role", metadata: { name: "uni" } }], + deleteStatus: 204, + deletes, + }); + const reg = new AgentDefRegistry(deps, { bindings: [binding], fetchFn, grants }); + await reg.loadAll(); + reconciled.length = 0; + + // The delete must NOT throw (grant GC is best-effort) but MUST report the partial + // success so the caller doesn't claim a clean full success (orphaned grants). + const removed = await reg.deleteDef("Agents/uni"); + expect(removed.name).toBe("uni"); + expect(removed.grantsReconciled).toBe(false); // FIX 5: the failure is surfaced, not swallowed. + // The note-delete + deregister STILL completed (the def IS gone). + expect(reg.findLiveByNote("Agents/uni")).toBeNull(); + expect(calls.deregistered).toEqual(["uni"]); + expect(deletes).toEqual(["Agents/uni"]); + // The reconcile WAS attempted (prune-all on the removed agent) — it just failed on the hub. + expect(reconciled).toEqual([{ agent: "uni", liveConnections: [] }]); + }); + + test("FIX 5: a SUCCESSFUL grant-reconcile on delete reports grantsReconciled:true", async () => { + const { deps } = recorderDeps(); + const reconciled: Array<{ agent: string; liveConnections: ConnectionSpec[] }> = []; + const grants = fakeGrantsClient({ reconciled }); // reconcile succeeds + const fetchFn = vaultFetchWithDelete({ + defs: [{ id: "Agents/uni", content: "role", metadata: { name: "uni" } }], + deleteStatus: 204, + }); + const reg = new AgentDefRegistry(deps, { bindings: [binding], fetchFn, grants }); + await reg.loadAll(); + const removed = await reg.deleteDef("Agents/uni"); + expect(removed.grantsReconciled).toBe(true); + }); +}); diff --git a/src/agent-defs.ts b/src/agent-defs.ts index 60fbf95..3e83994 100644 --- a/src/agent-defs.ts +++ b/src/agent-defs.ts @@ -1081,19 +1081,38 @@ export class AgentDefRegistry { this.seenDefs.set(vault, next); } - /** Reconcile a CONFIRMED-removed agent's grants away (prune ALL). Best-effort + no-op - * without a grants client / without a known name. */ - private async reconcileForRemovedAgent(name: string): Promise { - if (!this.grants || !name) return; + /** + * Reconcile a CONFIRMED-removed agent's grants away (prune ALL). Best-effort + no-op + * without a grants client / without a known name. + * + * FIX 5 (PR #3) — make the failure NON-SILENT. A hub-unreachable reconcile used to be + * caught + logged + ignored, ORPHANING the agent's approved grants on the hub (a + * re-created same-named agent resurrects them). It's still BEST-EFFORT (we don't block + * the note delete on grant cleanup — the def IS gone), but we now (a) `console.warn` + * loudly AND (b) RETURN a structured signal so the caller can surface a PARTIAL success + * (delete succeeded, grant cleanup didn't) rather than claiming a clean full success. + * `skipped` = no grants client / no name (nothing to reconcile — a true no-op). + */ + private async reconcileForRemovedAgent( + name: string, + ): Promise<{ ok: true; pruned: number } | { ok: false; error: string } | { skipped: true }> { + if (!this.grants || !name) return { skipped: true }; try { const { pruned } = await this.grants.reconcileGrants(name, []); if (pruned > 0) { console.log(`agent-defs: pruned ${pruned} stale grant(s) for removed agent "${name}".`); } + return { ok: true, pruned }; } catch (err) { + const error = (err as Error).message; + // NON-SILENT (FIX 5): a swallowed grant-GC failure orphans approved grants on the + // hub. Warn loudly + return the failure so the delete path reports partial success. console.warn( - `agent-defs: pruning grants for removed agent "${name}" failed (continuing): ${(err as Error).message}`, + `agent-defs: pruning grants for removed agent "${name}" FAILED — its approved hub ` + + `grants may be ORPHANED (re-creating a same-named agent would resurrect them); ` + + `the note delete still completed (best-effort grant cleanup): ${error}`, ); + return { ok: false, error }; } } @@ -1248,11 +1267,21 @@ export class AgentDefRegistry { /** * Delete a live def note, then deregister the agent immediately. The note MUST be a - * currently-live def we instantiated. Returns the (vault, name) of what was removed. - * Throws {@link AgentDefWriteError} on a miss or a delete failure (the deregister - * still runs even if it's already gone). + * currently-live def we instantiated. Returns the (vault, name) of what was removed, + * plus a `grantsReconciled` flag (FIX 5, PR #3) — `false` when the best-effort grant + * cleanup FAILED so the caller can report a PARTIAL success rather than a clean one. + * + * ORDERING (FIX 4, PR #3) — the VAULT NOTE DELETE happens FIRST; only after it + * SUCCEEDS do we deregister the live agent. So a vault-delete 502 throws here BEFORE + * any in-memory teardown — the def stays REGISTERED (it reappears coherently on the + * next poll), never orphaned (gone from memory but still in the vault, the confusing + * half-state). This mirrors the `agent-vaults` removal path's "persist the durable + * change first, then tear down in-memory state" discipline (daemon.ts #106). Throws + * {@link AgentDefWriteError} on a miss; a vault-delete failure throws (un-torn-down). */ - async deleteDef(noteId: string): Promise<{ vault: string; name: string }> { + async deleteDef( + noteId: string, + ): Promise<{ vault: string; name: string; grantsReconciled: boolean }> { const found = this.findLiveByNote(noteId); if (!found) { throw new AgentDefWriteError(`note ${noteId} is not a live agent definition`, 404); @@ -1261,11 +1290,15 @@ export class AgentDefRegistry { if (!client) { throw new AgentDefWriteError(`unknown def-vault "${found.vault}"`, 400); } + // STEP 1 — delete the vault note FIRST (the durable change). A non-ok (non-404) + // response throws out of here, BEFORE any deregister, so the in-memory def is left + // intact (FIX 4): no orphan, the next poll re-converges. (404 is fine — gone is gone.) await client.deleteNote(noteId); - // Tear the agent down + prune grants — the confirmed-removal path (a delete IS a - // confirmed removal); reload with "deleted" skips the (now-404) GET. - await this.reload(found.vault, noteId, "deleted"); - return { vault: found.vault, name: found.detail.name }; + // STEP 2 — the note is gone → tear the agent down + prune grants (the confirmed- + // removal path). Capture the grant-reconcile outcome to surface a partial success. + const reconcile = await this.confirmedRemoval(found.vault, noteId); + const grantsReconciled = !("ok" in reconcile) || reconcile.ok === true; + return { vault: found.vault, name: found.detail.name, grantsReconciled }; } /** @@ -1296,15 +1329,22 @@ export class AgentDefRegistry { * down AND prune ALL its grants (#96 grant-GC) so a deleted `#agent/definition` note * doesn't orphan live approved rows. The seen-set entry is cleared so a later loadAll * doesn't re-detect (and re-prune) the same removal. Reconcile is best-effort. + * + * Returns the grant-reconcile outcome (FIX 5, PR #3) so the API delete path can report + * a PARTIAL success when grant cleanup failed (delete done, grants possibly orphaned). */ - private async confirmedRemoval(vault: string, noteId: string): Promise { + private async confirmedRemoval( + vault: string, + noteId: string, + ): Promise<{ ok: true; pruned: number } | { ok: false; error: string } | { skipped: true }> { // The grant holder name comes from the live record if present, else the last-known // name we tracked for this note (a def removed before it ever instantiated). const name = this.live.get(this.keyOf(vault, noteId))?.name ?? this.seenDefs.get(vault)?.get(noteId); await this.deregisterByNote(vault, noteId); this.seenDefs.get(vault)?.delete(noteId); - if (name) await this.reconcileForRemovedAgent(name); + if (!name) return { skipped: true }; + return this.reconcileForRemovedAgent(name); } /** diff --git a/src/backends/channel-queue.test.ts b/src/backends/channel-queue.test.ts index 14e7c27..44f076d 100644 --- a/src/backends/channel-queue.test.ts +++ b/src/backends/channel-queue.test.ts @@ -16,6 +16,7 @@ import { type ChannelQueueStore, } from "./channel-queue.ts"; import type { AgentSpec } from "../sandbox/types.ts"; +import { InboundClaimConflictError } from "../transports/vault.ts"; import type { InboundQueueNote, InboundStatus } from "../transports/vault.ts"; /** @@ -44,7 +45,12 @@ class FakeStore implements ChannelQueueStore { .sort((a, b) => (a.ts < b.ts ? -1 : a.ts > b.ts ? 1 : 0)); } - async setInboundStatus(id: string, status: InboundStatus, claimedAt?: string | null): Promise { + async setInboundStatus( + id: string, + status: InboundStatus, + claimedAt?: string | null, + ifUpdatedAt?: string, + ): Promise { if (this.throwOnNextSetStatus) { const e = this.throwOnNextSetStatus; this.throwOnNextSetStatus = null; @@ -52,6 +58,17 @@ class FakeStore implements ChannelQueueStore { } const note = this.notes.get(id); if (!note) throw new Error(`fake store: no note ${id}`); + // CAS (FIX 3): when a precondition is supplied, the claim only lands if the note's + // `updatedAt` still matches what the caller last saw — else the race is lost (the + // real vault returns 409 → InboundClaimConflictError). On a successful CAS write we + // ADVANCE `updatedAt` (the vault bumps it on every write) so a second concurrent + // claimer with the now-stale precondition fails, modelling the real round-trip. + if (ifUpdatedAt !== undefined) { + if (note.updatedAt !== ifUpdatedAt) { + throw new InboundClaimConflictError(id, 409); + } + note.updatedAt = `${ifUpdatedAt}::bumped`; + } note.status = status; if (claimedAt === null) delete note.claimedAt; else if (claimedAt !== undefined) note.claimedAt = claimedAt; @@ -172,6 +189,63 @@ describe("ChannelQueueRegistry — claimNext (single-claim)", () => { await expect(reg.claimNext("laptop")).rejects.toThrow(/vault 500/); expect(store.notes.get("a")!.status).toBe("pending"); // not lost — retryable. }); + + test("FIX 3: a passed-through updatedAt is used as the CAS precondition on the claim", async () => { + const reg = new ChannelQueueRegistry(); + const store = new FakeStore(); + store.add({ ...inbound("a", "older", "2026-06-18T10:00:00Z"), updatedAt: "rev-1" }); + reg.register(specFor("laptop"), store); + const claimed = await reg.claimNext("laptop"); + expect(claimed!.id).toBe("a"); + // CAS landed → the store bumped updatedAt (modelling the vault advancing it on write). + expect(store.notes.get("a")!.status).toBe("in-flight"); + expect(store.notes.get("a")!.updatedAt).toBe("rev-1::bumped"); + }); + + test("FIX 3: a 428/409 conflict on the claim PATCH makes claimNext skip to the NEXT pending (no double-claim)", async () => { + const reg = new ChannelQueueRegistry(); + const store = new FakeStore(); + // Two pending notes, each with a known revision for the CAS precondition. + store.add({ ...inbound("a", "older", "2026-06-18T10:00:00Z"), updatedAt: "rev-a" }); + store.add({ ...inbound("b", "newer", "2026-06-18T10:05:00Z"), updatedAt: "rev-b" }); + reg.register(specFor("laptop"), store); + + // Simulate a CONCURRENT winner: between claimNext's list and its PATCH of "a", + // another session claims "a" and advances its revision. The next PATCH of "a" with the + // now-stale precondition will throw InboundClaimConflictError → re-list → claim "b". + const realSet = store.setInboundStatus.bind(store); + let firstPatch = true; + store.setInboundStatus = (async (id, status, claimedAt, ifUpdatedAt) => { + if (firstPatch && id === "a") { + firstPatch = false; + // The "other session" already claimed "a" (its revision moved on). + store.notes.get("a")!.updatedAt = "rev-a-claimed-by-someone-else"; + store.notes.get("a")!.status = "in-flight"; + } + return realSet(id, status, claimedAt, ifUpdatedAt); + }) as typeof store.setInboundStatus; + + const claimed = await reg.claimNext("laptop"); + // The conflict on "a" was caught + re-listed; we claimed "b" instead — never "a" twice. + expect(claimed!.id).toBe("b"); + expect(store.notes.get("b")!.status).toBe("in-flight"); + }); + + test("FIX 3: a conflict with NO other pending returns null (nothing claimable right now)", async () => { + const reg = new ChannelQueueRegistry(); + const store = new FakeStore(); + store.add({ ...inbound("a", "only", "2026-06-18T10:00:00Z"), updatedAt: "rev-a" }); + reg.register(specFor("laptop"), store); + // Every CAS on "a" loses (the precondition is always stale → conflict). After the + // conflict, "a" is left in-flight (by the simulated winner), so the re-list finds no + // pending and returns null — not a double-claim, not an error. + store.setInboundStatus = (async (id: string) => { + store.notes.get(id)!.status = "in-flight"; + throw new InboundClaimConflictError(id, 409); + }) as typeof store.setInboundStatus; + const claimed = await reg.claimNext("laptop"); + expect(claimed).toBeNull(); + }); }); describe("ChannelQueueRegistry — reply (outbound + mark handled)", () => { diff --git a/src/backends/channel-queue.ts b/src/backends/channel-queue.ts index 268387d..6e13b75 100644 --- a/src/backends/channel-queue.ts +++ b/src/backends/channel-queue.ts @@ -39,6 +39,7 @@ */ import type { AgentSpec } from "../sandbox/types.ts"; +import { InboundClaimConflictError } from "../transports/vault.ts"; import type { InboundQueueNote, InboundStatus } from "../transports/vault.ts"; /** @@ -49,12 +50,28 @@ import type { InboundQueueNote, InboundStatus } from "../transports/vault.ts"; export interface ChannelQueueStore { /** List this channel's inbound queue notes, ascending by ts (oldest first). */ listInboundQueue(opts?: { limit?: number }): Promise; - /** Set an inbound note's claim status (+ optionally claimedAt; `null` clears it). */ - setInboundStatus(id: string, status: InboundStatus, claimedAt?: string | null): Promise; + /** + * Set an inbound note's claim status (+ optionally claimedAt; `null` clears it). + * When `ifUpdatedAt` is given, the write is a COMPARE-AND-SWAP (the claim only lands + * if the note hasn't changed since it was read) and throws {@link + * InboundClaimConflictError} when the race is lost (agent#101); omitting it is the + * prior last-write-wins behavior (release / handled / sweep). + */ + setInboundStatus( + id: string, + status: InboundStatus, + claimedAt?: string | null, + ifUpdatedAt?: string, + ): Promise; /** Write an outbound reply (the SAME `#agent/message/outbound` path the worker uses). */ reply(args: { text: string; inReplyTo?: string }): Promise<{ sent: string[] }>; } +/** Bound on the CAS re-list retries in {@link ChannelQueueRegistry.claimNext} — a + * safety net against a pathological all-contended queue (each pass claims/eliminates + * one note, so the loop is naturally bounded by the pending count anyway). */ +const MAX_CLAIM_ATTEMPTS = 25; + /** The default in-flight claim TTL (design: 15 min comfortably covers an operator turn). */ export const DEFAULT_CLAIM_TTL_MS = 15 * 60 * 1000; @@ -205,32 +222,55 @@ export class ChannelQueueRegistry { * PATCH lands the note is no longer `pending`. Returns null when none pending (or for * an unregistered channel). * - * CLAIM-RACE SCOPE (honest): the claim PATCH is `force:true` (last-write-wins, no - * precondition), so TWO TRULY-CONCURRENT `claimNext` calls (e.g. the same channel - * connected from two sessions, both listing before either PATCHes) can both return the - * SAME note → a double-handle (two replies). The `channel` model is one-operator- - * session-at-a-time, so this is narrow; a double-handle is non-corrupting (a duplicate - * reply) and the TTL sweep can't even strand it. Hardening to a compare-and-swap claim - * (`if_updated_at` + re-list on 428) for the multi-session case is tracked as a - * follow-up (agent#101). Don't claim race-safety here that the `force:true` PATCH - * doesn't provide. + * SINGLE-CLAIM via COMPARE-AND-SWAP (agent#101). The claim PATCH carries + * `if_updated_at` (the note's last-seen `updated_at`), so it only lands if the note + * hasn't changed since this call read it. Two truly-concurrent `claimNext` calls read + * the SAME `updated_at`; the first claim advances it, so the second's precondition + * FAILS (the store throws {@link InboundClaimConflictError}) — we then RE-LIST and try + * the NEXT pending message, never double-claiming. The loop is bounded by the pending + * count (each pass either claims one or eliminates a now-contended one) plus a hard + * {@link MAX_CLAIM_ATTEMPTS} backstop. A note with no `updatedAt` (a vault that omitted + * it) falls back to the prior last-write-wins claim — the precondition is simply + * absent, so the narrow double-claim window remains only for that degenerate case. + * Returns null when none pending (or for an unregistered channel). */ async claimNext(channel: string, now: () => Date = () => new Date()): Promise { const rec = this.byChannel.get(channel); if (!rec) return null; - const notes = await rec.store.listInboundQueue(); - const oldest = notes.find((n) => n.status === "pending"); // listInboundQueue is ascending by ts. - if (!oldest) return null; - // Commit the claim (status → in-flight + claimedAt) BEFORE returning — the flip is - // the single-claim guarantee. If the PATCH throws, we don't return the note (the - // caller gets the error; the note stays pending for a retry). - await rec.store.setInboundStatus(oldest.id, "in-flight", now().toISOString()); - return { - id: oldest.id, - text: oldest.text, - inReplyTo: oldest.id, - systemPrompt: rec.spec.systemPrompt ?? "", - }; + for (let attempt = 0; attempt < MAX_CLAIM_ATTEMPTS; attempt++) { + // RE-LIST each attempt so a lost CAS race sees the queue as the winner left it + // (the contended note is now in-flight, so `find` skips past it to the next pending). + const notes = await rec.store.listInboundQueue(); + const oldest = notes.find((n) => n.status === "pending"); // ascending by ts. + if (!oldest) return null; + try { + // Commit the claim (status → in-flight + claimedAt) BEFORE returning, guarded by + // the note's `updated_at` so a concurrent claimer can't also win it. A non-conflict + // PATCH failure propagates (the caller gets the error; the note stays pending). + await rec.store.setInboundStatus( + oldest.id, + "in-flight", + now().toISOString(), + oldest.updatedAt, // CAS precondition; undefined → store falls back to force. + ); + } catch (err) { + if (err instanceof InboundClaimConflictError) { + // Another session claimed this note between our list and PATCH — re-list and + // try the next pending one (no double-claim). + continue; + } + throw err; + } + return { + id: oldest.id, + text: oldest.text, + inReplyTo: oldest.id, + systemPrompt: rec.spec.systemPrompt ?? "", + }; + } + // Exhausted the retry budget under sustained contention — treat as "none claimable + // right now" (a connected session retries `next-message`). Non-corrupting. + return null; } /** diff --git a/src/backends/programmatic.ts b/src/backends/programmatic.ts index 1fb0f8f..d3f8274 100644 --- a/src/backends/programmatic.ts +++ b/src/backends/programmatic.ts @@ -305,6 +305,19 @@ export class ProgrammaticBackend implements AgentBackend { // Mint the VAULT token only — no channel MCP in this backend (the daemon // mediates messaging). A spec with no vault gets an EMPTY mcpServers config // (the agent still runs; it just has no vault tools this turn). + // + // FIX 2 (PR #3) — mid-turn token expiry, ASSESSED + DEFERRED (no re-mint added). + // The vault write token is MINTED FRESH per turn here (no `expiresIn` override → the + // hub default ~90d non-ephemeral TTL), so it CANNOT expire during a single `claude -p` + // turn (which lasts minutes). And the vault WRITES are made by the OPAQUE `claude -p` + // subprocess via the token baked into its 0600 `.mcp.json` (below) — the backend has + // NO in-process seam to observe a 401 from those writes and re-inject a new token + // mid-turn. A re-mint-on-401 would require the MCP-client-in-subprocess to surface + // 401s back here, which the architecture doesn't provide. So a re-mint is INFEASIBLE + // (and unnecessary given the fresh-per-turn ~90d mint). If a future long-running / + // multi-day single turn or a short operator-pinned TTL ever makes mid-turn expiry + // real, the fix is at the MCP-client layer (refresh-on-401), tracked as a follow-up + // — NOT a forced backend re-mint that can't see the failure. let vaultArg: { url: string; entry: { name: string; token: string } } | undefined; if (spec.vault) { const v = spec.vault; diff --git a/src/backends/registry.test.ts b/src/backends/registry.test.ts index 67556aa..e458c05 100644 --- a/src/backends/registry.test.ts +++ b/src/backends/registry.test.ts @@ -12,6 +12,8 @@ import { describe, test, expect } from "bun:test"; import { ProgrammaticAgentRegistry, + OUTBOUND_MAX_RETRIES, + isTransientOutboundError, type WriteOutbound, type WriteThread, type ThreadNote, @@ -397,37 +399,148 @@ describe("ProgrammaticAgentRegistry — #agent/thread notes (unified lifecycle, expect(rec.calls).toHaveLength(0); }); - test("REGRESSION (c34db03, now BOTH modes): a turn whose outbound write THROWS still leaves exactly one #agent/thread note", async () => { + test("REGRESSION (c34db03, now BOTH modes): a turn whose outbound write THROWS still leaves a primary #agent/thread note (now re-recorded as error — FIX 1)", async () => { const backend = new FakeBackend(); const threads = threadRecorder(); // A THROWING WriteOutbound — the thread note is written BEFORE the additive outbound // (c34db03, now applied uniformly to BOTH modes), so the failed transcript write must // NOT cost us the primary record. Use a SINGLE-THREADED spec to prove the c34db03 // ordering now protects single-threaded too. (`recorder()` can't throw; inline variant.) + // The error message carries NO HTTP status → classified TRANSIENT → it RETRIES the + // bounded budget (FIX 1, PR #3) before giving up, then re-records the thread as error. let outboundAttempts = 0; const throwingWriteOutbound: WriteOutbound = async () => { outboundAttempts++; - throw new Error("vault write boom"); + throw new Error("vault write boom"); // no (NNN) status → transient → retried. }; const reg = new ProgrammaticAgentRegistry({ backend, writeOutbound: throwingWriteOutbound, writeThread: threads.fn, + outboundRetryBaseMs: 0, }); await reg.register(specFor("eng")); // single-threaded (default) — the ordering applies here now. reg.enqueue("eng", { content: "fire it" }); - await until(() => threads.threads.length === 1); - // Let the (throwing) outbound attempt settle. + // FIX 1: the primary `ok` thread note is written first, then after retries exhaust, a + // second `error` thread note re-records the UN-DELIVERED reply. + await until(() => threads.threads.length === 2); + + // First (optimistic) record was `ok`; the second re-records the failure so the durable + // thread record does NOT falsely claim the reply landed. + expect(threads.threads[0]!.status).toBe("ok"); + expect(threads.threads[0]!.output).toBe("reply:fire it"); + expect(threads.threads[1]!.status).toBe("error"); + expect(threads.threads[1]!.mode).toBe("single-threaded"); + // The undelivered reply text is preserved in the error record for recovery. + expect(threads.threads[1]!.output).toContain("reply:fire it"); + // Transient → the outbound was retried the full budget (1 initial + OUTBOUND_MAX_RETRIES). + expect(outboundAttempts).toBe(1 + OUTBOUND_MAX_RETRIES); + }); +}); + +describe("ProgrammaticAgentRegistry — outbound retry on transient failure (FIX 1, PR #3)", () => { + test("isTransientOutboundError: 5xx + network = transient; 4xx = permanent", () => { + expect(isTransientOutboundError(new Error("write reply failed (502) boom"))).toBe(true); + expect(isTransientOutboundError(new Error("write reply failed (503)"))).toBe(true); + expect(isTransientOutboundError(new Error("ECONNREFUSED"))).toBe(true); // no status → network. + expect(isTransientOutboundError(new Error("fetch failed"))).toBe(true); + expect(isTransientOutboundError(new Error("write reply failed (400) bad"))).toBe(false); + expect(isTransientOutboundError(new Error("write reply failed (401)"))).toBe(false); + expect(isTransientOutboundError(new Error("write reply failed (409)"))).toBe(false); + }); + + test("a transient-then-success outbound RETRIES and the reply LANDS (no loss, turn not re-run)", async () => { + const backend = new FakeBackend(); + const threads = threadRecorder(); + // Fail twice with a transient (5xx) error, then succeed — the retry must land the reply. + let attempts = 0; + const recorded: { reply: string }[] = []; + const flakyWriteOutbound: WriteOutbound = async (_channel, reply) => { + attempts++; + if (attempts <= 2) throw new Error("vault transport: write reply failed (502) blip"); + recorded.push({ reply }); + }; + const reg = new ProgrammaticAgentRegistry({ + backend, + writeOutbound: flakyWriteOutbound, + writeThread: threads.fn, + outboundRetryBaseMs: 0, + }); + await reg.register(specFor("eng")); + + reg.enqueue("eng", { content: "important" }); + await until(() => recorded.length === 1); await new Promise((r) => setTimeout(r, 5)); - // The thread note survived the outbound failure: exactly ONE, status ok, output = the reply. + // The reply landed on the 3rd attempt (1 initial + 2 retries == OUTBOUND_MAX_RETRIES). + expect(attempts).toBe(1 + OUTBOUND_MAX_RETRIES); + expect(recorded).toEqual([{ reply: "reply:important" }]); + // The backend ran the turn EXACTLY ONCE (no re-run / fork on the retry). + expect(backend.calls).toHaveLength(1); + // The thread note is the single `ok` record (the reply was ultimately delivered) — no + // error re-record because delivery succeeded. expect(threads.threads).toHaveLength(1); expect(threads.threads[0]!.status).toBe("ok"); - expect(threads.threads[0]!.mode).toBe("single-threaded"); - expect(threads.threads[0]!.output).toBe("reply:fire it"); - // The outbound WAS attempted (and threw) — proving the thread note was written first. - expect(outboundAttempts).toBe(1); + }); + + test("a PERSISTENT failure surfaces an error event + re-records the thread as error + does NOT claim success", async () => { + const backend = new FakeBackend(); + const threads = threadRecorder(); + const turn = turnRecorder(); + let attempts = 0; + const alwaysFail: WriteOutbound = async () => { + attempts++; + throw new Error("vault transport: write reply failed (503) down"); + }; + const reg = new ProgrammaticAgentRegistry({ + backend, + writeOutbound: alwaysFail, + writeThread: threads.fn, + onTurnEvent: turn.fn, + outboundRetryBaseMs: 0, + }); + await reg.register(specFor("eng")); + + reg.enqueue("eng", { content: "doomed" }); + await until(() => threads.threads.length === 2); + await new Promise((r) => setTimeout(r, 5)); + + // Retried the full budget then gave up (1 + OUTBOUND_MAX_RETRIES). + expect(attempts).toBe(1 + OUTBOUND_MAX_RETRIES); + // The live view resolved to ERROR (not `done`) — no silently-vanished reply. + const errorEvents = turn.events.filter((e) => e.event.kind === "error"); + expect(errorEvents.length).toBeGreaterThanOrEqual(1); + expect(turn.events.some((e) => e.event.kind === "done")).toBe(false); + // The thread record does NOT falsely claim a clean ok: the final record is error, + // carrying the un-delivered reply text for recovery. + expect(threads.threads[1]!.status).toBe("error"); + expect(threads.threads[1]!.output).toContain("reply:doomed"); + }); + + test("a PERMANENT (4xx) outbound failure does NOT retry — gives up immediately", async () => { + const backend = new FakeBackend(); + const threads = threadRecorder(); + let attempts = 0; + const reject4xx: WriteOutbound = async () => { + attempts++; + throw new Error("vault transport: write reply failed (400) bad request"); + }; + const reg = new ProgrammaticAgentRegistry({ + backend, + writeOutbound: reject4xx, + writeThread: threads.fn, + outboundRetryBaseMs: 0, + }); + await reg.register(specFor("eng")); + + reg.enqueue("eng", { content: "rejected" }); + await until(() => threads.threads.length === 2); + await new Promise((r) => setTimeout(r, 5)); + + // A 4xx is a real rejection → exactly ONE attempt, no retry. + expect(attempts).toBe(1); + expect(threads.threads[1]!.status).toBe("error"); }); }); diff --git a/src/backends/registry.ts b/src/backends/registry.ts index db582cb..32857b7 100644 --- a/src/backends/registry.ts +++ b/src/backends/registry.ts @@ -126,6 +126,37 @@ export interface ThreadNote { */ export type WriteThread = (thread: ThreadNote) => Promise; +/** How many times the outbound write is RETRIED on a transient failure (agent — PR #3 + * FIX 1) before giving up. Total attempts = 1 + this. */ +export const OUTBOUND_MAX_RETRIES = 2; +/** Base backoff (ms) between outbound retries — grows linearly (attempt 1 → BASE, 2 → 2×BASE). */ +export const OUTBOUND_RETRY_BASE_MS = 250; + +/** + * Classify an outbound-write error as TRANSIENT (worth retrying) vs PERMANENT (a real + * rejection). The VaultTransport's `reply()` throws `Error` whose message embeds the + * HTTP status as `(NNN)` for a non-ok vault response, or a raw network/fetch rejection + * (no status) when the vault is unreachable. So: + * - a parseable 5xx (502/503/504/…) → TRANSIENT (a vault blip; retry). + * - NO parseable status (a network error, DNS, connection refused) → TRANSIENT. + * - a parseable 4xx (400/401/403/409/…) → PERMANENT (a real rejection — auth, bad + * request; retrying just re-fails). Do NOT retry these. + * This keeps the retry to the case the audit flagged (a transient vault 5xx silently + * losing the reply) without papering over a genuine 4xx rejection. + */ +export function isTransientOutboundError(err: unknown): boolean { + const msg = (err as Error)?.message ?? ""; + const m = msg.match(/\((\d{3})\)/); + if (!m) return true; // no HTTP status → a network/connection error → transient. + const status = Number(m[1]); + return status >= 500 && status <= 599; // 5xx transient; 4xx permanent. +} + +/** Sleep helper for the outbound retry backoff (injectable-free; small + bounded). */ +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + /** A queued inbound message awaiting its serial turn. */ interface QueuedMessage { /** The inbound text handed to the `claude -p` turn as the prompt. */ @@ -177,17 +208,22 @@ export class ProgrammaticAgentRegistry { private readonly writeThread?: WriteThread; /** Optional streaming-view sink — push interim + lifecycle turn events per channel. */ private readonly onTurnEvent?: TurnEventSink; + /** Base backoff (ms) between outbound retries (FIX 1). Injectable so tests run fast. */ + private readonly outboundRetryBaseMs: number; constructor(deps: { backend: AgentBackend; writeOutbound: WriteOutbound; writeThread?: WriteThread; onTurnEvent?: TurnEventSink; + /** Override the outbound-retry backoff base (ms). Default {@link OUTBOUND_RETRY_BASE_MS}. */ + outboundRetryBaseMs?: number; }) { this.backend = deps.backend; this.writeOutbound = deps.writeOutbound; if (deps.writeThread) this.writeThread = deps.writeThread; if (deps.onTurnEvent) this.onTurnEvent = deps.onTurnEvent; + this.outboundRetryBaseMs = deps.outboundRetryBaseMs ?? OUTBOUND_RETRY_BASE_MS; } /** @@ -451,19 +487,36 @@ export class ProgrammaticAgentRegistry { // Empty reply → NO note (reviewer contract — `reply` can be ""): a turn that produced // no text (e.g. tool-only work) leaves the chat clean. if (result.reply && result.reply.length > 0) { - try { - await this.writeOutbound(channel, result.reply, msg.inReplyTo); - } catch (err) { - const reason = (err as Error).message; + const delivered = await this.deliverOutboundWithRetry(channel, result.reply, msg.inReplyTo); + if (!delivered.ok) { + // FIX 1 (PR #3) — the SCARY one. The reply was PRODUCED but, after the bounded + // retry, still NOT persisted to the transcript (a persistent vault 5xx / network + // fault, or a real 4xx rejection). We must NOT leave a clean `status:ok` record + // claiming the reply landed when it didn't: + // 1. RE-RECORD the thread note as `status:error` so the durable thread record + // reflects the UN-DELIVERED reply (overwrites the optimistic `ok` upsert for + // single-threaded; writes/overwrites the per-fire note for multi-threaded). + // 2. Resolve the live view to ERROR (not `done`) so the UI doesn't drop the + // in-progress bubble + poll for a note that isn't there (PR #83 nit). + // We do NOT re-run the `claude -p` turn (that forks/burns quota) — the reply text + // is preserved IN the error thread note's output for an operator to recover. console.error( - `parachute-agent: programmatic outbound write for channel "${channel}" failed: ${reason}`, + `parachute-agent: programmatic outbound write for channel "${channel}" failed ` + + `after ${OUTBOUND_MAX_RETRIES} retries: ${delivered.error}`, + ); + await this.recordThread( + handle, + msg, + "error", + `reply produced but NOT delivered (outbound write failed: ${delivered.error}). ` + + `Undelivered reply text: ${result.reply}`, + startedAt, + result.usage, ); - // The reply was produced but NOT persisted to the transcript. Resolve the live - // view to ERROR, not `done` — a `done` would drop the in-progress bubble and - // trigger a poll that finds no note, leaving the user with a silently vanished - // reply. (reviewer nit, PR #83.) The thread note above already captured the reply - // durably (BOTH modes), so the turn's record is not lost. - this.emitTurnEvent(channel, { kind: "error", error: `reply produced but not saved: ${reason}` }); + this.emitTurnEvent(channel, { + kind: "error", + error: `reply produced but not saved: ${delivered.error}`, + }); continue; } } @@ -518,4 +571,53 @@ export class ProgrammaticAgentRegistry { ); } } + + /** + * Deliver the outbound reply with a BOUNDED retry on a TRANSIENT failure (FIX 1, PR + * #3). A vault 5xx / network blip during the outbound write used to silently lose the + * reply (the turn resolved, the thread note said "ok", but the chat bubble never + * landed). We retry up to {@link OUTBOUND_MAX_RETRIES} times with a small linear + * backoff on a transient error ({@link isTransientOutboundError}: a 5xx or a + * no-status network error). A PERMANENT error (a 4xx — a real rejection) does NOT + * retry. Returns `{ ok: true }` once the write lands, or `{ ok: false, error }` after + * exhausting the retries / on a permanent failure — the caller then records the turn + * as un-delivered + surfaces it (never claims a clean success). We NEVER re-run the + * `claude -p` turn here (that would fork the conversation / burn quota); only the + * idempotent outbound WRITE is retried. + */ + private async deliverOutboundWithRetry( + channel: string, + reply: string, + inReplyTo?: string, + ): Promise<{ ok: true } | { ok: false; error: string }> { + let lastError = ""; + for (let attempt = 0; attempt <= OUTBOUND_MAX_RETRIES; attempt++) { + try { + await this.writeOutbound(channel, reply, inReplyTo); + return { ok: true }; + } catch (err) { + lastError = (err as Error).message; + const transient = isTransientOutboundError(err); + const more = attempt < OUTBOUND_MAX_RETRIES; + if (!transient || !more) { + // A permanent (4xx) error never retries; a transient one that exhausted the + // budget falls through to the failure return below. + if (!transient) { + console.warn( + `parachute-agent: outbound write for channel "${channel}" failed with a ` + + `non-transient error (not retrying): ${lastError}`, + ); + } + return { ok: false, error: lastError }; + } + // Transient + retries remain — back off (linear) and try again. + console.warn( + `parachute-agent: outbound write for channel "${channel}" transient failure ` + + `(attempt ${attempt + 1}/${OUTBOUND_MAX_RETRIES + 1}), retrying: ${lastError}`, + ); + await delay(this.outboundRetryBaseMs * (attempt + 1)); + } + } + return { ok: false, error: lastError }; + } } diff --git a/src/daemon.ts b/src/daemon.ts index 8876a30..f669db7 100755 --- a/src/daemon.ts +++ b/src/daemon.ts @@ -1966,6 +1966,16 @@ export function createFetchHandler( if (req.method === "DELETE") { try { const removed = await agentDefs.deleteDef(noteId); + // FIX 5 (PR #3) — surface a PARTIAL success: the note delete completed, but if + // best-effort grant cleanup failed, say so (the agent's approved hub grants may + // be orphaned) rather than reporting a clean full success. The delete itself is + // still a 200 (the def IS gone — grant GC is best-effort, not delete-blocking). + if (!removed.grantsReconciled) { + console.warn( + `parachute-agent: deleted agent def "${removed.name}" but grant cleanup failed — ` + + `its approved hub grants may be orphaned.`, + ); + } return json({ ok: true, ...removed, removed: true }); } catch (err) { if (err instanceof AgentDefWriteError) return json({ error: err.message }, err.status); diff --git a/src/transports/vault.test.ts b/src/transports/vault.test.ts index 5ac6173..2d16a71 100644 --- a/src/transports/vault.test.ts +++ b/src/transports/vault.test.ts @@ -26,7 +26,7 @@ */ import { describe, test, expect, afterEach } from "bun:test"; -import { VaultTransport, AGENT_VAULT_TAG_SCHEMA, AGENT_THREAD_TAG } from "./vault.ts"; +import { VaultTransport, AGENT_VAULT_TAG_SCHEMA, AGENT_THREAD_TAG, InboundClaimConflictError } from "./vault.ts"; import type { TransportContext, InboundMessage } from "../transport.ts"; import { instantiateTransport } from "../registry.ts"; @@ -1301,3 +1301,145 @@ describe("VaultTransport — scheduled-job notes (vault-native store)", () => { await expect(t.deleteJobNote("job-1")).rejects.toThrow(/delete job failed \(404\)/); }); }); + +// --------------------------------------------------------------------------- +// Channel-queue inbound notes — FIX 3 (CAS claim) + FIX 6 (handled exclusion). +// --------------------------------------------------------------------------- + +describe("VaultTransport — listInboundQueue", () => { + test("FIX 6: EXCLUDES handled notes so pending is never crowded out past the cap", async () => { + // The vault returns many `handled` notes plus one still-`pending` note. The handled + // ones must be dropped client-side so the pending one is always in the returned queue. + const handled = Array.from({ length: 50 }, (_, i) => ({ + id: `h${i}`, + content: `handled ${i}`, + metadata: { channel: "eng", direction: "inbound", sender: "operator", ts: `2026-01-01T00:${String(i).padStart(2, "0")}:00Z`, status: "handled" }, + updated_at: `2026-01-01T01:00:00Z`, + })); + const pending = { + id: "p1", + content: "still pending", + metadata: { channel: "eng", direction: "inbound", sender: "operator", ts: "2026-01-02T00:00:00Z", status: "pending" }, + updated_at: "2026-01-02T00:00:00Z", + }; + let listUrl = ""; + globalThis.fetch = (async (url: string | URL | Request) => { + const u = String(url); + // start() fires ensureSchema PUTs (.../api/tags/*); only capture the list GET. + if (u.includes("/api/notes?")) { + listUrl = u; + return new Response(JSON.stringify([...handled, pending]), { + status: 200, + headers: { "content-type": "application/json" }, + }); + } + return new Response(null, { status: 200 }); + }) as typeof fetch; + + const t = new VaultTransport(baseConfig()); + await t.start(fakeCtx("eng")); + const queue = await t.listInboundQueue(); + // No handled notes survive; the pending one IS present. + expect(queue.every((n) => n.status !== "handled")).toBe(true); + expect(queue.map((n) => n.id)).toEqual(["p1"]); + expect(queue[0]!.status).toBe("pending"); + // The list request asks the vault NEWEST-first (so a hard cap drops the oldest + // handled notes, never a recent pending). + expect(listUrl).toContain("sort=desc"); + }); + + test("FIX 6: in-flight notes are KEPT (only handled is excluded)", async () => { + globalThis.fetch = (async () => + new Response( + JSON.stringify([ + { id: "a", content: "p", metadata: { channel: "eng", ts: "t1", status: "pending" }, updated_at: "u1" }, + { id: "b", content: "f", metadata: { channel: "eng", ts: "t2", status: "in-flight", claimedAt: "c2" }, updated_at: "u2" }, + { id: "c", content: "h", metadata: { channel: "eng", ts: "t3", status: "handled" }, updated_at: "u3" }, + ]), + { status: 200, headers: { "content-type": "application/json" } }, + )) as unknown as typeof fetch; + const t = new VaultTransport(baseConfig()); + await t.start(fakeCtx("eng")); + const queue = await t.listInboundQueue(); + expect(queue.map((n) => n.id)).toEqual(["a", "b"]); + expect(queue.find((n) => n.id === "b")!.status).toBe("in-flight"); + }); + + test("FIX 3: threads the note's updated_at through as updatedAt (the CAS precondition)", async () => { + globalThis.fetch = (async () => + new Response( + JSON.stringify([ + { id: "n1", content: "hi", metadata: { channel: "eng", ts: "t1", status: "pending" }, updated_at: "2026-06-01T00:00:00Z" }, + ]), + { status: 200, headers: { "content-type": "application/json" } }, + )) as unknown as typeof fetch; + const t = new VaultTransport(baseConfig()); + await t.start(fakeCtx("eng")); + const queue = await t.listInboundQueue(); + expect(queue[0]!.updatedAt).toBe("2026-06-01T00:00:00Z"); + }); +}); + +describe("VaultTransport — setInboundStatus (FIX 3 compare-and-swap claim)", () => { + test("with ifUpdatedAt: sends if_updated_at (NOT force) as the precondition", async () => { + let body: any; + globalThis.fetch = (async (_url: string | URL | Request, init?: RequestInit) => { + body = JSON.parse(String(init?.body)); + return new Response(null, { status: 200 }); + }) as typeof fetch; + const t = new VaultTransport(baseConfig()); + await t.setInboundStatus("n1", "in-flight", "2026-06-01T00:00:01Z", "2026-06-01T00:00:00Z"); + expect(body.if_updated_at).toBe("2026-06-01T00:00:00Z"); + expect(body.force).toBeUndefined(); + expect(body.metadata.status).toBe("in-flight"); + expect(body.metadata.claimedAt).toBe("2026-06-01T00:00:01Z"); + }); + + test("without ifUpdatedAt: keeps the last-write-wins force:true (release/handled/sweep)", async () => { + let body: any; + globalThis.fetch = (async (_url: string | URL | Request, init?: RequestInit) => { + body = JSON.parse(String(init?.body)); + return new Response(null, { status: 200 }); + }) as typeof fetch; + const t = new VaultTransport(baseConfig()); + await t.setInboundStatus("n1", "handled", null); + expect(body.force).toBe(true); + expect(body.if_updated_at).toBeUndefined(); + }); + + test("a 409 (stale precondition) on a CAS write throws InboundClaimConflictError", async () => { + globalThis.fetch = (async () => + new Response(JSON.stringify({ error_type: "conflict" }), { status: 409 })) as unknown as typeof fetch; + const t = new VaultTransport(baseConfig()); + await expect( + t.setInboundStatus("n1", "in-flight", "now", "stale-updated-at"), + ).rejects.toBeInstanceOf(InboundClaimConflictError); + }); + + test("a 428 (precondition required) on a CAS write also throws InboundClaimConflictError", async () => { + globalThis.fetch = (async () => + new Response(JSON.stringify({ error: "precondition_required" }), { status: 428 })) as unknown as typeof fetch; + const t = new VaultTransport(baseConfig()); + await expect( + t.setInboundStatus("n1", "in-flight", "now", "some-updated-at"), + ).rejects.toBeInstanceOf(InboundClaimConflictError); + }); + + test("a 409 on a NON-CAS write (no ifUpdatedAt) throws a plain Error, not a conflict", async () => { + globalThis.fetch = (async () => + new Response("conflict", { status: 409 })) as unknown as typeof fetch; + const t = new VaultTransport(baseConfig()); + const err = await t.setInboundStatus("n1", "handled", null).catch((e) => e); + expect(err).toBeInstanceOf(Error); + expect(err).not.toBeInstanceOf(InboundClaimConflictError); + }); + + test("a 500 on a CAS write throws a plain Error (a real failure, not a lost race)", async () => { + globalThis.fetch = (async () => + new Response("boom", { status: 500 })) as unknown as typeof fetch; + const t = new VaultTransport(baseConfig()); + const err = await t.setInboundStatus("n1", "in-flight", "now", "u1").catch((e) => e); + expect(err).toBeInstanceOf(Error); + expect(err).not.toBeInstanceOf(InboundClaimConflictError); + }); +}); diff --git a/src/transports/vault.ts b/src/transports/vault.ts index 330ce17..f6ad3d9 100644 --- a/src/transports/vault.ts +++ b/src/transports/vault.ts @@ -208,10 +208,38 @@ export interface InboundQueueNote { status: InboundStatus; /** ISO timestamp the note was claimed (set with `in-flight`); used by the TTL sweep. */ claimedAt?: string; + /** + * The note's vault `updated_at` (the last-seen revision). Threaded through so a + * claim can use it as the `if_updated_at` compare-and-swap precondition (agent#101): + * two concurrent `claimNext` reads see the SAME `updated_at`; the first claim PATCH + * advances it, so the second's precondition fails (vault 409) and it re-lists rather + * than double-claiming. Absent when the vault response omitted it. + */ + updatedAt?: string; } const DEFAULT_VAULT_URL = "http://127.0.0.1:1940"; const DEFAULT_PATH_PREFIX = "channel"; + +/** + * Thrown by {@link VaultTransport.setInboundStatus} when a compare-and-swap claim + * (an `ifUpdatedAt` precondition) FAILED — the note changed since it was read, so + * another writer won the race (agent#101). The vault returns **409** (`error_type: + * "conflict"`) for a STALE `if_updated_at`, and **428** (`precondition_required`) when + * the precondition is absent; we treat both as "lost the claim race" so the caller + * (the channel queue's `claimNext`) re-lists and tries the next pending message rather + * than double-claiming. Distinct from a generic write error (any other non-ok status), + * which still throws a plain Error. + */ +export class InboundClaimConflictError extends Error { + constructor( + readonly id: string, + readonly status: number, + ) { + super(`vault transport: inbound claim ${id} lost the CAS race (${status})`); + this.name = "InboundClaimConflictError"; + } +} /** Parent tag (NEW, namespaced) — carried LITERALLY on every note WE write; query * this + metadata.channel to see BOTH directions of a channel (the slash children * are namespace, not inheritance). */ @@ -1182,13 +1210,26 @@ export class VaultTransport implements Transport { /** * List THIS channel's INBOUND queue notes (the `#agent/message/inbound` notes), - * ascending by `ts` (oldest first), carrying the claim `status`/`claimedAt`. The - * query is index-free, mirroring {@link loadTranscript}: query by the inbound + * ascending by `ts` (oldest first), carrying the claim `status`/`claimedAt`/`updatedAt`. + * The query is index-free, mirroring {@link loadTranscript}: query by the inbound * CHILD tag (we want inbound only — outbound replies are not queue items) and * filter to this channel CLIENT-SIDE on `metadata.channel` (we don't assume a * `channel` index). A note with NO `status` field reads as `pending` (a fresh * inbound the trigger just created). Throws on a non-ok vault response so the * caller surfaces a clear error rather than a silently-empty queue. + * + * QUEUE-CAP TRUNCATION FIX (agent#103). Over time `handled` notes accumulate; the + * tag query is capped (the vault limit), so once enough `handled` notes precede the + * still-`pending` ones, the pending notes fall OUTSIDE the cap and are never claimed + * (a silently-stuck queue). The vault's `status` metadata isn't indexed (we can't + * assume a per-vault schema), so we can't filter `status:pending` server-side. So we + * EXCLUDE `handled` notes CLIENT-SIDE — the live queue is only `pending` + `in-flight` + * — and additionally REQUEST the cap descending (newest first) so when the raw note + * count itself exceeds the cap, it's the OLDEST `handled` notes that get dropped, never + * a recent `pending`. The two together keep the actionable queue (pending + in-flight) + * intact regardless of how many `handled` notes have piled up. (Declaring `status` + * indexed for a true server-side `status != handled` filter is a future scale + * optimization, not a correctness requirement.) */ async listInboundQueue(opts?: { limit?: number }): Promise { const channel = this.channel; @@ -1199,13 +1240,23 @@ export class VaultTransport implements Transport { params.set("tag", AGENT_MESSAGE_INBOUND_TAG); // → %23agent%2Fmessage%2Finbound params.set("include_content", "true"); params.set("limit", String(fetchLimit)); + // NEWEST-first at the vault (default order_by is `updated_at`) so a hard cap drops + // the OLDEST notes (the long-settled `handled` ones), never a recent pending. We + // re-sort ascending below for the queue. The vault param is `sort` (asc|desc). + params.set("sort", "desc"); const url = `${this.vaultUrl}/vault/${this.vault}/api/notes?${params.toString()}`; const res = await fetch(url, { headers: { authorization: `Bearer ${this.token}` } }); if (!res.ok) { const detail = await res.text().catch(() => ""); throw new Error(`vault transport: list inbound queue failed (${res.status}) ${detail}`.trim()); } - type RawNote = { id?: string; content?: string; metadata?: Record }; + type RawNote = { + id?: string; + content?: string; + metadata?: Record; + updated_at?: string; + updatedAt?: string; + }; let notes: RawNote[]; try { const parsed = (await res.json()) as unknown; @@ -1222,15 +1273,27 @@ export class VaultTransport implements Transport { if (typeof note.id !== "string" || !note.id) continue; const meta = note.metadata ?? {}; if (meta.channel !== channel) continue; // client-side channel filter (index-free). + const status = coerceInboundStatus(meta[STATUS_META_KEY]); + // Drop `handled` notes — they are not queue items (#103). Only pending + in-flight + // make up the actionable queue; counting/returning handled would let them crowd + // the live queue out of the cap. + if (status === "handled") continue; + const updatedAt = + typeof note.updated_at === "string" + ? note.updated_at + : typeof note.updatedAt === "string" + ? note.updatedAt + : undefined; out.push({ id: note.id, text: typeof note.content === "string" ? note.content : "", sender: typeof meta.sender === "string" ? meta.sender : "", ts: typeof meta.ts === "string" ? meta.ts : "", - status: coerceInboundStatus(meta[STATUS_META_KEY]), + status, ...(typeof meta[CLAIMED_AT_META_KEY] === "string" ? { claimedAt: meta[CLAIMED_AT_META_KEY] as string } : {}), + ...(updatedAt ? { updatedAt } : {}), }); } // Ascending by ts; blank-ts notes sort first (stable, deterministic). @@ -1241,29 +1304,49 @@ export class VaultTransport implements Transport { /** * PATCH an inbound note's claim status (+ optionally `claimedAt`), by note id. * Sends ONLY the changed metadata; the vault MERGES it, so the channel/direction/ - * sender/ts are preserved. `force: true` satisfies the vault's 428 mutation - * precondition (the 4a precondition) — safe here: `status`/`claimedAt` are the - * module's OWN authoritative claim fields, the body carries no content. Passing - * `claimedAt: null` CLEARS the field (written as an empty string) — used on - * release/handled so a stale claim timestamp doesn't linger. Throws on a non-ok - * vault response (the caller decides whether to surface or swallow). + * sender/ts are preserved. Passing `claimedAt: null` CLEARS the field (written as + * an empty string) — used on release/handled so a stale claim timestamp doesn't + * linger. + * + * COMPARE-AND-SWAP (agent#101). When `ifUpdatedAt` is given, the PATCH carries + * `if_updated_at` (the note's last-seen `updated_at`) as the vault's optimistic- + * concurrency precondition instead of `force: true` — so a CLAIM only lands if the + * note hasn't changed since it was read. A STALE precondition (another session + * already claimed it) makes the vault return **409** (`conflict`); an ABSENT one (if + * the note carried no `updated_at` to send) would 428 — either way we throw + * {@link InboundClaimConflictError} so the caller re-lists and skips to the next + * pending message rather than double-claiming. When `ifUpdatedAt` is OMITTED (the + * release/handled/sweep paths, which are last-write-wins by design) the PATCH uses + * `force: true` as before. Any OTHER non-ok status throws a plain Error. */ async setInboundStatus( id: string, status: InboundStatus, claimedAt?: string | null, + ifUpdatedAt?: string, ): Promise { const metadata: Record = { [STATUS_META_KEY]: status }; if (claimedAt !== undefined) { metadata[CLAIMED_AT_META_KEY] = claimedAt === null ? "" : claimedAt; } const url = `${this.vaultUrl}/vault/${this.vault}/api/notes/${encodeURIComponent(id)}`; + // CAS when an `ifUpdatedAt` precondition is supplied; otherwise last-write-wins via + // `force` (the prior behavior, kept for release/handled/sweep). + const body = + ifUpdatedAt !== undefined + ? { metadata, if_updated_at: ifUpdatedAt } + : { metadata, force: true }; const res = await fetch(url, { method: "PATCH", headers: { "content-type": "application/json", authorization: `Bearer ${this.token}` }, - body: JSON.stringify({ metadata, force: true }), + body: JSON.stringify(body), }); if (!res.ok) { + // 409 (stale precondition) / 428 (precondition required) on a CAS attempt = the + // claim race was lost → a typed conflict the caller re-lists on. + if (ifUpdatedAt !== undefined && (res.status === 409 || res.status === 428)) { + throw new InboundClaimConflictError(id, res.status); + } const detail = await res.text().catch(() => ""); throw new Error( `vault transport: set inbound status ${id} failed (${res.status}) ${detail}`.trim(), From 8aca7793da73ffec9f31c944da4dc9b55af04ddd Mon Sep 17 00:00:00 2001 From: Aaron Gabriel Date: Fri, 19 Jun 2026 08:12:55 -0600 Subject: [PATCH 2/2] fix(thread): re-record the SAME turn on outbound failure (no double-count) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR #3 reviewer (af47de6b, LGTM-with-nits) caught the one structural nit in FIX 1: the outbound-failure path called recordThread a SECOND time for the same turn, so single-threaded re-read turn_count=N and wrote N+1 (the turn counted twice), and multi-threaded minted a SECOND per-fire note (fresh uuid) for one turn. Fix: thread a stable per-TURN id (turnThreadId) through every recordThread for the turn, and a `sameTurn` flag on the failure re-record. writeThread then reuses that leaf for multi-threaded (one note, not a duplicate) and, for single-threaded with sameTurn, keeps the existing turn_count instead of incrementing. The note still ends on status:error with the undelivered reply — just counted once. Tests: single-threaded same-turn re-record keeps turn_count==1 + flips to error; multi-threaded re-record reuses the threadId leaf (both writes hit one path). Declined the 3 cosmetic nits (dead camelCase fallback, two comment-only items). Gate: bun run test:all -> typecheck 0, bun 972/0, vitest 87/0. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/backends/registry.ts | 33 ++++++++++++++++-- src/transport.ts | 13 ++++++++ src/transports/vault.test.ts | 65 ++++++++++++++++++++++++++++++++++++ src/transports/vault.ts | 11 ++++-- 4 files changed, 117 insertions(+), 5 deletions(-) diff --git a/src/backends/registry.ts b/src/backends/registry.ts index 32857b7..343f11f 100644 --- a/src/backends/registry.ts +++ b/src/backends/registry.ts @@ -113,6 +113,17 @@ export interface ThreadNote { ended_at: string; /** Optional token/cost usage for observability. */ usage?: { inputTokens?: number; outputTokens?: number; totalCostUsd?: number }; + /** + * MULTI-threaded only: a stable per-TURN thread id (the per-fire note's leaf). The same + * id on a re-record (the outbound-failure status flip) reuses the SAME note instead of + * minting a duplicate. Single-threaded ignores it (deterministic name leaf). + */ + threadId?: string; + /** + * Re-record of the SAME turn — single-threaded keeps `turn_count` (the turn was already + * counted by the first record); no effect on multi-threaded. + */ + sameTurn?: boolean; } /** @@ -430,6 +441,11 @@ export class ProgrammaticAgentRegistry { // multi-threaded writes one note per fire. Read the mode off the spec so the // thread note carries it (it's the indexed query axis + governs the upsert). const startedAt = new Date().toISOString(); + // A stable per-TURN thread id, passed to every recordThread for this turn. For + // multi-threaded it's the per-fire note's leaf, so a re-record (the outbound-failure + // status flip below) updates the SAME note instead of minting a duplicate; single- + // threaded ignores it (deterministic name leaf). One uuid per turn. + const turnThreadId = crypto.randomUUID(); let result; try { @@ -452,7 +468,7 @@ export class ProgrammaticAgentRegistry { // thread note captures the turn outcome, so a failed turn is still a queryable // `status:error` (single-threaded upserts the rolling thread; multi-threaded writes // a per-fire note). - await this.recordThread(handle, msg, "error", reason, startedAt, undefined); + await this.recordThread(handle, msg, "error", reason, startedAt, undefined, { threadId: turnThreadId }); this.emitTurnEvent(channel, { kind: "error", error: reason }); continue; } @@ -467,7 +483,7 @@ export class ProgrammaticAgentRegistry { // BOTH modes record the failed turn (status:error) on the thread note so a failure // always leaves a queryable trace (single-threaded upserts the rolling thread, // marking it errored; multi-threaded writes a per-fire status:error note). - await this.recordThread(handle, msg, "error", result.error, startedAt, undefined); + await this.recordThread(handle, msg, "error", result.error, startedAt, undefined, { threadId: turnThreadId }); this.emitTurnEvent(channel, { kind: "error", error: result.error }); continue; } @@ -480,7 +496,7 @@ export class ProgrammaticAgentRegistry { // multi-threaded writes the per-fire note. Best-effort: a thread-note failure is // logged + the turn still resolves (we never re-run a `claude -p` turn — that would // burn quota for a duplicate). - await this.recordThread(handle, msg, "ok", result.reply ?? "", startedAt, result.usage); + await this.recordThread(handle, msg, "ok", result.reply ?? "", startedAt, result.usage, { threadId: turnThreadId }); // The outbound reply — the channel-transcript delivery (the chat bubble). It is // ADDITIVE to the primary thread-note record already written above (for BOTH modes). @@ -504,6 +520,10 @@ export class ProgrammaticAgentRegistry { `parachute-agent: programmatic outbound write for channel "${channel}" failed ` + `after ${OUTBOUND_MAX_RETRIES} retries: ${delivered.error}`, ); + // RE-RECORD the SAME turn as status:error — reuse the per-turn thread id + + // `sameTurn` so this updates the note the `ok` record above just wrote (one + // note, no turn_count double-count) rather than minting a duplicate / advancing + // the count (the FIX-1 re-record bug the reviewer caught). await this.recordThread( handle, msg, @@ -512,6 +532,7 @@ export class ProgrammaticAgentRegistry { `Undelivered reply text: ${result.reply}`, startedAt, result.usage, + { threadId: turnThreadId, sameTurn: true }, ); this.emitTurnEvent(channel, { kind: "error", @@ -548,6 +569,7 @@ export class ProgrammaticAgentRegistry { output: string, startedAt: string, usage: ThreadNote["usage"], + opts: { threadId?: string; sameTurn?: boolean } = {}, ): Promise { if (!this.writeThread) return; const thread: ThreadNote = { @@ -561,6 +583,11 @@ export class ProgrammaticAgentRegistry { started_at: startedAt, ended_at: new Date().toISOString(), ...(usage ? { usage } : {}), + // The per-turn thread id (stable across an ok→error re-record) + the same-turn flag, + // so a re-record updates the SAME note without minting a duplicate (multi) or + // double-counting turn_count (single). + ...(opts.threadId ? { threadId: opts.threadId } : {}), + ...(opts.sameTurn ? { sameTurn: true } : {}), }; try { await this.writeThread(thread); diff --git a/src/transport.ts b/src/transport.ts index c53deb2..b5bb7e8 100644 --- a/src/transport.ts +++ b/src/transport.ts @@ -69,6 +69,19 @@ export interface ThreadRecord { ended_at: string; /** Optional token/cost usage for this turn (single-threaded accumulates into the note). */ usage?: { inputTokens?: number; outputTokens?: number; totalCostUsd?: number }; + /** + * MULTI-threaded only: a stable per-TURN thread id (the note's path leaf). Passing the + * SAME id on a re-record (e.g. flipping `ok`→`error` after an outbound-delivery failure) + * makes both writes hit the SAME per-fire note instead of minting a duplicate. Absent → + * a fresh id is minted. Single-threaded ignores it (its leaf is the deterministic name). + */ + threadId?: string; + /** + * Re-record of the SAME turn (not a new turn). Single-threaded keeps the existing + * `turn_count` instead of incrementing (the turn was already counted by the first + * record). No effect on multi-threaded (turn_count is always 1). + */ + sameTurn?: boolean; } export interface ReactArgs { diff --git a/src/transports/vault.test.ts b/src/transports/vault.test.ts index 2d16a71..48df47a 100644 --- a/src/transports/vault.test.ts +++ b/src/transports/vault.test.ts @@ -349,6 +349,71 @@ describe("VaultTransport — writeThread (#agent/thread note, the unified model) expect(stored!.content).toContain("reply two"); }); + test("SINGLE-THREADED re-record of the SAME turn (sameTurn) flips status WITHOUT double-counting turn_count (PR #3 FIX 1)", async () => { + // The outbound-failure path: the turn was recorded `ok`, then the additive transcript + // write failed, so the same turn is re-recorded `error`. `sameTurn` must keep the count + // (the turn was already counted) — the reviewer caught the original re-record bumping it. + let stored: { metadata: Record; content: string } | undefined; + globalThis.fetch = (async (url: string | URL | Request, init?: RequestInit) => { + const u = String(url); + const method = init?.method ?? "GET"; + if (u.includes("/api/notes/") && method === "GET") { + if (!stored) return new Response("not found", { status: 404 }); + return new Response(JSON.stringify(stored), { status: 200 }); + } + if (u.includes("/api/notes/") && method === "PATCH") { + const body = JSON.parse(String(init?.body)) as { metadata: Record; content: string }; + stored = { metadata: body.metadata, content: body.content }; + return new Response(JSON.stringify({ id: "thread-eng" }), { status: 200 }); + } + return new Response("{}", { status: 200 }); + }) as typeof fetch; + const t = new VaultTransport(baseConfig()); + await t.start(fakeCtx("eng")); + await t.writeThread({ + channel: "eng", name: "eng", mode: "single-threaded", status: "ok", + input: "q", output: "a", started_at: "2026-06-18T07:00:00.000Z", + ended_at: "2026-06-18T07:00:05.000Z", threadId: "t1", + }); + expect(stored!.metadata.turn_count).toBe("1"); + // Re-record the SAME turn as error (outbound delivery failed). sameTurn → no increment. + await t.writeThread({ + channel: "eng", name: "eng", mode: "single-threaded", status: "error", + input: "q", output: "reply produced but NOT delivered", started_at: "2026-06-18T07:00:00.000Z", + ended_at: "2026-06-18T07:00:06.000Z", threadId: "t1", sameTurn: true, + }); + expect(stored!.metadata.turn_count).toBe("1"); // NOT 2 — the same turn, not a new one. + expect(stored!.metadata.status).toBe("error"); + expect(stored!.content).toContain("NOT delivered"); + }); + + test("MULTI-THREADED re-record reuses the passed threadId leaf — ONE note, not a duplicate (PR #3 FIX 1)", async () => { + const patchPaths: string[] = []; + globalThis.fetch = (async (url: string | URL | Request, init?: RequestInit) => { + const u = String(url); + if (u.includes("/api/notes/") && (init?.method ?? "GET") === "PATCH") { + patchPaths.push(decodeURIComponent(u)); + return new Response(JSON.stringify({ id: "x" }), { status: 200 }); + } + return new Response("{}", { status: 200 }); + }) as typeof fetch; + const t = new VaultTransport(baseConfig()); + await t.start(fakeCtx("eng")); + const base = { + channel: "eng", name: "d", mode: "multi-threaded" as const, + input: "q", started_at: "2026-06-18T07:00:00.000Z", ended_at: "2026-06-18T07:00:05.000Z", + threadId: "fixed-uuid", + }; + await t.writeThread({ ...base, status: "ok", output: "a" }); + await t.writeThread({ ...base, status: "error", output: "undelivered", sameTurn: true }); + // Both writes hit the SAME per-fire path (the reused threadId) — without the fix the + // second would mint a fresh uuid → a DIFFERENT path → a duplicate note for one turn. + const threadPatches = patchPaths.filter((p) => p.includes("/Threads/eng/")); + expect(threadPatches).toHaveLength(2); + expect(threadPatches[0]).toContain("/Threads/eng/fixed-uuid"); + expect(threadPatches[1]).toContain("/Threads/eng/fixed-uuid"); + }); + test("SINGLE-THREADED error on turn 2: turn_count==2, status:error, started_at preserved, last_turn_at advanced", async () => { // Same stored-note simulation as the two-turn test: turn 2 reads back turn 1's note. let stored: { metadata: Record; content: string } | undefined; diff --git a/src/transports/vault.ts b/src/transports/vault.ts index f6ad3d9..65c61e6 100644 --- a/src/transports/vault.ts +++ b/src/transports/vault.ts @@ -809,7 +809,11 @@ export class VaultTransport implements Transport { // registry enforces ONE agent per channel (byChannel index), so the collision can't // arise in practice. const safeName = (thread.name ?? thread.channel).replace(/[^a-zA-Z0-9_-]/g, "-"); - const leaf = singleThreaded ? safeName : crypto.randomUUID(); + // Multi-threaded leaf: a per-FIRE id. Reuse the caller's `threadId` when given (a + // re-record of the same turn — e.g. the outbound-failure status flip — targets the + // SAME per-fire note instead of minting a duplicate); else mint a fresh one. Single- + // threaded ignores it (deterministic name leaf so the one-per-channel note upserts). + const leaf = singleThreaded ? safeName : (thread.threadId ?? crypto.randomUUID()); const path = `${THREAD_PATH_PREFIX}/${safeChannel}/${leaf}`; // For single-threaded UPSERT, read the existing thread note (by its deterministic @@ -837,7 +841,10 @@ export class VaultTransport implements Transport { } } - const turnCount = singleThreaded ? priorTurnCount + 1 : 1; + // A re-record of the SAME turn (`sameTurn`) keeps the existing count — the first record + // already counted this turn; a status flip (ok→error on outbound-delivery failure) must + // not double-count it. A normal turn increments. Multi-threaded is always 1. + const turnCount = singleThreaded ? (thread.sameTurn ? priorTurnCount : priorTurnCount + 1) : 1; // `started_at` is set ONCE on create (preserve the prior on upsert); `last_turn_at` // advances every turn. const startedAt = priorStartedAt ?? thread.started_at;