From b9cce6fb254bfe5994321160025516d7afbc1f16 Mon Sep 17 00:00:00 2001 From: Luke Parke <5702154+LukasParke@users.noreply.github.com> Date: Thu, 13 Aug 2026 00:04:25 -0500 Subject: [PATCH] fix(platform-node): preserve queue and IPC ordering Port the queue, IPC, and subprocess identity portion of fork/port/openrouter-fixes commit b9905ac5. File storage remains separately reviewed in PR #70. Signed-off-by: Luke Parke <5702154+LukasParke@users.noreply.github.com> --- .../platform-node/src/agent-ipc-client.ts | 22 +- .../platform-node/src/agent-ipc-protocol.ts | 8 + .../platform-node/src/agent-ipc-server.ts | 46 +++- .../src/durable-outbound-queue.ts | 53 +++-- .../src/local-subprocess-adapter.ts | 53 ++++- .../local-subprocess/manifest-persistence.ts | 6 +- .../src/local-subprocess/step-spawner.ts | 4 +- .../src/local-subprocess/types.ts | 8 +- .../test/durable-outbound-queue.test.ts | 46 +++- .../test/ipc-ordering-errors.test.ts | 219 ++++++++++++++++++ specs/23-durable-execution.md | 11 +- 11 files changed, 424 insertions(+), 52 deletions(-) create mode 100644 packages/platform-node/test/ipc-ordering-errors.test.ts diff --git a/packages/platform-node/src/agent-ipc-client.ts b/packages/platform-node/src/agent-ipc-client.ts index 489174ed..b769bc52 100644 --- a/packages/platform-node/src/agent-ipc-client.ts +++ b/packages/platform-node/src/agent-ipc-client.ts @@ -544,15 +544,25 @@ export class AgentIpcClient { if (frame.type === 'error') { // Error frames are non-fatal: a single bad frame from a peer or a // transient handler exception should not tear down the chat - // connection. Surface to any in-flight `send` ack waiter so that - // call rejects, and stash the error so callers can inspect it via + // connection. Stash the error so callers can inspect it via // `getLastError()`. Streaming continues. const err = new Error(`server error (${frame.error.kind}): ${frame.error.message}`); this.lastError = err; - // Server only emits acks against `send`; the only deferred we can - // reasonably correlate is one of the pending send waiters. Without - // a messageId on the error frame we can't pinpoint which — fail - // them all so a chained `await client.send(...)` doesn't hang. + const correlatedId = frame.error.messageId; + if (correlatedId !== undefined) { + // The server told us exactly which send failed — reject only that + // waiter. Other in-flight sends may have executed fine; failing + // them too made callers retry messages that already ran. + const pending = this.pendingAcks.get(correlatedId); + if (pending) { + this.pendingAcks.delete(correlatedId); + pending.reject(err); + } + return; + } + // Uncorrelated error (unparseable frame, non-send handler): we can't + // pinpoint a waiter, so fail them all rather than hang a chained + // `await client.send(...)`. for (const pending of this.pendingAcks.values()) { pending.reject(err); } diff --git a/packages/platform-node/src/agent-ipc-protocol.ts b/packages/platform-node/src/agent-ipc-protocol.ts index 76e83804..20b47c80 100644 --- a/packages/platform-node/src/agent-ipc-protocol.ts +++ b/packages/platform-node/src/agent-ipc-protocol.ts @@ -164,6 +164,14 @@ const ErrorFrameSchema = z.object({ error: z.object({ kind: z.string().min(1), message: z.string().min(1), + /** + * When the failed frame was a `send`, the message id it carried — so + * the client can reject exactly the matching ack waiter instead of + * failing every in-flight send (which made unrelated messages that DID + * execute look failed, inviting duplicate retries). Optional: errors + * from unparseable or non-send frames have nothing to correlate. + */ + messageId: z.string().min(1).optional(), }), }); diff --git a/packages/platform-node/src/agent-ipc-server.ts b/packages/platform-node/src/agent-ipc-server.ts index 51108866..a06c3c16 100644 --- a/packages/platform-node/src/agent-ipc-server.ts +++ b/packages/platform-node/src/agent-ipc-server.ts @@ -175,6 +175,13 @@ interface ClientState { buffer: string; /** Whether this client has issued `subscribe`. */ subscribed: boolean; + /** + * Per-client dispatch chain. Frames from one client execute strictly in + * arrival order — a `send` and an `abort` in one TCP chunk must not race, + * and two `send`s must reach `harness.execute` in the order sent. A fire- + * and-forget `void dispatchLine(...)` per line gave no such guarantee. + */ + dispatchChain: Promise; } interface FrameContext { @@ -195,6 +202,17 @@ type FrameHandler = (frame: ClientFrame, ctx: FrameContext) => Promise; */ const MAX_UNIX_SOCKET_PATH_BYTES = 104; +/** + * Disconnect a client whose kernel-side socket buffer backs up past this + * many bytes. A stalled consumer (TUI suspended with ^Z, dead SSH hop) + * otherwise makes Node buffer every broadcast frame in process memory, + * unbounded, while both stream pumps keep producing. Disconnecting is the + * safe move in both modes: a durable client resumes by seq via + * `durableResume` replay and loses nothing; a non-durable client was + * always best-effort streaming. + */ +const MAX_SOCKET_BUFFERED_BYTES = 4 * 1024 * 1024; + function errorMessage(err: unknown): string { return err instanceof Error ? err.message : String(err); } @@ -243,6 +261,18 @@ function writeFrame(socket: Socket, frame: ServerFrame): void { } try { socket.write(encodeFrame(frame)); + /* Backpressure guard: `write()` returning false just means "buffering"; + * that's fine transiently. What is NOT fine is a consumer that never + * drains — cap the buffered bytes and cut the connection. The client's + * `close` handler removes it from the set; a durable client reconnects + * and resumes by seq. */ + if (socket.writableLength > MAX_SOCKET_BUFFERED_BYTES) { + socket.destroy( + new Error( + `agent-ipc-server: client write buffer exceeded ${MAX_SOCKET_BUFFERED_BYTES} bytes (stalled consumer)`, + ), + ); + } } catch { // Synchronous write throw is rare but possible (e.g. socket entered // an erroring state between the `destroyed` check and the call). @@ -673,6 +703,7 @@ export class AgentIpcServer { socket, buffer: '', subscribed: false, + dispatchChain: Promise.resolve(), }; this.clients.add(client); socket.setEncoding('utf8'); @@ -702,7 +733,13 @@ export class AgentIpcServer { const line = client.buffer.slice(0, nl); client.buffer = client.buffer.slice(nl + 1); if (line.length > 0) { - void this.dispatchLine(client, line); + /* Serialize per client: each frame's handler runs after the previous + * one settles, so frames arriving in one chunk keep their order + * (send-then-abort, send-then-send). dispatchLine never rejects + * (its own catch writes an error frame), but chain through catch + * anyway so an unexpected throw can't wedge the chain. */ + const next = client.dispatchChain.then(() => this.dispatchLine(client, line)); + client.dispatchChain = next.catch(() => undefined); } nl = client.buffer.indexOf('\n'); } @@ -745,6 +782,13 @@ export class AgentIpcServer { error: { kind: 'handler_error', message: errorMessage(err), + // Correlate send failures so the client rejects only the matching + // ack waiter — other in-flight sends may have executed fine. + ...(frame.type === 'send' + ? { + messageId: frame.messageId, + } + : {}), }, }); } diff --git a/packages/platform-node/src/durable-outbound-queue.ts b/packages/platform-node/src/durable-outbound-queue.ts index af8536c3..85cff56e 100644 --- a/packages/platform-node/src/durable-outbound-queue.ts +++ b/packages/platform-node/src/durable-outbound-queue.ts @@ -82,9 +82,15 @@ export interface DurableOutboundQueue { * frames as acknowledged. */ ackUpTo(throughSeq: number): Promise; - /** Remove every persisted frame + reset meta — used on a clean shutdown. */ + /** + * Remove every persisted frame and settle the ack watermark — used on a + * clean shutdown. `headSeq` is NOT reset (invariant 1: a seq is never + * reused), so a queue that appends again after `clear()` continues the + * sequence rather than restarting at 1 under clients that still hold a + * pre-clear delivery watermark. + */ clear(): Promise; - /** Number of frames currently persisted. Exposed for test assertions. */ + /** Number of frames currently un-acked (`headSeq - lastAckedSeq`). */ queueSize(): Promise; /** Monotonic head seq — every appended frame gets head+1, head+2, …. */ getHeadSeq(): number; @@ -242,16 +248,14 @@ export async function createDurableOutboundQueue( const effective = Math.min(throughSeq, headSeq); const previous = lastAckedSeq; lastAckedSeq = effective; - // Delete frames in `(previous, effective]`. - const keys = await storage.list(framePrefix); - for (const key of keys) { - const seq = parseSeqFromFrameKey(key, socketId); - if (seq === null) { - continue; - } - if (seq > previous && seq <= effective) { - await storage.delete(key); - } + /* Frame keys are computable from their seq — delete `(previous, + * effective]` directly instead of a storage.list() scan. Acks fire on + * every client watermark push, and a per-ack scan over a namespace that + * shares its backing directory with checkpoints and ledger shards made + * each ack O(total keys). `storage.delete` on an already-gone key is a + * no-op by contract, so gaps (from clear/compaction) cost nothing. */ + for (let seq = previous + 1; seq <= effective; seq++) { + await storage.delete(frameKey(socketId, seq)); } await flushMeta(); } @@ -261,21 +265,22 @@ export async function createDurableOutboundQueue( for (const key of keys) { await storage.delete(key); } - await storage.delete(metaKey(socketId)); - headSeq = 0; - lastAckedSeq = 0; + /* headSeq stays MONOTONIC through a clear — invariant 1 says a seq is + * never reused. Resetting to 0 here meant a queue cleared and then + * appended-to restarted at seq 1, and any client still holding a + * pre-clear `highestDeliveredDurableSeq` watermark would silently drop + * every new frame up to it (client dedupe is `seq <= highestDelivered`). + * Frames are gone; the counter is not. `lastAckedSeq` advances to + * headSeq (everything persisted is deleted ⇒ everything is settled), + * and the meta doc records both so a restart doesn't resurrect 0. */ + lastAckedSeq = headSeq; + await flushMeta(); } async function queueSize(): Promise { - const keys = await storage.list(framePrefix); - let count = 0; - for (const key of keys) { - const seq = parseSeqFromFrameKey(key, socketId); - if (seq !== null && seq > lastAckedSeq) { - count += 1; - } - } - return count; + /* In-memory bounds are authoritative: appends only persist inside + * `(lastAckedSeq, headSeq]` and acks delete below the watermark. */ + return headSeq - lastAckedSeq; } return { diff --git a/packages/platform-node/src/local-subprocess-adapter.ts b/packages/platform-node/src/local-subprocess-adapter.ts index 185d7f74..ad7ac037 100644 --- a/packages/platform-node/src/local-subprocess-adapter.ts +++ b/packages/platform-node/src/local-subprocess-adapter.ts @@ -1,4 +1,5 @@ import { execFileSync, spawn } from 'node:child_process'; +import { readFileSync } from 'node:fs'; import path from 'node:path'; import type { ProcessSubprocessRequest, @@ -68,7 +69,36 @@ function isErrnoException(err: unknown): err is NodeJS.ErrnoException { return typeof err === 'object' && err !== null && 'code' in err; } +/** + * Linux fast path: field 22 of `/proc//stat` is the process start time + * in clock ticks since boot — a stable pid-recycle discriminator readable in + * microseconds, vs ~10ms to fork `ps`. The value only needs to be a stable + * identity token (compared for equality against the persisted manifest), + * not a human-readable date, so the raw tick count is fine. Fields are + * located from the LAST `)` because field 2 (comm) can itself contain + * spaces and parens. + */ +function readProcStatStartTime(pid: number): string | null { + try { + const stat = readFileSync(`/proc/${pid}/stat`, 'utf8'); + const afterComm = stat.slice(stat.lastIndexOf(')') + 2); + const fields = afterComm.split(' '); + // afterComm starts at field 3 ("state"), so starttime (field 22) is index 19. + const starttime = fields[19]; + return starttime && starttime.length > 0 ? `proc:${starttime}` : null; + } catch { + return null; + } +} + function readPidStartTime(pid: number): string | null { + if (process.platform === 'linux') { + const viaProc = readProcStatStartTime(pid); + if (viaProc !== null) { + return viaProc; + } + // Fall through to ps — /proc may be restricted (containers, hardening). + } try { const out = execFileSync( 'ps', @@ -234,7 +264,7 @@ export function createLocalSubprocessAdapter( const now = nowIso(); const pid = child.pid; - const pidStarttime = signaller.startTime(pid); + const pidStarttime = await signaller.startTime(pid); const handle = await save({ id: `subprocess-${crypto.randomUUID()}`, status: 'running', @@ -328,7 +358,7 @@ export function createLocalSubprocessAdapter( ); } - const spawned = spawnStepChild({ + const spawned = await spawnStepChild({ spawnFn, signaller, request, @@ -470,7 +500,7 @@ export function createLocalSubprocessAdapter( if (!manifest) { return null; } - const handle = hydrateFromManifest(manifest, signaller); + const handle = await hydrateFromManifest(manifest, signaller); if (!handle) { // Pid drift or process gone — clear the stale manifest so // subsequent listLive calls don't re-surface it forever. @@ -492,11 +522,18 @@ export function createLocalSubprocessAdapter( } if (storage) { const manifests = await listLocalManifests(storage); - for (const manifest of manifests) { - if (live.has(manifest.handleId)) { - continue; - } - const hydrated = hydrateFromManifest(manifest, signaller); + const candidates = manifests.filter((m) => !live.has(m.handleId)); + /* Hydrate in parallel: each check may read /proc or fork `ps`, and a + * restart with N persisted manifests paid them serially before. Each + * result carries its own manifest so the pairing survives the reorder + * a bare `Promise.all` over handles would invite. */ + const hydratedAll = await Promise.all( + candidates.map(async (manifest) => ({ + manifest, + hydrated: await hydrateFromManifest(manifest, signaller), + })), + ); + for (const { manifest, hydrated } of hydratedAll) { if (!hydrated) { await clearIfDurable(manifest.handleId); continue; diff --git a/packages/platform-node/src/local-subprocess/manifest-persistence.ts b/packages/platform-node/src/local-subprocess/manifest-persistence.ts index 3708a306..e5eed164 100644 --- a/packages/platform-node/src/local-subprocess/manifest-persistence.ts +++ b/packages/platform-node/src/local-subprocess/manifest-persistence.ts @@ -230,14 +230,14 @@ function nowIso(): string { * in either case the process we recorded is no longer the one running * under that pid, so rebinding would be unsafe. */ -export function hydrateFromManifest( +export async function hydrateFromManifest( manifest: LocalManifest, signaller: ProcessSignaller, -): SubprocessHandle | null { +): Promise { if (!signaller.isAlive(manifest.pid)) { return null; } - const currentStart = signaller.startTime(manifest.pid); + const currentStart = await signaller.startTime(manifest.pid); if (manifest.pidStarttime !== null && currentStart !== null) { if (currentStart !== manifest.pidStarttime) { return null; diff --git a/packages/platform-node/src/local-subprocess/step-spawner.ts b/packages/platform-node/src/local-subprocess/step-spawner.ts index 732740d5..bb21b76c 100644 --- a/packages/platform-node/src/local-subprocess/step-spawner.ts +++ b/packages/platform-node/src/local-subprocess/step-spawner.ts @@ -72,7 +72,7 @@ export interface AttachCompletionArgs { * once the handle is registered (so the `on('close')` callback can * close over the factory's handle map + save/clearIfDurable closures). */ -export function spawnStepChild(args: SpawnStepChildArgs): SpawnStepChildResult { +export async function spawnStepChild(args: SpawnStepChildArgs): Promise { let asyncSpawnError: unknown = null; const child = args.spawnFn(args.bootstrapCommand, args.bootstrapArgs, { cwd: args.request.overrides.cwdInit, @@ -100,7 +100,7 @@ export function spawnStepChild(args: SpawnStepChildArgs): SpawnStepChildResult { } const pid = child.pid; - const pidStarttime = args.signaller.startTime(pid); + const pidStarttime = await args.signaller.startTime(pid); // Write the request envelope to stdin as a single newline-terminated JSON // frame. The child parses one frame on boot. diff --git a/packages/platform-node/src/local-subprocess/types.ts b/packages/platform-node/src/local-subprocess/types.ts index 27d20ab9..be8e6d98 100644 --- a/packages/platform-node/src/local-subprocess/types.ts +++ b/packages/platform-node/src/local-subprocess/types.ts @@ -10,5 +10,11 @@ export type SubprocessSignal = 'SIGTERM' | 'SIGSTOP' | 'SIGCONT'; export interface ProcessSignaller { kill(target: number, signal: SubprocessSignal): void; isAlive(pid: number): boolean; - startTime(pid: number): string | null; + /** + * Stable start-time identity token for `pid`, or null when unreadable. + * May be sync or async: the default signaller reads `/proc` (sync, µs) on + * Linux and shells out to `ps` elsewhere; an async implementation lets a + * host avoid blocking the event loop on that fork. Callers must await. + */ + startTime(pid: number): string | null | Promise; } diff --git a/packages/platform-node/test/durable-outbound-queue.test.ts b/packages/platform-node/test/durable-outbound-queue.test.ts index 1f5ea72b..a0af1520 100644 --- a/packages/platform-node/test/durable-outbound-queue.test.ts +++ b/packages/platform-node/test/durable-outbound-queue.test.ts @@ -152,7 +152,7 @@ describe('DurableOutboundQueue', () => { ]); }); - it('clear wipes every persisted frame + meta', async () => { + it('clear wipes every persisted frame but keeps headSeq monotonic', async () => { const storage = createInMemoryStorage(); const queue = await createDurableOutboundQueue({ storage, @@ -161,16 +161,52 @@ describe('DurableOutboundQueue', () => { await queue.append('a'); await queue.append('b'); await queue.clear(); - expect(queue.getHeadSeq()).toBe(0); - expect(queue.getLastAckedSeq()).toBe(0); + /* Invariant 1: a seq is NEVER reused, clear included. The old reset-to-0 + * behaviour made post-clear appends restart at seq 1 — and a client still + * holding a pre-clear delivery watermark (dedupe is `seq <= delivered`) + * silently dropped every new frame up to it. */ + expect(queue.getHeadSeq()).toBe(2); + expect(queue.getLastAckedSeq()).toBe(2); expect(await queue.queueSize()).toBe(0); + expect(await queue.frameRange(1)).toEqual([]); - // A fresh queue on the same storage + socketPath starts clean. + // Appends after clear continue the sequence — never restart. + const entry = await queue.append('c'); + expect(entry.seq).toBe(3); + + // A queue rehydrated from the same storage sees the preserved counters. const next = await createDurableOutboundQueue({ storage, socketPath: '/tmp/clear', }); - expect(next.getHeadSeq()).toBe(0); + expect(next.getHeadSeq()).toBe(3); + }); + + it('ack deletes by computed key: acked frames gone, later frames intact', async () => { + const storage = createInMemoryStorage(); + const queue = await createDurableOutboundQueue({ + storage, + socketPath: '/tmp/computed-ack', + }); + for (const f of [ + 'f1', + 'f2', + 'f3', + 'f4', + ]) { + await queue.append(f); + } + await queue.ackUpTo(2); + expect((await queue.frameRange(1)).map((e) => e.frame)).toEqual([ + 'f3', + 'f4', + ]); + expect(await queue.queueSize()).toBe(2); + // Ack past head clamps; ack below watermark is a no-op. + await queue.ackUpTo(99); + expect(await queue.queueSize()).toBe(0); + await queue.ackUpTo(1); + expect(queue.getLastAckedSeq()).toBe(4); }); it('queueSize reflects only frames above the ack watermark', async () => { diff --git a/packages/platform-node/test/ipc-ordering-errors.test.ts b/packages/platform-node/test/ipc-ordering-errors.test.ts new file mode 100644 index 00000000..ab39e645 --- /dev/null +++ b/packages/platform-node/test/ipc-ordering-errors.test.ts @@ -0,0 +1,219 @@ +/** + * IPC hardening (platform-node slice review P3/P4). + * + * P3: frames from one client execute strictly in arrival order — a batch of + * sends in a single TCP chunk must reach `harness.execute` in order (the + * old fire-and-forget dispatch let them interleave). + * P4: a server `error` frame carrying a `messageId` rejects ONLY that send's + * ack waiter — other in-flight sends may have executed and must not be + * failed into a duplicate retry. + */ + +import { afterEach, beforeEach, describe, expect, it } from 'bun:test'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import type { HarnessStatus, StreamEvent, StreamingItem } from '@noetic-tools/core'; +import { AgentIpcClient } from '../src/agent-ipc-client'; +import type { + ChatHistoryStore, + IpcAskUserService, + IpcHarness, + TaskLogger, +} from '../src/agent-ipc-server'; +import { AgentIpcServer } from '../src/agent-ipc-server'; +import { createLocalFsAdapter } from '../src/local-fs-adapter'; + +const noopChatStore: ChatHistoryStore = { + async readChatHistory() { + return []; + }, + async appendChatItem() {}, +}; +const noopLogger: TaskLogger = async () => {}; +const stubAskUser: IpcAskUserService = { + peek: () => null, + handleResolve: () => {}, + handleCancel: () => {}, + cancelAll: () => {}, +}; + +/** + * A stream that stays pending until `close()` is called, then ends. The + * server's pumps must be able to finish or `server.close()` never settles + * (same teardown contract the durable-resume tests follow). + */ +function closableStream(): { + iter: AsyncIterable; + close(): void; +} { + let closed = false; + let notify: (() => void) | null = null; + return { + iter: { + [Symbol.asyncIterator](): AsyncIterator { + return { + async next(): Promise> { + if (!closed) { + await new Promise((resolve) => { + notify = resolve; + }); + } + return { + value: undefined, + done: true, + }; + }, + }; + }, + }, + close() { + closed = true; + notify?.(); + }, + }; +} + +/** Both stub harnesses below report idle; neither test drives status. */ +const IDLE_STATUS: HarnessStatus = { + kind: 'idle', +}; + +let dir: string; +let socketPath: string; + +beforeEach(() => { + dir = mkdtempSync(path.join(tmpdir(), 'noetic-ipc-ord-')); + socketPath = path.join(dir, 's.sock'); +}); + +afterEach(() => { + rmSync(dir, { + recursive: true, + force: true, + }); +}); + +function makeServer(harness: IpcHarness): AgentIpcServer { + return new AgentIpcServer({ + harness, + chatHistoryStore: noopChatStore, + logger: noopLogger, + taskId: 'T-ord', + role: 'planner', + runnerId: 'r1', + threadId: 'thread-1', + socketPath, + askUserService: stubAskUser, + fs: createLocalFsAdapter(), + }); +} + +describe('P3 — per-client frame ordering', () => { + it('a burst of sends reaches harness.execute strictly in order', async () => { + const executed: string[] = []; + const items = closableStream(); + const events = closableStream(); + const harness: IpcHarness = { + async execute(input: string) { + // Async gap: the old fire-and-forget dispatch let a later frame's + // handler overtake this await. + await new Promise((resolve) => setTimeout(resolve, Math.random() * 5)); + executed.push(input); + }, + getItemStream: () => items.iter, + getFullStream: () => events.iter, + getStatus: () => IDLE_STATUS, + abort: async () => {}, + }; + const server = makeServer(harness); + await server.listen(); + + const client = new AgentIpcClient({ + socketPath, + }); + await client.connect(); + + const sends = Array.from( + { + length: 8, + }, + (_, i) => + client.send({ + messageId: `m-${i}`, + text: `msg-${i}`, + }), + ); + + await Promise.all(sends); + + expect(executed).toEqual( + Array.from( + { + length: 8, + }, + (_, i) => `msg-${i}`, + ), + ); + items.close(); + events.close(); + await server.close('test-end'); + client.close(); + }); +}); + +describe('P4 — correlated error frames', () => { + it('a failing send rejects only its own ack; concurrent sends still resolve', async () => { + const items = closableStream(); + const events = closableStream(); + const harness: IpcHarness = { + async execute(input: string) { + if (input === 'poison') { + throw new Error('scripted execute failure'); + } + }, + getItemStream: () => items.iter, + getFullStream: () => events.iter, + getStatus: () => IDLE_STATUS, + abort: async () => {}, + }; + const server = makeServer(harness); + await server.listen(); + const client = new AgentIpcClient({ + socketPath, + }); + await client.connect(); + + const good1 = client.send({ + messageId: 'good-1', + text: 'fine', + }); + const bad = client.send({ + messageId: 'bad-1', + text: 'poison', + }); + const good2 = client.send({ + messageId: 'good-2', + text: 'also fine', + }); + + // The poisoned send rejects with the handler error… + let thrown: unknown; + try { + await bad; + } catch (e) { + thrown = e; + } + expect(String(thrown)).toContain('scripted execute failure'); + // …while the unrelated sends resolve normally (the old blanket + // rejection failed all three). + await good1; + await good2; + expect(client.getLastError()).not.toBeNull(); + + items.close(); + events.close(); + await server.close('test-end'); + client.close(); + }); +}); diff --git a/specs/23-durable-execution.md b/specs/23-durable-execution.md index 404111f5..8b98968e 100644 --- a/specs/23-durable-execution.md +++ b/specs/23-durable-execution.md @@ -191,7 +191,7 @@ interface LocalSubprocessManifest { Entries live under the harness's `StorageAdapter`, default root `~/.noetic/subprocess/` (via `createFileStorage({root: resolveSubprocessRoot()})`). On `reattach(handleId)` the adapter: 1. Loads the manifest entry. -2. Re-queries `pidStarttime` against `ps -p -o lstart=`. A mismatch means the pid was recycled and the original child is gone — the handle is marked `stopped` and the manifest cleared. A match means the original child is still alive. +2. Re-queries `pidStarttime` — on Linux via the `/proc//stat` starttime fast path (field 22, read in microseconds), falling back to `ps -p -o lstart=` elsewhere or when `/proc` is restricted. `ProcessSignaller.startTime` may be sync or async; callers must await. A mismatch means the pid was recycled and the original child is gone — the handle is marked `stopped` and the manifest cleared. A match means the original child is still alive. 3. Rebinds the unix-domain socket at `socketPath` so the IPC server resumes accepting frames. 4. Returns a rehydrated `SubprocessHandle` whose `status` reflects current liveness. @@ -248,7 +248,8 @@ Invariants: - `headSeq` is the highest seq ever assigned. Appends start at `headSeq + 1`. Monotonic across the queue's lifetime; never reused even after full compaction. - `lastAckedSeq <= headSeq`. No frame with `seq <= lastAckedSeq` is persisted. -- `ackUpTo(seq)` advances `lastAckedSeq` and deletes every frame in `(previousAck, seq]`. +- `ackUpTo(seq)` advances `lastAckedSeq` and deletes every frame in `(previousAck, seq]` — by computed key, not a storage scan; `storage.delete` on an already-gone key is a no-op, so gaps cost nothing. +- `clear()` removes every persisted frame and settles the ack watermark (`lastAckedSeq := headSeq`) but does **not** reset `headSeq` — a queue that appends again continues the sequence, so clients holding a pre-clear delivery watermark don't silently drop new frames. `queueSize()` is `headSeq - lastAckedSeq`. Recovery: on load, the queue walks the `frame:` prefix and merges with the cached `meta` doc. A crash between frame write and meta flush leaves a frame whose seq is above the cached `headSeq`; the scan detects it and bumps `headSeq = max(cachedHead, scannedSeq)`. @@ -293,6 +294,12 @@ A server that composes a `DurableOutboundQueue` wraps every non-handshake outbou On a client reconnect that carries a `durableResume { ackedThrough }` handshake, the server replays `queue.frameRange(ackedThrough + 1)` in order before resuming live emission. On a `durableAck { throughSeq }` the server calls `queue.ackUpTo(throughSeq)` to compact. +Server-side dispatch and backpressure guarantees: + +- **Ordered dispatch** — frames from one client execute strictly in arrival order via a per-client promise chain, so a `send` and an `abort` arriving in one TCP chunk cannot race and two `send`s reach `harness.execute` in send order. +- **Broadcast backpressure** — a client whose kernel-side socket buffer exceeds 4 MiB (stalled consumer: suspended TUI, dead SSH hop) is disconnected rather than buffered unboundedly. A durable client reconnects and resumes by seq; a non-durable client was always best-effort. +- **Correlated errors** — error frames for a failed `send` carry that send's `messageId`, so the client rejects only the matching ack waiter instead of failing every in-flight send (which made successfully executed messages look failed and invited duplicate retries). Uncorrelated errors still fail all waiters rather than hang a chained `await client.send(...)`. + ### Client integration Clients track the highest seq they have successfully processed. On every reconnect they send `durableResume { ackedThrough: }` immediately after the server's `hello`. For each `durable` frame received, they re-parse `frame` as a `ServerFrame`, apply it, and eventually send a `durableAck` once the frame is durably consumed by the application (persisted to the chat JSONL, rendered to the TUI, etc.).