From 9f30eb10dddae987d47a7a949c077384ac8e0006 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sun, 20 Sep 2026 13:12:44 -0500 Subject: [PATCH] fix: reduce chat latency and preserve managed session reuse (#13710) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip manages AI agents and keeps their work attached to tasks. > - Chat connectors carry user messages and agent replies between a provider and those tasks. > - Each extra startup and context reset delays a reply. > - Managed account metadata was lost during adapter decoding, so compatible follow-ups started fresh. > - This branch fixes the reset and measures the remaining preparation, execution, and delivery costs. > - The changes must preserve account isolation, authorization, durable output, and recovery ownership. ## Linked Issues or Issue Description Refs #13699. The related service lifecycle work in #13410 and #13408 is separate; this branch focuses on task-bound chat response latency. **What happened?** Managed AI follow-ups started new provider sessions even after their configuration fingerprint stayed stable. The Codex codec removes unknown fields. The resume check then read the removed credential identity and treated it as a credential change. **Expected behavior** Compatible follow-ups resume the correct provider session. Changes to credentials, responsible users, permissions, or task configuration retain their reset behavior. **Steps to reproduce** 1. Use a Slack connector with a managed AI connection. 2. Send a message, then send a same-thread follow-up. 3. Inspect the configuration reset reason and the provider session identity. **Paperclip version or commit** Reproduced on `2a99de80ec52db01eead901f28323926ceaf3c1d`. **Deployment mode** Cloud staging with a native Codex runner. ## What Changed - Read saved credential identity before adapter decoding discards it. - Remove the internal credential identity from adapter-facing session params. - Test the real Codex codec and missing, changed, or unmanaged identity cases. - Preserve configured warm Codex runners and flush refreshed credentials after every turn. - Fence detached or closing session handles from successor credential ownership. - Stage current Codex launch credentials after restoring durable session history, uploading launch assets only once. - Reuse a runner binary already in the retained sandbox only when its SHA-256 matches the controller-owned artifact; still verify required capabilities before launch. - Lock the task before the run when saving results, preventing deadlocks with task updates. - Scope reusable projectless sandboxes to the company, environment, task, agent, and runtime configuration; verify Daytona sentinels for that scope. - Admit a new authorized chat message after a fully committed failed run and verified process cleanup. - Send compact deltas for verified plain-text Slack continuations. Match the actual prior run and current comment identity/body; exclude edited historical comments and prior agent output, preserve genuine brief edits and the full bootstrap fallback. - Keep attachments, omitted input, questions, approvals, recovery, and other providers on their existing framing. - Document managed session compatibility, credential lifecycle, and compact continuation boundaries. ## Verification - Workspace/session coverage: 156 tests passed. - Native session and credential ownership coverage: 390 tests passed, including exact artifact reuse, mismatches, failed probes, timeouts, and explicit artifact overrides. - Explicit continuation and durable chat authorization coverage: 172 tests passed. - Session resume and launch preparation coverage: 416 tests passed. - Result persistence coverage: 15 tests passed. The new concurrency test reproduced a PostgreSQL deadlock before the lock-order fix. - Environment lifecycle coverage: 92 tests passed, including projectless reuse and task/agent isolation at both selection and atomic handoff. - Daytona plugin coverage: 237 tests passed; 6 gated tests skipped. Standalone plugin build passed. - Compact Slack continuation and native resume coverage: 69 tests passed, including full-bootstrap retention, matching message authors/bodies, current-delivery selection, rejection of duplicate identities and historical comments, brief edits, and attachment/recovery fallbacks. - Final frozen-head `pnpm test:run` on repository-supported Node 26: 668 suites passed, 3 skipped, 1 failed; 12,797 tests passed and 82 skipped. The sole failure was a local `socket hang up` in `issue-recovery-actions.test.ts`, not an authorization assertion mismatch. All 57 tests in that suite passed three fresh reruns, and the suite passed latest-head CI. The full local invocation is therefore not claimed green. - An earlier Node 24 full run exposed an unrelated macOS symlink-cleanup failure; that 11-test catalog suite passes on Node 26 and in CI. No test behavior or timeout was relaxed. - Full local typecheck and build passed. Latest-head CI is green; Greptile is 5/5 with no unresolved review threads. - Two real Slack baseline replies took 25.1 and 24.6 seconds (24.9-second mean). Three same-thread signed probes on this head took 23.8, 23.9, and 22.7 seconds (23.5-second mean). This is a small sample and a modest wall-clock improvement, not a large or statistically established speedup. - In that same thread, uncached provider input fell from 8,514 tokens before compact input to 694–765 tokens afterward. The current delivery uses a 362-character delta; the full 19–21k-character bootstrap remains available for failed resume. Verified runner artifact preparation fell from about 1.2 seconds to 0.6 seconds. - A fresh thread created a separate task, sandbox, and provider session with full bootstrap (24.4 seconds). Its follow-up reused its own sandbox/session and compact input (28.3 seconds, including 16 seconds of model execution). Model variability and process startup remain substantial. - A signed duplicate webhook produced exactly one user comment, one successful run, and one final Slack reply. Slack's API independently confirmed the actual replies and a public task URL without an internal or pool hostname. - Earlier signed probes verified recovery after a failed run and reuse across a server deployment. The final idle test observed Daytona report the sandbox as stopped, then delivered a new reply in 19.9 seconds using the same sandbox/provider-session identity and compact input. Slack’s API confirmed that reply. - Live probes use signed synthetic inbound webhooks and real outbound Slack delivery, read back through Slack’s API. The final browser recheck found the Mac locked and the Slack tab blocked by another extension, so this is not claimed as full UI E2E proof. - This is a review branch. Do not merge until the maintainer reviews it. ## Risks - Incorrect session reuse could mix account or task context. Missing or changed identities continue to reset, and existing authorization checks remain in place. - Warm mode remains opt-in. Remote warm mode requires a reusable sandbox lease. Retained processes keep credentials until they close, so idle expiry and ownership fences are required. - A fresh user message may continue after a committed provider failure. Approval, current authorization, process termination, and prior-result checks remain required. - Projectless sandbox reuse is task- and agent-scoped. Missing or mismatched ownership cannot replace an existing lease; existing workspace-scoped leases keep their scope. Opt-in reuse retains a sandbox per task/agent, so provider auto-stop and deletion policies still determine idle compute and storage costs. Fleet defaults are unchanged. - Compact prompts apply only after proven resume and a matching prior-run delta. Missing or specialized context falls back to full input; fresh sessions always receive the full bootstrap. - No schema or migration changes. ## Model Used OpenAI GPT-6 through Codex, with code editing, tool use, and test execution. The exact serving model ID and context-window size are not exposed by this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [ ] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/connections/AI-CONNECTIONS.md | 40 +++++- .../daytona/src/plugin.test.ts | 6 +- .../sandbox-providers/daytona/src/plugin.ts | 6 +- .../src/__tests__/environment-runtime.test.ts | 64 +++++++++ .../heartbeat-workspace-session.test.ts | 28 +++- server/src/services/environment-runtime.ts | 32 +++-- server/src/services/environments.ts | 28 ++-- .../explicit-native-continuation.test.ts | 61 ++++++++- .../services/explicit-native-continuation.ts | 19 ++- server/src/services/heartbeat.ts | 55 ++++++-- .../managed-native-credentials.test.ts | 111 ++++++++++++++++ .../managed-native-credentials.ts | 77 +++++++++++ .../native-continuation.test.ts | 75 +++++++++++ .../native-runtime/native-continuation.ts | 48 ++++++- .../native-runtime/native-execution-input.ts | 1 + .../native-run-coordinator-store.ts | 6 + .../native-session-executor.test.ts | 75 +++++++++-- .../native-runtime/native-session-executor.ts | 123 +++++++++++++----- .../native-session-resume.test.ts | 26 ++++ .../native-runtime/native-session-resume.ts | 1 + .../paperclip-control-plane-port.test.ts | 47 ++++++- .../paperclip-control-plane-port.ts | 7 + 22 files changed, 841 insertions(+), 95 deletions(-) create mode 100644 server/src/services/native-runtime/managed-native-credentials.test.ts create mode 100644 server/src/services/native-runtime/managed-native-credentials.ts diff --git a/doc/connections/AI-CONNECTIONS.md b/doc/connections/AI-CONNECTIONS.md index 60afa650ade..f80623f5f06 100644 --- a/doc/connections/AI-CONNECTIONS.md +++ b/doc/connections/AI-CONNECTIONS.md @@ -142,9 +142,33 @@ Already-started native sessions retain their existing same-run recovery path; they must not be replaced by a fresh execution with a pre-provider receipt. Session reuse includes grant identity, responsible user, and credential -generation. A changed identity starts a fresh provider session. Managed native -executions use per-turn lifecycle cleanup; a suspended native execution whose -credential identity changed must restart as a new execution. +generation. A changed identity starts a fresh provider session. Native Codex +(`paperclip_runner`) honors the configured warm lifecycle. It copies refreshed +credentials back to the current invocation before deleting that invocation's +private home. The session-owned credential stays private until idle timeout or +explicit closure. Each follow-up rechecks current authorization and account +identity before reusing the session; changing identity retires the previous owner. +Other managed harnesses retain per-turn cleanup. A suspended native execution +whose credential identity changed must restart as a new execution. + +After a verified provider resume, plain-text Slack follow-ups send the new +authorized message delta instead of repeating the full task framing. The saved +run and current message identities and bodies must match. Actual brief edits +still arrive; historical Slack task titles are not repeated as new directions. +Attachments, omitted input, questions, approvals, and recovery retain their full +framing. A fresh provider session always receives the complete bootstrap. +Retained sandbox runner binaries are reused only after an exact SHA-256 match +with the controller artifact and the normal capability checks. Run-scoped +credential changes still require provider process rotation. + + +Warm sandbox execution requires both `reuseLease: true` and +`runnerLifecycleMode: "warm"` on the environment. `runnerIdleTimeoutMs` bounds +idle process retention. Chat tasks without a project reuse a sandbox only within +the same company, environment, task, agent, and runtime configuration. They do +not need an artificial project workspace. Other tasks, other agents, and ad-hoc +connection tests cannot claim that retained sandbox. Daytona verifies a matching +workspace sentinel before accepting either a workspace-scoped or task-scoped lease. Revocation blocks new invocations and refresh persistence. A running provider process may already hold credentials. The revoke confirmation lists attributed @@ -310,3 +334,13 @@ boundaries, and concurrent runs of one subscription for both providers. The opt- [`tests/hiring-ai-connections/README.md`](../../tests/hiring-ai-connections/README.md) describes real browser hiring, subtask, connection, and automatic-resume checks on local and Daytona environments, plus the production component Storybook checks. + +### Managed session compatibility + +Resume checks compare the selected account identity with the server-owned +metadata in the saved task session. Read this metadata before decoding the +adapter session: adapter codecs intentionally discard unknown fields. A missing +identity, a different grant or responsible user, or a changed credential generation +requires a fresh session. The metadata is removed before passing session params +to an adapter. Temporary authentication-home paths do not change the configuration +fingerprint. These checks do not relax current connection authorization. diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts index c01e83a309d..e29ac4780c5 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts @@ -3371,7 +3371,7 @@ describe("Daytona sandbox provider plugin", () => { expect(mockGet).toHaveBeenCalledTimes(2); }); - it("realizes a resumed lease after the provider fills in an unspecified target", async () => { + it.each(["workspace", "projectless task"])("realizes a resumed %s lease after the provider fills in an unspecified target", async (scope) => { process.env.DAYTONA_API_KEY = "host-key"; const sandbox = createMockSandbox({ id: "sandbox-default-target" }); mockCreate.mockResolvedValue(sandbox); @@ -3379,7 +3379,8 @@ describe("Daytona sandbox provider plugin", () => { const base = { driverKey: "daytona", companyId: "company-1", environmentId: "env-1" }; const config = { image: "node:20", timeoutMs: 300000, reuseLease: true }; const lease = await plugin.definition.onEnvironmentAcquireLease!({ - ...base, runId: "run-1", agentId: "agent-1", executionWorkspaceId: "workspace-1", config, + ...base, runId: "run-1", agentId: "agent-1", config, + ...(scope === "workspace" ? { executionWorkspaceId: "workspace-1" } : { issueId: "task-1" }), }); // The host materializes provider metadata into later operation config, // but resumes with the environment's original, target-less config. @@ -3404,6 +3405,7 @@ describe("Daytona sandbox provider plugin", () => { sandbox.state = "stopped"; const sentinel = lease.metadata!.workspaceSentinel as { token: string }; + expect(sentinel.token).toMatch(/^[a-f0-9]{64}$/); sandbox.process.executeCommand.mockResolvedValueOnce({ exitCode: 0, result: JSON.stringify({ token: sentinel.token }), artifacts: { stdout: JSON.stringify({ token: sentinel.token }) }, diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.ts index 151a22fa6a1..88ae8f3fdba 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.ts @@ -535,10 +535,10 @@ function parseProbeInteger(value: string | undefined | null): number | null { } function workspaceSentinelToken(input: { - params: Pick; + params: Pick; config: DaytonaDriverConfig; }): string | null { - if (!input.config.reuseLease || !input.params.agentId || !input.params.executionWorkspaceId) { + if (!input.config.reuseLease || !input.params.agentId || (!input.params.executionWorkspaceId && !input.params.issueId)) { return null; } return createHash("sha256") @@ -548,6 +548,7 @@ function workspaceSentinelToken(input: { environmentId: input.params.environmentId, agentId: input.params.agentId, executionWorkspaceId: input.params.executionWorkspaceId, + ...(input.params.executionWorkspaceId ? {} : { projectlessIssueId: input.params.issueId }), adapterType: input.params.adapterType ?? null, image: input.config.image, snapshot: input.config.snapshot, @@ -588,6 +589,7 @@ async function writeWorkspaceSentinel(input: { environmentId: input.params.environmentId, agentId: input.params.agentId, executionWorkspaceId: input.params.executionWorkspaceId, + ...(input.params.executionWorkspaceId ? {} : { projectlessIssueId: input.params.issueId }), adapterType: input.params.adapterType ?? null, provider: "daytona", writtenAt: new Date().toISOString(), diff --git a/server/src/__tests__/environment-runtime.test.ts b/server/src/__tests__/environment-runtime.test.ts index 9782336280a..a1eeab5167a 100644 --- a/server/src/__tests__/environment-runtime.test.ts +++ b/server/src/__tests__/environment-runtime.test.ts @@ -21,6 +21,7 @@ import { environments, executionWorkspaces, heartbeatRuns, + issues, plugins, projects, } from "@paperclipai/db"; @@ -216,6 +217,7 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { await rm(root, { recursive: true, force: true }).catch(() => undefined); } await db.delete(environmentLeases); + await db.delete(issues); await db.delete(heartbeatRuns); await db.delete(agents); await db.delete(environments); @@ -566,6 +568,68 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { ); }); + it.each(["same task", "different task", "different agent", "no task"])( + "scopes projectless reusable leases to their task and agent: %s", async (scenario) => { + const seeded = await seedReusablePluginSandboxLease(); + await environmentService(db).releaseLease(seeded.reusableLease.id, "expired"); + const issueId = randomUUID(); + const otherIssueId = randomUUID(); + await db.insert(issues).values([issueId, otherIssueId].map(id => ({ + id, companyId: seeded.companyId, title: "Projectless chat", status: "in_progress", + assigneeAgentId: seeded.agentId, + }))); + const nextAgentId = scenario === "different agent" ? randomUUID() : seeded.agentId; + if (nextAgentId !== seeded.agentId) await db.insert(agents).values({ + id: nextAgentId, companyId: seeded.companyId, name: "Other agent", adapterType: "paperclip_runner", + }); + let acquisitions = 0; + const call = vi.fn(async (_pluginId: string, method: string) => { + if (method === "environmentAcquireLease") return { + providerLeaseId: `projectless-${++acquisitions}`, metadata: { remoteCwd: "/workspace" }, + }; + if (method === "environmentResumeLease") return { + providerLeaseId: "projectless-1", metadata: { remoteCwd: "/workspace" }, + }; + if (method === "environmentReleaseLease") return undefined; + throw new Error(`Unexpected projectless lease method: ${method}`); + }); + const runtimeWithPlugin = environmentRuntimeService(db, { pluginWorkerManager: { + isRunning: () => true, call, + getWorker: () => ({ supportedMethods: ["environmentResumeLease", "environmentReleaseLease", "environmentDestroyLease"] }), + } as unknown as PluginWorkerManager }); + const first = await runtimeWithPlugin.acquireRunLease({ + companyId: seeded.companyId, environment: seeded.environment, issueId, + agentId: seeded.agentId, heartbeatRunId: seeded.runId, persistedExecutionWorkspace: null, + }); + expect(first.lease.metadata?.reusableSandboxLease).toMatchObject({ + projectlessIssueId: issueId, executionWorkspaceId: null, agentId: seeded.agentId, + }); + await runtimeWithPlugin.releaseRunLeases(seeded.runId, "released", undefined, "stop_and_retain"); + const nextRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: nextRunId, companyId: seeded.companyId, agentId: nextAgentId, status: "running", + }); + if (scenario === "different task" || scenario === "different agent") { + await expect(environmentService(db).acquireLease({ + companyId: seeded.companyId, environmentId: seeded.environment.id, + issueId: scenario === "different task" ? otherIssueId : issueId, + heartbeatRunId: nextRunId, executionWorkspaceId: null, + provider: "fake-plugin", providerLeaseId: first.lease.providerLeaseId, + leasePolicy: "reuse_by_environment", replacesReusableLeaseId: first.lease.id, + metadata: { agentId: nextAgentId }, + })).rejects.toMatchObject({ status: 409 }); + } + const next = await runtimeWithPlugin.acquireRunLease({ + companyId: seeded.companyId, environment: seeded.environment, + issueId: scenario === "no task" ? null : scenario === "different task" ? otherIssueId : issueId, + agentId: nextAgentId, heartbeatRunId: nextRunId, persistedExecutionWorkspace: null, + }); + const shouldResume = scenario === "same task"; + expect(next.lease.providerLeaseId).toBe(shouldResume ? "projectless-1" : "projectless-2"); + expect(acquisitions).toBe(shouldResume ? 1 : 2); + expect(call.mock.calls.filter((entry) => entry[1] === "environmentResumeLease")).toHaveLength(shouldResume ? 1 : 0); + }); + it("resumes the same provider lease on the next per-turn run", async () => { const seeded = await seedReusablePluginSandboxLease(); const workerManager = { diff --git a/server/src/__tests__/heartbeat-workspace-session.test.ts b/server/src/__tests__/heartbeat-workspace-session.test.ts index d644cb43620..04faeb834f9 100644 --- a/server/src/__tests__/heartbeat-workspace-session.test.ts +++ b/server/src/__tests__/heartbeat-workspace-session.test.ts @@ -43,6 +43,7 @@ import { stripConfiguredModelFromSessionParams, stripPaperclipSessionMetadataFromSessionParams, normalizeSessionParams, + isTaskSessionCredentialCompatible, shouldResetTaskSessionForWake, scrubGitCredentialText, buildAnchorFallbackWorkspaceNotes, @@ -2082,7 +2083,7 @@ describe("shouldResetTaskSessionForModelChange", () => { configuredModel: "gpt-5.4-mini", taskSessionParams: { sessionId: "thread-1", - __paperclipConfiguredModel: "gpt-5.4-mini", + __paperclipConfiguredModel: "gpt-5.4-mini", }, }), ).toBe(false); @@ -2619,6 +2620,7 @@ describe("stripPaperclipSessionMetadataFromSessionParams", () => { stripPaperclipSessionMetadataFromSessionParams({ sessionId: "thread-1", cwd: "/tmp/project", + paperclipAiCredentialIdentity: "grant:user:generation", __paperclipConfiguredModel: "gpt-5.4-mini", __paperclipConfigFingerprint: "v1:sha256:abc", __paperclipConfigFingerprintVersion: 1, @@ -2632,6 +2634,30 @@ describe("stripPaperclipSessionMetadataFromSessionParams", () => { }); }); +describe("isTaskSessionCredentialCompatible", () => { + it("retains the server-owned identity even when the Codex codec drops it", () => { + const saved = { sessionId: "thread-1", paperclipAiCredentialIdentity: "grant:user:generation" }; + const decoded = codexSessionCodec.deserialize(saved); + expect(decoded).toEqual({ sessionId: "thread-1" }); + expect(isTaskSessionCredentialCompatible(saved, "grant:user:generation")).toBe(true); + }); + + it.each([ + undefined, + null, + {}, + { paperclipAiCredentialIdentity: "other-grant:user:generation" }, + { paperclipAiCredentialIdentity: "grant:other-user:generation" }, + { paperclipAiCredentialIdentity: "grant:user:new-generation" }, + ])("requires the same saved grant, responsible user, and credential generation: %j", (saved) => { + expect(isTaskSessionCredentialCompatible(saved, "grant:user:generation")).toBe(false); + }); + + it("preserves unmanaged session behavior", () => { + expect(isTaskSessionCredentialCompatible({ sessionId: "thread-1" }, undefined)).toBe(true); + }); +}); + describe("normalizeSessionParams", () => { it("collapses an empty object to null", () => { expect(normalizeSessionParams({})).toBeNull(); diff --git a/server/src/services/environment-runtime.ts b/server/src/services/environment-runtime.ts index 5efccf36af5..323a3d98435 100644 --- a/server/src/services/environment-runtime.ts +++ b/server/src/services/environment-runtime.ts @@ -950,6 +950,7 @@ async function buildReusableSandboxLeaseFingerprint(input: { companyId: string; environment: Environment; executionWorkspaceId: string | null; + issueId: string | null; agentId: string | null; adapterType: string | null; provider: string; @@ -974,6 +975,7 @@ async function buildReusableSandboxLeaseFingerprint(input: { driver: input.environment.driver, }, executionWorkspaceId: input.executionWorkspaceId, + ...(input.executionWorkspaceId ? {} : { projectlessIssueId: input.issueId }), agentId: input.agentId, adapterType: input.adapterType, provider: input.provider, @@ -989,6 +991,7 @@ function buildReusableSandboxLeaseScope(input: { companyId: string; environmentId: string; executionWorkspaceId: string | null; + issueId: string | null; agentId: string | null; adapterType: string | null; provider: string; @@ -996,7 +999,9 @@ function buildReusableSandboxLeaseScope(input: { leaseFingerprint?: EffectiveRunConfigFingerprint | null; providerMetadata?: Record | null; }): Record | null { - if (!input.executionWorkspaceId || !input.agentId) return null; + // Chat tasks need not have a project workspace. Their task + agent identity + // is a stable, narrower reuse boundary; unscoped/ad-hoc probes cannot match it. + if ((!input.executionWorkspaceId && !input.issueId) || !input.agentId) return null; const providerMetadata = input.providerMetadata ?? {}; const adapterType = input.adapterType ?? null; const remoteCwd = readString(providerMetadata.remoteCwd); @@ -1008,6 +1013,7 @@ function buildReusableSandboxLeaseScope(input: { companyId: input.companyId, environmentId: input.environmentId, executionWorkspaceId: input.executionWorkspaceId, + ...(input.executionWorkspaceId ? {} : { projectlessIssueId: input.issueId }), agentId: input.agentId, adapterType, provider: input.provider, @@ -1029,6 +1035,7 @@ function reusableSandboxLeaseScopeMatches(input: { companyId: string; environmentId: string; executionWorkspaceId: string | null; + issueId: string | null; agentId: string | null; adapterType: string | null; provider: string; @@ -1036,7 +1043,7 @@ function reusableSandboxLeaseScopeMatches(input: { leaseFingerprint?: EffectiveRunConfigFingerprint | null; allowLegacyRuntimeFingerprint?: boolean; }): boolean { - if (!input.executionWorkspaceId || !input.agentId) return false; + if ((!input.executionWorkspaceId && !input.issueId) || !input.agentId) return false; const scope = input.lease.metadata?.reusableSandboxLease; if (!isRecord(scope)) return false; const adapterType = input.adapterType ?? null; @@ -1044,6 +1051,7 @@ function reusableSandboxLeaseScopeMatches(input: { scope.companyId === input.companyId && scope.environmentId === input.environmentId && scope.executionWorkspaceId === input.executionWorkspaceId && + (input.executionWorkspaceId !== null || scope.projectlessIssueId === input.issueId) && scope.agentId === input.agentId && scope.adapterType === adapterType && scope.provider === input.provider; @@ -1813,13 +1821,14 @@ function createSandboxEnvironmentDriver( supportsReusableLeases && parsed.config.reuseLease && input.heartbeatRunId !== null && - input.executionWorkspaceId !== null && + (input.executionWorkspaceId !== null || input.issueId !== null) && input.agentId !== null ? await buildReusableSandboxLeaseFingerprint({ db, companyId: input.companyId, environment: input.environment, executionWorkspaceId: input.executionWorkspaceId, + issueId: input.issueId, agentId: input.agentId, adapterType: input.adapterType, provider: parsed.config.provider, @@ -1842,13 +1851,14 @@ function createSandboxEnvironmentDriver( supportsReusableLeases && parsed.config.reuseLease && input.heartbeatRunId !== null && - input.executionWorkspaceId !== null && + (input.executionWorkspaceId !== null || input.issueId !== null) && input.agentId !== null ? (await environmentsSvc.listLeases(input.environment.id)) .filter((lease) => lease.leasePolicy === "reuse_by_environment" && reusableLeaseCanBeResumed({ lease, heartbeatRunId: input.heartbeatRunId }) && lease.executionWorkspaceId === input.executionWorkspaceId && + (input.executionWorkspaceId !== null || lease.issueId === input.issueId) && lease.metadata?.agentId === input.agentId, ) : []; @@ -1858,6 +1868,7 @@ function createSandboxEnvironmentDriver( companyId: input.companyId, environmentId: input.environment.id, executionWorkspaceId: input.executionWorkspaceId, + issueId: input.issueId, agentId: input.agentId, adapterType: input.adapterType, provider: parsed.config.provider, @@ -1880,7 +1891,7 @@ function createSandboxEnvironmentDriver( supportsReusableLeases && parsed.config.reuseLease && input.heartbeatRunId !== null && - input.executionWorkspaceId !== null && + (input.executionWorkspaceId !== null || input.issueId !== null) && input.agentId !== null ? findReusableSandboxLeaseId({ config: storedConfig, leases: reusableExistingLeases }) : null; @@ -2044,6 +2055,7 @@ function createSandboxEnvironmentDriver( companyId: input.companyId, environmentId: input.environment.id, executionWorkspaceId: input.executionWorkspaceId, + issueId: input.issueId, agentId: input.agentId, adapterType: input.adapterType, provider: parsed.config.provider, @@ -2179,13 +2191,14 @@ function createSandboxEnvironmentDriver( supportsReusableLeases && parsed.config.reuseLease && input.heartbeatRunId !== null && - input.executionWorkspaceId !== null && + (input.executionWorkspaceId !== null || input.issueId !== null) && input.agentId !== null ? await buildReusableSandboxLeaseFingerprint({ db, companyId: input.companyId, environment: input.environment, executionWorkspaceId: input.executionWorkspaceId, + issueId: input.issueId, agentId: input.agentId, adapterType: input.adapterType, provider: parsed.config.provider, @@ -2196,13 +2209,14 @@ function createSandboxEnvironmentDriver( supportsReusableLeases && parsed.config.reuseLease && input.heartbeatRunId !== null && - input.executionWorkspaceId !== null && + (input.executionWorkspaceId !== null || input.issueId !== null) && input.agentId !== null ? (await environmentsSvc.listLeases(input.environment.id)) .filter((lease) => lease.leasePolicy === "reuse_by_environment" && reusableLeaseCanBeResumed({ lease, heartbeatRunId: input.heartbeatRunId }) && lease.executionWorkspaceId === input.executionWorkspaceId && + (input.executionWorkspaceId !== null || lease.issueId === input.issueId) && lease.metadata?.agentId === input.agentId, ) : []; @@ -2212,6 +2226,7 @@ function createSandboxEnvironmentDriver( companyId: input.companyId, environmentId: input.environment.id, executionWorkspaceId: input.executionWorkspaceId, + issueId: input.issueId, agentId: input.agentId, adapterType: input.adapterType, provider: parsed.config.provider, @@ -2234,7 +2249,7 @@ function createSandboxEnvironmentDriver( supportsReusableLeases && parsed.config.reuseLease && input.heartbeatRunId !== null && - input.executionWorkspaceId !== null && + (input.executionWorkspaceId !== null || input.issueId !== null) && input.agentId !== null ? findReusableSandboxLeaseId({ config: parsed.config, leases: reusableExistingLeases }) : null; @@ -2284,6 +2299,7 @@ function createSandboxEnvironmentDriver( companyId: input.companyId, environmentId: input.environment.id, executionWorkspaceId: input.executionWorkspaceId, + issueId: input.issueId, agentId: input.agentId, adapterType: input.adapterType, provider: parsed.config.provider, diff --git a/server/src/services/environments.ts b/server/src/services/environments.ts index ea01905eb36..c63a3974adb 100644 --- a/server/src/services/environments.ts +++ b/server/src/services/environments.ts @@ -1,5 +1,5 @@ import { randomUUID } from "node:crypto"; -import { and, desc, eq, inArray, ne, sql } from "drizzle-orm"; +import { and, desc, eq, inArray, isNull, ne, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { agents, @@ -1443,10 +1443,10 @@ export function environmentService(db: Db) { }; if ( (input.replacesReusableLeaseId || input.reusesReusableLeaseId) && - (!input.executionWorkspaceId || !input.providerLeaseId) + ((!input.executionWorkspaceId && (!input.issueId || typeof input.metadata?.agentId !== "string")) || !input.providerLeaseId) ) { throw new Error( - "A reusable lease handoff requires an execution workspace and provider lease id.", + "A reusable lease handoff requires a workspace or agent-bound task, and a provider lease id.", ); } if (input.reusesReusableLeaseId && !input.heartbeatRunId) { @@ -1523,10 +1523,13 @@ export function environmentService(db: Db) { eq(environmentLeases.id, input.replacesReusableLeaseId), eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.environmentId, input.environmentId), - eq( - environmentLeases.executionWorkspaceId, - input.executionWorkspaceId!, - ), + input.executionWorkspaceId + ? eq(environmentLeases.executionWorkspaceId, input.executionWorkspaceId) + : and( + isNull(environmentLeases.executionWorkspaceId), + eq(environmentLeases.issueId, input.issueId!), + sql`${environmentLeases.metadata}->>'agentId' = ${input.metadata!.agentId}`, + ), eq(environmentLeases.leasePolicy, "reuse_by_environment"), eq( environmentLeases.providerLeaseId, @@ -1563,10 +1566,13 @@ export function environmentService(db: Db) { eq(environmentLeases.id, input.reusesReusableLeaseId), eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.environmentId, input.environmentId), - eq( - environmentLeases.executionWorkspaceId, - input.executionWorkspaceId!, - ), + input.executionWorkspaceId + ? eq(environmentLeases.executionWorkspaceId, input.executionWorkspaceId) + : and( + isNull(environmentLeases.executionWorkspaceId), + eq(environmentLeases.issueId, input.issueId!), + sql`${environmentLeases.metadata}->>'agentId' = ${input.metadata!.agentId}`, + ), eq( environmentLeases.heartbeatRunId, input.heartbeatRunId!, diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index e8bad1eb3ec..4c02a743e2e 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -6,17 +6,18 @@ import { mkdtemp, mkdir, writeFile, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { and, eq } from "drizzle-orm"; -import { beforeAll, afterAll, describe, it, expect } from "vitest"; +import { beforeAll, afterAll, describe, it, expect, vi } from "vitest"; import { approvals, issueApprovals, issueThreadInteractions, agentWakeupRequests, agents, companies, createDb, heartbeatRunEvents, heartbeatRuns, issueComments, issueRecoveryActions, - issues, nativeRunFinalizations, environmentLeases, environments, issueRelations, issueTreeHolds, issueTreeHoldMembers, + issues, nativeRunFinalizations, nativeRunResults, completionContracts, environmentLeases, environments, issueRelations, issueTreeHolds, issueTreeHoldMembers, } from "@paperclipai/db"; import { startEmbeddedPostgresTestDatabase, getEmbeddedPostgresTestSupport } from "../__tests__/helpers/embedded-postgres.js"; import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js"; import { buildExecutionContinuation } from "./execution-continuation.js"; import { heartbeatService, persistHeartbeatRunProcessMetadata, type HeartbeatEnvironmentRuntime } from "./heartbeat.js"; import { getExecutionBlocker } from "./execution-blocker.js"; +import { createDurableChatWakeupRequest } from "./durable-chat-wakeup.js"; const support = await getEmbeddedPostgresTestSupport(); (support.supported ? describe : describe.skip)("explicit native conversation continuation", () => { let database: Awaited>; @@ -952,6 +953,62 @@ const support = await getEmbeddedPostgresTestSupport(); expect(coordinator.attempt).toBe(3); expect(coordinator.failureDetail?.replacementDenied).toBe("explicit_user_continuation"); }); + it.each(["failed", "succeeded_result", "rejected_result", "still_committing", "controller", "successor", "live_process"])( + "requires a fully committed failed result and verified stop for a new turn (%s)", async scenario => { + const f = await seed(), contractId = randomUUID(), resultId = randomUUID(); + await db.insert(completionContracts).values({ id: contractId, companyId: f.companyId, issueId: f.issueId, + revision: 1, schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test", + risk: "standard", completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", + contractJson: {}, canonicalSha256: contractId, createdByActorType: "system", createdByActorId: "test" }); + await db.update(heartbeatRuns).set({ completionContractId: contractId, + ...(scenario === "live_process" ? { processPid: process.pid } : {}), + }).where(eq(heartbeatRuns.id, f.sourceRunId)); + await db.insert(nativeRunResults).values({ id: resultId, companyId: f.companyId, issueId: f.issueId, + runId: f.sourceRunId, completionContractId: contractId, serverFingerprint: resultId, + schemaStatus: scenario === "rejected_result" ? "rejected" : "accepted", canonicalSha256: resultId, + resultJson: { terminal: { runTerminalState: scenario === "succeeded_result" ? "succeeded" : "failed" } } }); + await db.update(nativeRunFinalizations).set({ resultId, + phase: scenario === "still_committing" ? "result_accepted" : "committed", + leaseOwner: scenario === "controller" ? "active-controller" : null, + failureDetail: scenario === "successor" ? { successorRunId: randomUUID() } : null, + }).where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + const result = await admit(f); + if (scenario === "failed") { + expect(result).toEqual({ previousRunId: f.sourceRunId, commentId: f.commentId }); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + const [saved] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId)); + expect(saved).toMatchObject({ phase: "committed", resultId }); + } else { + expect(result).toBeNull(); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull(); + } + }); + it.each(["linked", "revoked", "unlinked", "wrong_author"])( + "admits a fresh Slack message after failure only with current linked-user authority (%s)", async scenario => { + const f = await seed(); + // Keep execution queued so this tests real admission without launching a provider. + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }); + const authorize = vi.fn(async () => { if (scenario === "revoked") throw new Error("chat_authorization_revoked"); }); + const request = createDurableChatWakeupRequest({ id: randomUUID(), ...f, + requestedByActorType: scenario === "unlinked" ? "system" : "user", + requestedByActorId: scenario === "wrong_author" ? "another-user" : "board", requestedAt: new Date(), authorize }); + const wake = () => heartbeatService(db).wakeup(f.agentId, { + source: "assignment", triggerDetail: "system", reason: "External chat message received", + requestedByActorType: request.requestedByActorType, requestedByActorId: request.requestedByActorId, + payload: { issueId: f.issueId, wakeCommentId: f.commentId, mutation: "chat_message_received" }, + contextSnapshot: { issueId: f.issueId, wakeCommentId: f.commentId, source: "chat:slack" }, + durableChatRequest: request, + }); + if (scenario === "revoked") await expect(wake()).rejects.toThrow("chat_authorization_revoked"); + else await wake(); + expect(authorize).toHaveBeenCalled(); + const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued"))); + expect(successors).toHaveLength(scenario === "linked" ? 1 : 0); + if (scenario === "linked") { + expect(successors[0].contextSnapshot).toMatchObject({ forceFreshSession: true, previousRunId: f.sourceRunId }); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + } else expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull(); + }); it.each(["live_process", "missing_process", "lease", "coordinator", "successor", "agent_message", "old_comment", "wrong_author", "run_authored", "reassigned", "automatic", "approval", "question", "malformed_comment", "legacy_owner"])("keeps the hold for %s", async kind => { const f = await seed(); if (kind === "legacy_owner") await db.insert(heartbeatRuns).values({ companyId: f.companyId, diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index b9cff216ac5..dc7bf15d655 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -8,7 +8,7 @@ import { and, eq, inArray, isNull, ne, or, sql } from "drizzle-orm"; import { agents, agentWakeupRequests, approvals, issueApprovals, issueThreadInteractions, environmentLeases, heartbeatRuns, issueComments, issueRecoveryActions, - issues, nativeRunFinalizations, type Db, + issues, nativeRunFinalizations, nativeRunResults, type Db, } from "@paperclipai/db"; import { executionBlockerPredicate, getExecutionBlocker } from "./execution-blocker.js"; import { buildExecutionContinuation } from "./execution-continuation.js"; @@ -195,8 +195,21 @@ export async function admitExplicitNativeContinuation(input: { const cancelledStartup = await isCancelledNativeStartup(db, run, coordinator); if (cancelledStartup) cancelledStartupIds.add(run.id); if (run.runtimeMode !== "native" && !unusedAdmission && !legacyUserTurn && !cancelledStartup) return null; - if (!cancelledStartup && coordinator && (coordinator.phase !== "terminal_failure" || coordinator.leaseOwner || - coordinator.resultId || coordinator.failureDetail?.successorRunId)) return blocked("controller_settling", + // A provider failure can finish the normal result/assessment commit path. + // Its accepted failed result is immutable history, not a live controller. + // Only a new user turn may pass this gate; the process/lease stop proofs + // below remain mandatory and no previous action is automatically replayed. + const [committedFailure] = coordinator?.phase === "committed" && coordinator.resultId && run.status === "failed" + ? await db.select({ id: nativeRunResults.id }).from(nativeRunResults).where(and( + eq(nativeRunResults.companyId, companyId), eq(nativeRunResults.issueId, issueId), + eq(nativeRunResults.runId, run.id), eq(nativeRunResults.id, coordinator.resultId), + eq(nativeRunResults.schemaStatus, "accepted"), + sql`${nativeRunResults.resultJson}->'terminal'->>'runTerminalState' = 'failed'`, + )).limit(1) + : []; + if (!cancelledStartup && coordinator && ( + (coordinator.phase !== "terminal_failure" && !committedFailure) || coordinator.leaseOwner || + (coordinator.resultId && !committedFailure) || coordinator.failureDetail?.successorRunId)) return blocked("controller_settling", run.status === "cancelled" && !coordinator.leaseOwner ? "The cancelled run still needs verified cleanup. Your message is saved. Inspect the run and its environment for details." : "Waiting for the previous run to finish recovery. Your message will start automatically."); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index e572e0a4fac..e65149ac941 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -5746,6 +5746,7 @@ export function shouldDeferFollowupWakeForSameIssue(input: { return false; } +const SESSION_AI_CREDENTIAL_IDENTITY_KEY = "paperclipAiCredentialIdentity"; const SESSION_CONFIGURED_MODEL_KEY = "__paperclipConfiguredModel"; const SESSION_CONFIG_FINGERPRINT_KEY = "__paperclipConfigFingerprint"; const SESSION_CONFIG_FINGERPRINT_VERSION_KEY = @@ -5754,6 +5755,7 @@ const SESSION_CONFIG_CATEGORIES_KEY = "__paperclipConfigCategories"; const SESSION_CONFIG_CATEGORY_FINGERPRINTS_KEY = "__paperclipConfigCategoryFingerprints"; const PAPERCLIP_SESSION_METADATA_KEYS = new Set([ + SESSION_AI_CREDENTIAL_IDENTITY_KEY, SESSION_CONFIGURED_MODEL_KEY, SESSION_CONFIG_FINGERPRINT_KEY, SESSION_CONFIG_FINGERPRINT_VERSION_KEY, @@ -6746,6 +6748,15 @@ export function resolveExecutionWorkspaceConfigFreshness(input: { }; } +/** Read server-owned identity before adapter codecs discard unknown metadata. */ +export function isTaskSessionCredentialCompatible( + storedSessionParams: Record | null | undefined, + managedAiCredentialIdentity: string | undefined, +): boolean { + if (!managedAiCredentialIdentity) return true; + return storedSessionParams?.[SESSION_AI_CREDENTIAL_IDENTITY_KEY] === managedAiCredentialIdentity; +} + function readConfiguredModelFromAdapterConfig( adapterConfig: Record | null | undefined, ) { @@ -6761,7 +6772,7 @@ function attachPaperclipSessionMetadataToSessionParams( const next = { ...(sessionParams ?? {}) }; if (configuredModel) next[SESSION_CONFIGURED_MODEL_KEY] = configuredModel; if (configMetadata) { - if (configMetadata.aiCredentialIdentity) next.paperclipAiCredentialIdentity = configMetadata.aiCredentialIdentity; + if (configMetadata.aiCredentialIdentity) next[SESSION_AI_CREDENTIAL_IDENTITY_KEY] = configMetadata.aiCredentialIdentity; next[SESSION_CONFIG_FINGERPRINT_KEY] = configMetadata.fingerprint; next[SESSION_CONFIG_FINGERPRINT_VERSION_KEY] = configMetadata.version; next[SESSION_CONFIG_CATEGORIES_KEY] = configMetadata.categories; @@ -22487,9 +22498,13 @@ export function heartbeatService( delete context.paperclipPreviousSessionId; } + const taskSessionCredentialCompatible = isTaskSessionCredentialCompatible( + taskSession?.sessionParamsJson, + managedAiRuntime?.identity, + ); if (managedAiRuntime) { sessionConfigMetadata.aiCredentialIdentity = managedAiRuntime.identity; - if (taskSessionDecodedParams?.paperclipAiCredentialIdentity !== managedAiRuntime.identity) { + if (!taskSessionCredentialCompatible) { runtimeSessionIdForAdapter = null; runtimeSessionParamsForAdapter = null; previousSessionDisplayId = null; @@ -23021,7 +23036,7 @@ export function heartbeatService( return requests.length > 0 ? requests : undefined; })(), }); - const taskNativeSessionId = managedAiRuntime && taskSessionDecodedParams?.paperclipAiCredentialIdentity !== managedAiRuntime.identity ? null : readNonEmptyString( + const taskNativeSessionId = !taskSessionCredentialCompatible ? null : readNonEmptyString( taskSessionDecodedParams?.sessionId, ); // Compatibility for native retry rows created before same-run restart @@ -23141,7 +23156,14 @@ export function heartbeatService( executionTarget.transport === "sandbox" ? (executionTarget.runnerLifecyclePolicy ?? null) : null; - const effectiveLifecyclePolicy = managedAiRuntime ? { mode: "per_turn" as const, idleTimeoutMs: null } : environmentLifecyclePolicy ?? agentLifecyclePolicy; + // Native Codex owns a durable, session-scoped home. It flushes refreshed + // auth into each invocation's private home before that home is removed. + // Other managed harnesses still require per-turn credential cleanup. + const supportsManagedWarmSession = agent.adapterType === "paperclip_runner" && + nativeRuntimeResolution.profile.backend === "codex_app_server"; + const effectiveLifecyclePolicy = managedAiRuntime && !supportsManagedWarmSession + ? { mode: "per_turn" as const, idleTimeoutMs: null } + : environmentLifecyclePolicy ?? agentLifecyclePolicy; if ( effectiveLifecyclePolicy.mode === "warm" && executionTarget?.kind === "remote" && @@ -23352,10 +23374,17 @@ export function heartbeatService( previousTurn: (() => { if (!previousNativeRun || nativeReviewRequest) return null; try { - return { - runId: previousNativeRun.id, - task: parseNativeExecutionInput(parseObject(previousNativeRun.runnerProfileJson).nativeExecutionInput).task, - }; + const previousTask = parseNativeExecutionInput(parseObject(previousNativeRun.runnerProfileJson).nativeExecutionInput).task; + if (paperclipWakePayload?.externalChatProvider) { + // External native inputs use a neutral task title. Compare + // the saved canonical brief so old provider text is not + // repeated as a change, while genuine edits still arrive. + const savedIssue = parseObject(parseObject(previousNativeRun.contextSnapshot).paperclipIssue); + if (savedIssue.id !== issueRef.id || typeof savedIssue.title !== "string" || + (savedIssue.description !== null && typeof savedIssue.description !== "string")) return null; + return { runId: previousNativeRun.id, task: { title: savedIssue.title, description: savedIssue.description } }; + } + return { runId: previousNativeRun.id, task: previousTask }; } catch { // An invalid prior snapshot must use the fresh bootstrap. return null; @@ -24059,6 +24088,7 @@ export function heartbeatService( // Bootstrap with executable/home discovery while keeping // configured provider values and the server-selected // workspace boundary authoritative. + managedAiCredentialIdentity: managedAiRuntime?.identity, managedAiCredentialHome: managedAiRuntime ? String((managedAiRuntime.config.env as Record).CODEX_HOME) : undefined, runnerEnvironment: { ...buildNativeProviderEnvironment( @@ -26984,10 +27014,14 @@ export function heartbeatService( ); // Prove eligibility without retiring the hold. Later gates can still // decline this wake; hold retirement and successor creation stay atomic. + // A bound chat request has already rechecked its current principal + // above. Treat its new user message like a board comment, but keep + // failed-run retry actions on their separate exact-request path. if (executionBlocker && !(await admitExplicitNativeContinuation({ db: tx as unknown as Db, companyId: issue.companyId, issueId: issue.id, agentId, actorType: opts.requestedByActorType, actorId: opts.requestedByActorId, - reason, commentId: wakeCommentId ?? null, failedRunId: opts.failedRunId, successorRunId: explicitContinuationRunId, + reason: durableRequest && !failedChatRetry ? "issue_commented" : reason, + commentId: wakeCommentId ?? null, failedRunId: opts.failedRunId, successorRunId: explicitContinuationRunId, queuedCommentInterruptId: opts.queuedCommentInterruptId, queuedCommentRequestId: opts.queuedCommentRequestId, dryRun: true, @@ -27752,7 +27786,8 @@ export function heartbeatService( const explicitContinuation = await admitExplicitNativeContinuation({ db: tx as unknown as Db, companyId: issue.companyId, issueId: issue.id, agentId, actorType: opts.requestedByActorType, actorId: opts.requestedByActorId, - reason, commentId: wakeCommentId ?? null, failedRunId: opts.failedRunId, successorRunId: explicitContinuationRunId, + reason: durableRequest && !failedChatRetry ? "issue_commented" : reason, + commentId: wakeCommentId ?? null, failedRunId: opts.failedRunId, successorRunId: explicitContinuationRunId, queuedCommentInterruptId: opts.queuedCommentInterruptId, queuedCommentRequestId: opts.queuedCommentRequestId, }); diff --git a/server/src/services/native-runtime/managed-native-credentials.test.ts b/server/src/services/native-runtime/managed-native-credentials.test.ts new file mode 100644 index 00000000000..8075ba9f8f4 --- /dev/null +++ b/server/src/services/native-runtime/managed-native-credentials.test.ts @@ -0,0 +1,111 @@ +import { describe, expect, it, vi } from "vitest"; +import type { NativeSession } from "../../vendor/paperclip-runner/index.js"; +import { bindManagedNativeCredentialTurn, completeManagedNativeCredentialTurn } from "./managed-native-credentials.js"; + +function fixture() { + const close = vi.fn(async () => {}); + const detach = vi.fn(async () => {}); + const session = { close, detachControllerForRestart: detach } as unknown as NativeSession; + const first = { copyBack: vi.fn(async () => {}), remove: vi.fn(async () => {}) }; + bindManagedNativeCredentialTurn(session, first); + return { session, close, detach, first }; +} + +describe("managed native credential turns", () => { + it("flushes each warm turn into its own home without stopping the provider", async () => { + const { session, close, first } = fixture(); + await completeManagedNativeCredentialTurn(session); + expect(first.copyBack).toHaveBeenCalledOnce(); + expect(first.remove).not.toHaveBeenCalled(); + expect(close).not.toHaveBeenCalled(); + + const next = { copyBack: vi.fn(async () => {}), remove: vi.fn(async () => {}) }; + bindManagedNativeCredentialTurn(session, next); + await completeManagedNativeCredentialTurn(session); + await session.close({ reason: "idle timeout" }); + await session.close({ reason: "repeated cleanup" }); + expect(first.copyBack).toHaveBeenCalledOnce(); + expect(first.remove).not.toHaveBeenCalled(); + expect(next.copyBack).toHaveBeenCalledOnce(); + expect(next.remove).toHaveBeenCalledOnce(); + }); + + it("flushes and removes credentials when a per-turn session closes", async () => { + const { session, first } = fixture(); + await session.close({ reason: "turn complete" }); + await completeManagedNativeCredentialTurn(session); + expect(first.copyBack).toHaveBeenCalledOnce(); + expect(first.remove).toHaveBeenCalledOnce(); + }); + + it("keeps the current invocation exclusive until copy-back settles", async () => { + const { session } = fixture(); + expect(() => bindManagedNativeCredentialTurn(session, { copyBack: vi.fn(), remove: vi.fn() })) + .toThrow("managed_native_credential_turn_still_owned"); + await completeManagedNativeCredentialTurn(session); + let finish!: () => void; + const next = { copyBack: vi.fn(() => new Promise(resolve => { finish = resolve; })), remove: vi.fn(async () => {}) }; + bindManagedNativeCredentialTurn(session, next); + const one = completeManagedNativeCredentialTurn(session); + const two = completeManagedNativeCredentialTurn(session); + expect(() => bindManagedNativeCredentialTurn(session, { copyBack: vi.fn(), remove: vi.fn() })) + .toThrow("managed_native_credential_turn_still_owned"); + finish(); + await Promise.all([one, two]); + expect(next.copyBack).toHaveBeenCalledOnce(); + }); + + it("leaves retained credentials alone after controller detach", async () => { + const { session, close, first } = fixture(); + await session.detachControllerForRestart!(); + await completeManagedNativeCredentialTurn(session); + await session.close({ reason: "late old-owner cleanup" }); + expect(first.copyBack).not.toHaveBeenCalled(); + expect(first.remove).not.toHaveBeenCalled(); + expect(close).not.toHaveBeenCalled(); + expect(() => bindManagedNativeCredentialTurn(session, { copyBack: vi.fn(), remove: vi.fn() })) + .toThrow("managed_native_credential_session_retired"); + }); + + it("does not delete a successor's credentials when an old close finishes late", async () => { + let finish!: () => void; + const session = { close: () => new Promise(resolve => { finish = resolve; }), detachControllerForRestart: async () => {} } as unknown as NativeSession; + const first = { copyBack: vi.fn(async () => {}), remove: vi.fn(async () => {}) }; + bindManagedNativeCredentialTurn(session, first); + const closing = session.close({ reason: "old owner" }); + await session.detachControllerForRestart!(); + const next = { copyBack: vi.fn(async () => {}), remove: vi.fn(async () => {}) }; + expect(() => bindManagedNativeCredentialTurn(session, next)) + .toThrow("managed_native_credential_session_retired"); + const successorClose = vi.fn(async () => {}); + const successor = { close: successorClose } as unknown as NativeSession; + bindManagedNativeCredentialTurn(successor, next); + finish(); + await closing; + await session.close({ reason: "old reference after successor attached" }); + expect(first.remove).not.toHaveBeenCalled(); + expect(next.remove).not.toHaveBeenCalled(); + expect(next.copyBack).not.toHaveBeenCalled(); + expect(successorClose).not.toHaveBeenCalled(); + }); + + it("never rebinds a handle whose close is in progress", async () => { + let finish!: () => void; + const session = { close: () => new Promise(resolve => { finish = resolve; }) } as unknown as NativeSession; + bindManagedNativeCredentialTurn(session, { copyBack: vi.fn(async () => {}), remove: vi.fn(async () => {}) }); + await completeManagedNativeCredentialTurn(session); + const closing = session.close({ reason: "idle timeout" }); + expect(() => bindManagedNativeCredentialTurn(session, { copyBack: vi.fn(), remove: vi.fn() })) + .toThrow("managed_native_credential_session_retired"); + finish(); + await closing; + }); + + it("removes stopped-provider credentials even when refresh copy-back fails", async () => { + const session = { close: async () => {} } as unknown as NativeSession; + const remove = vi.fn(async () => {}); + bindManagedNativeCredentialTurn(session, { copyBack: async () => { throw new Error("copy-back failed"); }, remove }); + await expect(session.close({ reason: "completed" })).rejects.toThrow("copy-back failed"); + expect(remove).toHaveBeenCalledOnce(); + }); +}); diff --git a/server/src/services/native-runtime/managed-native-credentials.ts b/server/src/services/native-runtime/managed-native-credentials.ts new file mode 100644 index 00000000000..898211a845b --- /dev/null +++ b/server/src/services/native-runtime/managed-native-credentials.ts @@ -0,0 +1,77 @@ +import type { NativeSession } from "../../vendor/paperclip-runner/index.js"; + +type CredentialTurn = { + /** Copy refreshed credentials only into this invocation's private home. */ + copyBack: () => Promise; + /** Remove launch credentials after the provider owner has stopped. */ + remove: () => Promise; + detached: boolean; + closing?: boolean; + completed: boolean; + completion?: Promise; + removed?: boolean; +}; + +const credentialTurns = new WeakMap(); + +/** A warm session outlives an invocation's temporary credential destination. */ +export function bindManagedNativeCredentialTurn( + session: NativeSession, + callbacks: Pick, +): NativeSession { + const previous = credentialTurns.get(session); + // A successor controller must attach a new handle. Reusing a detached or + // closing handle would let old references close the successor's provider. + if (previous?.detached || previous?.closing) { + throw new Error("managed_native_credential_session_retired"); + } + if (previous && !previous.completed) { + throw new Error("managed_native_credential_turn_still_owned"); + } + const turn: CredentialTurn = { ...callbacks, detached: false, completed: false }; + credentialTurns.set(session, turn); + if (previous) return session; + + const close = session.close.bind(session); + const detach = session.detachControllerForRestart?.bind(session); + if (detach) { + session.detachControllerForRestart = async () => { + // Set before yielding: a late close from this controller cannot touch + // credentials retained by the replacement controller. + const owner = credentialTurns.get(session); + if (owner) owner.detached = true; + await detach(); + }; + } + session.close = async (input) => { + const owner = credentialTurns.get(session); + if (owner?.detached) return; + if (owner) owner.closing = true; + await close(input); + if (!owner || owner.detached || credentialTurns.get(session) !== owner) return; + try { + await completeCredentialTurn(owner); + } finally { + // Idle expiry removes the provider credential, but never writes back to + // the already-deleted temporary home of a completed invocation. + if (!owner.detached && !owner.removed && credentialTurns.get(session) === owner) { + owner.removed = true; + await owner.remove(); + } + } + }; + return session; +} + +async function completeCredentialTurn(turn: CredentialTurn) { + if (turn.detached || turn.completed) return; + turn.completion ??= turn.copyBack().finally(() => { turn.completed = true; }); + await turn.completion; +} + +/** Flush refreshes after each turn, without closing a healthy warm provider. */ +export async function completeManagedNativeCredentialTurn(session: NativeSession | undefined) { + if (!session) return; + const turn = credentialTurns.get(session); + if (turn) await completeCredentialTurn(turn); +} diff --git a/server/src/services/native-runtime/native-continuation.test.ts b/server/src/services/native-runtime/native-continuation.test.ts index dd52896fd29..cd2a9b8be91 100644 --- a/server/src/services/native-runtime/native-continuation.test.ts +++ b/server/src/services/native-runtime/native-continuation.test.ts @@ -44,3 +44,78 @@ describe("native continuation event projection", () => { expect(build({ ...wake, ...extra })).toBeNull(); }); }); + + +describe("plain-text Slack continuation projection", () => { + const slackComment = { ...message, author: { type: "user", id: "board" } }; + const slackWake = { + ...wake, + reason: "External chat message received", + issue: { id: "task", title: "OLD_EXACT_REPLY", description: "Started from Slack", workMode: "standard" }, + externalChatProvider: "slack", checkedOutByHarness: true, + comments: [slackComment], + commentIds: [message.id], latestCommentId: message.id, + commentWindow: { requestedCount: 1, includedCount: 1, missingCount: 0 }, + attachmentOmissions: [], + }; + const buildChat = (value: unknown = slackWake) => buildNativeContinuationPrompt({ + wakePayload: value, previousRunId: "prior", issue: slackWake.issue, + previousIssue: slackWake.issue, + allowExternalChat: true, + }); + it("sends the new message once without prior titles, history or instructions", () => { + const text = buildChat()!; + expect(text).not.toBeNull(); + expect(JSON.parse(text)).toEqual({ messages: [message] }); + expect(text).not.toContain("OLD_EXACT_REPLY"); + expect(text).not.toContain("OLD_HISTORY"); + expect(text.length).toBeLessThan(600); + }); + it("excludes edited history and prior agent output outside the current Slack delivery", () => { + const text = buildChat({ ...slackWake, executionContinuation: { + ...wake.executionContinuation, + resumeDelta: { baseRunId: "prior", messages: [ + { ...message, id: "old-edited", body: "STALE_DIRECTION" }, + { ...message, id: "old-reply", authorType: "agent", body: "OLD_AGENT_REPLY" }, + message, + ] }, + } }); + expect(JSON.parse(text!).messages).toEqual([message]); + expect(text).not.toContain("STALE_DIRECTION"); + expect(text).not.toContain("OLD_AGENT_REPLY"); + }); + it("preserves an actual task brief edit beside the current message", () => { + const text = buildNativeContinuationPrompt({ + wakePayload: slackWake, previousRunId: "prior", allowExternalChat: true, + issue: { ...slackWake.issue, description: "Updated scope" }, previousIssue: slackWake.issue, + }); + expect(JSON.parse(text!).taskChanges).toEqual({ description: "Updated scope" }); + }); + it("accepts the verified non-assignee chat binding", () => { + expect(buildChat({ ...slackWake, checkedOutByHarness: false, externalChatExecutionBound: true })).not.toBeNull(); + }); + it.each([ + { comments: [{ ...slackComment, attachments: [{ id: "file" }] }] }, + { comments: [{ ...slackComment, attachments: "malformed" }] }, + { comments: [{ ...slackComment, body: "EDITED_SINCE_DELTA" }] }, + { comments: [{ ...slackComment, author: { type: "user", id: "different-user" } }] }, + { attachmentOmissions: [{ commentId: message.id, notice: "Unavailable input" }] }, + { externalChatProvider: "discord" }, + { checkedOutByHarness: false }, + { truncated: true }, + { fallbackFetchNeeded: true }, + { comments: [{ ...slackComment, bodyTruncated: true }] }, + { interactionId: "approval", interactionKind: "request_confirmation" }, + { recovery: { cause: "interrupted" } }, + { executionContinuation: { ...wake.executionContinuation, resumeDelta: { baseRunId: "wrong", messages: [message] } } }, + { executionContinuation: { ...wake.executionContinuation, resumeDelta: { baseRunId: "prior", messages: [] } } }, + { commentIds: [message.id, "missing"], latestCommentId: "missing" }, + { commentIds: [message.id, message.id], latestCommentId: message.id }, + { executionContinuation: { ...wake.executionContinuation, resumeDelta: { baseRunId: "prior", messages: [message, message] } } }, + ])("keeps specialized or incomplete input on the full path: %j", (extra) => { + expect(buildChat({ ...slackWake, ...extra })).toBeNull(); + }); + it("requires the constructor to explicitly opt into chat continuation", () => { + expect(build(slackWake)).toBeNull(); + }); +}); diff --git a/server/src/services/native-runtime/native-continuation.ts b/server/src/services/native-runtime/native-continuation.ts index ee8714aee17..a81c62a4289 100644 --- a/server/src/services/native-runtime/native-continuation.ts +++ b/server/src/services/native-runtime/native-continuation.ts @@ -1,4 +1,4 @@ -import { normalizePaperclipWakePayload } from "@paperclipai/adapter-utils/server-utils"; +import { isPaperclipExternalChatContractTurn, normalizePaperclipWakePayload } from "@paperclipai/adapter-utils/server-utils"; /** Only new, authorized events belong in an already retained provider conversation. * Full bootstrap input is kept separately for an actual provider resume failure. @@ -8,16 +8,47 @@ export function buildNativeContinuationPrompt(input: { previousRunId: string; issue: { title: string; description: string | null }; previousIssue: { title: string; description: string | null }; + /** The constructor opts plain-text chat into the existing resume protocol. */ + allowExternalChat?: boolean; }): string | null { const wake = normalizePaperclipWakePayload(input.wakePayload); const continuation = wake?.executionContinuation; const delta = continuation?.resumeDelta; if (!wake || !delta || delta.baseRunId !== input.previousRunId) return null; - // These paths have specialized delivery/recovery contracts. Preserve their - // existing framing until they have an event-specific continuation projection. - if (!['issue_commented', 'issue_children_completed'].includes(wake.reason ?? '') || + const rawWake = input.wakePayload && typeof input.wakePayload === "object" + ? input.wakePayload as Record : {}; + // Attachment descriptors are intentionally omitted by the generic normalizer. + // Retain the complete bootstrap framing whenever current input needs them or + // an omission notice. The compact path never guesses an unavailable file. + const hasAttachmentContext = + (rawWake.attachmentOmissions != null && + (!Array.isArray(rawWake.attachmentOmissions) || rawWake.attachmentOmissions.length > 0)) || + (Array.isArray(rawWake.comments) && rawWake.comments.some((comment) => + comment && typeof comment === "object" && + (comment as Record).attachments != null && + (!Array.isArray((comment as Record).attachments) || + ((comment as Record).attachments as unknown[]).length > 0))); + const externalChat = input.allowExternalChat === true && + wake.externalChatProvider === "slack" && + isPaperclipExternalChatContractTurn(input.wakePayload) && + !hasAttachmentContext && wake.commentIds.length > 0 && + wake.latestCommentId === wake.commentIds.at(-1) && + new Set(wake.commentIds).size === wake.commentIds.length && + wake.commentIds.every((id) => { + const comments = wake.comments.filter((entry) => entry.id === id); + const messages = delta.messages.filter((entry) => entry.id === id); + return comments.length === 1 && messages.length === 1 && + messages[0]!.body === comments[0]!.body && + messages[0]!.authorType === comments[0]!.authorType && + messages[0]!.authorId === comments[0]!.authorId; + }); + // Other provider, attachment, question, approval, and recovery paths keep + // their specialized framing. A matching prior run still gates every delta. + if ((wake.externalChatProvider && !externalChat) || + (!['issue_commented', 'issue_children_completed'].includes(wake.reason ?? '') && + !(externalChat && wake.reason === "External chat message received")) || wake.fallbackFetchNeeded || wake.truncated || wake.recovery || continuation?.interruptedRunId || - wake.externalChatExecutionBound || wake.externalChatQuestionResponse || + (wake.externalChatExecutionBound && !externalChat) || wake.externalChatQuestionResponse || wake.taskWatchdog || wake.livenessContinuation || wake.activeTreeHold || wake.skillTest || wake.executionStage || wake.agentMessage || wake.documentReviewContext || wake.planReviewContext || wake.annotationDeltas.length > 0 @@ -31,7 +62,12 @@ export function buildNativeContinuationPrompt(input: { // Do not substitute an unverified interaction outcome for an authorized answer. if (wake.interactionId && humanResponses.length === 0) return null; const events = { - messages: delta.messages, + // Only the current authenticated Slack delivery belongs to this turn. + // The generic delta can also include edited historical comments or an old + // agent response; neither becomes a new Slack instruction. + messages: externalChat + ? wake.commentIds.map((id) => delta.messages.find((message) => message.id === id)!) + : delta.messages, ...(humanResponses.length ? { humanResponses } : {}), ...(Object.keys(taskChanges).length ? { taskChanges } : {}), ...(wake.childIssueSummaries.length ? { childResults: wake.childIssueSummaries } : {}), diff --git a/server/src/services/native-runtime/native-execution-input.ts b/server/src/services/native-runtime/native-execution-input.ts index 92d0121010a..fc46200f0df 100644 --- a/server/src/services/native-runtime/native-execution-input.ts +++ b/server/src/services/native-runtime/native-execution-input.ts @@ -175,6 +175,7 @@ export function buildNativeExecutionInput(input: { wakePayload: input.wakePayload, previousRunId: input.previousTurn.runId, previousIssue: input.previousTurn.task, + allowExternalChat: true, issue: input.issue, }), } : {}), diff --git a/server/src/services/native-runtime/native-run-coordinator-store.ts b/server/src/services/native-runtime/native-run-coordinator-store.ts index c29388e8be1..ad9f772023c 100644 --- a/server/src/services/native-runtime/native-run-coordinator-store.ts +++ b/server/src/services/native-runtime/native-run-coordinator-store.ts @@ -6,6 +6,7 @@ import type { Db } from "@paperclipai/db"; import { heartbeatRunEvents, heartbeatRuns, + issues, nativeRunFinalizations, nativeRunResults, } from "@paperclipai/db"; @@ -339,6 +340,11 @@ export class NativeRunCoordinatorStore { }); return this.#db.transaction(async (tx) => { + // Match task -> run lock ordering before the result's task foreign-key + // check; a concurrent task mutation may also need this run row. + await tx.select({ id: issues.id }).from(issues) + .where(and(eq(issues.id, this.#binding.issueId), eq(issues.companyId, this.#binding.companyId))) + .for("key share"); const [run] = await tx .select() .from(heartbeatRuns) 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 4f21c1153e7..e244964226d 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -6388,7 +6388,7 @@ describe("native warm session supervision", () => { }); }); - it("replaces an idle warm provider session when its pinned permission mode changes", async () => { + it.each(["permission", "managed credential"])("replaces an idle warm provider session when its %s changes", async (change) => { const firstClose = vi.fn(async () => undefined); const secondClose = vi.fn(async () => undefined); const firstSession = { close: firstClose }; @@ -6418,7 +6418,7 @@ describe("native warm session supervision", () => { } as unknown as NativeExecutionInputV1; const lowered = { ...base, - provider: { kind: "codex", model: null, approvalPolicy: "on-request" }, + provider: change === "permission" ? { kind: "codex", model: null, approvalPolicy: "on-request" } : base.provider, binding: { ...base.binding, runId: "run-permission-on-request" }, } as NativeExecutionInputV1; const result = { @@ -6453,11 +6453,13 @@ describe("native warm session supervision", () => { await executePaperclipNativeSession({ db: leaseDb(base), execution: base, + managedAiCredentialIdentity: "first-identity", runnerInstanceId: "runner", }); await executePaperclipNativeSession({ db: leaseDb(lowered), execution: lowered, + managedAiCredentialIdentity: change === "managed credential" ? "second-identity" : "first-identity", runnerInstanceId: "runner", }); expect(firstClose).toHaveBeenCalledWith({ @@ -7354,7 +7356,7 @@ describe("runnerd provider runtime wiring", () => { } expect(detach).toHaveBeenCalledOnce(); - expect(close).toHaveBeenCalledOnce(); + expect(close).toHaveBeenCalledTimes(timing === "during-close" ? 1 : 0); expect(state.copyBackCodexAuth).not.toHaveBeenCalled(); await expect(readFile(authPath, "utf8")).resolves.toBe(auth); }); @@ -9996,6 +9998,7 @@ describe("runnerd provider runtime wiring", () => { const sessionRoot = join(runtimeRoot, "sessions", createHash("sha256").update(execution.session.normalizedSessionId!).digest("hex")); if (scenario === "existing_state") await mkdir(sessionRoot, { recursive: true }); if (scenario === "symlink_parent") await symlink(isolatedStateDirectory, join(runtimeRoot, "sessions")); + const syncIn = vi.fn(async () => undefined); const remoteExecute = vi.fn(async (command: { command: string; args?: string[] }) => { if (command.args?.[2] === "paperclip-runner-claim-unstarted-session") { let exitCode = 1; @@ -10015,26 +10018,32 @@ describe("runnerd provider runtime wiring", () => { if (command.args?.[0] === "--version") return { exitCode: 0, timedOut: false, stdout: "codex-cli 0.153.4", stderr: "", }; + if (command.args?.[2] === "paperclip-runner-launch") { + throw new Error("fixture_stop_after_launch_staging"); + } if (command.args?.[1]?.includes("base64")) return { exitCode: 1, timedOut: false, stdout: "", stderr: "", }; return { exitCode: 0, timedOut: false, stdout: "", stderr: "" }; }); + const runtimeContext = nativeRuntimeContextFixture(); + const executionWithContext = { ...execution, runtimeContext }; const backend = await createRunnerdBackend({ - db: leaseDb(execution), execution, runnerInstanceId: "runner-new-in-retained-workspace", + db: leaseDb(execution), execution: executionWithContext, runnerInstanceId: "runner-new-in-retained-workspace", runnerIngressAuthorized: true, runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd, environmentId: "environment", leaseId: "lease-resumed", providerKey: "daytona", effectiveCapabilities: { runnerWebSocketIngress: true }, sandboxLeaseAcquisition: { outcome: "resumed", providerLeaseId: "sandbox-retained" }, - runner: { execute: remoteExecute }, + runner: { execute: remoteExecute, syncIn }, } as never, }); expect(backend).toBeDefined(); state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!(); const options = state.createTransport.mock.calls.at(-1)![0] as RunnerTransportOptions & { prepareExternalRunnerState: () => Promise; + runnerProcessLauncher: ReturnType; }; await mkdir(join(options.stateDirectory!, "control-plane"), { recursive: true }); await writeFile(join(options.stateDirectory!, "control-plane", "control-plane-state.json"), JSON.stringify({ @@ -10049,23 +10058,52 @@ describe("runnerd provider runtime wiring", () => { } if (scenario === "fresh") { await expect(options.prepareExternalRunnerState()).resolves.toBeUndefined(); - expect(remoteExecute.mock.calls.some(([command]) => command.args?.[1]?.includes("install -d"))).toBe(true); + expect(remoteExecute.mock.calls.some(([command]) => command.args?.[1]?.includes("install -d"))).toBe(false); + expect(syncIn).not.toHaveBeenCalled(); const claimCommand = remoteExecute.mock.calls.find(([command]) => command.args?.[2] === "paperclip-runner-claim-unstarted-session")![0]; expect((await lstat(sessionRoot)).mode & 0o777).toBe(0o700); // The exact same claim cannot silently reopen an existing partial root. expect(() => execFileSync("sh", claimCommand.args!, { stdio: "pipe" })).toThrow(); + // Authority rotation prepares history before Codex's resume path writes + // the current invocation's launch files. The actual launch must stage + // those files without trying to claim/restore the session a second time. + const home = join(options.stateDirectory!, "codex-home"); + await mkdir(home, { recursive: true }); + await writeFile(join(home, "auth.json"), "fixture-current-credential"); + await writeFile(join(home, "config.toml"), "fixture-current-config"); + await writeFile(join(options.stateDirectory!, "runtime-context.json"), JSON.stringify(runtimeContext)); + syncIn.mockClear(); + await expect(options.prepareExternalRunnerState()).resolves.toBeUndefined(); + expect(syncIn).not.toHaveBeenCalled(); + const launched = options.runnerProcessLauncher({ + command: "/controller/paperclip-runnerd", args: [], cwd: "/controller", environment: {}, + }); + await expect(launched.completion).rejects.toThrow("fixture_stop_after_launch_staging"); + for (const name of ["auth.json", "config.toml"]) { + expect(syncIn).toHaveBeenCalledWith([expect.objectContaining({ + files: [expect.objectContaining({ sourcePath: join(home, name), kind: "file", mode: 0o600 })], + })]); + } + expect(syncIn.mock.calls.flat(2).flatMap((entry: { files: Array<{ sourcePath: string }> }) => entry.files) + .filter((file: { sourcePath: string }) => file.sourcePath === runtimeContext.instructions.bundle.rootPath)).toHaveLength(1); + expect(syncIn).toHaveBeenCalledWith([expect.objectContaining({ + files: [expect.objectContaining({ sourcePath: join(options.stateDirectory!, "runtime-context.json"), kind: "file" })], + })]); + expect(remoteExecute.mock.calls.filter(([command]) => command.args?.[2] === "paperclip-runner-claim-unstarted-session")).toHaveLength(1); } else { await expect(options.prepareExternalRunnerState()).rejects.toThrow("runner_harness_state_mismatch"); expect(remoteExecute.mock.calls.some(([command]) => command.args?.[1]?.includes("install -d"))).toBe(false); } }); - it.each(["current", "stale", "missing"])("uses shared Codex and the server-owned replacement artifact (image=%s)", async (image) => { - const needsReplacement = image !== "current"; + it.each(["current", "stale", "missing", "retained", "retained-mismatch", "retained-error", "retained-timeout", "retained-explicit"])("uses shared Codex and the server-owned replacement artifact (image=%s)", async (image) => { + const retained = image.startsWith("retained"); + const exactRetained = image === "retained" || image === "retained-explicit"; + const needsReplacement = image !== "current" && !exactRetained; // The mocked remote executes metadata probes; artifact staging only needs bytes. // Keep this regression independent of a locally compiled Rust runner binary. const controllerArtifact = join(isolatedStateDirectory, "paperclip-runnerd"); - if (needsReplacement) { + if (needsReplacement || retained) { await writeFile(controllerArtifact, "fixture runner artifact"); state.resolveRunnerBinary.mockReturnValueOnce(controllerArtifact); } @@ -10074,13 +10112,19 @@ describe("runnerd provider runtime wiring", () => { async (command: { command: string; args?: string[] }) => { let stdout = ""; const script = command.args?.[1] ?? ""; - if (command.args?.[0] === "--build-metadata") { + if (script.includes('sha256sum "$1"')) { + if (image === "retained-error") throw new Error("checksum unavailable"); + return { + exitCode: 0, signal: null, timedOut: image === "retained-timeout", stderr: "", + stdout: `${createHash("sha256").update(image === "retained-mismatch" ? "stale artifact" : "fixture runner artifact").digest("hex")} ${command.args?.[3]}\n`, + }; + } else if (command.args?.[0] === "--build-metadata") { stdout = JSON.stringify({ schema: "paperclip-runner/runnerd-build-metadata/v1", binaryName: "paperclip-runnerd", packageName: "@paperclipai/paperclip-runner", binaryContractVersion: 2, - durableSessionCapabilities: image === "stale" && command.command === "/usr/local/bin/paperclip-runnerd" + durableSessionCapabilities: (image === "stale" || retained) && command.command === "/usr/local/bin/paperclip-runnerd" ? undefined : ["unlimited_runtime", "connection_lease_renewal"], prpTransportModes: ["listen_ws"], @@ -10121,6 +10165,7 @@ describe("runnerd provider runtime wiring", () => { db: leaseDb(execution), execution, runnerInstanceId: "runner-image-runtime", + ...(image === "retained-explicit" ? { runnerRemoteBinaryPath: controllerArtifact } : {}), runnerIngressAuthorized: true, runnerExecutionTarget: { kind: "remote", @@ -10128,6 +10173,7 @@ describe("runnerd provider runtime wiring", () => { remoteCwd: "/workspace", environmentId: "environment", leaseId: "lease", + ...(retained ? { sandboxLeaseAcquisition: { outcome: "resumed" }, reusableLeaseConfigured: true } : {}), providerKey: "daytona", effectiveCapabilities: { runnerWebSocketIngress: true }, runner: { execute: remoteExecute, syncIn }, @@ -10154,6 +10200,13 @@ describe("runnerd provider runtime wiring", () => { } else { expect(syncIn).not.toHaveBeenCalled(); } + if (exactRetained) { + expect(remoteExecute.mock.calls.some(([call]) => call.args?.[1]?.includes("command -v paperclip-runnerd"))).toBe(false); + expect(remoteExecute).toHaveBeenCalledWith(expect.objectContaining({ + command: "/workspace/.paperclip-runtime/paperclip-runner/bin/paperclip-runnerd", + args: ["--build-metadata"], + })); + } expect(remoteExecute).toHaveBeenCalledWith( expect.objectContaining({ command: "/opt/paperclip-runner/bin/codex", diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 6d1a3cbf05e..5abfbe1b55f 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -1,3 +1,4 @@ +import { bindManagedNativeCredentialTurn, completeManagedNativeCredentialTurn } from "./managed-native-credentials.js"; import { createLocalNativeQuestionBridge } from "./local-native-question-bridge.js"; import { readVerifiedRemoteWorkspaceFile } from "./remote-deliverable-file.js"; import { copyBackCodexAuth } from "@paperclipai/adapter-codex-local/server"; @@ -368,6 +369,7 @@ function clearNativeRuntimeRequestResolutions(runId: string): void { } type WarmNativeSession = { + managedAiCredentialIdentity?: string; credentialRunId?: string; githubAuthenticationMode?: string; networkAccess: boolean; @@ -6933,6 +6935,7 @@ export async function executePaperclipNativeSession(input: { runnerEnvironment?: NodeJS.ProcessEnv; /** Private grant materialization; never a user-configured host path. */ managedAiCredentialHome?: string; + managedAiCredentialIdentity?: string; runnerExecutionTarget?: AdapterExecutionTarget | null; /** Resolved per-run authorization; not an independent instance setting. */ runnerIngressAuthorized?: boolean; @@ -7771,6 +7774,7 @@ async function executePaperclipNativeSessionWithinScope( `native-warm-session:${input.execution.binding.runId}`, ); let existingWarmSession: NativeSession | undefined; + let managedCredentialSession: NativeSession | undefined; let persistedWarmSession: PersistedNativeSession | null | undefined; if (warmSessionId !== null && warmConfigDigest !== null) { const entry = warmNativeSessions.get(warmSessionId); @@ -7786,6 +7790,7 @@ async function executePaperclipNativeSessionWithinScope( entry.credentialRunId !== input.execution.binding.runId); if ( entry.configDigest !== warmConfigDigest || + entry.managedAiCredentialIdentity !== input.managedAiCredentialIdentity || credentialRunChanged || entry.githubAuthenticationMode !== input.runnerEnvironment?.PAPERCLIP_GITHUB_AUTH_MODE || @@ -8087,6 +8092,9 @@ async function executePaperclipNativeSessionWithinScope( ); }, onSession: async (session) => { + if (session && input.managedAiCredentialHome) { + managedCredentialSession = runnerdBackend?.bindManagedSession(session) ?? session; + } releaseRegisteredGoalController(); liveQuestions.close(); if (session?.goal) { @@ -8117,6 +8125,7 @@ async function executePaperclipNativeSessionWithinScope( existing.session = session; } else warmNativeSessions.set(warmSessionId, { + managedAiCredentialIdentity: input.managedAiCredentialIdentity, githubAuthenticationMode: input.runnerEnvironment?.PAPERCLIP_GITHUB_AUTH_MODE, networkAccess: @@ -8175,6 +8184,16 @@ async function executePaperclipNativeSessionWithinScope( }, { parentName: "task.run" }, ); + try { + await completeManagedNativeCredentialTurn(managedCredentialSession); + } catch { + // A durable result remains successful if optional credential refresh + // fails. Retire this owner so it cannot keep stale auth on a warm turn. + if (warmSessionId !== null && lifecyclePolicy.mode === "warm") { + await releaseWarmNativeSession(warmSessionId, warmSessionOwnerToken, lifecyclePolicy.idleTimeoutMs, true); + } + await input.onLog?.("stderr", "[paperclip-runner] managed credential refresh failed; provider session retired.\n"); + } if (native.terminal.runTerminalState === "succeeded") { // A truncated, verified external-chat wake cannot settle from the // provider's partial inline prompt. The run-scoped reader records a @@ -10132,7 +10151,7 @@ export async function createRunnerdBackend(input: { contextSnapshot: Record; }, ) => Promise; -}): Promise { +}): Promise { const sessionScopeId = nativeSessionScopeKey(input.execution); const scopeOwner = executingRunnerdSessionScopes.get(sessionScopeId); if (scopeOwner && scopeOwner !== input.execution.binding.runId) { @@ -10181,7 +10200,7 @@ async function createRunnerdBackendWithinSessionClaim( input: Parameters[0], sessionScopeId: string, retainedTransition?: VerifiedWarmTransitionBinding, -): Promise { +): Promise { let recoveryPending = retainedTransition !== undefined; const target = input.runnerExecutionTarget ?? { kind: "local" as const }; const remoteTarget = target.kind === "remote" ? target : null; @@ -10594,9 +10613,47 @@ async function createRunnerdBackendWithinSessionClaim( remotePrepared = true; return; } - let usedPreinstalledRunner = false; + // A resumed sandbox may already contain the exact controller-owned binary. + // Do not upload it every turn, but never treat compatible metadata alone as + // proof of artifact identity (especially for an explicit operator override). + let runnerArtifactPrepared = false; + if ( + sandboxLeaseAcquisition?.outcome === "resumed" && + existsSync(controllerRunnerBinary) + ) { + runnerArtifactPrepared = await measureNativeRunnerSpan( + input.trace, + "runner.artifact.verify_retained", + async () => { + const expected = createHash("sha256") + .update(readFileSync(controllerRunnerBinary)) + .digest("hex"); + try { + const probe = await remoteCommandRunner.execute({ + command: "sh", + args: [ + "-c", + 'test -x "$1" || exit 1; if command -v sha256sum >/dev/null 2>&1; then sha256sum "$1"; else shasum -a 256 "$1"; fi', + "paperclip-runner-artifact", + remoteBinary, + ], + cwd: remoteTarget.remoteCwd, + bypassSession: true, + timeoutMs: 10_000, + }); + return probe.exitCode === 0 && !probe.timedOut && + /^[a-f0-9]{64}\s/.test(probe.stdout) && + probe.stdout.trim().split(/\s+/)[0] === expected; + } catch { + // Missing binaries, checksum tools, or a failed probe use the + // ordinary verified staging path; none authorizes cached execution. + return false; + } + }, + ); + } const explicitRemoteBinary = input.runnerRemoteBinaryPath?.trim() || null; - if (mayUsePreinstalledRunnerArtifact(explicitRemoteBinary)) { + if (!runnerArtifactPrepared && mayUsePreinstalledRunnerArtifact(explicitRemoteBinary)) { const preinstalledRunner = await measureNativeRunnerSpan( input.trace, "runner.artifact.discover", @@ -10614,17 +10671,17 @@ async function createRunnerdBackendWithinSessionClaim( "runner.artifact.link", () => linkPreinstalledExecutable(preinstalledRunner, remoteBinary), ); - usedPreinstalledRunner = true; + runnerArtifactPrepared = true; await input.onLog?.( "stderr", "[paperclip-runner] using preinstalled runnerd from the sandbox image\n", ); } catch { - usedPreinstalledRunner = false; + runnerArtifactPrepared = false; } } } - if (!usedPreinstalledRunner) { + if (!runnerArtifactPrepared) { // Upload the same artifact used for the controller identity. The server // vendors the runner under vendor/paperclip-runner/bin, so the package // development fallback cannot locate it in a deployed server. @@ -11187,7 +11244,7 @@ async function createRunnerdBackendWithinSessionClaim( } }; - const ensureRemoteRunner = async () => { + const ensureRemoteRunner = async (stageLaunchAssets = true) => { await measureNativeRunnerSpan( input.trace, "stage.sync", @@ -11339,7 +11396,6 @@ async function createRunnerdBackendWithinSessionClaim( throw new Error("runner_harness_state_mismatch"); } } - await materializeRemoteHarnessLaunchState(); } else if (remoteTarget && remoteCommandRunner) { // Local and generic SSH execution retain their existing checkpoint // behavior. The manifest-only failover gate applies to managed sandbox @@ -11382,6 +11438,15 @@ async function createRunnerdBackendWithinSessionClaim( ); remoteHarnessStatePrepared = true; } + // Authority rotation needs durable history, before the transport writes + // this turn's launch files. Stage assets only at the actual launch so we + // neither upload stale context nor transfer every bundle twice on resume. + if (!stageLaunchAssets) return; + // Resume can prepare/rotate durable state before the transport creates + // this invocation's isolated Codex auth/config. The later launch must + // still stage those fresh files even when history was already restored. + // Launch material is never recovered from a failover backup. + await materializeRemoteHarnessLaunchState(); if ( remoteTarget && remoteCommandRunner && @@ -11632,7 +11697,7 @@ async function createRunnerdBackendWithinSessionClaim( target: remoteTarget, runnerIngressAuthorized: input.runnerIngressAuthorized === true, }); - await ensureRemoteRunner(); + await ensureRemoteRunner(false); } : undefined; const archiveExternalRunnerState = @@ -12284,27 +12349,14 @@ async function createRunnerdBackendWithinSessionClaim( }, }).transport, }); + const boundManagedSessions = new WeakSet(); const wrapManagedSession = (session: NativeSession): NativeSession => { - if (!input.managedAiCredentialHome || input.execution.provider.kind !== "codex") return session; - const close = session.close.bind(session); - const detach = session.detachControllerForRestart?.bind(session); - let detachedForRestart = false; - if (detach) { - session.detachControllerForRestart = async () => { - // Ownership is relinquished before the asynchronous detach completes. - // Late close finalizers must leave the live runner's credentials alone. - detachedForRestart = true; - await detach(); - }; - } - let copied = false; - session.close = async (closeInput) => { - await close(closeInput); - if (detachedForRestart || copied) return; - copied = true; - const remoteAuth = remoteRunnerFilesystemRoot ? posix.join(remoteRunnerFilesystemRoot, "codex-home", "auth.json") : null; - const localAuth = join(root, "codex-home", "auth.json"); - try { + if (!input.managedAiCredentialHome || input.execution.provider.kind !== "codex" || boundManagedSessions.has(session)) return session; + boundManagedSessions.add(session); + const remoteAuth = remoteRunnerFilesystemRoot ? posix.join(remoteRunnerFilesystemRoot, "codex-home", "auth.json") : null; + const localAuth = join(root, "codex-home", "auth.json"); + return bindManagedNativeCredentialTurn(session, { + copyBack: async () => { await copyBackCodexAuth({ hostAuthPath: join(input.managedAiCredentialHome!, "auth.json"), readSandboxAuth: async () => { @@ -12315,12 +12367,12 @@ async function createRunnerdBackendWithinSessionClaim( }, log: () => {}, }); - } finally { + }, + remove: async () => { rmSync(localAuth, { force: true }); if (remoteAuth && remoteCommandRunner) await remoteCommandRunner.execute({ command: "rm", args: ["-f", "--", remoteAuth], bypassSession: true, timeoutMs: 10000 }); - } - }; - return session; + }, + }); }; const priorAuthorityEpoch = sessionToolAuthorityEpochs.get(sessionScopeId); if (priorAuthorityEpoch && priorAuthorityEpoch !== authorityEpoch) { @@ -12328,6 +12380,7 @@ async function createRunnerdBackendWithinSessionClaim( } sessionToolAuthorityEpochs.set(sessionScopeId, authorityEpoch); return { + bindManagedSession: wrapManagedSession, descriptor: () => backend.descriptor(), openSession: async (sessionInput) => wrapManagedSession(await backend.openSession(sessionInput)), recoverSession: async (snapshot, options) => { @@ -12343,5 +12396,5 @@ async function createRunnerdBackendWithinSessionClaim( ); return wrapManagedSession(await backend.openSession(sessionInput)); }, - } satisfies NativeSessionBackend; + } satisfies NativeSessionBackend & { bindManagedSession(session: NativeSession): NativeSession }; } 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 beb81064384..47b564298d0 100644 --- a/server/src/services/native-runtime/native-session-resume.test.ts +++ b/server/src/services/native-runtime/native-session-resume.test.ts @@ -2288,6 +2288,32 @@ describe("rebindNativeSessionCheckpoint", () => { }); describe("buildNativeExecutionInput wake projection", () => { + it("uses compact Slack input only after provider resume, preserving fresh bootstrap fallback", () => { + const base = execution(currentRunId); + const issue = { id: issueId, identifier: "DOT-2", title: "Old greeting", description: null, workMode: "standard" }; + const message = { id: "wake", body: "NEW_SLACK_MESSAGE", authorType: "user", authorId: "board", sourceTrust: "user", createdByRunId: null }; + const input = buildNativeExecutionInput({ + companyId, runId: currentRunId, agentId, issue, + taskPrompt: "FULL_BOOTSTRAP_WITH_FILE_AND_AUTHORIZATION_INSTRUCTIONS", + resumedSession: true, + previousTurn: { runId: previousRunId, task: issue }, + wakePayload: { + reason: "External chat message received", issue, + externalChatProvider: "slack", checkedOutByHarness: true, + comments: [{ ...message, author: { type: "user", id: "board" } }], commentIds: [message.id], latestCommentId: message.id, + commentWindow: { requestedCount: 1, includedCount: 1, missingCount: 0 }, + executionContinuation: { version: 1, objective: message.body, messages: [message], resumeDelta: { baseRunId: previousRunId, messages: [message] } }, + }, + workspace: { id: currentRunId, ...base.workspace }, normalizedSessionId, + provider: "codex", completionContract: base.completionContract, + runtimeContext: nativeRuntimeContextFixture(), + }); + expect(input.continuationPrompt).toContain(message.body); + expect(input.continuationPrompt).not.toContain("FULL_BOOTSTRAP"); + expect(input.task.prompt).toContain("FULL_BOOTSTRAP"); + expect(input.task.prompt).toContain(message.body); + }); + it("uses a neutral turn title for authenticated external-chat follow-ups", () => { const staleRootTitle = "Reply with exactly STALE-ROOT-MARKER"; const input = buildNativeExecutionInput({ diff --git a/server/src/services/native-runtime/native-session-resume.ts b/server/src/services/native-runtime/native-session-resume.ts index 44d9e3c6c77..08182f652ba 100644 --- a/server/src/services/native-runtime/native-session-resume.ts +++ b/server/src/services/native-runtime/native-session-resume.ts @@ -219,6 +219,7 @@ export async function findNativeSessionResumeRun( processGroupId: heartbeatRuns.processGroupId, processStartedAt: heartbeatRuns.processStartedAt, runnerProfileJson: heartbeatRuns.runnerProfileJson, + contextSnapshot: heartbeatRuns.contextSnapshot, }) .from(heartbeatRuns) .where( 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 9d997f6ffda..bb16da33b0c 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 @@ -1,5 +1,6 @@ import { afterAll, beforeAll, describe, expect, it } from "vitest"; -import { asc, eq } from "drizzle-orm"; +import { asc, eq, sql } from "drizzle-orm"; +import { randomUUID } from "node:crypto"; import { renderPaperclipWakePrompt } from "@paperclipai/adapter-utils/server-utils"; import { readCompletedAssistantMessageCandidate, resolveHeartbeatRunResponse, selectHeartbeatRunFinalAgentMessage } from "../heartbeat-run-summary.js"; import { @@ -38,6 +39,7 @@ import { } from "../../vendor/paperclip-runner/testing.js"; import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js"; +import { NativeRunCoordinatorStore } from "./native-run-coordinator-store.js"; import { finalizeNativeRun } from "./native-run-finalizer.js"; import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js"; import { issueThreadInteractionService } from "../issue-thread-interactions.js"; @@ -957,6 +959,49 @@ describe("PaperclipControlPlanePort conformance", () => { await db.update(heartbeatRuns).set({ runnerInstanceId }).where(eq(heartbeatRuns.id, runId)); }); + it.each(["port", "coordinator"])("does not deadlock %s result persistence against a task mutation that updates its run", async (kind) => { + const identity = CONTROL_PLANE_CONFORMANCE_OPEN.identity; + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: runId, companyId: identity.companyId, agentId: identity.agentId, + status: "running", runtimeMode: "native", nativeIssueId: identity.issueId, + nativeSessionId: identity.sessionId, runnerInstanceId: conformanceRunnerId, + completionContractId: contractId, completionContractSha256: contractSha, + }); + const port = kind === "port" ? new PaperclipControlPlanePort(db, { + ...identity, runId, completionContractId: contractId, completionContractSha256: contractSha, + sourceInstanceId: conformanceRunnerId, controlPlaneSourceInstanceId: "result-lock-order", + }) : new NativeRunCoordinatorStore(db, { + ...identity, runId, completionContractId: contractId, completionContractSha256: contractSha, + normalizedSessionId: identity.sessionId, runnerSourceInstanceId: conformanceRunnerId, + completionContractRevision: "standalone-v1", completionContractCriterionIds: ["objective"], + }); + let completion: Promise | undefined; + try { + await db.transaction(async (tx) => { + await tx.select().from(issues).where(eq(issues.id, identity.issueId)).for("update"); + const [backend] = await tx.execute(sql`select pg_backend_pid() as pid`) as unknown as Array<{ pid: number }>; + completion = port.completeRun({ + result: CONTROL_PLANE_CONFORMANCE_RESULT, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL, + callerResultId: "result-lock-order", + }).then(() => null, (error: unknown) => 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); + await tx.execute(sql`set local lock_timeout = '1s'`); + await tx.update(heartbeatRuns).set({ updatedAt: new Date() }).where(eq(heartbeatRuns.id, runId)); + }); + } finally { + expect(await completion).toBeNull(); + } + }); + it("does not let native completion bypass a pending issue interaction", async () => { const identity = CONTROL_PLANE_CONFORMANCE_OPEN.identity; const port = new PaperclipControlPlanePort(db, { 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 a0a87735f68..328d6098e88 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.ts @@ -3,6 +3,7 @@ import type { Db } from "@paperclipai/db"; import { heartbeatRunEvents, heartbeatRuns, + issues, nativeRunFinalizations, nativeRunResults, } from "@paperclipai/db"; @@ -288,6 +289,12 @@ export class PaperclipControlPlanePort implements ControlPlanePort { }); await this.#db.transaction(async (tx) => { + // Result insertion checks the task foreign key. Acquire that parent lock + // before the run, matching task mutations that subsequently update a run. + // Otherwise concurrent chat/status writes can deadlock after generation. + await tx.select({ id: issues.id }).from(issues) + .where(and(eq(issues.id, this.#binding.issueId), eq(issues.companyId, this.#binding.companyId))) + .for("key share"); const run = await tx.select().from(heartbeatRuns) .where(eq(heartbeatRuns.id, this.#binding.runId)).for("update").limit(1) .then((rows) => rows[0] ?? null);