From aab87ce70744660a4a21c77278b0ede267c04720 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Mon, 2 Nov 2026 07:59:17 -0800 Subject: [PATCH] Prove workspace delivery before inspecting cleanup candidates (#18493) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Completed tasks leave workspaces that need safe cleanup. > - Providers without ancestry inspection currently need a recorded merged PR before cleanup can start. > - A squash merge changes commit identity, so direct ancestry cannot prove that delivery. > - This change proves delivery from the finalized HEAD and captured destination before storage inspection. > - Fresh storage checks and a second delivery check still protect removal. > - This reuses the existing worker and receipts. It adds no table or timer. ## Linked Issues or Issue Description Refs #18410. Related: #13400 addresses workspace identity during inspection; this change adds destination-aware delivery discovery. ## What Changed - Resolve direct ancestry or discover an associated merged GitHub PR without a recorded work product. Require exact HEAD, destination repository and branch, and a reachable integration commit. - Use the latest settled finalization receipt as a candidate. Reject missing, failed, pending, conflicting, or changed evidence. - Inspect storage only after positive remote proof for providers that disable ancestry inspection. Recheck HEAD, branch, source identity and delivery before removal intent. - Add an optional provider capability for a specific PR recovery hint and an optional prepared destination receipt. Older providers keep their existing protocol. - Wake cleanup when finalization writes settle or gain a workspace binding. Keep remote-only retries for concrete unresolved delivery. Park missing or unsupported candidate identity. - Reuse the source project's explicit GitHub consent for delivery checks through the existing authorization service. Revalidate the bound repository and consent before removal; missing or revoked consent retains storage without switching identities. Public sources remain anonymous. - Update the contracts, documentation, access map, and targeted regressions. ## Verification - Initial-head GitHub CI passed at `5acd4c818be6f1764f8319f42891f9956c357c26`, including full build and typecheck. Standard Greptile is 5/5 with no open threads. The first Apex review scored 3/5 and found startup latency from remote proof lookups and removal retry identity changes across provider capability changes. Both corrections are pushed in `9d1a38df3d`, with the qualified destination-ref length fix. Red regressions reproduced each issue. Latest targeted checks passed: cleanup/worker 26, provider integration 41, SDK contract 8. Head `9d1a38df3d` passed standard and Apex reviews at 5/5 with no open threads, plus all CI including full build/typecheck. Canary Dry Run initially failed on an npm 404 fetching `rollup-4.67.3.tgz`; the exact tarball subsequently returned HTTP 200, and one targeted retry passed without source changes. Aggregate verification also passed. After #18403 merged, master introduced conflicts in shared workspace cleanup and metadata handling. Resolved them in `7700832536`, preserving remote snapshot cleanup and Git delivery-proof cleanup. Targeted merge verification: 93 tests passed in the first run (59 opt-in skips), then all 16 remote lifecycle tests passed with explicit database opt-in; map checks passed. The isolated database is stopped. The merge head received standard review 5/5. Its CI found two failures in unchanged test cleanup paths: native runtime temporary-directory ENOTEMPTY and agent-chat cancellation lock timeout. Those failures remain recorded and require successful final-head verification. The subsequent project-consent change has targeted coverage: authorization36, resolver33 and cleanup19 tests passed, including revoked consent, anonymous public access and source-binding races. The positive consent test failed before the fix. Current head `c772679e05` has green GitHub CI, including full build/typecheck and both previously failing test jobs, plus standard Greptile 5/5 with no open threads. Neither cleanup failure required changes to its test path. Apex also completed at 5/5 on this exact head (check `124651548532`, summary updated 2026-10-31 20:45 UTC), with no open review threads. The PR remains draft. Earlier failures remain recorded above. - Final focused run on the rebased branch: **182 tests passed across six files**, covering the resolver, delivery cleanup, provider bridge, finalization notifications, existing workspace service and SDK contracts. The disposable test database was stopped afterward. - Targeted resolver tests: 33 passed. These cover ancestry, discovered squash PRs, wrong HEAD/base/repository, removed integration commits, bounded responses, cancellation and credential isolation. - Targeted cleanup integration tests: 17 passed. Provider bridge integration: 36 passed. SDK contract tests: 7 passed. Finalization notification tests: 9 passed. - A negative control that removed the final delivery check failed the race regression as expected. The check was restored. - Read-only live GitHub probe: the new resolver discovered PR #18410 from HEAD `3f5ade4cd42b707040e9a82fdaed343e3d736a09`, without PR hints. It proved integration commit `6484f96c28bfb11a793ce4be6cc5ac15643a5a02` reachable from observed master `5bc029bf6c1db6647e96acea323f3723fd1d3eba`. GitHub also advertised the exact original HEAD at `refs/pull/18410/head`. - `node scripts/check-maps.mjs` passed. Structural checks do not prove deployment behavior. - Local tests are targeted by user request. Full build, typecheck and test suite run in GitHub CI. A standalone server typecheck could not complete because local Runner build declarations were absent; no changed-file diagnostics were reported. - Initial local execution hit a stalled pnpm wrapper, sandbox socket denial and stale generated SDK declarations. Direct installed binaries, the authorized isolated test database and rebuilt SDK resolved those setup failures. No deployment was performed. ## Risks - Destructive cleanup still requires fresh storage safety and lifecycle checks. Remote proof alone never authorizes removal. A persisted removal intent preserves existing recovery semantics. - Legacy workspaces without an unambiguous destination or completion HEAD remain retained. This does not guess the original default branch. - Private repositories use the explicit standing GitHub consent on the bound source project, with current membership, grant, repository access and trust checks. No separate company token is required when that consent is valid. Missing or revoked consent and unavailable or rate-limited access retain storage; the worker does not borrow a task or agent identity. - A known undelivered commit keeps the existing five-minute retry because external GitHub changes have no guaranteed ingress to this lane. This can keep cleanup scheduled. Missing evidence parks instead. - Providers need the new recovery capability to clean a squash merge whose original HEAD is reachable only through its PR ref. Providers without it remain conservative. - The resolver currently supports GitHub.com. Existing local and other-provider ancestry paths stay in place. ## Model Used OpenAI Codex, GPT-6 family, with reasoning, repository tools and parallel code agents. The exact runtime model ID and context-window size were not exposed to 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 | 61 +++-- doc/SPEC-implementation.md | 6 +- doc/execution-github-identity.md | 3 + doc/plugins/PLUGIN_RUNNER_EXECUTION.md | 50 ++++- maps/agent-access-map/workspaces.md | 5 + maps/feature-map/workspaces.md | 9 +- .../plugins/sdk/src/workspace-provider.ts | 11 + .../sdk/tests/workspace-provider.test.ts | 33 +++ packages/shared/src/types/plugin.ts | 1 + packages/shared/src/validators/plugin.ts | 1 + .../github-workspace-delivery.test.ts | 159 +++++++++++++ .../plugin-workspaces.integration.test.ts | 144 +++++++----- .../project-git-authorization.test.ts | 55 +++++ .../__tests__/terminal-workspace-work.test.ts | 41 +++- ...space-delivery-cleanup.integration.test.ts | 212 ++++++++++++++++++ ...orkspace-operations-reconciliation.test.ts | 57 +++++ server/src/services/execution-workspaces.ts | 134 ++++++++--- .../src/services/github-workspace-delivery.ts | 189 ++++++++++++++++ server/src/services/plugin-workspaces.ts | 42 +++- .../src/services/project-git-authorization.ts | 4 +- .../src/services/terminal-workspace-work.ts | 6 +- server/src/services/workspace-operations.ts | 19 +- 22 files changed, 1105 insertions(+), 137 deletions(-) create mode 100644 server/src/__tests__/github-workspace-delivery.test.ts create mode 100644 server/src/__tests__/workspace-delivery-cleanup.integration.test.ts create mode 100644 server/src/services/github-workspace-delivery.ts diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 588d7f1b70e..5b6ea5e03d4 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -3697,29 +3697,54 @@ cleanup validation still inspect current Git state; neither uses a display resul to authorize removal. The terminal-workspace reaper uses the database work coordinator. Startup scans -recover existing workspace rows in bounded pages. Issue status and tree changes, -workspace lifecycle writes, and pull-request work-product writes signal work on +recover existing workspace rows in bounded pages. The first readiness sweep +defers the new remote delivery-proof checks while preserving candidate deadlines; +later scheduled work performs them. GitHub latency does not block readiness +through this path. Issue status and tree changes, +workspace lifecycle writes, finalization receipts, and pull-request work-product writes signal work on their transaction before writing. The coordinator wakes after the outer transaction settles and holds idle admission while writes or cleanup are active. Empty and nonterminal-only lanes have no recurring timer or query. Terminal trees schedule their actual cooldown deadline; unresolved delivery or cleanup eligibility is checked again after five minutes. Providers that explicitly declare -`supportsMergeAncestry: false` are inspected for automatic cleanup only when a -matching merged PR can provide delivery evidence. With no matching PR work -product, the unchanged workspace has no delivery-inspection deadline; existing -work-product notifications and startup recovery reconsider it. Open or unknown -remote PRs and unfinished cleanup retain their retries. Missing declarations keep -the previous ancestry-inspection behavior. Nonterminal trees, cooldowns, and active runs -are checked before Git. The worker shares Git readiness across compatible rows -that point at one directory within a bounded page. Company, branch, base ref, -and runtime ownership must match. Each row keeps its own task and delivery -assessment, which reuses the task tree read by the early gate. Unchanged terminal -candidates also back off for five minutes during unrelated commit wakes. Scoped -task/workspace hints invalidate -affected candidates; changed tree or workspace revisions invalidate them too. -Deadline selection reuses task trees read by those pages and performs no separate -full-inventory query. Commits during a paged rotation request another bounded -rotation so changes behind its cursor are recovered. +`supportsMergeAncestry: false` use the latest successful finalization receipt as a +candidate HEAD. Core checks GitHub before inspecting storage: the exact HEAD must +be an ancestor of the intended destination tip, or an associated merged PR must +have that exact final HEAD and destination, with its integration commit still +reachable from the destination. Recorded PR work products are optional lookup +hints. This supports squash merges without equating the squash commit with the +original workspace commit. + +Preparation captures the resolved destination branch when the provider supports +it. Cleanup never silently adopts a changed project source or today's default +branch for an ambiguous old workspace. Missing receipts or destinations park the +workspace; finalization, workspace, issue, provider and work-product changes and +startup reconsider it. A known commit awaiting delivery, or an unavailable remote +lookup, keeps a five-minute remote-only retry. GitHub merges have no guaranteed +local ingress for this lane, so this concrete unresolved delivery obligation can +keep the worker scheduled until delivery is proved. There is no new polling loop. +Background GitHub delivery checks use the source project's explicit GitHub +authorization, with the same consent checks as provider inspection and removal. +The stored source must still match the workspace repository. Each proof, including +the final removal check, revalidates the selected grant, project access and trust. +Public sources use anonymous access; missing or revoked private-source consent +retains storage without falling back to another identity. No task identity is borrowed. + +Only positive delivery evidence triggers a fresh storage inspection. The actual +HEAD and branch must match the receipt. Cleanup revalidates delivery after its +last inspection before recording removal intent. A provider can opt into +`recoveryPullRequestVersion: 1` to receive one specific PR number and original +HEAD for recovery checks. A merged PR alone does not prove that the original +commit remains recoverable after source-branch deletion. The provider must +independently prove recovery of that exact commit and reject any local data. +Missing capability declarations preserve existing provider protocols. +Nonterminal trees, cooldowns, and active runs are checked before Git. The worker +shares readiness across compatible rows within a bounded page. Each row retains +its own task and delivery assessment. Scoped task/workspace hints invalidate +backoff; changed tree or workspace revisions invalidate it too. Deadline selection +reuses task trees read by those pages without a separate full-inventory query. +Commits during a paged rotation request another bounded rotation so changes behind +its cursor are recovered. The backoff retains at most 256 candidates and only applies to the background worker. Final cleanup still checks the current task tree, delivery evidence, ownership, lifecycle generation, and Git state under the diff --git a/doc/SPEC-implementation.md b/doc/SPEC-implementation.md index 224925dbc54..59874d54e8b 100644 --- a/doc/SPEC-implementation.md +++ b/doc/SPEC-implementation.md @@ -1586,7 +1586,11 @@ The current app also exposes V1-supporting surfaces for: server-owned URL expansion, issue/comment/document scopes, status/date filters, issue-level pagination, a bounded `matchesPerIssue` override for machine consumers, and explicit issue/match truncation flags - execution workspaces, project workspaces, workspace runtime services, and workspace operations. Workspace reads - derive `deliveryState` as `merged_via_pr | merged_by_ancestry | unmerged | unknown`; terminal issue trees with a + return stored metadata without storage inspection; explicit readiness derives `deliveryState` as + `merged_via_pr | merged_by_ancestry | unmerged | unknown`. For providers without ancestry inspection, + automatic cleanup proves the finalized HEAD reached its captured destination through GitHub ancestry + or an exact-head merged PR, including squash merges. Recorded PR work products are optional hints. + Fresh storage, recoverability, and lifecycle checks remain mandatory. Terminal issue trees with a merged delivery and no active checkout run become cleanup-eligible with reason `issue_terminal` and are archived through the workspace cleanup path. Reopening the source issue records activity but does not restore that workspace. Before that, terminal isolated worktrees are compacted: ignored, untracked dependency, build, and cache directories diff --git a/doc/execution-github-identity.md b/doc/execution-github-identity.md index f5daa357289..53a4cc5d99a 100644 --- a/doc/execution-github-identity.md +++ b/doc/execution-github-identity.md @@ -210,6 +210,9 @@ tool, and proactive calls do not carry this authority. Repository URLs come from stored project workspaces. Foreground preparation uses the live run's accepted identity and standard trust checks, even when background project consent exists. Maintenance ignores originating run/agent context and uses project consent only. +Core delivery-proof cleanup uses the same project consent before contacting GitHub, +bound to the provider's stored source project and repository. It repeats authorization +for the final proof before removal; a changed source or revoked grant retains storage. After archival, existing consent remains usable only for host-issued workspace inspection and removal. New provisioning and new consent remain blocked. diff --git a/doc/plugins/PLUGIN_RUNNER_EXECUTION.md b/doc/plugins/PLUGIN_RUNNER_EXECUTION.md index 3142a596818..837aa2bdf0b 100644 --- a/doc/plugins/PLUGIN_RUNNER_EXECUTION.md +++ b/doc/plugins/PLUGIN_RUNNER_EXECUTION.md @@ -175,18 +175,44 @@ retain the inspection-based completion path. No stored binding migration is requ A provider whose inspection cannot prove merge ancestry may declare `workspaceProvider.supportsMergeAncestry: false`. Omission preserves the existing -ancestry inspection behavior. For an opted-out provider, automatic cleanup first -checks for a matching PR work product. No matching PR means no storage inspection -or recurring cleanup deadline for that workspace; work-product and workspace -notifications reconsider it, and startup reconstructs the decision. Provider -manifest changes, activation, and company re-enablement also wake cleanup, so a -provider that later gains ancestry inspection can reconsider parked workspaces. -A relevant open or unknown PR retains the existing bounded retry because external merge -state can change without a local notification. Only a merged PR with the expected -branch and a known HEAD triggers inspection; Core still compares that HEAD with -fresh storage evidence before archive. Removal's fresh data-safety and lifecycle -checks are unchanged. Explicit close-readiness and pending lifecycle recovery -continue to inspect even without a PR. +ancestry inspection behavior. For an opted-out provider, automatic cleanup uses +its latest successful finalization HEAD as a hint for destination-aware GitHub +checks. It discovers associated PRs without requiring a recorded work product. +A direct ancestry proof or a merged PR with the exact final HEAD, intended base +repository/branch, and reachable integration commit must precede storage +inspection. A fresh matching HEAD and branch, remote recoverability, no local +data, terminal tasks, and the existing lifecycle fences still govern deletion. +Open or unknown delivery for a known commit retains a bounded remote-only retry. +Missing candidate identity parks until a relevant write or startup recovery. +Explicit close-readiness and pending lifecycle recovery retain their checks. + +A provider may declare `workspaceProvider.recoveryPullRequestVersion: 1`. Core +can then include `recoveryPullRequest: { number, headSha }` in `inspect` and +`remove`. This is a narrow recovery hint, not permission to delete. The provider +must use its trusted source repository, validate that the specific PR recovery +reference still identifies `headSha`, and independently check the actual workspace +and all owned data. Arbitrary remote URLs and reference names are not accepted. +Absent capability means Core does not send the extra field. Before dispatch, +Core persists the negotiated removal payload and its operation ID. Retries replay +both unchanged, even if capabilities change, while still enforcing company, +provider enablement, binding and generation checks. A downgraded provider can +reject an unsupported payload; Core does not silently change its identity. +Legacy intents without a recovery hint retain their deterministic ID. A hinted +intent without a recorded ID is ambiguous and fails closed. None of these +receipts replaces the provider's checks. + +The optional `prepared.destinationRef` records the resolved delivery branch as +`refs/heads/`, up to 266 characters including the prefix. Providers should +return null when preparation cannot establish an unambiguous branch destination +or its canonical ref exceeds this receipt limit; an optional receipt must not +fail an otherwise successful preparation. Core stores this alongside the source +snapshot; callers cannot overwrite it through workspace metadata updates. Older +providers omit it. Core only falls back to their captured explicit source refs, +not a mutable project configuration or a newly chosen default branch. A source +SHA or tag may support direct ancestry, but does not imply a PR base branch. +Provider upgrades, activation, company re-enablement, and finalization settlement +wake cleanup transactionally. No table, independent timer, or background scan of +all remote PR refs is required. ## Warm task retention diff --git a/maps/agent-access-map/workspaces.md b/maps/agent-access-map/workspaces.md index fa5ab4075d0..c2b261a7f4f 100644 --- a/maps/agent-access-map/workspaces.md +++ b/maps/agent-access-map/workspaces.md @@ -320,6 +320,11 @@ Policy: Company/project/workspace visibility and runtime authority; active run, preset and task ownership where required. Workspace metadata reads do not perform storage inspection or certify delivery. The explicit close-readiness API and destructive cleanup retain live safety checks. +Automatic GitHub delivery checks reuse explicit source-project GitHub consent, +rechecking the stored repository, selected grant, project access and trust on each +proof. Missing or revoked consent never falls back to another identity. The resolved destination and +provider recovery hint cannot be supplied as an authorization override through +workspace metadata. Ambiguous or inaccessible proof retains storage. Evidence: `source-only` — [implementation](../../server/src/routes/workspace-runtime-service-authz.ts#L76) at the diff --git a/maps/feature-map/workspaces.md b/maps/feature-map/workspaces.md index 68035f4fb7f..83d5504ff63 100644 --- a/maps/feature-map/workspaces.md +++ b/maps/feature-map/workspaces.md @@ -206,9 +206,12 @@ Manual: Provision a disposable workspace, wait for readiness, start an available encryption key. Edits after the last save still depend on the sandbox; there is no periodic workspace backup. The experiment does not remove the instance volume or establish zero-outage deployment for other data paths. -- Terminal workspace cleanup wakes after committed task/workspace/PR changes and - restores cooldown deadlines at startup. Empty lanes stop scanning. Display Git - results can lag by thirty seconds. Compatible records share one advisory Git +- Terminal workspace cleanup wakes after committed task/workspace/PR/finalization changes and + restores cooldown deadlines at startup. Empty lanes stop scanning. Metadata reads + do not inspect storage. Providers without ancestry inspection use finalized HEAD + and destination-aware GitHub proof before storage checks. Associated squash PRs + need not be recorded manually; wrong HEAD/base, unreachable integration commits, + or unavailable credentials retain the workspace. Compatible records share one advisory Git inspection per bounded page, while each record keeps its own delivery check. Destructive cleanup starts a separate Git scan even if an older scan is running. The [worker tests](../../server/src/__tests__/terminal-workspace-work.test.ts) diff --git a/packages/plugins/sdk/src/workspace-provider.ts b/packages/plugins/sdk/src/workspace-provider.ts index bd28c5610fd..e46a3c4e98e 100644 --- a/packages/plugins/sdk/src/workspace-provider.ts +++ b/packages/plugins/sdk/src/workspace-provider.ts @@ -18,6 +18,14 @@ export const freshWorkspaceSchema = z.object({ export type PluginFreshWorkspace = z.infer; const sha = z.string().regex(/^[a-f0-9]{40,64}$/); +export const workspaceRecoveryPullRequestSchema = z.object({ + number: z.number().int().positive().max(Number.MAX_SAFE_INTEGER), + headSha: z.string().regex(/^(?:[a-f0-9]{40}|[a-f0-9]{64})$/), +}).strict(); +export type PluginWorkspaceRecoveryPullRequest = z.infer; +const destinationRef = z.string().max(266).regex(/^refs\/heads\/[^\x00-\x20\x7f~^:?*\[\]\\]+$/) + .refine(ref => !ref.includes("..") && !ref.includes("@{") && !ref.endsWith(".") && !ref.endsWith("/") + && ref.split("/").every(part => part && !part.startsWith(".") && !part.endsWith(".lock"))); export const workspaceProviderOperationSchema = z.discriminatedUnion("kind", [ z .object({ @@ -51,6 +59,7 @@ export const workspaceProviderOperationSchema = z.discriminatedUnion("kind", [ .object({ kind: z.literal("inspect"), binding: workspaceProviderBindingSchema, + recoveryPullRequest: workspaceRecoveryPullRequestSchema.optional(), }) .strict(), z @@ -58,6 +67,7 @@ export const workspaceProviderOperationSchema = z.discriminatedUnion("kind", [ kind: z.literal("remove"), binding: workspaceProviderBindingSchema, expectedHeadSha: sha.nullable(), + recoveryPullRequest: workspaceRecoveryPullRequestSchema.optional(), }) .strict(), ]); @@ -68,6 +78,7 @@ export const workspaceProviderResultSchema = z.discriminatedUnion("kind", [ binding: workspaceProviderBindingSchema, branchName: z.string().min(1).max(256).nullable(), baseSha: sha.nullable(), + destinationRef: destinationRef.nullable().optional(), created: z.boolean(), branchCreatedByRuntime: z.boolean(), }) diff --git a/packages/plugins/sdk/tests/workspace-provider.test.ts b/packages/plugins/sdk/tests/workspace-provider.test.ts index b21f7320857..74779cfa72c 100644 --- a/packages/plugins/sdk/tests/workspace-provider.test.ts +++ b/packages/plugins/sdk/tests/workspace-provider.test.ts @@ -50,6 +50,35 @@ const result = { branchCreatedByRuntime: true, }; describe("workspace provider SDK", () => { + it("bounds recovery hints to an exact PR number and commit", () => { + for (const kind of ["inspect", "remove"] as const) { + const operation = { kind, binding: result.binding, ...(kind === "remove" ? { expectedHeadSha: "a".repeat(40) } : {}) }; + const recoveryPullRequest = { number: 42, headSha: "a".repeat(40) }; + expect(workspaceProviderOperationSchema.parse({ ...operation, recoveryPullRequest })).toMatchObject({ recoveryPullRequest }); + for (const bad of [{ ...recoveryPullRequest, number: 0 }, { ...recoveryPullRequest, number: 1.5 }, + { ...recoveryPullRequest, headSha: "a".repeat(41) }, { ...recoveryPullRequest, ref: "refs/heads/main" }]) { + expect(() => workspaceProviderOperationSchema.parse({ ...operation, recoveryPullRequest: bad })).toThrow(); + } + } + }); + it("accepts only canonical optional preparation destinations", () => { + for (const destinationRef of [undefined, null, "refs/heads/main", "refs/heads/release/one"]) { + expect(parseWorkspaceProviderResult(params, { ...result, ...(destinationRef !== undefined ? { destinationRef } : {}) })) + .toMatchObject(result); + } + for (const destinationRef of ["main", "refs/tags/main", "refs/heads/main..other", "refs/heads/main/", "refs/heads/.hidden", "refs/heads/main.lock"]) { + expect(() => parseWorkspaceProviderResult(params, { ...result, destinationRef })).toThrow(); + } + }); + it("preserves a 255-character destination branch through preparation and restoration", () => { + const destinationRef = `refs/heads/release/${"a".repeat(247)}`; + expect(destinationRef).toHaveLength(266); + for (const input of [params, { ...params, operation: { ...params.operation, branchMode: "existing" as const, existing: result.binding } }]) { + expect(parseWorkspaceProviderResult(input as PluginWorkspaceParams, { ...result, destinationRef })) + .toMatchObject({ destinationRef }); + expect(() => parseWorkspaceProviderResult(input as PluginWorkspaceParams, { ...result, destinationRef: `${destinationRef}a` })).toThrow(); + } + }); it("separates completion acknowledgement from removal safety", () => { const input: PluginWorkspaceParams = { ...params, agentId: id(6), runId: id(7), operation: { kind: "finalize", binding: result.binding } }; @@ -122,6 +151,10 @@ describe("workspace provider SDK", () => { resourceLifecycle: ["project"], }; expect(pluginManifestV1Schema.safeParse(manifest).success).toBe(true); + expect(pluginManifestV1Schema.parse({ ...manifest, workspaceProvider: { ...manifest.workspaceProvider, + recoveryPullRequestVersion: 1 } }).workspaceProvider?.recoveryPullRequestVersion).toBe(1); + expect(pluginManifestV1Schema.safeParse({ ...manifest, workspaceProvider: { ...manifest.workspaceProvider, + recoveryPullRequestVersion: 2 } }).success).toBe(false); const withDefault = { ...manifest, workspaceProvider: { ...manifest.workspaceProvider, defaultForProjects: { mode: "isolated_workspace", strategy: "git_worktree" } } }; expect(pluginManifestV1Schema.parse(withDefault).workspaceProvider?.defaultForProjects).toEqual({ diff --git a/packages/shared/src/types/plugin.ts b/packages/shared/src/types/plugin.ts index 794f6cecdf4..07c409d7dac 100644 --- a/packages/shared/src/types/plugin.ts +++ b/packages/shared/src/types/plugin.ts @@ -720,6 +720,7 @@ export interface PaperclipPluginManifestV1 { /** Supports bounded run completion separately from removal-safety inspection. */ finalizationVersion?: 1; supportsMergeAncestry?: boolean; + recoveryPullRequestVersion?: 1; freshWorkspaceVersion?: 1; modes: Array<"shared_workspace" | "isolated_workspace" | "operator_branch">; strategies: Array<"project_primary" | "git_worktree">; diff --git a/packages/shared/src/validators/plugin.ts b/packages/shared/src/validators/plugin.ts index 841fcdc23a1..318fac91061 100644 --- a/packages/shared/src/validators/plugin.ts +++ b/packages/shared/src/validators/plugin.ts @@ -809,6 +809,7 @@ export const pluginManifestV1Schema = z.object({ workspaceProvider: z.object({ finalizationVersion: z.literal(1).optional(), supportsMergeAncestry: z.boolean().optional(), + recoveryPullRequestVersion: z.literal(1).optional(), freshWorkspaceVersion: z.literal(1).optional(), modes: z.array(z.enum(["shared_workspace", "isolated_workspace", "operator_branch"])).min(1).max(3), strategies: z.array(z.enum(["project_primary", "git_worktree"])).min(1).max(2), diff --git a/server/src/__tests__/github-workspace-delivery.test.ts b/server/src/__tests__/github-workspace-delivery.test.ts new file mode 100644 index 00000000000..c0fdd87671c --- /dev/null +++ b/server/src/__tests__/github-workspace-delivery.test.ts @@ -0,0 +1,159 @@ +import { describe, expect, it, vi } from "vitest"; +import type { Db } from "@paperclipai/db"; +import { createGitHubWorkspaceDeliveryResolver, type GitHubWorkspaceDeliveryInput } from "../services/github-workspace-delivery.js"; + +const HEAD = "1".repeat(40); +const TIP = "2".repeat(40); +const MERGE = "3".repeat(40); +const OTHER = "4".repeat(40); +const input: GitHubWorkspaceDeliveryInput = { companyId: "company-a", repoUrl: "https://github.com/acme/repo.git", headSha: HEAD, destinationRef: "refs/heads/main" }; +const ref = (name = "heads/main", tip = TIP) => ({ ref: `refs/${name}`, object: { type: "commit", sha: tip } }); +const comparison = (base: string, included: boolean) => ({ status: included ? "ahead" : "diverged", base_commit: { sha: base }, merge_base_commit: { sha: included ? base : OTHER }, behind_by: included ? 0 : 1, ahead_by: 1 }); +const pull = () => ({ number: 17, merged: true, head: { sha: HEAD, repo: { full_name: "fork/repo" } }, base: { ref: "main", repo: { full_name: "acme/repo" } }, merge_commit_sha: MERGE }); +function fixture(extra: Record = {}, ambient = false) { + const routes: Record = { + "/git/ref/heads/main": ref(), + [`/compare/${HEAD}...${TIP}?per_page=1`]: comparison(HEAD, false), + [`/commits/${HEAD}/pulls?per_page=100&page=1`]: [{ number: 17 }], + "/pulls/17": pull(), + [`/compare/${MERGE}...${TIP}?per_page=1`]: comparison(MERGE, true), + ...extra, + }; + const fetch = vi.fn(async (url, init) => { + expect(String(url).startsWith("https://api.github.com/repos/acme/repo")).toBe(true); + expect(init?.redirect).toBe("error"); + expect(init?.signal).toBeInstanceOf(AbortSignal); + const path = String(url).slice("https://api.github.com/repos/acme/repo".length); + const value = routes[path]; + if (typeof value === "function") return value(init); + if (value instanceof Response) return value.clone(); + return value === undefined ? new Response(null, { status: 404 }) : Response.json(value); + }); + const credentialProvider = vi.fn(async () => ambient ? { token: "test-credential", source: "server_env" as const, secretName: null } : null); + const resolve = createGitHubWorkspaceDeliveryResolver({} as Db, { fetch, credentialProvider }); + return { resolve, fetch, credentialProvider }; +} + +describe("GitHub workspace delivery proof", () => { + it("discovers an exact-head squash PR without a recorded work product, including a fork", async () => { + const f = fixture(); + expect(await f.resolve(input)).toEqual({ status: "delivered", headSha: HEAD, destinationRef: "refs/heads/main", destinationTipSha: TIP, evidence: { kind: "merged_pr", number: 17, mergeCommitSha: MERGE } }); + expect(f.fetch).toHaveBeenCalledTimes(5); + expect(f.credentialProvider).toHaveBeenCalledWith("company-a"); + }); + + it("accepts ordinary ancestry without a PR lookup", async () => { + const f = fixture({ [`/compare/${HEAD}...${TIP}?per_page=1`]: comparison(HEAD, true) }); + expect(await f.resolve(input)).toMatchObject({ status: "delivered", evidence: { kind: "ancestry" } }); + expect(f.fetch).toHaveBeenCalledTimes(2); + }); + + it.each([ + ["different source HEAD", { ...pull(), head: { sha: OTHER } }], + ["different destination branch", { ...pull(), base: { ref: "release", repo: { full_name: "acme/repo" } } }], + ["different destination repository", { ...pull(), base: { ref: "main", repo: { full_name: "other/repo" } } }], + ["open PR with test merge commit", { ...pull(), merged: false }], + ])("does not accept %s", async (_label, response) => { + const f = fixture({ "/pulls/17": response }); + expect(await f.resolve(input)).toEqual({ status: "not_delivered" }); + }); + + it("rejects an integration commit removed from the destination by force push", async () => { + const f = fixture({ [`/compare/${MERGE}...${TIP}?per_page=1`]: comparison(MERGE, false) }); + expect(await f.resolve(input)).toEqual({ status: "not_delivered" }); + }); + + it.each([null, "", "HEAD", "main~1", "refs/remotes/untrusted/main", "refs/heads/../main"])("rejects unresolved or unsupported destination %s before authentication", async (destinationRef) => { + const f = fixture(); + expect(await f.resolve({ ...input, destinationRef })).toMatchObject({ status: "unknown" }); + expect(f.credentialProvider).not.toHaveBeenCalled(); + expect(f.fetch).not.toHaveBeenCalled(); + }); + + it.each(["origin/main", "refs/remotes/origin/main", "main"])("normalizes branch %s while preserving exact destination", async (destinationRef) => { + const f = fixture(); + expect(await f.resolve({ ...input, destinationRef })).toMatchObject({ status: "delivered", destinationRef: "refs/heads/main" }); + }); + + it("keeps a bare branch/tag collision unknown even if they have the same tip", async () => { + const f = fixture({ "/git/ref/tags/main": ref("tags/main") }); + expect(await f.resolve({ ...input, destinationRef: "main" })).toEqual({ status: "unknown", reason: "ambiguous_destination_ref" }); + }); + + it.each(["refs/tags/release", TIP])("allows only direct ancestry for tag/SHA destination %s", async (destinationRef) => { + const f = fixture({ "/git/ref/tags/release": ref("tags/release"), [`/commits/${TIP}?per_page=1`]: { sha: TIP } }); + expect(await f.resolve({ ...input, destinationRef })).toEqual({ status: "not_delivered" }); + expect(f.fetch).toHaveBeenCalledTimes(2); + }); + + it("resolves an annotated tag to its commit before testing ancestry", async () => { + const f = fixture({ "/git/ref/tags/release": { ref: "refs/tags/release", object: { type: "tag", sha: OTHER } }, "/commits/tags%2Frelease?per_page=1": { sha: TIP }, [`/compare/${HEAD}...${TIP}?per_page=1`]: comparison(HEAD, true) }); + expect(await f.resolve({ ...input, destinationRef: "refs/tags/release" })).toMatchObject({ status: "delivered", destinationRef: "refs/tags/release", destinationTipSha: TIP }); + }); + + it("does not borrow instance credentials for a private repository", async () => { + const f = fixture({ "": { private: true, full_name: "acme/repo" } }, true); + expect(await f.resolve(input)).toEqual({ status: "unknown", reason: "ambient_credential_requires_public_repository" }); + expect(f.fetch).toHaveBeenCalledTimes(1); + }); + + it("probes ambient access again on each observation", async () => { + let calls = 0; + const f = fixture({ "": () => Response.json({ private: calls++ > 0, full_name: "acme/repo" }) }, true); + expect(await f.resolve(input)).toMatchObject({ status: "delivered" }); + expect(await f.resolve(input)).toMatchObject({ status: "unknown" }); + }); + + it.each([401, 403, 404, 429, 503])("preserves unknown on GitHub HTTP %s", async (status) => { + const f = fixture({ "/git/ref/heads/main": new Response(null, { status }) }); + expect(await f.resolve(input)).toMatchObject({ status: "unknown" }); + }); + + it("rejects a malformed positive comparison instead of trusting status alone", async () => { + const f = fixture({ [`/compare/${HEAD}...${TIP}?per_page=1`]: { ...comparison(HEAD, true), merge_base_commit: { sha: OTHER } } }); + expect(await f.resolve(input)).toEqual({ status: "unknown", reason: "invalid_comparison_response" }); + }); + + it("keeps unknown if association pagination exceeds the bounded search", async () => { + const page = () => Response.json([], { headers: { link: '; rel="next"' } }); + const f = fixture(Object.fromEntries([1, 2, 3].map((n) => [`/commits/${HEAD}/pulls?per_page=100&page=${n}`, page]))); + expect(await f.resolve(input)).toEqual({ status: "unknown", reason: "association_budget_exhausted" }); + expect(f.fetch).toHaveBeenCalledTimes(5); + }); + + it("bounds PR candidate lookups and does not mistake exhaustion for a negative", async () => { + const f = fixture({ [`/commits/${HEAD}/pulls?per_page=100&page=1`]: Array.from({ length: 100 }, (_, i) => ({ number: i + 100 })) }); + expect(await f.resolve(input)).toEqual({ status: "unknown", reason: "request_budget_exhausted" }); + expect(f.fetch).toHaveBeenCalledTimes(24); + }); + + it("propagates the shared request cancellation signal to an active transport", async () => { + const controller = new AbortController(); + const timeout = vi.spyOn(AbortSignal, "timeout").mockReturnValue(controller.signal); + let started!: () => void; + const waiting = new Promise((resolve) => { started = resolve; }); + const f = fixture({ "/git/ref/heads/main": (init: RequestInit) => new Promise((_resolve, reject) => { + init.signal!.addEventListener("abort", () => reject(new Error("cancelled")), { once: true }); + started(); + }) }); + try { + const result = f.resolve(input); + await waiting; + controller.abort(); + expect(await result).toEqual({ status: "unknown", reason: "github_unavailable" }); + expect(timeout).toHaveBeenCalledWith(20_000); + } finally { timeout.mockRestore(); } + }); + + it("redacts transport failures", async () => { + const f = fixture({ "/git/ref/heads/main": () => { throw new Error("test-credential secret provider response"); } }); + expect(await f.resolve(input)).toEqual({ status: "unknown", reason: "github_unavailable" }); + }); + + it("cancels and rejects an oversized provider body", async () => { + const cancel = vi.fn(); + const f = fixture({ "/git/ref/heads/main": () => new Response(new ReadableStream({ start(controller) { controller.enqueue(new Uint8Array(2 * 1024 * 1024 + 1)); }, cancel })) }); + expect(await f.resolve(input)).toEqual({ status: "unknown", reason: "response_too_large" }); + expect(cancel).toHaveBeenCalled(); + }); +}); diff --git a/server/src/__tests__/plugin-workspaces.integration.test.ts b/server/src/__tests__/plugin-workspaces.integration.test.ts index b838a7b7408..1e5af2ae97b 100644 --- a/server/src/__tests__/plugin-workspaces.integration.test.ts +++ b/server/src/__tests__/plugin-workspaces.integration.test.ts @@ -45,6 +45,90 @@ const support = await getEmbeddedPostgresTestSupport(); }); const fixture = (companyId?: string) => createPluginWorkspaceFixture(db, companyId); + it.each([1, undefined])("gates recovery hints by provider capability (%s)", async (version) => { + const f = await fixture(); + const [plugin] = await db.select().from(plugins).where(eq(plugins.id, f.pluginId)); + await db.update(plugins).set({ manifestJson: { ...plugin.manifestJson, + workspaceProvider: { ...plugin.manifestJson.workspaceProvider!, recoveryPullRequestVersion: version as 1 | undefined } } }).where(eq(plugins.id, f.pluginId)); + const workspace = await preparePluginWorkspace(db, f.workers, f.input); + const recoveryPullRequest = { number: 42, headSha: "a".repeat(40) }; + await inspectPluginWorkspace(db, f.workers, workspace!, recoveryPullRequest); + expect(f.calls.at(-1).operation.recoveryPullRequest).toEqual(version === 1 ? recoveryPullRequest : undefined); + let freshInspectionCompleted = false; + await expect(removePluginWorkspace(db, f.workers, workspace!, "a".repeat(40), recoveryPullRequest, async () => { + freshInspectionCompleted = f.calls.at(-1).operation.kind === "inspect"; + throw new Error("delivery changed"); + })).rejects.toThrow("delivery changed"); + expect(freshInspectionCompleted).toBe(true); + const [unchanged] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspace!.id)); + expect(unchanged.metadata?.workspaceProviderRemoval).toBeUndefined(); + f.loseRemoval(); + await expect(removePluginWorkspace(db, f.workers, workspace!, "a".repeat(40), recoveryPullRequest)).rejects.toThrow(); + const [pending] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspace!.id)); + expect((pending.metadata?.workspaceProviderRemoval as { operation: { recoveryPullRequest?: unknown } }).operation.recoveryPullRequest) + .toEqual(version === 1 ? recoveryPullRequest : undefined); + await removePluginWorkspace(db, f.workers, pending, "a".repeat(40), recoveryPullRequest, async () => { throw new Error("must not reauthorize pending removal"); }); + expect(f.calls.at(-1).operation.recoveryPullRequest).toEqual(version === 1 ? recoveryPullRequest : undefined); + }); + it.each([1, undefined])("replays a lost removal acknowledgement unchanged after capability flip (%s)", async (version) => { + const f = await fixture(); + const [plugin] = await db.select().from(plugins).where(eq(plugins.id, f.pluginId)); + const setCapability = (recoveryPullRequestVersion: 1 | undefined) => db.update(plugins).set({ manifestJson: { + ...plugin.manifestJson, workspaceProvider: { ...plugin.manifestJson.workspaceProvider!, recoveryPullRequestVersion }, + } }).where(eq(plugins.id, f.pluginId)); + await setCapability(version as 1 | undefined); + const workspace = await preparePluginWorkspace(db, f.workers, f.input); + f.loseRemoval(); + await expect(removePluginWorkspace(db, f.workers, workspace!, "a".repeat(40), { number: 42, headSha: "a".repeat(40) })).rejects.toThrow(); + const first = f.calls.at(-1); + await db.update(plugins).set({ status: "disabled" }).where(eq(plugins.id, f.pluginId)); + const count = f.calls.length; + await expect(removePluginWorkspace(db, f.workers, workspace!, "a".repeat(40))).rejects.toThrow("unavailable"); + expect(f.calls).toHaveLength(count); + await db.update(plugins).set({ status: "ready" }).where(eq(plugins.id, f.pluginId)); + await setCapability(version === 1 ? undefined : 1); + const [pending] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspace!.id)); + await removePluginWorkspace(db, f.workers, pending, "a".repeat(40)); + expect(f.calls.at(-1).operation).toEqual(first.operation); + expect(f.calls.at(-1).operationId).toBe(first.operationId); + expect(pending.metadata?.workspaceProviderRemoval).toMatchObject({ operation: first.operation, operationId: first.operationId }); + const [removed] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspace!.id)); + expect(removed.metadata?.workspaceProviderRemoved).toBe(true); + }); + it.each(["legacy", "ambiguous", "mismatched"] as const)("fails closed for uncertain removal identities (%s)", async (identity) => { + const hint = identity !== "legacy"; + const f = await fixture(); + const [plugin] = await db.select().from(plugins).where(eq(plugins.id, f.pluginId)); + await db.update(plugins).set({ manifestJson: { ...plugin.manifestJson, + workspaceProvider: { ...plugin.manifestJson.workspaceProvider!, recoveryPullRequestVersion: 1 } } }).where(eq(plugins.id, f.pluginId)); + const workspace = await preparePluginWorkspace(db, f.workers, f.input); + f.loseRemoval(); + await expect(removePluginWorkspace(db, f.workers, workspace!, "a".repeat(40), hint ? { number: 42, headSha: "a".repeat(40) } : undefined)).rejects.toThrow(); + const first = f.calls.at(-1); + const [pending] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspace!.id)); + const intent = pending.metadata!.workspaceProviderRemoval as Record; + if (identity === "mismatched") intent.operationId = "0".repeat(64); + else delete intent.operationId; + await db.update(executionWorkspaces).set({ metadata: { ...pending.metadata, workspaceProviderRemoval: intent } }).where(eq(executionWorkspaces.id, pending.id)); + if (hint) { + const count = f.calls.length; + await expect(removePluginWorkspace(db, f.workers, pending, "a".repeat(40))).rejects.toThrow(identity === "ambiguous" ? "identity is ambiguous" : "identity does not match"); + expect(f.calls).toHaveLength(count); + } else { + await removePluginWorkspace(db, f.workers, pending, "a".repeat(40)); + expect(f.calls.at(-1).operationId).toBe(first.operationId); + expect(f.calls.at(-1).operation).toEqual(first.operation); + } + }); + it.each(["refs/heads/main", null, undefined])("persists the actual prepared destination (%s)", async (destinationRef) => { + const f = await fixture(); + const workers: WorkspaceWorkers = { ...f.workers, call: async (...args: Parameters) => { + const result = await f.workers.call(...args); + return result.kind === "prepared" ? { ...result, ...(destinationRef !== undefined ? { destinationRef } : {}) } : result; + } }; + const workspace = await preparePluginWorkspace(db, workers, f.input); + expect(workspace!.metadata?.workspaceProviderDestinationRef).toBe(destinationRef); + }); it.each([true, undefined])("wakes parked cleanup when a provider upgrade enables ancestry (%s)", async (supportsMergeAncestry) => { const f = await fixture(); await preparePluginWorkspace(db, f.workers, f.input); @@ -103,68 +187,28 @@ const support = await getEmbeddedPostgresTestSupport(); expect(events).toEqual([]); } finally { unsubscribe(); } }); - it("keeps PR-only storage idle until delivery can qualify for fresh inspection", async () => { + it("keeps storage idle without a settled completion HEAD, even with a recorded PR", async () => { const f = await fixture(); - const workspace = (await preparePluginWorkspace(db, f.workers, f.input))!; + await preparePluginWorkspace(db, f.workers, f.input); const [plugin] = await db.select().from(plugins).where(eq(plugins.id, f.pluginId)); await db.update(plugins).set({ manifestJson: { ...plugin.manifestJson, workspaceProvider: { ...plugin.manifestJson.workspaceProvider!, supportsMergeAncestry: false } } }).where(eq(plugins.id, f.pluginId)); await db.update(issues).set({ status: "done", completedAt: new Date() }).where(eq(issues.id, f.issueId)); await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.runId)); - const originalCall = f.workers.call.bind(f.workers); - f.workers.call = async (...args) => { - const value = await originalCall(...args); - return (args[2] as any).operation.kind === "inspect" ? { ...value as object, isMergedIntoBase: null } : value; - }; - let clock = new Date(); - let state: "open" | "merged" = "open"; - let headSha = "b".repeat(40); - let pullRequestReads = 0; - const options = { pluginWorkerManager: f.workers, workspaceReaperCooldownDays: 0, - terminalWorkspaceInspectionBackoffMs: 300_000, now: () => clock, - resolvePullRequestDetails: async () => { pullRequestReads += 1; return { state, headRef: workspace.branchName, headSha }; } }; - const svc = executionWorkspaceService(db, options); - const inspections = () => f.calls.filter(call => call.operation.kind === "inspect").length; - await svc.sweepTerminalWorkspaces(100); - expect(inspections()).toBe(0); - expect(await svc.nextTerminalWorkspaceSweepAt()).toBeNull(); - svc.terminalWorkspaceWorkSettled(new Set([`issue:${randomUUID()}`])); - await svc.sweepTerminalWorkspaces(100); - expect(inspections()).toBe(0); - expect(await svc.nextTerminalWorkspaceSweepAt()).toBeNull(); - await executionWorkspaceService(db, options).sweepTerminalWorkspaces(100); - expect(inspections()).toBe(0); - - await db.insert(issueWorkProducts).values({ companyId: f.companyId, issueId: f.issueId, - type: "pull_request", provider: "github", title: "Unrelated work", status: "active", - url: "https://github.com/other/repository/pull/1" }); - svc.terminalWorkspaceWorkSettled(new Set([`issue:${f.issueId}`])); - await svc.sweepTerminalWorkspaces(100); - expect(inspections()).toBe(0); - expect(await svc.nextTerminalWorkspaceSweepAt()).toBeNull(); - await db.insert(issueWorkProducts).values({ companyId: f.companyId, issueId: f.issueId, type: "pull_request", provider: "github", title: "Work", status: "active", url: "https://github.com/example/repo/pull/1" }); - svc.terminalWorkspaceWorkSettled(new Set([`issue:${f.issueId}`])); + const options = { pluginWorkerManager: f.workers, workspaceReaperCooldownDays: 0, + resolveWorkspaceDelivery: async () => { throw new Error("No delivery lookup without a completion HEAD"); } }; + const svc = executionWorkspaceService(db, options); await svc.sweepTerminalWorkspaces(100); - expect(inspections()).toBe(0); - expect(await svc.nextTerminalWorkspaceSweepAt()).not.toBeNull(); - const readsBeforeUnrelatedWake = pullRequestReads; + expect(f.calls.some(call => call.operation.kind === "inspect")).toBe(false); + expect(await svc.nextTerminalWorkspaceSweepAt()).toBeNull(); svc.terminalWorkspaceWorkSettled(new Set([`issue:${randomUUID()}`])); await svc.sweepTerminalWorkspaces(100); - expect(pullRequestReads).toBe(readsBeforeUnrelatedWake); - expect(inspections()).toBe(0); - - state = "merged"; - clock = new Date(clock.getTime() + 300_001); - await svc.sweepTerminalWorkspaces(100); - expect(inspections()).toBe(1); - expect(f.calls.some(call => call.operation.kind === "remove")).toBe(false); - headSha = "a".repeat(40); - clock = new Date(clock.getTime() + 300_001); - await svc.sweepTerminalWorkspaces(100); - expect(f.calls.some(call => call.operation.kind === "remove")).toBe(true); + expect(await svc.nextTerminalWorkspaceSweepAt()).toBeNull(); + await executionWorkspaceService(db, options).sweepTerminalWorkspaces(100); + expect(f.calls.some(call => call.operation.kind === "inspect")).toBe(false); }); it("uses bounded finalization only for opted-in providers, preserving maintenance inspection", async () => { const f = await fixture(); diff --git a/server/src/__tests__/project-git-authorization.test.ts b/server/src/__tests__/project-git-authorization.test.ts index 9bd67aa5a3b..e250d3ae44c 100644 --- a/server/src/__tests__/project-git-authorization.test.ts +++ b/server/src/__tests__/project-git-authorization.test.ts @@ -8,6 +8,7 @@ import { createDb, authUsers, companyMemberships, connectionGrants, executionWor import { LOW_TRUST_REVIEW_PRESET } from "@paperclipai/shared"; import { ghFetch } from "../services/github-fetch.js"; import { projectRoutes } from "../routes/projects.js"; +import { createGitHubWorkspaceDeliveryResolver } from "../services/github-workspace-delivery.js"; import { projectGitAuthorizationService } from "../services/project-git-authorization.js"; import { errorHandler } from "../middleware/index.js"; import { githubCredentialsFixture } from "./helpers/github-credentials-fixture.js"; @@ -49,6 +50,60 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu const request = { companyId: input.companyId, projectId: project.id, workspaceId: workspace.id }; return { input, selected, project, workspace, service, request, scope: { projectId: project.id } }; } + it("uses project consent for private delivery checks and rechecks revocation without a task", async () => { + const f = await projectAuthorizationFixture(); + await f.service.set(f.input.companyId, f.project.id, "A", f.selected.id); + await db.update(projects).set({ archivedAt: new Date() }).where(eq(projects.id, f.project.id)); + await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.input.runId)); + const head = "a".repeat(40); + const fetch = vi.fn(async (_url, init) => { + expect(new Headers(init?.headers).get("authorization")).toBe("Bearer test-token-A"); + return Response.json({ ref: "refs/heads/main", object: { type: "commit", sha: head } }); + }); + const resolve = createGitHubWorkspaceDeliveryResolver(db, { fetch }); + const input = { companyId: f.input.companyId, repoUrl: f.workspace.repoUrl, headSha: head, + destinationRef: "refs/heads/main", projectSource: { projectId: f.project.id, workspaceId: f.workspace.id } }; + expect(await resolve(input)).toMatchObject({ status: "delivered" }); + expect(fetch).toHaveBeenCalledTimes(1); + await db.update(connectionGrants).set({ status: "revoked" }).where(eq(connectionGrants.id, f.selected.id)); + expect(await resolve(input)).toMatchObject({ status: "unknown" }); + expect(fetch).toHaveBeenCalledTimes(1); + }); + it.each(["missing consent", "different repository", "different company", "different project"])("retains delivery with %s without exporting credentials", async fault => { + const f = await projectAuthorizationFixture(); + if (fault !== "missing consent") await f.service.set(f.input.companyId, f.project.id, "A", f.selected.id); + vault.resolveUserSecretValue.mockClear(); + const fetch = vi.fn(); + const credentialProvider = vi.fn(async () => ({ token: "legacy-token", source: "company_secret" as const, secretName: "GITHUB_TOKEN" })); + const resolve = createGitHubWorkspaceDeliveryResolver(db, { fetch, credentialProvider }); + expect(await resolve({ companyId: fault === "different company" ? randomUUID() : f.input.companyId, + repoUrl: fault === "different repository" ? "https://github.com/other/repo" : f.workspace.repoUrl, + headSha: "a".repeat(40), destinationRef: "refs/heads/main", + projectSource: { projectId: fault === "different project" ? randomUUID() : f.project.id, workspaceId: f.workspace.id }, + })).toMatchObject({ status: "unknown" }); + expect(fetch).not.toHaveBeenCalled(); + expect(credentialProvider).not.toHaveBeenCalled(); + expect(vault.resolveUserSecretValue).not.toHaveBeenCalled(); + }); + it("checks public project delivery anonymously without acquiring a different credential", async () => { + const f = await projectAuthorizationFixture(); + const [saved] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, f.selected.id)); + await db.update(connectionGrants).set({ providerTenant: { ...saved.providerTenant, github: { ...saved.providerTenant!.github!, + repositories: [{ id: "1", fullName: "fixture/project", installationId: "1", private: false }] } } }).where(eq(connectionGrants.id, f.selected.id)); + const head = "a".repeat(40); + const fetch = vi.fn(async (_url, init) => { + expect(new Headers(init?.headers).has("authorization")).toBe(false); + return Response.json({ ref: "refs/heads/main", object: { type: "commit", sha: head } }); + }); + const credentialProvider = vi.fn(async () => ({ token: "legacy-token", source: "company_secret" as const, secretName: "GITHUB_TOKEN" })); + vault.resolveUserSecretValue.mockClear(); + const resolve = createGitHubWorkspaceDeliveryResolver(db, { fetch, credentialProvider }); + expect(await resolve({ companyId: f.input.companyId, repoUrl: f.workspace.repoUrl, headSha: head, + destinationRef: "refs/heads/main", projectSource: { projectId: f.project.id, workspaceId: f.workspace.id }, + })).toMatchObject({ status: "delivered" }); + expect(credentialProvider).not.toHaveBeenCalled(); + expect(vault.resolveUserSecretValue).not.toHaveBeenCalled(); + }); it("uses explicit project consent without a task and journals consent changes", async () => { const f = await projectAuthorizationFixture(); await expect(f.service.resolve(f.request, f.scope)).rejects.toThrow("Authorize this project"); diff --git a/server/src/__tests__/terminal-workspace-work.test.ts b/server/src/__tests__/terminal-workspace-work.test.ts index df72716ce3f..e44d37f14a6 100644 --- a/server/src/__tests__/terminal-workspace-work.test.ts +++ b/server/src/__tests__/terminal-workspace-work.test.ts @@ -22,7 +22,7 @@ function setup() { }); coordinators.push(coordinator); const service = { - sweepTerminalWorkspaces: vi.fn(async (_limit?: number, _signal?: AbortSignal, _options?: { compact?: boolean }) => ({ checked: 0 })), + sweepTerminalWorkspaces: vi.fn(async (_limit?: number, _signal?: AbortSignal, _options?: { compact?: boolean; allowRemoteDeliveryChecks?: boolean }) => ({ checked: 0 })), hasTerminalWorkspaceSweepContinuation: vi.fn(() => false), nextTerminalWorkspaceSweepAt: vi.fn(async (): Promise => null), }; @@ -44,15 +44,48 @@ describe("terminal workspace work", () => { expect(worker.hasUnrecoveredWork()).toBe(false); }); - it("skips compaction only in the startup sweep that readiness waits for", async () => { + it("defers compaction and remote delivery checks only in the startup sweep that readiness waits for", async () => { const s = setup(); s.service.hasTerminalWorkspaceSweepContinuation.mockReturnValueOnce(true); await s.start().ready; expect(s.service.sweepTerminalWorkspaces).toHaveBeenCalledTimes(1); - expect(s.service.sweepTerminalWorkspaces.mock.calls[0]?.[2]).toEqual({ compact: false }); + expect(s.service.sweepTerminalWorkspaces.mock.calls[0]?.[2]).toEqual({ compact: false, allowRemoteDeliveryChecks: false }); await vi.advanceTimersByTimeAsync(1_000); expect(s.service.sweepTerminalWorkspaces).toHaveBeenCalledTimes(2); - expect(s.service.sweepTerminalWorkspaces.mock.calls[1]?.[2]).toEqual({ compact: true }); + expect(s.service.sweepTerminalWorkspaces.mock.calls[1]?.[2]).toEqual({ compact: true, allowRemoteDeliveryChecks: true }); + }); + + it("retains deferred delivery across drain and resumes it outside readiness", async () => { + const s = setup(); + let deadline: number | null = Date.now(); + let release!: () => void; + const remote = vi.fn(() => new Promise(resolve => { release = resolve; })); + s.service.hasTerminalWorkspaceSweepContinuation.mockReturnValueOnce(true); + s.service.nextTerminalWorkspaceSweepAt.mockImplementation(async () => deadline); + s.service.sweepTerminalWorkspaces.mockImplementation(async (_limit, _signal, options) => { + if (options?.allowRemoteDeliveryChecks === false) return { checked: 1 }; + await remote(); + deadline = null; + return { checked: 1 }; + }); + const worker = s.start(); + try { + await worker.ready; + expect(remote).not.toHaveBeenCalled(); + expect(s.coordinator.nextWakeAt()).toBe(Date.now() + 1000); + s.enabled.mockReturnValue(false); + await vi.advanceTimersByTimeAsync(1000); + expect(remote).not.toHaveBeenCalled(); + expect(s.service.sweepTerminalWorkspaces).toHaveBeenCalledTimes(1); + expect(s.coordinator.nextWakeAt()).not.toBeNull(); + s.enabled.mockReturnValue(true); + await vi.advanceTimersByTimeAsync(TERMINAL_WORKSPACE_RETRY_MS); + expect(remote).toHaveBeenCalledTimes(1); + release(); + await vi.advanceTimersByTimeAsync(1); + expect(s.coordinator.nextWakeAt()).toBeNull(); + expect(worker.hasUnrecoveredWork()).toBe(false); + } finally { release?.(); } }); it("restores the cooldown deadline, lets commits preempt it, and parks after cleanup", async () => { diff --git a/server/src/__tests__/workspace-delivery-cleanup.integration.test.ts b/server/src/__tests__/workspace-delivery-cleanup.integration.test.ts new file mode 100644 index 00000000000..19e23b62e75 --- /dev/null +++ b/server/src/__tests__/workspace-delivery-cleanup.integration.test.ts @@ -0,0 +1,212 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { eq } from "drizzle-orm"; +import { createDb, executionWorkspaces, heartbeatRuns, issues, plugins, projectWorkspaces, workspaceOperations } from "@paperclipai/db"; +import { createPluginWorkspaceFixture } from "./helpers/plugin-workspaces.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { preparePluginWorkspace } from "../services/plugin-workspaces.js"; +import { executionWorkspaceService } from "../services/execution-workspaces.js"; +import type { GitHubWorkspaceDeliveryInput, GitHubWorkspaceDeliveryResult } from "../services/github-workspace-delivery.js"; + +const support = await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("remote delivery cleanup", () => { + let db: ReturnType; + let stopDb: (() => Promise) | undefined; + beforeAll(async () => { + const database = await startEmbeddedPostgresTestDatabase("workspace-delivery-cleanup"); + db = createDb(database.connectionString, { maxConnections: 1 }); + stopDb = database.stop; + }); + afterAll(async () => { await db?.$client.end(); await stopDb?.(); }); + afterEach(async () => { await db.update(issues).set({ status: "in_progress" }); }); + async function fixture() { + const f = await createPluginWorkspaceFixture(db); + await db.update(projectWorkspaces).set({ repoRef: "main" }).where(eq(projectWorkspaces.id, f.input.projectWorkspaceId)); + const [plugin] = await db.select().from(plugins).where(eq(plugins.id, f.pluginId)); + await db.update(plugins).set({ manifestJson: { ...plugin.manifestJson, + workspaceProvider: { ...plugin.manifestJson.workspaceProvider!, supportsMergeAncestry: false, recoveryPullRequestVersion: 1 } } }).where(eq(plugins.id, f.pluginId)); + const workspace = (await preparePluginWorkspace(db, f.workers, f.input))!; + await db.update(issues).set({ status: "done", completedAt: new Date() }).where(eq(issues.id, f.issueId)); + await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.runId)); + const [receipt] = await db.insert(workspaceOperations).values({ companyId: f.companyId, executionWorkspaceId: workspace.id, + heartbeatRunId: f.runId, issueId: f.issueId, phase: "workspace_finalize", status: "succeeded", finishedAt: new Date(), + metadata: { pluginWorkspace: { providerType: "plugin", inspection: { kind: "finalized", headSha: "a".repeat(40), branchName: workspace.branchName } } } }).returning(); + let clock = new Date(); + let proof: GitHubWorkspaceDeliveryResult = { status: "not_delivered" }; + const resolve = vi.fn(async (_input: GitHubWorkspaceDeliveryInput) => proof); + const original = f.workers.call.bind(f.workers); + f.workers.call = async (...args) => { + const result = await original(...args); + return (args[2] as any).operation.kind === "inspect" ? { ...result as object, isMergedIntoBase: null } : result; + }; + const service = executionWorkspaceService(db, { pluginWorkerManager: f.workers, workspaceReaperCooldownDays: 0, + terminalWorkspaceInspectionBackoffMs: 300_000, now: () => clock, resolveWorkspaceDelivery: resolve }); + const delivered = (kind: "ancestry" | "merged_pr" = "ancestry"): GitHubWorkspaceDeliveryResult => ({ status: "delivered", + headSha: "a".repeat(40), destinationRef: "refs/heads/main", destinationTipSha: "c".repeat(40), + evidence: kind === "ancestry" ? { kind } : { kind, number: 42, mergeCommitSha: "b".repeat(40) } }); + return { ...f, workspace, receipt, resolve, service, delivered, + setProof: (next: GitHubWorkspaceDeliveryResult) => { proof = next; }, + advance: () => { clock = new Date(clock.getTime() + 300_001); }, + inspections: () => f.calls.filter(call => call.operation.kind === "inspect"), + removals: () => f.calls.filter(call => call.operation.kind === "remove") }; + } + it("defers stalled remote proofs at startup without losing paged cleanup deadlines", async () => { + const f = await fixture(); f.setProof(f.delivered()); + let release!: (proof: GitHubWorkspaceDeliveryResult) => void; + const stalled = new Promise(resolve => { release = resolve; }); + f.resolve.mockImplementationOnce(async () => stalled); + let startupFinished = false; + const startup = f.service.sweepTerminalWorkspaces(1, undefined, { allowRemoteDeliveryChecks: false }) + .then(() => { startupFinished = true; }); + try { + await vi.waitFor(() => expect(startupFinished).toBe(true), { timeout: 1500 }); + expect(f.resolve).not.toHaveBeenCalled(); + expect(f.inspections()).toHaveLength(0); + expect(f.service.hasTerminalWorkspaceSweepContinuation()).toBe(true); + const deadline = await f.service.nextTerminalWorkspaceSweepAt(); + expect(deadline).not.toBeNull(); + await f.service.sweepTerminalWorkspaces(1); + expect(f.service.hasTerminalWorkspaceSweepContinuation()).toBe(false); + expect(await f.service.nextTerminalWorkspaceSweepAt()).toBe(deadline); + let maintenanceFinished = false; + const maintenance = f.service.sweepTerminalWorkspaces(1).then(() => { maintenanceFinished = true; }); + await vi.waitFor(() => expect(f.resolve).toHaveBeenCalledTimes(1)); + expect(maintenanceFinished).toBe(false); + release(f.delivered()); + await maintenance; + expect(f.removals()).toHaveLength(1); + } finally { + release(f.delivered()); + await startup; + } + }); + it("retries a known commit remotely and inspects only after delivery without a work product", async () => { + const f = await fixture(); + await f.service.sweepTerminalWorkspaces(); + expect(f.resolve).toHaveBeenCalledWith(expect.objectContaining({ headSha: "a".repeat(40), destinationRef: "main", pullRequests: [], projectSource: { projectId: f.input.projectId, workspaceId: f.input.projectWorkspaceId } })); + expect(f.inspections()).toHaveLength(0); + expect(await f.service.nextTerminalWorkspaceSweepAt()).not.toBeNull(); + f.service.terminalWorkspaceWorkSettled(new Set([`issue:${randomUUID()}`])); + await f.service.sweepTerminalWorkspaces(); + expect(f.resolve).toHaveBeenCalledTimes(1); + f.setProof(f.delivered()); f.advance(); + await f.service.sweepTerminalWorkspaces(); + expect(f.removals()).toHaveLength(1); + }); + it("parks unsupported destinations across unrelated wakes", async () => { + const f = await fixture(); f.setProof({ status: "unknown", reason: "unsupported_destination_ref" }); + await f.service.sweepTerminalWorkspaces(); + expect(await f.service.nextTerminalWorkspaceSweepAt()).toBeNull(); + f.service.terminalWorkspaceWorkSettled(new Set([`issue:${randomUUID()}`])); + await f.service.sweepTerminalWorkspaces(); + expect(await f.service.nextTerminalWorkspaceSweepAt()).toBeNull(); + expect(f.resolve).toHaveBeenCalledTimes(1); + expect(f.inspections()).toHaveLength(0); + }); + it("passes exact discovered PR recovery evidence through inspection and removal", async () => { + const f = await fixture(); f.setProof(f.delivered("merged_pr")); + await f.service.sweepTerminalWorkspaces(); + expect(f.removals()).toHaveLength(1); + const expected = { number: 42, headSha: "a".repeat(40) }; + expect(f.inspections().every(call => JSON.stringify(call.operation.recoveryPullRequest) === JSON.stringify(expected))).toBe(true); + expect(f.removals()[0].operation.recoveryPullRequest).toEqual(expected); + }); + it.each(["running", "failed", "wrong_branch", "missing_head"])("retains storage when the latest receipt is %s", async (fault) => { + const f = await fixture(); f.setProof(f.delivered()); + await db.insert(workspaceOperations).values({ companyId: f.companyId, executionWorkspaceId: f.workspace.id, + phase: "workspace_finalize", status: fault === "running" || fault === "failed" ? fault : "succeeded", + createdAt: new Date(f.receipt.createdAt.getTime() + 1000), finishedAt: fault === "running" ? null : new Date(), + metadata: { pluginWorkspace: { inspection: { headSha: fault === "missing_head" ? null : "a".repeat(40), branchName: "different" } } } }); + await f.service.sweepTerminalWorkspaces(); + expect(f.resolve).not.toHaveBeenCalled(); + expect(f.inspections()).toHaveLength(0); + expect(await f.service.nextTerminalWorkspaceSweepAt()).toBeNull(); + }); + it("retains storage when conflicting receipts have indistinguishable ordering", async () => { + const f = await fixture(); f.setProof(f.delivered()); + await db.insert(workspaceOperations).values({ companyId: f.companyId, executionWorkspaceId: f.workspace.id, + phase: "workspace_finalize", status: "failed", createdAt: f.receipt.createdAt, finishedAt: new Date() }); + await f.service.sweepTerminalWorkspaces(); + expect(f.resolve).not.toHaveBeenCalled(); + expect(f.inspections()).toHaveLength(0); + }); + it.each([null, "HEAD"])("parks ambiguous explicit destination %s despite a source fallback", async destination => { + const f = await fixture(); f.setProof(f.delivered()); + await db.update(executionWorkspaces).set({ metadata: { ...f.workspace.metadata, + workspaceProviderDestinationRef: destination } }).where(eq(executionWorkspaces.id, f.workspace.id)); + await f.service.sweepTerminalWorkspaces(); + expect(f.resolve).not.toHaveBeenCalled(); + expect(f.inspections()).toHaveLength(0); + expect(await f.service.nextTerminalWorkspaceSweepAt()).toBeNull(); + }); + it.each(["head", "branch", "dirty", "new_receipt"])("rejects %s changes during storage inspection", async fault => { + const f = await fixture(); f.setProof(f.delivered()); + const original = f.workers.call.bind(f.workers); + f.workers.call = async (...args) => { + const result = await original(...args); + if ((args[2] as any).operation.kind !== "inspect") return result; + if (fault === "new_receipt") await db.insert(workspaceOperations).values({ companyId: f.companyId, + executionWorkspaceId: f.workspace.id, phase: "workspace_finalize", status: "failed", + createdAt: new Date(f.receipt.createdAt.getTime() + 1000), finishedAt: new Date() }); + return { ...result as object, ...(fault === "head" ? { headSha: "b".repeat(40) } + : fault === "branch" ? { branchName: "changed" } : fault === "dirty" ? { hasDirtyTrackedFiles: true } : {}) }; + }; + await f.service.sweepTerminalWorkspaces(); + expect(f.removals()).toHaveLength(0); + const [row] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspace.id)); + expect(row.closedAt).toBeNull(); + }); + it("retains storage when destination proof disappears after the final inspection", async () => { + const f = await fixture(); + f.resolve.mockResolvedValueOnce(f.delivered("merged_pr")).mockResolvedValue({ status: "not_delivered" }); + await f.service.sweepTerminalWorkspaces(); + expect(f.inspections()).toHaveLength(2); + expect(f.removals()).toHaveLength(0); + expect(f.resolve).toHaveBeenCalledTimes(2); + const [row] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspace.id)); + expect(row.status).toBe("cleanup_failed"); + expect(row.metadata?.workspaceProviderRemoval).toBeUndefined(); + }); + it("retains storage when the source project changes during the final inspection", async () => { + const f = await fixture(); f.setProof(f.delivered()); + const original = f.workers.call.bind(f.workers); + let inspections = 0; + f.workers.call = async (...args) => { + const result = await original(...args); + if ((args[2] as any).operation.kind === "inspect" && ++inspections === 2) { + const [row] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspace.id)); + await db.update(executionWorkspaces).set({ metadata: { ...row.metadata, + workspaceProvider: { ...row.metadata!.workspaceProvider as object, projectId: randomUUID() }, + } }).where(eq(executionWorkspaces.id, f.workspace.id)); + } + return result; + }; + await f.service.sweepTerminalWorkspaces(); + expect(inspections).toBe(2); + expect(f.resolve).toHaveBeenCalledTimes(1); + expect(f.removals()).toHaveLength(0); + }); + it("fences destination edits made during inspection before archiving", async () => { + const f = await fixture(); f.setProof(f.delivered()); + const original = f.workers.call.bind(f.workers); + f.workers.call = async (...args) => { + const result = await original(...args); + if ((args[2] as any).operation.kind === "inspect") await db.update(executionWorkspaces).set({ + metadata: { ...f.workspace.metadata, workspaceProviderDestinationRef: "refs/heads/other" }, + }).where(eq(executionWorkspaces.id, f.workspace.id)); + return result; + }; + await f.service.sweepTerminalWorkspaces(); + expect(f.removals()).toHaveLength(0); + const [row] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspace.id)); + expect(row.closedAt).toBeNull(); + expect(row.metadata?.workspaceProviderDestinationRef).toBe("refs/heads/other"); + }); + it("re-resolves delivery and recovery evidence for an archived cleanup retry", async () => { + const f = await fixture(); f.setProof(f.delivered("merged_pr")); + await db.update(executionWorkspaces).set({ status: "cleanup_failed", closedAt: new Date() }).where(eq(executionWorkspaces.id, f.workspace.id)); + await f.service.sweepTerminalWorkspaces(); + expect(f.resolve).toHaveBeenCalled(); + expect(f.removals()).toHaveLength(1); + }); +}); diff --git a/server/src/__tests__/workspace-operations-reconciliation.test.ts b/server/src/__tests__/workspace-operations-reconciliation.test.ts index 4414d511432..0488294a9c7 100644 --- a/server/src/__tests__/workspace-operations-reconciliation.test.ts +++ b/server/src/__tests__/workspace-operations-reconciliation.test.ts @@ -12,6 +12,8 @@ import { workspaceOperations, } from "@paperclipai/db"; import { eq } from "drizzle-orm"; +import { subscribeDatabaseWork } from "@paperclipai/db/work-signals"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; import { getWorkspaceOperationLogStore } from "../services/workspace-operation-log-store.js"; import { RUNTIME_CONTROL_STALE_AFTER_MS, @@ -105,6 +107,61 @@ describeEmbeddedPostgres("workspace operation reconciliation", () => { await tempDb?.cleanup(); }); + async function seedFinalizationWorkspace() { + const companyId = randomUUID(), projectId = randomUUID(), executionWorkspaceId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Delivery receipt wake", + issuePrefix: `W${companyId.replace(/-/g, "").slice(0, 7).toUpperCase()}` }); + await db.insert(projects).values({ id: projectId, companyId, name: "Delivery", status: "in_progress" }); + await db.insert(executionWorkspaces).values({ id: executionWorkspaceId, companyId, projectId, + mode: "isolated_workspace", strategyType: "git_worktree", name: "Finalization", status: "active" }); + return { companyId, executionWorkspaceId }; + } + + it.each(["succeeded", "failed"] as const)("wakes cleanup after %s finalization commits, never at a nested savepoint", async status => { + const scope = await seedFinalizationWorkspace(); + const settled: string[][] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.terminalWorkspaces, (event, hints) => { + if (event === "settled") settled.push([...(hints ?? [])]); + }); + try { + await db.transaction(async tx => { + const recorder = workspaceOperationService(tx as unknown as typeof db).createRecorder(scope); + const operation = await recorder.recordOperation({ phase: "workspace_finalize", run: async () => ({ status }) }); + expect(operation.status).toBe(status); + expect(settled).toEqual([]); + }); + expect(settled).toEqual([[`workspace:${scope.executionWorkspaceId}`]]); + const rows = await db.select().from(workspaceOperations); + expect(rows).toHaveLength(1); + expect(rows[0]?.status).toBe(status); + } finally { unsubscribe(); } + }); + + it("wakes the attached workspace when an already settled finalization gains its identity", async () => { + const scope = await seedFinalizationWorkspace(); + const recorder = workspaceOperationService(db).createRecorder({ companyId: scope.companyId }); + const operation = await recorder.recordOperation({ phase: "workspace_finalize", run: async () => ({}) }); + const settled: string[][] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.terminalWorkspaces, (event, hints) => { + if (event === "settled") settled.push([...(hints ?? [])]); + }); + try { + await recorder.attachExecutionWorkspaceId(scope.executionWorkspaceId); + expect(settled).toEqual([[`workspace:${scope.executionWorkspaceId}`]]); + expect((await workspaceOperationService(db).getById(operation.id))?.executionWorkspaceId).toBe(scope.executionWorkspaceId); + } finally { unsubscribe(); } + }); + + it("does not publish a finalization receipt rolled back by the outer transaction", async () => { + const scope = await seedFinalizationWorkspace(); + await expect(db.transaction(async tx => { + await workspaceOperationService(tx as unknown as typeof db).createRecorder(scope) + .recordOperation({ phase: "workspace_finalize", run: async () => ({}) }); + throw new Error("rollback receipt"); + })).rejects.toThrow("rollback receipt"); + expect(await db.select().from(workspaceOperations)).toEqual([]); + }); + it("terminalizes an orphaned runtime start with an inspectable reconciliation log", async () => { const companyId = randomUUID(); const projectId = randomUUID(); diff --git a/server/src/services/execution-workspaces.ts b/server/src/services/execution-workspaces.ts index 04b22af37f2..ba0d2fa8353 100644 --- a/server/src/services/execution-workspaces.ts +++ b/server/src/services/execution-workspaces.ts @@ -1,6 +1,6 @@ import { notifyStatusCardIssueChange } from "./status-card-notifications.js"; import { effectiveProjectWorkspacePolicy } from "./project-workspace-provider.js"; -import { readRemoteWorkspaceAuthority } from "@paperclipai/shared"; +import { readRemoteWorkspaceAuthority, readWorkspaceProviderBinding } from "@paperclipai/shared"; import { cleanupRemoteWorkspaceAllocation, inspectRemoteWorkspacePersistence, type DestroyRemoteWorkspaceLease } from "./remote-workspace-lifecycle.js"; import { inspectPluginWorkspace, removePluginWorkspace, restorePluginWorkspace, pluginWorkspaceSupportsMergeAncestry, type WorkspaceWorkers } from "./plugin-workspaces.js"; import { acquireExecutionWorkspaceLifecycleLock, withExecutionWorkspaceLifecycleLock } from "./execution-workspace-lifecycle-lock.js"; @@ -101,6 +101,7 @@ import { type GitHubPullRequestReference, type PullRequestMergeDetailsResolver, } from "./github-pull-request-merge.js"; +import { createGitHubWorkspaceDeliveryResolver, type GitHubWorkspaceDeliveryResolver, type GitHubWorkspaceDeliveryResult } from "./github-workspace-delivery.js"; import { visibleIssueCondition } from "./issue-visibility.js"; import { createIssueGitRemoteAuthProvider } from "./git-credentials.js"; import { readProjectWorkspaceRuntimeConfig } from "./project-workspace-runtime-config.js"; @@ -285,7 +286,7 @@ function preserveWorkspaceMetadataForPatch( } if (patch.metadata !== undefined) { nextPatch.metadata = { ...nextPatch.metadata }; - for (const key of ["workspaceProvider", "workspaceProviderOwner", "workspaceProviderSource", "workspaceProviderBaseSha", "workspaceProviderRemoved", "workspaceProviderBranchMode", "workspaceProviderRemoval", "workspaceProviderRetired", "workspaceProviderRestoration", "lifecycleGeneration", "reopenPendingConsumption", "reopenPendingConsumptionSince"] as const) { + for (const key of ["workspaceProvider", "workspaceProviderOwner", "workspaceProviderSource", "workspaceProviderDestinationRef", "workspaceProviderBaseSha", "workspaceProviderRemoved", "workspaceProviderBranchMode", "workspaceProviderRemoval", "workspaceProviderRetired", "workspaceProviderRestoration", "lifecycleGeneration", "reopenPendingConsumption", "reopenPendingConsumptionSince"] as const) { if (current.metadata?.[key] !== undefined) nextPatch.metadata[key] = current.metadata[key]; else delete nextPatch.metadata[key]; } @@ -340,6 +341,7 @@ export type ExecutionWorkspaceServiceOptions = { taskDirectoryRetention?: { dryRun: boolean; terminalDelayMs: number; orphanGraceMs: number; perSweep?: number; trackedLimit?: number }; terminalWorkspaceInspectionBackoffMs?: number; resolvePullRequestDetails?: PullRequestMergeDetailsResolver; + resolveWorkspaceDelivery?: GitHubWorkspaceDeliveryResolver; now?: () => Date; beforeTerminalWorkspaceCleanup?: (workspace: ExecutionWorkspaceRow) => Promise; // The terminal-workspace reaper waits this many days after an issue tree @@ -1443,7 +1445,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic const destroyRemoteLease: DestroyRemoteWorkspaceLease = input => opts.destroyRemoteWorkspaceLease ? opts.destroyRemoteWorkspaceLease(input) : import("./environment-runtime.js").then(({ environmentRuntimeService }) => environmentRuntimeService(db).destroyRemoteWorkspaceLease(input)); - async function inspectWorkspace(workspace: ExecutionWorkspace) { + async function inspectWorkspace(workspace: ExecutionWorkspace, recoveryPullRequest?: { number: number; headSha: string }) { if (readRemoteWorkspaceAuthority(workspace.metadata)) { const state = await inspectRemoteWorkspacePersistence(db, workspace); return { statusInspectionSucceeded: state.ready, warnings: state.reason ? [state.reason] : [], git: null }; @@ -1452,7 +1454,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic const [row] = await db.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.id, workspace.id), eq(executionWorkspaces.companyId, workspace.companyId))); if (!row) throw notFound("Workspace is unavailable"); let state; - try { state = await inspectPluginWorkspace(db, opts.pluginWorkerManager, row); } + try { state = await inspectPluginWorkspace(db, opts.pluginWorkerManager, row, recoveryPullRequest); } catch { return { statusInspectionSucceeded: false, warnings: ["Workspace storage inspection is unavailable; data was retained."], git: null }; } return { statusInspectionSucceeded: true, warnings: [], git: { repoRoot: `/workspaces/${workspace.id}`, workspacePath: `/workspaces/${workspace.id}`, @@ -1748,6 +1750,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic const recoveryActionsSvc = issueRecoveryActionService(db); const resolvePullRequestDetails = opts.resolvePullRequestDetails ?? createPullRequestMergeDetailsResolver(db); + const resolveWorkspaceDelivery = opts.resolveWorkspaceDelivery ?? createGitHubWorkspaceDeliveryResolver(db); const now = opts.now ?? (() => new Date()); // The reaper waits this long after an issue tree becomes terminal before it // archives the workspace. A value of 0 disables the cooldown, so the reaper @@ -1787,7 +1790,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic // sweep and starves eligible rows behind them. The cursor resets to the start // when a sweep reaches the end of the candidate set. let terminalSweepCursor: { updatedAt: string; id: string } | null = null; - const terminalInspections = new Map; nextAt: number }>(); + const terminalInspections = new Map; nextAt: number; parked?: boolean }>(); const terminalWorkspaceDeadlines = new Map(); const isolatedRehomeDeadlines = new Map(); const isolatedRehomeNotBefore = new Map(); @@ -1964,18 +1967,34 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic return details; } - async function hasMergedDeliveryPullRequest(workspace: ExecutionWorkspaceRow, references: GitHubPullRequestReference[]) { - for (const reference of references) { - const details = await cachedDeliveryPullRequestDetails(workspace, reference); - if (details.state === "merged" && details.headRef === workspace.branchName && details.headSha) return true; - } - return false; + async function workspaceDeliveryCandidate(workspace: ExecutionWorkspaceRow) { + const binding = readWorkspaceProviderBinding(workspace.metadata); + if (!binding) return null; + const source = parseObject(workspace.metadata?.workspaceProviderSource); + const destinationRef = Object.hasOwn(workspace.metadata ?? {}, "workspaceProviderDestinationRef") + ? readNullableString(workspace.metadata?.workspaceProviderDestinationRef) + : readNullableString(workspace.baseRef) ?? readNullableString(source.repoRef) ?? readNullableString(source.defaultRef); + if (!destinationRef || destinationRef === "HEAD" || !parseGitHubRepository(workspace.repoUrl)) return null; + const operations = await db.select({ id: workspaceOperations.id, status: workspaceOperations.status, createdAt: workspaceOperations.createdAt, + finishedAt: workspaceOperations.finishedAt, metadata: workspaceOperations.metadata }) + .from(workspaceOperations).where(and(eq(workspaceOperations.companyId, workspace.companyId), + eq(workspaceOperations.executionWorkspaceId, workspace.id), eq(workspaceOperations.phase, "workspace_finalize"))) + .orderBy(desc(workspaceOperations.createdAt), desc(workspaceOperations.id)).limit(2); + const [operation, previous] = operations; + if (operation && previous && operation.createdAt.getTime() === previous.createdAt.getTime()) return null; + if (!operation || operation.status !== "succeeded" || !operation.finishedAt) return null; + const inspection = parseObject(parseObject(operation.metadata?.pluginWorkspace).inspection); + const headSha = readNullableString(inspection.headSha); + if (!headSha || !/^[a-f0-9]{40}$/i.test(headSha) || !workspace.branchName || inspection.branchName !== workspace.branchName) return null; + return { operationId: operation.id, headSha: headSha.toLowerCase(), destinationRef, + projectSource: { projectId: binding.projectId, workspaceId: binding.projectWorkspaceId } }; } async function assessDelivery( workspace: ExecutionWorkspaceRow, git: ExecutionWorkspaceCloseGitReadiness | null, precomputedIssueTree?: Awaited>, + deliveryProof?: Extract, ) { const issueTree = precomputedIssueTree ?? await listWorkspaceIssueTree(workspace); const sourceIssue = issueTree.find((issue) => issue.id === workspace.sourceIssueId) ?? null; @@ -2000,7 +2019,9 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic .catch(() => null) : null; - if (sourceIssueTerminal) { + if (deliveryProof) { + mergedPullRequest = deliveryProof.evidence.kind === "merged_pr" && deliveryProof.headSha === workspaceHeadSha; + } else if (sourceIssueTerminal) { for (const reference of await listDeliveryPullRequestReferences(workspace)) { const details = await cachedDeliveryPullRequestDetails(workspace, reference); if ( @@ -2026,7 +2047,9 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic sourceIssueTerminal, mergedPullRequest, pullRequestStateUnknown, - isMergedIntoBase: git?.isMergedIntoBase ?? null, + isMergedIntoBase: deliveryProof + ? deliveryProof.evidence.kind === "ancestry" && deliveryProof.headSha === workspaceHeadSha + : git?.isMergedIntoBase ?? null, }), sourceIssueTerminal, subtreeTerminal, @@ -2439,6 +2462,8 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic workspace: ExecutionWorkspaceRow, expectedHeadSha: string | null, capturedGeneration: number, + recoveryPullRequest?: { number: number; headSha: string }, + beforeRemove?: () => Promise, ): Promise<{ cleaned: boolean; warnings: string[]; skippedReopened?: boolean }> { // The gateway holds the per-workspace lifecycle lock across the destructive // actions. A reopen takes the same lock, so a reopen cannot rebuild the @@ -2457,18 +2482,19 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic message: "execution workspace cleanup skipped because it was reopened", }, onSkip: () => ({ cleaned: false, warnings: [], skippedReopened: true }), - write: () => runTerminalWorkspaceCleanup(workspace, expectedHeadSha), + write: () => runTerminalWorkspaceCleanup(workspace, expectedHeadSha, recoveryPullRequest, beforeRemove), }); } - async function runTerminalWorkspaceCleanup(workspace: ExecutionWorkspaceRow, expectedHeadSha: string | null) { + async function runTerminalWorkspaceCleanup(workspace: ExecutionWorkspaceRow, expectedHeadSha: string | null, + recoveryPullRequest?: { number: number; headSha: string }, beforeRemove?: () => Promise) { if (readRemoteWorkspaceAuthority(workspace.metadata)) { if (await workspaceHasActiveRun(workspace)) throw conflict("Workspace still has active execution"); return cleanupRemoteWorkspaceAllocation({ db, workspace, destroyLease: destroyRemoteLease }); } if (workspace.providerType === "plugin") { if (await workspaceHasActiveRun(workspace)) throw conflict("Workspace still has active execution"); - await removePluginWorkspace(db, opts.pluginWorkerManager, workspace, expectedHeadSha); + await removePluginWorkspace(db, opts.pluginWorkerManager, workspace, expectedHeadSha, recoveryPullRequest, beforeRemove); return { cleaned: true, warnings: [] }; } const [ @@ -3354,7 +3380,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic // `compact: false` skips workspace compaction and compaction-trash deletion // for this sweep only. Pending compactions keep their deadlines, so a later // sweep runs them. - sweepTerminalWorkspaces: async (limit = 50, signal?: AbortSignal, options: { compact?: boolean } = {}) => { + sweepTerminalWorkspaces: async (limit = 50, signal?: AbortSignal, options: { compact?: boolean; allowRemoteDeliveryChecks?: boolean } = {}) => { // Skip this sweep while another sweep runs. A concurrent sweep would share // the cursor and the boundary and could corrupt the rotation state. A // skipped tick is safe: the next tick runs the sweep with intact state. @@ -3726,13 +3752,14 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic } continue; } - let deliveryReferences: GitHubPullRequestReference[] | null = null; - if (workspace.providerType === "plugin" && !isClosedExecutionWorkspaceStatus(workspace.status) - && !reopenPending && !workspace.metadata?.workspaceProviderRemoval && !workspace.metadata?.workspaceProviderRestoration) { + let deliveryCandidate: Awaited> = null; + let deliveryProof: Extract | undefined; + if (workspace.providerType === "plugin" && !reopenPending + && !workspace.metadata?.workspaceProviderRemoval && !workspace.metadata?.workspaceProviderRestoration) { try { if (!(await pluginWorkspaceSupportsMergeAncestry(db, opts.pluginWorkerManager, workspace))) { - deliveryReferences = await listDeliveryPullRequestReferences(workspace); - if (deliveryReferences.length === 0) { + deliveryCandidate = await workspaceDeliveryCandidate(workspace); + if (!deliveryCandidate) { result.skippedUndelivered += 1; if (compactionPending && compactedTerminalAt.get(workspace.id) !== latestTerminalAt) terminalWorkspaceDeadlines.set(workspace.id, compactionDueAt); @@ -3745,13 +3772,22 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic continue; } } + if (deliveryCandidate && options.allowRemoteDeliveryChecks === false) { + result.skippedBackoff += 1; + continue; + } if (opts.terminalWorkspaceInspectionBackoffMs) { const revision = createHash("sha256").update(JSON.stringify([ workspace.updatedAt, workspace.metadata, [...issueTree].sort((a, b) => a.id.localeCompare(b.id)), ])).digest("hex"); const previous = terminalInspections.get(workspace.id); - if (previous?.revision === revision && previous.nextAt > now().getTime()) { + if (previous?.revision === revision && (previous.parked || previous.nextAt > now().getTime())) { + if (previous.parked) { + if (compactionPending && compactedTerminalAt.get(workspace.id) !== latestTerminalAt) + terminalWorkspaceDeadlines.set(workspace.id, compactionDueAt); + else terminalWorkspaceDeadlines.delete(workspace.id); + } result.skippedBackoff += 1; continue; } @@ -3761,17 +3797,50 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic nextAt: now().getTime() + opts.terminalWorkspaceInspectionBackoffMs, }); } - if (deliveryReferences !== null) { + if (deliveryCandidate) { try { - if (!(await hasMergedDeliveryPullRequest(workspace, deliveryReferences))) { + const proof = await resolveWorkspaceDelivery({ companyId: workspace.companyId, repoUrl: workspace.repoUrl, + headSha: deliveryCandidate.headSha, destinationRef: deliveryCandidate.destinationRef, projectSource: deliveryCandidate.projectSource, + pullRequests: await listDeliveryPullRequestReferences(workspace) }); + if (proof.status !== "delivered" || proof.headSha !== deliveryCandidate.headSha) { result.skippedUndelivered += 1; + if (proof.status === "unknown" && ["unsupported_source_or_destination", "unsupported_destination_ref"].includes(proof.reason)) { + const inspection = terminalInspections.get(workspace.id); + if (inspection) inspection.parked = true; + if (compactionPending && compactedTerminalAt.get(workspace.id) !== latestTerminalAt) + terminalWorkspaceDeadlines.set(workspace.id, compactionDueAt); + else terminalWorkspaceDeadlines.delete(workspace.id); + } continue; } + deliveryProof = proof; } catch { result.skippedUndelivered += 1; continue; } } + const recoveryPullRequest = deliveryProof?.evidence.kind === "merged_pr" + ? { number: deliveryProof.evidence.number, headSha: deliveryProof.headSha } : undefined; + const expectedDeliveryCandidate = deliveryCandidate; + const expectedDeliveryProof = deliveryProof; + const verifyDeliveryBeforeRemove = expectedDeliveryCandidate && expectedDeliveryProof ? async () => { + const [current] = await db.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.id, workspace.id), + eq(executionWorkspaces.companyId, workspace.companyId))); + const currentCandidate = current ? await workspaceDeliveryCandidate(current) : null; + if (!current || current.repoUrl !== workspace.repoUrl || current.branchName !== workspace.branchName + || current.providerRef !== workspace.providerRef || currentCandidate?.operationId !== expectedDeliveryCandidate.operationId + || currentCandidate.headSha !== expectedDeliveryCandidate.headSha || currentCandidate.destinationRef !== expectedDeliveryCandidate.destinationRef + || currentCandidate.projectSource.projectId !== expectedDeliveryCandidate.projectSource.projectId + || currentCandidate.projectSource.workspaceId !== expectedDeliveryCandidate.projectSource.workspaceId) + throw conflict("Workspace changed after delivery was verified"); + const freshProof = await resolveWorkspaceDelivery({ companyId: current.companyId, repoUrl: current.repoUrl, + headSha: currentCandidate.headSha, destinationRef: currentCandidate.destinationRef, projectSource: currentCandidate.projectSource, + pullRequests: await listDeliveryPullRequestReferences(current) }); + if (freshProof.status !== "delivered" || freshProof.headSha !== expectedDeliveryProof.headSha + || freshProof.destinationRef !== expectedDeliveryProof.destinationRef || freshProof.evidence.kind !== expectedDeliveryProof.evidence.kind + || (recoveryPullRequest && (freshProof.evidence.kind !== "merged_pr" || freshProof.evidence.number !== recoveryPullRequest.number))) + throw conflict("Workspace delivery could not be confirmed before removal"); + } : undefined; const workspacePath = readNullableString(workspace.providerRef) ?? readNullableString(workspace.cwd); const createdByRuntime = workspace.providerType === "git_worktree" ? isRuntimeOwnedGitBranch(workspace.metadata) @@ -3781,7 +3850,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic : `workspace:${workspace.id}`; let inspection = readinessByWorkingTree.get(readinessKey); if (!inspection) { - inspection = inspectWorkspace(toExecutionWorkspace(workspace)); + inspection = inspectWorkspace(toExecutionWorkspace(workspace), recoveryPullRequest); readinessByWorkingTree.set(readinessKey, inspection); } const { git, statusInspectionSucceeded } = await inspection; @@ -3790,7 +3859,12 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic result.skippedUndelivered += 1; continue; } - const assessment = await assessDelivery(workspace, git, issueTree); + if (deliveryCandidate && (git?.headSha !== deliveryCandidate.headSha || git.branchName !== workspace.branchName + || (await workspaceDeliveryCandidate(workspace))?.operationId !== deliveryCandidate.operationId)) { + result.skippedUndelivered += 1; + continue; + } + const assessment = await assessDelivery(workspace, git, issueTree, deliveryProof); if (!assessment.sourceIssueTerminal || !assessment.subtreeTerminal) { if (reopenPending) { // The source issue left the terminal state, so the reopen transition @@ -3899,7 +3973,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic if (workspace.providerType === "plugin" && isClosedExecutionWorkspaceStatus(workspace.status)) { const generation = readExecutionWorkspaceLifecycleGeneration(workspace.metadata); try { - const cleanup = await cleanupTerminalWorkspace(workspace, assessment.workspaceHeadSha, generation); + const cleanup = await cleanupTerminalWorkspace(workspace, assessment.workspaceHeadSha, generation, recoveryPullRequest, verifyDeliveryBeforeRemove); if (cleanup.skippedReopened) result.skippedReopened += 1; else if (!cleanup.cleaned) result.cleanupFailed += 1; else { result.archived += 1; terminalWorkspaceDeadlines.delete(workspace.id); } @@ -3938,6 +4012,8 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic .where(and( eq(executionWorkspaces.id, workspace.id), eq(executionWorkspaces.companyId, workspace.companyId), + ...(deliveryProof ? workspaceSourceKeys.map(key => sql`${executionWorkspaces[key]} IS NOT DISTINCT FROM ${workspace[key]}`) : []), + deliveryProof ? sql`${executionWorkspaces.metadata} IS NOT DISTINCT FROM ${JSON.stringify(workspace.metadata)}::jsonb` : undefined, inArray(executionWorkspaces.status, ["active", "idle", "in_review"]), isNull(executionWorkspaces.closedAt), sql`(${executionWorkspaces.metadata} ->> ${EXECUTION_WORKSPACE_REOPEN_PENDING_METADATA_KEY}) IS DISTINCT FROM 'true'`, @@ -4033,7 +4109,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic archived.metadata as Record | null, ); try { - const cleanup = await cleanupTerminalWorkspace(archived, assessment.workspaceHeadSha, capturedGeneration); + const cleanup = await cleanupTerminalWorkspace(archived, assessment.workspaceHeadSha, capturedGeneration, recoveryPullRequest, verifyDeliveryBeforeRemove); if (cleanup.skippedReopened) result.skippedReopened += 1; else if (!cleanup.cleaned) result.cleanupFailed += 1; else result.archived += 1; diff --git a/server/src/services/github-workspace-delivery.ts b/server/src/services/github-workspace-delivery.ts new file mode 100644 index 00000000000..2563fad2ef3 --- /dev/null +++ b/server/src/services/github-workspace-delivery.ts @@ -0,0 +1,189 @@ +import type { Db } from "@paperclipai/db"; +import { resolveGitHubApiCredential, type GitHubApiCredential } from "./git-credentials.js"; +import { projectGitAuthorizationService } from "./project-git-authorization.js"; +import type { GitHubPullRequestReference } from "./github-pull-request-merge.js"; + +export type GitHubWorkspaceDeliveryInput = { + companyId: string; + projectSource?: { projectId: string; workspaceId: string }; + repoUrl: string | null; + headSha: string | null; + destinationRef: string | null; + pullRequests?: GitHubPullRequestReference[]; +}; +export type GitHubWorkspaceDeliveryResult = + | { + status: "delivered"; + headSha: string; + destinationRef: string; + destinationTipSha: string; + evidence: { kind: "ancestry" } | { kind: "merged_pr"; number: number; mergeCommitSha: string }; + } + | { status: "not_delivered" } + | { status: "unknown"; reason: string }; +export type GitHubWorkspaceDeliveryResolver = (input: GitHubWorkspaceDeliveryInput) => Promise; + +type Options = { + fetch?: typeof fetch; + credentialProvider?: (companyId: string) => Promise; +}; +type RecordValue = Record; +const record = (value: unknown): RecordValue | null => value !== null && typeof value === "object" && !Array.isArray(value) ? value as RecordValue : null; +const sha = (value: unknown): string | null => typeof value === "string" && /^[a-f0-9]{40}$/i.test(value) ? value.toLowerCase() : null; +const unknown = (reason: string): GitHubWorkspaceDeliveryResult => ({ status: "unknown", reason }); + +class Unavailable extends Error {} + +export function createGitHubWorkspaceDeliveryResolver(db: Db, opts: Options = {}): GitHubWorkspaceDeliveryResolver { + const fetchImpl = opts.fetch ?? fetch; + const credentialProvider = opts.credentialProvider ?? ((companyId: string) => resolveGitHubApiCredential(db, companyId)); + return async (input) => { + const repository = /^(?:https?:\/\/(?:www\.)?github\.com\/|ssh:\/\/git@github\.com\/|git@github\.com:)([A-Za-z0-9_.-]+)\/([A-Za-z0-9_.-]+?)(?:\.git)?\/?$/i.exec(input.repoUrl?.trim() ?? ""); + const headSha = sha(input.headSha); + let ref = input.destinationRef?.trim(); + if (!repository || !headSha || !ref || ref === "HEAD") return unknown("unsupported_source_or_destination"); + if (ref.startsWith("refs/remotes/origin/")) ref = `refs/heads/${ref.slice(20)}`; + else if (ref.startsWith("origin/")) ref = `refs/heads/${ref.slice(7)}`; + if (ref.startsWith("refs/") && !ref.startsWith("refs/heads/") && !ref.startsWith("refs/tags/")) return unknown("unsupported_destination_ref"); + if (/[\s~^:?*\[\\\x00-\x1f\x7f]/.test(ref) || ref.includes("..") || ref.includes("@{") || ref.endsWith("/") || ref.endsWith(".") || ref.endsWith(".lock") || ref.split("/").some((part) => !part || part.startsWith("."))) return unknown("unsupported_destination_ref"); + const owner = repository[1]!.toLowerCase(); + const repo = repository[2]!.toLowerCase(); + const fullName = `${owner}/${repo}`; + const prefix = `/repos/${encodeURIComponent(owner)}/${encodeURIComponent(repo)}`; + try { + const token = input.projectSource ? await projectGitAuthorizationService(db).resolve({ + companyId: input.companyId, ...input.projectSource, expectedRepository: fullName, + }, { ...input.projectSource, operation: "inspect" }) : undefined; + const credential = token === undefined ? await credentialProvider(input.companyId) + : token ? { token, source: "project_authorization" as const } : null; + const signal = AbortSignal.timeout(20_000); + let requests = 0; + async function read(path: string, allowMissing = false): Promise<{ body: unknown; next: boolean } | null> { + if (++requests > 24) throw new Unavailable("request_budget_exhausted"); + const response = await fetchImpl(`https://api.github.com${prefix}${path}`, { + redirect: "error", signal, + headers: { + accept: "application/vnd.github+json", "x-github-api-version": "2022-11-28", + ...(credential?.token ? { authorization: `Bearer ${credential.token}` } : {}), + }, + }); + if (response.status === 404 && allowMissing) { await response.body?.cancel(); return null; } + if (!response.ok) { await response.body?.cancel(); throw new Unavailable(`github_http_${response.status}`); } + const reader = response.body?.getReader(); + if (!reader) throw new Unavailable("invalid_response"); + const chunks: Uint8Array[] = []; + let size = 0; + try { + while (true) { + const part = await reader.read(); + if (part.done) break; + size += part.value.byteLength; + if (size > 2 * 1024 * 1024) throw new Unavailable("response_too_large"); + chunks.push(part.value); + } + } finally { await reader.cancel().catch(() => {}); reader.releaseLock(); } + return { body: JSON.parse(Buffer.concat(chunks).toString("utf8")), next: /rel="next"/.test(response.headers.get("link") ?? "") }; + } + if (credential?.source === "server_env") { + const repoData = record((await read(""))?.body); + if (repoData?.private !== false || typeof repoData.full_name !== "string" || repoData.full_name.toLowerCase() !== fullName) return unknown("ambient_credential_requires_public_repository"); + } + async function reference(kind: "heads" | "tags", name: string) { + const result = await read(`/git/ref/${kind}/${encodeURIComponent(name)}`, true); + if (!result) return null; + const body = record(result.body); + const object = record(body?.object); + if (body?.ref !== `refs/${kind}/${name}` || !sha(object?.sha) || (object?.type !== "commit" && !(kind === "tags" && object?.type === "tag"))) throw new Unavailable("invalid_ref_response"); + if (object.type === "commit") return sha(object.sha)!; + const commit = record((await read(`/commits/${encodeURIComponent(`tags/${name}`)}?per_page=1`))?.body); + if (!sha(commit?.sha)) throw new Unavailable("invalid_commit_response"); + return sha(commit?.sha)!; + } + let destinationTipSha: string; + let destinationRef: string; + let branch: string | null = null; + if (sha(ref)) { + const commit = record((await read(`/commits/${ref}?per_page=1`))?.body); + if (sha(commit?.sha) !== sha(ref)) return unknown("invalid_commit_response"); + destinationTipSha = sha(ref)!; + destinationRef = destinationTipSha; + } else if (ref.startsWith("refs/heads/")) { + branch = ref.slice(11); + const tip = await reference("heads", branch); + if (!tip) return unknown("destination_missing_or_inaccessible"); + destinationTipSha = tip; destinationRef = ref; + } else if (ref.startsWith("refs/tags/")) { + const tip = await reference("tags", ref.slice(10)); + if (!tip) return unknown("destination_missing_or_inaccessible"); + destinationTipSha = tip; destinationRef = ref; + } else { + const branchTip = await reference("heads", ref); + const tagTip = await reference("tags", ref); + if (branchTip && tagTip) return unknown("ambiguous_destination_ref"); + if (!branchTip && !tagTip) return unknown("destination_missing_or_inaccessible"); + branch = branchTip ? ref : null; + destinationTipSha = (branchTip ?? tagTip)!; + destinationRef = `refs/${branchTip ? "heads" : "tags"}/${ref}`; + } + const delivered = (evidence: Extract["evidence"]): GitHubWorkspaceDeliveryResult => ({ status: "delivered", headSha, destinationRef, destinationTipSha, evidence }); + async function ancestor(candidate: string): Promise { + if (candidate === destinationTipSha) return true; + const response = await read(`/compare/${candidate}...${destinationTipSha}?per_page=1`, true); + if (!response) return null; + const comparison = record(response.body); + const base = sha(record(comparison?.base_commit)?.sha); + const mergeBase = sha(record(comparison?.merge_base_commit)?.sha); + if (base !== candidate || !mergeBase) throw new Unavailable("invalid_comparison_response"); + if (comparison?.status === "ahead") { + if (mergeBase !== candidate || comparison.behind_by !== 0 || typeof comparison.ahead_by !== "number" || comparison.ahead_by <= 0) throw new Unavailable("invalid_comparison_response"); + return true; + } + if (mergeBase !== candidate && typeof comparison?.behind_by === "number" && comparison.behind_by > 0 + && ((comparison.status === "behind" && comparison.ahead_by === 0) + || (comparison.status === "diverged" && typeof comparison.ahead_by === "number" && comparison.ahead_by > 0))) return false; + throw new Unavailable("invalid_comparison_response"); + } + const direct = await ancestor(headSha); + if (direct === true) return delivered({ kind: "ancestry" }); + if (!branch) return direct === false ? { status: "not_delivered" } : unknown("comparison_unavailable"); + const checked = new Set(); + let incomplete = direct === null; + async function checkPull(number: number): Promise { + if (checked.has(number)) return null; + checked.add(number); + const result = await read(`/pulls/${number}`, true); + if (!result) { incomplete = true; return null; } + const pull = record(result.body); + const base = record(pull?.base); + const baseRepo = record(base?.repo); + const pullHead = record(pull?.head); + if (pull?.number !== number || typeof pull.merged !== "boolean" || !sha(pullHead?.sha) || typeof base?.ref !== "string" || typeof baseRepo?.full_name !== "string") throw new Unavailable("invalid_pull_response"); + if (!pull.merged || sha(pullHead?.sha) !== headSha || base.ref !== branch || baseRepo.full_name.toLowerCase() !== fullName) return null; + const mergeCommitSha = sha(pull.merge_commit_sha); + if (!mergeCommitSha) throw new Unavailable("invalid_merge_commit"); + const reachable = await ancestor(mergeCommitSha); + if (reachable === null) incomplete = true; + return reachable ? delivered({ kind: "merged_pr", number, mergeCommitSha }) : null; + } + for (const hint of input.pullRequests ?? []) { + if (hint.host !== "github.com" || hint.owner.toLowerCase() !== owner || hint.repo.toLowerCase() !== repo || !Number.isSafeInteger(hint.number) || hint.number < 1) continue; + const proof = await checkPull(hint.number); + if (proof) return proof; + } + for (let page = 1; page <= 3; page++) { + const response = await read(`/commits/${headSha}/pulls?per_page=100&page=${page}`); + if (!Array.isArray(response?.body)) throw new Unavailable("invalid_association_response"); + for (const value of response.body) { + const number = record(value)?.number; + if (typeof number !== "number" || !Number.isSafeInteger(number) || number < 1) throw new Unavailable("invalid_association_response"); + const proof = await checkPull(number); + if (proof) return proof; + } + if (!response.next) return incomplete ? unknown("comparison_unavailable") : { status: "not_delivered" }; + } + return unknown("association_budget_exhausted"); + } catch (error) { + return unknown(error instanceof Unavailable ? error.message : "github_unavailable"); + } + }; +} diff --git a/server/src/services/plugin-workspaces.ts b/server/src/services/plugin-workspaces.ts index 09316a0be45..0792926670e 100644 --- a/server/src/services/plugin-workspaces.ts +++ b/server/src/services/plugin-workspaces.ts @@ -21,6 +21,7 @@ import { parseWorkspaceProviderResult, workspaceProviderOperationSchema, type PluginWorkspaceOperation, + type PluginWorkspaceRecoveryPullRequest, type PluginWorkspaceParams, type PluginFreshWorkspace, } from "@paperclipai/plugin-sdk"; @@ -88,6 +89,7 @@ export async function callWorkspaceProvider( advanceGeneration?: boolean; runId?: string | null; agentId?: string | null; + removalOperationId?: string; }, ) { const { workspace } = input; @@ -107,6 +109,11 @@ export async function callWorkspaceProvider( : 0; const operationGeneration = generation + (input.advanceGeneration ? 1 : 0); const operation = workspaceProviderOperationSchema.parse(input.operation); + if (input.removalOperationId !== undefined && (operation.kind !== "remove" || + input.removalOperationId !== createHash("sha256").update(JSON.stringify([workspace.id, operationGeneration, operation])).digest("hex"))) + throw conflict("Workspace removal identity does not match its recorded payload"); + if ((operation.kind === "inspect" || operation.kind === "remove") && input.removalOperationId === undefined + && selectedProvider.manifestJson.workspaceProvider?.recoveryPullRequestVersion !== 1) delete operation.recoveryPullRequest; if (operation.kind === "prepare" && operation.fresh) { if (!workspace.sourceIssueId) throw conflict("Fresh workspace task ownership is missing"); operation.fresh = await stageProviderFreshWorkspace({ db, companyId: workspace.companyId, @@ -117,7 +124,7 @@ export async function callWorkspaceProvider( companyId: workspace.companyId, projectId: workspace.projectId, executionWorkspaceId: workspace.id, - operationId: + operationId: input.removalOperationId ?? ( operation.kind === "inspect" ? randomUUID() : createHash("sha256") @@ -125,7 +132,7 @@ export async function callWorkspaceProvider( JSON.stringify([workspace.id, operationGeneration, operation, ...(operation.kind === "finalize" ? [input.runId, input.agentId] : [])]), ) - .digest("hex"), + .digest("hex")), generation: operationGeneration, deadline: new Date(Date.now() + (operation.kind === "finalize" ? 30_000 : 300_000)).toISOString(), runId: input.runId ?? null, @@ -513,6 +520,7 @@ export async function preparePluginWorkspace( ...workspace.metadata, workspaceProvider: result.binding, workspaceProviderBaseSha: result.baseSha, + ...(result.destinationRef !== undefined ? { workspaceProviderDestinationRef: result.destinationRef } : {}), workspaceProviderSource: sourceInput, workspaceProviderRemoved: false, createdByRuntime: result.branchCreatedByRuntime, @@ -583,6 +591,7 @@ async function reconcilePluginWorkspaceReservation( ...workspace.metadata, workspaceProvider: result.binding, workspaceProviderBaseSha: result.baseSha, + ...(result.destinationRef !== undefined ? { workspaceProviderDestinationRef: result.destinationRef } : {}), createdByRuntime: result.branchCreatedByRuntime, }, }) @@ -605,12 +614,13 @@ async function inspectBinding( db: Db, workers: WorkspaceWorkers | undefined, workspace: Workspace, + recoveryPullRequest?: PluginWorkspaceRecoveryPullRequest, ) { const binding = readWorkspaceProviderBinding(workspace.metadata); if (!binding) throw conflict("Workspace storage binding is unavailable"); const result = await callWorkspaceProvider(db, workers, { workspace, - operation: { kind: "inspect", binding }, + operation: { kind: "inspect", binding, ...(recoveryPullRequest ? { recoveryPullRequest } : {}) }, }); if (result.kind !== "inspection") throw conflict("Workspace inspection is incomplete"); @@ -630,10 +640,10 @@ async function withRetainedWorkspace(db: Db, workers: WorkspaceWorkers | unde }); } -export async function inspectPluginWorkspace(db: Db, workers: WorkspaceWorkers | undefined, workspace: Workspace) { +export async function inspectPluginWorkspace(db: Db, workers: WorkspaceWorkers | undefined, workspace: Workspace, recoveryPullRequest?: PluginWorkspaceRecoveryPullRequest) { return withRetainedWorkspace(db, workers, workspace, (current) => current.metadata?.workspaceProviderRemoved === true - ? Promise.resolve(removedInspection()) : inspectBinding(db, workers, current)); + ? Promise.resolve(removedInspection()) : inspectBinding(db, workers, current, recoveryPullRequest)); } /** Completion does not need the deletion-safety proof supplied by inspect. */ @@ -710,6 +720,7 @@ async function finishPendingRestoration( lifecycleGeneration: generation + 1, workspaceProvider: result.binding, workspaceProviderBaseSha: result.baseSha, + ...(result.destinationRef !== undefined ? { workspaceProviderDestinationRef: result.destinationRef } : {}), workspaceProviderRemoved: false, workspaceProviderRestoration: null, }, @@ -738,7 +749,7 @@ async function finishPendingRemoval( ): Promise { if (workspace.metadata?.workspaceProviderRemoved === true) return workspace; const pending = workspace.metadata?.workspaceProviderRemoval as - | { generation?: unknown; operation?: unknown } + | { generation?: unknown; operation?: unknown; operationId?: unknown } | undefined; const parsed = workspaceProviderOperationSchema.safeParse(pending?.operation); const binding = readWorkspaceProviderBinding(workspace.metadata); @@ -749,9 +760,15 @@ async function finishPendingRemoval( JSON.stringify(parsed.data.binding) !== JSON.stringify(binding) ) throw conflict("Workspace removal intent is unavailable or stale"); + if (pending.operationId === undefined && parsed.data.recoveryPullRequest) + throw conflict("Workspace removal identity is ambiguous; reconciliation requires the original request"); + const operationId = createHash("sha256").update(JSON.stringify([workspace.id, pending.generation, parsed.data])).digest("hex"); + if (pending.operationId !== undefined && pending.operationId !== operationId) + throw conflict("Workspace removal identity does not match its recorded payload"); const result = await callWorkspaceProvider(db, workers, { workspace, operation: parsed.data, + removalOperationId: operationId, }); if (result.kind !== "removed") throw conflict("Workspace removal is incomplete"); @@ -791,6 +808,8 @@ export async function removePluginWorkspace( workers: WorkspaceWorkers | undefined, workspace: Workspace, expectedHeadSha: string | null, + recoveryPullRequest?: PluginWorkspaceRecoveryPullRequest, + beforeRemove?: () => Promise, ) { workspace = await reconcilePluginWorkspaceReservation(db, workers, workspace); if (workspace.metadata?.workspaceProviderRestoration) @@ -806,7 +825,7 @@ export async function removePluginWorkspace( } const binding = readWorkspaceProviderBinding(workspace.metadata); if (!binding) throw conflict("Workspace storage binding is unavailable"); - const inspection = await inspectBinding(db, workers, workspace); + const inspection = await inspectBinding(db, workers, workspace, recoveryPullRequest); if ( inspection.hasDirtyTrackedFiles || inspection.hasUntrackedFiles || @@ -819,8 +838,12 @@ export async function removePluginWorkspace( "Workspace has local changes or cannot be recovered remotely", ); } + await beforeRemove?.(); const generation = workspace.metadata?.lifecycleGeneration ?? 0; - const operation = { kind: "remove" as const, binding, expectedHeadSha }; + const selectedProvider = await provider(db, workers, workspace.companyId, binding.pluginId); + const operation = workspaceProviderOperationSchema.parse({ kind: "remove", binding, expectedHeadSha, + ...(recoveryPullRequest && selectedProvider.manifestJson.workspaceProvider?.recoveryPullRequestVersion === 1 ? { recoveryPullRequest } : {}) }); + const operationId = createHash("sha256").update(JSON.stringify([workspace.id, generation, operation])).digest("hex"); const [pending] = await db.transaction(async (tx) => { await notifyTerminalWorkspaceWork(tx, `workspace:${workspace.id}`); return tx @@ -828,7 +851,7 @@ export async function removePluginWorkspace( .set({ metadata: { ...workspace.metadata, - workspaceProviderRemoval: { generation, operation }, + workspaceProviderRemoval: { generation, operation, operationId }, }, updatedAt: new Date(), }) @@ -962,5 +985,6 @@ export async function restorePluginWorkspace( workspaceProviderRemoved: false, workspaceProviderRestoration: null, workspaceProviderBaseSha: result.baseSha, + ...(result.destinationRef !== undefined ? { workspaceProviderDestinationRef: result.destinationRef } : {}), }; } diff --git a/server/src/services/project-git-authorization.ts b/server/src/services/project-git-authorization.ts index 3124b1f23e0..a017c620dda 100644 --- a/server/src/services/project-git-authorization.ts +++ b/server/src/services/project-git-authorization.ts @@ -136,7 +136,7 @@ export function projectGitAuthorizationService(db: Db) { return read(companyId, projectId, userId); }, async resolve( - input: { companyId: string; projectId: string; workspaceId: string }, + input: { companyId: string; projectId: string; workspaceId: string; expectedRepository?: string }, scope?: { projectId: string; workspaceId?: string; @@ -166,6 +166,8 @@ export function projectGitAuthorizationService(db: Db) { ); if (!workspace?.repoUrl) throw forbidden("Git credentials require a configured project repository"); const repo = normalizeProjectRepositoryUrl(workspace.repoUrl); + if (input.expectedRepository && repo.fullName.toLowerCase() !== input.expectedRepository.toLowerCase()) + throw forbidden("Project repository changed before GitHub authorization"); const isPrivate = await projectRepositoryIsPrivate(db, input.companyId, workspace); if (scope.runId || scope.agentId) { if (!scope.runId || !scope.agentId) throw forbidden("Incomplete foreground Git identity"); diff --git a/server/src/services/terminal-workspace-work.ts b/server/src/services/terminal-workspace-work.ts index ea8b0af51f9..6c86d1ac907 100644 --- a/server/src/services/terminal-workspace-work.ts +++ b/server/src/services/terminal-workspace-work.ts @@ -13,7 +13,7 @@ export function registerTerminalWorkspaceWork( sweepTerminalWorkspaces( limit?: number, signal?: AbortSignal, - options?: { compact?: boolean }, + options?: { compact?: boolean; allowRemoteDeliveryChecks?: boolean }, ): Promise<{ checked: number }>; nextTerminalWorkspaceSweepAt(): Promise; hasTerminalWorkspaceSweepContinuation(): boolean; @@ -33,9 +33,9 @@ export function registerTerminalWorkspaceWork( run: signal => { recovered = false; next = Date.now() + TERMINAL_WORKSPACE_RETRY_MS; - const compact = startupSweepStarted; + const startup = !startupSweepStarted; startupSweepStarted = true; - return service.sweepTerminalWorkspaces(50, signal, { compact }); + return service.sweepTerminalWorkspaces(50, signal, { compact: !startup, allowRemoteDeliveryChecks: !startup }); }, hasPending: async () => { if (!canRun()) next = Date.now() + TERMINAL_WORKSPACE_RETRY_MS; diff --git a/server/src/services/workspace-operations.ts b/server/src/services/workspace-operations.ts index ff39d61a518..acf39177275 100644 --- a/server/src/services/workspace-operations.ts +++ b/server/src/services/workspace-operations.ts @@ -2,6 +2,7 @@ import { notifyAttentionRetentionWork, notifyIssueAttentionRetentionWork } from import { notifyNativeResultCommit } from "./native-result-commit-notifications.js"; import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; +import { notifyTerminalWorkspaceWork } from "./terminal-workspace-work.js"; import { randomUUID } from "node:crypto"; import type { Db } from "@paperclipai/db"; import { workspaceOperations } from "@paperclipai/db"; @@ -473,24 +474,27 @@ export function workspaceOperationService(db: Db) { }): WorkspaceOperationRecorder { let executionWorkspaceId = input.executionWorkspaceId ?? null; const createdIds: string[] = []; + let hasFinalization = false; return { async attachExecutionWorkspaceId(nextExecutionWorkspaceId) { executionWorkspaceId = nextExecutionWorkspaceId ?? null; if (!executionWorkspaceId || createdIds.length === 0) return; - await db - .update(workspaceOperations) - .set({ - executionWorkspaceId, - updatedAt: new Date(), - }) - .where(inArray(workspaceOperations.id, createdIds)); + await db.transaction(async tx => { + if (hasFinalization) await notifyTerminalWorkspaceWork(tx, `workspace:${executionWorkspaceId}`); + await tx + .update(workspaceOperations) + .set({ executionWorkspaceId, updatedAt: new Date() }) + .where(and(eq(workspaceOperations.companyId, input.companyId), inArray(workspaceOperations.id, createdIds))); + }); }, async recordOperation(recordInput) { async function writeState(write: (database: Db) => Promise): Promise { if (recordInput.phase !== "workspace_finalize" || (!input.heartbeatRunId && !executionWorkspaceId && !input.issueId)) return write(db); return db.transaction(async tx => { + if (executionWorkspaceId) await notifyTerminalWorkspaceWork(tx, `workspace:${executionWorkspaceId}`); + else if (input.issueId) await notifyTerminalWorkspaceWork(tx, `issue:${input.issueId}`); if (!input.heartbeatRunId) { // An export clears barriers for every issue sharing this workspace. if (executionWorkspaceId) await notifyAttentionRetentionWork(tx, { companyId: input.companyId, sourceKind: "blocker_attention" }); @@ -579,6 +583,7 @@ export function workspaceOperationService(db: Db) { throw insertError; } createdIds.push(id); + if (recordInput.phase === "workspace_finalize") hasFinalization = true; let heartbeatTimer: NodeJS.Timeout | null = null; let timeoutTimer: NodeJS.Timeout | null = null;