diff --git a/doc/SPEC-implementation.md b/doc/SPEC-implementation.md index f50252a776d..931de428964 100644 --- a/doc/SPEC-implementation.md +++ b/doc/SPEC-implementation.md @@ -1554,6 +1554,20 @@ for contracts, recovery behavior, Storybook, and acceptance workflows. Board may override by raising budget or explicitly resuming agent. +Native runs retain final usage receipts during a bounded accounting-only drain +after a governed wait cancels provider work. This does not accept late messages, +tool calls, or new completion proposals. Missing or incomplete receipts continue +to block budget admission. +A controller that detaches for server restart loses checkpoint and completion +write authority. A closing event stream is not proof that the governed run +finished; the replacement controller must adopt and settle the original run. + +Complete direct Anthropic API receipts for `claude-sonnet-5` can use a versioned +list-price estimate when the provider supplies no run price. The receipt records +the rates and assumptions. Aggregate cache writes use the one-hour rate because +their TTL is unknown. Estimates are not invoices. Other models, billers, unknown +billing types, and incomplete receipts remain unpriced. + ## 13.3 Cost Event Ingestion `POST /companies/:companyId/cost-events` body: diff --git a/doc/connections/CONNECTION-SEARCH.md b/doc/connections/CONNECTION-SEARCH.md index 9c0630e21e1..594b10a66ec 100644 --- a/doc/connections/CONNECTION-SEARCH.md +++ b/doc/connections/CONNECTION-SEARCH.md @@ -57,6 +57,12 @@ grants consent. A provider connection is not proof that its underlying app is authorized. Multiple matching aggregator apps are returned as candidates; the agent searches the selected `aggregator.targetService` to get its provider question. +A fuzzy or partial external app match remains a suggestion alongside authorized +installed capability matches. It cannot replace those matches with a mandatory +provider question. For example, a query for a page service can return the +installed page tools alongside a Page X suggestion; selecting the exact Page X +service is required before that external route asks for provider consent. + ## Regression coverage ```sh @@ -69,3 +75,10 @@ Circleback/Attio/ClickUp, indexed Executor tools, and provider consent. The nati tool authority test searches a paragraph longer than the old 200-character limit and then creates a real connection interaction. Existing tests retain checks for private metadata, administrative denials, stale identities, and saved declines. + +When `connection_request` rejects an unknown `toolNames` entry, its error includes +up to twenty exact names from that eligible connection's active tool catalog. +This is discovery metadata: the request still fails with 422, grants no access, +and creates no approval card. The agent must select the needed exact names and +submit a new request; normal user approval, catalog-version and policy checks +still apply. Ineligible connections and inactive catalog entries are excluded. diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 1819311080a..c923e18c66f 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1382,6 +1382,23 @@ controller must recover it through the authenticated ownership checks. This preserves active work and queued messages without treating a server restart as user cancellation. +When a native turn creates a governed approval wait, its checkpoint retains the +exact committed triggering event and yielded disposition before interrupting the +provider. Shutdown gives these turns a shared, bounded 20-second window to finish +settling their usage and terminal evidence before detaching. Recovery verifies +the trigger against the retained event ledger and settles that same turn, even +if the user has already answered the interaction. It replays retained accounting +through the ordinary control-plane observers without starting another model turn. +When a later heartbeat run reuses the provider session, it clears that prior +run's wait marker along with its terminal, result and pending-request authority. +Already-numbered cancellation acknowledgements and cancelled/expired runtime requests +are retained during shutdown so recovery can replay a contiguous journal. They +carry no permission to start work. Stream closure, stream failure or drain timeout +after complete usage leaves the saved wait unfinalized until a provider terminal +is retained; complete token counters alone cannot certify a stopped provider. +A timeout, missing accounting, or an unproven provider terminal remains a failure; +the saved wait does not certify successful provider execution. + Before either shutdown path exits, idle warm sessions close through their normal suspend-and-checkpoint path. Remote sessions therefore leave verified backup authority for the next controller even though their last run is already diff --git a/packages/paperclip-runner/src/backends/harness-driver-backend.test.ts b/packages/paperclip-runner/src/backends/harness-driver-backend.test.ts index d550d594d64..a2cf6f970cd 100644 --- a/packages/paperclip-runner/src/backends/harness-driver-backend.test.ts +++ b/packages/paperclip-runner/src/backends/harness-driver-backend.test.ts @@ -974,6 +974,28 @@ describe("HarnessDriverBackend", () => { await expect(iterator.next()).rejects.toThrow("provider transport lost after resolution"); }); + it.each([false, true])("retains final usage without accepting late output (journal boundary only: %s)", async journalOnly => { + class AccountingAfterCancelSession extends FakeHarnessSession { + async interrupt() {} + override async *events() { + yield prpEvent(1, "item.completed", { kind: "agentMessage", text: "late answer" }); + yield prpEvent(2, "item.completed", { kind: "usage", usage: { runDelta: { inputTokens: 7, outputTokens: 3 }, runDeltaComplete: true } }); + yield prpEvent(3, "turn.completed", { status: "completed" }); + } + } + const backend = new HarnessDriverBackend({ ...driver, async openSession() { return new AccountingAfterCancelSession(); } }); + const session = await backend.openSession({ + identity: { runId: "run-1", sessionId: "session-1", companyId: "company-1", issueId: "issue-1", agentId: "agent-1" }, workingDirectory: "/workspace", + }); + if (journalOnly) session.revokeTurnPublication(); + else await session.cancel({ reason: "durable governed wait", signal: new AbortController().signal }).cleanup; + const events = []; + for await (const event of session.events()) events.push(event); + expect(events.filter(event => event.payload.kind === "agentMessage")).toHaveLength(0); + expect(events.find(event => event.payload.kind === "usage")?.payload.usage).toMatchObject({ runDelta: { inputTokens: 7, outputTokens: 3 } }); + expect(events.some(event => event.eventType === "turn.cancelled")).toBe(true); + }); + it("does not start provider work when Stop arrives before the first turn", async () => { const provider = new FakeHarnessSession(); const start = vi.spyOn(provider, "startTurn"); diff --git a/packages/paperclip-runner/src/backends/harness-driver-backend.ts b/packages/paperclip-runner/src/backends/harness-driver-backend.ts index bf5496c097e..4597400ff3b 100644 --- a/packages/paperclip-runner/src/backends/harness-driver-backend.ts +++ b/packages/paperclip-runner/src/backends/harness-driver-backend.ts @@ -549,7 +549,10 @@ class HarnessNativeSession implements NativeSession { (event.eventType === "run.terminal" && ["failed", "cancelled"].includes(String(event.payload.runTerminalState))) || (event.eventType === "item.completed" && event.payload.kind === "interrupt_acknowledgement"); - if (this.#explicitlyCancelled && !isCancellationEvent) continue; + // Accounting for work already performed survives cancellation. It grants + // no tool, message, semantic-result, or continuation authority. + const isUsageReceipt = event.eventType === "item.completed" && event.payload.kind === "usage"; + if (this.#explicitlyCancelled && !isCancellationEvent && !isUsageReceipt) continue; sourceInstanceId = event.sourceInstanceId; lastSourceSequence = Math.max(lastSourceSequence, event.sourceSeq); if (event.eventType === "runtime_request.created") { @@ -735,6 +738,10 @@ class HarnessNativeSession implements NativeSession { return this.#session.interrupt(input); } + revokeTurnPublication() { + this.#explicitlyCancelled = true; + } + cancel(input: { reason: string; signal: AbortSignal }) { if (input.signal.aborted) { throw ( @@ -744,7 +751,7 @@ class HarnessNativeSession implements NativeSession { // This flag is the adapter's synchronous publication boundary. Provider // interruption happens afterward as passive cleanup, so a slow or broken // transport cannot synthesize or publish new accepted output for the turn. - this.#explicitlyCancelled = true; + this.revokeTurnPublication(); const interrupt = this.#session.interrupt; return { cleanup: diff --git a/packages/paperclip-runner/src/contracts/native-session-backend.ts b/packages/paperclip-runner/src/contracts/native-session-backend.ts index 87af9e2ad05..f473b56a03e 100644 --- a/packages/paperclip-runner/src/contracts/native-session-backend.ts +++ b/packages/paperclip-runner/src/contracts/native-session-backend.ts @@ -63,6 +63,10 @@ export interface PersistedNativeSession { terminal?: PrpTerminalState | null; activeTurnId?: string | null; terminalTurns?: PersistedHarnessTurnTerminal[]; + /** Control-plane disposition bound to the exact committed tool event. It is + * independent of the interaction's later answer and proves no provider terminal. */ + governedWait?: { sourceEvent: PrpEvent; result: PrpStructuredRunResult }; + /** At-most-once marker for resultless disposition repair or restart continuation. */ dispositionOnlyRecoveryConsumed?: boolean; dispositionOnlyRecoveryTurnId?: string | null; @@ -200,6 +204,9 @@ export interface NativeSession { correlationId?: string; }): Promise; interrupt?(input: { turnId?: string; reason?: string }): Promise; + /** Revoke publication synchronously while the control plane journals a wait. + * Does not interrupt the provider; cancel owns the subsequent passive cleanup. */ + revokeTurnPublication?(): void; /** Commit cancellation synchronously; the returned promise owns cleanup only. */ cancel?(input: { reason: string; diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index db0344becc7..e65944682cd 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -4403,7 +4403,9 @@ describe("executeNativeSession recovery", () => { expect(lifecycle).toEqual(["closed"]); }); - it("commits a governed wait without waiting for abort-insensitive provider cleanup", async () => { + it("commits a proved governed wait without waiting for abort-insensitive provider cleanup", async () => { + vi.useFakeTimers(); + try { const lifecycle: string[] = []; let cancellationSignal: AbortSignal | undefined; const cancel = vi.fn(({ signal }: { signal: AbortSignal }) => { @@ -4429,6 +4431,8 @@ describe("executeNativeSession recovery", () => { }, async *events() { yield runnerEvent(1, "item.completed"); + expect(cancel).toHaveBeenCalledOnce(); + yield runnerEvent(2, "turn.interrupted", { providerTerminalObserved: true }); }, async startTurn() { return { turnId: "turn-recovery" }; @@ -4498,7 +4502,13 @@ describe("executeNativeSession recovery", () => { keepSessionOpen: true, onSession: (current) => retainedSessions.push(current), }); - await expect(execution).resolves.toMatchObject({ result: yieldedResult }); + let settled = false; + const completed = execution.then(value => { settled = true; return value; }); + void completed.catch(() => undefined); + await vi.advanceTimersByTimeAsync(9_900); + expect(settled).toBe(false); + await vi.advanceTimersByTimeAsync(200); + await expect(completed).resolves.toMatchObject({ result: yieldedResult }); expect(cancel).toHaveBeenCalledOnce(); expect(cancellationSignal?.aborted).toBe(true); expect(close).toHaveBeenCalledOnce(); @@ -4506,9 +4516,11 @@ describe("executeNativeSession recovery", () => { expect(retainedSessions.at(-1)).toBeNull(); expect(events.map((event) => event.eventType)).toEqual([ "item.completed", + "turn.interrupted", "run.result.accepted", "run.terminal", ]); + } finally { vi.useRealTimers(); } }); it("uses the execution timeout when a completion report has no provider terminal", async () => { @@ -4637,7 +4649,7 @@ describe("executeNativeSession recovery", () => { expect(retainedSessions.at(-1)).toBeNull(); }); - it("retains a reusable session while a semantic terminal releases its remote subscription", async () => { + it.each([false, true])("retains a reusable session while terminal teardown settles (governed=%s)", async (governed) => { const close = vi.fn(async () => undefined); const retainedSessions: Array = []; const session: NativeSession = { @@ -4653,9 +4665,9 @@ describe("executeNativeSession recovery", () => { }, async *events() { try { - yield runnerEvent(1, "run.result.proposed", result); + yield { ...runnerEvent(1, governed ? "item.completed" : "run.result.proposed", governed ? { kind: "dynamicToolCall" } : result), turnId: "turn-recovery" }; yield { - ...runnerEvent(2, "turn.completed"), + ...runnerEvent(2, governed ? "turn.interrupted" : "turn.completed"), turnId: "turn-recovery", }; } finally { @@ -4680,6 +4692,7 @@ describe("executeNativeSession recovery", () => { lineage: [], }; }, + cancel: () => ({ cleanup: Promise.resolve() }), close, }; const backend: NativeSessionBackend = { @@ -4720,10 +4733,11 @@ describe("executeNativeSession recovery", () => { controlPlane: port, runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery", + resolveGovernedWait: governed ? () => yieldedResult : undefined, keepSessionOpen: true, onSession: (current) => retainedSessions.push(current), }), - ).resolves.toMatchObject({ result, terminal }); + ).resolves.toMatchObject({ result: governed ? yieldedResult : result }); expect(close).not.toHaveBeenCalled(); expect(retainedSessions).toEqual([session]); @@ -7641,7 +7655,192 @@ describe("executeNativeSession recovery", () => { ]); }); - it("parks a provider turn immediately after a durable governed wait appears", async () => { + it.each(["end", "throw", "timeout"])("keeps a governed wait unfinalized when complete usage is followed by %s without a provider terminal", async fault => { + vi.useFakeTimers(); + try { + const trigger = { ...runnerEvent(1, "item.completed", { kind: "dynamicToolCall" }), turnId: "turn-waiting" }; + const usage = { ...runnerEvent(2, "item.completed", { kind: "usage", usage: { runDelta: { inputTokens: 7, outputTokens: 3 }, runDeltaComplete: true } }), turnId: trigger.turnId }; + const history: PrpEvent[] = []; + let checkpoint: PersistedNativeSession | undefined; + let releaseStream!: () => void; + const streamClosed = new Promise(resolve => { releaseStream = resolve; }); + const complete = vi.fn(async () => undefined); + const session: NativeSession = { + identity: () => identity, + async capabilities() { return { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; }, + async startTurn() { return { turnId: trigger.turnId }; }, + cancel() { return { cleanup: Promise.resolve() }; }, + async snapshot() { return { backendKind: "mock", sessionId: identity.sessionId, identity, activeTurnId: trigger.turnId }; }, + async result() { return null; }, async close() { releaseStream(); }, + async *events() { + yield trigger; yield usage; + if (fault === "throw") throw new Error("provider stream lost"); + if (fault === "timeout") await streamClosed; + }, + }; + const backend: NativeSessionBackend = { + async descriptor() { return { kind: "mock", name: "missing-stop", version: "1", capabilities: await session.capabilities() }; }, + async openSession() { return session; }, + }; + const port: ControlPlanePort = { + async openRun() {}, completeRun: complete, + async checkpointSession(value) { checkpoint = structuredClone(value); }, + async appendEvent(event) { history.push(event); return { cursor: history.length, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; }, + async replayEvents(request) { return { events: history.filter(row => row.sourceInstanceId === request.sourceInstanceId && row.sourceSeq > request.afterSourceSeq), highestContiguousSourceSeq: 0 }; }, + }; + const outcome = expect(executeNativeSession({ input, backend, controlPlane: port, timeoutMs: 1_000, + runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery", resolveGovernedWait: () => yieldedResult })) + .rejects.toMatchObject({ providerCode: "governed_wait_provider_terminal_unproven" }); + void outcome.catch(() => undefined); + await vi.advanceTimersByTimeAsync(600); + await outcome; + expect(checkpoint?.governedWait).toEqual({ sourceEvent: trigger, result: yieldedResult }); + expect(complete).not.toHaveBeenCalled(); + expect(history).toEqual([trigger, usage]); + } finally { vi.useRealTimers(); } + }); + + it.each(["interrupt_acknowledgement", "runtime_request.cancelled", "runtime_request.expired"])("retains %s in contiguous governed-wait history for recovery", async cancellationKind => { + const trigger = { ...runnerEvent(1, "item.completed", { kind: "dynamicToolCall" }), turnId: "turn-waiting" }; + const acknowledgement: PrpEvent = { ...runnerEvent(2, cancellationKind === "interrupt_acknowledgement" ? "item.completed" : cancellationKind as "runtime_request.cancelled" | "runtime_request.expired", + cancellationKind === "interrupt_acknowledgement" ? { kind: cancellationKind } : { requestId: "request-1" }), turnId: trigger.turnId }; + const usage = { ...runnerEvent(3, "item.completed", { kind: "usage", usage: { runDelta: { inputTokens: 7, outputTokens: 3 }, runDeltaComplete: true } }), turnId: trigger.turnId }; + const stopped = { ...runnerEvent(4, "turn.interrupted", { providerTerminalObserved: true }), turnId: trigger.turnId }; + const history: PrpEvent[] = []; + let checkpoint: PersistedNativeSession | undefined; + const start = vi.fn(async () => ({ turnId: trigger.turnId })); + const cancel = vi.fn(() => ({ cleanup: Promise.resolve() })); + const complete = vi.fn(async () => undefined); + const session: NativeSession = { + identity: () => identity, + async capabilities() { return { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; }, + startTurn: start, cancel, + async snapshot() { return { backendKind: "mock", sessionId: identity.sessionId, identity, activeTurnId: trigger.turnId }; }, + async result() { return null; }, async close() {}, + async *events() { yield trigger; yield acknowledgement; yield usage; yield stopped; }, + }; + const backend: NativeSessionBackend = { + async descriptor() { return { kind: "mock", name: "cancel-history", version: "1", capabilities: await session.capabilities() }; }, + async openSession() { return session; }, async recoverSession() { return { recovered: true, session }; }, + }; + const port: ControlPlanePort = { + async openRun() {}, completeRun: complete, + async checkpointSession(value) { checkpoint = structuredClone(value); }, + async appendEvent(event) { + const duplicate = history.some(row => row.sourceEventId === event.sourceEventId); + if (!duplicate) history.push(event); + return { cursor: history.length, highestContiguousSourceSeq: event.sourceSeq, disposition: duplicate ? "duplicate" : "committed" }; + }, + async replayEvents(request) { + const events = history.filter(row => row.sourceInstanceId === request.sourceInstanceId && row.sourceSeq > request.afterSourceSeq); + return { events, highestContiguousSourceSeq: events.at(-1)?.sourceSeq ?? request.afterSourceSeq }; + }, + }; + const options = { input, backend, controlPlane: port, runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery" }; + await expect(executeNativeSession({ ...options, resolveGovernedWait: () => yieldedResult })) + .resolves.toMatchObject({ result: yieldedResult }); + expect(history.filter(event => event.sourceInstanceId === trigger.sourceInstanceId)).toEqual([trigger, acknowledgement, usage, stopped]); + await expect(executeNativeSession({ ...options, persistedSession: checkpoint })).resolves.toMatchObject({ result: yieldedResult }); + expect(start).toHaveBeenCalledOnce(); + expect(cancel).toHaveBeenCalledOnce(); + }); + + it.each(["trigger", "usage", "terminal"])("recovers a journaled governed wait after its answer is saved (last committed: %s)", async lastCommitted => { + const terminalCommitted = lastCommitted === "terminal"; + const trigger = { ...runnerEvent(1, "item.completed", { kind: "dynamicToolCall", item: { id: "ask-1", name: "request_human_input" } }), turnId: "turn-waiting" }; + const usage = { ...runnerEvent(2, "item.completed", { kind: "usage", usage: { runDelta: { inputTokens: 7, outputTokens: 3 }, runDeltaComplete: true } }), turnId: trigger.turnId }; + const stopped = { ...runnerEvent(3, "turn.interrupted", { providerTerminalObserved: true }), turnId: trigger.turnId }; + const history: PrpEvent[] = []; + let checkpoint: PersistedNativeSession | undefined; + let crashed = false; + let interrupted = false; + const revoke = vi.fn(); + const cancel = vi.fn(() => { interrupted = true; return { cleanup: Promise.resolve() }; }); + const start = vi.fn(async () => ({ turnId: trigger.turnId })); + const complete = vi.fn(async () => undefined); + const snapshot = (): PersistedNativeSession => ({ backendKind: "mock", sessionId: identity.sessionId, identity, + providerSessionId: "same-provider", activeTurnId: trigger.turnId, cursor: String(history.at(-1)?.sourceSeq ?? 0) }); + const makeSession = (recovering: boolean): NativeSession => ({ identity: () => identity, + async capabilities() { return { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; }, + startTurn: start, revokeTurnPublication: revoke, cancel, + async snapshot() { return snapshot(); }, async result() { return null; }, async close() {}, + async *events() { + if (!recovering) { yield trigger; return; } + // The restart resumes settlement only; it must not submit a model turn. + expect(interrupted).toBe(true); + yield usage; + yield stopped; + }, + }); + const backend: NativeSessionBackend = { + async descriptor() { return { kind: "mock", name: "durable-wait", version: "1", capabilities: await makeSession(false).capabilities() }; }, + async openSession() { return makeSession(false); }, + async recoverSession() { return { recovered: true, session: makeSession(true) }; }, + }; + const port: ControlPlanePort = { + async openRun() {}, completeRun: complete, + async checkpointSession(value) { + checkpoint = structuredClone(value); + if (value.governedWait && !crashed) { + expect(revoke).toHaveBeenCalledOnce(); + expect(cancel).not.toHaveBeenCalled(); + crashed = true; + throw new Error("controller lost after governed checkpoint"); + } + }, + async appendEvent(event) { + if (!history.some(row => row.sourceEventId === event.sourceEventId)) history.push(event); + return { cursor: history.length, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; + }, + async replayEvents(request) { + const events = history.filter(row => row.sourceInstanceId === request.sourceInstanceId && row.sourceSeq > request.afterSourceSeq); + return { events, highestContiguousSourceSeq: events.at(-1)?.sourceSeq ?? request.afterSourceSeq }; + }, + }; + const options = { input, backend, controlPlane: port, runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery" }; + await expect(executeNativeSession({ ...options, resolveGovernedWait: () => yieldedResult })) + .rejects.toThrow("controller lost after governed checkpoint"); + expect(checkpoint?.governedWait).toEqual({ sourceEvent: trigger, result: yieldedResult }); + expect(complete).not.toHaveBeenCalled(); + // The failed controller's ordinary cleanup also invokes cancel. Its + // checkpoint assertion above proves interruption followed persistence. + cancel.mockClear(); interrupted = false; + if (lastCommitted !== "trigger") history.push(usage); + if (terminalCommitted) history.push(stopped); + const answerAlreadySaved = vi.fn(() => null); + await expect(executeNativeSession({ ...options, persistedSession: checkpoint, resolveGovernedWait: answerAlreadySaved })) + .resolves.toMatchObject({ result: yieldedResult, turnId: trigger.turnId, terminal: { runTerminalState: "succeeded" } }); + expect(start).toHaveBeenCalledOnce(); + expect(cancel).toHaveBeenCalledTimes(terminalCommitted ? 0 : 1); + expect(answerAlreadySaved).not.toHaveBeenCalled(); + expect(complete).toHaveBeenCalledOnce(); + expect(history.filter(event => event.payload.kind === "usage")).toEqual([usage]); + }); + + it.each(["missing-trigger", "changed-trigger", "unproven-terminal", "different-turn"])("rejects unsafe governed-wait recovery: %s", async fault => { + const trigger = { ...runnerEvent(1, "item.completed", { kind: "dynamicToolCall" }), turnId: "turn-waiting" }; + const checkpoint: PersistedNativeSession = { backendKind: "mock", sessionId: identity.sessionId, identity, + activeTurnId: trigger.turnId, governedWait: { sourceEvent: trigger, result: yieldedResult } }; + const history = fault === "missing-trigger" ? [] : [fault === "changed-trigger" ? { ...trigger, payload: {} } : trigger]; + if (fault === "unproven-terminal") history.push({ ...trigger, sourceSeq: 2, sourceEventId: "unproven", eventType: "turn.failed", payload: { providerTerminalObserved: false } }); + if (fault === "different-turn") history.push({ ...trigger, sourceSeq: 2, sourceEventId: "new-turn", turnId: "different", eventType: "turn.started" }); + const start = vi.fn(); + const complete = vi.fn(); + const session: NativeSession = { identity: () => identity, startTurn: start, async result() { return null; }, + async capabilities() { return { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; }, + async snapshot() { return checkpoint; }, async *events() {}, async close() {} }; + const backend: NativeSessionBackend = { async descriptor() { return { kind: "mock", name: "unsafe-wait", version: "1", capabilities: await session.capabilities() }; }, + async openSession() { return session; }, async recoverSession() { return { recovered: true, session }; } }; + const port: ControlPlanePort = { async openRun() {}, async checkpointSession() {}, async appendEvent() { throw new Error("unexpected append"); }, + completeRun: complete, async replayEvents(request) { return { events: history.filter(row => row.sourceSeq > request.afterSourceSeq), highestContiguousSourceSeq: history.at(-1)?.sourceSeq ?? 0 }; } }; + await expect(executeNativeSession({ input, backend, controlPlane: port, persistedSession: checkpoint, + runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery" })).rejects.toThrow(/governed_wait|failed terminal/); + expect(start).not.toHaveBeenCalled(); expect(complete).not.toHaveBeenCalled(); + }); + + it.each([0, 6_000, 12_000])("retains actual shutdown usage after %s ms before finalizing a governed wait", async (receiptDelayMs) => { + vi.useFakeTimers(); + try { const yielded: PrpStructuredRunResult = { schema: "paperclip.run_result.v1", reportedWorkDisposition: "yielded", @@ -7681,8 +7880,8 @@ describe("executeNativeSession recovery", () => { turnId: "turn-waiting", }; const turnInterrupted: PrpEvent = { - ...controlEvent(2, "turn.interrupted", { reason: "governed_wait" }), - sourceEventId: "provider-recovery:2", + ...controlEvent(3, "turn.interrupted", { reason: "governed_wait" }), + sourceEventId: "provider-recovery:3", sourceInstanceId: "provider-recovery", sourceKind: "provider", turnId: "turn-waiting", @@ -7711,6 +7910,9 @@ describe("executeNativeSession recovery", () => { async *events() { yield itemCompleted; await cancelled; + await new Promise(resolve => setTimeout(resolve, receiptDelayMs)); + yield { ...itemCompleted, sourceSeq: 2, sourceEventId: "provider-recovery:2", + payload: { kind: "usage", usage: { runDelta: { inputTokens: 7, outputTokens: 3 }, runDeltaComplete: true } } }; yield turnInterrupted; }, async startTurn() { @@ -7778,7 +7980,7 @@ describe("executeNativeSession recovery", () => { completeRun, }; - const completed = await executeNativeSession({ + const execution = executeNativeSession({ input, backend, controlPlane: port, @@ -7788,7 +7990,10 @@ describe("executeNativeSession recovery", () => { event.eventType === "item.completed" ? yielded : null, }); + await vi.advanceTimersByTimeAsync(receiptDelayMs + 1); + const completed = await execution; expect(cancel).toHaveBeenCalledOnce(); + expect(events.find(event => event.payload.kind === "usage")?.payload.usage).toMatchObject({ runDelta: { inputTokens: 7, outputTokens: 3 } }); expect(completed).toMatchObject({ result: yielded, terminal: { @@ -7799,9 +8004,12 @@ describe("executeNativeSession recovery", () => { }); expect(events.map((event) => event.eventType)).toEqual([ "item.completed", + "item.completed", + "turn.interrupted", "run.result.accepted", "run.terminal", ]); + } finally { vi.useRealTimers(); } }); it("hands a committed structured input to the durable wait after its live window", async () => { diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index df65c4450ab..e49b4f6b612 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -47,6 +47,11 @@ const FAILED_OPERATION_SETTLEMENT_GRACE_MS = 100; // bounded by the transport. Other iterator and handoff cleanup keeps the short // fail-closed quarantine boundary below. const REUSABLE_SESSION_CANCELLATION_SETTLEMENT_GRACE_MS = 10_000; +// Governed cancellation can traverse the ACPX turn cancellation (2s), host +// cancellation (2s), protocol/TERM/KILL shutdown (7s), and final status read +// (1s). Allow that bounded settlement plus transport scheduling; never treat +// elapsed time or a cancellation acknowledgement as usage/terminal evidence. +const GOVERNED_ACCOUNTING_DRAIN_MS = 15_000; const DEFAULT_NATIVE_CHECKPOINT_TIMEOUT_MS = 30_000; type NativeSessionCleanupDomain = string; @@ -870,12 +875,15 @@ async function consumeTurn( }, initialGoal?: HarnessThreadGoal | null, resumeInterruptedTurn?: (event: PrpEvent, signal: AbortSignal) => Promise, + checkpointGovernedWait?: (wait: NonNullable, signal: AbortSignal) => Promise, + initialGovernedWait?: PersistedNativeSession["governedWait"], ) { let timer: ReturnType | undefined; const appendAbort = new AbortController(); const governedCleanupOperations = new Set>(); let governedCancellationCommitted = false; let semanticResultObserved = false; + let governedProviderTerminalObserved = false; let deferredGovernedCleanupSettlement: Promise | null = null; let deferredSessionCancellationSettlement: Promise | null = null; const inputTimers = new Map>(); @@ -922,7 +930,7 @@ async function consumeTurn( let latestSessionGoal: HarnessThreadGoal | null = null; let resultSource: "semantic_result" | "governed_wait" | null = null; let providerFailure: NativeProviderTerminalFailure | null = null; - const settleDurableResult = ( + const settleDurableResult = async ( event: PrpEvent, result: PrpStructuredRunResult, reason: string, @@ -930,16 +938,77 @@ async function consumeTurn( if (session.cancel === undefined) { throw new Error("native_governed_wait_cancellation_unavailable"); } - const cancellation = session.cancel({ - reason, - signal: appendAbort.signal, - }); + // Revoke new publication before any asynchronous journal write. Persist + // the disposition before provider interruption, retaining usage and terminal + // evidence even when the interaction is answered during the restart. + session.revokeTurnPublication?.(); + await checkpointGovernedWait?.({ sourceEvent: event, result }, appendAbort.signal); + appendAbort.signal.throwIfAborted(); + const cancellation = session.cancel({ reason, signal: appendAbort.signal }); governedCancellationCommitted = true; const cleanup = cancellation.cleanup; governedCleanupOperations.add(cleanup); void cleanup .catch(() => quarantineSession("governed_cleanup_failed")) .finally(() => governedCleanupOperations.delete(cleanup)); + // Cancellation revokes provider work immediately, but shutdown can still + // supply its final usage. Keep a bounded accounting-only drain before + // acknowledging the run; otherwise the next wake sees permanent debt. + // Never let a stuck stream or cleanup revoke the durable governed wait. + const drainAbort = new AbortController(); + const stopDrain = () => drainAbort.abort(appendAbort.signal.reason); + appendAbort.signal.addEventListener("abort", stopDrain, { once: true }); + const drainMs = Math.min(GOVERNED_ACCOUNTING_DRAIN_MS, timeoutMs > 0 ? Math.max(1, Math.floor(timeoutMs / 4)) : GOVERNED_ACCOUNTING_DRAIN_MS); + let drainTimer: ReturnType | undefined; + const drained = (async () => { + while (!drainAbort.signal.aborted) { + const next = await eventIterator.next(); + if (next.done || drainAbort.signal.aborted) return; + const trailing = next.value; + const terminal = isTurnTerminal(trailing); + const usage = trailing.eventType === "item.completed" && trailing.payload.kind === "usage"; + // These already-numbered cancellation facts carry no new work + // authority. Retain them so later receipts have contiguous replay. + const cancellation = trailing.eventType === "runtime_request.cancelled" + || trailing.eventType === "runtime_request.expired" + || (trailing.eventType === "item.completed" && trailing.payload.kind === "interrupt_acknowledgement") + || (trailing.eventType === "run.terminal" && ["failed", "cancelled"].includes(String(trailing.payload.runTerminalState))); + if (trailing.turnId === event.turnId && (usage || terminal || cancellation)) { + const receipt = await controlPlane.appendEvent(trailing, { signal: drainAbort.signal }); + if (drainAbort.signal.aborted) return; + eventCount += receipt.disposition === "committed" ? 1 : 0; + highestContiguousSourceSeq = Math.max(highestContiguousSourceSeq, receipt.highestContiguousSourceSeq); + if (terminal) { + if (trailing.eventType === "turn.failed" || trailing.payload.providerTerminalObserved === false) { + throw new NativeProviderTerminalFailure("governed_wait_provider_terminal_unproven", false); + } + governedProviderTerminalObserved = true; + return; + } + } + } + })(); + await Promise.race([ + drained.catch(error => { + quarantineSession("governed_accounting_drain_failed"); + if (error instanceof NativeProviderTerminalFailure) throw error; + }), + new Promise(resolve => { drainTimer = setTimeout(() => { + drainAbort.abort(new Error("governed accounting drain timed out")); + quarantineSession("governed_accounting_drain_timed_out"); + resolve(); + }, drainMs); }), + ]).finally(() => { + clearTimeout(drainTimer); + appendAbort.signal.removeEventListener("abort", stopDrain); + drainAbort.abort(); + }); + // A durable wait and complete usage cannot prove that the provider + // stopped. Leave the journaled wait recoverable, without a successful + // result or an accounting-completion boundary, if shutdown is unproven. + if (!governedProviderTerminalObserved) { + throw new NativeProviderTerminalFailure("governed_wait_provider_terminal_unproven", false); + } return { event, eventCount, @@ -947,6 +1016,10 @@ async function consumeTurn( governedResult: result, }; }; + if (initialGovernedWait) { + return settleDurableResult(initialGovernedWait.sourceEvent, initialGovernedWait.result, + "Paperclip resumed settlement of this durable governed interaction."); + } while (true) { const next = await eventIterator.next().catch((error) => { throw providerFailure ?? error; @@ -1309,15 +1382,14 @@ async function consumeTurn( ); } else { // Iterator and provider cleanup own no control-plane mutation authority. - // A reusable session with a semantic result needs a longer bounded - // window for the remote event subscription to release. This applies - // after the provider terminal, which still crosses the remote PRP - // acknowledgement boundary and routinely takes longer than - // the generic local cleanup grace. Governed waits and unrelated stalled - // cleanup retain the short fail-closed boundary. + // A reusable session needs a bounded window for its remote terminal + // subscription to release. A governed wait qualifies only after its + // actual provider terminal was retained during the accounting drain. + // Missing terminal evidence and unrelated stalled cleanup retain the + // short fail-closed boundary. const teardownSettled = await settlesWithin( passiveTeardownSettlement, - semanticResultObserved + semanticResultObserved || governedProviderTerminalObserved ? reusableSessionCancellationGraceMs : FAILED_OPERATION_SETTLEMENT_GRACE_MS, ); @@ -1871,6 +1943,42 @@ export async function executeNativeSession( issueId: input.binding.issueId, agentId: input.binding.agentId, }; + let governedWait = persistedSession?.governedWait; + const governedReplay: PrpEvent[] = []; + if (governedWait) { + const event = governedWait.sourceEvent; + const validation = validatePrpStructuredRunResult(governedWait.result); + if (!event?.turnId || event.runId !== identity.runId || event.normalizedSessionId !== identity.sessionId + || event.sourceInstanceId !== options.runnerInstanceId || !Number.isSafeInteger(event.sourceSeq) || event.sourceSeq < 1 + || !validation.ok || validation.result.reportedWorkDisposition !== "yielded" + || validation.result.completionClaim?.contractRevision !== input.completionContract.contract.revision) { + throw new Error("native_governed_wait_checkpoint_mismatch"); + } + await runAbortableOperationWithin({ + timeoutMs: options.timeoutMs ?? 900_000, + timeoutMessage: "native governed-wait recovery replay timed out", + operation: async signal => { + let afterSourceSeq = event.sourceSeq - 1; + while (true) { + const page = await options.controlPlane.replayEvents({ runId: identity.runId, + sourceInstanceId: options.runnerInstanceId, afterSourceSeq, limit: 1_000 }, { signal }); + signal.throwIfAborted(); + if (!page.events.length) break; + for (const entry of page.events) { + if (entry.runId !== identity.runId || entry.normalizedSessionId !== identity.sessionId + || entry.sourceInstanceId !== event.sourceInstanceId || entry.sourceSeq !== afterSourceSeq + 1 + || governedReplay.length >= 10_000) throw new Error("native_governed_wait_replay_mismatch"); + governedReplay.push(entry); + afterSourceSeq = entry.sourceSeq; + } + } + }, + }); + if (canonicalJson(governedReplay[0]) !== canonicalJson(event)) throw new Error("native_governed_wait_trigger_unproven"); + if (governedReplay.some(entry => entry.eventType === "turn.started" && entry.turnId !== event.turnId)) { + throw new Error("native_governed_wait_turn_mismatch"); + } + } let recovered = false; let session: NativeSession | null = null; let continuityBreak: { @@ -2152,7 +2260,7 @@ export async function executeNativeSession( externalSignal?: AbortSignal, ) => persistCheckpointWithin({ - snapshot, + snapshot: governedWait ? { ...snapshot, governedWait } : snapshot, controlPlane: options.controlPlane, onCheckpoint: options.onCheckpoint, timeoutMs: checkpointTimeoutMs, @@ -2240,6 +2348,21 @@ export async function executeNativeSession( await persistCheckpoint(recoveredSnapshot); } + const governedTerminal = governedWait ? governedReplay.find(event => + event.turnId === governedWait!.sourceEvent.turnId && isTurnTerminal(event)) : undefined; + if (governedTerminal?.eventType === "turn.failed" || governedTerminal?.payload.providerTerminalObserved === false) { + throw new NativeProviderTerminalFailure("governed_wait_provider_terminal_unproven", false); + } + if (governedWait && recoveredActiveTurnId && recoveredActiveTurnId !== governedWait.sourceEvent.turnId) { + throw new Error("native_governed_wait_turn_mismatch"); + } + if (governedWait) { + for (const event of governedReplay) { + if (event.turnId === governedWait.sourceEvent.turnId && (event.payload.kind === "usage" || isTurnTerminal(event))) { + await options.controlPlane.appendEvent(event); + } + } + } let consumed = { event: null as PrpEvent | null, eventCount: 0, @@ -2322,7 +2445,10 @@ export async function executeNativeSession( replayedDisposition?.terminal ?? dispositionFallback; const consumptionAbort = new AbortController(); const consuming = - recoveryTerminal === null + governedTerminal && governedWait + ? Promise.resolve({ event: governedWait.sourceEvent, governedResult: governedWait.result, + eventCount: 0, highestContiguousSourceSeq: governedTerminal.sourceSeq }) + : recoveryTerminal === null ? consumeTurn( session, options.controlPlane, @@ -2345,6 +2471,11 @@ export async function executeNativeSession( : undefined, recoveredSnapshot.goal, resumeInterruptedTurn, + async (wait, signal) => { + governedWait = structuredClone(wait); + await checkpoint(signal); + }, + governedWait, ) : Promise.resolve({ event: recoveryTerminal, @@ -2360,12 +2491,12 @@ export async function executeNativeSession( // that later rejection becomes process-fatal under Node's strict policy. void consuming.catch(() => undefined); try { - const shouldStartFreshTurn = + const shouldStartFreshTurn = !governedWait && ( !recovered || (!recoveredActiveTurnId && !adoptedProviderTerminal && !checkpointedDispositionTerminal && - !dispositionRecoveryStillOwned); + !dispositionRecoveryStillOwned)); if (options.sessionGoalControl) { // Explicit controls remain authoritative after controller loss. In // particular, pause/clear/edit must reach a recovered session even diff --git a/server/src/__tests__/adapter-cost-failure-paths.test.ts b/server/src/__tests__/adapter-cost-failure-paths.test.ts index 0c15c0081e4..4703f374321 100644 --- a/server/src/__tests__/adapter-cost-failure-paths.test.ts +++ b/server/src/__tests__/adapter-cost-failure-paths.test.ts @@ -29,6 +29,8 @@ const cases: Array<{ name: string; execute: () => Promise; event: unkno ]; describe("CLI adapter accounting on timeout", () => { + // Keep OpenCode runtime config copies inside each fixture; host config may + // include plugins and large dependency trees unrelated to accounting. const directories: string[] = []; afterEach(async () => { vi.clearAllMocks(); for (const dir of directories.splice(0)) await rm(dir, { recursive: true, force: true }); }); it.each(cases.filter(f => ["codex", "cursor", "opencode"].includes(f.name)))("$name preserves malformed optional counters through the checkpoint wrapper", async fixture => { @@ -49,7 +51,7 @@ describe("CLI adapter accounting on timeout", () => { runId: "invalid-counter-run", agent: { id: "agent", companyId: "company", name: "Worker", adapterType: `${fixture.name}_local`, adapterConfig: {} }, runtime: { sessionId: null, sessionParams: null, sessionDisplayId: null, taskKey: null }, config: { engine: "cli", command, cwd: dir, model: fixture.name === "opencode" ? "openai/test" : "gpt-6-astra", paperclipRuntimeSkills: [], - env: { OPENAI_API_KEY: "fixture", OPENCODE_ALLOW_ALL_MODELS: "1" } }, + env: { XDG_CONFIG_HOME: dir, OPENAI_API_KEY: "fixture", OPENCODE_ALLOW_ALL_MODELS: "1" } }, context: {}, onLog: async () => {}, onUsage, }); const valid = cached === 0 || cached === undefined; @@ -77,7 +79,7 @@ describe("CLI adapter accounting on timeout", () => { runId: "test-run", agent: { id: "test-agent", companyId: "test-company", name: "Accounting", adapterType: `${fixture.name}_local`, adapterConfig: {} }, runtime: { sessionId: null, sessionParams: null, sessionDisplayId: null, taskKey: null }, config: { engine: "cli", command, cwd: dir, model: ["opencode", "pi"].includes(fixture.name) ? "openai/test" : fixture.name === "cursor" ? "gpt-5" : "test", paperclipRuntimeSkills: [], - env: { OPENAI_API_KEY: "test-placeholder", ANTHROPIC_API_KEY: "test-placeholder", GEMINI_API_KEY: "test-placeholder", OPENCODE_ALLOW_ALL_MODELS: "1", CLAUDE_CONFIG_DIR: dir } }, + env: { XDG_CONFIG_HOME: dir, OPENAI_API_KEY: "test-placeholder", ANTHROPIC_API_KEY: "test-placeholder", GEMINI_API_KEY: "test-placeholder", OPENCODE_ALLOW_ALL_MODELS: "1", CLAUDE_CONFIG_DIR: dir } }, context: {}, onLog: async () => {}, onUsage, }; const execute = await fixture.execute(); @@ -211,7 +213,7 @@ describe("CLI adapter accounting on timeout", () => { const result = await execute({ runId: "tail-run", agent: { id: "test-agent", companyId: "test-company", name: "Accounting", adapterType: `${fixture.name}_local`, adapterConfig: {} }, runtime: { sessionId: null, sessionParams: null, sessionDisplayId: null, taskKey: null }, - config: { engine: "cli", command, cwd: dir, model: "anthropic/test", paperclipRuntimeSkills: [], env: { ANTHROPIC_API_KEY: "fixture", OPENAI_BASE_URL: "https://unrelated-proxy.example/v1", OPENROUTER_API_KEY: "unrelated-key", OPENCODE_ALLOW_ALL_MODELS: "1" } }, + config: { engine: "cli", command, cwd: dir, model: "anthropic/test", paperclipRuntimeSkills: [], env: { XDG_CONFIG_HOME: dir, ANTHROPIC_API_KEY: "fixture", OPENAI_BASE_URL: "https://unrelated-proxy.example/v1", OPENROUTER_API_KEY: "unrelated-key", OPENCODE_ALLOW_ALL_MODELS: "1" } }, context: {}, onLog: async () => {}, onUsage, }); expect(result.usage).toMatchObject({ inputTokens: 40, outputTokens: 20, cachedInputTokens: 200 }); @@ -231,7 +233,7 @@ describe("CLI adapter accounting on timeout", () => { runId: "test-run", agent: { id: "test-agent", companyId: "test-company", name: "Accounting", adapterType: `${fixture.name}_local`, adapterConfig: {} }, runtime: { sessionId: null, sessionParams: null, sessionDisplayId: null, taskKey: null }, config: { engine: "cli", command, cwd: dir, model: ["opencode", "pi"].includes(fixture.name) ? "openai/test" : fixture.name === "cursor" ? "gpt-5" : "test", paperclipRuntimeSkills: [], - env: { OPENAI_API_KEY: "test-placeholder", ANTHROPIC_API_KEY: "test-placeholder", GEMINI_API_KEY: "test-placeholder", MOONSHOT_API_KEY: "test-placeholder", OPENCODE_ALLOW_ALL_MODELS: "1", CLAUDE_CONFIG_DIR: dir } }, + env: { XDG_CONFIG_HOME: dir, OPENAI_API_KEY: "test-placeholder", ANTHROPIC_API_KEY: "test-placeholder", GEMINI_API_KEY: "test-placeholder", MOONSHOT_API_KEY: "test-placeholder", OPENCODE_ALLOW_ALL_MODELS: "1", CLAUDE_CONFIG_DIR: dir } }, context: {}, onLog: async () => {}, }); expect(result.timedOut).toBe(true); diff --git a/server/src/__tests__/connection-aggregator-fallback.test.ts b/server/src/__tests__/connection-aggregator-fallback.test.ts index 40cee41710b..c129a6e092f 100644 --- a/server/src/__tests__/connection-aggregator-fallback.test.ts +++ b/server/src/__tests__/connection-aggregator-fallback.test.ts @@ -317,9 +317,34 @@ const support = await getEmbeddedPostgresTestSupport(); expect(result.results.some(item => item.service === "google-sheets")).toBe(true); expect(result.instruction).not.toContain("search its exact name"); }); + it("keeps installed capability matches when a generic query resembles an external app name", async () => { + await resetQuestions(); + const installed = await seedProvider("Studio library", "notion:list_pages"); + await db.update(toolCatalogEntries).set({ description: "Read recent pages and return their titles and verification code" }) + .where(eq(toolCatalogEntries.connectionId, installed.id)); + const service = connectionIntentService(db); + const result = await service.search(claims, + "connected page service that can find or list recent pages and return page titles plus a verification code"); + expect(result.results).toEqual(expect.arrayContaining([expect.objectContaining({ + service: `connection:${installed.id}`, connectionId: installed.id, state: "ready", + })])); + expect(result.providerQuestion).toBeUndefined(); + expect(result.instruction).not.toContain("Ask the responsible user with providerQuestion"); + // A deliberate app name still reaches the existing governed provider choice. + const named = await service.search(claims, "Page X"); + expect(named.providerQuestion?.id).toBe("connection-provider:page-x"); + expect(named.results[0]?.service).toBe("via:composio:page-x"); + }); + it("keeps an external typo as a discovery suggestion until its app is selected", async () => { + await resetQuestions(); + const result = await connectionIntentService(db).search(claims, + "Find Circlebak meeting transcripts and action items from yesterday"); + expect(result.results.some(item => item.service === "via:composio:circleback-mcp")).toBe(true); + expect(result.providerQuestion).toBeUndefined(); + expect(result.instruction).toContain("aggregator.targetService"); + }); it.each([ ["help me find tools for circle back", "circleback-mcp"], - ["Find Circlebak meeting transcripts and action items from yesterday", "circleback-mcp"], ["Can you connect Circleback MCP to get all our meeting notes?", "circleback-mcp"], ["Find Attio tools to review all our customer contacts before next week's meeting", "attio"], ["Help me find a ClickUp connection to organize our team's projects", "clickup"], @@ -583,6 +608,51 @@ const support = await getEmbeddedPostgresTestSupport(); ); expect(result.instruction).toContain("Arcade"); }); + it("returns exact eligible catalog names after a guessed tool without granting access", async () => { + await resetQuestions(); + const connection = await seedProvider("arcade", "Hubspot_ListContacts", "responsible-user", false); + await db.delete(toolConnectionInstalls).where(eq(toolConnectionInstalls.connectionId, connection.id)); + await seedProvider("composio", "Unrelated_PrivateTool", "other-user", false); + await db.insert(toolCatalogEntries).values({ companyId: claims.company_id, connectionId: connection.id, + toolName: "Hubspot_RemovedTool", name: "Removed", versionHash: "old", entryKind: "tool", status: "removed" }); + const answer = await selectProvider("via:arcade:hubspot"); + const service = connectionIntentService(db); + const error = await service.request(claims, "via:arcade:hubspot", { + selectionInteractionId: answer.id, toolNames: ["hubspot_list_contacts"], + }).catch(error => error); + expect(error).toMatchObject({ status: 422 }); + expect(error.message).toContain('"Hubspot_ListContacts"'); + expect(error.message).toContain("Request only the needed exact names"); + expect(error.message).not.toMatch(/Unrelated_PrivateTool|Hubspot_RemovedTool/); + expect(await db.select().from(toolProfiles).where(eq(toolProfiles.companyId, claims.company_id))).toEqual([]); + const interactions = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.companyId, claims.company_id)); + expect(interactions.map(row => [row.id, row.status])).toEqual([[answer.id, "answered"]]); + const requested = await service.request(claims, "via:arcade:hubspot", { + selectionInteractionId: answer.id, toolNames: ["Hubspot_ListContacts"], + }); + expect(requested.state).toBe("needs_user_action"); + const loaded = await service.loadIntent(requested.interactionId!); + expect(loaded.interaction.payload.accessRequest?.tools.map(tool => tool.toolName)).toEqual(["Hubspot_ListContacts"]); + expect(loaded.interaction.status).toBe("pending"); + expect(await db.select().from(toolProfiles).where(eq(toolProfiles.companyId, claims.company_id))).toEqual([]); + }); + it("bounds catalog-name recovery and does not disclose an ineligible connection", async () => { + await resetQuestions(); + const connection = await seedProvider("arcade", "Read_00", "responsible-user", false); + await db.insert(toolCatalogEntries).values(Array.from({ length: 24 }, (_, index) => ({ + companyId: claims.company_id, connectionId: connection.id, toolName: `Read_${String(index + 1).padStart(2, "0")}`, + name: `Read_${String(index + 1).padStart(2, "0")}`, versionHash: "v1", entryKind: "tool" as const, status: "active" as const, + }))); + const service = connectionIntentService(db); + const error = await service.request(claims, "arcade", { connectionId: connection.id, toolNames: ["guessed"] }).catch(error => error); + expect(error).toMatchObject({ status: 422 }); + expect(error.message).toContain("first 20 of 25"); + expect(error.message.match(/Read_\d{2}/g)).toHaveLength(20); + await db.update(connectionGrants).set({ status: "revoked" }).where(eq(connectionGrants.connectionId, connection.id)); + const denied = await service.request(claims, "arcade", { connectionId: connection.id, toolNames: ["guessed"] }).catch(error => error); + expect(denied.message).not.toContain("Read_"); + expect(denied.message).toContain("not eligible"); + }); it("does not switch providers when the chosen route loses permission", async () => { await resetQuestions(); const answer = await selectProvider("via:arcade:hubspot"); diff --git a/server/src/services/anthropic-pricing.test.ts b/server/src/services/anthropic-pricing.test.ts new file mode 100644 index 00000000000..760b18d2eef --- /dev/null +++ b/server/src/services/anthropic-pricing.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, it } from "vitest"; +import type { AdapterUsageCheckpoint } from "@paperclipai/adapter-utils"; +import { priceAnthropicReceipt } from "./anthropic-pricing.js"; + +const receipt: AdapterUsageCheckpoint = { complete: true, usageBasis: "per_run", provider: "anthropic", biller: "anthropic", + billingType: "metered_api", model: "claude-sonnet-5", costUsd: null, costStatus: "unpriced", + usage: { inputTokens: 22462, cachedInputTokens: 208218, cacheWriteTokens: 22450, outputTokens: 2450 } }; +describe("complete Anthropic receipt pricing", () => { + it("prices complete counters conservatively and labels their provenance as an estimate", () => { + const priced = priceAnthropicReceipt(receipt); + expect(priced.costUsdExact).toBe("0.155967600"); + expect(priced.costStatus).toBe("estimated"); + expect(priced.pricingProvenance).toMatchObject({ source: "rate_card", version: "anthropic-standard-2026-10-07" }); + expect(priced.pricingProvenance?.evidence).toContain("one-hour upper rate"); + expect(priced.usage).toBe(receipt.usage); + expect(receipt.costStatus).toBe("unpriced"); + }); + it.each([ + { complete: false }, { billingType: "unknown" }, { billingType: "subscription_included" }, + { provider: "unknown" }, { biller: "aws_bedrock" }, { model: "claude-sonnet-latest" }, + { usageBasis: "session_cumulative" }, { costUsd: 0.12 }, { costUsdExact: "0.12" }, { cacheAdjustedCostUsd: 0.12 }, + { pricingContext: { serviceTier: "priority" } }, { usageByModel: [{ model: "other", usage: { inputTokens: 1, outputTokens: 1 }, costUsd: 1 }] }, + { usage: undefined }, { usage: { inputTokens: 1, outputTokens: 1 } }, + { usage: { inputTokens: 1, outputTokens: 1, cacheWriteTokens: 2, cachedInputTokens: 0 } }, + ] as Partial[])("preserves incomplete, incompatible or already priced receipts: %j", override => { + const original = { ...receipt, ...override }; + expect(priceAnthropicReceipt(original)).toBe(original); + }); +}); diff --git a/server/src/services/anthropic-pricing.ts b/server/src/services/anthropic-pricing.ts new file mode 100644 index 00000000000..6a9d3cfdae3 --- /dev/null +++ b/server/src/services/anthropic-pricing.ts @@ -0,0 +1,31 @@ +import { centsToUsd, unitsToCents, usdToUnits } from "@paperclipai/shared"; +import type { AdapterUsageCheckpoint } from "@paperclipai/adapter-utils"; + +// Direct Anthropic API list prices, verified 2026-10-07: +// https://platform.claude.com/docs/en/about-claude/pricing +// Keep this bounded to the qualified model; aliases and other billers stay unknown. +const RATES = ["2", "0.2", "4", "10"] as const; + +/** A complete token receipt can receive an explicit estimate, never an invoice + * claim. ACP aggregates cache writes without TTL; conservatively price them at + * the documented one-hour rate rather than silently assume five-minute writes. */ +export function priceAnthropicReceipt(receipt: AdapterUsageCheckpoint): AdapterUsageCheckpoint { + if (!receipt.complete || !receipt.usage || receipt.usageBasis !== "per_run" + || receipt.provider !== "anthropic" || receipt.biller !== "anthropic" + || !["api", "metered_api"].includes(receipt.billingType ?? "") + || receipt.model !== "claude-sonnet-5" || receipt.usageByModel?.length + || receipt.costUsd != null || receipt.costUsdExact != null || receipt.cacheAdjustedCostUsd != null + || (receipt.pricingContext?.serviceTier && !["standard", "default"].includes(receipt.pricingContext.serviceTier))) return receipt; + const usage = receipt.usage; + if (usage.cacheWriteTokens === undefined || usage.cachedInputTokens === undefined) return receipt; + const counts = [usage.inputTokens - usage.cacheWriteTokens, usage.cachedInputTokens, usage.cacheWriteTokens, usage.outputTokens]; + if (counts.some(count => !Number.isSafeInteger(count) || count < 0)) return receipt; + const numerator = counts.reduce((total, count, index) => total + BigInt(count) * usdToUnits(RATES[index]), 0n); + return { ...receipt, costUsdExact: centsToUsd(unitsToCents((numerator + 500_000n) / 1_000_000n)), costStatus: "estimated", + pricingProvenance: { + source: "rate_card", version: "anthropic-standard-2026-10-07", + evidence: "https://platform.claude.com/docs/en/about-claude/pricing; standard global API pricing assumed; cache-write TTL unavailable, one-hour upper rate used; not an invoice", + inputCentsPerMillion: "200", cachedInputCentsPerMillion: "20", cacheWriteCentsPerMillion: "400", outputCentsPerMillion: "1000", serviceTier: "standard", + }, + }; +} diff --git a/server/src/services/connection-intents.ts b/server/src/services/connection-intents.ts index c53a42b76f5..68005f19f4c 100644 --- a/server/src/services/connection-intents.ts +++ b/server/src/services/connection-intents.ts @@ -466,7 +466,8 @@ export function connectionIntentService(db: Db) { } const publicMatches = searchAggregatorServices(preparedQuery); const bestMatches = publicMatches.filter(match => Math.floor(match.nameScore / 100) === Math.floor((publicMatches[0]?.nameScore ?? 0) / 100)); - const publicService = findAggregatorService(serviceQuery, publicMatches) ?? (bestMatches.length === 1 ? bestMatches[0]!.service : undefined); + const namedPublicService = findAggregatorService(serviceQuery, publicMatches); + const publicService = namedPublicService ?? (bestMatches.length === 1 ? bestMatches[0]!.service : undefined); // Extract service namespaces only from the current identity's indexed tools. // Generic provider search/execute descriptions do not prove app support. const indexedServices = new Set(); @@ -553,6 +554,14 @@ export function connectionIntentService(db: Db) { return alternatives; } const alternatives = await aggregatorAlternatives(targetService, targetName, publicService); + // A typo/prefix match is a discovery hint, not a selected app. Generic + // capabilities (e.g. "pages") can resemble a catalog name ("Page X"). + // Retain authorized installed matches instead of prescribing provider consent. + const namedTarget = namedPublicService !== undefined + || indexedMatches.some(match => match.slug === targetService && match.nameScore >= 500); + if (alternatives.length && !namedTarget) { + return discoverySuggestions([...ranked(), ...alternatives]); + } if (explicit && explicitConsent) { const selected = alternatives.find(item => item.aggregator?.provider === explicit.provider); if (!selected) return { version: 1, query, results: [], instruction: "The explicitly requested external provider is unavailable or its app support could not be verified. Explain the limitation. Do not switch providers automatically." }; @@ -586,7 +595,7 @@ export function connectionIntentService(db: Db) { function discoverySuggestions(results: ConnectionSearchResultItem[]): ConnectionsSearchResult { return { version: 1, query, results: results.slice(0, 40), - instruction: "Multiple apps match this query. Choose the relevant service and method using descriptions and purposes. For available or needs_user_action results, tool methods and AgentMail use connection_request to show the inline setup card; other channel or AI methods use setupPath. Use ready tools as installed; ready AI authentication applies to the next execution without reconnection. For an aggregator result, search its aggregator.targetService to obtain the provider-choice question and follow that instruction before requesting a connection. Respect unavailable states. Do not treat a search match as provider consent or app authorization." }; + instruction: "These are possible connection matches. Choose the relevant service and method using descriptions and purposes. For available or needs_user_action results, tool methods and AgentMail use connection_request to show the inline setup card; other channel or AI methods use setupPath. Use ready tools as installed; ready AI authentication applies to the next execution without reconnection. For an aggregator result, search its aggregator.targetService to obtain the provider-choice question and follow that instruction before requesting a connection. Respect unavailable states. Do not treat a search match as provider consent or app authorization." }; } } @@ -730,7 +739,12 @@ export function connectionIntentService(db: Db) { const catalog = (await indexedCatalog(connection.id, context.run.companyId)).filter(tool => tool.entryKind === "tool"); const names = options.toolNames ?? (app.slug === "composio" ? ["COMPOSIO_SEARCH_TOOLS", "COMPOSIO_MANAGE_CONNECTIONS"] : catalog.map(tool => tool.toolName)); const tools = names.map(name => catalog.find(tool => tool.toolName === name)); - if (options.toolNames && tools.some(tool => !tool)) throw unprocessable("A requested tool is not in this connection's active catalog"); + if (options.toolNames && tools.some(tool => !tool)) { + // Return discovery metadata only after validating this connection's + // eligibility. Never substitute guessed names or grant access here. + const availableNames = catalog.map(tool => tool.toolName).sort(); + throw unprocessable(`A requested tool is not in this connection's active catalog. Available indexed tool names (first ${Math.min(20, availableNames.length)} of ${availableNames.length}): ${JSON.stringify(availableNames.slice(0, 20))}. Request only the needed exact names with connection_request. No access was granted.`); + } const effective = await access.getEffectiveProfilesForAgent(context.run.companyId, context.agent.id); const installed = effective.installedConnections.some(item => item.id === connection.id); const missing = !installed || tools.some(tool => tool && !effective.allowedTools.some(allowed => allowed.id === tool.id)); diff --git a/server/src/services/native-runtime/native-completion-feedback.test.ts b/server/src/services/native-runtime/native-completion-feedback.test.ts index e52702dd173..d49a5cc1339 100644 --- a/server/src/services/native-runtime/native-completion-feedback.test.ts +++ b/server/src/services/native-runtime/native-completion-feedback.test.ts @@ -54,6 +54,47 @@ describe("native final-response feedback", () => { await db.update(issueThreadInteractions).set({ sourceRunId: null }).where(eq(issueThreadInteractions.id, interactionId)); await expect(nativeCompletionFeedback(db, value.runId, waiting)).resolves.toContain("Completion report accepted"); }); + it.each(["blocked", "needs_review", "yielded"] as const)("rejects a %s approval report bound to an already declined invocation", async (disposition) => { + const value = await fixture(), interactionId = randomUUID(), invocationId = randomUUID(), actionRequestId = randomUUID(); + await db.insert(issueThreadInteractions).values({ id: interactionId, companyId: value.companyId, issueId: value.issueId, + sourceRunId: value.runId, createdByAgentId: value.agentId, kind: "request_confirmation", status: "rejected", + continuationPolicy: "wake_assignee", result: { version: 1, outcome: "rejected" }, + payload: { version: 1, prompt: "Approve service read", toolAction: { version: 1, actionRequestId, invocationId, toolName: "pages.read", toolDisplayName: "Read pages", connectionId: null, applicationId: null, appDisplayName: null, risk: "read", previewMarkdown: "Read pages", argumentsSummaryJson: "{}", argumentsHash: "test-hash", expiresAt: "2026-10-07T12:00:00Z" } }, + }); + // The provider can name its exact invocation in the summary without citing + // an interaction evidence ref, as in the retained OpenCode failure. + const report: PrpStructuredRunResult = { ...done, reportedWorkDisposition: disposition, + summary: disposition === "blocked" ? "Waiting on the page-service approval." : `Waiting on the page-service approval (invocationId ${invocationId}).`, + ...(disposition === "blocked" ? { blocker: { reasonCode: "approval_required", owner: { kind: "user" as const, name: "Release Owner" }, scope: "task_wide" as const, unblockAction: `Resolve invocation ${invocationId}` } } : {}), + completionClaim: { ...done.completionClaim, objectiveSatisfied: false, remainingWork: [{ description: "Wait for the decision", blocksCompletion: true }] }, + attentionRequests: [{ kind: "approval", ownerClass: "human", summary: "Approve the page-service call" }], + ...(disposition === "yielded" ? { continuation: { kind: "response_wake" as const, idempotencyKey: "wait", summary: "Wait for approval" } } : {}), + }; + await expect(nativeCompletionFeedback(db, value.runId, report)).rejects.toThrow("already resolved"); + // A blocked report may name the exact action only in an unmet criterion. + const criterionOnly: PrpStructuredRunResult = { ...report, blocker: undefined, + summary: "Await the requested approval", evidence: [{ ref: `approval:${actionRequestId}` }], + completionClaim: { ...report.completionClaim, criteria: [{ criterionId: "objective", status: "not_satisfied", evidenceRefs: [`approval:${actionRequestId}`] }] }, + }; + await expect(nativeCompletionFeedback(db, value.runId, criterionOnly)).rejects.toThrow("already resolved"); + // A provider can mark the approval criterion unknown while citing the exact + // declined interaction; that still is not a new review target. + await expect(nativeCompletionFeedback(db, value.runId, { + ...criterionOnly, completionClaim: { ...criterionOnly.completionClaim, criteria: [{ criterionId: "objective", status: "unknown", evidenceRefs: [`interaction:${interactionId}`] }] }, + })).rejects.toThrow("already resolved"); + // Satisfied historical evidence does not bind the independent review target. + if (disposition !== "yielded") await expect(nativeCompletionFeedback(db, value.runId, { + ...criterionOnly, completionClaim: { ...criterionOnly.completionClaim, criteria: [{ criterionId: "past-action", status: "satisfied", evidenceRefs: [`approval:${actionRequestId}`] }] }, + })).resolves.toContain("report accepted"); + // Completed-action evidence is not the target of an independent review. + if (disposition !== "yielded") await expect(nativeCompletionFeedback(db, value.runId, { + ...report, blocker: undefined, summary: "Review a separate deliverable", evidence: [{ ref: `interaction:${interactionId}` }], + })).resolves.toContain("report accepted"); + // Neither a similar description nor another run's card is identity proof. + await expect(nativeCompletionFeedback(db, value.runId, { ...report, blocker: undefined, summary: "Wait for a separate approval" })).resolves.toContain("report accepted"); + await db.update(issueThreadInteractions).set({ sourceRunId: null }).where(eq(issueThreadInteractions.id, interactionId)); + await expect(nativeCompletionFeedback(db, value.runId, report)).resolves.toContain("report accepted"); + }); it.each(["accepted", "rejected"] as const)("allows a new question after a %s tool approval cited as completed evidence", async (status) => { const value = await fixture(), interactionId = randomUUID(), questionId = randomUUID(); await db.insert(issueThreadInteractions).values({ id: interactionId, companyId: value.companyId, issueId: value.issueId, diff --git a/server/src/services/native-runtime/native-completion-feedback.ts b/server/src/services/native-runtime/native-completion-feedback.ts index d814ca3a232..9bbb9d39d26 100644 --- a/server/src/services/native-runtime/native-completion-feedback.ts +++ b/server/src/services/native-runtime/native-completion-feedback.ts @@ -176,21 +176,31 @@ export async function nativeCompletionFeedback( .limit(1) .then((rows) => rows[0]), ]); - // A response wake for this run's tool action already has a durable approval - // surface. Reject a repeated approval request before it becomes a new review. - // General evidence can also cite completed actions while waiting on something - // else, so a resolved card alone is not a stale wait target. - if (result.reportedWorkDisposition === "yielded" && result.continuation?.kind === "response_wake") { - const referencedIds = result.evidence.flatMap(({ ref }) => { - const match = typeof ref === "string" ? /^interaction:([a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12})$/i.exec(ref) : null; - return match ? [match[1]!] : []; - }); - const referenced = referencedIds.length ? await db.select().from(issueThreadInteractions).where(and( + // Match an existing action by its exact durable identity, regardless of how + // the provider labels its final disposition. A summary can cite the invocation + // ID instead of an interaction evidence ref; prose similarity is not identity. + if (signals.actionableAttentionRequests.some(request => request.kind === "approval" && request.ownerClass === "human")) { + const approvalTarget = { + summary: result.summary, + attentionRequests: signals.actionableAttentionRequests.filter(request => request.kind === "approval" && request.ownerClass === "human"), + blocker: result.blocker, + // An unmet criterion can name the action still being requested. Completed + // evidence elsewhere in the report is not authority to suppress a review. + unmetCriteria: result.completionClaim.criteria.filter(criterion => criterion.status !== "satisfied"), + continuation: result.continuation, + // The existing response-wake contract also names its card through evidence. + evidence: result.reportedWorkDisposition === "yielded" && result.continuation?.kind === "response_wake" ? result.evidence : [], + }; + const referencedIds = new Set(JSON.stringify(approvalTarget).match(/\b[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}\b/gi)?.map(id => id.toLowerCase()) ?? []); + const cards = referencedIds.size ? await db.select().from(issueThreadInteractions).where(and( eq(issueThreadInteractions.companyId, run.companyId), eq(issueThreadInteractions.issueId, issue.id), eq(issueThreadInteractions.sourceRunId, run.id), eq(issueThreadInteractions.createdByAgentId, run.agentId), eq(issueThreadInteractions.kind, "request_confirmation"), eq(issueThreadInteractions.continuationPolicy, "wake_assignee"), - inArray(issueThreadInteractions.id, referencedIds), )) : []; + const referenced = cards.filter(card => { + const action = record(record(card.payload).toolAction); + return [card.id, action.actionRequestId, action.invocationId].some(id => typeof id === "string" && referencedIds.has(id.toLowerCase())); + }); for (const card of referenced) { const action = record(record(card.payload).toolAction); if (action.version !== 1 || typeof action.actionRequestId !== "string") continue; diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 3fbe173df1e..f165e004b43 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -5141,6 +5141,62 @@ describe("native startup cancellation fence", () => { }); describe("native startup restart detachment", () => { + it("waits for a checkpointed governed settlement before relinquishing its controller", async () => { + const restarting = structuredClone(execution); + restarting.binding.runId = "restart-governed-settlement"; + let release!: () => void, announce!: () => void; + const accounting = new Promise(resolve => { release = resolve; }); + const checkpointed = new Promise(resolve => { announce = resolve; }); + const detach = vi.fn(async () => undefined); + state.execute.mockReset().mockImplementationOnce(async options => { + await options.onSession({ detachControllerForRestart: detach }); + await options.onCheckpoint({ governedWait: { sourceEvent: { turnId: "settling-turn" } } }); + announce(); + await accounting; + await options.onSession(null); + throw new Error("settlement test complete"); + }); + const outcome = executePaperclipNativeSession({ db: leaseDb(restarting), execution: restarting, runnerInstanceId: "runner" }).catch(error => error); + await checkpointed; + const detached = detachNativeSessionsForRestart([restarting.binding.runId]); + try { + await new Promise(resolve => setImmediate(resolve)); + expect(detach).not.toHaveBeenCalled(); + release(); + await outcome; + expect(await detached).toMatchObject({ inactiveRunIds: [restarting.binding.runId] }); + expect(detach).not.toHaveBeenCalled(); + } finally { release(); await outcome; await detached; } + }); + it("bounds governed settlement without allowing the detached controller to publish success", async () => { + const restarting = structuredClone(execution); + restarting.binding.runId = "restart-governed-settlement-timeout"; + let release!: () => void, announce!: () => void; + const accounting = new Promise(resolve => { release = resolve; }); + const checkpointed = new Promise(resolve => { announce = resolve; }); + const detach = vi.fn(async () => undefined); + const onUsage = vi.fn(async () => undefined); + state.execute.mockReset().mockImplementationOnce(async options => { + await options.onSession({ detachControllerForRestart: detach }); + await options.onCheckpoint({ governedWait: { sourceEvent: { turnId: "settling-turn" } } }); + announce(); await accounting; + await options.onSession(null); + return { result: {}, terminal: { runTerminalState: "succeeded" }, usage: null }; + }); + const outcome = executePaperclipNativeSession({ db: leaseDb(restarting), execution: restarting, runnerInstanceId: "runner", onUsage }).catch(error => error); + await checkpointed; + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); + const detaching = detachNativeSessionsForRestart([restarting.binding.runId]); + try { + await vi.advanceTimersByTimeAsync(19_999); + expect(detach).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1); + expect(await detaching).toMatchObject({ detachedRunIds: [restarting.binding.runId] }); + release(); + expect(await outcome).toBeInstanceOf(NativeControllerDetachedForRestartError); + expect(onUsage).not.toHaveBeenCalled(); + } finally { vi.useRealTimers(); release(); await outcome; await detaching; } + }); it("waits for in-flight runner startup and its detach acknowledgement before shutdown returns", async () => { const root = await mkdtemp(join(tmpdir(), "native-startup-detach-")); const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR; @@ -5186,6 +5242,24 @@ describe("native startup restart detachment", () => { } }); + it("rejects a governed completion returned successfully by an already detached controller", async () => { + const restarting = structuredClone(execution); + restarting.binding.runId = "restart-successful-old-consumer"; + const detach = vi.fn(async () => undefined); + const onUsage = vi.fn(async () => undefined); + state.execute.mockReset().mockImplementationOnce(async options => { + await options.onSession({ detachControllerForRestart: detach }); + await detachNativeSessionsForRestart([restarting.binding.runId]); + await options.onSession(null); + return { result: {}, + terminal: { runTerminalState: "succeeded" }, usage: null }; + }); + await expect(executePaperclipNativeSession({ db: leaseDb(restarting), execution: restarting, + runnerInstanceId: "runner", onUsage })).rejects.toBeInstanceOf(NativeControllerDetachedForRestartError); + expect(detach).toHaveBeenCalledOnce(); + expect(onUsage).not.toHaveBeenCalled(); + }); + it.each([false, true])("settles failed startup without claiming detachment (deadline exceeded: %s)", async (exceedDeadline) => { const root = await mkdtemp(join(tmpdir(), "native-startup-failure-")); const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR; @@ -5240,6 +5314,23 @@ describe("native startup restart detachment", () => { }); describe("native terminal-turn accounting", () => { + it.each([false, true])("prices only complete direct Claude API turn accounting (%s)", async complete => { + const { rehydrateRunnerdUsageNotification } = await import("@paperclipai/paperclip-runner/live"); + const { priceAnthropicReceipt } = await import("../anthropic-pricing.js"); + const onUsage = vi.fn(async (_receipt: import("@paperclipai/adapter-utils").AdapterUsageCheckpoint) => {}); + const counts = { inputTokens: 12, outputTokens: 4, cacheReadTokens: 30, cacheWriteTokens: 20, providerCostUsd: 0 }; + state.execute.mockReset().mockImplementationOnce(async () => { + const notification = rehydrateRunnerdUsageNotification({ provider: "acpx", cumulative: { ...counts, providerCostUsd: 10 }, runDelta: counts, runDeltaAvailable: complete }, "session", "turn"); + await accountingEvents.committed!({ eventType: "item.completed", turnId: "turn", payload: { kind: "usage", usage: notification.tokenUsage } } as unknown as PrpEvent); + await accountingEvents.committed!({ eventType: "turn.interrupted", turnId: "turn", payload: {} } as unknown as PrpEvent); + return { result: { summary: "Waiting for approval" }, terminal: { runTerminalState: "succeeded" }, turnId: "turn", normalizedSessionId: "session", providerSessionId: null, driverKind: "acpx_runtime", driverVersion: "1", nativeEventCount: 2, highestContiguousSourceSeq: 2, usage: null }; + }); + await executePaperclipNativeSession({ db: leaseDb(), execution: { ...execution, provider: { kind: "acpx", agent: "claude", model: "claude-sonnet-5" } } as NativeExecutionInput, runnerInstanceId: "runner", runnerEnvironment: { ANTHROPIC_API_KEY: "fixture" }, onUsage }); + const receipt = onUsage.mock.calls.at(-1)![0]; + const priced = priceAnthropicReceipt(receipt); + if (complete) expect(priced).toMatchObject({ complete: true, costStatus: "estimated", costUsdExact: "0.000150000" }); + else expect(priced).toMatchObject({ complete: false, costStatus: "unpriced", costUsd: null }); + }); it.each(["claude", "codex"].flatMap(agent => [false, true].flatMap(restart => [false, true].flatMap(partialFirst => [false, true].map(zeroFirst => ({ agent, restart, partialFirst, zeroFirst }))), )))("accumulates ACPX turns for $agent (restart: $restart, partial: $partialFirst, zero first: $zeroFirst)", async ({ agent, restart, partialFirst, zeroFirst }) => { diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index fec05347384..997845791cc 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -242,6 +242,9 @@ type NativeSessionStartup = { resolve: (session: ActiveNativeSession | null) => void; stopRequested?: boolean; cancellationSettled?: Promise; + governedWait?: boolean; + settled: Promise; + settle: () => void; }; const nativeSessionStartups = new Map(); @@ -279,7 +282,20 @@ export async function detachNativeSessionsForRestart( const detachedRunIds: string[] = []; const inactiveRunIds: string[] = []; const unsupportedRunIds: string[] = []; + const settlementDeadline = Date.now() + 20_000; for (const runId of new Set(runIds)) { + // A governed stop already revoked new work and is collecting terminal + // accounting. Give that exact owner a bounded chance to persist it before + // fencing callbacks and relinquishing the runner. Timeout is not success. + const settling = nativeSessionStartups.get(runId); + if (settling?.governedWait) { + let timer: ReturnType | undefined; + try { + await Promise.race([settling.settled, new Promise(resolve => { + timer = setTimeout(resolve, Math.max(0, settlementDeadline - Date.now())); + })]); + } finally { clearTimeout(timer); } + } nativeRunsDetachingForRestart.add(runId); const active = activeNativeSessions.get(runId) ?? await waitForNativeSessionStartup(runId); if (!active) { @@ -6002,6 +6018,7 @@ function loadWarmNativeCheckpoint( activeTurnId: null, terminalTurns: [], pendingRuntimeRequests: [], + governedWait: undefined, }; if (path !== scopedPath || envelope.configDigest !== configDigest) { // Upgrade the validated checkpoint atomically. When moving from a legacy @@ -7381,7 +7398,10 @@ export async function executePaperclipNativeSession(input: { // Register before the first asynchronous operation on either backend path. // A duplicate execution must not replace the original startup handoff. let resolveStartup!: (session: ActiveNativeSession | null) => void; + let settle!: () => void; const startup: NativeSessionStartup = { + settled: new Promise(resolve => { settle = resolve; }), + settle: () => settle(), promise: new Promise(resolve => { resolveStartup = resolve; }), resolve: session => resolveStartup(session), }; @@ -7478,6 +7498,7 @@ export async function executePaperclipNativeSession(input: { executingNativeOwnerScopes.delete(ownerScope); } startup.resolve(null); + startup.settle(); if (nativeSessionStartups.get(runId) === startup) { nativeSessionStartups.delete(runId); } @@ -7993,6 +8014,11 @@ async function executePaperclipNativeSessionWithinScope( } }; let completedConversationReply: PrpEvent | null = null; + const assertControllerActive = () => { + if (nativeRunsDetachingForRestart.has(input.execution.binding.runId)) { + throw new NativeControllerDetachedForRestartError(); + } + }; const controlPlane = new PaperclipControlPlanePort( input.db, { @@ -8008,6 +8034,7 @@ async function executePaperclipNativeSessionWithinScope( }, { privateKeyPem: input.runnerEnvironment?.PAPERCLIP_AGENT_PRIVATE_KEY, + assertControllerActive, onCommittedEvent: async (event) => { await observeAccountingEvent(event); await toolTrace.observe(event); @@ -8541,8 +8568,15 @@ async function executePaperclipNativeSessionWithinScope( controlPlane, runnerInstanceId: effectiveRunnerInstanceId, controlPlaneInstanceId, - resolveGovernedWait: ({ event }) => - governedWaitObservation.consume(event), + resolveGovernedWait: ({ event }) => { + assertControllerActive(); + const result = governedWaitObservation.consume(event); + if (result) { + const startup = nativeSessionStartups.get(input.execution.binding.runId); + if (startup) startup.governedWait = true; + } + return result; + }, resolveMissingResult: async ({ terminalEvent }) => { // Governed waits take precedence over an ordinary chat reply. // Execution tasks still require their normal semantic finish. @@ -8581,6 +8615,10 @@ async function executePaperclipNativeSessionWithinScope( requireSessionCloseBeforeReturn: runnerdBackend !== null || input.instructionWorkingCopy !== undefined, onSessionClosed: input.instructionWorkingCopy?.collectStopped, onCheckpoint: async (snapshot) => { + if (snapshot.governedWait) { + const startup = nativeSessionStartups.get(input.execution.binding.runId); + if (startup) startup.governedWait = true; + } snapshot = identityRedactor.redact(snapshot); if (warmSessionId !== null && warmConfigDigest !== null) { await persistWarmNativeCheckpoint( @@ -8774,6 +8812,9 @@ async function executePaperclipNativeSessionWithinScope( }, { parentName: "task.run" }, ); + // A detached consumer can resolve successfully after its stream closes. + // Only the replacement controller may settle the run or certify accounting. + assertControllerActive(); // Persist provider accounting before any workspace/issue finalization. A // detached controller or failed finalizer must not lose a completed turn. // session.usage() may be an attachment baseline, a partial report, or a diff --git a/server/src/services/native-runtime/native-session-resume.test.ts b/server/src/services/native-runtime/native-session-resume.test.ts index 60290408caf..574e0dda15e 100644 --- a/server/src/services/native-runtime/native-session-resume.test.ts +++ b/server/src/services/native-runtime/native-session-resume.test.ts @@ -2125,10 +2125,17 @@ describe("rebindNativeSessionCheckpoint", () => { }); it("retains provider identity but clears prior turn and event state", () => { + const prior = previousRun(); + const profile = prior.runnerProfileJson as Record; + const checkpoint = profile.sessionCheckpoint as Record; + checkpoint.governedWait = { sourceEvent: { runId: previousRunId, turnId: "old-turn" }, result: { reportedWorkDisposition: "yielded" } }; + const before = structuredClone(prior); const rebound = rebindNativeSessionCheckpoint({ - previousRun: previousRun(), + previousRun: prior, currentExecution: execution(currentRunId), }); + expect(rebound?.governedWait).toBeUndefined(); + expect(prior).toEqual(before); expect(rebound).toMatchObject({ sessionId: "provider-thread-123", providerSessionId: "provider-thread-123", diff --git a/server/src/services/native-runtime/native-session-resume.ts b/server/src/services/native-runtime/native-session-resume.ts index c7090ea8639..051fbc0cf22 100644 --- a/server/src/services/native-runtime/native-session-resume.ts +++ b/server/src/services/native-runtime/native-session-resume.ts @@ -568,6 +568,9 @@ export function rebindNativeSessionCheckpoint(input: { activeTurnId: null, terminalTurns: [], pendingRuntimeRequests: [], + // Settlement evidence belongs to the prior heartbeat run, even when the + // provider session is reused for the user's response. + governedWait: undefined, providerRecoveryPolicy, }; } diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts index f6651aedc88..129ffdcf23a 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts @@ -299,6 +299,67 @@ describe("PaperclipControlPlanePort conformance", () => { } }); + it.each([ + ["checkpoint", "detach"], ["result", "detach"], ["event", "detach"], + ["checkpoint", "abort"], ["result", "abort"], ["event", "abort"], + ] as const)("revokes an in-flight %s on %s", async (operation, reason) => { + const identity = { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, runId: randomUUID(), sessionId: randomUUID() }; + const runnerId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: identity.runId, companyId: identity.companyId, agentId: identity.agentId, + status: "running", runtimeMode: "native", nativeIssueId: identity.issueId, + nativeSessionId: identity.sessionId, runnerInstanceId: runnerId, + completionContractId: contractId, completionContractSha256: contractSha, + contextSnapshot: { issueId: identity.issueId }, + }); + const binding = { ...identity, completionContractId: contractId, completionContractSha256: contractSha, + sourceInstanceId: runnerId, controlPlaneSourceInstanceId: `detached-${identity.runId}` }; + let detached = false; + const controller = new AbortController(); + const port = new PaperclipControlPlanePort(db, binding, { + assertControllerActive: () => { if (detached) throw new Error("controller_detached"); }, + }); + await port.openRun({ identity, backendKind: "mock", sourceInstanceId: runnerId }); + const mutate = () => operation === "event" ? port.appendEvent({ + schema: "paperclip.prp.event.v1", sourceEventId: `${runnerId}:1`, sourceSeq: 1, + sourceInstanceId: runnerId, sourceKind: "runner", runId: identity.runId, + normalizedSessionId: identity.sessionId, eventType: "turn.interrupted", schemaVersion: 1, + priority: 0, emittedAt: new Date().toISOString(), payload: { reason: "governed_wait" }, + }, { signal: controller.signal }) : operation === "checkpoint" + ? port.checkpointSession({ backendKind: "mock", sessionId: identity.sessionId, identity }, { signal: controller.signal }) + : port.completeRun({ result: CONTROL_PLANE_CONFORMANCE_RESULT, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL }, { signal: controller.signal }); + let pending: Promise | undefined; + await db.transaction(async tx => { + await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, identity.runId)).for("update"); + const [backend] = await tx.execute(sql`select pg_backend_pid() as pid`) as unknown as Array<{ pid: number }>; + pending = mutate().then(() => null, error => error); + let waiting = false; + for (let attempt = 0; attempt < 100; attempt++) { + const [state] = await db.execute(sql`select exists ( + select 1 from pg_stat_activity where ${backend.pid} = any(pg_blocking_pids(pid)) + ) as waiting`) as unknown as Array<{ waiting: boolean }>; + if (state.waiting) { waiting = true; break; } + await new Promise(resolve => setTimeout(resolve, 10)); + } + expect(waiting).toBe(true); + if (reason === "detach") detached = true; + else controller.abort(new Error("controller_detached")); + }); + expect(await pending).toMatchObject({ message: "controller_detached" }); + await expect(mutate()).rejects.toThrow("controller_detached"); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, identity.runId)); + expect(run.runnerProfileJson?.sessionCheckpoint).toBeUndefined(); + expect(run.nextEventSeq).toBe(1); + expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, identity.runId))).toEqual([]); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, identity.runId))).toEqual([]); + expect(await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, identity.runId))).toEqual([]); + // A new controller's independently authorized port can still settle it. + const replacement = new PaperclipControlPlanePort(db, binding); + await replacement.openRun({ identity, backendKind: "mock", sourceInstanceId: runnerId }); + await expect(replacement.completeRun({ result: CONTROL_PLANE_CONFORMANCE_RESULT, + terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL })).resolves.toBeUndefined(); + }); + it("redacts identity private material before persisting native events and checkpoints", async () => { const identity = { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, runId: randomUUID(), sessionId: randomUUID() }; const runnerId = randomUUID(); diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.ts b/server/src/services/native-runtime/paperclip-control-plane-port.ts index e41fd522c8e..bfd1cdc56b2 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.ts @@ -80,6 +80,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort { #redactedDeltaBytes = 0; readonly #binding: PaperclipControlPlaneBinding; #sessionId: string | null = null; + readonly #assertControllerActive?: () => void; readonly #onCommittedEvent?: (event: PrpEvent) => Promise; readonly #onDuplicateEvent?: (event: PrpEvent) => Promise; @@ -89,17 +90,25 @@ export class PaperclipControlPlanePort implements ControlPlanePort { options: { /** Runtime-only secret; never part of the persisted binding. */ privateKeyPem?: string; + /** Synchronous revocation fence for a controller handing off on restart. */ + assertControllerActive?: () => void; onCommittedEvent?: (event: PrpEvent) => Promise; onDuplicateEvent?: (event: PrpEvent) => Promise; } = {}, ) { this.#db = db; + this.#assertControllerActive = options.assertControllerActive; this.#identityRedactor = createAgentIdentityRedactor(options.privateKeyPem); this.#binding = structuredClone(binding); this.#onCommittedEvent = options.onCommittedEvent; this.#onDuplicateEvent = options.onDuplicateEvent; } + #assertActive(options?: { signal: AbortSignal }): void { + this.#assertControllerActive?.(); + options?.signal.throwIfAborted(); + } + #matchesPersistedBinding(run: typeof heartbeatRuns.$inferSelect): boolean { return run.companyId === this.#binding.companyId && run.agentId === this.#binding.agentId @@ -112,6 +121,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort { } async openRun(input: OpenControlPlaneRunInput): Promise { + this.#assertControllerActive?.(); const identity = input.identity; if ( identity.companyId !== this.#binding.companyId @@ -134,6 +144,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort { if (!run || !this.#matchesPersistedBinding(run)) { throw new Error("native_open_run_not_authorized"); } + this.#assertControllerActive?.(); this.#sessionId = identity.sessionId; } @@ -160,7 +171,8 @@ export class PaperclipControlPlanePort implements ControlPlanePort { return structuredClone(snapshot as PersistedNativeSession); } - async checkpointSession(snapshot: PersistedNativeSession): Promise { + async checkpointSession(snapshot: PersistedNativeSession, options?: { signal: AbortSignal }): Promise { + this.#assertActive(options); snapshot = this.#identityRedactor.redact(snapshot); const identity = snapshot.identity; if ( @@ -174,6 +186,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort { const run = await tx.select().from(heartbeatRuns) .where(eq(heartbeatRuns.id, this.#binding.runId)).for("update").limit(1) .then((rows) => rows[0] ?? null); + this.#assertActive(options); if (!run || !this.#matchesPersistedBinding(run)) { throw new Error("native_session_checkpoint_binding_mismatch"); } @@ -187,10 +200,12 @@ export class PaperclipControlPlanePort implements ControlPlanePort { nativePhaseUpdatedAt: new Date(), updatedAt: new Date(), }).where(eq(heartbeatRuns.id, this.#binding.runId)); + this.#assertActive(options); }); } - async appendEvent(value: NativeRunEvent | PrpEvent) { + async appendEvent(value: NativeRunEvent | PrpEvent, options?: { signal: AbortSignal }) { + this.#assertActive(options); if (!isPrpEvent(value)) throw new Error("native_legacy_event_not_supported"); const validated = validatePrpEvent(value); if (!validated.ok) throw new Error(`native_event_schema_invalid:${validated.issues[0]?.message ?? "unknown"}`); @@ -235,22 +250,30 @@ export class PaperclipControlPlanePort implements ControlPlanePort { } } } - const persisted = await appendHeartbeatRunEvent(this.#db, { - companyId: this.#binding.companyId, - runId: this.#binding.runId, - agentId: this.#binding.agentId, - eventType: event.eventType, - stream: "system", - level: event.eventType.includes("failed") ? "error" : "info", - payload: { prpEvent: event as unknown as Record }, - nativeSource: { - sourceInstanceId: event.sourceInstanceId, - sourceEventId: event.sourceEventId, - sourceSeq: event.sourceSeq, - protocolSchemaVersion: event.schemaVersion, - canonicalPayload: event as unknown as Record, - }, + const persisted = await this.#db.transaction(async tx => { + this.#assertActive(options); + const receipt = await appendHeartbeatRunEvent(tx as unknown as Db, { + companyId: this.#binding.companyId, + runId: this.#binding.runId, + agentId: this.#binding.agentId, + eventType: event.eventType, + stream: "system", + level: event.eventType.includes("failed") ? "error" : "info", + payload: { prpEvent: event as unknown as Record }, + nativeSource: { + sourceInstanceId: event.sourceInstanceId, + sourceEventId: event.sourceEventId, + sourceSeq: event.sourceSeq, + protocolSchemaVersion: event.schemaVersion, + canonicalPayload: event as unknown as Record, + }, + }); + // The nested append has no publication side effects. Revocation rolls + // back its event and sequence allocation before the outer commit. + this.#assertActive(options); + return receipt; }); + this.#assertActive(options); if (persisted.disposition === "committed") { publishChatPublicationCommitSignal({ companyId: this.#binding.companyId, @@ -305,7 +328,8 @@ export class PaperclipControlPlanePort implements ControlPlanePort { return { events, highestContiguousSourceSeq: cursor }; } - async completeRun(value: NativeRunResult | CompleteControlPlaneRunInput): Promise { + async completeRun(value: NativeRunResult | CompleteControlPlaneRunInput, options?: { signal: AbortSignal }): Promise { + this.#assertActive(options); value = this.#identityRedactor.redact(value); if (!isCompleteInput(value)) throw new Error("native_structured_result_required"); // Capture compatibility diagnostics before canonical validation removes @@ -341,6 +365,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort { const run = await tx.select().from(heartbeatRuns) .where(eq(heartbeatRuns.id, this.#binding.runId)).for("update").limit(1) .then((rows) => rows[0] ?? null); + this.#assertActive(options); if (!run || !this.#matchesPersistedBinding(run)) { throw new Error("native_result_binding_mismatch"); } @@ -352,6 +377,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort { .where(and(eq(nativeRunResults.runId, this.#binding.runId), or(...callerConditions))) .limit(1).then((rows) => rows[0] ?? null); if (existing) { + this.#assertActive(options); if (existing.canonicalSha256 !== canonicalSha256) throw new Error("structured_result_replay_conflict"); return; } @@ -404,6 +430,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort { }, updatedAt: new Date(), }).where(eq(heartbeatRuns.id, this.#binding.runId)); + this.#assertActive(options); }); } } diff --git a/server/src/services/usage-receipts.ts b/server/src/services/usage-receipts.ts index 7d02f0c047c..3606cd3a0ce 100644 --- a/server/src/services/usage-receipts.ts +++ b/server/src/services/usage-receipts.ts @@ -1,3 +1,4 @@ +import { priceAnthropicReceipt } from "./anthropic-pricing.js"; import { priceCodexReceipt } from "./codex-pricing.js"; import { randomUUID } from "node:crypto"; import { promises as fs } from "node:fs"; @@ -237,8 +238,8 @@ export async function createRunUsageRecorder(db: Db, input: { companyId: string; const attempt = parsed.attemptId ?? currentAttempt; if (!attempts.has(attempt)) currentAttempt = attempt; const previous = attempts.get(attempt); - const priced = priceCodexReceipt({ ...parsed, usage: parsed.usage ?? previous?.usage, - complete: parsed.complete && !(previous?.usage && !parsed.usage) }); + const priced = priceAnthropicReceipt(priceCodexReceipt({ ...parsed, usage: parsed.usage ?? previous?.usage, + complete: parsed.complete && !(previous?.usage && !parsed.usage) })); priced.pricingProvenance ??= { source: priced.costUsd != null || priced.costUsdExact != null || priced.cacheAdjustedCostUsd != null ? "provider_reported" : "unknown", version: "accounting-receipt/v1",