From 44efdd9259dc30886549b100b1dcff527f3a0800 Mon Sep 17 00:00:00 2001 From: Harry Yep Date: Sun, 26 Jul 2026 18:28:36 +0800 Subject: [PATCH 1/2] fix: stop gate settles reaped cancellations instead of looping on TaskStop demands --- plugins/fusion/scripts/worker-lifecycle.mjs | 72 ++++-- tests/fusion-worker-lifecycle.test.mjs | 233 ++++++++++++++++++++ 2 files changed, 292 insertions(+), 13 deletions(-) diff --git a/plugins/fusion/scripts/worker-lifecycle.mjs b/plugins/fusion/scripts/worker-lifecycle.mjs index abfd76a..6ec6e6e 100644 --- a/plugins/fusion/scripts/worker-lifecycle.mjs +++ b/plugins/fusion/scripts/worker-lifecycle.mjs @@ -658,6 +658,22 @@ function canonicalToolResponseUsage(response) { } function handlePreToolUse(input, env) { + if (!input.agent_id && ["TaskStop", "TaskOutput"].includes(input.tool_name) && typeof input.tool_input?.task_id === "string") { + const taskId = input.tool_input.task_id; + const now = new Date().toISOString(); + for (const record of readWorkerRecords(env, { strict: true }).filter((candidate) => candidate.sessionId === input.session_id && !isTerminalWorkerStatus(candidate.transportStatus) && [candidate.backgroundTaskId, candidate.agentId].includes(taskId))) { + updateLifecycleWorkerRecord(record.taskId, env, (current) => { + if (!current || current.sessionId !== input.session_id || isTerminalWorkerStatus(current.transportStatus) || ![current.backgroundTaskId, current.agentId].includes(taskId)) { + return null; + } + return { + ...current, + cancelAttemptCount: (current.cancelAttemptCount ?? 0) + 1, + lastCancelAttemptAt: now + }; + }); + } + } if (input.tool_name === "Agent" || input.tool_name === "Task") { const agentType = input.tool_input?.subagent_type; if (PEER_WRAPPER_AGENTS.has(agentType)) { @@ -959,6 +975,16 @@ function settleQueuedVerdict(record, now, env) { } } +function settleReapedWorker(current, now, env) { + return settleQueuedVerdict({ + ...markWorkerCollected(current, WORKER_COLLECTION_METHODS.TASK_REAPED, now), + transportStatus: "failed", + failureKind: "task_reaped", + finishedAt: now, + lastActivityAt: now + }, now, env); +} + function handlePostToolUse(input, env, failed = false) { if (!input.agent_id && ["Read", "TaskOutput", "TaskStop"].includes(input.tool_name)) { const taskId = typeof input.tool_input?.task_id === "string" ? input.tool_input.task_id : null; @@ -971,13 +997,7 @@ function handlePostToolUse(input, env, failed = false) { if (!current || current.sessionId !== input.session_id || isTerminalWorkerStatus(current.transportStatus) || ![current.backgroundTaskId, current.agentId].includes(taskId)) { return null; } - return settleQueuedVerdict({ - ...markWorkerCollected(current, WORKER_COLLECTION_METHODS.TASK_REAPED, now), - transportStatus: "failed", - failureKind: "task_reaped", - finishedAt: now, - lastActivityAt: now - }, now, env); + return settleReapedWorker(current, now, env); }); } return; @@ -1849,7 +1869,8 @@ function clearSettleOnlyAdvisory(sessionId, env) { } function handleStop(input, env) { - const tasks = Array.isArray(input.background_tasks) ? input.background_tasks : []; + const hasBackgroundTasks = Array.isArray(input.background_tasks); + const tasks = hasBackgroundTasks ? input.background_tasks : []; reconcileTaskNotifications(input, env); const initialRecords = sessionRecords(input.session_id, env); if (initialRecords.every((record) => !terminalCollectionPending(record)) && collectorStopGate(input, env)) { @@ -1893,13 +1914,38 @@ function handleStop(input, env) { return settleQueuedVerdict({ ...current, transportStatus: "failed", failureKind: current.failureKind ?? "owner_lost", finishedAt: now }, now, env); }); } - const cancelIds = cancellations.map((record) => record.backgroundTaskId ?? record.agentId).filter(Boolean); - if (cancelIds.length === 0) { - writeOutput(blockStop(`Fusion task${missingRuntimeIds.length === 1 ? "" : "s"} ${missingRuntimeIds.map((record) => record.taskId).join(", ")} failed before a runtime task id was available. Report the failure before finishing.`)); + const reapedTaskIds = []; + for (const record of cancellations.filter((candidate) => candidate.backgroundTaskId || candidate.agentId)) { + if (hasBackgroundTasks ? runtimeTaskForRecord(record, tasks) : (record.cancelAttemptCount ?? 0) < 2) { + continue; + } + let settled = false; + updateLifecycleWorkerRecord(record.taskId, env, (current) => { + if (!current || isTerminalWorkerStatus(current.transportStatus)) { + return null; + } + settled = true; + return settleReapedWorker(current, now, env); + }); + if (settled) { + reapedTaskIds.push(record.taskId); + } + } + const reapedSentence = reapedTaskIds.length > 0 ? `Fusion task IDs ${reapedTaskIds.join(", ")} were settled as task_reaped because the harness no longer tracks their runtime tasks.` : ""; + const reapedSuffix = reapedSentence ? ` ${reapedSentence}` : ""; + const cancelIds = cancellations.filter((record) => !reapedTaskIds.includes(record.taskId)).map((record) => record.backgroundTaskId ?? record.agentId).filter(Boolean); + if (cancelIds.length > 0) { + writeOutput(blockStop(`Call TaskStop for over-budget task${cancelIds.length === 1 ? "" : "s"} ${cancelIds.join(", ")} and report the cancellation before finishing.${reapedSuffix}`)); + return; + } + if (missingRuntimeIds.length > 0) { + writeOutput(blockStop(`Fusion task${missingRuntimeIds.length === 1 ? "" : "s"} ${missingRuntimeIds.map((record) => record.taskId).join(", ")} failed before a runtime task id was available. Report the failure before finishing.${reapedSuffix}`)); + return; + } + if (reapedSentence) { + writeOutput(hookOutput("Stop", reapedSentence)); return; } - writeOutput(blockStop(`Call TaskStop for over-budget task${cancelIds.length === 1 ? "" : "s"} ${cancelIds.join(", ")} and report the cancellation before finishing.`)); - return; } const currentRecords = sessionRecords(input.session_id, env); const inFlight = armInFlightRecords(currentRecords, tasks, env); diff --git a/tests/fusion-worker-lifecycle.test.mjs b/tests/fusion-worker-lifecycle.test.mjs index ec64c72..20b5bf6 100644 --- a/tests/fusion-worker-lifecycle.test.mjs +++ b/tests/fusion-worker-lifecycle.test.mjs @@ -1000,6 +1000,239 @@ test("a TaskStop no-task error marks a matching unexpected async peer wrapper as assert.ok(reaped.collectedAt); }); +test("Stop emits one reaped notice when the harness no longer tracks a cancel-requested worker", (t) => { + const box = sandbox(t); + const workerDispatch = dispatch(box); + run(box, workerDispatch); + run(box, { + ...workerDispatch, + hook_event_name: "PostToolUse", + tool_response: { isAsync: true, status: "async_launched", agentId: "reaped-cancellation" } + }); + const taskId = record(box).taskId; + updateWorkerRecord(taskId, envFor(box), (current) => ({ ...current, transportStatus: "cancel_requested", failureKind: "timeout" })); + + const stopped = run(box, { + hook_event_name: "Stop", + session_id: "session-1", + cwd: box.cwd, + transcript_path: box.transcript, + stop_hook_active: true, + background_tasks: [] + }); + + const output = JSON.parse(stopped.stdout); + assert.strictEqual(output.decision, undefined); + assert.match(output.hookSpecificOutput.additionalContext, new RegExp(`Fusion task IDs ${taskId} were settled as task_reaped because the harness no longer tracks their runtime tasks`)); + assert.doesNotMatch(stopped.stdout, /Call TaskStop/); + const settled = record(box); + assert.strictEqual(settled.transportStatus, "failed"); + assert.strictEqual(settled.failureKind, "task_reaped"); + assert.strictEqual(settled.collectionMethod, "task_reaped"); + assert.ok(settled.finishedAt); + assert.ok(settled.collectedAt); +}); + +test("Stop emits one TaskStop demand while the harness tracks a cancel-requested worker", (t) => { + const box = sandbox(t); + const workerDispatch = dispatch(box); + run(box, workerDispatch); + run(box, { + ...workerDispatch, + hook_event_name: "PostToolUse", + tool_response: { isAsync: true, status: "async_launched", agentId: "live-cancellation" } + }); + updateWorkerRecord(record(box).taskId, envFor(box), (current) => ({ ...current, transportStatus: "cancel_requested", failureKind: "timeout" })); + + const stopped = run(box, { + hook_event_name: "Stop", + session_id: "session-1", + cwd: box.cwd, + transcript_path: box.transcript, + stop_hook_active: false, + background_tasks: [{ id: "live-cancellation", type: "subagent", status: "running", agent_type: "fusion:fast-worker" }] + }); + + assert.deepStrictEqual(JSON.parse(stopped.stdout), { + decision: "block", + reason: "Call TaskStop for over-budget task live-cancellation and report the cancellation before finishing." + }); + assert.strictEqual(record(box).transportStatus, "cancel_requested"); +}); + +test("Stop emits one combined TaskStop demand and reaped notice for mixed cancellations", (t) => { + const box = sandbox(t); + const absent = createWorkerRecord({ + taskId: `fusion-${"c".repeat(24)}`, + sessionId: "session-1", + agentType: "fusion:fast-worker", + workspaceRoot: box.cwd + }, envFor(box)); + const live = createWorkerRecord({ + taskId: `fusion-${"d".repeat(24)}`, + sessionId: "session-1", + agentType: "fusion:fast-worker", + workspaceRoot: box.cwd + }, envFor(box)); + updateWorkerRecord(absent.taskId, envFor(box), (current) => ({ ...current, agentId: "absent-cancellation", backgroundTaskId: "absent-cancellation", transportStatus: "cancel_requested", failureKind: "timeout" })); + updateWorkerRecord(live.taskId, envFor(box), (current) => ({ ...current, agentId: "live-cancellation", backgroundTaskId: "live-cancellation", transportStatus: "cancel_requested", failureKind: "timeout" })); + + const stopped = run(box, { + hook_event_name: "Stop", + session_id: "session-1", + cwd: box.cwd, + transcript_path: box.transcript, + stop_hook_active: true, + background_tasks: [{ id: "live-cancellation", type: "subagent", status: "running", agent_type: "fusion:fast-worker" }] + }); + + const output = JSON.parse(stopped.stdout); + assert.deepStrictEqual(output, { + decision: "block", + reason: `Call TaskStop for over-budget task live-cancellation and report the cancellation before finishing. Fusion task IDs ${absent.taskId} were settled as task_reaped because the harness no longer tracks their runtime tasks.` + }); + const records = readWorkerRecords(envFor(box)); + const settled = records.find((candidate) => candidate.taskId === absent.taskId); + const pending = records.find((candidate) => candidate.taskId === live.taskId); + assert.strictEqual(settled.transportStatus, "failed"); + assert.strictEqual(settled.failureKind, "task_reaped"); + assert.strictEqual(pending.transportStatus, "cancel_requested"); +}); + +test("Stop emits one reaped notice while applying a queued rejected verdict", (t) => { + const box = sandbox(t); + const workerDispatch = dispatch(box); + run(box, workerDispatch); + run(box, { + ...workerDispatch, + hook_event_name: "PostToolUse", + tool_response: { isAsync: true, status: "async_launched", agentId: "reaped-rejected" } + }); + const worker = updateWorkerRecord(record(box).taskId, envFor(box), (current) => ({ ...current, transportStatus: "cancel_requested", failureKind: "timeout" })); + const queued = recordWorkerAcceptance({ taskId: worker.taskId, acceptance: "rejected", env: envFor(box), source: "main-loop", reason: "Runtime result was not delivered." }); + assert.strictEqual(queued.queued, true); + + const stopped = run(box, { + hook_event_name: "Stop", + session_id: "session-1", + cwd: box.cwd, + transcript_path: box.transcript, + stop_hook_active: true, + background_tasks: [] + }); + + const output = JSON.parse(stopped.stdout); + assert.strictEqual(output.decision, undefined); + assert.match(output.hookSpecificOutput.additionalContext, new RegExp(`Fusion task IDs ${worker.taskId} were settled as task_reaped because the harness no longer tracks their runtime tasks`)); + const settled = record(box); + assert.strictEqual(settled.transportStatus, "failed"); + assert.strictEqual(settled.failureKind, "task_reaped"); + assert.strictEqual(settled.acceptance, "rejected"); + assert.strictEqual(settled.acceptanceReason, "Runtime result was not delivered."); + assert.ok(settled.acceptanceRecordedAt); + assert.strictEqual(settled.pendingVerdict, undefined); +}); + +test("Stop without a runtime task list emits one response and settles after two unobserved cancellation attempts", (t) => { + const box = sandbox(t); + const workerDispatch = dispatch(box); + run(box, workerDispatch); + run(box, { + ...workerDispatch, + hook_event_name: "PostToolUse", + tool_response: { isAsync: true, status: "async_launched", agentId: "unobserved-cancellation" } + }); + updateWorkerRecord(record(box).taskId, envFor(box), (current) => ({ ...current, transportStatus: "cancel_requested", failureKind: "timeout" })); + const attempt = { + hook_event_name: "PreToolUse", + session_id: "session-1", + cwd: box.cwd, + transcript_path: box.transcript, + tool_name: "TaskStop", + tool_input: { task_id: "unobserved-cancellation" } + }; + + run(box, attempt); + const firstStopped = run(box, { + hook_event_name: "Stop", + session_id: "session-1", + cwd: box.cwd, + transcript_path: box.transcript, + stop_hook_active: true + }); + const firstOutput = JSON.parse(firstStopped.stdout); + assert.strictEqual(firstOutput.decision, "block"); + assert.match(firstOutput.reason, /Call TaskStop for over-budget task unobserved-cancellation/); + assert.strictEqual(record(box).cancelAttemptCount, 1); + assert.strictEqual(record(box).transportStatus, "cancel_requested"); + + run(box, attempt); + const secondStopped = run(box, { + hook_event_name: "Stop", + session_id: "session-1", + cwd: box.cwd, + transcript_path: box.transcript, + stop_hook_active: true + }); + const secondOutput = JSON.parse(secondStopped.stdout); + assert.strictEqual(secondOutput.decision, undefined); + assert.match(secondOutput.hookSpecificOutput.additionalContext, /settled as task_reaped because the harness no longer tracks their runtime tasks/); + assert.strictEqual(record(box).cancelAttemptCount, 2); + assert.strictEqual(record(box).transportStatus, "failed"); + assert.strictEqual(record(box).failureKind, "task_reaped"); +}); + +test("PreToolUse task tools ignore subagents and stamp only matching active main-session workers", (t) => { + const box = sandbox(t); + const active = createWorkerRecord({ + taskId: `fusion-${"a".repeat(24)}`, + sessionId: "session-1", + agentType: "fusion:fast-worker", + workspaceRoot: box.cwd + }, envFor(box)); + const terminal = createWorkerRecord({ + taskId: `fusion-${"b".repeat(24)}`, + sessionId: "session-1", + agentType: "fusion:fast-worker", + workspaceRoot: box.cwd + }, envFor(box)); + updateWorkerRecord(active.taskId, envFor(box), (current) => ({ ...current, agentId: "shared-cancellation", backgroundTaskId: "shared-cancellation", transportStatus: "cancel_requested" })); + const terminalBefore = updateWorkerRecord(terminal.taskId, envFor(box), (current) => ({ ...current, agentId: "shared-cancellation", backgroundTaskId: "shared-cancellation", transportStatus: "done", finishedAt: new Date().toISOString() })); + + for (const toolName of ["TaskStop", "TaskOutput"]) { + const subagentAttempt = run(box, { + hook_event_name: "PreToolUse", + session_id: "session-1", + cwd: box.cwd, + transcript_path: box.transcript, + agent_id: "calling-worker", + agent_type: "fusion:fast-worker", + tool_name: toolName, + tool_input: { task_id: "shared-cancellation" } + }); + + assert.strictEqual(subagentAttempt.stdout, ""); + assert.strictEqual(readWorkerRecords(envFor(box)).find((candidate) => candidate.taskId === active.taskId).cancelAttemptCount, undefined); + } + + const mainAttempt = run(box, { + hook_event_name: "PreToolUse", + session_id: "session-1", + cwd: box.cwd, + transcript_path: box.transcript, + tool_name: "TaskStop", + tool_input: { task_id: "shared-cancellation" } + }); + + assert.strictEqual(mainAttempt.stdout, ""); + const records = readWorkerRecords(envFor(box)); + const stamped = records.find((candidate) => candidate.taskId === active.taskId); + const untouched = records.find((candidate) => candidate.taskId === terminal.taskId); + assert.strictEqual(stamped.cancelAttemptCount, 1); + assert.ok(stamped.lastCancelAttemptAt); + assert.deepStrictEqual(untouched, terminalBefore); +}); + test("no-task errors ignore unmatched task ids and terminal worker records", (t) => { const box = sandbox(t); const peerDispatch = dispatch(box, { subagent_type: "codex:codex-rescue", prompt: "bounded peer brief" }); From 1886d66dfc89c20f57d94422267b21e87af85b0c Mon Sep 17 00:00:00 2001 From: Harry Yep Date: Sun, 26 Jul 2026 18:28:37 +0800 Subject: [PATCH 2/2] chore: release 0.0.39 --- .claude-plugin/marketplace.json | 8 ++++---- CHANGELOG.md | 6 ++++++ plugins/codex/.claude-plugin/plugin.json | 2 +- plugins/fusion/.claude-plugin/plugin.json | 2 +- plugins/grok/.claude-plugin/plugin.json | 2 +- 5 files changed, 13 insertions(+), 7 deletions(-) diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json index 30c617c..203d7b5 100644 --- a/.claude-plugin/marketplace.json +++ b/.claude-plugin/marketplace.json @@ -6,14 +6,14 @@ "email": "hi@okis.dev" }, "description": "Multi-model orchestration marketplace for Claude Code.", - "version": "0.0.38", + "version": "0.0.39", "plugins": [ { "name": "grok", "source": "./plugins/grok", "displayName": "Grok Companion", "description": "Local Grok CLI delegation: task, review, resumable history, best-of-n tournaments, background jobs, stats, and setup health checks.", - "version": "0.0.38", + "version": "0.0.39", "author": { "name": "Harry Yep" }, @@ -34,7 +34,7 @@ "source": "./plugins/codex", "displayName": "Codex Companion", "description": "First party local Codex CLI delegation for tasks, reviews, resumable threads, and durable background jobs.", - "version": "0.0.38", + "version": "0.0.39", "author": { "name": "Harry Yep" }, @@ -55,7 +55,7 @@ "source": "./plugins/fusion", "displayName": "Fusion Orchestrator", "description": "Multi-model orchestration: tier agents, routing rules, blind panel, ultra fleet, model config, and drift doctor.", - "version": "0.0.38", + "version": "0.0.39", "author": { "name": "Harry Yep" }, diff --git a/CHANGELOG.md b/CHANGELOG.md index ccdba0f..94312cd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # changelog +## 0.0.39 + +- the over budget stop gate settles instead of looping when the harness has already reaped a worker's runtime task: a `cancel_requested` record whose runtime id is absent from a provided `background_tasks` array settles as `task_reaped` through the same `settleReapedWorker` helper the no task found escape hatch uses, because that hatch never fires for this class; `TaskStop` on a reaped id returns a tool_use_error, which emits neither PostToolUse nor PostToolUseFailure, so the demand was unsatisfiable (observed live: two user interrupted workers held eight consecutive blocking stop rounds while twelve errored `TaskStop` and `TaskOutput` calls delivered zero hook events, and only SessionEnd closed the records) +- reaped settlements keep the stop hook's single output document invariant: the notice rides the block reason while live cancellations or missing runtime failures still demand, and emits alone only when nothing blocks, so a blocking decision can never be lost to a second stdout document +- PreToolUse stamps `cancelAttemptCount` on matching non terminal main session workers when `TaskStop` or `TaskOutput` targets their runtime id, and a stop input without any `background_tasks` array settles a cancellation after two unobserved attempts, covering harness inputs that omit the runtime task list + ## 0.0.38 - Fable and Opus 5 are peer orchestrators: the routing preamble drops the Opus as fallback framing (switching is lateral; posture, gates, fleet defaults, and delegation never condition on which of the two is active), the readme reframes `best[1m]` floating as lateral within the top tier, and doctor treats a deliberate Opus 5 session as healthy instead of a fallback to recover diff --git a/plugins/codex/.claude-plugin/plugin.json b/plugins/codex/.claude-plugin/plugin.json index ef97fe6..d30f872 100644 --- a/plugins/codex/.claude-plugin/plugin.json +++ b/plugins/codex/.claude-plugin/plugin.json @@ -2,7 +2,7 @@ "$schema": "https://json.schemastore.org/claude-code-plugin-manifest.json", "name": "codex", "displayName": "Codex Companion", - "version": "0.0.38", + "version": "0.0.39", "description": "First party local Codex CLI delegation for tasks, reviews, resumable threads, and durable background jobs.", "author": { "name": "Harry Yep" diff --git a/plugins/fusion/.claude-plugin/plugin.json b/plugins/fusion/.claude-plugin/plugin.json index 2d67d46..fc72815 100644 --- a/plugins/fusion/.claude-plugin/plugin.json +++ b/plugins/fusion/.claude-plugin/plugin.json @@ -2,7 +2,7 @@ "$schema": "https://json.schemastore.org/claude-code-plugin-manifest.json", "name": "fusion", "displayName": "Fusion Orchestrator", - "version": "0.0.38", + "version": "0.0.39", "description": "Multi-model orchestration: tier agents, routing rules, blind panel, ultra fleet, model config, and drift doctor.", "author": { "name": "Harry Yep" diff --git a/plugins/grok/.claude-plugin/plugin.json b/plugins/grok/.claude-plugin/plugin.json index 32e4f18..0d5b294 100644 --- a/plugins/grok/.claude-plugin/plugin.json +++ b/plugins/grok/.claude-plugin/plugin.json @@ -2,7 +2,7 @@ "$schema": "https://json.schemastore.org/claude-code-plugin-manifest.json", "name": "grok", "displayName": "Grok Companion", - "version": "0.0.38", + "version": "0.0.39", "description": "Local Grok CLI delegation: task, review, resumable history, background jobs, stats, and setup health checks.", "author": { "name": "Harry Yep"