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);