mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-18 05:23:39 +02:00
fix(hermes): settle native usage before question yield
Stop native Hermes at a committed assigned question. Preserve final prompt usage before its completed tool result reaches the controller cancellation boundary. Keep ordinary tools streaming and add an immediate-shutdown native regression with pinned runtime closure updates. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
e3be06f25e
commit
23919a8787
12 files changed
+358
-21
No files matched your search
@@ -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.
|
||||
@@ -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.
|
||||
|
||||
@@ -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');
|
||||
|
||||
@@ -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" });
|
||||
|
||||
@@ -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<AcpRuntimeEvent>();
|
||||
const normalizeMessage = initializedAgent === "grok"
|
||||
? createGrokMessageNormalizer<AcpRuntimeEvent>() : (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 = {
|
||||
|
||||
@@ -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 });
|
||||
|
||||
@@ -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<AcpRuntimeEvent>();
|
||||
const normalizeMessage = this.#agent === "grok"
|
||||
? createGrokMessageNormalizer<AcpRuntimeEvent>() : (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") {
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/** Reviewed native closures. Setup cannot adopt a digest from downloaded files. */
|
||||
export const HERMES_CLOSURES: Readonly<Record<string, string>> = Object.freeze({
|
||||
"darwin-arm64": "ad1e555296e51be6d20a7234daeb7028570c7ee8fee9d27fb88c4f550491a07f",
|
||||
"linux-x64": "353a942b2a88542db8537611535898ae6088045608b88b4ddab491a6b000b8cf",
|
||||
"darwin-arm64": "f443a6867c914f49d7308c7421dbce9ba4a0d024b0c718a41a78845c784a7bd0",
|
||||
"linux-x64": "c9879c3b69357d17d375d8fd4897f18496cebb21256c1e8bb9220d779d995921",
|
||||
});
|
||||
@@ -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" };
|
||||
|
||||
@@ -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<string, unknown> | 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<string, unknown> : 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");
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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={})
|
||||
|
||||
Reference in new issue
Block a user