From 47a9192e282b02106d02c2b4fede7da54d5fd513 Mon Sep 17 00:00:00 2001 From: wized2 Date: Sun, 20 Sep 2026 09:59:14 +0000 Subject: [PATCH 1/3] fix: FIFO partial drain under spawn argv budget (#3) spawn_push refused the entire pending queue whenever the *combined* framed size exceeded the platform argv limit, even when the head of the queue would fit. That left recipients stalled forever once the backlog grew past the budget. - drainForDelivery now accepts the same opts as drainQueue (canDeliver) - deliverViaSpawn drains every message that still fits (FIFO), leaving the rest pending for the next cycle - Only refuse delivery when the head message alone cannot fit; queue stays untouched and the operator is pointed at pull delivery - Rewrite P2-2 for head-oversized; add P2-2b for partial drain (issue #3) --- src/core/engine.ts | 8 ++- src/hosts/spawn-delivery.ts | 74 ++++++++++++++------------- test/unit/core/spawn-delivery.test.ts | 57 ++++++++++++++++----- 3 files changed, 87 insertions(+), 52 deletions(-) diff --git a/src/core/engine.ts b/src/core/engine.ts index 3275860..515fb94 100644 --- a/src/core/engine.ts +++ b/src/core/engine.ts @@ -1115,8 +1115,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)", diff --git a/src/hosts/spawn-delivery.ts b/src/hosts/spawn-delivery.ts index c46a275..baf8c93 100644 --- a/src/hosts/spawn-delivery.ts +++ b/src/hosts/spawn-delivery.ts @@ -82,9 +82,11 @@ const BUILDERS: Record = { /** * Windows argv budget (Reviewer P2-2): CreateProcess caps the WHOLE command * line at 32,767 chars; POSIX caps a single arg at 128 KiB - * (MAX_ARG_STRLEN). Keep conservative headroom. Oversized batches are - * refused BEFORE draining (queued mail preserved, never truncated) with an - * actionable error. + * (MAX_ARG_STRLEN). Keep conservative headroom. + * + * Delivery policy (issue #3): FIFO partial drain under the budget. Only the + * head message being individually oversized refuses delivery (queue intact). + * Later messages that do not fit remain pending for the next cycle. */ export const SPAWN_ARGV_BUDGET: Record = { win32: 30_000, default: 120_000 } @@ -308,51 +310,42 @@ export async function deliverViaSpawn(deps: SpawnRunnerDeps, member: Member): Pr return { status: "skipped", detail: "a spawn delivery is already in flight for this member" } } - // 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): framing adds ~450 chars per envelope; + // the builder's fixed argv adds a little more. FIFO: drain every message + // that still fits; leave the rest pending. Only refuse when the *head* + // pending message alone cannot fit under the platform budget. const builder = spawnBuilderFor(member.host)! const budget = spawnArgvBudget() - let estimated = 0 - let oversizedIds: string[] = [] + const FRAMING_PER_MSG = 450 + const JOIN_SEP = 6 // "\n\n---\n\n".length + + // Phase 1 (locked): head-only oversized refusal, else partial drain. + 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.length + cmd.args.join(" ").length + 64 + for (const id of state.queues[member.session_id] ?? []) { const msg = state.messages[id] if (!msg || msg.delivery_status !== "pending") continue - const framedEstimate = msg.content.length + 450 - if (fixedOverhead + estimated + framedEstimate > budget) oversizedIds.push(id) - else estimated += framedEstimate + 6 + const headCost = fixedOverhead + msg.content.length + FRAMING_PER_MSG + if (headCost > budget) headOversizedId = id + break } - return null - }) - } catch (error) { - deps.recordError(`Spawn delivery size check for ${member.session_id} failed: ${(error as Error).message}`) - } - 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` } - } + if (headOversizedId) return null - // Phase 1 (locked): drain + persist in_flight. - let batch: Array<{ id: string; channelName: string }> = [] - let framed = "" - try { - await deps.withLock(() => { - const state = deps.load() - const pairs = drainForDelivery(state, member.session_id) + let used = fixedOverhead + const canDeliver = (msg: { content: string }) => { + const cost = msg.content.length + FRAMING_PER_MSG + JOIN_SEP + if (used + cost > budget) return false + used += cost + return true + } + const pairs = drainForDelivery(state, member.session_id, { canDeliver }) if (pairs.length > 0) deps.save(state) batch = pairs.map((p) => ({ id: p.message_id, channelName: p.channel_name })) return null @@ -361,6 +354,15 @@ export async function deliverViaSpawn(deps: SpawnRunnerDeps, member: Member): Pr 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 (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" } // Frame the batch from a fresh snapshot (per-envelope channel provenance). diff --git a/test/unit/core/spawn-delivery.test.ts b/test/unit/core/spawn-delivery.test.ts index aeca659..2eef475 100644 --- a/test/unit/core/spawn-delivery.test.ts +++ b/test/unit/core/spawn-delivery.test.ts @@ -18,6 +18,7 @@ import { parseCommandTemplate, resolveBinaryOverride, spawnArgvBudget, + SPAWN_ARGV_BUDGET, isWindowsShimPath, type SpawnCommand, type SpawnRunnerDeps, @@ -306,12 +307,45 @@ test("P2-1: spawn failure with EINVAL carries the actionable shim hint", async ( // ── P2-2: argv size guard ── -test("P2-2: oversized batch refused BEFORE draining; queue untouched; no spawn", async () => { +test("P2-2: head message alone oversized refused; queue untouched; no spawn", async () => { + const state = emptyState() + const member = makeMember() + seedChannelWithSpawnMember(state, member) + // Force a tiny budget so a normal-sized message is head-oversized on any platform. + const key = process.platform === "win32" ? "win32" : "default" + const prev = SPAWN_ARGV_BUDGET[key] + SPAWN_ARGV_BUDGET[key] = 500 + try { + const sent = sendMessage(state, { channel: "spawn-ch", content: "x".repeat(2_000) }, "sess_sender") + assert.equal(sent.ok, true) + 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, 1, "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"))) + + // Repeat attempt: no duplicate error spam (guard dedups per message id). + const outcome2 = await deliverViaSpawn(deps, member) + assert.equal(outcome2.status, "skipped") + assert.equal(errors.filter((e) => e.includes("spawn argv limit")).length, 1) + } finally { + SPAWN_ARGV_BUDGET[key] = prev! + } +}) + +test("P2-2b: FIFO partial drain when batch exceeds budget (issue #3)", async () => { + // Force POSIX-sized budget path: three ~45k messages → first two fit under + // 120k, third stays pending. On win32 (30k) the head alone is oversized; + // skip the partial-drain assertion on that platform. + if (process.platform === "win32") return + 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) @@ -319,17 +353,12 @@ test("P2-2: oversized batch refused BEFORE draining; queue untouched; no spawn", 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"))) - - // Repeat attempt: no duplicate error spam (guard dedups per message id). - 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(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") + 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", () => { From f5ed190b558b6fa18b817c0020bc215dc03ae9d3 Mon Sep 17 00:00:00 2001 From: wized2 Date: Wed, 21 Oct 2026 03:04:31 +0000 Subject: [PATCH 2/3] fix: strict FIFO on argv budget; UTF-8 byte sizing; JOIN_SEP=7 Once the first pending message does not fit, stop draining so a smaller later message cannot leapfrog. Measure argv costs with UTF-8 byteLength. Correct separator length for "\\n\\n---\\n\\n" (7). Add regression test. --- src/core/engine.ts | 6 ++++- src/hosts/spawn-delivery.ts | 8 +++---- test/unit/core/spawn-delivery.test.ts | 32 +++++++++++++++++++++++++++ 3 files changed, 41 insertions(+), 5 deletions(-) diff --git a/src/core/engine.ts b/src/core/engine.ts index 515fb94..3acd037 100644 --- a/src/core/engine.ts +++ b/src/core/engine.ts @@ -1286,9 +1286,13 @@ export function drainQueue( continue } + // Strict FIFO under size/budget 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 baf8c93..a5b3c91 100644 --- a/src/hosts/spawn-delivery.ts +++ b/src/hosts/spawn-delivery.ts @@ -317,7 +317,7 @@ export async function deliverViaSpawn(deps: SpawnRunnerDeps, member: Member): Pr const builder = spawnBuilderFor(member.host)! const budget = spawnArgvBudget() const FRAMING_PER_MSG = 450 - const JOIN_SEP = 6 // "\n\n---\n\n".length + const JOIN_SEP = Buffer.byteLength("\n\n---\n\n", "utf8") // 7 // Phase 1 (locked): head-only oversized refusal, else partial drain. let batch: Array<{ id: string; channelName: string }> = [] @@ -327,12 +327,12 @@ export async function deliverViaSpawn(deps: SpawnRunnerDeps, member: Member): Pr 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.length + cmd.args.join(" ").length + 64 + const fixedOverhead = Buffer.byteLength(cmd.command, "utf8") + Buffer.byteLength(cmd.args.join(" "), "utf8") + 64 for (const id of state.queues[member.session_id] ?? []) { const msg = state.messages[id] if (!msg || msg.delivery_status !== "pending") continue - const headCost = fixedOverhead + msg.content.length + FRAMING_PER_MSG + const headCost = fixedOverhead + Buffer.byteLength(msg.content, "utf8") + FRAMING_PER_MSG if (headCost > budget) headOversizedId = id break } @@ -340,7 +340,7 @@ export async function deliverViaSpawn(deps: SpawnRunnerDeps, member: Member): Pr let used = fixedOverhead const canDeliver = (msg: { content: string }) => { - const cost = msg.content.length + FRAMING_PER_MSG + JOIN_SEP + const cost = Buffer.byteLength(msg.content, "utf8") + FRAMING_PER_MSG + JOIN_SEP if (used + cost > budget) return false used += cost return true diff --git a/test/unit/core/spawn-delivery.test.ts b/test/unit/core/spawn-delivery.test.ts index 2eef475..dd039d2 100644 --- a/test/unit/core/spawn-delivery.test.ts +++ b/test/unit/core/spawn-delivery.test.ts @@ -337,6 +337,38 @@ test("P2-2: head message alone oversized refused; queue untouched; no spawn", as } }) +test("P2-2b-strict: smaller later message does not leapfrog a blocked head", async () => { + // 80k head exceeds budget after fixed overhead on win32 (30k) and nearly fills + // linux 120k with framing; 50k is still too large to follow an 80k head; + // 5k must remain queued behind both — never delivered alone. + if (process.platform === "win32") return + + const state = emptyState() + const member = makeMember() + seedChannelWithSpawnMember(state, member) + const sizes = [80_000, 50_000, 5_000] + 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) + } + const { deps, spawned, errors } = makeFakeDeps(state) + const outcome = await deliverViaSpawn(deps, member) + // Head alone may or may not fit; either way the 5k message must not spawn alone. + const pending = Object.values(state.messages).filter((m) => m.delivery_status === "pending") + if (outcome.status === "delivered") { + // Only a prefix from the front is allowed. + assert.ok(spawned.length === 1) + assert.ok(pending.length >= 1) + // The 5k message (last) must still be pending if 50k was not delivered first. + const lastId = (state.queues[member.session_id] ?? []).at(-1) + assert.ok(lastId) + } else { + assert.equal(spawned.length, 0) + assert.equal(pending.length, 3) + } + assert.equal(errors.filter((e) => e.includes("spawn argv limit")).length, 0) +}) + test("P2-2b: FIFO partial drain when batch exceeds budget (issue #3)", async () => { // Force POSIX-sized budget path: three ~45k messages → first two fit under // 120k, third stays pending. On win32 (30k) the head alone is oversized; From eeb1d305e18a0ac0d219057916793b9f1eaf7820 Mon Sep 17 00:00:00 2001 From: wized2 Date: Fri, 30 Oct 2026 12:32:29 +0000 Subject: [PATCH 3/3] =?UTF-8?q?fix:=20address=20PR=20review=20=E2=80=94=20?= =?UTF-8?q?TS=20imports,=20eligible=20head=20size,=20regression=20fixtures?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Import MessageEnvelope from core/types; export budgetRefusal for spawn path - Oversized-head check skips paused/budget-exhausted channels (eligibility first) - P2-2 uses 100k multi-byte UTF-8 so framed cost exceeds POSIX argv budget - P2-2b-strict restores 80k/50k/5k with payload + queue order assertions - P2-2 boundary builds a deterministic under-budget payload and requires delivery - Drop unused SPAWN_ARGV_BUDGET test import --- src/core/engine.ts | 2 +- src/hosts/spawn-delivery.ts | 19 +++++--- test/unit/core/spawn-delivery.test.ts | 63 +++++++++++++++++---------- 3 files changed, 54 insertions(+), 30 deletions(-) diff --git a/src/core/engine.ts b/src/core/engine.ts index 23257ab..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) { diff --git a/src/hosts/spawn-delivery.ts b/src/hosts/spawn-delivery.ts index fb4f11e..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,13 +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, - type MessageEnvelope, 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). */ @@ -339,12 +340,20 @@ async function deliverViaSpawnInner(deps: SpawnRunnerDeps, member: Member): Prom 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) - // Head-only oversized refusal (consistent accounting: no JOIN on first msg). + // 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 - const channelName = - Object.values(state.channels).find((channel) => channel.id === msg.channel_id)?.name ?? msg.channel_id + 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 diff --git a/test/unit/core/spawn-delivery.test.ts b/test/unit/core/spawn-delivery.test.ts index 44a797f..39cdfaa 100644 --- a/test/unit/core/spawn-delivery.test.ts +++ b/test/unit/core/spawn-delivery.test.ts @@ -11,13 +11,13 @@ import { parseCommandTemplate, resolveBinaryOverride, spawnArgvBudget, - SPAWN_ARGV_BUDGET, + 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" @@ -347,8 +347,9 @@ test("P2-2: head message alone oversized refused; queue untouched; no spawn", as const state = emptyState() const member = makeMember() seedChannelWithSpawnMember(state, member) - // Max content is 100_000 chars; still oversized once framed on both platforms. - const huge = "x".repeat(99_000) + // 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]! @@ -372,19 +373,31 @@ test("P2-2 boundary: first-message JOIN_SEP must not be charged (UTF-8)", async const state = emptyState() const member = makeMember() seedChannelWithSpawnMember(state, member) - const content = "é".repeat(59_723) + "x" - const sent = sendMessage(state, { channel: "spawn-ch", content }, "sess_sender") - assert.equal(sent.ok, true) + 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.notEqual(outcome.detail, "no pending messages") - if (outcome.status === "delivered") { - assert.equal(spawned.length, 1) - } else { - assert.equal(outcome.status, "skipped") - assert.match(outcome.detail, /argv limit/) - assert.equal((state.queues[member.session_id] ?? []).length, 1) - } + 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 () => { @@ -392,8 +405,9 @@ test("P2-2b-strict: only the first fitting message drains; later ids stay queued const state = emptyState() const member = makeMember() seedChannelWithSpawnMember(state, member) - // ~55k / ~55k / ~5k — only first fits under ~120k framed budget. - const sizes = [55_000, 55_000, 5_000] + // 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( @@ -406,18 +420,19 @@ test("P2-2b-strict: only the first fitting message drains; later ids stay queued } const { deps, spawned } = makeFakeDeps(state) const outcome = await deliverViaSpawn(deps, member) - assert.equal(outcome.status, "delivered") - assert.equal(spawned.length, 1) - const payload = spawned[0]!.message - assert.match(payload, /m0:/) - assert.doesNotMatch(payload, /m1:/) - assert.doesNotMatch(payload, /m2:/) - assert.deepEqual(state.queues[member.session_id], [ids[1], ids[2]]) + 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()