diff --git a/doc/plans/2026-10-06-hermes-native-runner.md b/doc/plans/2026-10-06-hermes-native-runner.md index 02e50f5d365..88638779923 100644 --- a/doc/plans/2026-10-06-hermes-native-runner.md +++ b/doc/plans/2026-10-06-hermes-native-runner.md @@ -997,3 +997,48 @@ forbidden/tracked-import guards still report existing violations in unchanged files; their failed logs are retained. These are not passing guard results. The total reserved amount remains `$23` against the `$25` cap; unknown spend retains its reservation. Hermes remains pending qualification. + +### 2026-10-08 committed-question native stop boundary + +The clean `8c6c82a213` Mac question campaign failed after 644.667 seconds. +Its original infrastructure classification and `not_started` cleanup verdict +remain unchanged. Independent public records show that the yielded first run +had pending unpriced accounting. The company paused, so submitting Cobalt +could not launch the continuation. Independent cleanup checks found no owned +process group, server listener or temporary instance directory. These checks +do not replace the original cleanup verdict or establish billing completeness. + +A credential-free native reproduction showed two races. Hermes could start +another model request after saving the question. Even after stopping that +request, immediate Runner shutdown could overtake the terminal prompt receipt. +The failed diagnostic logs remain retained. + +The bridge now recognizes the authenticated assigned question result and uses +Hermes's native hard interrupt before returning from the completion callback. +It publishes that completion after native finalization and usage provenance. +Both Runner event pumps retain the completed question event until ACPX's final +prompt receipt has been read. Ordinary tools still stream. The deferred result +must be applied, pending, wake the assignee and match the current run. The +controller still validates durable wait and cost authority independently. + +Both predecessor manifests were authenticated before the bridge-only update. +The task-owned Mac runtime passed complete file verification before and after +the transformation. Its new closure is +`f443a6867c914f49d7308c7421dbce9ba4a0d024b0c718a41a78845c784a7bd0`. +The derived Linux closure is +`c9879c3b69357d17d375d8fd4897f18496cebb21256c1e8bb9220d779d995921`. +Dependencies and Rust inputs remain unchanged. Fresh cloud materialization, +native regression, clean-head live billing and review proof remain required. +No new paid retry has started. Docker stays stopped, and the full unknown-spend +reservation remains held. + +The correction passes 38 pinned Python tests, 108 focused Runner tests and +seven distribution fixtures. Runner TypeScript and generated contracts pass. +All four native fixtures pass in 106.153 seconds. The new Rust-sidecar test +immediately cancels and closes after the question result, observes exactly one +model request and retains the reported 10 input and five output tokens with a +complete turn delta. A provider that has already ended may reject interruption +as `already_terminal`; the test still requires verified owned close and the +receipt. This is deterministic transport proof, not live billing or Daytona +proof. The first focused sidecar run timed out under restricted local IPC; +its failure log is retained separately from the passing authorized IPC retry. diff --git a/packages/paperclip-runner/docs/hermes.md b/packages/paperclip-runner/docs/hermes.md index 261a9467e44..856cffa540a 100644 --- a/packages/paperclip-runner/docs/hermes.md +++ b/packages/paperclip-runner/docs/hermes.md @@ -135,6 +135,13 @@ extend the task's execution deadline. authority. Missing, invalid or timed-out reads do not invent usage or cost. This optional execution-result field carries an existing v1 PRP event; it does not change persisted execution inputs or the Rust wire contract. +- An applied pending `request_human_input` response stops Hermes in its native + tool completion callback, before another model request. The bridge publishes + that tool's completed result after native finalization. Both Runner event + pumps read the prompt usage receipt before forwarding this completion to the + controller. Other tool activity still streams immediately. Tool identities + and results are preserved; wait and accounting authority remain with the + controller. Missing usage remains unknown. - Paperclip owns task titles. Hermes keeps its immediate derived session title; its paid background title upgrade is disabled in the managed profile so it cannot start inference after a turn's usage receipt has closed. diff --git a/packages/paperclip-runner/scripts/qualify-hermes-runnerd.test.mjs b/packages/paperclip-runner/scripts/qualify-hermes-runnerd.test.mjs index 32cbec9110b..748e844f371 100644 --- a/packages/paperclip-runner/scripts/qualify-hermes-runnerd.test.mjs +++ b/packages/paperclip-runner/scripts/qualify-hermes-runnerd.test.mjs @@ -23,7 +23,9 @@ const completion = { evidence: [], verification: [{ commandOrCheck: 'native transport fixture', status: 'passed' }], attentionRequests: [], artifacts: [], }; -test('pinned Hermes streams images and semantic completion through Rust PRP and the production sidecar', { +for (const question of [false, true]) test(question + ? 'pinned Hermes retains prompt usage before a committed question triggers immediate Rust-sidecar shutdown' + : 'pinned Hermes streams images and semantic completion through Rust PRP and the production sidecar', { skip: process.env.PAPERCLIP_HERMES_QUALIFY !== '1', timeout: 180_000, }, async t => { const root = await mkdtemp(join(tmpdir(), 'paperclip-hermes-runnerd-')); @@ -50,9 +52,10 @@ test('pinned Hermes streams images and semantic completion through Rust PRP and if (!body.messages.some(message => message.role === 'tool')) { send({ role: 'assistant', reasoning_content: 'Checking the native runner path.' }); await new Promise(resolve => setTimeout(resolve, 80)); - const finish = body.tools.find(tool => tool.function.name.endsWith('paperclip_finish')); + const finish = body.tools.find(tool => tool.function.name.endsWith(question ? 'request_human_input' : 'paperclip_finish')); assert.ok(finish, 'Paperclip completion authority was not exposed to the native model'); - send({ tool_calls: [{ index: 0, id: 'native-prp-finish', type: 'function', function: { name: finish.function.name, arguments: JSON.stringify(completion) } }] }); + send({ tool_calls: [{ index: 0, id: 'native-prp-finish', type: 'function', function: { name: finish.function.name, + arguments: JSON.stringify(question ? { marker: 'governed-question' } : completion) } }] }); send({}, 'tool_calls'); } else { send({ role: 'assistant', content: 'Native Rust ' }); @@ -105,14 +108,43 @@ test('pinned Hermes streams images and semantic completion through Rust PRP and }, prpIdentity: { runnerInstanceId: 'hermes-prp-runner', environmentLeaseId: 'hermes-prp-lease', runId: identity.runId, normalizedSessionId: identity.sessionId, turnId: 'hermes-prp-turn', itemId: 'hermes-prp-item' }, }); - const backend = createRunnerdNativeSessionBackend(input, { runnerInstanceId: 'hermes-prp-runner', transportFactory: () => bundle.transport }); + const backend = createRunnerdNativeSessionBackend(input, { runnerInstanceId: 'hermes-prp-runner', transportFactory: () => bundle.transport, + ...(question ? { + dynamicTools: [{ name: 'request_human_input', description: 'Simulated authenticated question operation', + inputSchema: { type: 'object', properties: { marker: { type: 'string' } }, required: ['marker'] } }], + dynamicToolHandler: async () => ({ disposition: 'applied', interaction: { + id: 'fixture-question', companyId: identity.companyId, issueId: identity.issueId, sourceRunId: identity.runId, + kind: 'ask_user_questions', status: 'pending', continuationPolicy: 'wake_assignee', + } }), + } : {}), + }); session = await backend.openSession({ identity, workingDirectory: workspace }); await session.startTurn({ message: { role: 'user', text: input.task.prompt, attachments: input.attachments } }); const events = []; + let cancellationOutcome; for await (const event of session.events()) { events.push(event); + if (question && event.eventType === 'tool.execution.completed') { + cancellationOutcome = session.cancel({ reason: 'committed question wait', signal: new AbortController().signal }).cleanup + .then(() => ({ stopped: true }), error => ({ error })); + break; + } if (['run.terminal', 'turn.completed', 'turn.failed', 'turn.cancelled', 'turn.interrupted'].includes(event.eventType)) break; } + if (question) { + assert.ok(cancellationOutcome, 'The committed question completion did not reach the native boundary'); + await session.close({ reason: 'immediate question shutdown' }); + const stop = await cancellationOutcome; + if (stop.error) assert.equal(stop.error.code, 'already_terminal', 'Provider cancellation failed before terminal settlement'); + const usage = await session.accountingUsageEvent?.(); + assert.deepEqual(usage?.payload.usage.runDelta && { + input: usage.payload.usage.runDelta.inputTokens, output: usage.payload.usage.runDelta.outputTokens, + complete: usage.payload.usage.runDeltaComplete, + }, { input: 10, output: 5, complete: true }, 'Immediate shutdown lost the native prompt receipt'); + assert.ok(events.some(event => event.payload.kind === 'usage'), 'The question completion reached PRP before its usage receipt'); + assert.equal(requests.length, 1, 'Hermes started another model request after the committed question'); + return; + } assert.ok(events.some(event => event.eventType === 'turn.completed'), JSON.stringify(events)); assert.ok(events.some(event => JSON.stringify(event.payload).includes('Checking the native runner path.')), 'Reasoning did not cross PRP'); assert.ok(events.some(event => JSON.stringify(event.payload).includes('Native Rust')), 'Text did not cross PRP'); diff --git a/packages/paperclip-runner/scripts/qualify-hermes-runtime.test.mjs b/packages/paperclip-runner/scripts/qualify-hermes-runtime.test.mjs index 8a307f958df..51a064393ce 100644 --- a/packages/paperclip-runner/scripts/qualify-hermes-runtime.test.mjs +++ b/packages/paperclip-runner/scripts/qualify-hermes-runtime.test.mjs @@ -72,16 +72,18 @@ test("pinned Hermes streams through the production ACPX host using a no-auth loc return; } const toolDone = body.messages.slice(lastUser + 1).some(message => message.role === "tool"); - if (!toolDone && (userText.includes("NATIVE_TOOL_FIXTURE") || userText.includes("NATIVE_QUESTION_FIXTURE") || userText.includes("NATIVE_MCP_FIXTURE") || userText.includes("NATIVE_MEMORY_FIXTURE") || userText.includes("NATIVE_PLAN_FIXTURE"))) { + if (!toolDone && (userText.includes("NATIVE_TOOL_FIXTURE") || userText.includes("NATIVE_QUESTION_FIXTURE") || userText.includes("NATIVE_MCP_FIXTURE") || userText.includes("NATIVE_MEMORY_FIXTURE") || userText.includes("NATIVE_PLAN_FIXTURE") || userText.includes("NATIVE_GOVERNED_QUESTION_FIXTURE"))) { const question = userText.includes("NATIVE_QUESTION_FIXTURE"); + const governed = userText.includes("NATIVE_GOVERNED_QUESTION_FIXTURE"); const bridge = userText.includes("NATIVE_MCP_FIXTURE"); const memory = userText.includes("NATIVE_MEMORY_FIXTURE"); const plan = userText.includes("NATIVE_PLAN_FIXTURE"); - const name = question ? "clarify" : memory ? "memory" : bridge ? body.tools.find(tool => tool.function?.name.endsWith("fixture_probe"))?.function.name ?? "missing_assigned_tool" + const name = governed ? body.tools.find(tool => tool.function?.name.endsWith("request_human_input"))?.function.name ?? "missing_question_tool" + : question ? "clarify" : memory ? "memory" : bridge ? body.tools.find(tool => tool.function?.name.endsWith("fixture_probe"))?.function.name ?? "missing_assigned_tool" : plan ? body.tools.find(tool => tool.function?.name.endsWith("write_document"))?.function.name ?? "missing_plan_tool" : "terminal"; const args = question ? { question: "Choose the fixture color", choices: ["Cobalt", "Amber"] } : memory ? { action: "add", target: "memory", content: "NATIVE_PERSISTED_MEMORY: Project uses Cobalt." } - : bridge ? { marker: "assigned-tool-roundtrip" } + : governed || bridge ? { marker: "assigned-tool-roundtrip" } : plan ? { marker: "planning-workflow-roundtrip" } : { command: "printf 'native-hermes-command\\n' > command-proof.txt", timeout: 10 }; res.write(`data: ${JSON.stringify({ id: "tool-fixture", object: "chat.completion.chunk", model: "hermes-fixture", created: 1, choices: [{ index: 0, delta: { role: "assistant", tool_calls: [{ index: 0, id: `native-${name}`, type: "function", function: { name, arguments: JSON.stringify(args) } }] }, finish_reason: null }] })}\n\n`); @@ -111,8 +113,14 @@ test("pinned Hermes streams through the production ACPX host using a no-auth loc clientCapabilities: acpxProfileClientCapabilities("hermes"), systemInstructions: "This is a transport fixture. Reply briefly.", semanticTools: { - tools: ["fixture_probe", "write_document"].map(name => ({ name, description: "Return the supplied marker (simulated semantic authority).", inputSchema: { type: "object", properties: { marker: { type: "string" } }, required: ["marker"], additionalProperties: false } })), - handler: async call => { bridgeCalls.push({ tool: call.tool, arguments: call.arguments }); return { marker: "assigned-tool-returned" }; }, + tools: ["fixture_probe", "write_document", "request_human_input"].map(name => ({ name, description: "Return the supplied marker (simulated semantic authority).", inputSchema: { type: "object", properties: { marker: { type: "string" } }, required: ["marker"], additionalProperties: false } })), + handler: async call => { + bridgeCalls.push({ tool: call.tool, arguments: call.arguments }); + return call.tool === "request_human_input" ? { disposition: "applied", interaction: { + id: "fixture-question", companyId: "fixture-company", issueId: "fixture-issue", sourceRunId: "fixture-run", + kind: "ask_user_questions", status: "pending", continuationPolicy: "wake_assignee", + } } : { marker: "assigned-tool-returned" }; + }, }, runtimeContext: { instructions: { workingCopy: { kind: "agent_files", rootPath: agentFiles, entryPath: "AGENTS.md" } }, skills: [], mcp: {} }, environment: { PATH: process.env.PATH, @@ -170,12 +178,33 @@ test("pinned Hermes streams through the production ACPX host using a no-auth loc assert.equal((await questionTurn.result).status, "completed"); assert.equal(questions, 1, `Native clarify failed: ${JSON.stringify(requests.at(-1)?.body.messages?.filter(message => message.role === "tool"))}`); assert.ok(requests.some(row => row.body.messages?.some(message => message.role === "tool" && String(message.content).includes("Cobalt"))), "The native callback did not receive the submitted answer"); + const governedRequestsBefore = requests.length; + const governedOrder = []; + const governed = host.startTurn({ text: "NATIVE_GOVERNED_QUESTION_FIXTURE: Save the assigned Paperclip question.", requestId: "governed-question-turn", + onExtensionNotification: async (method, params) => { + if (method === "_hermes/usage") { + assert.equal(params.tokens, "reported"); + governedOrder.push("usage"); + } + }, + }); + const governedEvents = []; + for await (const event of governed.events) { + governedEvents.push(event); + if (event.type === "tool_call" && event.tag === "tool_call_update" && event.status === "completed") governedOrder.push("question"); + } + assert.equal((await governed.result).status, "cancelled", "Hermes continued after the committed question"); + assert.equal(requests.length - governedRequestsBefore, 1, "Hermes started another model request after saving the question"); + assert.deepEqual(governedOrder, ["usage", "question"], "The committed question result preceded final native usage"); + assert.ok(governedEvents.some(event => event.type === "tool_call" && event.tag === "tool_call_update" + && event.status === "completed" && JSON.stringify(event.rawOutput).includes("fixture-question")), + `The committed native tool result was lost at interruption: ${JSON.stringify(governedEvents).slice(0, 8000)}`); for (const kind of ["MCP", "MEMORY"]) { const turn = host.startTurn({ text: `NATIVE_${kind}_FIXTURE: Use the native tool.`, requestId: `${kind.toLowerCase()}-turn` }); for await (const _event of turn.events) { /* drain */ } assert.equal((await turn.result).status, "completed"); } - assert.deepEqual(bridgeCalls, [{ tool: "fixture_probe", arguments: { marker: "assigned-tool-roundtrip" } }]); + assert.deepEqual(bridgeCalls, ["request_human_input", "fixture_probe"].map(tool => ({ tool, arguments: { marker: "assigned-tool-roundtrip" } }))); assert.equal(host.steeringCapability()?.steering, true); assert.equal(host.steeringCapability()?.queuedFollowUp, false); const steering = host.startTurn({ text: "NATIVE_STEER_FIXTURE: Continue working until redirected.", requestId: "steering-turn" }); diff --git a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts index 20bd4d3c202..fc3596b8151 100644 --- a/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts +++ b/packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts @@ -15,7 +15,7 @@ import type { AcpPermissionDecision, } from "acpx/runtime"; -import { acpxProfileClientCapabilities, bindAcpxExtensionTurn, validateAcpxRichEvent, createAcpxProfileExtensionAdapter, type AcpxExtensionInput } from "../drivers/acpx/profile-extensions.js"; +import { acpxProfileClientCapabilities, bindAcpxExtensionTurn, validateAcpxRichEvent, createAcpxProfileExtensionAdapter, isHermesCommittedQuestionCompletion, type AcpxExtensionInput } from "../drivers/acpx/profile-extensions.js"; import type { PaperclipQuestionSet } from "../contracts/question-set.js"; import { readProviderUsageBilling, type ProviderUsageBilling } from "../contracts/usage-billing.js"; import { createAcpxToolEventNormalizer, createGrokMessageNormalizer } from "../provider-events.js"; @@ -630,13 +630,19 @@ async function pumpTurn( const normalizeToolEvent = createAcpxToolEventNormalizer(); const normalizeMessage = initializedAgent === "grok" ? createGrokMessageNormalizer() : (event: AcpRuntimeEvent) => event; + const committedQuestions: AcpRuntimeEvent[] = []; for await (const event of runtimeTurn.events) { toolEvidence?.tool(event); + const normalized = normalizeMessage(normalizeToolEvent(boundRuntimeEventForNormalization(event))); + if (openParams?.agent === "hermes" && event.type === "tool_call" + && isHermesCommittedQuestionCompletion({ ...normalized, rawOutput: event.rawOutput }, runId ?? "")) { + if (committedQuestions.length >= MAX_PENDING_INPUTS) throw new Error("Hermes committed question limit exceeded"); + committedQuestions.push(normalized); + continue; + } emit( "runtime.event", - sanitizeRuntimeEvent( - normalizeMessage(normalizeToolEvent(boundRuntimeEventForNormalization(event))), - ), + sanitizeRuntimeEvent(normalized), currentTurnId, ); } @@ -671,6 +677,9 @@ async function pumpTurn( // not replace the provider's authoritative completed/cancelled result. diagnostic("acpx_terminal_usage_unavailable", safeMessage(error)); } + // The native bridge has already stopped work. Reading its prompt receipt + // precedes the tool result that lets the controller cancel this transport. + for (const event of committedQuestions) emit("runtime.event", sanitizeRuntimeEvent(event), currentTurnId); terminal = boundedSidecarValue(result); } catch (error) { terminal = { diff --git a/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.test.ts b/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.test.ts index 8bdc507d8a1..353819dd755 100644 --- a/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.test.ts +++ b/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.test.ts @@ -22,6 +22,45 @@ import type { import type { AcpxRecoveryWorkspaceLease } from "./runtime-sandbox.js"; describe("Codex ACPX harness driver", () => { + it("reads Hermes prompt usage before the committed question completion while ordinary tools stream", async () => { + const result = { disposition: "applied", interaction: { id: "question", companyId: "company", issueId: "issue", + sourceRunId: "run-question", kind: "ask_user_questions", status: "pending", continuationPolicy: "wake_assignee" } }; + const fixture = driverFixture({ agent: "hermes", providerPolicy: { readOnly: false } }, { runtimeEvents: [ + { type: "tool_call", tag: "tool_call", toolCallId: "ordinary", title: "Read", kind: "read", text: "Reading" }, + { type: "tool_call", tag: "tool_call_update", toolCallId: "ordinary", status: "completed", rawOutput: "Read done", text: "Read done" }, + { type: "tool_call", tag: "tool_call", toolCallId: "question", title: "mcp__paperclip__request_human_input", kind: "other", text: "Asking" }, + { type: "tool_call", tag: "tool_call_update", toolCallId: "question", status: "completed", rawOutput: JSON.stringify({ result: JSON.stringify(result) }), text: "Saved" }, + ] }); + const session = await fixture.driver.openSession({ runId: "run-question", normalizedSessionId: "session-1", workingDirectory: "/workspace" }); + let settled = false; + fixture.host.status.mockImplementation(async () => ({ agentSessionId: "agent-1", + models: { currentModelId: "gpt-5.6-sol", availableModelIds: ["gpt-5.6-sol"] }, + ...(settled ? { lastRequestId: "provider-turn-1", requestTokenUsage: { receipt: { + input_tokens: 10, output_tokens: 5, total_tokens: 15, cache_read_input_tokens: 0, cache_creation_input_tokens: 0, + } } } : { requestTokenUsage: {} }), + })); + await session.startTurn({ message: { text: "Ask the assigned question" } }); + const events: PrpEvent[] = []; + let usageAtQuestion: unknown; + const collected = (async () => { + for await (const event of session.events()) { + events.push(event); + if (event.eventType === "tool.execution.completed" && event.payload.executionId === "question") usageAtQuestion = await session.usage(); + if (event.eventType === "turn.interrupted") return; + } + })(); + const completedTool = (event: PrpEvent, id: string) => event.eventType === "tool.execution.completed" && event.payload.executionId === id; + await vi.waitFor(() => expect(events.some(event => completedTool(event, "ordinary"))).toBe(true)); + expect(events.some(event => completedTool(event, "question"))).toBe(false); + settled = true; + fixture.finishTurn({ status: "cancelled", stopReason: "cancelled" }); + await collected; + const questionIndex = events.findIndex(event => completedTool(event, "question")); + expect(questionIndex).toBeGreaterThan(events.findIndex(event => completedTool(event, "ordinary"))); + expect(usageAtQuestion).toMatchObject({ cumulative: { inputTokens: 10, outputTokens: 5, totalTokens: 15 } }); + await session.close({ reason: "question receipt ordering verified" }); + }); + it("delivers negotiated steering and follow-up distinctly, once, for the active turn", async () => { const fixture = driverFixture({ agent: "pi", model: "openrouter/deepseek/deepseek-v4-flash-0731", providerPolicy: { readOnly: true } }); fixture.host.steeringCapability.mockReturnValue({ steering: true, queuedFollowUp: true }); diff --git a/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.ts b/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.ts index de18908c84b..b51c31aeaca 100644 --- a/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.ts +++ b/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.ts @@ -1,7 +1,7 @@ import { isProviderMode } from "../../contracts/provider-mode.js"; import { acpxProfileActivity, type AcpxActivityAdapter, type AcpxToolEvidence } from "./profile-activity.js"; import { requireAcpxResponseDelivery } from "./response-delivery.js"; -import { acpxProfileClientCapabilities, bindAcpxExtensionTurn, validateAcpxRichEvent, createAcpxProfileExtensionAdapter, type AcpxExtensionInput } from "./profile-extensions.js"; +import { acpxProfileClientCapabilities, bindAcpxExtensionTurn, validateAcpxRichEvent, createAcpxProfileExtensionAdapter, isHermesCommittedQuestionCompletion, type AcpxExtensionInput } from "./profile-extensions.js"; import { createHash, randomBytes } from "node:crypto"; import type { @@ -1461,11 +1461,18 @@ class CodexAcpxSession implements HarnessSession { createAcpxToolEventNormalizer(); const normalizeMessage = this.#agent === "grok" ? createGrokMessageNormalizer() : (event: AcpRuntimeEvent) => event; + const committedQuestions: AcpRuntimeEvent[] = []; for await (const event of turn.events) { toolEvidence?.tool(event); const projected = activity.toolExecutionId && event.type === "tool_call" && typeof event.toolCallId === "string" ? { ...event, toolCallId: activity.toolExecutionId(event.toolCallId) } : event; - this.#mapRuntimeEvent(normalizeMessage(normalizeToolEvent(projected)), turnId, ++index); + const normalized = normalizeMessage(normalizeToolEvent(projected)); + if (this.#agent === "hermes" && isHermesCommittedQuestionCompletion(normalized, this.#input.runId)) { + if (committedQuestions.length >= 16) throw new Error("Hermes committed question limit exceeded"); + committedQuestions.push(normalized); + continue; + } + this.#mapRuntimeEvent(normalized, turnId, ++index); } const result = await turn.result; await drainExtensions(); @@ -1481,6 +1488,7 @@ class CodexAcpxSession implements HarnessSession { const estimate = acpxUsageEstimateNotice(receipt, `${turnId}:usage-estimate`); if (estimate) { validateAcpxRichEvent(estimate); this.#emit(estimate.eventType, estimate.payload, { turnId, itemId: estimate.itemId }); } } + for (const event of committedQuestions) this.#mapRuntimeEvent(event, turnId, ++index); this.#cancelPendingRuntimeRequests("provider turn settled", turnId); if (this.#terminalTurns.has(turnId)) return; if (result.status === "completed") { diff --git a/packages/paperclip-runner/src/drivers/acpx/hermes-distributions.ts b/packages/paperclip-runner/src/drivers/acpx/hermes-distributions.ts index 8dfb88a4bef..a28406bade1 100644 --- a/packages/paperclip-runner/src/drivers/acpx/hermes-distributions.ts +++ b/packages/paperclip-runner/src/drivers/acpx/hermes-distributions.ts @@ -1,5 +1,5 @@ /** Reviewed native closures. Setup cannot adopt a digest from downloaded files. */ export const HERMES_CLOSURES: Readonly> = Object.freeze({ - "darwin-arm64": "ad1e555296e51be6d20a7234daeb7028570c7ee8fee9d27fb88c4f550491a07f", - "linux-x64": "353a942b2a88542db8537611535898ae6088045608b88b4ddab491a6b000b8cf", + "darwin-arm64": "f443a6867c914f49d7308c7421dbce9ba4a0d024b0c718a41a78845c784a7bd0", + "linux-x64": "c9879c3b69357d17d375d8fd4897f18496cebb21256c1e8bb9220d779d995921", }); diff --git a/packages/paperclip-runner/src/drivers/acpx/hermes-extensions.test.ts b/packages/paperclip-runner/src/drivers/acpx/hermes-extensions.test.ts index 5836465de57..4838ba7b9f9 100644 --- a/packages/paperclip-runner/src/drivers/acpx/hermes-extensions.test.ts +++ b/packages/paperclip-runner/src/drivers/acpx/hermes-extensions.test.ts @@ -1,7 +1,25 @@ import { describe, expect, it } from "vitest"; -import { createAcpxProfileExtensionAdapter, validateAcpxRichEvent } from "./profile-extensions.js"; +import { createAcpxProfileExtensionAdapter, isHermesCommittedQuestionCompletion, validateAcpxRichEvent } from "./profile-extensions.js"; describe("Hermes native extensions", () => { + const committed = { disposition: "applied", interaction: { id: "question", companyId: "company", issueId: "issue", + sourceRunId: "run", kind: "ask_user_questions", status: "pending", continuationPolicy: "wake_assignee" } }; + const completion = { type: "tool_call", tag: "tool_call_update", status: "completed", title: "mcp__paperclip__request_human_input" }; + it("delays only the current run's structured committed question completion", () => { + for (const rawOutput of [committed, JSON.stringify(committed), { result: committed }, + { result: JSON.stringify(committed) }, { result: "Saved", structuredContent: committed }]) { + expect(isHermesCommittedQuestionCompletion({ ...completion, rawOutput }, "run")).toBe(true); + } + for (const event of [{ ...completion, title: "terminal", rawOutput: committed }, + { ...completion, status: "failed", rawOutput: committed }, { ...completion, tag: "tool_call", rawOutput: committed }, + { ...completion, rawOutput: "Saved a pending question" }, { ...completion, rawOutput: { ...committed, error: "denied" } }, + { ...completion, rawOutput: { result: { ...committed, error: "denied" } } }, + { ...completion, rawOutput: { ...committed, disposition: "rejected" } }, + ...["status", "kind", "continuationPolicy", "sourceRunId", "id", "companyId", "issueId"].map(key => ({ + ...completion, rawOutput: { ...committed, interaction: { ...committed.interaction, [key]: "" } }, + }))]) expect(isHermesCommittedQuestionCompletion(event, "run")).toBe(false); + expect(isHermesCommittedQuestionCompletion({ ...completion, rawOutput: committed }, "other-run")).toBe(false); + }); const adapter = () => createAcpxProfileExtensionAdapter("hermes", { sessionId: "session", turnId: "turn", workspacePath: "/workspace" })!; it("keeps native child identities stable across start, progress and completion", async () => { const context = { version: 1, sessionId: "session", turnToken: "token", childId: "child", delegationId: "batch", model: "exact-model", summary: "Read source", status: "completed" }; diff --git a/packages/paperclip-runner/src/drivers/acpx/profile-extensions.ts b/packages/paperclip-runner/src/drivers/acpx/profile-extensions.ts index a0075a2ddf2..e07fcbbf9fe 100644 --- a/packages/paperclip-runner/src/drivers/acpx/profile-extensions.ts +++ b/packages/paperclip-runner/src/drivers/acpx/profile-extensions.ts @@ -2,7 +2,7 @@ import { createCursorProfileExtensionAdapter, CURSOR_CLIENT_CAPABILITIES } from import { createHash } from "node:crypto"; import type { HarnessRuntimeRequestResolution } from "../../contracts/harness-driver.js"; import { parsePaperclipQuestionSet, parsePaperclipQuestionResponse, type PaperclipQuestionSet } from "../../contracts/question-set.js"; -import { isCanonicalProviderEventType, type CanonicalProviderEvent } from "../../provider-events.js"; +import { isCanonicalProviderEventType, type AcpRuntimeEventShape, type CanonicalProviderEvent } from "../../provider-events.js"; import { validatePrpEvent } from "../../protocol/replay-contract.js"; import type { QualifiedAcpxAgent } from "./qualified-profiles.js"; import { parseProviderUsageBilling, type ProviderUsageBilling } from "../../contracts/usage-billing.js"; @@ -112,6 +112,32 @@ export function acpxProfileClientCapabilities(agent: QualifiedAcpxAgent): Record return agent === "hermes" ? { _meta: { paperclipHermes: { version: 1, billingReceipts: 1 } } } : {}; } +/** Delay this tool's display completion until its native prompt receipt is read. + * This grants no wait or accounting authority; the controller validates both. + */ +export function isHermesCommittedQuestionCompletion(event: AcpRuntimeEventShape, runId: string): boolean { + if (event.type !== "tool_call" || event.tag !== "tool_call_update" || event.status !== "completed" + || event.title !== "mcp__paperclip__request_human_input") return false; + const object = (value: unknown): Record | null => { + if (typeof value === "string") { + if (value.length > 65_536) return null; + try { value = JSON.parse(value); } catch { return null; } + } + return value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record : null; + }; + let result = object(event.rawOutput); + if (!result || "error" in result) return false; + if (!("disposition" in result)) { + if (Object.keys(result).some(key => !["result", "structuredContent", "_meta"].includes(key))) return false; + result = object(result.structuredContent ?? result.result); + } + if (!result || "error" in result || result.disposition !== "applied") return false; + const interaction = object(result.interaction); + return interaction !== null && interaction.kind === "ask_user_questions" && interaction.status === "pending" + && interaction.continuationPolicy === "wake_assignee" && interaction.sourceRunId === runId + && ["id", "companyId", "issueId"].every(key => typeof interaction[key] === "string" && Boolean(interaction[key])); +} + /** Reject an oversized approval document; never silently approve a truncated revision. */ export function validateAcpxExtensionInput(input: AcpxExtensionInput): void { if (!isAcpxCanonicalInputMethod(input.method)) throw new Error("Unsupported ACP input method"); diff --git a/packages/paperclip-runner/src/providers/hermes/bridge.py b/packages/paperclip-runner/src/providers/hermes/bridge.py index f278e090b91..82f04c7a4e6 100644 --- a/packages/paperclip-runner/src/providers/hermes/bridge.py +++ b/packages/paperclip-runner/src/providers/hermes/bridge.py @@ -20,6 +20,7 @@ from acp_adapter.server import HermesACPAgent, _history_replay_updates, _mcp_ser from acp_adapter.session import SessionManager, _parse_model_config, _expand_acp_enabled_toolsets from acp_adapter.events import _send_update, make_step_cb from acp_adapter.tools import build_tool_start, build_tool_complete +from agent.interrupt_compat import request_hard_interrupt from policy import authorize_tool, REPORTING_TOOLS from tool_process import tool_policy, install_tool_process_policy from billing import close_turn, fence_background_dispatch, install_billing_capture, start_turn @@ -29,6 +30,41 @@ ANSWER_MAX_LENGTH = 65536 USAGE_COUNTERS = ("input_tokens", "output_tokens", "cache_read_tokens", "cache_write_tokens", "reasoning_tokens", "total_tokens", "estimated_cost_usd") +def committed_human_input_wait(name, result): + # Only the authenticated assigned Paperclip operation can end native work + # at this boundary. Other tool output and model text cannot signal a wait. + if name != "mcp__paperclip__request_human_input" or not isinstance(result, dict): + return False + if "error" in result: + return False + # The pinned MCP handler wraps the authenticated server's JSON text or + # structuredContent in this carrier. Decode one level, never prose or an + # arbitrary tool's JSON. Direct structured results are accepted too. + if "disposition" not in result: + if set(result) - {"result", "structuredContent", "_meta"}: + return False + structured = result.get("structuredContent", result.get("result")) + if isinstance(structured, str): + if len(structured) > 65536: + return False + try: + structured = json.loads(structured) + except ValueError: + return False + if not isinstance(structured, dict): + return False + result = structured + if "error" in result: + return False + interaction = result.get("interaction") + return (result.get("disposition") == "applied" and isinstance(interaction, dict) + and interaction.get("kind") == "ask_user_questions" + and interaction.get("status") == "pending" + and interaction.get("continuationPolicy") == "wake_assignee" + and all(isinstance(interaction.get(key), str) and bool(interaction[key]) + for key in ("id", "companyId", "issueId", "sourceRunId"))) + + def usage_snapshot(agent): return {key: getattr(agent, "session_" + key, 0) for key in USAGE_COUNTERS} @@ -222,6 +258,7 @@ class ManagedHermesACPAgent(HermesACPAgent): self._usage_before = None self._billing = None self._billing_negotiated = False + self._committed_wait_updates = [] def _build_model_state(self, state): from acp.schema import ModelInfo, SessionModelState @@ -322,6 +359,7 @@ class ManagedHermesACPAgent(HermesACPAgent): self._billing = None self._active = None self._usage_before = None + self._committed_wait_updates.clear() for future in tuple(self._pending_questions): future.cancel() @@ -359,6 +397,12 @@ class ManagedHermesACPAgent(HermesACPAgent): **({"estimatedUsd": after["estimated_cost_usd"] - before["estimated_cost_usd"]} if estimated else {}), **({"billing": billing} if billing is not None else {}), }) + # The controller yields when it sees the committed question result. + # Publish it only after native work and its turn receipt have finished; + # otherwise owned shutdown can overtake the final accounting response. + for update in self._committed_wait_updates: + await conn.session_update(session_id, update) + self._committed_wait_updates.clear() return response def _handle_slash_command(self, *args, **kwargs): @@ -418,6 +462,15 @@ class ManagedHermesACPAgent(HermesACPAgent): envelope = json.loads(result_text) except (ValueError, TypeError): envelope = None + committed_wait = committed_human_input_wait(name, envelope) + if committed_wait: + # Stop inside the native completion callback, before publishing + # the tool result lets the controller observe the durable wait. + # Otherwise Hermes can start another paid request while the + # controller's cancellation crosses the transport boundary. + if not request_hard_interrupt(state.agent, tool_reason="user_interrupt"): + raise ValueError("Hermes cannot stop at a committed Paperclip question") + state.cancel_event.set() # Native policy and MCP transport failures use this exact envelope; # an arbitrary successful payload mentioning errors is not a failure. is_error = (isinstance(envelope, dict) and set(envelope) == {"error"} @@ -429,7 +482,11 @@ class ManagedHermesACPAgent(HermesACPAgent): # Hermes deliberately omits raw output for its polished/structured # cards. ACPX consumes rawOutput for the runner transcript; keep # native content (including diffs) and the original result together. - _send_update(conn, session_id, loop, update.model_copy(update={"raw_output": result_text})) + update = update.model_copy(update={"raw_output": result_text}) + if committed_wait: + self._committed_wait_updates.append(update) + else: + _send_update(conn, session_id, loop, update) billing = self._billing diff --git a/packages/paperclip-runner/src/providers/hermes/test_bridge.py b/packages/paperclip-runner/src/providers/hermes/test_bridge.py index 0c1591348c0..2ee3507f3e8 100644 --- a/packages/paperclip-runner/src/providers/hermes/test_bridge.py +++ b/packages/paperclip-runner/src/providers/hermes/test_bridge.py @@ -261,6 +261,73 @@ class Controls(unittest.IsolatedAsyncioTestCase): self.assertEqual(updates[3].status, "failed") self.assertEqual(json.loads(updates[3].raw_output), {"error": "denied"}) + async def test_committed_question_stops_before_its_native_completion_is_published(self): + callbacks = SimpleNamespace(tool_call_ids={}, tool_call_meta={}) + result = {"disposition": "applied", "interaction": {"id": "question", "companyId": "company", + "issueId": "issue", "sourceRunId": "run", "kind": "ask_user_questions", + "status": "pending", "continuationPolicy": "wake_assignee"}} + name = "mcp__paperclip__request_human_input" + updates = [] + for carrier in [result, {"result": result}, {"result": json.dumps(result)}, + {"result": "Question saved", "structuredContent": result}]: + with self.subTest(carrier=carrier): + self.state.cancel_event.clear() + self.bridge._committed_wait_updates.clear() + callbacks = SimpleNamespace(tool_call_ids={}, tool_call_meta={}) + with patch("bridge.HermesACPAgent._wire_turn_callbacks", return_value=callbacks), \ + patch("bridge._send_update", side_effect=lambda c, s, l, event: updates.append(event)), \ + patch("bridge.request_hard_interrupt", return_value=True) as stop: + self.bridge._wire_turn_callbacks(self.state, "session", Mock(), asyncio.get_running_loop()) + self.state.agent.tool_start_callback("wait", name, {}) + self.state.agent.tool_complete_callback("wait", name, {}, json.dumps(carrier)) + self.assertTrue(self.state.cancel_event.is_set()) + stop.assert_called_once_with(self.state.agent, tool_reason="user_interrupt") + self.assertNotEqual(updates[-1].status, "completed") + self.assertEqual(len(self.bridge._committed_wait_updates), 1) + self.assertEqual(json.loads(self.bridge._committed_wait_updates[0].raw_output), carrier) + + async def test_committed_question_completion_follows_final_native_usage(self): + self.state.agent = SimpleNamespace(**{"session_" + key: 0 for key in USAGE_COUNTERS}, + _paperclip_usage_receipts=[True], _paperclip_cost_receipts=[True]) + self.bridge._usage_before = dict.fromkeys(USAGE_COUNTERS, 0) + self.bridge._billing = TurnBilling() + self.bridge._billing.begin().json(b'{"usage":{"prompt_tokens":20,"completion_tokens":5,"cost":0.0042}}') + update = object() + self.bridge._committed_wait_updates.append(update) + order = [] + conn = SimpleNamespace(ext_notification=AsyncMock(side_effect=lambda *args: order.append("usage")), + session_update=AsyncMock(side_effect=lambda *args: order.append("question"))) + with patch("bridge.HermesACPAgent._finish_turn", new=AsyncMock(return_value=SimpleNamespace(usage=None))): + response = await self.bridge._finish_turn(self.state, "session", conn, {}, None, True) + self.assertEqual(order, ["usage", "question"]) + self.assertEqual(response.usage.total_tokens, 25) + conn.session_update.assert_awaited_once_with("session", update) + self.assertEqual(self.bridge._committed_wait_updates, []) + + async def test_other_or_uncommitted_tool_results_cannot_stop_the_native_turn(self): + original = {"disposition": "applied", "interaction": {"id": "question", "companyId": "company", + "issueId": "issue", "sourceRunId": "run", "kind": "ask_user_questions", + "status": "pending", "continuationPolicy": "wake_assignee"}} + cases = [("mcp__paperclip__read_file", original), ("terminal", original), + ("mcp__paperclip__request_human_input", {"error": "denied"}), + ("mcp__paperclip__request_human_input", {"result": "Saved a pending question"}), + ("mcp__paperclip__request_human_input", {"result": original, "error": "denied"}), + ("mcp__paperclip__request_human_input", {**original, "disposition": "rejected"})] + for key, value in [("status", "resolved"), ("kind", "request_confirmation"), + ("continuationPolicy", "none"), ("sourceRunId", None), ("id", "")]: + cases.append(("mcp__paperclip__request_human_input", + {**original, "interaction": {**original["interaction"], key: value}})) + with patch("bridge.request_hard_interrupt", return_value=True) as stop: + for index, (name, result) in enumerate(cases): + with self.subTest(name=name, result=result): + callbacks = SimpleNamespace(tool_call_ids={}, tool_call_meta={}) + with patch("bridge.HermesACPAgent._wire_turn_callbacks", return_value=callbacks), patch("bridge._send_update"): + self.bridge._wire_turn_callbacks(self.state, "session", Mock(), asyncio.get_running_loop()) + self.state.agent.tool_start_callback(str(index), name, {}) + self.state.agent.tool_complete_callback(str(index), name, {}, result) + self.assertFalse(self.state.cancel_event.is_set()) + stop.assert_not_called() + async def test_pre_dispatch_denials_keep_native_ids_and_complete_exactly_once(self): from hermes_cli import middleware callbacks = SimpleNamespace(tool_call_ids={}, tool_call_meta={})