mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-21 08:22:20 +02:00
refactor(heartbeat): extract environment selection (#15825)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Heartbeat execution selects an environment before it prepares a workspace and starts an agent. > - Workspace preparation now has its own module, but its selection policy still occupies the executor. > - Environment defaults, recovery bindings, chat isolation, and workspace contention must retain their order. > - This pull request moves that selection phase behind one typed function. > - The executor keeps dispatch gates and lease cleanup. Tests cover the public path and the new boundary. ## Linked Issues or Issue Description Refs #15807. This continues the heartbeat extraction series after workspace preparation. **Current behavior** The 5,599-line heartbeat service contains a 301-line environment selection phase inside run execution. **Proposed behavior** Move that phase to `heartbeat/environment-selection.ts`. Keep the same policy, statement order, errors, and public exports. **Reason and benefit** Readers can find selection policy in one module and follow the executor's phases without reading each policy branch. ## What Changed - Add `selectHeartbeatEnvironment` with typed run, task, policy, configuration, and service inputs. Return the values needed by configuration and workspace preparation. - Preserve native recovery bindings, environment precedence, managed sandbox enforcement, Kubernetes provisioning, native chat isolation, and shared workspace contention. - Reduce `heartbeat.ts` from 5,599 lines / 234,599 bytes to 5,314 lines / 220,784 bytes. The new module is 402 lines / 18,215 bytes. - Add six public-path characterization cases before moving production code and six boundary cases. Repair plugin-test teardown to drain active execution and call the database cleanup helper. - Assert the selected local environment in the browser fixture's real worker and document the new boundary. ## Verification - Before the move: 71 tests passed, including all six new public-path cases. Three plugin tests skipped when the host could not start their database probe. - After the move: all 112 focused tests passed across eight files, with no skips. These cover environment policy, local and plugin execution, contention, recovery bindings, workspace preparation, chat isolation, and low-trust directories. - Passed `pnpm -r typecheck` and `pnpm build`. - A normalized TypeScript AST comparison confirms the 301 moved lines and remaining executor statements are unchanged, apart from the new call and imports. - Browser acceptance: the task UI recovered automatically and the worker verified its selected local environment, preserved uncommitted files, and completed the task. The persistent-failure case passed in isolation: three total runs, then a blocked task that remained blocked after reload. - The first persistent-failure browser attempt missed the retry banner within five seconds, despite a persisted scheduled retry. Its unchanged isolated rerun passed. The trace is retained; the intermittent UI failure is not yet attributed to a root cause. No assertion or timeout was relaxed. - Greptile reviewed `20953dd164decb565bc8d85081aad31a23fd33e5`: 5/5, no actionable findings. - Hosted CI passed on `20953dd164decb565bc8d85081aad31a23fd33e5`: 52 successful checks and two intentionally skipped Storybook checks, with no pending or failing checks. This includes all general and serialized test groups, Runner verification, eight browser shards, typecheck, and build. - The duplicate local `pnpm test:run` was interrupted after hosted CI completed. It is not reported as a local full-suite pass. Local completed checks are the 112 focused tests, typecheck, build, and browser cases described above. ## Risks - This selection phase enforces execution restrictions. A wrong input or import could select the wrong environment or change retry behavior. Public-path tests cover refusal, redirect, contention, and dispatch; boundary tests cover recovery and isolation. - The helper can provision a managed environment row and add concurrency notes to the supplied task context. It acquires no execution lease. The executor retains its catch/finally boundaries and resource ownership. - Kubernetes selection is tested with real database state and bootstrap settings. These checks do not claim execution against a live Kubernetes cluster or hosted sandbox provider. - No schema, migration, API contract, or intended behavior change. ## Model Used OpenAI Codex (GPT-6), with repository inspection, code editing, compiler checks, database integration tests, and browser automation. The exact serving model ID and context-window size are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
de9ab8e7e0
commit
30e338fba6
7 files changed
+676
-317
No files matched your search
+9
-1
@@ -1360,12 +1360,20 @@ lifecycle and ownership callbacks update its teardown state immediately, includi
|
||||
when preparation later throws. Public sandbox lifecycle and model fallback helpers
|
||||
remain re-exported by `heartbeat.ts`.
|
||||
|
||||
Environment selection is in `server/src/services/heartbeat/environment-selection.ts`.
|
||||
`selectHeartbeatEnvironment` resolves environment policy and workspace reuse,
|
||||
including immutable native recovery bindings, managed sandbox enforcement, and
|
||||
Kubernetes provisioning. It validates native chat isolation and checks shared
|
||||
workspace contention before acquisition. It can defer a sandbox run or add a
|
||||
concurrency note to the task context. The executor retains lease ownership,
|
||||
dispatch gates, and terminal cleanup.
|
||||
|
||||
Workspace preparation is in `server/src/services/heartbeat/workspace-preparation.ts`.
|
||||
`prepareHeartbeatWorkspace` resolves and provisions the workspace, persists its
|
||||
metadata and issue binding, then acquires and realizes the execution environment.
|
||||
It preserves the immutable workspace binding used by native recovery and records
|
||||
preparation spans even when acquisition or realization fails. The executor retains
|
||||
environment selection, dispatch gates, and terminal lease cleanup. Persistence
|
||||
dispatch gates and terminal lease cleanup. Persistence
|
||||
failures still clean newly created workspace artifacts before propagating.
|
||||
|
||||
Runtime execution is in `server/src/services/heartbeat/runtime-execution.ts`.
|
||||
|
||||
@@ -59,14 +59,16 @@ describeEmbeddedPostgres("heartbeat plugin environments", () => {
|
||||
let stopDb: (() => Promise<void>) | null = null;
|
||||
let db!: ReturnType<typeof createDb>;
|
||||
const tempRoots: string[] = [];
|
||||
const heartbeats: ReturnType<typeof heartbeatService>[] = [];
|
||||
|
||||
beforeAll(async () => {
|
||||
const started = await startEmbeddedPostgresTestDatabase("heartbeat-plugin-environment");
|
||||
stopDb = started.stop;
|
||||
stopDb = started.cleanup;
|
||||
db = createDb(started.connectionString);
|
||||
}, 20_000);
|
||||
|
||||
afterEach(async () => {
|
||||
for (const heartbeat of heartbeats.splice(0)) await heartbeat.drainActiveRunExecutions();
|
||||
adapterExecute.mockClear();
|
||||
while (tempRoots.length > 0) {
|
||||
const root = tempRoots.pop();
|
||||
@@ -195,6 +197,7 @@ describeEmbeddedPostgres("heartbeat plugin environments", () => {
|
||||
});
|
||||
|
||||
const heartbeat = heartbeatService(db, { pluginWorkerManager: workerManager });
|
||||
heartbeats.push(heartbeat);
|
||||
const run = await heartbeat.wakeup(agentId, {
|
||||
source: "on_demand",
|
||||
triggerDetail: "manual",
|
||||
@@ -432,6 +435,7 @@ describeEmbeddedPostgres("heartbeat plugin environments", () => {
|
||||
]);
|
||||
|
||||
const heartbeat = heartbeatService(db, { pluginWorkerManager: workerManager });
|
||||
heartbeats.push(heartbeat);
|
||||
const sharedRun = await heartbeat.wakeup(agentAId, {
|
||||
source: "on_demand",
|
||||
triggerDetail: "manual",
|
||||
@@ -657,6 +661,7 @@ describeEmbeddedPostgres("heartbeat plugin environments", () => {
|
||||
});
|
||||
|
||||
const heartbeat = heartbeatService(db, { pluginWorkerManager: workerManager });
|
||||
heartbeats.push(heartbeat);
|
||||
const run = await heartbeat.wakeup(agentId, {
|
||||
source: "assignment",
|
||||
triggerDetail: "manual",
|
||||
|
||||
@@ -156,6 +156,8 @@ describeEmbeddedPostgres("shared-workspace run serialization", () => {
|
||||
failExecutionRunIds.clear();
|
||||
captureRunFailure.mockClear();
|
||||
await instanceSettingsService(db).updateGeneral({ executionMode: "any" });
|
||||
await instanceSettingsService(db).updateExperimental({ enableManagedSandboxOnly: false });
|
||||
vi.unstubAllEnvs();
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
@@ -489,6 +491,90 @@ describeEmbeddedPostgres("shared-workspace run serialization", () => {
|
||||
expect(executedRunIds).not.toContain(run!.id);
|
||||
});
|
||||
|
||||
it.each(["any", "invalid"])("refuses local fallback when Kubernetes is required but bootstrap mode is %s", async (mode) => {
|
||||
const fixture = await seedWorkspaceFixture({ agentEnvironmentDriver: "local" });
|
||||
vi.stubEnv("PAPERCLIP_EXECUTION_MODE", mode);
|
||||
await instanceSettingsService(db).updateGeneral({ executionMode: "kubernetes" });
|
||||
|
||||
const run = await heartbeat.invoke(fixture.agentId, "assignment", { issueId: fixture.issueId }, "system");
|
||||
expect(run).not.toBeNull();
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
|
||||
const finished = await heartbeat.getRun(run!.id);
|
||||
expect(finished?.status).toBe("failed");
|
||||
expect(finished?.error).toContain("no managed Kubernetes environment is configured");
|
||||
expect(executedRunIds).not.toContain(run!.id);
|
||||
expect(await db.select().from(environmentLeases).where(eq(environmentLeases.heartbeatRunId, run!.id))).toHaveLength(0);
|
||||
expect((await heartbeat.getRun(fixture.holderRunId))?.status).toBe("running");
|
||||
});
|
||||
|
||||
it("lazily provisions the required Kubernetes environment before applying workspace contention", async () => {
|
||||
const fixture = await seedWorkspaceFixture({ agentEnvironmentDriver: "local" });
|
||||
for (const key of Object.keys(process.env).filter(key => key.startsWith("PAPERCLIP_K8S_"))) {
|
||||
vi.stubEnv(key, undefined);
|
||||
}
|
||||
vi.stubEnv("PAPERCLIP_EXECUTION_MODE", "kubernetes");
|
||||
vi.stubEnv("PAPERCLIP_K8S_IN_CLUSTER", "true");
|
||||
vi.stubEnv("PAPERCLIP_K8S_BACKEND", "job");
|
||||
await instanceSettingsService(db).updateGeneral({ executionMode: "kubernetes" });
|
||||
|
||||
const run = await heartbeat.invoke(fixture.agentId, "assignment", { issueId: fixture.issueId }, "system");
|
||||
expect(run).not.toBeNull();
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
|
||||
const configured = await db.select().from(environments).where(eq(environments.driver, "sandbox"));
|
||||
expect(configured).toMatchObject([{
|
||||
status: "active",
|
||||
config: { provider: "kubernetes", inCluster: true, backend: "job" },
|
||||
metadata: { managedKubernetesSandbox: true },
|
||||
}]);
|
||||
expect((await heartbeat.getRun(run!.id))?.errorCode).toBe(WORKSPACE_BUSY_ERROR_CODE);
|
||||
expect(executedRunIds).not.toContain(run!.id);
|
||||
expect(await db.select().from(environmentLeases).where(eq(environmentLeases.heartbeatRunId, run!.id))).toHaveLength(0);
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, run!.id))).toHaveLength(1);
|
||||
});
|
||||
|
||||
it.each(["missing", "archived"])("refuses local execution when the managed sandbox is %s", async (state) => {
|
||||
const fixture = await seedWorkspaceFixture({ agentEnvironmentDriver: "local" });
|
||||
if (state === "archived") {
|
||||
await db.insert(environments).values({
|
||||
name: "Unavailable managed sandbox", driver: "sandbox", status: "archived",
|
||||
config: { provider: "fake" }, metadata: { managedByPaperclip: true },
|
||||
});
|
||||
}
|
||||
await instanceSettingsService(db).updateExperimental({ enableManagedSandboxOnly: true });
|
||||
|
||||
const run = await heartbeat.invoke(fixture.agentId, "assignment", { issueId: fixture.issueId }, "system");
|
||||
expect(run).not.toBeNull();
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
|
||||
const finished = await heartbeat.getRun(run!.id);
|
||||
expect(finished?.status).toBe("failed");
|
||||
expect(finished?.error).toContain("no active managed sandbox environment exists");
|
||||
expect(executedRunIds).not.toContain(run!.id);
|
||||
expect(await db.select().from(environmentLeases).where(eq(environmentLeases.heartbeatRunId, run!.id))).toHaveLength(0);
|
||||
expect((await heartbeat.getRun(fixture.holderRunId))?.status).toBe("running");
|
||||
});
|
||||
|
||||
it("redirects a local selection to the managed sandbox before applying workspace contention", async () => {
|
||||
const fixture = await seedWorkspaceFixture({ agentEnvironmentDriver: "local" });
|
||||
await db.insert(environments).values({
|
||||
name: "Managed sandbox", driver: "sandbox", status: "active",
|
||||
config: { provider: "fake" }, metadata: { managedByPaperclip: true },
|
||||
});
|
||||
await instanceSettingsService(db).updateExperimental({ enableManagedSandboxOnly: true });
|
||||
|
||||
const run = await heartbeat.invoke(fixture.agentId, "assignment", { issueId: fixture.issueId }, "system");
|
||||
expect(run).not.toBeNull();
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
|
||||
expect((await heartbeat.getRun(run!.id))?.errorCode).toBe(WORKSPACE_BUSY_ERROR_CODE);
|
||||
expect(executedRunIds).not.toContain(run!.id);
|
||||
expect(await db.select().from(environmentLeases).where(eq(environmentLeases.heartbeatRunId, run!.id))).toHaveLength(0);
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, run!.id))).toHaveLength(1);
|
||||
expect((await heartbeat.getRun(fixture.holderRunId))?.status).toBe("running");
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ driver: "local", policySource: "issue" },
|
||||
{ driver: "local", policySource: "project" },
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { selectHeartbeatEnvironment } from "./heartbeat/environment-selection.js";
|
||||
import { prepareHeartbeatWorkspace } from "./heartbeat/workspace-preparation.js";
|
||||
import { executeHeartbeatRuntime, NativeSessionResumeScheduledError, NativeWorkspaceFinalizeScheduledError } from "./heartbeat/runtime-execution.js";
|
||||
import { selectHeartbeatRuntime } from "./heartbeat/runtime-selection.js";
|
||||
@@ -55,7 +56,6 @@ import {
|
||||
} from "./heartbeat/recovery.js";
|
||||
import {
|
||||
BOUNDED_TRANSIENT_HEARTBEAT_RETRY_MAX_ATTEMPTS,
|
||||
WorkspaceBusyDeferral,
|
||||
isWorkspaceBusyDeferral,
|
||||
createHeartbeatRetries,
|
||||
} from "./heartbeat/retries.js";
|
||||
@@ -138,8 +138,6 @@ export {
|
||||
buildPaperclipWakePayload,
|
||||
} from "./heartbeat/run-preparation.js";
|
||||
import {
|
||||
resolveNativeRecoveryExecutionWorkspaceBinding,
|
||||
resolveExecutionWorkspaceReuseRequestForIssue,
|
||||
buildExecutionWorkspaceConfigSnapshot,
|
||||
stripWorkspaceRuntimeFromExecutionRunConfig,
|
||||
buildEffectiveRunSessionConfigMetadata,
|
||||
@@ -421,12 +419,6 @@ import {
|
||||
isNativeRunnerOwnershipHeld,
|
||||
NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE,
|
||||
} from "./native-runtime/native-runner-ownership.js";
|
||||
import {
|
||||
findNativeChatWorkspaceScope,
|
||||
nativeChatWorkspaceCwd,
|
||||
nativeChatWorkspaceMatches,
|
||||
} from "./native-runtime/native-chat-workspace.js";
|
||||
import { shouldUseIsolatedTaskDirectory } from "./isolated-task-directory.js";
|
||||
import { trackAgentFirstHeartbeat } from "@paperclipai/shared/telemetry";
|
||||
import { getTelemetryClient } from "../telemetry.js";
|
||||
import {
|
||||
@@ -505,18 +497,13 @@ import {
|
||||
gateProjectExecutionWorkspacePolicy,
|
||||
parseIssueExecutionWorkspaceSettings,
|
||||
parseProjectExecutionWorkspacePolicy,
|
||||
resolveExecutionWorkspaceEnvironmentId,
|
||||
resolveExecutionWorkspaceMode,
|
||||
resolveSharedWorkspaceConcurrency,
|
||||
selectEnvironmentExecutionWorkspaceSettings,
|
||||
} from "./execution-workspace-policy.js";
|
||||
import {
|
||||
instanceSettingsService,
|
||||
resolveWorktreeRunExecutionActivation,
|
||||
} from "./instance-settings.js";
|
||||
import {
|
||||
isExecutionForcedToKubernetes,
|
||||
} from "./execution-allowlist.js";
|
||||
|
||||
import {
|
||||
buildConfigurationIncompleteRecoveryNoticeSeed,
|
||||
@@ -571,7 +558,6 @@ import {
|
||||
} from "@paperclipai/adapter-utils/server-utils";
|
||||
|
||||
import { environmentService } from "./environments.js";
|
||||
import { parseExecutionPolicyBootstrapEnv } from "./execution-policy-bootstrap.js";
|
||||
|
||||
import {
|
||||
environmentRuntimeService,
|
||||
@@ -3034,306 +3020,35 @@ export function heartbeatService(
|
||||
} else {
|
||||
delete context.paperclipTurnContext;
|
||||
}
|
||||
// A native run's execution input is immutable once persisted. Recovery must therefore
|
||||
// restore the workspace bound to that input rather than consulting the issue's current
|
||||
// workspace pointer: a newer run may already have moved or cleared the issue binding while
|
||||
// this older provider session is still recoverable.
|
||||
const persistedRunnerProfile = parseObject(run.runnerProfileJson);
|
||||
const persistedNativeExecutionInput =
|
||||
run.runtimeMode === "native" &&
|
||||
persistedRunnerProfile.nativeExecutionInput !== undefined
|
||||
? parseNativeExecutionInput(
|
||||
persistedRunnerProfile.nativeExecutionInput,
|
||||
)
|
||||
: null;
|
||||
const isDotRun = persistedNativeExecutionInput?.provider.kind === "openai_dot"
|
||||
|| (!persistedNativeExecutionInput && agent.adapterType === "paperclip_runner" && parseObject(agent.adapterConfig).provider === "openai_dot");
|
||||
const persistedNativeExecutionWorkspaceId =
|
||||
persistedNativeExecutionInput?.binding.executionWorkspaceId ?? null;
|
||||
const requestedExecutionWorkspaceId =
|
||||
persistedNativeExecutionWorkspaceId ??
|
||||
readNonEmptyString(issueRef?.executionWorkspaceId);
|
||||
const existingExecutionWorkspace = requestedExecutionWorkspaceId
|
||||
? await executionWorkspacesSvc.getById(requestedExecutionWorkspaceId)
|
||||
: null;
|
||||
const nativeRecoveryExecutionWorkspaceId =
|
||||
resolveNativeRecoveryExecutionWorkspaceBinding({
|
||||
bindingId: persistedNativeExecutionWorkspaceId,
|
||||
persistedWorkspaceFound: existingExecutionWorkspace !== null,
|
||||
});
|
||||
const workspaceReuseRequest =
|
||||
resolveExecutionWorkspaceReuseRequestForIssue({
|
||||
issueExecutionWorkspaceId: requestedExecutionWorkspaceId,
|
||||
issueExecutionWorkspacePreference: nativeRecoveryExecutionWorkspaceId
|
||||
? "reuse_existing"
|
||||
: (issueRef?.executionWorkspacePreference ?? null),
|
||||
existingExecutionWorkspaceStatus:
|
||||
existingExecutionWorkspace?.status ?? null,
|
||||
});
|
||||
const requestedShouldReuseExisting =
|
||||
workspaceReuseRequest.requestedShouldReuseExisting;
|
||||
const reusableExistingExecutionWorkspace =
|
||||
workspaceReuseRequest.existingExecutionWorkspaceAvailable
|
||||
? existingExecutionWorkspace
|
||||
: null;
|
||||
const requestedReusableExecutionWorkspaceConfig =
|
||||
reusableExistingExecutionWorkspace?.config ?? null;
|
||||
const localEnvironment = await environmentsSvc.ensureLocalEnvironment(
|
||||
agent.companyId,
|
||||
);
|
||||
const resolvedInstanceSettings = await instanceSettings.get();
|
||||
// Managed-sandbox-only policy: a run that would land on the local
|
||||
// environment is redirected onto the platform-managed sandbox row, and
|
||||
// with no active managed row the resolution fails closed
|
||||
// (ManagedSandboxUnavailableError) — never local. Mirrors the forced
|
||||
// kubernetes execution mode below, which takes precedence when both
|
||||
// regimes are active.
|
||||
const managedSandboxOnly =
|
||||
(await instanceSettings.getExperimental()).enableManagedSandboxOnly ===
|
||||
true;
|
||||
const managedSandboxEnvironment = managedSandboxOnly
|
||||
? await environmentsSvc.findManagedSandboxEnvironment(agent.companyId)
|
||||
: null;
|
||||
const environmentResolution = resolveExecutionWorkspaceEnvironmentId({
|
||||
lowTrustIssueEnvironmentId: trustPreset.kind === "low_trust_review"
|
||||
? parseIssueExecutionWorkspaceSettings(issueContext?.executionWorkspaceSettings, {includeEnvironmentId: true})?.environmentId : null,
|
||||
agentDefaultEnvironmentId: agent.defaultEnvironmentId,
|
||||
instanceDefaultEnvironmentId:
|
||||
resolvedInstanceSettings.defaultEnvironmentId ?? null,
|
||||
localDefaultEnvironmentId: localEnvironment.id,
|
||||
managedSandboxOnly,
|
||||
managedSandboxEnvironmentId: managedSandboxEnvironment?.id ?? null,
|
||||
});
|
||||
const effectiveExecutionWorkspaceMode: ReturnType<
|
||||
typeof resolveExecutionWorkspaceMode
|
||||
> = requestedExecutionWorkspaceMode;
|
||||
const executionPolicy = {
|
||||
executionMode: resolvedInstanceSettings.general.executionMode,
|
||||
// Backstop behind the resolver's local→managed redirect: the run-time
|
||||
// allowlist below fails any run that still resolved to a `local`
|
||||
// environment under managed-sandbox-only, so no selection path or
|
||||
// tenant-set env var can land untrusted execution on the tenant
|
||||
// container.
|
||||
managedSandboxOnly,
|
||||
};
|
||||
const executionForcedToKubernetes =
|
||||
isExecutionForcedToKubernetes(executionPolicy);
|
||||
let selectedEnvironmentId = environmentResolution.environmentId;
|
||||
if (executionForcedToKubernetes) {
|
||||
let kubernetesEnvironment =
|
||||
await environmentsSvc.findKubernetesEnvironment(agent.companyId);
|
||||
if (!kubernetesEnvironment) {
|
||||
// Lazy recovery for companies created after the startup bootstrap ran
|
||||
// (the boot hook only provisions environments for companies that exist
|
||||
// at boot). Re-derive the managed-env config from the bootstrap env.
|
||||
// If the process env no longer forces Kubernetes (rollback / config
|
||||
// drift relative to the persisted executionMode setting), skip the
|
||||
// provisioning gracefully: the guard below still refuses local
|
||||
// fallback with the explicit error, instead of crashing here on
|
||||
// undefined config.
|
||||
let bootstrap: ReturnType<typeof parseExecutionPolicyBootstrapEnv> =
|
||||
null;
|
||||
let bootstrapSkipReason: string | null = null;
|
||||
try {
|
||||
bootstrap = parseExecutionPolicyBootstrapEnv(process.env);
|
||||
if (!bootstrap) {
|
||||
bootstrapSkipReason =
|
||||
'PAPERCLIP_EXECUTION_MODE bootstrap env is not kubernetes-forced (absent or "any")';
|
||||
}
|
||||
} catch (err) {
|
||||
bootstrapSkipReason = `PAPERCLIP_EXECUTION_MODE bootstrap env failed to parse: ${
|
||||
err instanceof Error ? err.message : String(err)
|
||||
}`;
|
||||
}
|
||||
if (bootstrap) {
|
||||
await environmentsSvc.ensureKubernetesEnvironment(
|
||||
agent.companyId,
|
||||
bootstrap.kubernetesConfig,
|
||||
);
|
||||
kubernetesEnvironment =
|
||||
await environmentsSvc.findKubernetesEnvironment(agent.companyId);
|
||||
} else {
|
||||
logger.warn(
|
||||
{
|
||||
runId: run.id,
|
||||
agentId: agent.id,
|
||||
companyId: agent.companyId,
|
||||
reason: bootstrapSkipReason,
|
||||
},
|
||||
"executionMode=kubernetes is persisted but the bootstrap env cannot provision a managed Kubernetes environment; skipping lazy provisioning for this company (the run will fail with the explicit no-managed-environment error)",
|
||||
);
|
||||
}
|
||||
}
|
||||
if (!kubernetesEnvironment) {
|
||||
throw new Error(
|
||||
"Instance execution policy requires the Kubernetes sandbox provider " +
|
||||
"(executionMode=kubernetes) but no managed Kubernetes environment is " +
|
||||
"configured for this company. Configure one (PAPERCLIP_K8S_* env on the " +
|
||||
"cloud instance) before running agents; refusing to fall back to local execution.",
|
||||
);
|
||||
}
|
||||
if (kubernetesEnvironment.id !== selectedEnvironmentId) {
|
||||
logger.info(
|
||||
{
|
||||
runId: run.id,
|
||||
issueId,
|
||||
agentId: agent.id,
|
||||
resolvedEnvironmentId: selectedEnvironmentId,
|
||||
forcedKubernetesEnvironmentId: kubernetesEnvironment.id,
|
||||
},
|
||||
"Forcing run onto the managed Kubernetes environment (executionMode=kubernetes)",
|
||||
);
|
||||
}
|
||||
selectedEnvironmentId = kubernetesEnvironment.id;
|
||||
}
|
||||
const selectedEnvironmentForConfig =
|
||||
selectedEnvironmentId === localEnvironment.id
|
||||
? localEnvironment
|
||||
: selectedEnvironmentId
|
||||
? await environmentsSvc.getById(selectedEnvironmentId)
|
||||
: null;
|
||||
const nativeChatWorkspaceScope = await findNativeChatWorkspaceScope(db, {
|
||||
adapterType: agent.adapterType,
|
||||
environmentDriver: selectedEnvironmentForConfig?.driver ?? null,
|
||||
companyId: agent.companyId,
|
||||
agentId: agent.id,
|
||||
issueId,
|
||||
});
|
||||
const nativeChatExpectedCwd = nativeChatWorkspaceScope
|
||||
? nativeChatWorkspaceCwd(
|
||||
nativeChatWorkspaceScope,
|
||||
reusableExistingExecutionWorkspace,
|
||||
requestedShouldReuseExisting,
|
||||
)
|
||||
: null;
|
||||
if (
|
||||
nativeChatWorkspaceScope &&
|
||||
persistedNativeExecutionInput &&
|
||||
persistedNativeExecutionInput.schema !== "paperclip.native-execution-input.v6" &&
|
||||
!nativeChatWorkspaceMatches({
|
||||
scope: nativeChatWorkspaceScope,
|
||||
expectedCwd: nativeChatExpectedCwd,
|
||||
execution: persistedNativeExecutionInput,
|
||||
})
|
||||
) {
|
||||
// Never rewrite an admitted provider input or release ownership of an
|
||||
// older process whose permissions still include the shared agent home.
|
||||
throw new NativeRunnerOwnershipUnverifiedError(
|
||||
"native_chat_workspace_scope_mismatch",
|
||||
);
|
||||
}
|
||||
if (
|
||||
nativeChatWorkspaceScope &&
|
||||
(!nativeChatExpectedCwd ||
|
||||
executionProjectId !== nativeChatWorkspaceScope.projectId)
|
||||
) {
|
||||
throw new ConfigurationIncompleteFailure(
|
||||
"External chat requires a task-owned isolated workspace. Configure and select an existing isolated worktree for this project task; shared project workspaces cannot be used for external chat.",
|
||||
{
|
||||
configurationIncomplete: {
|
||||
reason: "native_chat_workspace_isolation_required",
|
||||
issueId,
|
||||
},
|
||||
},
|
||||
);
|
||||
}
|
||||
const sharedWorkspaceConcurrency = resolveSharedWorkspaceConcurrency({
|
||||
projectPolicy: projectExecutionWorkspacePolicy,
|
||||
issueSettings: issueExecutionWorkspaceSettings,
|
||||
});
|
||||
// A live holder is always consulted for shared workspaces. Depending on policy and the final
|
||||
// execution target it either remains the existing deferral gate or becomes dispatch context.
|
||||
// Local/SSH folders never take an exclusive workspace lock, including when older
|
||||
// project or issue settings request serialization. Sandbox protection still uses
|
||||
// the existing holder staleness and workspace_busy retry ladder.
|
||||
if (
|
||||
issueRef?.projectWorkspaceId &&
|
||||
effectiveExecutionWorkspaceMode === "shared_workspace"
|
||||
) {
|
||||
const workspaceHolder = await findSharedWorkspaceHolder({
|
||||
companyId: agent.companyId,
|
||||
projectWorkspaceId: issueRef.projectWorkspaceId,
|
||||
excludeIssueId: issueRef.id,
|
||||
excludeRunId: run.id,
|
||||
honorIsolatedWorkspaceModes: isolatedWorkspacesEnabled,
|
||||
});
|
||||
if (workspaceHolder) {
|
||||
const environmentDriver =
|
||||
selectedEnvironmentForConfig?.driver ?? null;
|
||||
const shouldSerialize =
|
||||
sharedWorkspaceConcurrency !== "allow" &&
|
||||
(executionForcedToKubernetes ||
|
||||
(environmentDriver !== "local" &&
|
||||
environmentDriver !== "ssh"));
|
||||
if (shouldSerialize) {
|
||||
throw new WorkspaceBusyDeferral({
|
||||
holder: workspaceHolder,
|
||||
projectWorkspaceId: issueRef.projectWorkspaceId,
|
||||
deferralAttempt:
|
||||
run.scheduledRetryReason === WORKSPACE_BUSY_RETRY_REASON
|
||||
? (run.scheduledRetryAttempt ?? 0)
|
||||
: 0,
|
||||
wasIssueAssignee: issueContext?.assigneeAgentId === agent.id,
|
||||
});
|
||||
}
|
||||
|
||||
const holderIssueLabel =
|
||||
workspaceHolder.issueIdentifier ?? workspaceHolder.issueId;
|
||||
const concurrentWorkspaceNote =
|
||||
`shared workspace is concurrently held by run ${workspaceHolder.runId} (issue ${holderIssueLabel}); ` +
|
||||
"expect concurrent mutations, coordinate via commits";
|
||||
const appendConcurrentWorkspaceNote = (value: unknown) => {
|
||||
const existing = typeof value === "string" ? value.trimEnd() : "";
|
||||
return existing
|
||||
? `${existing}\n${concurrentWorkspaceNote}`
|
||||
: concurrentWorkspaceNote;
|
||||
};
|
||||
context.paperclipTaskMarkdown = appendConcurrentWorkspaceNote(
|
||||
context.paperclipTaskMarkdown,
|
||||
);
|
||||
context.paperclipTaskMarkdownAssignment = appendConcurrentWorkspaceNote(
|
||||
context.paperclipTaskMarkdownAssignment,
|
||||
);
|
||||
if (typeof context.paperclipTaskMarkdownCompact === "string") {
|
||||
context.paperclipTaskMarkdownCompact =
|
||||
appendConcurrentWorkspaceNote(
|
||||
context.paperclipTaskMarkdownCompact,
|
||||
);
|
||||
}
|
||||
if (typeof context.paperclipTaskMarkdownAssignmentCompact === "string") {
|
||||
context.paperclipTaskMarkdownAssignmentCompact =
|
||||
appendConcurrentWorkspaceNote(
|
||||
context.paperclipTaskMarkdownAssignmentCompact,
|
||||
);
|
||||
}
|
||||
logger.info(
|
||||
{
|
||||
event: "shared_workspace_concurrent_dispatch",
|
||||
runId: run.id,
|
||||
issueId: issueRef.id,
|
||||
projectWorkspaceId: issueRef.projectWorkspaceId,
|
||||
holderRunId: workspaceHolder.runId,
|
||||
holderIssueId: workspaceHolder.issueId,
|
||||
sharedWorkspaceConcurrency,
|
||||
environmentDriver,
|
||||
executionForcedToKubernetes,
|
||||
},
|
||||
"Dispatching alongside a live shared-workspace holder",
|
||||
);
|
||||
}
|
||||
}
|
||||
const useIsolatedTaskDirectory = issueRef !== null && shouldUseIsolatedTaskDirectory({
|
||||
trustPreset: trustPreset.kind,
|
||||
environmentDriver: selectedEnvironmentForConfig?.driver ?? null,
|
||||
mode: requestedExecutionWorkspaceMode,
|
||||
hasProjectWorkspace: projectContext?.hasWorkspace ?? false,
|
||||
projectWorkspaceId: issueRef.projectWorkspaceId,
|
||||
workspaceStrategies: [
|
||||
config.workspaceStrategy,
|
||||
issueAssigneeOverrides?.adapterConfig?.workspaceStrategy,
|
||||
projectExecutionWorkspacePolicy?.workspaceStrategy,
|
||||
issueExecutionWorkspaceSettings?.workspaceStrategy,
|
||||
],
|
||||
const {
|
||||
persistedRunnerProfile,
|
||||
persistedNativeExecutionInput,
|
||||
isDotRun,
|
||||
existingExecutionWorkspace,
|
||||
nativeRecoveryExecutionWorkspaceId,
|
||||
workspaceReuseRequest,
|
||||
requestedShouldReuseExisting,
|
||||
reusableExistingExecutionWorkspace,
|
||||
requestedReusableExecutionWorkspaceConfig,
|
||||
localEnvironment,
|
||||
resolvedInstanceSettings,
|
||||
environmentResolution,
|
||||
effectiveExecutionWorkspaceMode,
|
||||
executionPolicy,
|
||||
selectedEnvironmentId,
|
||||
selectedEnvironmentForConfig,
|
||||
nativeChatWorkspaceScope,
|
||||
nativeChatExpectedCwd,
|
||||
useIsolatedTaskDirectory,
|
||||
} = await selectHeartbeatEnvironment(db, {
|
||||
run, agent,
|
||||
task: { issueRef, issueContext, issueId, context, executionProjectId, projectContext },
|
||||
policy: {
|
||||
trustPreset, requestedExecutionWorkspaceMode, isolatedWorkspacesEnabled,
|
||||
projectExecutionWorkspacePolicy, issueExecutionWorkspaceSettings,
|
||||
},
|
||||
config: { config, issueAssigneeOverrides },
|
||||
services: { executionWorkspacesSvc, environmentsSvc, instanceSettings, findSharedWorkspaceHolder },
|
||||
});
|
||||
const workspaceManagedConfig = buildExecutionWorkspaceAdapterConfig({
|
||||
agentConfig: config,
|
||||
|
||||
@@ -0,0 +1,142 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { mkdtemp, mkdir, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { agents, companies, createDb, environments, heartbeatRuns, issues, projects, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, type Db } from "@paperclipai/db";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
|
||||
import { resolvePaperclipInstanceRoot } from "../../home-paths.js";
|
||||
import { environmentService } from "../environments.js";
|
||||
import { executionWorkspaceService } from "../execution-workspaces.js";
|
||||
import { instanceSettingsService } from "../instance-settings.js";
|
||||
import { buildNativeExecutionInput } from "../native-runtime/native-execution-input.js";
|
||||
import { nativeRuntimeContextFixture } from "../native-runtime/runtime-context.test-fixture.js";
|
||||
import { selectHeartbeatEnvironment, type HeartbeatEnvironmentSelectionInput } from "./environment-selection.js";
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
describe.skipIf(!support.supported)("heartbeat environment selection boundary", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: Db;
|
||||
let home: string;
|
||||
beforeAll(async () => {
|
||||
home = await mkdtemp(join(tmpdir(), "heartbeat-environment-selection-"));
|
||||
vi.stubEnv("PAPERCLIP_HOME", home);
|
||||
vi.stubEnv("PAPERCLIP_INSTANCE_ID", "environment-selection");
|
||||
await mkdir(resolvePaperclipInstanceRoot(), { recursive: true });
|
||||
database = await startEmbeddedPostgresTestDatabase("heartbeat-environment-selection-");
|
||||
db = createDb(database.connectionString);
|
||||
await db.execute(sql`set client_min_messages = warning`);
|
||||
}, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
|
||||
afterAll(async () => {
|
||||
await db?.$client.end();
|
||||
await database?.cleanup();
|
||||
vi.unstubAllEnvs();
|
||||
if (home) await rm(home, { recursive: true, force: true });
|
||||
});
|
||||
beforeEach(async () => {
|
||||
await db.execute(sql`truncate table companies restart identity cascade`);
|
||||
});
|
||||
|
||||
async function fixture() {
|
||||
const [company] = await db.insert(companies).values({ name: "Environment selection", issuePrefix: "ENV" }).returning();
|
||||
const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Selecting agent", status: "running", adapterType: "process" }).returning();
|
||||
const [project] = await db.insert(projects).values({ companyId: company.id, name: "Project" }).returning();
|
||||
const [issue] = await db.insert(issues).values({ companyId: company.id, projectId: project.id, title: "Select this environment", assigneeAgentId: agent.id }).returning();
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id, issueId: issue.id, scopeKind: "issue", status: "running", runtimeMode: "legacy" }).returning();
|
||||
const input: HeartbeatEnvironmentSelectionInput = {
|
||||
run, agent,
|
||||
task: { issueRef: issue, issueContext: issue, issueId: issue.id, executionProjectId: project.id, projectContext: { hasWorkspace: false }, context: { issueId: issue.id } },
|
||||
policy: {
|
||||
trustPreset: { kind: "standard", preset: "standard", boundary: null, sourcePresets: {} },
|
||||
requestedExecutionWorkspaceMode: "shared_workspace", isolatedWorkspacesEnabled: true,
|
||||
projectExecutionWorkspacePolicy: null, issueExecutionWorkspaceSettings: null,
|
||||
},
|
||||
config: { config: {}, issueAssigneeOverrides: null },
|
||||
services: { environmentsSvc: environmentService(db), executionWorkspacesSvc: executionWorkspaceService(db), instanceSettings: instanceSettingsService(db), findSharedWorkspaceHolder: vi.fn(async () => null) },
|
||||
};
|
||||
const workspace = async () => {
|
||||
const created = await executionWorkspaceService(db).create({ companyId: company.id, projectId: project.id, sourceIssueId: issue.id, name: "Saved workspace", mode: "isolated_workspace", strategyType: "git_worktree", cwd: home, providerRef: home });
|
||||
if (!created) throw new Error("Workspace fixture was not created");
|
||||
return created;
|
||||
};
|
||||
const recover = (workspaceId: string) => {
|
||||
input.run.runtimeMode = "native";
|
||||
input.run.runnerProfileJson = { nativeExecutionInput: buildNativeExecutionInput({
|
||||
companyId: company.id, runId: run.id, agentId: agent.id,
|
||||
issue: { id: issue.id, identifier: "ENV-1", title: issue.title, description: null, workMode: "standard" },
|
||||
taskPrompt: "Continue the same task", normalizedSessionId: randomUUID(), provider: "codex",
|
||||
workspace: { id: workspaceId, cwd: home, repoUrl: null, repoRef: null, branchName: null },
|
||||
completionContract: { id: randomUUID(), sha256: `sha256:${"a".repeat(64)}`, schemaVersion: "paperclip.run-result.v1", contract: { revision: "1", objective: "Select the environment", criteria: [{ id: "objective", requirement: "Preserve the workspace" }] } },
|
||||
runtimeContext: nativeRuntimeContextFixture(),
|
||||
}) };
|
||||
};
|
||||
return { input, issue, project, workspace, recover };
|
||||
}
|
||||
|
||||
it("keeps the admitted workspace binding when the issue points at a newer workspace", async () => {
|
||||
const { input, workspace, recover } = await fixture();
|
||||
const original = await workspace();
|
||||
const newer = await workspace();
|
||||
input.task.issueRef!.executionWorkspaceId = newer.id;
|
||||
input.task.issueRef!.executionWorkspacePreference = null;
|
||||
recover(original.id);
|
||||
const result = await selectHeartbeatEnvironment(db, input);
|
||||
expect(result.nativeRecoveryExecutionWorkspaceId).toBe(original.id);
|
||||
expect(result.reusableExistingExecutionWorkspace?.id).toBe(original.id);
|
||||
expect(result.requestedShouldReuseExisting).toBe(true);
|
||||
expect(input.task.issueRef!.executionWorkspaceId).toBe(newer.id);
|
||||
expect(result.persistedNativeExecutionInput?.binding.executionWorkspaceId).toBe(original.id);
|
||||
});
|
||||
|
||||
it("does not treat a synthetic run-id recovery binding as a persisted workspace", async () => {
|
||||
const { input, recover } = await fixture();
|
||||
recover(input.run.id);
|
||||
const result = await selectHeartbeatEnvironment(db, input);
|
||||
expect(result).toMatchObject({ existingExecutionWorkspace: null, nativeRecoveryExecutionWorkspaceId: null, requestedShouldReuseExisting: false, reusableExistingExecutionWorkspace: null });
|
||||
expect(result.persistedNativeExecutionInput?.binding.executionWorkspaceId).toBe(input.run.id);
|
||||
});
|
||||
|
||||
it("does not make an archived issue workspace available for reuse", async () => {
|
||||
const { input, workspace } = await fixture();
|
||||
const saved = await workspace();
|
||||
await executionWorkspaceService(db).update(saved.id, { status: "archived" });
|
||||
input.task.issueRef!.executionWorkspaceId = saved.id;
|
||||
input.task.issueRef!.executionWorkspacePreference = "reuse_existing";
|
||||
const result = await selectHeartbeatEnvironment(db, input);
|
||||
expect(result.requestedShouldReuseExisting).toBe(true);
|
||||
expect(result.reusableExistingExecutionWorkspace).toBeNull();
|
||||
expect(result.requestedReusableExecutionWorkspaceConfig).toBeNull();
|
||||
});
|
||||
|
||||
it("uses the low-trust task sandbox even when workspace feature gates hide issue settings", async () => {
|
||||
const { input, issue } = await fixture();
|
||||
const [sandbox] = await db.insert(environments).values({ name: "Task sandbox", driver: "sandbox", status: "active", config: { provider: "fake" } }).returning();
|
||||
input.task.issueContext!.executionWorkspaceSettings = { mode: "isolated_workspace", environmentId: sandbox.id };
|
||||
input.policy.trustPreset = { kind: "low_trust_review", preset: "low_trust_review", sourcePresets: {}, boundary: { mode: "low_trust_review", companyId: input.agent.companyId, rootIssueId: issue.id } };
|
||||
input.policy.requestedExecutionWorkspaceMode = "isolated_workspace";
|
||||
input.policy.isolatedWorkspacesEnabled = false;
|
||||
const result = await selectHeartbeatEnvironment(db, input);
|
||||
expect(result.selectedEnvironmentForConfig?.id).toBe(sandbox.id);
|
||||
expect(result.useIsolatedTaskDirectory).toBe(true);
|
||||
});
|
||||
|
||||
it("returns the task-owned local chat root before workspace realization", async () => {
|
||||
const { input, issue } = await fixture();
|
||||
await db.update(issues).set({ originKind: "chat_channel", projectId: null }).where(eq(issues.id, issue.id));
|
||||
input.agent.adapterType = "paperclip_runner";
|
||||
input.task.executionProjectId = null;
|
||||
input.task.projectContext = null;
|
||||
const result = await selectHeartbeatEnvironment(db, input);
|
||||
expect(result.nativeChatWorkspaceScope).toMatchObject({ companyId: input.agent.companyId, agentId: input.agent.id, issueId: issue.id, projectId: null });
|
||||
expect(result.nativeChatExpectedCwd).toBe(result.nativeChatWorkspaceScope?.taskRoot);
|
||||
expect(result.nativeChatExpectedCwd).toContain(`chat-workspaces/${input.agent.companyId}/${input.agent.id}/${issue.id}`);
|
||||
});
|
||||
|
||||
it("rejects a project chat without a task-owned isolated workspace", async () => {
|
||||
const { input, issue } = await fixture();
|
||||
await db.update(issues).set({ originKind: "chat_channel" }).where(eq(issues.id, issue.id));
|
||||
input.agent.adapterType = "paperclip_runner";
|
||||
await expect(selectHeartbeatEnvironment(db, input)).rejects.toThrow("External chat requires a task-owned isolated workspace");
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,402 @@
|
||||
import type { Db, agents, heartbeatRuns, issues } from "@paperclipai/db";
|
||||
import type { IssueExecutionWorkspaceSettings, ProjectExecutionWorkspacePolicy } from "@paperclipai/shared";
|
||||
import { parseObject } from "../../adapters/utils.js";
|
||||
import { logger } from "../../middleware/logger.js";
|
||||
import { parseNativeExecutionInput } from "../../vendor/paperclip-runner/index.js";
|
||||
import { isExecutionForcedToKubernetes } from "../execution-allowlist.js";
|
||||
import { parseExecutionPolicyBootstrapEnv } from "../execution-policy-bootstrap.js";
|
||||
import {
|
||||
parseIssueExecutionWorkspaceSettings,
|
||||
resolveExecutionWorkspaceEnvironmentId,
|
||||
resolveExecutionWorkspaceMode,
|
||||
resolveSharedWorkspaceConcurrency,
|
||||
type ParsedExecutionWorkspaceMode,
|
||||
} from "../execution-workspace-policy.js";
|
||||
import { shouldUseIsolatedTaskDirectory } from "../isolated-task-directory.js";
|
||||
import {
|
||||
findNativeChatWorkspaceScope,
|
||||
nativeChatWorkspaceCwd,
|
||||
nativeChatWorkspaceMatches,
|
||||
} from "../native-runtime/native-chat-workspace.js";
|
||||
import { NativeRunnerOwnershipUnverifiedError } from "../native-runtime/native-runner-ownership.js";
|
||||
import type { environmentService } from "../environments.js";
|
||||
import type { executionWorkspaceService } from "../execution-workspaces.js";
|
||||
import type { instanceSettingsService } from "../instance-settings.js";
|
||||
import type { TrustPresetResolution } from "../trust-preset-resolver.js";
|
||||
import { WORKSPACE_BUSY_RETRY_REASON } from "../../modules/run-dispatch/index.js";
|
||||
import { WorkspaceBusyDeferral, type createHeartbeatRetries } from "./retries.js";
|
||||
import { ConfigurationIncompleteFailure } from "./run-preparation.js";
|
||||
import { resolveNativeRecoveryExecutionWorkspaceBinding, resolveExecutionWorkspaceReuseRequestForIssue } from "./workspaces.js";
|
||||
|
||||
type Issue = typeof issues.$inferSelect;
|
||||
|
||||
/** Selects policy and reuse inputs; lease acquisition and cleanup belong to the executor. */
|
||||
export interface HeartbeatEnvironmentSelectionInput {
|
||||
run: Pick<typeof heartbeatRuns.$inferSelect,
|
||||
"id" | "runtimeMode" | "runnerProfileJson" | "scheduledRetryReason" | "scheduledRetryAttempt">;
|
||||
agent: Pick<typeof agents.$inferSelect,
|
||||
"id" | "companyId" | "adapterType" | "adapterConfig" | "defaultEnvironmentId">;
|
||||
task: {
|
||||
issueRef: Pick<Issue, "id" | "projectWorkspaceId" | "executionWorkspaceId" | "executionWorkspacePreference"> | null;
|
||||
issueContext: Pick<Issue, "assigneeAgentId" | "executionWorkspaceSettings"> | null;
|
||||
issueId: string | null;
|
||||
context: Record<string, unknown>;
|
||||
executionProjectId: string | null;
|
||||
projectContext: { hasWorkspace: boolean } | null;
|
||||
};
|
||||
policy: {
|
||||
trustPreset: TrustPresetResolution;
|
||||
requestedExecutionWorkspaceMode: ParsedExecutionWorkspaceMode;
|
||||
isolatedWorkspacesEnabled: boolean;
|
||||
projectExecutionWorkspacePolicy: ProjectExecutionWorkspacePolicy | null;
|
||||
issueExecutionWorkspaceSettings: IssueExecutionWorkspaceSettings | null;
|
||||
};
|
||||
config: {
|
||||
config: Record<string, unknown>;
|
||||
issueAssigneeOverrides: { adapterConfig: Record<string, unknown> | null } | null;
|
||||
};
|
||||
services: {
|
||||
executionWorkspacesSvc: Pick<ReturnType<typeof executionWorkspaceService>, "getById">;
|
||||
environmentsSvc: Pick<ReturnType<typeof environmentService>,
|
||||
"ensureLocalEnvironment" | "findManagedSandboxEnvironment" | "findKubernetesEnvironment" | "ensureKubernetesEnvironment" | "getById">;
|
||||
instanceSettings: Pick<ReturnType<typeof instanceSettingsService>, "get" | "getExperimental">;
|
||||
findSharedWorkspaceHolder: ReturnType<typeof createHeartbeatRetries>["findSharedWorkspaceHolder"];
|
||||
};
|
||||
}
|
||||
|
||||
function readNonEmptyString(value: unknown): string | null {
|
||||
return typeof value === "string" && value.trim().length > 0 ? value : null;
|
||||
}
|
||||
|
||||
export async function selectHeartbeatEnvironment(db: Db, input: HeartbeatEnvironmentSelectionInput) {
|
||||
const { run, agent } = input;
|
||||
const { issueRef, issueContext, issueId, context, executionProjectId, projectContext } = input.task;
|
||||
const {
|
||||
trustPreset, requestedExecutionWorkspaceMode, isolatedWorkspacesEnabled,
|
||||
projectExecutionWorkspacePolicy, issueExecutionWorkspaceSettings,
|
||||
} = input.policy;
|
||||
const { config, issueAssigneeOverrides } = input.config;
|
||||
const { executionWorkspacesSvc, environmentsSvc, instanceSettings, findSharedWorkspaceHolder } = input.services;
|
||||
// A native run's execution input is immutable once persisted. Recovery must therefore
|
||||
// restore the workspace bound to that input rather than consulting the issue's current
|
||||
// workspace pointer: a newer run may already have moved or cleared the issue binding while
|
||||
// this older provider session is still recoverable.
|
||||
const persistedRunnerProfile = parseObject(run.runnerProfileJson);
|
||||
const persistedNativeExecutionInput =
|
||||
run.runtimeMode === "native" &&
|
||||
persistedRunnerProfile.nativeExecutionInput !== undefined
|
||||
? parseNativeExecutionInput(
|
||||
persistedRunnerProfile.nativeExecutionInput,
|
||||
)
|
||||
: null;
|
||||
const isDotRun = persistedNativeExecutionInput?.provider.kind === "openai_dot"
|
||||
|| (!persistedNativeExecutionInput && agent.adapterType === "paperclip_runner" && parseObject(agent.adapterConfig).provider === "openai_dot");
|
||||
const persistedNativeExecutionWorkspaceId =
|
||||
persistedNativeExecutionInput?.binding.executionWorkspaceId ?? null;
|
||||
const requestedExecutionWorkspaceId =
|
||||
persistedNativeExecutionWorkspaceId ??
|
||||
readNonEmptyString(issueRef?.executionWorkspaceId);
|
||||
const existingExecutionWorkspace = requestedExecutionWorkspaceId
|
||||
? await executionWorkspacesSvc.getById(requestedExecutionWorkspaceId)
|
||||
: null;
|
||||
const nativeRecoveryExecutionWorkspaceId =
|
||||
resolveNativeRecoveryExecutionWorkspaceBinding({
|
||||
bindingId: persistedNativeExecutionWorkspaceId,
|
||||
persistedWorkspaceFound: existingExecutionWorkspace !== null,
|
||||
});
|
||||
const workspaceReuseRequest =
|
||||
resolveExecutionWorkspaceReuseRequestForIssue({
|
||||
issueExecutionWorkspaceId: requestedExecutionWorkspaceId,
|
||||
issueExecutionWorkspacePreference: nativeRecoveryExecutionWorkspaceId
|
||||
? "reuse_existing"
|
||||
: (issueRef?.executionWorkspacePreference ?? null),
|
||||
existingExecutionWorkspaceStatus:
|
||||
existingExecutionWorkspace?.status ?? null,
|
||||
});
|
||||
const requestedShouldReuseExisting =
|
||||
workspaceReuseRequest.requestedShouldReuseExisting;
|
||||
const reusableExistingExecutionWorkspace =
|
||||
workspaceReuseRequest.existingExecutionWorkspaceAvailable
|
||||
? existingExecutionWorkspace
|
||||
: null;
|
||||
const requestedReusableExecutionWorkspaceConfig =
|
||||
reusableExistingExecutionWorkspace?.config ?? null;
|
||||
const localEnvironment = await environmentsSvc.ensureLocalEnvironment(
|
||||
agent.companyId,
|
||||
);
|
||||
const resolvedInstanceSettings = await instanceSettings.get();
|
||||
// Managed-sandbox-only policy: a run that would land on the local
|
||||
// environment is redirected onto the platform-managed sandbox row, and
|
||||
// with no active managed row the resolution fails closed
|
||||
// (ManagedSandboxUnavailableError) — never local. Mirrors the forced
|
||||
// kubernetes execution mode below, which takes precedence when both
|
||||
// regimes are active.
|
||||
const managedSandboxOnly =
|
||||
(await instanceSettings.getExperimental()).enableManagedSandboxOnly ===
|
||||
true;
|
||||
const managedSandboxEnvironment = managedSandboxOnly
|
||||
? await environmentsSvc.findManagedSandboxEnvironment(agent.companyId)
|
||||
: null;
|
||||
const environmentResolution = resolveExecutionWorkspaceEnvironmentId({
|
||||
lowTrustIssueEnvironmentId: trustPreset.kind === "low_trust_review"
|
||||
? parseIssueExecutionWorkspaceSettings(issueContext?.executionWorkspaceSettings, {includeEnvironmentId: true})?.environmentId : null,
|
||||
agentDefaultEnvironmentId: agent.defaultEnvironmentId,
|
||||
instanceDefaultEnvironmentId:
|
||||
resolvedInstanceSettings.defaultEnvironmentId ?? null,
|
||||
localDefaultEnvironmentId: localEnvironment.id,
|
||||
managedSandboxOnly,
|
||||
managedSandboxEnvironmentId: managedSandboxEnvironment?.id ?? null,
|
||||
});
|
||||
const effectiveExecutionWorkspaceMode: ReturnType<
|
||||
typeof resolveExecutionWorkspaceMode
|
||||
> = requestedExecutionWorkspaceMode;
|
||||
const executionPolicy = {
|
||||
executionMode: resolvedInstanceSettings.general.executionMode,
|
||||
// Backstop behind the resolver's local→managed redirect: the run-time
|
||||
// allowlist below fails any run that still resolved to a `local`
|
||||
// environment under managed-sandbox-only, so no selection path or
|
||||
// tenant-set env var can land untrusted execution on the tenant
|
||||
// container.
|
||||
managedSandboxOnly,
|
||||
};
|
||||
const executionForcedToKubernetes =
|
||||
isExecutionForcedToKubernetes(executionPolicy);
|
||||
let selectedEnvironmentId = environmentResolution.environmentId;
|
||||
if (executionForcedToKubernetes) {
|
||||
let kubernetesEnvironment =
|
||||
await environmentsSvc.findKubernetesEnvironment(agent.companyId);
|
||||
if (!kubernetesEnvironment) {
|
||||
// Lazy recovery for companies created after the startup bootstrap ran
|
||||
// (the boot hook only provisions environments for companies that exist
|
||||
// at boot). Re-derive the managed-env config from the bootstrap env.
|
||||
// If the process env no longer forces Kubernetes (rollback / config
|
||||
// drift relative to the persisted executionMode setting), skip the
|
||||
// provisioning gracefully: the guard below still refuses local
|
||||
// fallback with the explicit error, instead of crashing here on
|
||||
// undefined config.
|
||||
let bootstrap: ReturnType<typeof parseExecutionPolicyBootstrapEnv> =
|
||||
null;
|
||||
let bootstrapSkipReason: string | null = null;
|
||||
try {
|
||||
bootstrap = parseExecutionPolicyBootstrapEnv(process.env);
|
||||
if (!bootstrap) {
|
||||
bootstrapSkipReason =
|
||||
'PAPERCLIP_EXECUTION_MODE bootstrap env is not kubernetes-forced (absent or "any")';
|
||||
}
|
||||
} catch (err) {
|
||||
bootstrapSkipReason = `PAPERCLIP_EXECUTION_MODE bootstrap env failed to parse: ${
|
||||
err instanceof Error ? err.message : String(err)
|
||||
}`;
|
||||
}
|
||||
if (bootstrap) {
|
||||
await environmentsSvc.ensureKubernetesEnvironment(
|
||||
agent.companyId,
|
||||
bootstrap.kubernetesConfig,
|
||||
);
|
||||
kubernetesEnvironment =
|
||||
await environmentsSvc.findKubernetesEnvironment(agent.companyId);
|
||||
} else {
|
||||
logger.warn(
|
||||
{
|
||||
runId: run.id,
|
||||
agentId: agent.id,
|
||||
companyId: agent.companyId,
|
||||
reason: bootstrapSkipReason,
|
||||
},
|
||||
"executionMode=kubernetes is persisted but the bootstrap env cannot provision a managed Kubernetes environment; skipping lazy provisioning for this company (the run will fail with the explicit no-managed-environment error)",
|
||||
);
|
||||
}
|
||||
}
|
||||
if (!kubernetesEnvironment) {
|
||||
throw new Error(
|
||||
"Instance execution policy requires the Kubernetes sandbox provider " +
|
||||
"(executionMode=kubernetes) but no managed Kubernetes environment is " +
|
||||
"configured for this company. Configure one (PAPERCLIP_K8S_* env on the " +
|
||||
"cloud instance) before running agents; refusing to fall back to local execution.",
|
||||
);
|
||||
}
|
||||
if (kubernetesEnvironment.id !== selectedEnvironmentId) {
|
||||
logger.info(
|
||||
{
|
||||
runId: run.id,
|
||||
issueId,
|
||||
agentId: agent.id,
|
||||
resolvedEnvironmentId: selectedEnvironmentId,
|
||||
forcedKubernetesEnvironmentId: kubernetesEnvironment.id,
|
||||
},
|
||||
"Forcing run onto the managed Kubernetes environment (executionMode=kubernetes)",
|
||||
);
|
||||
}
|
||||
selectedEnvironmentId = kubernetesEnvironment.id;
|
||||
}
|
||||
const selectedEnvironmentForConfig =
|
||||
selectedEnvironmentId === localEnvironment.id
|
||||
? localEnvironment
|
||||
: selectedEnvironmentId
|
||||
? await environmentsSvc.getById(selectedEnvironmentId)
|
||||
: null;
|
||||
const nativeChatWorkspaceScope = await findNativeChatWorkspaceScope(db, {
|
||||
adapterType: agent.adapterType,
|
||||
environmentDriver: selectedEnvironmentForConfig?.driver ?? null,
|
||||
companyId: agent.companyId,
|
||||
agentId: agent.id,
|
||||
issueId,
|
||||
});
|
||||
const nativeChatExpectedCwd = nativeChatWorkspaceScope
|
||||
? nativeChatWorkspaceCwd(
|
||||
nativeChatWorkspaceScope,
|
||||
reusableExistingExecutionWorkspace,
|
||||
requestedShouldReuseExisting,
|
||||
)
|
||||
: null;
|
||||
if (
|
||||
nativeChatWorkspaceScope &&
|
||||
persistedNativeExecutionInput &&
|
||||
persistedNativeExecutionInput.schema !== "paperclip.native-execution-input.v6" &&
|
||||
!nativeChatWorkspaceMatches({
|
||||
scope: nativeChatWorkspaceScope,
|
||||
expectedCwd: nativeChatExpectedCwd,
|
||||
execution: persistedNativeExecutionInput,
|
||||
})
|
||||
) {
|
||||
// Never rewrite an admitted provider input or release ownership of an
|
||||
// older process whose permissions still include the shared agent home.
|
||||
throw new NativeRunnerOwnershipUnverifiedError(
|
||||
"native_chat_workspace_scope_mismatch",
|
||||
);
|
||||
}
|
||||
if (
|
||||
nativeChatWorkspaceScope &&
|
||||
(!nativeChatExpectedCwd ||
|
||||
executionProjectId !== nativeChatWorkspaceScope.projectId)
|
||||
) {
|
||||
throw new ConfigurationIncompleteFailure(
|
||||
"External chat requires a task-owned isolated workspace. Configure and select an existing isolated worktree for this project task; shared project workspaces cannot be used for external chat.",
|
||||
{
|
||||
configurationIncomplete: {
|
||||
reason: "native_chat_workspace_isolation_required",
|
||||
issueId,
|
||||
},
|
||||
},
|
||||
);
|
||||
}
|
||||
const sharedWorkspaceConcurrency = resolveSharedWorkspaceConcurrency({
|
||||
projectPolicy: projectExecutionWorkspacePolicy,
|
||||
issueSettings: issueExecutionWorkspaceSettings,
|
||||
});
|
||||
// A live holder is always consulted for shared workspaces. Depending on policy and the final
|
||||
// execution target it either remains the existing deferral gate or becomes dispatch context.
|
||||
// Local/SSH folders never take an exclusive workspace lock, including when older
|
||||
// project or issue settings request serialization. Sandbox protection still uses
|
||||
// the existing holder staleness and workspace_busy retry ladder.
|
||||
if (
|
||||
issueRef?.projectWorkspaceId &&
|
||||
effectiveExecutionWorkspaceMode === "shared_workspace"
|
||||
) {
|
||||
const workspaceHolder = await findSharedWorkspaceHolder({
|
||||
companyId: agent.companyId,
|
||||
projectWorkspaceId: issueRef.projectWorkspaceId,
|
||||
excludeIssueId: issueRef.id,
|
||||
excludeRunId: run.id,
|
||||
honorIsolatedWorkspaceModes: isolatedWorkspacesEnabled,
|
||||
});
|
||||
if (workspaceHolder) {
|
||||
const environmentDriver =
|
||||
selectedEnvironmentForConfig?.driver ?? null;
|
||||
const shouldSerialize =
|
||||
sharedWorkspaceConcurrency !== "allow" &&
|
||||
(executionForcedToKubernetes ||
|
||||
(environmentDriver !== "local" &&
|
||||
environmentDriver !== "ssh"));
|
||||
if (shouldSerialize) {
|
||||
throw new WorkspaceBusyDeferral({
|
||||
holder: workspaceHolder,
|
||||
projectWorkspaceId: issueRef.projectWorkspaceId,
|
||||
deferralAttempt:
|
||||
run.scheduledRetryReason === WORKSPACE_BUSY_RETRY_REASON
|
||||
? (run.scheduledRetryAttempt ?? 0)
|
||||
: 0,
|
||||
wasIssueAssignee: issueContext?.assigneeAgentId === agent.id,
|
||||
});
|
||||
}
|
||||
|
||||
const holderIssueLabel =
|
||||
workspaceHolder.issueIdentifier ?? workspaceHolder.issueId;
|
||||
const concurrentWorkspaceNote =
|
||||
`shared workspace is concurrently held by run ${workspaceHolder.runId} (issue ${holderIssueLabel}); ` +
|
||||
"expect concurrent mutations, coordinate via commits";
|
||||
const appendConcurrentWorkspaceNote = (value: unknown) => {
|
||||
const existing = typeof value === "string" ? value.trimEnd() : "";
|
||||
return existing
|
||||
? `${existing}\n${concurrentWorkspaceNote}`
|
||||
: concurrentWorkspaceNote;
|
||||
};
|
||||
context.paperclipTaskMarkdown = appendConcurrentWorkspaceNote(
|
||||
context.paperclipTaskMarkdown,
|
||||
);
|
||||
context.paperclipTaskMarkdownAssignment = appendConcurrentWorkspaceNote(
|
||||
context.paperclipTaskMarkdownAssignment,
|
||||
);
|
||||
if (typeof context.paperclipTaskMarkdownCompact === "string") {
|
||||
context.paperclipTaskMarkdownCompact =
|
||||
appendConcurrentWorkspaceNote(
|
||||
context.paperclipTaskMarkdownCompact,
|
||||
);
|
||||
}
|
||||
if (typeof context.paperclipTaskMarkdownAssignmentCompact === "string") {
|
||||
context.paperclipTaskMarkdownAssignmentCompact =
|
||||
appendConcurrentWorkspaceNote(
|
||||
context.paperclipTaskMarkdownAssignmentCompact,
|
||||
);
|
||||
}
|
||||
logger.info(
|
||||
{
|
||||
event: "shared_workspace_concurrent_dispatch",
|
||||
runId: run.id,
|
||||
issueId: issueRef.id,
|
||||
projectWorkspaceId: issueRef.projectWorkspaceId,
|
||||
holderRunId: workspaceHolder.runId,
|
||||
holderIssueId: workspaceHolder.issueId,
|
||||
sharedWorkspaceConcurrency,
|
||||
environmentDriver,
|
||||
executionForcedToKubernetes,
|
||||
},
|
||||
"Dispatching alongside a live shared-workspace holder",
|
||||
);
|
||||
}
|
||||
}
|
||||
const useIsolatedTaskDirectory = issueRef !== null && shouldUseIsolatedTaskDirectory({
|
||||
trustPreset: trustPreset.kind,
|
||||
environmentDriver: selectedEnvironmentForConfig?.driver ?? null,
|
||||
mode: requestedExecutionWorkspaceMode,
|
||||
hasProjectWorkspace: projectContext?.hasWorkspace ?? false,
|
||||
projectWorkspaceId: issueRef.projectWorkspaceId,
|
||||
workspaceStrategies: [
|
||||
config.workspaceStrategy,
|
||||
issueAssigneeOverrides?.adapterConfig?.workspaceStrategy,
|
||||
projectExecutionWorkspacePolicy?.workspaceStrategy,
|
||||
issueExecutionWorkspaceSettings?.workspaceStrategy,
|
||||
],
|
||||
});
|
||||
return {
|
||||
persistedRunnerProfile,
|
||||
persistedNativeExecutionInput,
|
||||
isDotRun,
|
||||
existingExecutionWorkspace,
|
||||
nativeRecoveryExecutionWorkspaceId,
|
||||
workspaceReuseRequest,
|
||||
requestedShouldReuseExisting,
|
||||
reusableExistingExecutionWorkspace,
|
||||
requestedReusableExecutionWorkspaceConfig,
|
||||
localEnvironment,
|
||||
resolvedInstanceSettings,
|
||||
environmentResolution,
|
||||
effectiveExecutionWorkspaceMode,
|
||||
executionPolicy,
|
||||
selectedEnvironmentId,
|
||||
selectedEnvironmentForConfig,
|
||||
nativeChatWorkspaceScope,
|
||||
nativeChatExpectedCwd,
|
||||
useIsolatedTaskDirectory,
|
||||
};
|
||||
}
|
||||
@@ -12,6 +12,7 @@ const api = async (route, method = "GET", body) => {
|
||||
return response.json();
|
||||
};
|
||||
const run = await api(`/heartbeat-runs/${env.PAPERCLIP_RUN_ID}`);
|
||||
assert.equal(run.contextSnapshot.paperclipEnvironment.driver, "local");
|
||||
const issueId = run.contextSnapshot.issueId;
|
||||
const repositories = await readdir(path.join(process.cwd(), ".paperclip-repositories"));
|
||||
const repo = path.join(process.cwd(), ".paperclip-repositories", repositories.find((name) => !name.includes(".clone-")));
|
||||
|
||||
Reference in new issue
Block a user