mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-22 00:30:19 +02:00
Merge branch 'codex/boat-environment-core' into codex/boat-environment
* codex/boat-environment-core: perf(chat): stop polling provider action and session sync queues (#15822) refactor(heartbeat): extract environment selection (#15825) perf(runner): compile the TypeScript surface once per build (#15853)
This commit is contained in:
commit
add1bea3ad
19 files changed
+1108
-510
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`.
|
||||
|
||||
@@ -434,9 +434,19 @@ Longer polling intervals alone do not complete an item.
|
||||
Startup restores existing rows, and empty workers disarm. No new tables.
|
||||
Tests cover an idle hour, real producer notifications, PostgreSQL deadline
|
||||
selection, stale claims, outer-transaction wakes, and callback-driven setup.
|
||||
Remaining **1s** maintenance: provider runtimes, other provider action
|
||||
outboxes, run milestone/wake-notice projection, GitHub operation recovery,
|
||||
Teams protocol maintenance, and Slack session sync. GitHub pre-ingress recovery
|
||||
Three more existing action queues now use commit/deadline workers: provider
|
||||
messages/notices, accepted GitHub webhook receipts, and Slack session sync.
|
||||
Foreground claims, retries, operator resolution, and settlement writes signal
|
||||
in their transaction. Shared predicates select pending rows and deadlines;
|
||||
terminal history and uncertain provider sends have no deadline. Stale provider
|
||||
sends remain quarantined, while GitHub receipts retain bounded replay and
|
||||
their existing attempt fences. Slack sync retains the 30-minute refresh for
|
||||
running sessions and pauses without rescheduling until an endpoint change.
|
||||
Tests exercise real producers with deferred fast paths disabled, retry and
|
||||
startup recovery, stale-send quarantine, and zero scans after settlement.
|
||||
Remaining **1s** maintenance: provider runtimes, Telegram maintenance, failed
|
||||
chat-run retries, run milestone/wake-notice projection, GitHub operation
|
||||
recovery, and Teams protocol maintenance. GitHub pre-ingress recovery
|
||||
must still inspect provider history: a lost callback has no local queue row.
|
||||
This slice does **not** yet make the whole chat subsystem quiet. Independent
|
||||
DB roots/direct SQL require startup recovery, as with the other migrated queues.
|
||||
|
||||
@@ -63,7 +63,7 @@
|
||||
},
|
||||
"sideEffects": false,
|
||||
"scripts": {
|
||||
"build": "pnpm run check:protocol-manifest && pnpm run build:typescript && pnpm run check:capability-contract && pnpm run check:capability-inventory && pnpm run check:protocol-coverage && pnpm run check:semantic-contracts && pnpm run check:runner-workflow-traceability && pnpm run build:binary && node scripts/generate-replay-goldens.mjs --check && node scripts/generate-semantic-action-catalog.mjs --check",
|
||||
"build": "pnpm run check:protocol-manifest && pnpm run build:typescript && pnpm run check:capability-contract && pnpm run check:capability-inventory && pnpm run check:protocol-coverage && node scripts/generate-semantic-contracts.mjs --check && node scripts/check-runner-workflow-traceability.mjs && pnpm run build:binary && node scripts/generate-replay-goldens.mjs --check && node scripts/generate-semantic-action-catalog.mjs --check",
|
||||
"build:typescript": "pnpm run check:acpx-profiles && pnpm run ensure:eval-build-deps && pnpm run check:protocol-types && node ./node_modules/typescript/bin/tsc --version && node ./node_modules/typescript/bin/tsc -p tsconfig.json && node ./node_modules/typescript/bin/tsc -p tsconfig.surfaces.json && node scripts/build-verified-provider-entrypoints.mjs",
|
||||
"build:rust": "cargo build --manifest-path runner/Cargo.toml --locked --workspace --bins",
|
||||
"build:binary": "cargo build --release --manifest-path runner/Cargo.toml --locked -p paperclip-runner-core --bin paperclip-runnerd && node scripts/stage-runner-binary.mjs",
|
||||
|
||||
@@ -106,6 +106,12 @@ describe("createChatReconciliationCoordinator", () => {
|
||||
// The service integration tests hold real publication workers while this
|
||||
// scheduled method returns; app shutdown must also join those workers.
|
||||
expect(source).toContain("await chatChannels.shutdown()");
|
||||
expect(source).not.toContain("chatChannels.processPendingSlackSessionSyncs()");
|
||||
const service = readFileSync(new URL("../services/chat-channels.ts", import.meta.url), "utf8");
|
||||
const actionMaintenance = service.slice(service.indexOf("async function processPendingChatMaintenance("),
|
||||
service.indexOf("async function processPendingDeliveries("));
|
||||
expect(actionMaintenance).not.toContain("processPendingProviderEffects(");
|
||||
expect(actionMaintenance).not.toContain("processPendingGitHubWebhookIngress(");
|
||||
});
|
||||
|
||||
it.each([false, true])(
|
||||
|
||||
@@ -19,6 +19,9 @@ function setup() {
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner, canRun: enabled, canReconcile: enabled, onError });
|
||||
coordinators.push(coordinator);
|
||||
const service = {
|
||||
processPendingProviderEffects: vi.fn(async () => {}),
|
||||
processPendingGitHubWebhookIngress: vi.fn(async () => {}),
|
||||
processPendingSlackSessionSyncs: vi.fn(async () => {}),
|
||||
processPendingSlackBoardMessages: vi.fn(async () => {}),
|
||||
processPendingSlackTaskStarts: vi.fn(async () => {}),
|
||||
processPendingReceiptReactions: vi.fn(async () => {}),
|
||||
@@ -27,7 +30,7 @@ function setup() {
|
||||
nextChatActionAt: vi.fn(async (): Promise<number | null> => null),
|
||||
nextVerificationMessageAt: vi.fn(async (): Promise<number | null> => null),
|
||||
};
|
||||
const runs = [service.processPendingSlackBoardMessages, service.processPendingSlackTaskStarts,
|
||||
const runs = [service.processPendingProviderEffects, service.processPendingGitHubWebhookIngress, service.processPendingSlackSessionSyncs, service.processPendingSlackBoardMessages, service.processPendingSlackTaskStarts,
|
||||
service.processPendingReceiptReactions, service.processPendingSlackSessionStops, service.processPendingVerificationMessages];
|
||||
return { owner, service, runs, coordinator, enabled, onError, start: () => registerChatActionWork(coordinator, service, enabled) };
|
||||
}
|
||||
@@ -37,7 +40,7 @@ describe("chat action scheduling", () => {
|
||||
await s.start().ready;
|
||||
await vi.advanceTimersByTimeAsync(3_600_000);
|
||||
for (const run of s.runs) expect(run).toHaveBeenCalledTimes(1);
|
||||
expect(s.service.nextChatActionAt).toHaveBeenCalledTimes(4);
|
||||
expect(s.service.nextChatActionAt).toHaveBeenCalledTimes(7);
|
||||
expect(s.service.nextVerificationMessageAt).toHaveBeenCalledTimes(1);
|
||||
expect(s.coordinator.nextWakeAt()).toBeNull();
|
||||
expect(vi.getTimerCount()).toBe(0);
|
||||
|
||||
@@ -13250,6 +13250,49 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
await service.shutdown();
|
||||
});
|
||||
|
||||
it("drains accepted GitHub webhooks from commit wakes and durable retry deadlines", async () => {
|
||||
const fixture = await seedCompany();
|
||||
const { endpoint, runtime, service, webhookSecret } = await configuredGitHubEndpoint(fixture, {
|
||||
deferWebhookProcessing: true, scheduleDeferredWork: () => undefined,
|
||||
});
|
||||
const providerRuntime = runtime.endpoints.get(endpoint.id)!;
|
||||
let attempts = 0;
|
||||
providerRuntime.webhookHook = async () => {
|
||||
providerRuntime.webhookResponse = new Response("result", { status: ++attempts === 1 ? 503 : 202 });
|
||||
};
|
||||
const delivery = randomUUID();
|
||||
const send = () => service.handleWebhook(endpoint.publicId, "github", signedGitHubWebhookRequest({
|
||||
delivery, event: "installation_repositories", payload: { action: "added", installation: { id: 2468 } }, webhookSecret,
|
||||
}));
|
||||
const onError = vi.fn();
|
||||
let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
|
||||
const receipt = () => db.select().from(chatActions).where(and(eq(chatActions.endpointId, endpoint.id),
|
||||
eq(chatActions.providerActionId, `github_webhook_ingress:${delivery}`))).then(rows => rows[0]);
|
||||
try {
|
||||
await registerChatActionWork(coordinator, service, () => true).ready;
|
||||
expect((await send()).status).toBe(202);
|
||||
await expect.poll(async () => (await receipt())?.status, { timeout: 10_000 }).toBe("processed");
|
||||
expect(attempts).toBe(2);
|
||||
expect((await receipt())?.result).toMatchObject({ attempts: 2, httpStatus: 202 });
|
||||
// A duplicate callback wakes the lane but cannot repeat a settled effect.
|
||||
await send();
|
||||
await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 10_000 });
|
||||
expect(attempts).toBe(2);
|
||||
await coordinator.stop();
|
||||
// Work committed while no worker exists is recovered by the next startup.
|
||||
const restartDelivery = randomUUID();
|
||||
await service.handleWebhook(endpoint.publicId, "github", signedGitHubWebhookRequest({
|
||||
delivery: restartDelivery, event: "installation_repositories",
|
||||
payload: { action: "added", installation: { id: 2468 } }, webhookSecret,
|
||||
}));
|
||||
expect(attempts).toBe(2);
|
||||
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
|
||||
await registerChatActionWork(coordinator, service, () => true).ready;
|
||||
await expect.poll(() => attempts, { timeout: 10_000 }).toBe(3);
|
||||
expect(onError).not.toHaveBeenCalled();
|
||||
} finally { await coordinator.stop(); }
|
||||
});
|
||||
|
||||
it("safely re-arms an exact terminal GitHub delivery after an operator redelivery", async () => {
|
||||
const fixture = await seedCompany();
|
||||
const { endpoint, runtime, service, webhookSecret } =
|
||||
@@ -32819,7 +32862,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
.update(chatEndpoints)
|
||||
.set({ status: "active" })
|
||||
.where(eq(chatEndpoints.id, endpoint.id));
|
||||
const channelId = `C-LANE-${endpoint.id}`;
|
||||
const channelId = `C${endpoint.id.replaceAll("-", "").toUpperCase()}`;
|
||||
await db.insert(chatEndpointResources).values({
|
||||
companyId: fixture.companyId,
|
||||
endpointId: endpoint.id,
|
||||
@@ -32892,6 +32935,45 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
return { ...fixture, storage, runtime, service, lanes, enqueue, cleanup };
|
||||
}
|
||||
|
||||
it("wakes Slack session sync from publication commit and leaves settled action queues quiet", async () => {
|
||||
const f = await publicationLaneFixture(1);
|
||||
const lane = f.lanes[0]!;
|
||||
const onError = vi.fn();
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
|
||||
try {
|
||||
await registerChatActionWork(coordinator, f.service, () => true).ready;
|
||||
await f.enqueue(0, "A completed answer");
|
||||
await f.service.processPendingPublications();
|
||||
await expect.poll(async () => db.select({ status: chatActions.status }).from(chatActions)
|
||||
.where(and(eq(chatActions.endpointId, lane.endpoint.id), eq(chatActions.kind, "slack_session_sync"))),
|
||||
{ timeout: 10_000 }).toEqual([{ status: "processed" }]);
|
||||
await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 10_000 });
|
||||
const scans = vi.spyOn(db, "select");
|
||||
try {
|
||||
await new Promise(resolve => setTimeout(resolve, 1200));
|
||||
expect(scans).not.toHaveBeenCalled();
|
||||
} finally { scans.mockRestore(); }
|
||||
expect(onError).not.toHaveBeenCalled();
|
||||
} finally { await coordinator.stop(); await f.cleanup(); }
|
||||
});
|
||||
|
||||
it("recovers stale provider sends on startup without replaying uncertain effects", async () => {
|
||||
const f = await publicationLaneFixture(1);
|
||||
const lane = f.lanes[0]!;
|
||||
const [action] = await db.insert(chatActions).values({ companyId: f.companyId, endpointId: lane.endpoint.id,
|
||||
kind: "provider_effect", providerActionId: randomUUID(), status: "processing", result: { attempts: 1 },
|
||||
updatedAt: new Date(Date.now() - 120_000), payload: {} }).returning();
|
||||
const before = lane.providerRuntime.posts.length;
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true });
|
||||
try {
|
||||
await registerChatActionWork(coordinator, f.service, () => true).ready;
|
||||
await expect.poll(async () => db.select({ status: chatActions.status }).from(chatActions)
|
||||
.where(eq(chatActions.id, action!.id)), { timeout: 10_000 }).toEqual([{ status: "delivery_unknown" }]);
|
||||
expect(await f.service.nextChatActionAt("provider_effect")).toBeNull();
|
||||
expect(lane.providerRuntime.posts).toHaveLength(before);
|
||||
} finally { await coordinator.stop(); await f.cleanup(); }
|
||||
});
|
||||
|
||||
it("restores action retry and stale-claim deadlines and parks paused or uncertain work", async () => {
|
||||
const f = await publicationLaneFixture(1);
|
||||
const lane = f.lanes[0]!;
|
||||
@@ -32929,6 +33011,21 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
await set("delivery_unknown");
|
||||
expect(await f.service.nextChatActionAt(kind)).toBeNull();
|
||||
expect(await due()).toBe(false);
|
||||
} else if (kind === "slack_session_sync") {
|
||||
expect(await due()).toBe(true);
|
||||
await set("received", { retryAt: later.toISOString() });
|
||||
expect(await f.service.nextChatActionAt(kind)).toBe(later.getTime());
|
||||
expect(await due()).toBe(false);
|
||||
await set("processing");
|
||||
expect(await f.service.nextChatActionAt(kind)).toBe(at.getTime() + 60_000);
|
||||
for (const status of ["paused", "attention"]) {
|
||||
await db.update(chatEndpoints).set({ status }).where(eq(chatEndpoints.id, lane.endpoint.id));
|
||||
expect(await f.service.nextChatActionAt(kind)).toBeNull();
|
||||
expect(await due()).toBe(false);
|
||||
}
|
||||
await db.update(chatEndpoints).set({ status: "active" }).where(eq(chatEndpoints.id, lane.endpoint.id));
|
||||
await set("failed", { retryable: true });
|
||||
expect(await f.service.nextChatActionAt(kind)).toBeNull();
|
||||
} else {
|
||||
expect(await due()).toBe(true);
|
||||
await set("processing");
|
||||
@@ -53999,6 +54096,75 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
);
|
||||
});
|
||||
|
||||
it("delivers queued provider notices from commit wakes without the deferred fast path", async () => {
|
||||
const fixture = await seedCompany();
|
||||
const { callbacks, endpoint, service, runtime } =
|
||||
await configuredSlackEndpoint(fixture, { scheduleDeferredWork: () => undefined });
|
||||
if (!callbacks.onSlashCommand) {
|
||||
throw new Error("Slack slash command callback was not registered");
|
||||
}
|
||||
await db.insert(chatEndpointResources).values({
|
||||
companyId: fixture.companyId,
|
||||
endpointId: endpoint.id,
|
||||
type: "channel",
|
||||
providerResourceId: "C-CONTROL-GUIDANCE",
|
||||
label: "control-guidance",
|
||||
availability: "available",
|
||||
enabled: true,
|
||||
});
|
||||
const command = endpoint.setup.command;
|
||||
if (!command) throw new Error("Slack endpoint did not expose its command");
|
||||
const post = vi.fn(async () => ({
|
||||
id: "unexpected-channel-control-post",
|
||||
threadId: "slack:C-CONTROL-GUIDANCE:root",
|
||||
}));
|
||||
const postEphemeral = vi.fn(async () => ({
|
||||
id: "channel-control-guidance",
|
||||
threadId: "slack:C-CONTROL-GUIDANCE:root",
|
||||
}));
|
||||
const providerRuntime = runtime.endpoints.get(endpoint.id)!;
|
||||
vi.spyOn(providerRuntime, "thread").mockReturnValue({ post, postEphemeral } as never);
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true });
|
||||
onTestFinished(async () => { await coordinator.stop(); });
|
||||
await registerChatActionWork(coordinator, service, () => true).ready;
|
||||
for (const control of ["status"] as const) {
|
||||
await callbacks.onSlashCommand({
|
||||
endpointId: endpoint.id,
|
||||
provider: "slack",
|
||||
event: {
|
||||
channel: {
|
||||
id: "slack:C-CONTROL-GUIDANCE",
|
||||
name: "control-guidance",
|
||||
isDM: false,
|
||||
post,
|
||||
postEphemeral,
|
||||
} as never,
|
||||
command,
|
||||
text: control,
|
||||
triggerId: `channel-control-${control}`,
|
||||
user: {
|
||||
userId: "U-CHANNEL-CONTROLLER",
|
||||
userName: "channel-controller",
|
||||
fullName: "Channel Controller",
|
||||
isBot: false,
|
||||
isMe: false,
|
||||
isSystem: false,
|
||||
},
|
||||
raw: { command, text: control },
|
||||
adapter: {} as never,
|
||||
openModal: async () => undefined,
|
||||
},
|
||||
});
|
||||
}
|
||||
await expect.poll(async () => db.select({ status: chatActions.status }).from(chatActions)
|
||||
.where(and(eq(chatActions.endpointId, endpoint.id), eq(chatActions.kind, "provider_effect"))),
|
||||
{ timeout: 10_000 }).toEqual([{ status: "processed" }]);
|
||||
expect(postEphemeral).toHaveBeenCalledTimes(1);
|
||||
expect(post).not.toHaveBeenCalled();
|
||||
expect(await service.nextChatActionAt("provider_effect")).toBeNull();
|
||||
await coordinator.stop();
|
||||
});
|
||||
|
||||
it("returns ephemeral guidance for exact Slack controls in channels without creating tasks or actions", async () => {
|
||||
const fixture = await seedCompany();
|
||||
const { callbacks, endpoint, service } =
|
||||
@@ -54438,6 +54604,9 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
const fixture = await seedCompany();
|
||||
const { callbacks, endpoint, runtime, service } =
|
||||
await configuredTelegramEndpoint(fixture);
|
||||
const workWake = vi.fn();
|
||||
const stopWorkObserver = subscribeDeliveryWork(db, DELIVERY_QUEUES.chatProviderEffects, workWake);
|
||||
onTestFinished(() => { stopWorkObserver(); expect(workWake).toHaveBeenCalled(); });
|
||||
if (!callbacks.onSlashCommand) {
|
||||
throw new Error("Telegram slash command callback was not registered");
|
||||
}
|
||||
|
||||
@@ -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" },
|
||||
|
||||
@@ -103,6 +103,36 @@ describe("server package build script", () => {
|
||||
);
|
||||
});
|
||||
|
||||
it("compiles the vendored Runner's TypeScript surface once per build", () => {
|
||||
// prepare:runner-vendor runs the Runner's full build, and `pnpm -r build`
|
||||
// has already run it once by the time the server builds. Each extra
|
||||
// build:typescript inside that chain (two tsc projects plus the eval-kernel
|
||||
// rebuild) costs about 7s on the CI Build job's critical path, twice.
|
||||
// Checks that read dist/ must call their script directly, the way
|
||||
// generate-replay-goldens.mjs already does, not through a `check:*`
|
||||
// wrapper that rebuilds first for standalone use.
|
||||
const runner = JSON.parse(readFileSync(new URL(
|
||||
"../../../packages/paperclip-runner/package.json", import.meta.url,
|
||||
), "utf8")) as { scripts: Record<string, string> };
|
||||
const steps = runner.scripts.build.split(" && ");
|
||||
const wrapped = steps
|
||||
.filter((step) => step.startsWith("pnpm run "))
|
||||
.map((step) => step.slice("pnpm run ".length));
|
||||
for (const name of wrapped) {
|
||||
expect(runner.scripts[name], `build step ${name}`).toBeDefined();
|
||||
}
|
||||
const compiles = (step: string) =>
|
||||
step === "build:typescript" || (runner.scripts[step] ?? "").includes("build:typescript");
|
||||
|
||||
expect(wrapped.filter(compiles)).toEqual(["build:typescript"]);
|
||||
expect(steps).toContain("node scripts/generate-semantic-contracts.mjs --check");
|
||||
expect(steps).toContain("node scripts/check-runner-workflow-traceability.mjs");
|
||||
expect(steps.indexOf("node scripts/generate-semantic-contracts.mjs --check"))
|
||||
.toBeGreaterThan(steps.indexOf("pnpm run build:typescript"));
|
||||
expect(steps.indexOf("node scripts/check-runner-workflow-traceability.mjs"))
|
||||
.toBeGreaterThan(steps.indexOf("pnpm run build:typescript"));
|
||||
});
|
||||
|
||||
it("ships the pinned OpenCode dependency resolved by the vendored Runner", () => {
|
||||
const server = JSON.parse(readFileSync(packageJsonPath, "utf8"));
|
||||
const runner = JSON.parse(readFileSync(new URL(
|
||||
|
||||
+2
-4
@@ -327,7 +327,7 @@ export function createChatReconciliationCoordinator(input: {
|
||||
projectRunMilestones: () => Promise<number>;
|
||||
flushPublications: () => Promise<unknown>;
|
||||
processPendingSlackFileUploadReceipts?: () => Promise<unknown>;
|
||||
processPendingSlackSessionSyncs: () => Promise<unknown>;
|
||||
processPendingSlackSessionSyncs?: () => Promise<unknown>;
|
||||
onError: (lane: ChatReconciliationLane, error: unknown) => void;
|
||||
}) {
|
||||
let stopped = false;
|
||||
@@ -374,7 +374,7 @@ export function createChatReconciliationCoordinator(input: {
|
||||
milestoneReconciliation.poll();
|
||||
publicationReconciliation.poll();
|
||||
if (input.processPendingSlackFileUploadReceipts) start("Slack file receipts", input.processPendingSlackFileUploadReceipts);
|
||||
start("Slack session status", input.processPendingSlackSessionSyncs);
|
||||
if (input.processPendingSlackSessionSyncs) start("Slack session status", input.processPendingSlackSessionSyncs);
|
||||
},
|
||||
notifyPublications() {
|
||||
milestoneReconciliation.notify();
|
||||
@@ -1273,8 +1273,6 @@ export async function createApp(
|
||||
publicBaseUrl: opts.authPublicBaseUrl,
|
||||
}),
|
||||
flushPublications: () => reconcileChatPublicationMaintenance(),
|
||||
processPendingSlackSessionSyncs: () =>
|
||||
chatChannels.processPendingSlackSessionSyncs(),
|
||||
onError: (lane, err) => {
|
||||
logger.error({ err, lane }, `Failed to reconcile chat ${lane}`);
|
||||
},
|
||||
|
||||
@@ -23,6 +23,10 @@ export function chatActionDeadline(kind: ScheduledChatAction) {
|
||||
when ${chatActions.status} = 'admitting' then ${updated} + ${SLACK_COMMAND_ADMISSION_STALE_MS}
|
||||
when ${chatActions.status} = 'resolving' then ${updated} + ${SLACK_COMMAND_EXPLICIT_RETRY_STALE_MS}
|
||||
when ${chatActions.status} = 'validating' then ${updated} + ${SLACK_COMMAND_POST_STALE_MS} end`;
|
||||
if (kind === "slack_session_sync") return sql<number>`case
|
||||
when ${chatEndpoints.status} in ('paused', 'attention') then null
|
||||
when ${chatActions.status} = 'received' then ${retry}
|
||||
when ${chatActions.status} = 'processing' then ${updated} + ${PROVIDER_EFFECT_STALE_MS} end`;
|
||||
return sql<number>`case
|
||||
when ${chatActions.status} = 'received' then 0
|
||||
when ${chatActions.status} = 'processing' then ${updated} + ${PROVIDER_EFFECT_STALE_MS}
|
||||
@@ -30,13 +34,15 @@ export function chatActionDeadline(kind: ScheduledChatAction) {
|
||||
}
|
||||
|
||||
function pendingChatAction(kind: ScheduledChatAction) {
|
||||
// Filter history before the endpoint join and aggregate. Failed reactions
|
||||
// and stops only remain live when the saved result permits another attempt.
|
||||
// Filter history before the endpoint join and aggregate. Failed actions
|
||||
// only remain live when the saved result permits another attempt.
|
||||
const status = kind === "slack_board_message"
|
||||
? eq(chatActions.status, "received")
|
||||
: kind === "slash_task_start"
|
||||
? inArray(chatActions.status, ["queued", "provider_confirmed", "admitting", "resolving", "validating"])
|
||||
: or(inArray(chatActions.status, ["received", "processing"]),
|
||||
: kind === "slack_session_sync"
|
||||
? inArray(chatActions.status, ["received", "processing"])
|
||||
: or(inArray(chatActions.status, ["received", "processing"]),
|
||||
and(eq(chatActions.status, "failed"), sql`${chatActions.result}->>'retryable' = 'true'`));
|
||||
return and(eq(chatActions.kind, kind), status);
|
||||
}
|
||||
@@ -54,6 +60,9 @@ export async function nextChatActionAt(db: Db, kind: ScheduledChatAction): Promi
|
||||
}
|
||||
|
||||
type ChatActionService = {
|
||||
processPendingProviderEffects(): Promise<unknown>;
|
||||
processPendingGitHubWebhookIngress(): Promise<unknown>;
|
||||
processPendingSlackSessionSyncs(): Promise<unknown>;
|
||||
processPendingSlackBoardMessages(): Promise<unknown>;
|
||||
processPendingSlackTaskStarts(): Promise<unknown>;
|
||||
processPendingReceiptReactions(): Promise<unknown>;
|
||||
@@ -69,6 +78,9 @@ export function registerChatActionWork(
|
||||
canRun: () => boolean,
|
||||
) {
|
||||
const actions = [
|
||||
{ kind: "provider_effect", run: service.processPendingProviderEffects },
|
||||
{ kind: "github_webhook_ingress", run: service.processPendingGitHubWebhookIngress },
|
||||
{ kind: "slack_session_sync", run: service.processPendingSlackSessionSyncs },
|
||||
{ kind: "slack_board_message", run: service.processPendingSlackBoardMessages },
|
||||
{ kind: "slash_task_start", run: service.processPendingSlackTaskStarts },
|
||||
{ kind: "receipt_reaction", run: service.processPendingReceiptReactions },
|
||||
|
||||
@@ -4581,6 +4581,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
runtimeContext: LifecycleRuntimeFence;
|
||||
},
|
||||
) {
|
||||
await notifyChatActionWork(database, "provider_effect");
|
||||
const [inserted] = await database
|
||||
.insert(chatActions)
|
||||
.values({
|
||||
@@ -4718,6 +4719,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
["verifying", "active"],
|
||||
)
|
||||
: null;
|
||||
await notifyChatActionWork(tx, "provider_effect");
|
||||
const [ownedAction] = await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
@@ -4819,6 +4821,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
const failure =
|
||||
"Provider delivery could not be confirmed after the worker stopped. The effect will not be replayed automatically.";
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyChatActionWork(tx, "provider_effect");
|
||||
const [quarantined] = await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
@@ -5093,6 +5096,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
action: typeof chatActions.$inferSelect,
|
||||
attempt: number,
|
||||
): Promise<boolean> {
|
||||
await notifyChatActionWork(tx, "provider_effect");
|
||||
const [cancelled] = await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
@@ -5161,14 +5165,17 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
if (!action) return "failed";
|
||||
const payload = providerEffectPayload(action.payload);
|
||||
if (!payload) {
|
||||
await db
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "failed",
|
||||
result: { code: "provider_effect_payload_invalid" },
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(chatActions.id, action.id));
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyChatActionWork(tx, "provider_effect");
|
||||
return tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "failed",
|
||||
result: { code: "provider_effect_payload_invalid" },
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(chatActions.id, action.id));
|
||||
});
|
||||
return "failed";
|
||||
}
|
||||
if (action.status === "processed") return "processed";
|
||||
@@ -5238,6 +5245,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
payload,
|
||||
)
|
||||
: false;
|
||||
await notifyChatActionWork(tx, "provider_effect");
|
||||
const [transportClaim] = await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
@@ -5355,6 +5363,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
},
|
||||
["verifying", "active"],
|
||||
);
|
||||
await notifyChatActionWork(tx, "provider_effect");
|
||||
const [completed] = await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
@@ -5481,23 +5490,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
const actions = await db
|
||||
.select()
|
||||
.from(chatActions)
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.kind, "provider_effect"),
|
||||
or(
|
||||
eq(chatActions.status, "received"),
|
||||
and(
|
||||
eq(chatActions.status, "processing"),
|
||||
lte(chatActions.updatedAt, staleBefore),
|
||||
),
|
||||
and(
|
||||
eq(chatActions.status, "failed"),
|
||||
sql`coalesce(${chatActions.result}->>'retryable', 'false') = 'true'`,
|
||||
sql`(${chatActions.result}->>'retryAt' is null or (${chatActions.result}->>'retryAt')::timestamptz <= ${now.toISOString()}::timestamptz)`,
|
||||
),
|
||||
),
|
||||
),
|
||||
)
|
||||
.where(dueChatAction("provider_effect", now.getTime()))
|
||||
.orderBy(asc(chatActions.createdAt))
|
||||
.limit(limit);
|
||||
for (const action of actions) {
|
||||
@@ -25203,6 +25196,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
runtimeGeneration: context.generation,
|
||||
credentialFingerprint: context.credentialFingerprint,
|
||||
};
|
||||
await notifyChatActionWork(tx, "github_webhook_ingress");
|
||||
const inserted = await tx
|
||||
.insert(chatActions)
|
||||
.values({
|
||||
@@ -25229,6 +25223,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
// currently authenticated body after a terminal processing failure.
|
||||
// Successful, cancelled, retryable, and in-flight receipts remain
|
||||
// immutable deduplication fences.
|
||||
await notifyChatActionWork(tx, "github_webhook_ingress");
|
||||
const [rearmed] = await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
@@ -25312,32 +25307,35 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
Date.now() +
|
||||
Math.min(60_000, 1_000 * 2 ** Math.max(0, input.attempts - 1)),
|
||||
);
|
||||
await db
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: cancelled ? "cancelled" : processed ? "processed" : "failed",
|
||||
payload: terminal
|
||||
? redactedGitHubWebhookIngressPayload(input.payload)
|
||||
: input.payload,
|
||||
result: {
|
||||
attempts: input.attempts,
|
||||
...(input.cancelledCode ? { code: input.cancelledCode } : {}),
|
||||
...(input.response
|
||||
? { httpStatus: input.response.status }
|
||||
: { error: redactError(input.error) }),
|
||||
retryable: !terminal,
|
||||
...(!terminal ? { retryAt: retryAt.toISOString() } : {}),
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, actionId),
|
||||
eq(chatActions.kind, "github_webhook_ingress"),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`(${chatActions.result}->>'attempts')::int = ${input.attempts}`,
|
||||
),
|
||||
);
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyChatActionWork(tx, "github_webhook_ingress");
|
||||
return tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: cancelled ? "cancelled" : processed ? "processed" : "failed",
|
||||
payload: terminal
|
||||
? redactedGitHubWebhookIngressPayload(input.payload)
|
||||
: input.payload,
|
||||
result: {
|
||||
attempts: input.attempts,
|
||||
...(input.cancelledCode ? { code: input.cancelledCode } : {}),
|
||||
...(input.response
|
||||
? { httpStatus: input.response.status }
|
||||
: { error: redactError(input.error) }),
|
||||
retryable: !terminal,
|
||||
...(!terminal ? { retryAt: retryAt.toISOString() } : {}),
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, actionId),
|
||||
eq(chatActions.kind, "github_webhook_ingress"),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`(${chatActions.result}->>'attempts')::int = ${input.attempts}`,
|
||||
),
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
async function processGitHubWebhookIngress(
|
||||
@@ -25361,18 +25359,21 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
return githubIngressAcceptedResponse();
|
||||
const payload = githubWebhookIngressPayload(action.payload);
|
||||
if (!payload) {
|
||||
await db
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "failed",
|
||||
payload: {},
|
||||
result: {
|
||||
code: "github_webhook_ingress_payload_invalid",
|
||||
retryable: false,
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(chatActions.id, action.id));
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyChatActionWork(tx, "github_webhook_ingress");
|
||||
return tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "failed",
|
||||
payload: {},
|
||||
result: {
|
||||
code: "github_webhook_ingress_payload_invalid",
|
||||
retryable: false,
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(chatActions.id, action.id));
|
||||
});
|
||||
return githubIngressAcceptedResponse();
|
||||
}
|
||||
const now = new Date();
|
||||
@@ -25407,21 +25408,24 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
sql`coalesce((${chatActions.result}->>'attempts')::int, 0) = ${previousAttempts}`,
|
||||
)
|
||||
: eq(chatActions.status, "received");
|
||||
const [claimed] = await db
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "processing",
|
||||
result: { attempts },
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, action.id),
|
||||
eq(chatActions.kind, "github_webhook_ingress"),
|
||||
claimGuard,
|
||||
),
|
||||
)
|
||||
.returning();
|
||||
const [claimed] = await db.transaction(async (tx) => {
|
||||
await notifyChatActionWork(tx, "github_webhook_ingress");
|
||||
return tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "processing",
|
||||
result: { attempts },
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, action.id),
|
||||
eq(chatActions.kind, "github_webhook_ingress"),
|
||||
claimGuard,
|
||||
),
|
||||
)
|
||||
.returning();
|
||||
});
|
||||
if (!claimed) return null;
|
||||
action = claimed;
|
||||
|
||||
@@ -26192,28 +26196,13 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
onlyActionId?: string,
|
||||
) {
|
||||
const now = new Date();
|
||||
const staleBefore = new Date(now.getTime() - PROVIDER_EFFECT_STALE_MS);
|
||||
const actions = await db
|
||||
.select({ id: chatActions.id })
|
||||
.from(chatActions)
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.kind, "github_webhook_ingress"),
|
||||
...(onlyActionId ? [eq(chatActions.id, onlyActionId)] : []),
|
||||
or(
|
||||
eq(chatActions.status, "received"),
|
||||
and(
|
||||
eq(chatActions.status, "processing"),
|
||||
lte(chatActions.updatedAt, staleBefore),
|
||||
),
|
||||
and(
|
||||
eq(chatActions.status, "failed"),
|
||||
sql`coalesce(${chatActions.result}->>'retryable', 'false') = 'true'`,
|
||||
sql`(${chatActions.result}->>'retryAt' is null or (${chatActions.result}->>'retryAt')::timestamptz <= ${now.toISOString()}::timestamptz)`,
|
||||
),
|
||||
),
|
||||
),
|
||||
)
|
||||
.where(and(
|
||||
onlyActionId ? eq(chatActions.id, onlyActionId) : undefined,
|
||||
dueChatAction("github_webhook_ingress", now.getTime()),
|
||||
))
|
||||
.orderBy(asc(chatActions.createdAt))
|
||||
.limit(limit);
|
||||
for (const action of actions) await processGitHubWebhookIngress(action.id);
|
||||
@@ -28092,8 +28081,6 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
|
||||
async function processPendingChatMaintenance(limit = 25) {
|
||||
return Promise.allSettled([
|
||||
processPendingGitHubWebhookIngress(limit),
|
||||
processPendingProviderEffects(limit),
|
||||
processPendingTelegramMaintenance(limit),
|
||||
processFailedChatRunRetries(limit),
|
||||
]);
|
||||
@@ -28106,7 +28093,8 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
// so a crashed processing claim is quarantined before the delivery worker
|
||||
// could otherwise replay it.
|
||||
const actionRecovery = onlyDeliveryId || !maintenance ? null : Promise.allSettled([
|
||||
processPendingChatMaintenance(limit), processPendingSlackBoardMessages(limit),
|
||||
processPendingChatMaintenance(limit), processPendingGitHubWebhookIngress(limit),
|
||||
processPendingProviderEffects(limit), processPendingSlackBoardMessages(limit),
|
||||
processPendingSlackTaskStarts(limit), processPendingReceiptReactions(limit),
|
||||
processPendingSlackSessionStops(limit), slackRegistration.processPendingVerificationMessages(limit),
|
||||
]);
|
||||
@@ -30095,6 +30083,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
);
|
||||
}
|
||||
const now = new Date();
|
||||
await notifyChatActionWork(tx, "provider_effect");
|
||||
const [claimed] = await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
@@ -30206,6 +30195,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
const now = new Date();
|
||||
const nextStatus =
|
||||
resolution === "mark_delivered" ? "processed" : "cancelled";
|
||||
await notifyChatActionWork(tx, "provider_effect");
|
||||
const [resolved] = await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
@@ -36931,43 +36921,36 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
) {
|
||||
const now = new Date();
|
||||
const actions = await db
|
||||
.select()
|
||||
.select({ action: chatActions })
|
||||
.from(chatActions)
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.kind, "slack_session_sync"),
|
||||
...(onlyActionId ? [eq(chatActions.id, onlyActionId)] : []),
|
||||
or(
|
||||
and(
|
||||
eq(chatActions.status, "received"),
|
||||
sql`(${chatActions.result}->>'retryAt' is null or (${chatActions.result}->>'retryAt')::timestamptz <= ${now.toISOString()}::timestamptz)`,
|
||||
),
|
||||
and(
|
||||
eq(chatActions.status, "processing"),
|
||||
lte(chatActions.updatedAt, new Date(now.getTime() - 60_000)),
|
||||
),
|
||||
),
|
||||
),
|
||||
)
|
||||
.leftJoin(chatEndpoints, and(eq(chatEndpoints.id, chatActions.endpointId), eq(chatEndpoints.companyId, chatActions.companyId)))
|
||||
.where(and(
|
||||
onlyActionId ? eq(chatActions.id, onlyActionId) : undefined,
|
||||
dueChatAction("slack_session_sync", now.getTime()),
|
||||
))
|
||||
.orderBy(asc(chatActions.updatedAt))
|
||||
.limit(limit);
|
||||
.limit(limit)
|
||||
.then(rows => rows.map(row => row.action));
|
||||
await options.slackSessionSyncSelectionBarrier?.();
|
||||
for (const action of actions) {
|
||||
const payload = slackSessionSyncPayload(action.payload);
|
||||
if (!payload || !action.conversationId) {
|
||||
await db
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "failed",
|
||||
result: { code: "slack_session_payload_invalid" },
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, action.id),
|
||||
eq(chatActions.updatedAt, action.updatedAt),
|
||||
),
|
||||
);
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyChatActionWork(tx, "slack_session_sync");
|
||||
return tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "failed",
|
||||
result: { code: "slack_session_payload_invalid" },
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, action.id),
|
||||
eq(chatActions.updatedAt, action.updatedAt),
|
||||
),
|
||||
);
|
||||
});
|
||||
continue;
|
||||
}
|
||||
const revisionWhere = and(
|
||||
@@ -36980,26 +36963,19 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
eq(chatActions.updatedAt, action.updatedAt),
|
||||
);
|
||||
const record = await endpointRecord(action.endpointId);
|
||||
if (record && ["paused", "attention"].includes(record.endpoint.status)) {
|
||||
await db
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "received",
|
||||
result: { retryAt: new Date(Date.now() + 30_000).toISOString() },
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(selectedSnapshotWhere);
|
||||
continue;
|
||||
}
|
||||
if (record && ["paused", "attention"].includes(record.endpoint.status)) continue;
|
||||
if (!record || record.endpoint.provider !== "slack") {
|
||||
await db
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "cancelled",
|
||||
result: { code: "slack_session_endpoint_unavailable" },
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(selectedSnapshotWhere);
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyChatActionWork(tx, "slack_session_sync");
|
||||
return tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "cancelled",
|
||||
result: { code: "slack_session_endpoint_unavailable" },
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(selectedSnapshotWhere);
|
||||
});
|
||||
continue;
|
||||
}
|
||||
const attempts =
|
||||
@@ -37084,6 +37060,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
? endpoint.allowDirectMessages
|
||||
: nonDirectDestinationAllowed(endpoint, resource))
|
||||
) {
|
||||
await notifyChatActionWork(tx, "slack_session_sync");
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
@@ -37118,6 +37095,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!publication) {
|
||||
await notifyChatActionWork(tx, "slack_session_sync");
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
@@ -37150,6 +37128,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
if (!run || !["queued", "running"].includes(run.status))
|
||||
status = "active";
|
||||
}
|
||||
await notifyChatActionWork(tx, "slack_session_sync");
|
||||
const claimed = await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
@@ -37195,6 +37174,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
outcome === "updated" && claim.status === "processing"
|
||||
? 30 * 60_000
|
||||
: null;
|
||||
await notifyChatActionWork(tx, "slack_session_sync");
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
@@ -37232,35 +37212,38 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
disposition.kind === "retry"
|
||||
? disposition.retryAfterMs
|
||||
: Math.min(5 * 60_000, 1_000 * 2 ** Math.min(attempts, 8));
|
||||
await db
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: retryable ? "received" : "failed",
|
||||
result: {
|
||||
attempts,
|
||||
code: retryable
|
||||
? "slack_session_sync_retry"
|
||||
: "slack_session_sync_rejected",
|
||||
...(retryable
|
||||
? { retryAt: new Date(Date.now() + retryMs).toISOString() }
|
||||
: {}),
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
revisionWhere,
|
||||
ownsAttempt
|
||||
? and(
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`${chatActions.result}->>'ownerToken' = ${ownerToken}`,
|
||||
)
|
||||
: and(
|
||||
eq(chatActions.status, "received"),
|
||||
eq(chatActions.updatedAt, action.updatedAt),
|
||||
),
|
||||
),
|
||||
);
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyChatActionWork(tx, "slack_session_sync");
|
||||
return tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: retryable ? "received" : "failed",
|
||||
result: {
|
||||
attempts,
|
||||
code: retryable
|
||||
? "slack_session_sync_retry"
|
||||
: "slack_session_sync_rejected",
|
||||
...(retryable
|
||||
? { retryAt: new Date(Date.now() + retryMs).toISOString() }
|
||||
: {}),
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
revisionWhere,
|
||||
ownsAttempt
|
||||
? and(
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`${chatActions.result}->>'ownerToken' = ${ownerToken}`,
|
||||
)
|
||||
: and(
|
||||
eq(chatActions.status, "received"),
|
||||
eq(chatActions.updatedAt, action.updatedAt),
|
||||
),
|
||||
),
|
||||
);
|
||||
});
|
||||
logger.warn(
|
||||
{
|
||||
endpointId: action.endpointId,
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
import { sql } from "drizzle-orm";
|
||||
import { chatActions, type Db } from "@paperclipai/db";
|
||||
|
||||
import { notifyChatActionWork } from "./chat-work-notifications.js";
|
||||
|
||||
type DbOrTransaction = Db | Parameters<Parameters<Db["transaction"]>[0]>[0];
|
||||
|
||||
export interface SlackSessionSyncPayload {
|
||||
@@ -47,6 +49,7 @@ export async function stageSlackSessionSync(
|
||||
runtimeGeneration: input.runtimeGeneration,
|
||||
credentialFingerprint: input.credentialFingerprint,
|
||||
};
|
||||
await notifyChatActionWork(database, "slack_session_sync");
|
||||
await database
|
||||
.insert(chatActions)
|
||||
.values({
|
||||
|
||||
@@ -11,6 +11,9 @@ export async function notifyChatPublicationWork(tx: object): Promise<void> {
|
||||
}
|
||||
|
||||
export const CHAT_ACTION_QUEUES = {
|
||||
provider_effect: DELIVERY_QUEUES.chatProviderEffects,
|
||||
github_webhook_ingress: DELIVERY_QUEUES.chatGitHubIngress,
|
||||
slack_session_sync: DELIVERY_QUEUES.chatSessionSyncs,
|
||||
slack_board_message: DELIVERY_QUEUES.chatBoardMessages,
|
||||
slash_task_start: DELIVERY_QUEUES.chatTaskStarts,
|
||||
receipt_reaction: DELIVERY_QUEUES.chatReceiptReactions,
|
||||
|
||||
@@ -6,6 +6,9 @@ export const DELIVERY_QUEUES = {
|
||||
chatTaskStarts: "chat-task-starts",
|
||||
chatReceiptReactions: "chat-receipt-reactions",
|
||||
chatSessionStops: "chat-session-stops",
|
||||
chatProviderEffects: "chat-provider-effects",
|
||||
chatGitHubIngress: "chat-github-ingress",
|
||||
chatSessionSyncs: "chat-session-syncs",
|
||||
chatVerification: "chat-verification",
|
||||
chatInbound: "chat-inbound",
|
||||
chatPublications: "chat-publications",
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { ComputerStopPendingError } from "../modules/computers/index.js";
|
||||
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";
|
||||
@@ -56,7 +57,6 @@ import {
|
||||
} from "./heartbeat/recovery.js";
|
||||
import {
|
||||
BOUNDED_TRANSIENT_HEARTBEAT_RETRY_MAX_ATTEMPTS,
|
||||
WorkspaceBusyDeferral,
|
||||
isWorkspaceBusyDeferral,
|
||||
createHeartbeatRetries,
|
||||
} from "./heartbeat/retries.js";
|
||||
@@ -139,8 +139,6 @@ export {
|
||||
buildPaperclipWakePayload,
|
||||
} from "./heartbeat/run-preparation.js";
|
||||
import {
|
||||
resolveNativeRecoveryExecutionWorkspaceBinding,
|
||||
resolveExecutionWorkspaceReuseRequestForIssue,
|
||||
buildExecutionWorkspaceConfigSnapshot,
|
||||
stripWorkspaceRuntimeFromExecutionRunConfig,
|
||||
buildEffectiveRunSessionConfigMetadata,
|
||||
@@ -422,12 +420,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 {
|
||||
@@ -506,18 +498,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,
|
||||
@@ -572,7 +559,6 @@ import {
|
||||
} from "@paperclipai/adapter-utils/server-utils";
|
||||
|
||||
import { environmentService } from "./environments.js";
|
||||
import { parseExecutionPolicyBootstrapEnv } from "./execution-policy-bootstrap.js";
|
||||
|
||||
import {
|
||||
environmentRuntimeService,
|
||||
@@ -3037,306 +3023,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,161 @@
|
||||
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, environmentLeases, 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.each(["agent", "instance"] as const)("selects the %s Boat environment without acquiring a lease", async source => {
|
||||
const { input } = await fixture();
|
||||
const [boat] = await db.insert(environments).values({
|
||||
name: `Selected Boat ${source}`, driver: "computer", status: "active", config: { provider: "boat" },
|
||||
}).returning();
|
||||
const [other] = await db.insert(environments).values({
|
||||
name: `Other environment ${source}`, driver: "ssh", status: "active", config: {},
|
||||
}).returning();
|
||||
input.agent.defaultEnvironmentId = source === "agent" ? boat.id : null;
|
||||
const settings = await input.services.instanceSettings.get();
|
||||
input.services.instanceSettings.get = vi.fn(async () => ({
|
||||
...settings, defaultEnvironmentId: source === "instance" ? boat.id : other.id,
|
||||
}));
|
||||
const result = await selectHeartbeatEnvironment(db, input);
|
||||
expect(result.selectedEnvironmentId).toBe(boat.id);
|
||||
expect(result.selectedEnvironmentForConfig).toMatchObject({ id: boat.id, driver: "computer" });
|
||||
expect(await db.select().from(environmentLeases).where(eq(environmentLeases.heartbeatRunId, input.run.id))).toEqual([]);
|
||||
});
|
||||
|
||||
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