From 410e5c6eedba98588a87e4104153898fb0437532 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Sat, 24 Oct 2026 10:54:40 -0700 Subject: [PATCH] fix(native): use current authority for warm plugin completion (#17444) --- .../native-session-executor.test.ts | 25 ++++++++++++++++--- .../native-runtime/native-session-executor.ts | 7 +++++- 2 files changed, 28 insertions(+), 4 deletions(-) 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 66a235cc2ba..51e0c3af818 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -1,5 +1,6 @@ import * as environmentTaskRuntime from "../environment-task-runtime.js"; import * as pluginTaskRuntime from "./plugin-runner-task.js"; +import * as completionFeedback from "./native-completion-feedback.js"; import { environmentTaskOperationSchema } from "@paperclipai/plugin-sdk"; import { trace } from "@opentelemetry/api"; import { InMemorySpanExporter, SimpleSpanProcessor, TracerProvider } from "@opentelemetry/sdk-trace"; @@ -119,6 +120,7 @@ type BackendFactoryOptions = { }; }) => unknown; dynamicToolHandler?: (call: unknown) => Promise; + completionFeedback?: (result: Parameters[2]) => Promise; onSpawn?: (meta: { pid: number; processGroupId: number | null; @@ -10521,15 +10523,32 @@ describe("runnerd provider runtime wiring", () => { await writeRunnerAuthority(root, identity, "suspended"); }) }; const nextSession = { close: vi.fn(async () => undefined) }; - const result = { result: { summary: "done" }, terminal: { runTerminalState: "succeeded" }, + const validated = validatePrpStructuredRunResult({ schema: "paperclip.run_result.v1", reportedWorkDisposition: "done", + summary: "done", completionClaim: { contractRevision: first.completionContract.contract.revision, objectiveSatisfied: true, + criteria: [], remainingWork: [] }, evidence: [], verification: [], artifacts: [], attentionRequests: [] }); + if (!validated.ok) throw new Error("Invalid completion fixture"); + const result = { result: validated.result, terminal: { runTerminalState: "succeeded" }, turnId: "turn", normalizedSessionId: first.session.normalizedSessionId, providerSessionId: "provider", driverKind: "test", driverVersion: "1", nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null }; + const feedback = vi.spyOn(completionFeedback, "nativeCompletionFeedback").mockResolvedValue("accepted"); + const firstDb = leaseDb(first); + let retainedFeedback: BackendFactoryOptions["completionFeedback"]; state.execute.mockReset().mockImplementationOnce(async options => { + retainedFeedback = state.createBackend.mock.calls.at(-1)![1].completionFeedback; await options.onSession?.(firstSession); return result; }).mockImplementationOnce(async options => { order.push("execute-next"); + if (reusable) { + await expect(retainedFeedback!(result.result)).resolves.toBe("accepted"); + expect(feedback).toHaveBeenLastCalledWith(firstDb, next.binding.runId, result.result); + expect(state.toolAuthorityDefinitions).toHaveBeenLastCalledWith(expect.objectContaining({ runId: next.binding.runId })); + feedback.mockClear(); + state.toolAuthorityDefinitions.mockRejectedValueOnce(new Error("current authority denied")); + await expect(retainedFeedback!(result.result)).rejects.toThrow("current authority denied"); + expect(feedback).not.toHaveBeenCalled(); + } expect(options.existingSession).toBe(reusable ? firstSession : undefined); await options.onSession?.(reusable ? firstSession : nextSession); return result; @@ -10542,7 +10561,7 @@ describe("runnerd provider runtime wiring", () => { ? priorNativeRunDb(scenario === "stale-owner" ? older : first, scenario === "active-owner" ? "running" : "succeeded").select(fields as never) : ordinarySelect(fields as never)) as Db["select"]; try { - await executePaperclipNativeSession({ db: leaseDb(first), execution: first, + await executePaperclipNativeSession({ db: firstDb, execution: first, runnerInstanceId: identity.runnerInstanceId, useRunnerd: true, runnerExecutionTarget: pluginTarget as never,...pluginOptions }); if (scenario === "stale-owner") await writeRunnerAuthority(root, { ...identity, runId: older.binding.runId }, "ready"); if (scenario === "missing-owner") { @@ -10598,7 +10617,7 @@ describe("runnerd provider runtime wiring", () => { } } finally { failRetirement = false; - resume?.mockRestore();owners?.mockRestore(); + resume?.mockRestore();owners?.mockRestore();feedback.mockRestore(); await closeWarmNativeSessionsForEnvironment({ environmentId, reason: "test cleanup" }); } }); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 61092b3ce54..fe6d4732e76 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -11916,7 +11916,12 @@ async function createRunnerdBackendWithinSessionClaim( // its host environment out of the plugin and records the prepared directory. runnerInstanceId: prior?.runnerInstanceId ?? effectiveRunnerInstanceId, environment: {}, workingDirectoryAuthority: "runner_prepared_launch", dynamicTools, dynamicToolHandler: executeTool, - completionFeedback: async result => { await authorityEpoch.definitions(); return nativeCompletionFeedback(input.db, input.execution.binding.runId, result); }, + completionFeedback: async result => { + const current = sessionToolAuthorityEpochs.get(sessionScopeId); + if (!current) throw new Error("native_session_tool_authority_unavailable"); + await current.definitions(); + return nativeCompletionFeedback(input.db, current.runId, result); + }, codexTransportFactory: recovery => createRunnerdCodexTransport({ provider: provider.kind, stateDirectory: root, ...owner.driver, runnerArtifact: description.runner, ...(provider.kind === "acpx" ? { acpxAgent: provider.agent,