From de9ab8e7e066b865908389633e5a2b5c3024651f Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sat, 10 Oct 2026 22:27:38 -0700 Subject: [PATCH 1/3] perf(runner): compile the TypeScript surface once per build (#15853) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Pull requests to this repository run a large CI matrix. The slowest check sets the time a contributor waits for a green run. > - After #15849 hid the verifier's downloads behind package staging, the `Build` lane is still the slowest check in five of the eight fully green PR runs created after that merge (535s to 581s; the top server shard is 529s to 560s in the other three). `Build` is about 100s of setup, 160s to 176s of `pnpm build`, and 265s to 312s of the public install verifier. > - Inside `pnpm build`, every package after `@paperclipai/shared` waits on the Runner: its `build` script takes 76s to 80s, and for 65s of that it is the only job running. The chain runs `build:typescript` three times: once directly, then again inside `check:semantic-contracts` and `check:runner-workflow-traceability`, which rebuild first so they work standalone. The server's `prepare:runner-vendor` then runs the whole Runner build a second time, so the job pays for four redundant TypeScript passes. > - Each pass is two `tsc` projects, the eval-kernel rebuild, and the schema checks, 7s to 8s on the CI runner. The dist they rebuild is the dist the first pass just wrote; nothing between the passes changes it. > - This pull request calls the two check scripts directly after the chain's own `build:typescript`, the way `generate-replay-goldens.mjs --check` already does in the same line. The standalone `check:*` scripts keep their rebuild. > - The benefit is about 30s off the `Build` job's critical path, about 15s off the Typecheck job and every Docker or release build that runs the server build, with no change to what is checked. ## Linked Issues or Issue Description **Describe the enhancement** The `Build` PR check is the slowest check in most green PR runs. Its `pnpm build` step compiles the Runner's TypeScript surface four times: three passes inside one Runner `build`, and a second full Runner `build` from the server's `prepare:runner-vendor`. Only the first pass produces a dist that the later steps read; the other three rewrite the same output. **Why is this needed** The slowest check decides the wall-clock time of every PR run. In the eight fully green PR runs created after 2026-10-11 04:10Z, `Build` topped five of them at 535s to 581s. The Runner build sits alone on the `pnpm build` critical path for about 65s, and the server build repeats it. From the job logs of four of those runs, the second and third `build:typescript` passes cost 7s to 8s each, in both Runner builds, so about 30s per job. **Proposed solution** In the Runner's `build` script, replace `pnpm run check:semantic-contracts && pnpm run check:runner-workflow-traceability` with `node scripts/generate-semantic-contracts.mjs --check && node scripts/check-runner-workflow-traceability.mjs`. Both scripts read `dist/`, which `build:typescript` wrote earlier in the same chain, and the steps between them are `--check` runs that write nothing to `dist/`. Add a guard test so the chain compiles TypeScript exactly once and both checks still run after it. **Alternatives considered** - Skip the server's `prepare:runner-vendor` rebuild when `pnpm -r build` already built the Runner. That rebuild is what makes `pnpm --filter @paperclipai/server build` correct standalone (the Dockerfile uses it), and the server's build-script test pins the command. A freshness check would be a larger semantic change; this PR shrinks the rebuild instead of removing it, and the next cycle can measure what remains. - Make `tsc` incremental across the passes. It would still re-run the schema checks and the eval-kernel build each time, and would add build-info files to the package. - Change `check:semantic-contracts` and `check:runner-workflow-traceability` themselves to stop rebuilding. They are documented standalone commands (`pnpm --dir packages/paperclip-runner check:semantic-contracts` in the capability docs) and must work from a stale dist; unchanged here. ## What Changed - `packages/paperclip-runner/package.json`: the `build` script runs `node scripts/generate-semantic-contracts.mjs --check` and `node scripts/check-runner-workflow-traceability.mjs` directly, after its own `build:typescript`, instead of the `check:semantic-contracts` and `check:runner-workflow-traceability` wrappers that rebuild first. Every other step, its order, and the standalone `check:*` scripts are unchanged. - `server/src/__tests__/server-package-build-script.test.ts`: a new test in the suite that already pins the server's Runner vendoring reads the Runner's `build` script, resolves every `pnpm run` step, and asserts that exactly one step compiles TypeScript (directly or through a wrapper), that both check scripts are in the chain, and that both come after `build:typescript`. It fails against the previous script with `expected [ 'build:typescript', …(2) ] to deeply equal [ 'build:typescript' ]`. ## Verification - `pnpm --filter @paperclipai/server exec vitest run src/__tests__/server-package-build-script.test.ts`: 9 pass (1 new). With the `package.json` change stashed, the new test fails as described above and the other 8 pass. - In `packages/paperclip-runner`, the new chain without the cargo step (`check:protocol-manifest`, `build:typescript`, `check:capability-contract`, `check:capability-inventory`, `check:protocol-coverage`, `generate-semantic-contracts.mjs --check`, `check-runner-workflow-traceability.mjs`, `generate-replay-goldens.mjs --check`, `generate-semantic-action-catalog.mjs --check`) exits 0: `Runner stress traceability passed: 44 findings, 12 workflows, 1 explicit exclusion.` and `Replay golden snapshots and parity summaries are current.` Cargo is not installed on the host that produced this PR, so `build:binary` was not run locally; it is unchanged and runs in this PR's `Build` job. - `pnpm run build:typescript` alone takes 5.8s to 5.9s on this host, so each removed pass is in the range the CI logs show. - This PR's own `Build` job runs `pnpm build` from the branch checkout, so its log shows the new chain in both Runner builds. Expected on that job: `pnpm build` about 130s to 146s (was 160s to 176s on the four measured runs), `Build` about 505s to 551s (was 535s to 581s). **Measured on the four post-#15849 `master` runs that motivated this PR** (`Build` job logs): | Run | `pnpm build` | Runner build in `pnpm -r build` | Extra `build:typescript` passes in it | Runner rebuild in server build | Extra passes in it | | --- | --- | --- | --- | --- | --- | | 38112764439 | 169s | 76s | 04:51:52 and 04:52:00 (about 14s) | 04:53:12 to 04:53:41 | 04:53:26 and 04:53:34 (about 15s) | | 38111191543 | 176s | 79s | 04:22:40 and 04:22:48 (about 14s) | 04:24:04 to 04:24:34 | 04:24:18 and 04:24:26 (about 15s) | | 38110914319 | 160s | 75s | 04:19:07 and 04:19:14 (about 14s) | 04:20:25 to 04:20:53 | 04:20:39 and 04:20:46 (about 14s) | | 38110897745 | 172s | 80s | 04:15:16 and 04:15:25 (about 15s) | 04:16:41 to 04:17:11 | 04:16:55 and 04:17:03 (about 15s) | ## Risks - Low risk. The same two checks run against the same dist, in the same position in the chain. The only difference is that the dist is not rewritten twice between them. - If a future step between `build:typescript` and the two checks starts writing to `dist/`, the checks would read that output instead of a fresh rebuild. The steps there today (`check:capability-contract`, `check:capability-inventory`, `check:protocol-coverage`) are `--check` runs that only compare generated spec files. - The standalone `check:semantic-contracts` and `check:runner-workflow-traceability` scripts still rebuild first, so documented usage is unchanged. - The guard test lives in the server's suite because that suite already pins the Runner vendoring contract and runs in the server shards on every PR. ## Model Used - Claude (Anthropic), model ID `claude-fable-5-1`, running as a Paperclip agent through the Claude local adapter with tool use (shell, file edits, GitHub CLI). The agent collected the CI data, made the change, ran the tests, and wrote this description. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [ ] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: devinfoley <139239+devinfoley@users.noreply.github.com> Co-authored-by: Claude Fable 5.1 --- packages/paperclip-runner/package.json | 2 +- .../server-package-build-script.test.ts | 30 +++++++++++++++++++ 2 files changed, 31 insertions(+), 1 deletion(-) diff --git a/packages/paperclip-runner/package.json b/packages/paperclip-runner/package.json index 8c1249044ca..f962fe6c20b 100644 --- a/packages/paperclip-runner/package.json +++ b/packages/paperclip-runner/package.json @@ -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", diff --git a/server/src/__tests__/server-package-build-script.test.ts b/server/src/__tests__/server-package-build-script.test.ts index 64cbbf17ab2..0c972569472 100644 --- a/server/src/__tests__/server-package-build-script.test.ts +++ b/server/src/__tests__/server-package-build-script.test.ts @@ -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 }; + 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( From 30e338fba6262385c0c8bc4b01732cb6d94c5386 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sun, 11 Oct 2026 06:04:31 -0500 Subject: [PATCH 2/3] refactor(heartbeat): extract environment selection (#15825) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Heartbeat execution selects an environment before it prepares a workspace and starts an agent. > - Workspace preparation now has its own module, but its selection policy still occupies the executor. > - Environment defaults, recovery bindings, chat isolation, and workspace contention must retain their order. > - This pull request moves that selection phase behind one typed function. > - The executor keeps dispatch gates and lease cleanup. Tests cover the public path and the new boundary. ## Linked Issues or Issue Description Refs #15807. This continues the heartbeat extraction series after workspace preparation. **Current behavior** The 5,599-line heartbeat service contains a 301-line environment selection phase inside run execution. **Proposed behavior** Move that phase to `heartbeat/environment-selection.ts`. Keep the same policy, statement order, errors, and public exports. **Reason and benefit** Readers can find selection policy in one module and follow the executor's phases without reading each policy branch. ## What Changed - Add `selectHeartbeatEnvironment` with typed run, task, policy, configuration, and service inputs. Return the values needed by configuration and workspace preparation. - Preserve native recovery bindings, environment precedence, managed sandbox enforcement, Kubernetes provisioning, native chat isolation, and shared workspace contention. - Reduce `heartbeat.ts` from 5,599 lines / 234,599 bytes to 5,314 lines / 220,784 bytes. The new module is 402 lines / 18,215 bytes. - Add six public-path characterization cases before moving production code and six boundary cases. Repair plugin-test teardown to drain active execution and call the database cleanup helper. - Assert the selected local environment in the browser fixture's real worker and document the new boundary. ## Verification - Before the move: 71 tests passed, including all six new public-path cases. Three plugin tests skipped when the host could not start their database probe. - After the move: all 112 focused tests passed across eight files, with no skips. These cover environment policy, local and plugin execution, contention, recovery bindings, workspace preparation, chat isolation, and low-trust directories. - Passed `pnpm -r typecheck` and `pnpm build`. - A normalized TypeScript AST comparison confirms the 301 moved lines and remaining executor statements are unchanged, apart from the new call and imports. - Browser acceptance: the task UI recovered automatically and the worker verified its selected local environment, preserved uncommitted files, and completed the task. The persistent-failure case passed in isolation: three total runs, then a blocked task that remained blocked after reload. - The first persistent-failure browser attempt missed the retry banner within five seconds, despite a persisted scheduled retry. Its unchanged isolated rerun passed. The trace is retained; the intermittent UI failure is not yet attributed to a root cause. No assertion or timeout was relaxed. - Greptile reviewed `20953dd164decb565bc8d85081aad31a23fd33e5`: 5/5, no actionable findings. - Hosted CI passed on `20953dd164decb565bc8d85081aad31a23fd33e5`: 52 successful checks and two intentionally skipped Storybook checks, with no pending or failing checks. This includes all general and serialized test groups, Runner verification, eight browser shards, typecheck, and build. - The duplicate local `pnpm test:run` was interrupted after hosted CI completed. It is not reported as a local full-suite pass. Local completed checks are the 112 focused tests, typecheck, build, and browser cases described above. ## Risks - This selection phase enforces execution restrictions. A wrong input or import could select the wrong environment or change retry behavior. Public-path tests cover refusal, redirect, contention, and dispatch; boundary tests cover recovery and isolation. - The helper can provision a managed environment row and add concurrency notes to the supplied task context. It acquires no execution lease. The executor retains its catch/finally boundaries and resource ownership. - Kubernetes selection is tested with real database state and bootstrap settings. These checks do not claim execution against a live Kubernetes cluster or hosted sandbox provider. - No schema, migration, API contract, or intended behavior change. ## Model Used OpenAI Codex (GPT-6), with repository inspection, code editing, compiler checks, database integration tests, and browser automation. The exact serving model ID and context-window size are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Co-authored-by: Paperclip --- doc/DEVELOPING.md | 10 +- .../heartbeat-plugin-environment.test.ts | 7 +- .../heartbeat-workspace-busy.test.ts | 86 ++++ server/src/services/heartbeat.ts | 345 ++------------- .../heartbeat/environment-selection.test.ts | 142 +++++++ .../heartbeat/environment-selection.ts | 402 ++++++++++++++++++ tests/e2e/workspace-bootstrap/worker.mjs | 1 + 7 files changed, 676 insertions(+), 317 deletions(-) create mode 100644 server/src/services/heartbeat/environment-selection.test.ts create mode 100644 server/src/services/heartbeat/environment-selection.ts diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index fd665cdb6db..3425955fed7 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -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`. diff --git a/server/src/__tests__/heartbeat-plugin-environment.test.ts b/server/src/__tests__/heartbeat-plugin-environment.test.ts index 2cd19d301da..667bd68ac21 100644 --- a/server/src/__tests__/heartbeat-plugin-environment.test.ts +++ b/server/src/__tests__/heartbeat-plugin-environment.test.ts @@ -59,14 +59,16 @@ describeEmbeddedPostgres("heartbeat plugin environments", () => { let stopDb: (() => Promise) | null = null; let db!: ReturnType; const tempRoots: string[] = []; + const heartbeats: ReturnType[] = []; 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", diff --git a/server/src/__tests__/heartbeat-workspace-busy.test.ts b/server/src/__tests__/heartbeat-workspace-busy.test.ts index af0d1f63c94..cb45daa7445 100644 --- a/server/src/__tests__/heartbeat-workspace-busy.test.ts +++ b/server/src/__tests__/heartbeat-workspace-busy.test.ts @@ -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" }, diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index b0ddbe610ce..4c2cd3da028 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { selectHeartbeatEnvironment } from "./heartbeat/environment-selection.js"; import { prepareHeartbeatWorkspace } from "./heartbeat/workspace-preparation.js"; import { executeHeartbeatRuntime, NativeSessionResumeScheduledError, NativeWorkspaceFinalizeScheduledError } from "./heartbeat/runtime-execution.js"; import { selectHeartbeatRuntime } from "./heartbeat/runtime-selection.js"; @@ -55,7 +56,6 @@ import { } from "./heartbeat/recovery.js"; import { BOUNDED_TRANSIENT_HEARTBEAT_RETRY_MAX_ATTEMPTS, - WorkspaceBusyDeferral, isWorkspaceBusyDeferral, createHeartbeatRetries, } from "./heartbeat/retries.js"; @@ -138,8 +138,6 @@ export { buildPaperclipWakePayload, } from "./heartbeat/run-preparation.js"; import { - resolveNativeRecoveryExecutionWorkspaceBinding, - resolveExecutionWorkspaceReuseRequestForIssue, buildExecutionWorkspaceConfigSnapshot, stripWorkspaceRuntimeFromExecutionRunConfig, buildEffectiveRunSessionConfigMetadata, @@ -421,12 +419,6 @@ import { isNativeRunnerOwnershipHeld, NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE, } from "./native-runtime/native-runner-ownership.js"; -import { - findNativeChatWorkspaceScope, - nativeChatWorkspaceCwd, - nativeChatWorkspaceMatches, -} from "./native-runtime/native-chat-workspace.js"; -import { shouldUseIsolatedTaskDirectory } from "./isolated-task-directory.js"; import { trackAgentFirstHeartbeat } from "@paperclipai/shared/telemetry"; import { getTelemetryClient } from "../telemetry.js"; import { @@ -505,18 +497,13 @@ import { gateProjectExecutionWorkspacePolicy, parseIssueExecutionWorkspaceSettings, parseProjectExecutionWorkspacePolicy, - resolveExecutionWorkspaceEnvironmentId, resolveExecutionWorkspaceMode, - resolveSharedWorkspaceConcurrency, selectEnvironmentExecutionWorkspaceSettings, } from "./execution-workspace-policy.js"; import { instanceSettingsService, resolveWorktreeRunExecutionActivation, } from "./instance-settings.js"; -import { - isExecutionForcedToKubernetes, -} from "./execution-allowlist.js"; import { buildConfigurationIncompleteRecoveryNoticeSeed, @@ -571,7 +558,6 @@ import { } from "@paperclipai/adapter-utils/server-utils"; import { environmentService } from "./environments.js"; -import { parseExecutionPolicyBootstrapEnv } from "./execution-policy-bootstrap.js"; import { environmentRuntimeService, @@ -3034,306 +3020,35 @@ export function heartbeatService( } else { delete context.paperclipTurnContext; } - // A native run's execution input is immutable once persisted. Recovery must therefore - // restore the workspace bound to that input rather than consulting the issue's current - // workspace pointer: a newer run may already have moved or cleared the issue binding while - // this older provider session is still recoverable. - const persistedRunnerProfile = parseObject(run.runnerProfileJson); - const persistedNativeExecutionInput = - run.runtimeMode === "native" && - persistedRunnerProfile.nativeExecutionInput !== undefined - ? parseNativeExecutionInput( - persistedRunnerProfile.nativeExecutionInput, - ) - : null; - const isDotRun = persistedNativeExecutionInput?.provider.kind === "openai_dot" - || (!persistedNativeExecutionInput && agent.adapterType === "paperclip_runner" && parseObject(agent.adapterConfig).provider === "openai_dot"); - const persistedNativeExecutionWorkspaceId = - persistedNativeExecutionInput?.binding.executionWorkspaceId ?? null; - const requestedExecutionWorkspaceId = - persistedNativeExecutionWorkspaceId ?? - readNonEmptyString(issueRef?.executionWorkspaceId); - const existingExecutionWorkspace = requestedExecutionWorkspaceId - ? await executionWorkspacesSvc.getById(requestedExecutionWorkspaceId) - : null; - const nativeRecoveryExecutionWorkspaceId = - resolveNativeRecoveryExecutionWorkspaceBinding({ - bindingId: persistedNativeExecutionWorkspaceId, - persistedWorkspaceFound: existingExecutionWorkspace !== null, - }); - const workspaceReuseRequest = - resolveExecutionWorkspaceReuseRequestForIssue({ - issueExecutionWorkspaceId: requestedExecutionWorkspaceId, - issueExecutionWorkspacePreference: nativeRecoveryExecutionWorkspaceId - ? "reuse_existing" - : (issueRef?.executionWorkspacePreference ?? null), - existingExecutionWorkspaceStatus: - existingExecutionWorkspace?.status ?? null, - }); - const requestedShouldReuseExisting = - workspaceReuseRequest.requestedShouldReuseExisting; - const reusableExistingExecutionWorkspace = - workspaceReuseRequest.existingExecutionWorkspaceAvailable - ? existingExecutionWorkspace - : null; - const requestedReusableExecutionWorkspaceConfig = - reusableExistingExecutionWorkspace?.config ?? null; - const localEnvironment = await environmentsSvc.ensureLocalEnvironment( - agent.companyId, - ); - const resolvedInstanceSettings = await instanceSettings.get(); - // Managed-sandbox-only policy: a run that would land on the local - // environment is redirected onto the platform-managed sandbox row, and - // with no active managed row the resolution fails closed - // (ManagedSandboxUnavailableError) — never local. Mirrors the forced - // kubernetes execution mode below, which takes precedence when both - // regimes are active. - const managedSandboxOnly = - (await instanceSettings.getExperimental()).enableManagedSandboxOnly === - true; - const managedSandboxEnvironment = managedSandboxOnly - ? await environmentsSvc.findManagedSandboxEnvironment(agent.companyId) - : null; - const environmentResolution = resolveExecutionWorkspaceEnvironmentId({ - lowTrustIssueEnvironmentId: trustPreset.kind === "low_trust_review" - ? parseIssueExecutionWorkspaceSettings(issueContext?.executionWorkspaceSettings, {includeEnvironmentId: true})?.environmentId : null, - agentDefaultEnvironmentId: agent.defaultEnvironmentId, - instanceDefaultEnvironmentId: - resolvedInstanceSettings.defaultEnvironmentId ?? null, - localDefaultEnvironmentId: localEnvironment.id, - managedSandboxOnly, - managedSandboxEnvironmentId: managedSandboxEnvironment?.id ?? null, - }); - const effectiveExecutionWorkspaceMode: ReturnType< - typeof resolveExecutionWorkspaceMode - > = requestedExecutionWorkspaceMode; - const executionPolicy = { - executionMode: resolvedInstanceSettings.general.executionMode, - // Backstop behind the resolver's local→managed redirect: the run-time - // allowlist below fails any run that still resolved to a `local` - // environment under managed-sandbox-only, so no selection path or - // tenant-set env var can land untrusted execution on the tenant - // container. - managedSandboxOnly, - }; - const executionForcedToKubernetes = - isExecutionForcedToKubernetes(executionPolicy); - let selectedEnvironmentId = environmentResolution.environmentId; - if (executionForcedToKubernetes) { - let kubernetesEnvironment = - await environmentsSvc.findKubernetesEnvironment(agent.companyId); - if (!kubernetesEnvironment) { - // Lazy recovery for companies created after the startup bootstrap ran - // (the boot hook only provisions environments for companies that exist - // at boot). Re-derive the managed-env config from the bootstrap env. - // If the process env no longer forces Kubernetes (rollback / config - // drift relative to the persisted executionMode setting), skip the - // provisioning gracefully: the guard below still refuses local - // fallback with the explicit error, instead of crashing here on - // undefined config. - let bootstrap: ReturnType = - 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, diff --git a/server/src/services/heartbeat/environment-selection.test.ts b/server/src/services/heartbeat/environment-selection.test.ts new file mode 100644 index 00000000000..c55395c3d7a --- /dev/null +++ b/server/src/services/heartbeat/environment-selection.test.ts @@ -0,0 +1,142 @@ +import { randomUUID } from "node:crypto"; +import { mkdtemp, mkdir, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { agents, companies, createDb, environments, heartbeatRuns, issues, projects, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, type Db } from "@paperclipai/db"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { resolvePaperclipInstanceRoot } from "../../home-paths.js"; +import { environmentService } from "../environments.js"; +import { executionWorkspaceService } from "../execution-workspaces.js"; +import { instanceSettingsService } from "../instance-settings.js"; +import { buildNativeExecutionInput } from "../native-runtime/native-execution-input.js"; +import { nativeRuntimeContextFixture } from "../native-runtime/runtime-context.test-fixture.js"; +import { selectHeartbeatEnvironment, type HeartbeatEnvironmentSelectionInput } from "./environment-selection.js"; + +const support = await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("heartbeat environment selection boundary", () => { + let database: Awaited>; + let db: Db; + let home: string; + beforeAll(async () => { + home = await mkdtemp(join(tmpdir(), "heartbeat-environment-selection-")); + vi.stubEnv("PAPERCLIP_HOME", home); + vi.stubEnv("PAPERCLIP_INSTANCE_ID", "environment-selection"); + await mkdir(resolvePaperclipInstanceRoot(), { recursive: true }); + database = await startEmbeddedPostgresTestDatabase("heartbeat-environment-selection-"); + db = createDb(database.connectionString); + await db.execute(sql`set client_min_messages = warning`); + }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); + afterAll(async () => { + await db?.$client.end(); + await database?.cleanup(); + vi.unstubAllEnvs(); + if (home) await rm(home, { recursive: true, force: true }); + }); + beforeEach(async () => { + await db.execute(sql`truncate table companies restart identity cascade`); + }); + + async function fixture() { + const [company] = await db.insert(companies).values({ name: "Environment selection", issuePrefix: "ENV" }).returning(); + const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Selecting agent", status: "running", adapterType: "process" }).returning(); + const [project] = await db.insert(projects).values({ companyId: company.id, name: "Project" }).returning(); + const [issue] = await db.insert(issues).values({ companyId: company.id, projectId: project.id, title: "Select this environment", assigneeAgentId: agent.id }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id, issueId: issue.id, scopeKind: "issue", status: "running", runtimeMode: "legacy" }).returning(); + const input: HeartbeatEnvironmentSelectionInput = { + run, agent, + task: { issueRef: issue, issueContext: issue, issueId: issue.id, executionProjectId: project.id, projectContext: { hasWorkspace: false }, context: { issueId: issue.id } }, + policy: { + trustPreset: { kind: "standard", preset: "standard", boundary: null, sourcePresets: {} }, + requestedExecutionWorkspaceMode: "shared_workspace", isolatedWorkspacesEnabled: true, + projectExecutionWorkspacePolicy: null, issueExecutionWorkspaceSettings: null, + }, + config: { config: {}, issueAssigneeOverrides: null }, + services: { environmentsSvc: environmentService(db), executionWorkspacesSvc: executionWorkspaceService(db), instanceSettings: instanceSettingsService(db), findSharedWorkspaceHolder: vi.fn(async () => null) }, + }; + const workspace = async () => { + const created = await executionWorkspaceService(db).create({ companyId: company.id, projectId: project.id, sourceIssueId: issue.id, name: "Saved workspace", mode: "isolated_workspace", strategyType: "git_worktree", cwd: home, providerRef: home }); + if (!created) throw new Error("Workspace fixture was not created"); + return created; + }; + const recover = (workspaceId: string) => { + input.run.runtimeMode = "native"; + input.run.runnerProfileJson = { nativeExecutionInput: buildNativeExecutionInput({ + companyId: company.id, runId: run.id, agentId: agent.id, + issue: { id: issue.id, identifier: "ENV-1", title: issue.title, description: null, workMode: "standard" }, + taskPrompt: "Continue the same task", normalizedSessionId: randomUUID(), provider: "codex", + workspace: { id: workspaceId, cwd: home, repoUrl: null, repoRef: null, branchName: null }, + completionContract: { id: randomUUID(), sha256: `sha256:${"a".repeat(64)}`, schemaVersion: "paperclip.run-result.v1", contract: { revision: "1", objective: "Select the environment", criteria: [{ id: "objective", requirement: "Preserve the workspace" }] } }, + runtimeContext: nativeRuntimeContextFixture(), + }) }; + }; + return { input, issue, project, workspace, recover }; + } + + it("keeps the admitted workspace binding when the issue points at a newer workspace", async () => { + const { input, workspace, recover } = await fixture(); + const original = await workspace(); + const newer = await workspace(); + input.task.issueRef!.executionWorkspaceId = newer.id; + input.task.issueRef!.executionWorkspacePreference = null; + recover(original.id); + const result = await selectHeartbeatEnvironment(db, input); + expect(result.nativeRecoveryExecutionWorkspaceId).toBe(original.id); + expect(result.reusableExistingExecutionWorkspace?.id).toBe(original.id); + expect(result.requestedShouldReuseExisting).toBe(true); + expect(input.task.issueRef!.executionWorkspaceId).toBe(newer.id); + expect(result.persistedNativeExecutionInput?.binding.executionWorkspaceId).toBe(original.id); + }); + + it("does not treat a synthetic run-id recovery binding as a persisted workspace", async () => { + const { input, recover } = await fixture(); + recover(input.run.id); + const result = await selectHeartbeatEnvironment(db, input); + expect(result).toMatchObject({ existingExecutionWorkspace: null, nativeRecoveryExecutionWorkspaceId: null, requestedShouldReuseExisting: false, reusableExistingExecutionWorkspace: null }); + expect(result.persistedNativeExecutionInput?.binding.executionWorkspaceId).toBe(input.run.id); + }); + + it("does not make an archived issue workspace available for reuse", async () => { + const { input, workspace } = await fixture(); + const saved = await workspace(); + await executionWorkspaceService(db).update(saved.id, { status: "archived" }); + input.task.issueRef!.executionWorkspaceId = saved.id; + input.task.issueRef!.executionWorkspacePreference = "reuse_existing"; + const result = await selectHeartbeatEnvironment(db, input); + expect(result.requestedShouldReuseExisting).toBe(true); + expect(result.reusableExistingExecutionWorkspace).toBeNull(); + expect(result.requestedReusableExecutionWorkspaceConfig).toBeNull(); + }); + + it("uses the low-trust task sandbox even when workspace feature gates hide issue settings", async () => { + const { input, issue } = await fixture(); + const [sandbox] = await db.insert(environments).values({ name: "Task sandbox", driver: "sandbox", status: "active", config: { provider: "fake" } }).returning(); + input.task.issueContext!.executionWorkspaceSettings = { mode: "isolated_workspace", environmentId: sandbox.id }; + input.policy.trustPreset = { kind: "low_trust_review", preset: "low_trust_review", sourcePresets: {}, boundary: { mode: "low_trust_review", companyId: input.agent.companyId, rootIssueId: issue.id } }; + input.policy.requestedExecutionWorkspaceMode = "isolated_workspace"; + input.policy.isolatedWorkspacesEnabled = false; + const result = await selectHeartbeatEnvironment(db, input); + expect(result.selectedEnvironmentForConfig?.id).toBe(sandbox.id); + expect(result.useIsolatedTaskDirectory).toBe(true); + }); + + it("returns the task-owned local chat root before workspace realization", async () => { + const { input, issue } = await fixture(); + await db.update(issues).set({ originKind: "chat_channel", projectId: null }).where(eq(issues.id, issue.id)); + input.agent.adapterType = "paperclip_runner"; + input.task.executionProjectId = null; + input.task.projectContext = null; + const result = await selectHeartbeatEnvironment(db, input); + expect(result.nativeChatWorkspaceScope).toMatchObject({ companyId: input.agent.companyId, agentId: input.agent.id, issueId: issue.id, projectId: null }); + expect(result.nativeChatExpectedCwd).toBe(result.nativeChatWorkspaceScope?.taskRoot); + expect(result.nativeChatExpectedCwd).toContain(`chat-workspaces/${input.agent.companyId}/${input.agent.id}/${issue.id}`); + }); + + it("rejects a project chat without a task-owned isolated workspace", async () => { + const { input, issue } = await fixture(); + await db.update(issues).set({ originKind: "chat_channel" }).where(eq(issues.id, issue.id)); + input.agent.adapterType = "paperclip_runner"; + await expect(selectHeartbeatEnvironment(db, input)).rejects.toThrow("External chat requires a task-owned isolated workspace"); + }); +}); diff --git a/server/src/services/heartbeat/environment-selection.ts b/server/src/services/heartbeat/environment-selection.ts new file mode 100644 index 00000000000..78509cdac29 --- /dev/null +++ b/server/src/services/heartbeat/environment-selection.ts @@ -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; + agent: Pick; + task: { + issueRef: Pick | null; + issueContext: Pick | null; + issueId: string | null; + context: Record; + executionProjectId: string | null; + projectContext: { hasWorkspace: boolean } | null; + }; + policy: { + trustPreset: TrustPresetResolution; + requestedExecutionWorkspaceMode: ParsedExecutionWorkspaceMode; + isolatedWorkspacesEnabled: boolean; + projectExecutionWorkspacePolicy: ProjectExecutionWorkspacePolicy | null; + issueExecutionWorkspaceSettings: IssueExecutionWorkspaceSettings | null; + }; + config: { + config: Record; + issueAssigneeOverrides: { adapterConfig: Record | null } | null; + }; + services: { + executionWorkspacesSvc: Pick, "getById">; + environmentsSvc: Pick, + "ensureLocalEnvironment" | "findManagedSandboxEnvironment" | "findKubernetesEnvironment" | "ensureKubernetesEnvironment" | "getById">; + instanceSettings: Pick, "get" | "getExperimental">; + findSharedWorkspaceHolder: ReturnType["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 = + 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, + }; +} diff --git a/tests/e2e/workspace-bootstrap/worker.mjs b/tests/e2e/workspace-bootstrap/worker.mjs index e9af0d7e41e..07222bfb2bc 100644 --- a/tests/e2e/workspace-bootstrap/worker.mjs +++ b/tests/e2e/workspace-bootstrap/worker.mjs @@ -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-"))); From e733bddc45e00efb1819fffa8677f7b0d07802c5 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sun, 11 Oct 2026 06:08:05 -0500 Subject: [PATCH 3/3] perf(chat): stop polling provider action and session sync queues (#15822) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Chat stores pending provider actions in durable database rows. > - Three action queues still scan every second when empty. > - These scans keep idle databases active. > - This pull request wakes those queues after a commit or at their next deadline. > - Empty queues stop querying while existing recovery rules remain in place. ## Linked Issues or Issue Description Refs #15808, #15802, #15625. Related sleep work: #15750, #15757. **What existing behavior does this improve?** Provider messages and notices, accepted GitHub webhook receipts, and Slack session status synchronization. **Current behavior** Chat maintenance scans these three queues every second. It also rechecks paused Slack sync rows every 30 seconds. **Proposed behavior** Reuse the transaction-aware action workers and existing chat_actions rows. Wake after commits. Restore unfinished work on startup. Use stored retry and stale-claim deadlines. Stop scanning when empty. Park paused Slack sync rows until an endpoint change. ## What Changed - Add three topics to the existing action worker registration. No tables, migrations, triggers, or external services. - Signal producers, claims, retry updates, terminal settlements, and operator resolution inside their owning transaction. Provider I/O remains outside transactions. - Share pending-state and deadline predicates between dispatch and scheduling. Exclude terminal history and uncertain provider sends. - Preserve GitHub receipt attempt fences and provider-send quarantine. Preserve Slack's 30-minute refresh while a session is processing. - Remove these workers from periodic app maintenance. Update the sleep inventory and add regression coverage. ## Verification - Passed: 13 focused tests across scheduler, app wiring, and real PostgreSQL integration. Covers actual commit wakes with deferred fast paths disabled, GitHub retries and startup recovery, stale-send quarantine, Slack publication-to-sync delivery, paused deadlines, and idle queues. - Passed: pnpm typecheck. - Passed: full chat integration and related coordinator suites, **1,274 tests across five files**. - Passed: pnpm build. - Greptile: **5/5**, no actionable findings, completed on `dc284cd275e5f6df624d60b27954421f2bcdbee2`. - Hosted CI: **54 checks passed**, two intentionally skipped Storybook checks, no failures or pending checks on `dc284cd275e5f6df624d60b27954421f2bcdbee2`. Provider responses are simulated. PostgreSQL, transactions, service dispatch, and scheduling are real. No live Slack or GitHub App was used. ## Risks - Writers must emit the existing process-local transaction signal. Independent database clients and direct SQL still need explicit integration or startup recovery. - Accepted GitHub receipts are covered. Finding callbacks GitHub never delivered still uses the separate provider recovery lane. - Other chat maintenance and persistent transports still run. This does not establish whole-instance sleep. - Existing claim, revision, credential, and authorization checks remain authoritative during concurrent foreground and scheduled dispatch. ## Model Used OpenAI Codex (GPT-6), with repository inspection, code execution, and automated tests. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Co-authored-by: Paperclip --- ...-instance-sleep-and-scheduler-inventory.md | 16 +- .../app-private-hostname-gate.test.ts | 6 + server/src/__tests__/chat-action-work.test.ts | 7 +- .../chat-channels.integration.test.ts | 171 ++++++++- server/src/app.ts | 6 +- server/src/services/chat-action-work.ts | 18 +- server/src/services/chat-channels.ts | 341 +++++++++--------- .../src/services/chat-slack-session-outbox.ts | 3 + .../src/services/chat-work-notifications.ts | 3 + .../services/delivery-work-notifications.ts | 3 + 10 files changed, 382 insertions(+), 192 deletions(-) diff --git a/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md b/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md index d818caa41c6..e40716791f3 100644 --- a/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md +++ b/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md @@ -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. diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index 61893417dfd..8f087241913 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -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])( diff --git a/server/src/__tests__/chat-action-work.test.ts b/server/src/__tests__/chat-action-work.test.ts index af541a60c07..b96541e79ee 100644 --- a/server/src/__tests__/chat-action-work.test.ts +++ b/server/src/__tests__/chat-action-work.test.ts @@ -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 => null), nextVerificationMessageAt: vi.fn(async (): Promise => 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); diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 118b47a8a81..a7d9fad74fa 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -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"); } diff --git a/server/src/app.ts b/server/src/app.ts index 003d587c551..238a5bad698 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -326,7 +326,7 @@ export function createChatReconciliationCoordinator(input: { projectRunMilestones: () => Promise; flushPublications: () => Promise; processPendingSlackFileUploadReceipts?: () => Promise; - processPendingSlackSessionSyncs: () => Promise; + processPendingSlackSessionSyncs?: () => Promise; onError: (lane: ChatReconciliationLane, error: unknown) => void; }) { let stopped = false; @@ -373,7 +373,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(); @@ -1271,8 +1271,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}`); }, diff --git a/server/src/services/chat-action-work.ts b/server/src/services/chat-action-work.ts index 42993950548..f794cc6a0c2 100644 --- a/server/src/services/chat-action-work.ts +++ b/server/src/services/chat-action-work.ts @@ -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`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`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; + processPendingGitHubWebhookIngress(): Promise; + processPendingSlackSessionSyncs(): Promise; processPendingSlackBoardMessages(): Promise; processPendingSlackTaskStarts(): Promise; processPendingReceiptReactions(): Promise; @@ -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 }, diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index d81f4ea8744..12bff8621d9 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -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 { + 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, diff --git a/server/src/services/chat-slack-session-outbox.ts b/server/src/services/chat-slack-session-outbox.ts index 4be13a63e46..21b4a0d64a8 100644 --- a/server/src/services/chat-slack-session-outbox.ts +++ b/server/src/services/chat-slack-session-outbox.ts @@ -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[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({ diff --git a/server/src/services/chat-work-notifications.ts b/server/src/services/chat-work-notifications.ts index 50da6074d55..f76ac13c72e 100644 --- a/server/src/services/chat-work-notifications.ts +++ b/server/src/services/chat-work-notifications.ts @@ -11,6 +11,9 @@ export async function notifyChatPublicationWork(tx: object): Promise { } 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, diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 686d860f02c..fdee30932f8 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -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",