diff --git a/src/core/engine.ts b/src/core/engine.ts index 6a21328..1097a56 100644 --- a/src/core/engine.ts +++ b/src/core/engine.ts @@ -367,7 +367,7 @@ function clampOptionalPositive(value: number | undefined | null, min: number, ma } /** Conversation budget guard shared by sends and every handover attempt. */ -function budgetRefusal(channel: Channel, now: number): string | null { +export function budgetRefusal(channel: Channel, now: number): string | null { const budgets = channel.budgets if (!budgets) return null if (budgets.max_runtime_ms !== null && now - channel.created_at > budgets.max_runtime_ms) { @@ -1083,8 +1083,12 @@ export function pendingRecipients(state: State): string[] { * OWN channel's name (a session may sit in multiple channels; provenance in * the untrusted-message framing must never borrow another channel's label). */ -export function drainForDelivery(state: State, recipientSessionId: string): DeliveryPair[] { - const delivered = drainQueue(state, recipientSessionId) +export function drainForDelivery( + state: State, + recipientSessionId: string, + opts: { now?: number; canDeliver?: (msg: MessageEnvelope) => boolean } = {}, +): DeliveryPair[] { + const delivered = drainQueue(state, recipientSessionId, opts) return delivered.map((envelope) => ({ message_id: envelope.message_id, channel_name: Object.values(state.channels).find((c) => c.id === envelope.channel_id)?.name ?? "(unknown channel)", @@ -1253,18 +1257,21 @@ export function drainQueue( continue } - if (opts.canDeliver && !opts.canDeliver(msg)) { + // Channel lifetime/runtime budget before argv gates so an exhausted + // channel cannot consume the batch argv allowance and block another + // channel's eligible mail. + if (budgetRefusal(channel, now)) { remaining.push(id) continue } - // Sending can queue many envelopes before the first handover. Recheck - // for each attempt, including retries, so a batch cannot overrun its - // lifetime cap or deliver work after its runtime budget has elapsed. - // Keep the pending envelope visible for an explicit budget extension. - if (budgetRefusal(channel, now)) { + // Strict FIFO under argv/size gates: once the head does not fit, stop. + // Do not skip a large message to deliver a later smaller one. + if (opts.canDeliver && !opts.canDeliver(msg)) { remaining.push(id) - continue + const idx = queue.indexOf(id) + for (const rest of queue.slice(idx + 1)) remaining.push(rest) + break } msg.delivery_status = "in_flight" diff --git a/src/hosts/spawn-delivery.ts b/src/hosts/spawn-delivery.ts index 318029e..48a76e8 100644 --- a/src/hosts/spawn-delivery.ts +++ b/src/hosts/spawn-delivery.ts @@ -1,4 +1,4 @@ -/** +/** * Resume delivery uses argv-only child processes and persisted two-phase queues. * The host exposes no busy-check API; concurrent TUI turns may interleave. * Codex TUI-session resume remains unverified. Hosts without a verified native @@ -9,12 +9,14 @@ import { execFile } from "node:child_process" import { existsSync } from "node:fs" import type { Member, State } from "../core/types.js" import { + budgetRefusal, commitDelivery, drainForDelivery, effectiveEndpointCapabilities, formatUntrustedMessage, requeueFailedDelivery, } from "../core/engine.js" +import type { MessageEnvelope } from "../core/types.js" export interface SpawnCommand { /** Executable to spawn (resolved or bare name — resolved on PATH). */ @@ -320,81 +322,68 @@ async function deliverViaSpawnInner(deps: SpawnRunnerDeps, member: Member): Prom const refusal = spawnDeliveryRefusal(member) if (refusal) return { status: "skipped", detail: refusal } - // Pre-spawn argv size guard (P2-2): estimate the framed batch size from - // the PENDING queue without draining. Framing adds ~450 chars per - // envelope; the builder's fixed argv adds a little more. + // Argv size guard (issue #3 / P2-2): partial FIFO drain under platform budget. + // JOIN separators are only charged *between* framed messages (not on the head). + // Framing uses spawnArgumentCost(formatUntrustedMessage(...)) so estimates match + // the argv that is actually spawned. const builder = spawnBuilderFor(member.host)! const budget = spawnArgvBudget() - let estimated = 0 - let oversizedIds: string[] = [] + const JOIN = "\n\n---\n\n" + const joinCost = spawnArgumentCost(JOIN) + + let batch: Array<{ id: string; channelName: string }> = [] + let framed = "" + let headOversizedId: string | null = null try { await deps.withLock(() => { const state = deps.load() const cmd = builder.buildResumeCommand({ hostSessionId: member.host_session_id as string, cwd: deps.cwd }) const fixedOverhead = [cmd.command, ...cmd.args].reduce((sum, arg) => sum + spawnArgumentCost(arg), 64) + + // First *eligible* message oversized? Refuse without draining. + // Channel eligibility (paused / budget) is checked before argv size so an + // oversized ineligible head cannot block other channels' mail. + // Accounting matches canDeliver: no JOIN cost on the first framed message. + const now = Date.now() for (const id of state.queues[member.session_id] ?? []) { const msg = state.messages[id] if (!msg || msg.delivery_status !== "pending") continue + if (msg.recipient_session_id !== member.session_id) continue + const channel = Object.values(state.channels).find((c) => c.id === msg.channel_id) + if (!channel) continue + if (channel.paused) continue + if (budgetRefusal(channel, now)) continue + const channelName = channel.name + const headCost = fixedOverhead + spawnArgumentCost(formatUntrustedMessage(msg, channelName)) + if (headCost > budget) headOversizedId = id + break + } + if (headOversizedId) return null + + // Running argv cost of messages already accepted into this batch. + let batchArgv = 0 + let count = 0 + const canDeliver = (msg: MessageEnvelope) => { const channelName = Object.values(state.channels).find((channel) => channel.id === msg.channel_id)?.name ?? msg.channel_id - const framedEstimate = spawnArgumentCost(formatUntrustedMessage(msg, channelName)) - if (fixedOverhead + estimated + framedEstimate > budget) oversizedIds.push(id) - else estimated += framedEstimate + 6 + const framedCost = spawnArgumentCost(formatUntrustedMessage(msg, channelName)) + const sep = count === 0 ? 0 : joinCost + if (fixedOverhead + batchArgv + sep + framedCost > budget) return false + batchArgv += sep + framedCost + count += 1 + return true } - return null - }) - } catch (error) { - deps.recordError(`Spawn delivery size check for ${member.session_id} failed: ${(error as Error).message}`) - return { status: "failed", detail: "invalid host command; pending queue was preserved" } - } - if (oversizedIds.length > 0) { - // Do NOT drain: leave the queue untouched so the operator can shrink - // content or switch the member to pull. Record once per message id per - // process lifetime to avoid error spam. - const fresh = oversizedIds.filter((id) => !sizeGuardReported.has(id)) - for (const id of fresh) sizeGuardReported.add(id) - if (fresh.length > 0) { - deps.recordError( - `Spawn delivery refused for ${member.session_id}: batch exceeds the spawn argv limit (~${budget} chars on ${process.platform}). Reduce message size or switch this member to pull delivery. Message ids: ${fresh.join(", ")}`, - ) - } - return { status: "skipped", detail: `batch exceeds the spawn argv limit (~${budget} chars); queue left untouched` } - } - // Phase 1 (locked): drain + persist in_flight. - let batch: Array<{ id: string; channelName: string }> = [] - let framed = "" - let budgetExceeded = false - try { - await deps.withLock(() => { - const state = deps.load() - // Recheck under the SAME lock as drain: new mail can arrive after the - // earlier diagnostic guard. Snapshot the exact envelopes before save. - const cmd = builder.buildResumeCommand({ hostSessionId: member.host_session_id as string, cwd: deps.cwd }) - const pending = (state.queues[member.session_id] ?? []) - .flatMap((id) => { - const msg = state.messages[id] - if (!msg || msg.delivery_status !== "pending") return [] - const name = - Object.values(state.channels).find((channel) => channel.id === msg.channel_id)?.name ?? msg.channel_id - return [formatUntrustedMessage(msg, name)] - }) - .join("\n\n---\n\n") - const cost = [cmd.command, ...cmd.args, pending].reduce((sum, arg) => sum + spawnArgumentCost(arg), 64) - if (cost > budget) { - budgetExceeded = true - return null - } const queueSnapshot = () => JSON.stringify((state.queues[member.session_id] ?? []).map((id) => [id, state.messages[id]?.delivery_status])) const before = queueSnapshot() - const pairs = drainForDelivery(state, member.session_id) + const pairs = drainForDelivery(state, member.session_id, { canDeliver }) framed = pairs .flatMap((pair) => { const msg = state.messages[pair.message_id] return msg ? [formatUntrustedMessage(msg, pair.channel_name)] : [] }) - .join("\n\n---\n\n") + .join(JOIN) // Terminal queue cleanup is a durable outcome even without a spawn. if (pairs.length > 0 || before !== queueSnapshot()) deps.save(state) batch = pairs.map((p) => ({ id: p.message_id, channelName: p.channel_name })) @@ -404,11 +393,18 @@ async function deliverViaSpawnInner(deps: SpawnRunnerDeps, member: Member): Prom deps.recordError(`Spawn delivery drain for ${member.session_id} failed: ${(error as Error).message}`) return { status: "failed", detail: `drain failed: ${(error as Error).message}` } } - if (budgetExceeded) + if (headOversizedId) { + if (!sizeGuardReported.has(headOversizedId)) { + sizeGuardReported.add(headOversizedId) + deps.recordError( + `Spawn delivery refused for ${member.session_id}: message exceeds the spawn argv limit (~${budget} chars on ${process.platform}). Reduce message size or switch this member to pull delivery. Message ids: ${headOversizedId}`, + ) + } return { status: "skipped", detail: `batch exceeds the spawn argv limit (~${budget} chars); queue left untouched` } + } if (batch.length === 0) return { status: "skipped", detail: "no pending messages" } if (!framed) { - // Snapshots lost the envelopes (pruned concurrently) — requeue by ids. + // Snapshots lost the envelopes (pruned concurrently) — requeue by ids. try { await deps.withLock(() => { const state = deps.load() @@ -430,11 +426,9 @@ async function deliverViaSpawnInner(deps: SpawnRunnerDeps, member: Member): Prom cwd: deps.cwd, }) - deps.setDelivering?.(member.session_id, true) try { const result = await deps.spawn(cmd, framed) if (result.ok) { - // Phase 2 (locked): the host CLI accepted + ran the turn. await deps.withLock(() => { const state = deps.load() commitDelivery( @@ -450,13 +444,10 @@ async function deliverViaSpawnInner(deps: SpawnRunnerDeps, member: Member): Prom detail: `resumed ${builder.binary} session ${member.host_session_id} with ${batch.length} message(s)`, } } - // Failure: requeue in original FIFO order + visible error. Windows - // npm .cmd/.bat shims cannot be spawned without a shell (Node's - // CVE-2024-27980 fix) — enrich EINVAL with the actionable override. const failureDetail = /\bEINVAL\b/.test(result.error) || isWindowsShimPath(cmd.command) ? result.error + - ' | Windows shim/no-shell spawn failed: set OPENCOMMS_CLAUDE_BIN / OPENCOMMS_CODEX_BIN to a native executable or a command template, e.g. OPENCOMMS_CODEX_BIN="node C:\\path\\to\\codex.js"' + ' | Windows shim/no-shell spawn failed: set OPENCOMMS_CLAUDE_BIN / OPENCOMMS_CODEX_BIN to a native executable or a command template, e.g. OPENCOMMS_CODEX_BIN="node C:\\\\path\\\\to\\\\codex.js"' : result.error await deps.withLock(() => { const state = deps.load() @@ -488,7 +479,6 @@ async function deliverViaSpawnInner(deps: SpawnRunnerDeps, member: Member): Prom /* best effort */ } return { status: "failed", detail: (error as Error).message } - } finally { - deps.setDelivering?.(member.session_id, false) } } + diff --git a/test/unit/core/spawn-delivery.test.ts b/test/unit/core/spawn-delivery.test.ts index fa5b215..39cdfaa 100644 --- a/test/unit/core/spawn-delivery.test.ts +++ b/test/unit/core/spawn-delivery.test.ts @@ -11,12 +11,13 @@ import { parseCommandTemplate, resolveBinaryOverride, spawnArgvBudget, + spawnArgumentCost, isWindowsShimPath, defaultSpawn, type SpawnCommand, type SpawnRunnerDeps, } from "../../../src/hosts/spawn-delivery.js" -import { createChannel, joinChannel, sendMessage } from "../../../src/core/engine.js" +import { createChannel, joinChannel, sendMessage, formatUntrustedMessage } from "../../../src/core/engine.js" import { emptyState } from "../../../src/core/store.js" import type { Member, State } from "../../../src/core/types.js" @@ -340,29 +341,119 @@ test("P2-1: spawn failure with EINVAL carries the actionable shim hint", async ( assert.equal((state.queues[member.session_id] ?? []).length, 1) }) -test("P2-2: oversized batch refused BEFORE draining; queue untouched; no spawn", async () => { +// ── P2-2: argv size guard (partial FIFO drain) ── + +test("P2-2: head message alone oversized refused; queue untouched; no spawn", async () => { const state = emptyState() const member = makeMember() seedChannelWithSpawnMember(state, member) - // Three 45k-char messages (each under the 100k single-message send limit) - // sum past every platform budget (win32 30k / POSIX 120k). - for (let i = 0; i < 3; i++) { - const sent = sendMessage(state, { channel: "spawn-ch", content: `m${i} ` + "x".repeat(45_000) }, "sess_sender") - assert.equal(sent.ok, true) - } + // Max content is 100_000 chars. Multi-byte UTF-8 makes framed cost exceed the 120k POSIX budget. + // (99k ASCII still fits after framing on Linux — that fixture was incorrect.) + const huge = "\u00e9".repeat(100_000) + const sent = sendMessage(state, { channel: "spawn-ch", content: huge }, "sess_sender") + assert.equal(sent.ok, true) + const headId = (sent.data as { message_ids: string[] }).message_ids[0]! const { deps, spawned, errors } = makeFakeDeps(state) const outcome = await deliverViaSpawn(deps, member) assert.equal(outcome.status, "skipped") assert.match(outcome.detail, /argv limit/) assert.equal(spawned.length, 0, "no doomed spawn attempted") - assert.equal((state.queues[member.session_id] ?? []).length, 3, "queue left untouched") - for (const m of Object.values(state.messages)) assert.equal(m.delivery_status, "pending", "not drained") - assert.ok(errors.some((e) => e.includes("spawn argv limit") && e.includes("switch this member to pull"))) + assert.equal((state.queues[member.session_id] ?? []).length, 1, "queue left untouched") + assert.equal(state.messages[headId]!.delivery_status, "pending") + assert.ok(errors.some((e) => e.includes("spawn argv limit"))) const outcome2 = await deliverViaSpawn(deps, member) assert.equal(outcome2.status, "skipped") - assert.equal(errors.filter((e) => e.includes("spawn argv limit")).length, 1) + assert.equal(errors.filter((e) => e.includes("spawn argv limit")).length, 1, "error deduped") +}) + +test("P2-2 boundary: first-message JOIN_SEP must not be charged (UTF-8)", async () => { + if (process.platform === "win32") return + const state = emptyState() + const member = makeMember() + seedChannelWithSpawnMember(state, member) + const sent0 = sendMessage(state, { channel: "spawn-ch", content: "probe" }, "sess_sender") + assert.equal(sent0.ok, true) + const headId = (sent0.data as { message_ids: string[] }).message_ids[0]! + const msg = state.messages[headId]! + const cmd = CLAUDE_CODE_SPAWN.buildResumeCommand({ + hostSessionId: member.host_session_id as string, + cwd: "C:\\proj", + }) + const fixedOverhead = [cmd.command, ...cmd.args].reduce((sum, arg) => sum + spawnArgumentCost(arg), 64) + const budget = spawnArgvBudget() + const emptyFramed = formatUntrustedMessage({ ...msg, content: "" }, "spawn-ch") + const wrapperCost = spawnArgumentCost(emptyFramed) + const room = budget - fixedOverhead - wrapperCost - 200 + assert.ok(room > 1000, "expected room for a fitting payload under the platform budget") + const content = "x".repeat(Math.min(room, 100_000)) + msg.content = content + const headCost = fixedOverhead + spawnArgumentCost(formatUntrustedMessage(msg, "spawn-ch")) + assert.ok(headCost <= budget, `expected fitting headCost=${headCost} budget=${budget}`) + const { deps, spawned } = makeFakeDeps(state) + const outcome = await deliverViaSpawn(deps, member) + assert.equal(outcome.status, "delivered", outcome.detail) + assert.equal(spawned.length, 1) + assert.ok(String(spawned[0]?.message ?? "").includes(content.slice(0, 32))) + assert.equal((state.queues[member.session_id] ?? []).length, 0) + assert.equal(state.messages[headId]!.delivery_status, "delivered") +}) + +test("P2-2b-strict: only the first fitting message drains; later ids stay queued in order", async () => { + if (process.platform === "win32") return + const state = emptyState() + const member = makeMember() + seedChannelWithSpawnMember(state, member) + // 80k / 50k / 5k — only the first fits under the ~120k POSIX framed budget. + // A smaller third message must not leapfrog the blocked 50k second message. + const sizes = [80_000, 50_000, 5_000] + const ids: string[] = [] + for (let i = 0; i < sizes.length; i++) { + const sent = sendMessage( + state, + { channel: "spawn-ch", content: `m${i}:` + "x".repeat(sizes[i]!) }, + "sess_sender", + ) + assert.equal(sent.ok, true) + ids.push((sent.data as { message_ids: string[] }).message_ids[0]!) + } + const { deps, spawned } = makeFakeDeps(state) + const outcome = await deliverViaSpawn(deps, member) + assert.equal(outcome.status, "delivered", outcome.detail) + assert.equal(spawned.length, 1, "one spawn with only the first message") + const payload = String(spawned[0]?.message ?? "") + assert.ok(payload.includes("m0:"), "first message content present") + assert.equal(payload.includes("m1:"), false, "second message must not appear in spawn payload") + assert.equal(payload.includes("m2:"), false, "third message must not appear in spawn payload") + assert.deepEqual(state.queues[member.session_id], [ids[1], ids[2]], "remaining queue is exact second then third") + assert.equal(state.messages[ids[0]!]!.delivery_status, "delivered") + assert.equal(state.messages[ids[1]!]!.delivery_status, "pending") + assert.equal(state.messages[ids[2]!]!.delivery_status, "pending") +}) + + +test("P2-2b: FIFO partial drain when batch exceeds budget (issue #3)", async () => { + if (process.platform === "win32") return + const state = emptyState() + const member = makeMember() + seedChannelWithSpawnMember(state, member) + const ids: string[] = [] + for (let i = 0; i < 3; i++) { + const sent = sendMessage(state, { channel: "spawn-ch", content: `m${i} ` + "x".repeat(45_000) }, "sess_sender") + assert.equal(sent.ok, true) + ids.push((sent.data as { message_ids: string[] }).message_ids[0]!) + } + const { deps, spawned, errors } = makeFakeDeps(state) + + const outcome = await deliverViaSpawn(deps, member) + assert.equal(outcome.status, "delivered", "first fitting prefix should deliver") + assert.equal(spawned.length, 1, "one spawn with the fitting prefix") + assert.equal((state.queues[member.session_id] ?? []).length, 1, "third message remains queued") + assert.deepEqual(state.queues[member.session_id], [ids[2]]) + const pending = Object.values(state.messages).filter((m) => m.delivery_status === "pending") + assert.equal(pending.length, 1, "exactly one message still pending") + assert.equal(errors.filter((e) => e.includes("spawn argv limit")).length, 0, "deferred messages are not errors") }) test("spawn argv budget is platform-aware and conservative", () => {