Skip to content

feat: add durable delivery receipts and inbox cursors #395

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
EtanHey merged 3 commits into from
Aug 12, 2026
Merged

feat: add durable delivery receipts and inbox cursors #395

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e9b1a0c
feat: add durable delivery receipts and inbox cursors
EtanHey Aug 11, 2026
937c827
fix: make queued delivery durable across agent posture
EtanHey Aug 11, 2026
9d2c7d7
test: pin retryable busy delivery classification
EtanHey Aug 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions docs/plans/2026-08-11-pr395-review-fixes.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
# PR 395 High-Severity Review Fixes

## Goal

Close the two high-severity review gaps without widening Phase 1 scope.

## Design

1. Bootstrap inbox state for every spawned role and append the concrete per-agent mailbox monitor/cursor contract to every delivered boot instruction. Preserve the caller's task text as the task summary.
2. Make the guarded `deliverAgentInput` route state the sole interactive-state authority for queued delivery. Represent a pre-mutation posture rejection as retryable, retain the durable queued receipt, and retry with exponential backoff capped at a fixed maximum. Mark a receipt terminal only when the target is gone or the submission outcome is uncertain.
3. Pin worker and workspace spawn boot delivery, posture disagreement retry, backoff, eventual submission, and target-gone resolution with focused tests.
4. Run focused tests, typecheck/build/full suite, compiled probes, push the signed commit, update PR 395, and inbox the lead.

291 changes: 290 additions & 1 deletion src/agent-engine.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,16 @@
* These 7 functions are the engine that MCP tools (and later the 2-tool facade) drive.
*/

import { existsSync, readFileSync, statSync } from "node:fs";
import { randomUUID } from "node:crypto";
import {
existsSync,
mkdirSync,
readFileSync,
renameSync,
statSync,
unlinkSync,
writeFileSync,
} from "node:fs";
import { homedir } from "node:os";
import { dirname, isAbsolute, join, resolve } from "node:path";
import { StateManager } from "./state-manager.js";
Expand Down Expand Up @@ -46,6 +55,7 @@ import {
type AgentState,
type CliType,
type CloseForensicsEvent,
type DeliveryEventType,
type PublicAgent,
type WaitResult,
} from "./agent-types.js";
Expand Down Expand Up @@ -152,6 +162,39 @@ import {

type ProcessLiveness = "alive" | "gone" | "unknown";

export type AgentDeliveryState = "submitted" | "queued" | "failed";

export interface AgentDeliveryReceipt {
delivery_id: string;
agent_id: string;
text: string;
press_enter: boolean;
source_event: DeliveryEventType;
delivery_state: AgentDeliveryState;
terminal: boolean;
created_at: string;
resolved_at: string | null;
retry_count: number;
submit_verified: boolean | null;
error: string | null;
/** Persisted before terminal mutation; a nonterminal value is never replayed after restart. */
submission_started_at?: string | null;
/** Earliest wall-clock time at which a known pre-mutation rejection may retry. */
next_attempt_at?: string | null;
}

/** A known pre-mutation delivery rejection that is safe to retry. */
export class RetryableDeliveryError extends Error {
constructor(message: string) {
super(message);
this.name = "RetryableDeliveryError";
}
}

type DeliverySubmitter = (
receipt: AgentDeliveryReceipt,
) => Promise<{ retry_count: number; submit_verified: boolean | null }>;

export interface SpawnAgentParams {
repo: string;
model?: string;
Expand Down Expand Up @@ -381,6 +424,8 @@ export interface AgentEngineOptions {
fleetSidebarPublisher?: FleetSidebarPublisherLike;
/** Render-only timeout for a working seat whose transcript/output stops advancing. */
fleetWorkingNoProgressTimeoutMs?: number;
/** Bound one queued terminal submission so lifecycle sweeps cannot hang forever. */
deliverySubmitTimeoutMs?: number;
}

export type RolePlacementReconcileTrigger = "spawn" | "idle" | "boot";
Expand Down Expand Up @@ -904,13 +949,27 @@ export class AgentEngine {
private fleetWorkingNoProgressTimeoutMs: number;
private startupInitializePromise: Promise<void> | null = null;
private lifecycleMutationTail: Promise<void> = Promise.resolve();
private deliveryReceipts = new Map<string, AgentDeliveryReceipt>();
private deliveryReceiptsPath: string;
private deliverySubmitter: DeliverySubmitter | null = null;
private deliveryDrainInFlight = false;
private deliverySubmitTimeoutMs: number;
constructor(
stateMgr: StateManager,
registry: AgentRegistry,
client: AgentEngineClient,
opts?: AgentEngineOptions,
) {
this.stateMgr = stateMgr;
this.deliveryReceiptsPath = join(
stateMgr.getBaseDir(),
"delivery-receipts.json",
);
this.deliverySubmitTimeoutMs = Math.max(
1,
opts?.deliverySubmitTimeoutMs ?? 30_000,
);
this.loadDeliveryReceipts();
this.registry = registry;
this.client = client;
this.roleSurfaceIdsProvider = opts?.roleSurfaceIdsProvider;
Expand Down Expand Up @@ -4463,6 +4522,236 @@ export class AgentEngine {

async runSweep(): Promise<void> {

@macroscopeapp macroscopeapp Bot Aug 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟠 High src/agent-engine.ts:4513

If runSweepOnce() (or the lifecycle-mutation wrapper) rejects — e.g. due to a setStatus/setStatuses failure inside syncSidebar — drainDeliveryQueue() is never reached. Durable queued deliveries that were already accepted and persisted will remain stuck for as long as the sweep keeps failing, even when the target agent is ready/idle. Wrap the sweep in try/finally so the drain always runs.

🚀 Reply "fix it for me" or copy this AI Prompt for your agent: 
In file @src/agent-engine.ts around line 4513:

If `runSweepOnce()` (or the lifecycle-mutation wrapper) rejects — e.g. due to a `setStatus`/`setStatuses` failure inside `syncSidebar` — `drainDeliveryQueue()` is never reached. Durable queued deliveries that were already accepted and persisted will remain stuck for as long as the sweep keeps failing, even when the target agent is `ready`/`idle`. Wrap the sweep in `try/finally` so the drain always runs.

await this.runLifecycleMutation(() => this.runSweepOnce());
await this.drainDeliveryQueue();
}
Comment on lines 4523 to +4526

@coderabbitai coderabbitai Bot Aug 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Description: Find the delivery submitter registration and its mutation path.
rg -n -C15 'setDeliverySubmitter|deliverAgentInput' src --glob '*.ts'

Repository: EtanHey/cmuxlayer

Length of output: 21881

🏁 Script executed:

#!/bin/bash
set -eu
printf '%s\n' '--- lifecycle lock and delivery methods ---'
rg -n -C35 'runLifecycleMutation|drainDeliveryQueue|deliverySubmitter|deliveryDrainInFlight' src/agent-engine.ts
printf '%s\n' '--- server delivery implementation ---'
sed -n '9327,9549p' src/server.ts
printf '%s\n' '--- lifecycle mutation call sites ---'
rg -n -C8 'runLifecycleMutation\(' src --glob '*.ts'

Repository: EtanHey/cmuxlayer

Length of output: 32543

🏁 Script executed:

#!/bin/bash
set -eu
printf '%s\n' '--- delivery metadata refresh path ---'
rg -n -C20 'refreshManagedMetadataBestEffort|lifecycleRefreshManagedMetadata' src/server.ts src/agent-engine.ts
printf '%s\n' '--- lifecycle mutation implementations and terminal operations ---'
rg -n -C12 'stopAgent|recoverCrashedAgents|reconcileRolePlacements|closeSurface|bind.*Surface|withSurfaceWrite' src/agent-engine.ts src/server.ts
printf '%s\n' '--- write serialization implementation ---'
rg -n -C25 'function withSurfaceWrite|const withSurfaceWrite|withSurfaceWrite =' src --glob '*.ts'

Repository: EtanHey/cmuxlayer

Length of output: 50374

🏁 Script executed:

#!/bin/bash
set -eu
printf '%s\n' '--- refresh lock and delivery write lock ---'
sed -n '2938,2964p' src/server.ts
sed -n '9137,9155p' src/server.ts
rg -n -C30 'withSurfaceWrite' src/server.ts src/agent-engine.ts
printf '%s\n' '--- direct lifecycle mutation wrappers near relevant tools ---'
rg -n -C10 'runLifecycleMutation' src/server.ts
printf '%s\n' '--- sweep startup and scheduling ---'
rg -n -C20 'startSweep|runSweep\(' src/agent-engine.ts src/server.ts

Repository: EtanHey/cmuxlayer

Length of output: 50373

🏁 Script executed:

#!/bin/bash
set -eu
printf '%s\n' '--- surface-write lock implementation ---'
sed -n '3310,3575p' src/server.ts
printf '%s\n' '--- lifecycle surface mutation calls ---'
rg -n -C14 'this\.client\.(closeSurface|moveSurface|newSplit)|reconcileRolePlacements\(' src/agent-engine.ts
printf '%s\n' '--- stable identity propagation in recovery/reflow ---'
rg -n -C10 'stableSurfaceIdentity|surface_uuid' src/agent-engine.ts | head -240

Repository: EtanHey/cmuxlayer

Length of output: 31403

🏁 Script executed:

#!/bin/bash
set -eu
printf '%s\n' '--- stopAgent implementation and callers ---'
sed -n '6200,6355p' src/agent-engine.ts
rg -n -C20 '"stop_agent"|stopAgent\(' src/server.ts src/agent-engine.ts
printf '%s\n' '--- crash recovery close/write options ---'
sed -n '3144,3330p' src/agent-engine.ts
printf '%s\n' '--- lifecycle scheduler and direct sweep callers ---'
rg -n -C18 'startSweep|runSweep\(' src/agent-engine.ts src/server.ts

Repository: EtanHey/cmuxlayer

Length of output: 32145

🏁 Script executed:

#!/bin/bash
set -eu
node - <<'JS'
"use strict";

// Model the lock semantics shown by withSurfaceWrite and runLifecycleMutation.
// acquireSurfaceWrite rejects only an identical key; different keys do not
// serialize. The lifecycle tail is a non-reentrant FIFO lock.

const active = new Map();
function acquire(key, owner) {
  if (active.has(key)) throw new Error(`conflict:${key}`);
  active.set(key, owner);
}
function release(key, owner) {
  if (active.get(key) === owner) active.delete(key);
}

acquire("uuid:agent-1", "delivery");
let overlap = false;
try {
  acquire("surface-ref-1", "stop-close");
  overlap = true;
} finally {
  release("surface-ref-1", "stop-close");
  release("uuid:agent-1", "delivery");
}

let released = false;
let nestedCompleted = false;
const outer = Promise.resolve().then(async () => {
  const nestedWait = new Promise((resolve) => {
    if (released) resolve();
    else setTimeout(() => resolve(), 0);
  });
  await Promise.race([
    nestedWait.then(() => { nestedCompleted = true; }),
    new Promise((resolve) => setTimeout(resolve, 10)),
  ]);
  released = true;
});
outer.then(() => {});
setTimeout(() => {
  if (!overlap) throw new Error("expected distinct surface keys to overlap");
  if (nestedCompleted) throw new Error("nested lock unexpectedly completed");
  console.log(JSON.stringify({
    distinct_surface_keys_overlap: overlap,
    nested_lifecycle_lock_waits: !nestedCompleted,
  }));
}, 20);
JS

Repository: EtanHey/cmuxlayer

Length of output: 481

🏁 Script executed:

#!/bin/bash
set -eu
node - <<'JS'
"use strict";

const active = new Map();
function acquire(key, owner) {
  if (active.has(key)) throw new Error(`conflict:${key}`);
  active.set(key, owner);
}
function release(key, owner) {
  if (active.get(key) === owner) active.delete(key);
}

acquire("uuid:agent-1", "delivery");
let distinctKeysOverlap = false;
try {
  acquire("surface-ref-1", "stop-close");
  distinctKeysOverlap = true;
} finally {
  release("surface-ref-1", "stop-close");
  release("uuid:agent-1", "delivery");
}

let tail = Promise.resolve();
function runLifecycleMutation(operation) {
  const previous = tail;
  let release;
  tail = new Promise((resolve) => { release = resolve; });
  return previous.then(async () => {
    try {
      return await operation();
    } finally {
      release();
    }
  });
}

let nestedCompleted = false;
const outer = runLifecycleMutation(async () => {
  await Promise.race([
    runLifecycleMutation(async () => {
      nestedCompleted = true;
    }),
    new Promise((resolve) => setTimeout(resolve, 20)),
  ]);
});
await outer;

console.log(JSON.stringify({
  distinct_surface_keys_overlap: distinctKeysOverlap,
  nested_lifecycle_lock_pending_during_outer_operation: !nestedCompleted,
}));
JS

Repository: EtanHey/cmuxlayer

Length of output: 255

🏁 Script executed:

#!/bin/bash
set -eu
node - <<'JS'
"use strict";

let tail = Promise.resolve();
function runLifecycleMutation(operation) {
  const previous = tail;
  let release;
  tail = new Promise((resolve) => { release = resolve; });
  return previous.then(async () => {
    try {
      return await operation();
    } finally {
      release();
    }
  });
}

let nestedCompleted = false;
let nestedCompletedWhileOuterHeld = false;
let outerReleased = false;
let observedWhileOuterHeld = false;

const outer = runLifecycleMutation(async () => {
  void runLifecycleMutation(async () => {
    nestedCompleted = true;
    if (!outerReleased) nestedCompletedWhileOuterHeld = true;
  });
  await new Promise((resolve) => setTimeout(resolve, 10));
  observedWhileOuterHeld = !nestedCompleted;
});
await outer;
outerReleased = true;
await new Promise((resolve) => setTimeout(resolve, 0));

console.log(JSON.stringify({
  nested_lock_pending_while_outer_held: observedWhileOuterHeld,
  nested_completed_while_outer_held: nestedCompletedWhileOuterHeld,
  nested_completed_after_outer_release: nestedCompleted,
}));
JS

Repository: EtanHey/cmuxlayer

Length of output: 286

🏁 Script executed:

#!/bin/bash
set -eu
node - <<'JS'
"use strict";

let tail = Promise.resolve();
let outerReleased = false;

function runLifecycleMutation(id, operation) {
  const previous = tail;
  let release;
  tail = new Promise((resolve) => { release = resolve; });
  return previous.then(async () => {
    try {
      return await operation();
    } finally {
      if (id === "outer") outerReleased = true;
      release();
    }
  });
}

let nestedCompleted = false;
let nestedCompletedWhileOuterHeld = false;
let observedWhileOuterHeld = false;

const outer = runLifecycleMutation("outer", async () => {
  void runLifecycleMutation("nested", async () => {
    nestedCompleted = true;
    if (!outerReleased) nestedCompletedWhileOuterHeld = true;
  });
  await new Promise((resolve) => setTimeout(resolve, 10));
  observedWhileOuterHeld = !nestedCompleted;
});
await outer;
await new Promise((resolve) => setTimeout(resolve, 0));

console.log(JSON.stringify({
  nested_lock_pending_while_outer_held: observedWhileOuterHeld,
  nested_completed_while_outer_held: nestedCompletedWhileOuterHeld,
  nested_completed_after_outer_release: nestedCompleted && outerReleased,
}));
JS

Repository: EtanHey/cmuxlayer

Length of output: 287

Use the same surface-write lock key for delivery and stop cleanup.

deliverAgentInput() locks uuid:${surface_uuid}. stopAgent() closes the route without stableSurfaceIdentity, so it locks the mutable surface reference instead. These keys can overlap on the same surface.

Do not move drainDeliveryQueue() inside runLifecycleMutation(). Its metadata refresh uses the non-reentrant lifecycle lock and can block until the delivery timeout. Pass stableSurfaceIdentity: route.surface_uuid to the stop close path.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/agent-engine.ts` around lines 4523 - 4526, Update the stop cleanup path
used by runSweep and stopAgent so the route close operation receives
stableSurfaceIdentity set to route.surface_uuid, matching deliverAgentInput’s
uuid-based lock key. Keep drainDeliveryQueue() outside runLifecycleMutation();
do not change the delivery queue ordering or lifecycle lock behavior.


setDeliverySubmitter(submitter: DeliverySubmitter | null): void {
this.deliverySubmitter = submitter;
}

private loadDeliveryReceipts(): void {
try {
const parsed: unknown = JSON.parse(
readFileSync(this.deliveryReceiptsPath, "utf8"),
);
if (!Array.isArray(parsed)) return;
let repairedUncertainReceipt = false;
for (const candidate of parsed) {
if (
candidate &&
typeof candidate === "object" &&
typeof (candidate as AgentDeliveryReceipt).delivery_id === "string"
) {
const receipt: AgentDeliveryReceipt = {
submission_started_at: null,
next_attempt_at: null,
...(candidate as AgentDeliveryReceipt),
};
if (
receipt.delivery_state === "queued" &&
receipt.submission_started_at
) {
receipt.delivery_state = "failed";
receipt.terminal = true;
receipt.resolved_at = new Date().toISOString();
receipt.error =
"Delivery outcome uncertain after process restart; refusing automatic replay";
repairedUncertainReceipt = true;
}
this.deliveryReceipts.set(receipt.delivery_id, receipt);
}
}
if (repairedUncertainReceipt) {
try {
this.persistDeliveryReceipts();
} catch {
// In-memory terminal state still prevents replay in this process.
}
}
} catch {
// Missing or corrupt legacy state must not prevent lifecycle startup.
}
}

private persistDeliveryReceipts(): void {

@macroscopeapp macroscopeapp Bot Aug 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟠 High src/agent-engine.ts:4565

Terminal delivery receipts (both resolved and drained) are never removed from deliveryReceipts. Every call to resolveDelivery, queueDelivery, or the drain loop appends to the map and then persistDeliveryReceipts rewrites the entire collection — including the full text payload of every historical receipt — as a single synchronous JSON file. On a long-running server this causes unbounded memory and disk growth, and each write gets progressively slower.

Consider pruning terminal receipts (e.g. receipts where terminal === true and resolved_at is older than a retention window) before persisting, or cap the number of retained receipts.

🚀 Reply "fix it for me" or copy this AI Prompt for your agent: 
In file @src/agent-engine.ts around line 4565:

Terminal delivery receipts (both resolved and drained) are never removed from `deliveryReceipts`. Every call to `resolveDelivery`, `queueDelivery`, or the drain loop appends to the map and then `persistDeliveryReceipts` rewrites the entire collection — including the full `text` payload of every historical receipt — as a single synchronous JSON file. On a long-running server this causes unbounded memory and disk growth, and each write gets progressively slower.

Consider pruning terminal receipts (e.g. receipts where `terminal === true` and `resolved_at` is older than a retention window) before persisting, or cap the number of retained receipts.

mkdirSync(dirname(this.deliveryReceiptsPath), { recursive: true });
const tempPath = `${this.deliveryReceiptsPath}.${process.pid}.${randomUUID()}.tmp`;
try {
writeFileSync(
tempPath,
`${JSON.stringify([...this.deliveryReceipts.values()], null, 2)}\n`,
"utf8",
Comment on lines +4580 to +4583

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Prune terminal receipts before rewriting the receipt store

Every immediate or queued delivery remains in deliveryReceipts forever, including its full text, and each state change synchronously serializes and rewrites the entire map. On a long-running daemon this makes disk usage unbounded and turns ordinary sends into progressively larger O(total historical payload) writes; apply retention or compact terminal receipts before persisting.

Useful? React with 👍 / 👎.

);
renameSync(tempPath, this.deliveryReceiptsPath);
} finally {
if (existsSync(tempPath)) unlinkSync(tempPath);
}
}
Comment on lines +4576 to +4589

@coderabbitai coderabbitai Bot Aug 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🚀 Performance & Scalability | 🟠 Major | 🏗️ Heavy lift

Delivery receipts are never pruned, and the whole file is rewritten on every mutation.

Three consequences follow from the same root cause:

  1. this.deliveryReceipts only ever grows. No code path deletes a terminal receipt. For a long-lived daemon, delivery-receipts.json grows without bound.
  2. persistDeliveryReceipts serializes the entire map with 2-space indentation on every change. drainDeliveryQueue calls it at least twice per queued receipt. Cost per drain is O(total receipts × queued receipts), and every sweep re-scans all historical receipts at Line 4650.
  3. AgentDeliveryReceipt.text stores the full delivery payload. Agent-to-agent message bodies are retained on disk in plaintext for the lifetime of the state directory.

Add retention: drop terminal receipts after a bounded age or count, and truncate or omit text once a receipt becomes terminal. Batch the persist calls in drainDeliveryQueue so one drain writes once.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/agent-engine.ts` around lines 4576 - 4589, Update delivery-receipt
lifecycle handling around persistDeliveryReceipts and drainDeliveryQueue: retain
only bounded-age or bounded-count receipts, remove terminal receipts when they
exceed retention, and clear or omit AgentDeliveryReceipt.text when a receipt
becomes terminal. Batch queue mutations so drainDeliveryQueue invokes
persistence once per drain rather than once per receipt, while preserving atomic
file replacement.


queueDelivery(input: {
agent_id: string;
text: string;
press_enter: boolean;
source_event: DeliveryEventType;
}): AgentDeliveryReceipt {
const receipt: AgentDeliveryReceipt = {
delivery_id: randomUUID(),
...input,
delivery_state: "queued",
terminal: false,
created_at: new Date().toISOString(),
resolved_at: null,
retry_count: 0,
submit_verified: null,
error: null,
submission_started_at: null,
next_attempt_at: null,
};
this.deliveryReceipts.set(receipt.delivery_id, receipt);
try {
// Acceptance is not returned until the full replay payload is durable.
this.persistDeliveryReceipts();

@macroscopeapp macroscopeapp Bot Aug 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟠 High src/agent-engine.ts:4601

persistDeliveryReceipts() serializes the entire in-memory deliveryReceipts map via atomic rename without first reading and merging the current file contents. When two AgentEngine instances share the same deliveryReceiptsPath (same state directory), the second writer's rename silently overwrites the first writer's receipt. The first caller already received an accepted receipt (the method returned successfully with delivery_state: "queued"), but its payload is permanently lost on disk and will never be drained after a restart.

The root cause is that persistDeliveryReceipts is a blind full-file replace with no read-merge step and no inter-process lock. Consider either (a) reloading the file and merging unknown entries before writing, ideally under an advisory file lock, or (b) using an append-only format (one JSON line per receipt) so concurrent writers cannot destroy each other's data.

🚀 Reply "fix it for me" or copy this AI Prompt for your agent: 
In file @src/agent-engine.ts around line 4601:

`persistDeliveryReceipts()` serializes the entire in-memory `deliveryReceipts` map via atomic rename without first reading and merging the current file contents. When two `AgentEngine` instances share the same `deliveryReceiptsPath` (same state directory), the second writer's rename silently overwrites the first writer's receipt. The first caller already received an accepted receipt (the method returned successfully with `delivery_state: "queued"`), but its payload is permanently lost on disk and will never be drained after a restart.

The root cause is that `persistDeliveryReceipts` is a blind full-file replace with no read-merge step and no inter-process lock. Consider either (a) reloading the file and merging unknown entries before writing, ideally under an advisory file lock, or (b) using an append-only format (one JSON line per receipt) so concurrent writers cannot destroy each other's data.

} catch (error) {
this.deliveryReceipts.delete(receipt.delivery_id);
throw error;
}
this.appendDeliveryReceiptEventBestEffort(receipt);
return { ...receipt };
}

resolveDelivery(
input: Omit<AgentDeliveryReceipt, "created_at" | "resolved_at"> & {
created_at?: string;
},
opts?: { appendFailureEvent?: boolean },
): AgentDeliveryReceipt {
const receipt: AgentDeliveryReceipt = {
...input,
created_at: input.created_at ?? new Date().toISOString(),
resolved_at: new Date().toISOString(),
};
this.deliveryReceipts.set(receipt.delivery_id, receipt);
this.persistDeliveryReceipts();
if (receipt.delivery_state === "failed" && opts?.appendFailureEvent) {
this.appendDeliveryReceiptEventBestEffort(receipt);
}
return { ...receipt };
}

getDeliveryReceipt(deliveryId: string): AgentDeliveryReceipt | null {
const receipt = this.deliveryReceipts.get(deliveryId);
return receipt ? { ...receipt } : null;
}

async drainDeliveryQueue(): Promise<void> {

@macroscopeapp macroscopeapp Bot Aug 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟠 High src/agent-engine.ts:4646

drainDeliveryQueue processes queued receipts sequentially with a per-item timeout, so a drain with N hung submissions blocks runSweep for N * deliverySubmitTimeoutMs — e.g. 100 stuck receipts stall the lifecycle sweep for ~50 minutes at the default 30s timeout. The deliverySubmitTimeoutMs bound only limits each item, not the whole drain, so an unbounded queue of hung submissions can halt all periodic reconciliation and lifecycle work. Consider bounding the total drain time or processing receipts concurrently so one slow queue cannot block the sweep indefinitely.

🚀 Reply "fix it for me" or copy this AI Prompt for your agent: 
In file @src/agent-engine.ts around line 4646:

`drainDeliveryQueue` processes queued receipts sequentially with a per-item timeout, so a drain with N hung submissions blocks `runSweep` for `N * deliverySubmitTimeoutMs` — e.g. 100 stuck receipts stall the lifecycle sweep for ~50 minutes at the default 30s timeout. The `deliverySubmitTimeoutMs` bound only limits each item, not the whole drain, so an unbounded queue of hung submissions can halt all periodic reconciliation and lifecycle work. Consider bounding the total drain time or processing receipts concurrently so one slow queue cannot block the sweep indefinitely.

if (this.deliveryDrainInFlight || !this.deliverySubmitter) return;
this.deliveryDrainInFlight = true;
try {
for (const receipt of this.deliveryReceipts.values()) {
if (receipt.delivery_state !== "queued") continue;
const agent = this.getAgentState(receipt.agent_id);

@macroscopeapp macroscopeapp Bot Aug 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟠 High src/agent-engine.ts:4640

drainDeliveryQueue looks up the agent via this.getAgentState(receipt.agent_id), which returns null after a restart if the agent was renamed (e.g. pending → final ID during session capture). In-memory registry aliases resolve the old ID within the same process, but reconstitute clears aliases on restart, so durable queued receipts referencing the old agent_id are silently skipped forever.

transferAgentRenameMemory rekeys every other in-memory map (readyPatternMatches, cliExitShellMatches, fleetScreenProgress, etc.) but does not update deliveryReceipts. Add rekeying of delivery receipts (updating each receipt's agent_id field) in transferAgentRenameMemory and call persistDeliveryReceipts() so the renamed ID survives a restart.

🚀 Reply "fix it for me" or copy this AI Prompt for your agent: 
In file @src/agent-engine.ts around line 4640:

`drainDeliveryQueue` looks up the agent via `this.getAgentState(receipt.agent_id)`, which returns `null` after a restart if the agent was renamed (e.g. pending → final ID during session capture). In-memory registry aliases resolve the old ID within the same process, but `reconstitute` clears aliases on restart, so durable queued receipts referencing the old `agent_id` are silently skipped forever.

`transferAgentRenameMemory` rekeys every other in-memory map (`readyPatternMatches`, `cliExitShellMatches`, `fleetScreenProgress`, etc.) but does not update `deliveryReceipts`. Add rekeying of delivery receipts (updating each receipt's `agent_id` field) in `transferAgentRenameMemory` and call `persistDeliveryReceipts()` so the renamed ID survives a restart.

if (!agent) {
receipt.delivery_state = "failed";
receipt.terminal = true;
receipt.resolved_at = new Date().toISOString();
receipt.error = `Delivery target ${receipt.agent_id} is gone or no longer exists`;
this.persistDeliveryReceipts();
this.appendDeliveryReceiptEventBestEffort(receipt);
continue;
}
if (
receipt.next_attempt_at &&
Date.parse(receipt.next_attempt_at) > Date.now()
) {
continue;
Comment on lines +4652 to +4666

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Drain queued sends before marking agents done

When a working agent completes its current task, runSweepOnce() first transitions it to done in maybeMarkTaskDone(), and only afterward calls this drain, which accepts only ready or idle. Normal sweeps do not transition a working agent to idle, so a follow-up queued while it is busy commonly remains queued forever precisely when the TUI becomes available; eligibility should use fresh interactive-screen evidence or drain before the terminal transition.

AGENTS.md reference: AGENTS.md:L8-L16

Useful? React with 👍 / 👎.

}
try {
receipt.submission_started_at = new Date().toISOString();
// This is the no-replay boundary. A crash after this write leaves an
// uncertain terminal receipt instead of re-sending terminal input.
this.persistDeliveryReceipts();
let timeout: ReturnType<typeof setTimeout> | null = null;
const result = await Promise.race([
this.deliverySubmitter(receipt),
new Promise<never>((_resolve, reject) => {
Comment on lines +4673 to +4676

@macroscopeapp macroscopeapp Bot Aug 11, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟠 High src/agent-engine.ts:4649

drainDeliveryQueue checks this.deliverySubmitter only once at entry, then awaits multiple submissions in a loop using the mutable field directly. If setDeliverySubmitter(null) is called while a submission is in flight, the next loop iteration calls null(receipt), throws a TypeError, and permanently records that never-attempted delivery as failed. Capture the submitter reference before entering the loop, or re-check it before each invocation.

           let timeout: ReturnType<typeof setTimeout> | null = null;
+          const submitter = this.deliverySubmitter;
+          if (!submitter) break;
           const result = await Promise.race([
-            this.deliverySubmitter(receipt),
+            submitter(receipt),
🚀 Reply "fix it for me" or copy this AI Prompt for your agent: 
In file @src/agent-engine.ts around lines 4649-4652:

`drainDeliveryQueue` checks `this.deliverySubmitter` only once at entry, then awaits multiple submissions in a loop using the mutable field directly. If `setDeliverySubmitter(null)` is called while a submission is in flight, the next loop iteration calls `null(receipt)`, throws a `TypeError`, and permanently records that never-attempted delivery as `failed`. Capture the submitter reference before entering the loop, or re-check it before each invocation.

timeout = setTimeout(
() =>
reject(
new Error(
`Delivery submission timed out after ${this.deliverySubmitTimeoutMs}ms; outcome uncertain and will not be retried`,
),
),
this.deliverySubmitTimeoutMs,
);
}),
]).finally(() => {
if (timeout) clearTimeout(timeout);
});
receipt.delivery_state = "submitted";
receipt.terminal = true;
receipt.resolved_at = new Date().toISOString();
receipt.retry_count += result.retry_count;
receipt.submit_verified = result.submit_verified;
receipt.error = null;
receipt.next_attempt_at = null;
} catch (error) {
if (error instanceof RetryableDeliveryError) {
receipt.submission_started_at = null;
receipt.retry_count += 1;
const backoffMs = Math.min(
30_000,
250 * 2 ** Math.min(receipt.retry_count - 1, 16),
);
receipt.next_attempt_at = new Date(
Date.now() + backoffMs,
).toISOString();
receipt.error = error.message;
} else {
receipt.delivery_state = "failed";
receipt.terminal = true;
receipt.resolved_at = new Date().toISOString();
receipt.error =
error instanceof Error ? error.message : String(error);
}
}
this.persistDeliveryReceipts();
// Successful delivery already emitted the correlated source event;
// failures have no such event and need an explicit terminal transition.
if (receipt.delivery_state === "failed") {
this.appendDeliveryReceiptEventBestEffort(receipt);
}
}
} finally {
this.deliveryDrainInFlight = false;
}
}

private appendDeliveryReceiptEvent(receipt: AgentDeliveryReceipt): void {
const agent = this.getAgentState(receipt.agent_id);
this.stateMgr.getEventLog().appendDelivery({
ts: receipt.resolved_at ?? receipt.created_at,
event_type: receipt.source_event,
source_agent: null,
target_surface: agent?.surface_id ?? "unknown",
target_agent: receipt.agent_id,
bytes: Buffer.byteLength(receipt.text),
press_enter: receipt.press_enter,
submit_verified: receipt.submit_verified,
retry_count: receipt.retry_count,
delivery_id: receipt.delivery_id,
delivery_state: receipt.delivery_state,
});
}

private appendDeliveryReceiptEventBestEffort(
receipt: AgentDeliveryReceipt,
): void {
try {
this.appendDeliveryReceiptEvent(receipt);
} catch {
// Receipt persistence is authoritative; telemetry must not invalidate
// acceptance or tempt a caller to duplicate terminal input.
}
}

requestFleetSidebarRepublish(): void {
Expand Down
5 changes: 5 additions & 0 deletions src/agent-types.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -188,6 +188,11 @@ export interface DeliveryTelemetryEvent {
press_enter: boolean | null;
submit_verified: boolean | null;
retry_count: number;
/** Stable receipt identity for agent-routed delivery state transitions. */
delivery_id?: string;
/** Nonterminal acceptance or terminal resolution. */
delivery_state?: "submitted" | "queued" | "failed";
target_agent?: string;
}

export interface ControlHealthTelemetryEvent {
Expand Down
Loading
Loading