fix: FIFO partial drain under spawn argv budget (#3)

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"

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)
   }
 }
+

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", () => {