diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 61bffc56443..3546241c639 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1377,6 +1377,16 @@ agent, workspace, retained session, and termination of prior execution. The checkpoint must match the exact Runner, lease, run, session, turn, and item. A missing restart claim does not remove these checks or authorize a fresh session. +After a plugin session passes provider recovery and durable run admission, its +local suspension receipt is removed. The running session then owns authority; +the old receipt must not conflict with a later warm run's journal identity. +Failed admission retains the receipt. Normal suspension writes a new receipt. + +A repository-free fresh plugin task directory remains the task's directory on +ordinary follow-ups. Reuse requires its exact company, task, workspace, provider +owner, fresh-attempt identity and path. Mode-only isolation settings do not turn +that directory into a Git worktree. Explicit strategy settings still apply. + A prepared remote workspace interrupted before the executor's first durable claim may resume its untouched bootstrap on the same run. This requires attempt and controller generation zero, no prior owner, and no process, provider session, diff --git a/maps/agent-access-map/runs-adapters.md b/maps/agent-access-map/runs-adapters.md index d5892c37ec5..286665f80e5 100644 --- a/maps/agent-access-map/runs-adapters.md +++ b/maps/agent-access-map/runs-adapters.md @@ -271,6 +271,15 @@ session-control access. Deployed diagnostic usefulness remains unverified. ## `session` +Plugin suspension receipts remain required through recovery and admission, then +are removed when the admitted session becomes live. A later warm run therefore +does not inherit an obsolete stopped-run identity. Repository-free fresh plugin +directories retain their saved company, task, workspace, provider owner and +attempt bindings on follow-up. This does not grant session-reset authority or +discard an explicit repository strategy. See the +[session recipe](../feature-map/runs-adapters.md#session) for the continuation +checks; source regressions do not establish live provider acceptance. + Scoped source evidence: completed warm-runner retirement can persist maintenance under the same process's live, task-bound preparation lease before native coordinator admission. Original controller generation, committed-result and diff --git a/maps/feature-map/runs-adapters.md b/maps/feature-map/runs-adapters.md index 5d86501621a..104aee6ae38 100644 --- a/maps/feature-map/runs-adapters.md +++ b/maps/feature-map/runs-adapters.md @@ -74,6 +74,15 @@ mutations. Qualify a useful model task separately from saving the account. ### `session` +For a plugin computer, complete a task, let its warm session suspend, then send +two follow-ups within the next warm window. Verify the first resumes the saved +session and the second keeps the same provider history without an identity +mismatch. After a fresh recovery into a repository-free directory, send another +ordinary follow-up. Verify it retains that directory without Core Git worktree +preparation. Explicit repository strategies and changed provider owners must +remain denied. Executor and workspace-preparation regressions cover control +behavior; they do not establish live provider acceptance. + For a remote OpenCode agent configured with `per_turn`, complete a task, then send two follow-ups without resetting its session. Verify that each continuation restores its external controller checkpoint, retains provider context and saved diff --git a/server/src/services/heartbeat/environment-selection.ts b/server/src/services/heartbeat/environment-selection.ts index 484a6937907..41ba6e44d20 100644 --- a/server/src/services/heartbeat/environment-selection.ts +++ b/server/src/services/heartbeat/environment-selection.ts @@ -19,6 +19,7 @@ import { import { assertTaskWorkspaceExecutionAccess, assertTaskWorkspaceSourceProjectAccess } from "../task-workspace-access.js"; import { isBoundIsolatedTaskDirectory, shouldUseIsolatedTaskDirectory } from "../isolated-task-directory.js"; import { readTaskDirectoryRetention } from "../isolated-task-directory-retention.js"; +import { isBoundFreshPluginTaskDirectory } from "../provider-fresh-workspace.js"; import { findNativeChatWorkspaceScope, nativeChatWorkspaceCwd, @@ -478,15 +479,23 @@ export async function selectHeartbeatEnvironment(db: Db, input: HeartbeatEnviron preserveImplicitDefaultSettings: ["project", "task_default"].includes(taskWorkspaceIntent.source) }); } const explicitTaskDirectory = !isDotRun && taskWorkspaceIntent?.selection.kind === "task_directory"; - const explicitDirectoryStrategies = [issueAssigneeOverrides?.adapterConfig?.workspaceStrategy, + const explicitDirectoryStrategies = [ + ...(reusableExistingExecutionWorkspace?.providerType === "adapter_managed" ? [config.workspaceStrategy] : []), + issueAssigneeOverrides?.adapterConfig?.workspaceStrategy, projectExecutionWorkspacePolicy?.workspaceStrategy, issueExecutionWorkspaceSettings?.workspaceStrategy]; const boundTaskDirectory = !isDotRun && !fresh && !nativeRecoveryExecutionWorkspaceId && issueId !== null && issueRef !== null && requestedShouldReuseExisting && requestedExecutionWorkspaceMode === "isolated_workspace" && !workspaceProjectContext?.hasWorkspace && !issueRef.projectWorkspaceId && taskWorkspaceIntent?.selection.kind !== "configured_source" && (taskWorkspaceIntent?.selection.kind !== "existing" || taskWorkspaceIntent.selection.workspaceId === issueRef.executionWorkspaceId) && - await isBoundIsolatedTaskDirectory({ companyId: agent.companyId, issueId, - workspaceId: issueRef.executionWorkspaceId, workspace: reusableExistingExecutionWorkspace }); + (await isBoundIsolatedTaskDirectory({ companyId: agent.companyId, issueId, + workspaceId: issueRef.executionWorkspaceId, workspace: reusableExistingExecutionWorkspace }) || + (selectedEnvironmentForConfig?.driver === "plugin" && agent.adapterType === "paperclip_runner" && + isBoundFreshPluginTaskDirectory({ companyId: agent.companyId, issueId, + workspaceId: issueRef.executionWorkspaceId, workspace: reusableExistingExecutionWorkspace, + owner: { agentId: agent.id, environmentId: selectedEnvironmentForConfig.id, + pluginKey: parseObject(selectedEnvironmentForConfig.config).pluginKey ?? null, + driverKey: parseObject(selectedEnvironmentForConfig.config).driverKey ?? null } }))); const reuseTaskDirectory = boundTaskDirectory && explicitDirectoryStrategies.every(value => { const strategy = parseObject(value); return Object.keys(strategy).length === 0 || (Object.keys(strategy).length === 1 && strategy.type === "project_primary"); diff --git a/server/src/services/heartbeat/workspace-preparation.test.ts b/server/src/services/heartbeat/workspace-preparation.test.ts index 6192dccbae3..59b53740e79 100644 --- a/server/src/services/heartbeat/workspace-preparation.test.ts +++ b/server/src/services/heartbeat/workspace-preparation.test.ts @@ -37,6 +37,7 @@ import { resolveExecutionWorkspaceMode, buildExecutionWorkspaceAdapterConfig } f import { createLocalDiskStorageProvider } from "../../storage/local-disk-provider.js"; import { createStorageService } from "../../storage/service.js"; import { InMemorySpanExporter, SimpleSpanProcessor, TracerProvider } from "@opentelemetry/sdk-trace"; +import { selectHeartbeatEnvironment } from "./environment-selection.js"; const execFile = promisify(execFileCallback); const externalDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_WORKSPACE_PREPARATION_TEST_DATABASE_URL"); @@ -230,6 +231,44 @@ describe.skipIf(!support.supported && !externalDatabaseUrl)("heartbeat workspace const [task] = await db.select().from(issues).where(eq(issues.id, issue.id)); expect(task.projectWorkspaceId).toBe(project ? issue.projectWorkspaceId : null); }); + it.each(["implicit", "task", "agent"] as const)("keeps provider directory selection after a fresh task succeeds (strategy: %s)", async strategySource => { + const { input, issue, envOrchestrator } = await fixture(); + await db.update(issues).set({ projectId: null, projectWorkspaceId: null }).where(eq(issues.id, issue.id)); + input.task.issueRef = { ...issue, projectId: null, projectWorkspaceId: null }; + input.task.context = { issueId: issue.id }; + const pluginKey = "test.fresh-followup"; + await db.insert(plugins).values({ pluginKey, packageName: "test-tasks", version: "1.0.0", apiVersion: 1, + status: "ready", categories: ["workspace"], manifestJson: { id: pluginKey, apiVersion: 1, version: "1.0.0", + displayName: "Tasks", description: "Tasks", author: "Test", categories: ["workspace"], + capabilities: ["environment.drivers.register"], entrypoints: { worker: "./worker.js" }, + environmentDrivers: [{ driverKey: "tasks", displayName: "Task VM", configSchema: { type: "object" }, supportsTasks: true, freshWorkspaceVersion: 1 }] } }); + const environment = await environmentService(db).create({ name: `Task computer ${issue.id}`, driver: "plugin", status: "active", envVars: {}, config: { pluginKey, driverKey: "tasks" } }); + input.agent = { ...input.agent, adapterType: "paperclip_runner", defaultEnvironmentId: environment.id }; + input.environment.selectedEnvironmentForConfig = environment; + input.policy.requestedExecutionWorkspaceMode = input.policy.effectiveExecutionWorkspaceMode = "isolated_workspace"; + input.task.freshWorkspace = { mode: "fresh_isolated_workspace", retryIntentId: input.run.id, sourceRunId: input.run.id, retainedSourceIds: [] }; + const stopped = new Error("Provider directory prepared"); + vi.spyOn(envOrchestrator, "acquireForRun").mockRejectedValue(stopped); + await expect(prepareHeartbeatWorkspace(db, input)).rejects.toBe(stopped); + const [task] = await db.select().from(issues).where(eq(issues.id, issue.id)); + expect(task.executionWorkspaceSettings).toEqual({ mode: "isolated_workspace" }); + const settings = { mode: "isolated_workspace" as const, ...(strategySource === "task" ? { workspaceStrategy: { type: "git_worktree" as const } } : {}) }; + const agentConfig = strategySource === "agent" ? { workspaceStrategy: { type: "git_worktree" as const } } : {}; + const selection = await selectHeartbeatEnvironment(db, { + run: input.run, agent: input.agent, + task: { issueRef: task, issueContext: task, issueId: issue.id, context: input.task.context, + executionProjectId: null, workspaceProjectContext: null, boundSourceWorkspace: null, + hasSelectedWorkspaceSource: false, taskWorkspaceIntent: null, responsibleUserId: null }, + policy: { ...input.policy, issueExecutionWorkspaceSettings: settings }, + config: { config: agentConfig, sourceWorkspaceConfig: buildExecutionWorkspaceAdapterConfig({ agentConfig, projectPolicy: null, + issueSettings: settings, mode: "isolated_workspace", legacyUseProjectWorkspace: null }), issueAssigneeOverrides: null }, + services: { environmentsSvc: environmentService(db), executionWorkspacesSvc: executionWorkspaceService(db), + instanceSettings: instanceSettingsService(db), findSharedWorkspaceHolder: async () => null, findExecutionWorkspaceHolder: async () => null }, + }); + expect(selection.useIsolatedTaskDirectory).toBe(strategySource === "implicit"); + expect(selection.reusableExistingExecutionWorkspace?.id).toBe(task.executionWorkspaceId); + }); + const readIssue = (id: string) => db.select().from(issues).where(eq(issues.id, id)).then(rows => rows[0]!); const readRun = (id: string) => db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)).then(rows => rows[0]!); const leases = (id: string) => db.select().from(environmentLeases).where(eq(environmentLeases.heartbeatRunId, id)); 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 add9cfac21e..b3af82ecd05 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -11260,6 +11260,80 @@ describe("runnerd provider runtime wiring", () => { }; } + it("admits a warm follow-up after resuming a suspended plugin session", async () => { + const first = { ...scopedNativeExecution("plugin-cold-then-warm"), + schema: "paperclip.native-execution-input.v5" as const, executionMode: "default" as const, + provider: { kind: "codex" as const, model: null, approvalPolicy: "never" as const }, + planningContext: null, runtimeContext: nativeRuntimeContextFixture(), + session: { ...execution.session, normalizedSessionId: "plugin-cold-then-warm", + lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 } }, + }; + parseNativeExecutionInput(first); + const identity = { runId: first.binding.runId, normalizedSessionId: first.session.normalizedSessionId, + runnerInstanceId: "plugin-runner", environmentLeaseId: "original-lease", turnId: "original-turn", itemId: "original-item" }; + const root = await initializeScopedRunnerRoot(first, identity.runnerInstanceId); + await mkdir(join(root, "control-plane"), { recursive: true }); + const journal = (runId: string) => JSON.stringify({ ...durableControlPlaneState({ ...identity, runId }), + commands: [{ status: "completed" }], committedEvents: [{ eventType: "run.terminal", envelope: { ...identity, runId } }] }); + await writeFile(join(root, "control-plane/control-plane-state.json"), journal(identity.runId)); + const receiptPath = join(root, "plugin-runner-state.json"); + await writeFile(receiptPath, JSON.stringify(durableRunnerState(identity, "suspended"))); + const home = join(isolatedStateDirectory, "plugin-account"); + await mkdir(home); + await writeFile(join(home, "auth.json"), JSON.stringify({ OPENAI_API_KEY: "synthetic-fixture" }), { mode: 0o600 }); + const task = vi.fn(async (operation: any) => operation.kind === "read_state" + ? { kind: "state", taskId: "replacement-task", state: { ...durableRunnerState(identity, "suspended"), + outbox: [], pendingTerminalDelivery: null } } + : { kind: "description", taskId: "replacement-task", launchPreparationVersion: 1, + projectAuthCheckVersion: 1, warmTaskVersion: 1, runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } }); + const resumed = { ...first, binding: { ...first.binding, runId: `${first.binding.runId}-resumed` } }; + const followup = { ...first, binding: { ...first.binding, runId: `${first.binding.runId}-followup` } }; + const session = { close: vi.fn(async () => undefined) }; + const result = { result: { summary: "done" }, terminal: { runTerminalState: "succeeded" }, turnId: "turn", + normalizedSessionId: first.session.normalizedSessionId, providerSessionId: "original-provider-history", + driverKind: "codex_app_server", driverVersion: "1", nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null }; + const resume = vi.spyOn(environmentTaskRuntime, "resolvePluginTaskResume").mockResolvedValue("original-task"); + const createOwner = pluginTaskRuntime.createPluginRunnerTaskOwner; + const rebind = vi.fn(); + const owner = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner") + .mockImplementation(input => ({ ...createOwner(input), rebind })); + state.execute.mockReset().mockImplementation(async options => { + if (options.input.binding.runId === resumed.binding.runId) expect(existsSync(receiptPath)).toBe(true); + await writeFile(join(root, "control-plane/control-plane-state.json"), journal(options.input.binding.runId)); + await options.onSession?.(session); + return result; + }); + const options = { runnerInstanceId: identity.runnerInstanceId, useRunnerd: true, runnerIngressAuthorized: true, + managedAiCredentialHome: home, managedAiCredentialIdentity: "revision", managedAiAccountIdentity: "account" }; + const dbFor = (current: NativeExecutionInput, previous: NativeExecutionInput) => { + const db = leaseDb(current); + const select = db.select.bind(db); + db.select = ((fields?: Record) => fields?.status === heartbeatRuns.status + && fields?.runnerProfileJson === heartbeatRuns.runnerProfileJson ? priorNativeRunDb(previous).select(fields as never) + : select(fields as never)) as Db["select"]; + return db; + }; + const target = { kind: "remote" as const, transport: "sandbox" as const, environmentId: "plugin-environment", + remoteCwd: ".", leaseId: "replacement-lease", runnerTask: task, + sandboxLeaseAcquisition: { outcome: "replacement" as const, providerLeaseId: "replacement-task" } }; + try { + await expect(executePaperclipNativeSession({ ...options, execution: resumed, db: dbFor(resumed, first), + runnerExecutionTarget: target as never })).resolves.toMatchObject({ exitCode: 0 }); + const checkpointReads = task.mock.calls.filter(([operation]) => operation.kind === "read_state").length; + await expect(executePaperclipNativeSession({ ...options, execution: followup, db: dbFor(followup, resumed), + runnerExecutionTarget: { ...target, leaseId: "followup-lease", sandboxLeaseAcquisition: { + outcome: "resumed", providerLeaseId: "replacement-task" } } as never })).resolves.toMatchObject({ exitCode: 0 }); + expect(state.execute.mock.calls[1]![0].existingSession).toBe(session); + expect(rebind).toHaveBeenCalledOnce(); + expect(session.close).not.toHaveBeenCalled(); + expect(task.mock.calls.filter(([operation]) => operation.kind === "read_state")).toHaveLength(checkpointReads); + expect(existsSync(receiptPath)).toBe(false); + } finally { + owner.mockRestore(); + resume.mockRestore(); + } + }); + function priorNativeRunDb(priorExecution: NativeExecutionInput, status = "succeeded"): Db { return { select: () => ({ from: (table: unknown) => { diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index e19a19689b0..7b1ae6608a6 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -9298,6 +9298,9 @@ async function executePaperclipNativeSessionWithinScope( ); }, onSession: async (session) => { + if (session && usesPluginTask) { + rmSync(resolve(scopedRunnerdStateRoot(input.execution), "plugin-runner-state.json"), { force: true }); + } if (session && session === existingWarmSession) remoteWarmRecoveryPending = false; if (session && input.managedAiCredentialHome) { managedCredentialSession = runnerdBackend?.bindManagedSession(session) ?? session; diff --git a/server/src/services/provider-fresh-workspace.test.ts b/server/src/services/provider-fresh-workspace.test.ts index 324f6b9542a..e26fad13e89 100644 --- a/server/src/services/provider-fresh-workspace.test.ts +++ b/server/src/services/provider-fresh-workspace.test.ts @@ -5,12 +5,34 @@ import { tmpdir } from "node:os"; import path from "node:path"; import { afterEach, describe, expect, it, vi } from "vitest"; import type { Db } from "@paperclipai/db"; -import { releaseProviderRecoveryTransfers, stageProviderFreshWorkspace } from "./provider-fresh-workspace.js"; +import { isBoundFreshPluginTaskDirectory, releaseProviderRecoveryTransfers, stageProviderFreshWorkspace } from "./provider-fresh-workspace.js"; +import type { ExecutionWorkspace } from "@paperclipai/shared"; const state = vi.hoisted(() => ({ getObject: vi.fn() })); vi.mock("../storage/index.js", () => ({ getStorageService: () => ({ getObject: state.getObject }) })); let home: string | undefined; afterEach(async () => { vi.unstubAllEnvs(); state.getObject.mockReset(); if (home) await rm(home, { recursive: true, force: true }); }); describe("fresh provider recovery transfer staging", () => { + it.each(["valid", "company", "task", "binding", "owner", "path", "attempt", "closed", "repository", "strategy"])( + "recognizes only the bound provider directory: %s", scenario => { + const companyId = randomUUID(), issueId = randomUUID(), workspaceId = randomUUID(); + const owner = { agentId: randomUUID(), environmentId: randomUUID(), pluginKey: "test.tasks", driverKey: "tasks" }; + const workspace = { id: workspaceId, companyId, sourceIssueId: issueId, status: "active", mode: "isolated_workspace", + strategyType: "project_primary", providerType: "adapter_managed", repoUrl: null, branchName: null, + cwd: `/workspaces/${workspaceId}`, metadata: { runnerTaskWorkspaceOwner: owner, + runnerTaskFreshWorkspace: { attemptId: workspaceId, sourceRunId: randomUUID(), previousWorkspaceId: null, recoverySnapshots: [] } }, + } as unknown as ExecutionWorkspace; + if (scenario === "company") workspace.companyId = randomUUID(); + if (scenario === "task") workspace.sourceIssueId = randomUUID(); + if (scenario === "binding") workspace.id = randomUUID(); + if (scenario === "owner") workspace.metadata!.runnerTaskWorkspaceOwner = { ...owner, agentId: randomUUID() }; + if (scenario === "path") workspace.cwd = "/workspaces/other"; + if (scenario === "attempt") workspace.metadata!.runnerTaskFreshWorkspace = { attemptId: randomUUID() }; + if (scenario === "closed") workspace.status = "archived"; + if (scenario === "repository") workspace.repoUrl = "https://github.com/example/repo.git"; + if (scenario === "strategy") workspace.strategyType = "git_worktree"; + expect(isBoundFreshPluginTaskDirectory({ companyId, issueId, workspaceId, workspace, owner })).toBe(scenario === "valid"); + }); + it("verifies immutable transfer files and reuses stable descriptors after interruption", async () => { home = await mkdtemp(path.join(tmpdir(), "provider-recovery-")); vi.stubEnv("PAPERCLIP_HOME", home); const companyId = randomUUID(), sourceRunId = randomUUID(), leaseId = randomUUID(), operationId = randomUUID(); diff --git a/server/src/services/provider-fresh-workspace.ts b/server/src/services/provider-fresh-workspace.ts index e3e0ecfca0e..fcc184e15f4 100644 --- a/server/src/services/provider-fresh-workspace.ts +++ b/server/src/services/provider-fresh-workspace.ts @@ -9,6 +9,7 @@ import { Transform } from "node:stream"; import { and, asc, eq } from "drizzle-orm"; import { executionWorkspaces, heartbeatRuns, issues, projectWorkspaces, workspaceOperations, type Db } from "@paperclipai/db"; import { freshWorkspaceSchema, type PluginFreshWorkspace } from "@paperclipai/plugin-sdk"; +import type { ExecutionWorkspace } from "@paperclipai/shared"; import { resolvePaperclipInstanceRoot } from "../home-paths.js"; import { getStorageService } from "../storage/index.js"; import { recoveryRunAncestry } from "./recovery-run-ancestry.js"; @@ -17,6 +18,22 @@ import { conflict, forbidden } from "../errors.js"; import { notifyTerminalWorkspaceWork } from "./terminal-workspace-work.js"; import { SystemRecoveryFailure } from "./system-recovery-state.js"; +export function isBoundFreshPluginTaskDirectory(input: { + companyId: string; issueId: string; workspaceId: string | null; + workspace: ExecutionWorkspace | null; + owner: { agentId: string; environmentId: string; pluginKey: unknown; driverKey: unknown }; +}): boolean { + const workspace = input.workspace; + if (!workspace || workspace.id !== input.workspaceId || workspace.companyId !== input.companyId || + workspace.sourceIssueId !== input.issueId || workspace.status !== "active" || + workspace.mode !== "isolated_workspace" || workspace.strategyType !== "project_primary" || + workspace.providerType !== "adapter_managed" || workspace.repoUrl || workspace.branchName || + workspace.cwd !== `/workspaces/${workspace.id}` || + !isDeepStrictEqual(workspace.metadata?.runnerTaskWorkspaceOwner, input.owner)) return false; + const fresh = freshWorkspaceSchema.safeParse(workspace.metadata?.runnerTaskFreshWorkspace); + return fresh.success && fresh.data.attemptId === workspace.id; +} + export async function stageProviderFreshWorkspace(input: { db: Db; companyId: string; issueId: string; attemptId: string; sourceRunId: string; previousWorkspaceId: string | null; existing?: PluginFreshWorkspace;