diff --git a/.github/ci-task-matrix.json b/.github/ci-task-matrix.json index b0caf7a5412..e6602d115dd 100644 --- a/.github/ci-task-matrix.json +++ b/.github/ci-task-matrix.json @@ -153,12 +153,12 @@ "task": "ci:runner-rust" }, { - "lane_label": "vitest 1/2", + "lane_label": "vitest package", "command": "test:typescript:vitest --shard=1/2", "task": "ci:runner-vitest" }, { - "lane_label": "vitest 2/2", + "lane_label": "vitest native server", "command": "test:typescript:vitest --shard=2/2", "release_crate": true, "task": "ci:runner-vitest" diff --git a/.github/scripts/tests/ci-dependency-cache.test.mjs b/.github/scripts/tests/ci-dependency-cache.test.mjs index 17c1f274061..1b79b00c401 100644 --- a/.github/scripts/tests/ci-dependency-cache.test.mjs +++ b/.github/scripts/tests/ci-dependency-cache.test.mjs @@ -86,3 +86,17 @@ test('cache restores are exact and only a frozen canonical master push can publi const workflow = readFileSync(new URL('../../workflows/release-verify.yml', import.meta.url), 'utf8'); assert.match(workflow, /save-cache: .*inputs\.ref == github\.sha/); }); + + +test('paired test timings remain available when lifecycle hooks disable materialized caching', () => { + const workflow = readFileSync(new URL('../../workflows/ci-work-benchmark.yml', import.meta.url), 'utf8'); + const step = name => workflow.split(` - name: ${name}\n`)[1].split(' - name:')[0]; + for (const name of ['Save PR-scoped qualification dependencies', 'Save PR-scoped qualification builds', + 'Remove materialized outputs before restore', 'Qualify a real warm restore', + 'Verify exact cache hits and record setup comparison']) { + assert.match(step(name), /if: steps.qualification.outputs.cacheable == 'true'/); + } + assert.match(step('Explain disabled materialized reuse'), /cacheable != 'true'/); + assert.doesNotMatch(step('Prepare unchanged baseline'), /if:/); + assert.doesNotMatch(step('Compare complete commands and executed case inventories'), /if:/); +}); diff --git a/.github/scripts/tests/release-runner-cache.test.mjs b/.github/scripts/tests/release-runner-cache.test.mjs index b6f478122e6..e358cedfde8 100644 --- a/.github/scripts/tests/release-runner-cache.test.mjs +++ b/.github/scripts/tests/release-runner-cache.test.mjs @@ -41,7 +41,7 @@ test("parallel lanes cover check:all exactly once and never bypass verification" assert.deepEqual([...runner.matchAll(/^ - lane: (.+)$/gm)].map(([, value]) => value), ["protocol", "rust", "server-integration"]); assert.match(runner, /fail-fast: false/); assert.doesNotMatch(runner, /max-parallel: 1|^ needs:|continue-on-error:/m); - const verify = runner.split(" - name: Verify Paperclip Runner\n")[1].split(" - name: Warm debug")[0]; + const verify = runner.split(" - name: Verify Paperclip Runner\n")[1].split(" - name:")[0]; assert.match(verify, /RUNNER_CHECKS: \$\{\{ matrix.checks \}\}/); assert.match(verify, /set -euo pipefail/); assert.match(verify, /for check in \$RUNNER_CHECKS; do\s+pnpm --filter @paperclipai\/paperclip-runner "\$check"\s+done/); diff --git a/.github/workflows/ci-work-benchmark.yml b/.github/workflows/ci-work-benchmark.yml index 2ff13f8c7cc..fe7ad35782d 100644 --- a/.github/workflows/ci-work-benchmark.yml +++ b/.github/workflows/ci-work-benchmark.yml @@ -67,18 +67,21 @@ jobs: node scripts/ci-dependency-cache.mjs - name: Save PR-scoped qualification dependencies + if: steps.qualification.outputs.cacheable == 'true' uses: actions/cache/save@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 with: key: ${{ steps.qualification.outputs.dependency-key }} path: ${{ steps.qualification.outputs.dependency-paths }} - name: Save PR-scoped qualification builds + if: steps.qualification.outputs.cacheable == 'true' uses: actions/cache/save@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 with: key: ${{ steps.qualification.outputs.build-key }} path: ${{ steps.qualification.outputs.build-paths }} - name: Remove materialized outputs before restore + if: steps.qualification.outputs.cacheable == 'true' run: | node --input-type=module <<'NODE' import { execFileSync } from 'node:child_process'; @@ -93,12 +96,14 @@ jobs: echo "CI_WARM_STARTED=$(date +%s)" >> "$GITHUB_ENV" - name: Qualify a real warm restore + if: steps.qualification.outputs.cacheable == 'true' id: warm uses: ./candidate/.github/actions/setup-ci-dependencies with: working-directory: candidate - name: Verify exact cache hits and record setup comparison + if: steps.qualification.outputs.cacheable == 'true' env: DEPS_RESTORED: ${{ steps.warm.outputs.dependencies-restored }} BUILDS_RESTORED: ${{ steps.warm.outputs.builds-restored }} @@ -111,6 +116,12 @@ jobs: printf '{"coldSeconds":%s,"warmSeconds":%s}\n' "$CI_COLD_SECONDS" "$warm_seconds" > "$OUTPUT/dependency-setup.json" echo "Dependency setup: ${CI_COLD_SECONDS}s cold, ${warm_seconds}s with exact materialized restores." >> "$GITHUB_STEP_SUMMARY" + - name: Explain disabled materialized reuse + if: steps.qualification.outputs.cacheable != 'true' + run: | + mkdir -p "$RUNNER_TEMP/ci-work-benchmark" + echo "Materialized dependency reuse is disabled by an unmodelled install hook. Retaining cold setup and comparing test commands only." >> "$GITHUB_STEP_SUMMARY" + - name: Prepare unchanged baseline run: | base_sha=$(git rev-parse HEAD^1) diff --git a/.github/workflows/pr-trusted.yml b/.github/workflows/pr-trusted.yml index e0db2a58a2a..1eecbe823b2 100644 --- a/.github/workflows/pr-trusted.yml +++ b/.github/workflows/pr-trusted.yml @@ -646,7 +646,7 @@ jobs: timeout-minutes: 20 strategy: fail-fast: false - matrix: ${{ fromJSON(needs.gate.outputs.plan != '' && toJSON(fromJSON(needs.gate.outputs.plan).jobs.verify_paperclip_runner.matrix) || '{"include":[{"lane_label":"static checks","command":"check:static","task":"ci:runner-static"},{"lane_label":"rust","command":"check:runner","release_crate":true,"task":"ci:runner-rust"},{"lane_label":"vitest 1/2","command":"test:typescript:vitest --shard=1/2","task":"ci:runner-vitest"},{"lane_label":"vitest 2/2","command":"test:typescript:vitest --shard=2/2","release_crate":true,"task":"ci:runner-vitest"}]}') }} + matrix: ${{ fromJSON(needs.gate.outputs.plan != '' && toJSON(fromJSON(needs.gate.outputs.plan).jobs.verify_paperclip_runner.matrix) || '{"include":[{"lane_label":"static checks","command":"check:static","task":"ci:runner-static"},{"lane_label":"rust","command":"check:runner","release_crate":true,"task":"ci:runner-rust"},{"lane_label":"vitest package","command":"test:typescript:vitest --shard=1/2","task":"ci:runner-vitest"},{"lane_label":"vitest native server","command":"test:typescript:vitest --shard=2/2","release_crate":true,"task":"ci:runner-vitest"}]}') }} env: PAPERCLIP_CI_PROFILE: ${{ needs.gate.outputs.profile_ci }} @@ -831,6 +831,23 @@ jobs: if: ${{ steps.runner_release_crate.outputs.cache-hit == 'true' }} run: node scripts/runner-release-crate-cache.mjs mark-current + - name: Compute the Runner debug crate cache key + id: runner_debug_crate_key + if: hashFiles('scripts/runner-debug-crate-cache.mjs') != '' + run: node scripts/runner-debug-crate-cache.mjs plan + + - name: Restore the Runner debug crate (read only) + id: runner_debug_crate + if: hashFiles('scripts/runner-debug-crate-cache.mjs') != '' + uses: actions/cache/restore@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0 + with: + path: ${{ steps.runner_debug_crate_key.outputs.paths }} + key: ${{ steps.runner_debug_crate_key.outputs.key }} + + - name: Mark the restored Runner debug crate current + if: ${{ steps.runner_debug_crate.outputs.cache-hit == 'true' }} + run: node scripts/runner-debug-crate-cache.mjs mark-current + - name: Verify Runner Rust through Turbo if: matrix.command == 'check:runner' && hashFiles('turbo.json') != '' uses: ./.github/actions/run-ci-task diff --git a/.github/workflows/release-verify.yml b/.github/workflows/release-verify.yml index 7daaf98d13b..76128fd1a89 100644 --- a/.github/workflows/release-verify.yml +++ b/.github/workflows/release-verify.yml @@ -433,12 +433,38 @@ jobs: pnpm --filter @paperclipai/paperclip-runner "$check" done + - name: Compute the Runner debug crate cache key + id: runner_debug_crate_key + if: ${{ matrix.lane == 'rust' && github.repository == 'paperclipai/paperclip' && github.event_name == 'push' && github.ref == 'refs/heads/master' && inputs.ref == github.sha }} + run: node scripts/runner-debug-crate-cache.mjs plan + + - name: Check for this push's Runner debug crate + id: runner_debug_crate + if: ${{ matrix.lane == 'rust' && github.repository == 'paperclipai/paperclip' && github.event_name == 'push' && github.ref == 'refs/heads/master' && inputs.ref == github.sha }} + uses: actions/cache/restore@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0 + with: + lookup-only: true + path: ${{ steps.runner_debug_crate_key.outputs.paths }} + key: ${{ steps.runner_debug_crate_key.outputs.key }} + + - name: Clean debug test variants before publishing workspace binaries + if: ${{ matrix.lane == 'rust' && github.repository == 'paperclipai/paperclip' && github.event_name == 'push' && github.ref == 'refs/heads/master' && inputs.ref == github.sha && steps.runner_debug_crate.outputs.cache-hit != 'true' }} + working-directory: packages/paperclip-runner + run: cargo clean --profile dev --manifest-path runner/Cargo.toml -p paperclip-runner-core + - name: Warm debug dependencies for the shared Runner cache # Protocol tests need debug binaries. Populate their dependencies in # the sole cache writer, so a cold save also serves the protocol lane. if: ${{ matrix.lane == 'rust' && github.repository == 'paperclipai/paperclip' && github.event_name == 'push' && github.ref == 'refs/heads/master' && inputs.ref == github.sha }} run: pnpm --filter @paperclipai/paperclip-runner build:rust + - name: Save the Runner debug crate + if: ${{ matrix.lane == 'rust' && github.repository == 'paperclipai/paperclip' && github.event_name == 'push' && github.ref == 'refs/heads/master' && inputs.ref == github.sha && steps.runner_debug_crate.outputs.cache-hit != 'true' }} + uses: actions/cache/save@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0 + with: + path: ${{ steps.runner_debug_crate_key.outputs.paths }} + key: ${{ steps.runner_debug_crate_key.outputs.key }} + # The dependency cache above deliberately leaves the Runner's own crate # out, so every pull request build compiled paperclip-runner-core in # release mode (28-49s) although most pushes leave its inputs alone. diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index cb6a7d2a57e..3cf16c21fab 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -353,11 +353,11 @@ When changing the workflow path or branch, authorize the new reference before updating the caller. Retain older authorized SHA references while queued runs or supported reruns still use them. -General server shards run at most two test files at a time in separate Vitest +General server shards run at most three test files at a time in separate Vitest forks. Each file retains its own home, config and process state, and its test cases run serially. An explicit `PAPERCLIP_*_TEST_DATABASE_URL`, including `PAPERCLIP_TEST_DATABASE_URL`, keeps those shards serial because the database -is shared. Native Runner suites also use two isolated file workers. Their long +is shared. Native Runner suites use four isolated file workers. Their long lease, preparation, disk-loss and checkpoint tests have separate files so real observation windows can overlap. Issue recovery suites split by producer and recovery behavior for the same reason. Unsharded local tests and route/authz @@ -369,6 +369,22 @@ config; an explicit shared test database retains one partition per job. The Runner transport lease-expiry and cold-admission regressions live in their own file so those real-clock waits overlap other files in the existing pool. +The two PR Runner Vitest jobs separate package tests from native server tests. +The first runs the complete package inventory; the second runs the complete +native server group. The existing `--shard=1/2` and `--shard=2/2` commands route +through `run-pr-vitest-lane.mjs` for compatibility with the master-pinned +workflow. Other workflow names and shard counts keep their existing coverage. +Both jobs still prepare debug binaries. A separate exact-key debug cache holds +the workspace library and every Cargo binary target. Its key includes source, +protocol and fixture content, Cargo configuration, compiler flags, toolchain +and checkout path. Only the Rust lane of a verified master push saves it. +PR jobs only restore it; a missing or incomplete entry rebuilds normally. + +Affected-test selection treats Vite's exact `?raw` imports as file inputs. +Changing that file selects its consumers. Generated/untracked files and unknown +query loaders retain their consumers conservatively. Other filesystem, process +and network effects keep their existing fallback. + For a full wall-clock profile of all main PR CI jobs, add `profile-ci` to the PR. See [CI profiling](ci-profiling.md) for the Honeycomb waterfall, coverage, operator setup, and activation requirements. @@ -4178,6 +4194,12 @@ Focused coverage lives in `task-workspace-binding.test.ts`, `task-workspace-repo Channel defaults can be configured in Apps connection settings and per destination. They affect new conversations only. A queued selection or repository request is applied on the next normal admission; configuration changes do not interrupt a live run. See [the architecture](plans/2026-10-09-task-workspaces.md). +GitHub repository discovery and admission recheck current grant access. Installation +changes advance the access revision and invalidate an in-flight refresh. Ordinary +pull-request webhooks update delivery health without invalidating repository +access. A refresh preserves their latest webhook timestamp and health when it +commits. Concurrent access refreshes retain the existing one-retry limit. + ### Slack provisioning paths Self-hosted development continues to use independent Slack app setup. Managed diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index b5b2af1a789..325148106d2 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1683,6 +1683,19 @@ remains protected. Cleanup never interprets this registration as a stop receipt or permission to replay provider work. Repeated cleanup visits register the source once; settled sources add no continuing discovery or new maintenance timer. +Retry and authenticated new messages also inspect the task's bound remote +workspace for its exact failed writer. A newer setup failure cannot hide that +writer's file obligation. A released reusable source can register preservation +when its prepared descriptor, current workspace authority, and dedicated stop +receipts agree. Registration keeps the lease's policy, release state, and stop +receipts. The existing worker verifies allocation ownership before file-only +resume. It saves the files before ordinary continuation admission; repeated +requests do not create another recovery generation or replay provider work. +If Retry names a newer failed preparation, continuation separately verifies that +it never reached provider dispatch. Contradictory execution evidence keeps the +hold. After preservation, the accepted continuation selects the bound saved +workspace through its verified repair receipt instead of discarding its files. + Every continuation carries the triggering request, ordered user direction, interaction outcomes, completed work, and explicit history coverage. A delivered message remains part of the task's request after its connection or approval resolves. The original title is background; a completed Notion read does not satisfy a later Gmail request. Author and source-trust boundaries survive rendering into both native and legacy prompts. Missing required history must be fetched before dispatch rather than described as complete. ### Remote runner process evidence diff --git a/package.json b/package.json index 9e2bb3714a6..d4e1c1afe35 100644 --- a/package.json +++ b/package.json @@ -61,7 +61,7 @@ "smoke:notion-generic-live": "node scripts/smoke/notion-generic-live.mjs", "smoke:posthog-live": "node scripts/smoke/posthog-live.mjs", "smoke:terminal-bench-loop-skill": "node scripts/smoke/terminal-bench-loop-skill-smoke.mjs", - "test:release-registry": "node scripts/ci-profile/node-test.mjs scripts/ci-profile/reporters.test.mjs scripts/__tests__/canary-startup-playwright.test.mjs scripts/__tests__/cursor-public-install-sandbox.test.mjs scripts/__tests__/grok-public-install-sandbox.test.mjs scripts/__tests__/public-install-closure.test.mjs scripts/__tests__/public-install-registry.test.mjs scripts/__tests__/prefetch-child.test.mjs scripts/__tests__/stage-bundled-packages.test.mjs scripts/__tests__/pack-staged-packages.test.mjs scripts/__tests__/retain-runner-qualification-packages.test.mjs scripts/__tests__/runner-release-crate-cache.test.mjs scripts/__tests__/vitest-worker-isolation.test.mjs scripts/verify-release-registry-state.test.mjs scripts/release-package-map.test.mjs scripts/check-release-package-bootstrap.test.mjs scripts/check-no-git-push.test.mjs scripts/release-lib.test.mjs scripts/release-registry-versions.test.mjs scripts/link-plugin-dev-sdk.test.js scripts/acpx-patch-packaging.test.mjs scripts/service-onboard-smoke.test.mjs scripts/docker-onboard-smoke.test.mjs scripts/preview-artifacts.test.mjs scripts/cloud-migrator-artifacts.test.mjs", + "test:release-registry": "node scripts/ci-profile/node-test.mjs scripts/ci-profile/reporters.test.mjs scripts/__tests__/canary-startup-playwright.test.mjs scripts/__tests__/cursor-public-install-sandbox.test.mjs scripts/__tests__/grok-public-install-sandbox.test.mjs scripts/__tests__/public-install-closure.test.mjs scripts/__tests__/public-install-registry.test.mjs scripts/__tests__/prefetch-child.test.mjs scripts/__tests__/stage-bundled-packages.test.mjs scripts/__tests__/pack-staged-packages.test.mjs scripts/__tests__/retain-runner-qualification-packages.test.mjs scripts/__tests__/runner-release-crate-cache.test.mjs scripts/__tests__/runner-debug-crate-cache.test.mjs scripts/__tests__/vitest-worker-isolation.test.mjs scripts/verify-release-registry-state.test.mjs scripts/release-package-map.test.mjs scripts/check-release-package-bootstrap.test.mjs scripts/check-no-git-push.test.mjs scripts/release-lib.test.mjs scripts/release-registry-versions.test.mjs scripts/link-plugin-dev-sdk.test.js scripts/acpx-patch-packaging.test.mjs scripts/service-onboard-smoke.test.mjs scripts/docker-onboard-smoke.test.mjs scripts/preview-artifacts.test.mjs scripts/cloud-migrator-artifacts.test.mjs", "storybook-visual:baseline": "node scripts/storybook-visual-baseline.mjs", "test:storybook-visual": "node scripts/storybook-visual-baseline.mjs download && node scripts/storybook-visual-baseline.mjs verify && pnpm build-storybook && npx playwright test --config tests/storybook-visual/playwright.config.ts", "test:storybook-visual:update": "node scripts/storybook-visual-baseline.mjs download && pnpm build-storybook && npx playwright test --config tests/storybook-visual/playwright.config.ts --update-snapshots && node scripts/storybook-visual-baseline.mjs pack", diff --git a/packages/paperclip-runner/scripts/run-pr-vitest-lane.mjs b/packages/paperclip-runner/scripts/run-pr-vitest-lane.mjs index 51e980aee91..40e5bd2f89b 100644 --- a/packages/paperclip-runner/scripts/run-pr-vitest-lane.mjs +++ b/packages/paperclip-runner/scripts/run-pr-vitest-lane.mjs @@ -1,26 +1,9 @@ -// Runs this package's vitest shard for a Verify Paperclip Runner lane, then — -// on the final shard of the PR workflow only — the server package's -// general-server-native-runner group. -// -// That server suite rebuilds the Runner release binaries with cargo in -// beforeAll. The PR workflow's plain General tests server shards carry no -// Rust cache, so hosting it there cold-compiled every third-party crate on -// each run (277s of a 291s shard vitest step, actions run 35246999382, -// 2026-09-17) and made that shard the slowest check of the whole run. The -// Verify Paperclip Runner lanes already restore the shared release-runner-v1 -// Rust cache read-only, which turns that build into an incremental rebuild, -// and the workflow files themselves list this lane's command as a package -// script — so the suite moves here without a workflow-file change. -// -// Contract, mirrored in scripts/run-vitest-stable.mjs (prWorkflowName) and -// pinned by scripts/__tests__/run-vitest-stable-shard.test.mjs: -// - Only the PR workflow (pr.yml, whose GITHUB_WORKFLOW the reusable -// pr-trusted.yml jobs inherit) excludes these suites from the existing -// without-chat group, and only there does this wrapper run them. Other -// callers keep that group complete. Release verification instead selects -// an explicit partition and its own required cached native lane. -// - The suite runs on the lane whose --shard=N/M has N === M (or an unsharded -// invocation), so exactly one PR lane carries it. +// The two PR Vitest slots cover separate workloads: the first runs every +// package test and the second runs the native server integration group. +// Both need debug fixtures; only the native group also builds release units. +// Keep the existing --shard commands compatible with master-pinned workflows. +// Local/other callers retain ordinary package sharding. Other PR shard counts +// keep package shards plus native tests on the final lane, preserving coverage. import { spawnSync } from "node:child_process"; import { dirname, resolve } from "node:path"; import { fileURLToPath } from "node:url"; @@ -33,18 +16,30 @@ const dryRun = process.argv.includes("--dry-run"); const shardArg = args.find((value) => value.startsWith("--shard=")); const shardMatch = shardArg ? /^--shard=(\d+)\/(\d+)$/.exec(shardArg) : null; -if (shardArg && !shardMatch) { +if (args.filter(value => value.startsWith("--shard=")).length > 1 || (shardArg && !shardMatch)) { console.error(`[pr-vitest-lane] unrecognized shard argument: ${shardArg}`); process.exit(1); } +if (shardMatch && (Number(shardMatch[1]) < 1 || Number(shardMatch[1]) > Number(shardMatch[2]))) { + throw new Error(`Invalid shard: ${shardArg}`); +} const isFinalShard = !shardArg || shardMatch[1] === shardMatch[2]; const isPrWorkflow = process.env.GITHUB_WORKFLOW === "PR"; const runNativeRunnerGroup = isPrWorkflow && isFinalShard; +// Reuse the two existing CI slots: all package files in the first, all native +// server files in the second. Keeping the old command arguments makes this +// safe while the trusted workflow still comes from master. Other shard counts +// and local callers retain ordinary Vitest sharding and the final-lane append. +const separateWorkloads = isPrWorkflow && shardMatch?.[2] === "2"; +const runPackageTests = !separateWorkloads || !isFinalShard; +const packageArgs = separateWorkloads ? args.filter((arg) => arg !== shardArg) : args; const plannedCommands = [ { command: "pnpm", args: ["run", "ensure:eval-build-deps"], cwd: packageRoot }, { command: "pnpm", args: ["run", "build:rust"], cwd: packageRoot }, - { command: "pnpm", args: ["exec", "vitest", "run", ...args], cwd: packageRoot }, + ...(runPackageTests + ? [{ command: "pnpm", args: ["exec", "vitest", "run", ...packageArgs], cwd: packageRoot }] + : []), ...(runNativeRunnerGroup ? [{ command: "pnpm", @@ -55,7 +50,7 @@ const plannedCommands = [ ]; if (dryRun) { - console.log(JSON.stringify({ isPrWorkflow, isFinalShard, runNativeRunnerGroup, plannedCommands }, null, 2)); + console.log(JSON.stringify({ isPrWorkflow, isFinalShard, runNativeRunnerGroup, runPackageTests, plannedCommands }, null, 2)); process.exit(0); } diff --git a/scripts/__tests__/run-vitest-stable-shard.test.mjs b/scripts/__tests__/run-vitest-stable-shard.test.mjs index 99a2ff239a2..bce13ef5285 100644 --- a/scripts/__tests__/run-vitest-stable-shard.test.mjs +++ b/scripts/__tests__/run-vitest-stable-shard.test.mjs @@ -469,25 +469,24 @@ function wrapperPlan(args, envOverrides = {}) { return JSON.parse(result.stdout); } -test("the PR vitest lane wrapper runs the native-runner group exactly on the final PR shard", () => { - for (const [args, envOverrides, expected] of [ - [["--shard=1/2"], { GITHUB_WORKFLOW: "PR" }, false], - [["--shard=2/2"], { GITHUB_WORKFLOW: "PR" }, true], - [[], { GITHUB_WORKFLOW: "PR" }, true], - [["--shard=2/2"], { GITHUB_WORKFLOW: "Release" }, false], - [["--shard=2/2"], {}, false], - [[], {}, false], +test("the PR lanes separate complete package and native workloads without changing local shards", () => { + for (const [args, envOverrides, packageArgs, native] of [ + [["--shard=1/2"], { GITHUB_WORKFLOW: "PR" }, [], false], + [["--shard=2/2"], { GITHUB_WORKFLOW: "PR" }, null, true], + [[], { GITHUB_WORKFLOW: "PR" }, [], true], + [["--shard=2/3"], { GITHUB_WORKFLOW: "PR" }, ["--shard=2/3"], false], + [["--shard=3/3"], { GITHUB_WORKFLOW: "PR" }, ["--shard=3/3"], true], + [["--shard=2/2"], { GITHUB_WORKFLOW: "Release" }, ["--shard=2/2"], false], + [["--shard=2/2"], {}, ["--shard=2/2"], false], + [[], {}, [], false], ]) { const plan = wrapperPlan(args, envOverrides); - assert.equal(plan.runNativeRunnerGroup, expected, - `args ${JSON.stringify(args)} env ${JSON.stringify(envOverrides)}`); - const commands = plan.plannedCommands.map((planned) => planned.args.join(" ")); - assert.ok(commands.some((command) => command.startsWith(`exec vitest run${args.length ? ` ${args.join(" ")}` : ""}`)), - "the wrapper must pass shard flags through to vitest"); - assert.equal( - commands.some((command) => command.includes("--group general-server-native-runner")), - expected, - ); + assert.equal(plan.runNativeRunnerGroup, native); + assert.equal(plan.runPackageTests, packageArgs !== null); + const vitest = plan.plannedCommands.find(command => command.args[0] === "exec"); + assert.deepEqual(vitest?.args ?? null, packageArgs === null ? null : ["exec", "vitest", "run", ...packageArgs]); + assert.equal(plan.plannedCommands.some(command => command.args.includes("general-server-native-runner")), native); + assert.ok(plan.plannedCommands.some(command => command.args.includes("build:rust")), "both workloads require debug fixtures"); } const malformed = spawnSync(process.execPath, [laneWrapper, "--shard=nonsense", "--dry-run"], { @@ -728,7 +727,10 @@ test("a failed chat child fails the job after its sibling finishes", { skip: pro } }); -test("native files overlap only when their database fixtures are isolated", { skip: process.platform === "win32" }, () => { +for (const [group, parallelWorkers, shardArgs] of [ + ["general-server-native-runner", 4, []], + ["general-server-without-chat-or-native-runner", 3, ["--shard-index", "0", "--shard-count", "12"]], +]) test(`${group} files overlap only when their database fixtures are isolated`, { skip: process.platform === "win32" }, () => { const directory = mkdtempSync(path.join(os.tmpdir(), "native-file-workers-")); const receipt = path.join(directory, "args.json"); const fakePnpm = path.join(directory, "pnpm"); @@ -736,22 +738,25 @@ test("native files overlap only when their database fixtures are isolated", { sk chmodSync(fakePnpm, 0o755); try { for (const [databaseEnv, workers] of [ - [{}, 2], + [{}, parallelWorkers], [{ PAPERCLIP_TEST_DATABASE_URL: "postgresql://shared/test" }, 1], [{ PAPERCLIP_ISSUE_RECOVERY_TEST_DATABASE_URL: "postgresql://shared/recovery_test" }, 1], - [{ PAPERCLIP_TEST_POSTGRES_ADMIN_URL: "postgresql://owned/postgres" }, 2], + [{ PAPERCLIP_TEST_POSTGRES_ADMIN_URL: "postgresql://owned/postgres" }, parallelWorkers], ]) { const env = workflowEnv({ PATH: `${directory}${path.delimiter}${process.env.PATH}`, PAPERCLIP_CI_PROFILE: "false" }); for (const key of Object.keys(env)) if (key.endsWith("_TEST_DATABASE_URL")) delete env[key]; Object.assign(env, databaseEnv); - const result = spawnSync(process.execPath, [script, "--mode", "general", "--group", "general-server-native-runner"], { + const result = spawnSync(process.execPath, [script, "--mode", "general", "--group", group, ...shardArgs], { cwd: repoRoot, encoding: "utf8", env, }); assert.equal(result.status, 0, result.stdout + result.stderr); const args = JSON.parse(readFileSync(receipt, "utf8")); assert.ok(args.includes(`--maxWorkers=${workers}`)); - assert.ok(args.includes(workers === 2 ? "--fileParallelism" : "--no-file-parallelism")); - for (const file of [restartRecoverySuitePath, ...restartWaitSuites]) assert.ok(args.includes(file)); + assert.ok(args.includes(workers > 1 ? "--fileParallelism" : "--no-file-parallelism")); + if (workers > 1) assert.ok(args.includes("--pool=forks") && args.includes("--isolate")); + if (group === "general-server-native-runner") { + for (const file of [restartRecoverySuitePath, ...restartWaitSuites]) assert.ok(args.includes(file)); + } } } finally { rmSync(directory, { recursive: true, force: true }); } }); diff --git a/scripts/__tests__/runner-debug-crate-cache.test.mjs b/scripts/__tests__/runner-debug-crate-cache.test.mjs new file mode 100644 index 00000000000..08df149227d --- /dev/null +++ b/scripts/__tests__/runner-debug-crate-cache.test.mjs @@ -0,0 +1,86 @@ +import assert from "node:assert/strict"; +import { execFileSync } from "node:child_process"; +import { mkdirSync, mkdtempSync, readFileSync, rmSync, statSync, writeFileSync } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import test from "node:test"; +import { debugContract, debugKey } from "../runner-debug-crate-cache.mjs"; +import { markCurrent } from "../runner-release-crate-cache.mjs"; + +const bins = [...readFileSync("packages/paperclip-runner/runner/crates/runner-core/Cargo.toml", "utf8") + .matchAll(/\[\[bin\]\]\nname = "([^"]+)"/g)].map(match => match[1]); +const metadata = { workspace_members: ["core"], packages: [{ id: "core", name: "paperclip-runner-core", + targets: bins.map(name => ({ name, kind: ["bin"] })) }] }; + +test("every debug binary must be complete before the restored units are marked fresh", t => { + const root = mkdtempSync(path.join(os.tmpdir(), "runner-debug-cache-")); + t.after(() => rmSync(root, { recursive: true, force: true })); + const contract = debugContract(root, metadata); + assert.equal(contract.bins.length, bins.length); + assert.ok(contract.bins.length >= 6); + for (const item of contract.required) { + const filename = path.join(root, item.glob.replaceAll("*", "0123456789abcdef")); + mkdirSync(path.dirname(filename), { recursive: true }); + writeFileSync(filename, "cached artifact"); + } + const dependency = path.join(root, "packages/paperclip-runner/runner/target/debug/deps/libserde-123.rlib"); + writeFileSync(dependency, "dependency"); + const previous = statSync(dependency).mtimeMs; + assert.equal(markCurrent({ root, ...contract }).action, "marked"); + assert.equal(statSync(dependency).mtimeMs, previous); + const last = contract.required.at(-1); + rmSync(path.join(root, last.glob.replaceAll("*", "0123456789abcdef"))); + const discarded = markCurrent({ root, ...contract }); + assert.equal(discarded.action, "discarded"); + assert.match(discarded.reason, new RegExp(last.label)); + assert.equal(readFileSync(dependency, "utf8"), "dependency"); + assert.throws(() => debugContract(root, { ...metadata, workspace_members: [] }), /Review/); +}); + +test("debug cache keys invalidate source, included inputs, deletions and compiler flags", t => { + const root = mkdtempSync(path.join(os.tmpdir(), "runner-debug-key-")); + t.after(() => rmSync(root, { recursive: true, force: true })); + const source = path.join(root, "packages/paperclip-runner/runner/crates/runner-core/src/lib.rs"); + mkdirSync(path.dirname(source), { recursive: true }); + writeFileSync(source, "pub fn run() {}\n"); + execFileSync("git", ["init", "-q"], { cwd: root }); + execFileSync("git", ["add", "."], { cwd: root }); + const env = { CARGO_HOME: path.join(root, "cargo-home") }; + const first = debugKey(root, env); + assert.equal(debugKey(root, env), first); + for (const name of ["RUST_LOG", "RUST_BACKTRACE", "RUSTUP_HOME"]) { + assert.equal(debugKey(root, { ...env, [name]: "different-runner-value" }), first, name); + } + assert.notEqual(debugKey(root, { ...env, RUSTFLAGS: "-C target-cpu=native" }), first); + assert.notEqual(debugKey(root, { ...env, RUSTC_WORKSPACE_WRAPPER: "/compiler-wrapper" }), first); + writeFileSync(source, "pub fn run() { panic!(); }\n"); + assert.notEqual(debugKey(root, env), first); + rmSync(source); + assert.notEqual(debugKey(root, env), first); + writeFileSync(source, "pub fn run() {}\n"); + const protocol = path.join(root, "packages/paperclip-runner/protocol/input.json"); + mkdirSync(path.dirname(protocol), { recursive: true }); + writeFileSync(protocol, "{}"); + execFileSync("git", ["add", "."], { cwd: root }); + assert.notEqual(debugKey(root, env), first); +}); + +test("PRs only restore exact debug keys and only verified master pushes publish them", () => { + const pr = readFileSync(".github/workflows/pr-trusted.yml", "utf8"); + const release = readFileSync(".github/workflows/release-verify.yml", "utf8"); + const step = (source, name) => source.split(` - name: ${name}\n`)[1]?.split(" - name:")[0]; + const reader = step(pr, "Restore the Runner debug crate (read only)"); + assert.match(reader, /actions\/cache\/restore@/); + assert.match(reader, /hashFiles\('scripts\/runner-debug-crate-cache.mjs'\) != ''/); + assert.doesNotMatch(reader, /restore-keys/); + for (const field of ["key", "paths"]) assert.ok(reader.includes(`steps.runner_debug_crate_key.outputs.${field}`)); + assert.equal(step(pr, "Save the Runner debug crate"), undefined); + for (const name of ["Compute the Runner debug crate cache key", "Check for this push's Runner debug crate", + "Clean debug test variants before publishing workspace binaries", "Save the Runner debug crate"]) { + const body = step(release, name); + assert.match(body, /matrix.lane == 'rust' && github.repository == 'paperclipai\/paperclip' && github.event_name == 'push' && github.ref == 'refs\/heads\/master' && inputs.ref == github.sha/); + } + assert.match(step(release, "Save the Runner debug crate"), /runner_debug_crate.outputs.cache-hit != 'true'/); + assert.match(step(release, "Warm debug dependencies for the shared Runner cache"), /build:rust/); + assert.ok(release.indexOf("Warm debug dependencies") < release.indexOf("Save the Runner debug crate")); +}); diff --git a/scripts/__tests__/runner-release-crate-cache.test.mjs b/scripts/__tests__/runner-release-crate-cache.test.mjs index ac98c949fdf..e595bcd28a9 100644 --- a/scripts/__tests__/runner-release-crate-cache.test.mjs +++ b/scripts/__tests__/runner-release-crate-cache.test.mjs @@ -361,8 +361,8 @@ test("the Verify Paperclip Runner job restores the crate only in the lanes that assert.deepEqual(lanes, [ ["static checks", false], ["rust", true], - ["vitest 1/2", false], - ["vitest 2/2", true], + ["vitest package", false], + ["vitest native server", true], ]); for (const stepName of [KEY_STEP, RESTORE_STEP]) { assert.match(step(body, stepName), /if: \$\{\{ matrix\.release_crate == true \}\}/, stepName); @@ -414,7 +414,7 @@ test("only the master Rust lane writes the crate, after a lookup, with the reade // The dependency cache's own writer gate is the same expression. assert.match(body, new RegExp(`save-if: \\$\\{\\{ ${WRITER_GATE.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")} \\}\\}`)); - assert.equal((releaseWorkflow.match(/actions\/cache\/save@/g) ?? []).length, 1); + assert.equal((releaseWorkflow.match(/actions\/cache\/save@/g) ?? []).length, 2, "release and debug workspace caches have separate guarded writers"); }); test("the scope names the platform, the compiler and the checkout root, and survives a missing rustc", () => { diff --git a/scripts/ci/test-selection.integration.test.mjs b/scripts/ci/test-selection.integration.test.mjs index a959dc54e8c..38818dd5b34 100644 --- a/scripts/ci/test-selection.integration.test.mjs +++ b/scripts/ci/test-selection.integration.test.mjs @@ -36,6 +36,8 @@ test("real graph retains direct and indirect filesystem, process and opaque depe write("ui/src/builtin-loader.test.ts", 'process.getBuiltinModule("fs").readFileSync("input");'); write("ui/src/opaque.test.ts", 'import {parse} from "yaml"; console.log(parse("value: true"));'); write("ui/src/independent.test.ts", 'export const independent = true;'); + write("ui/src/raw.test.ts", 'import text from "./leaf.ts?raw"; console.log(text);'); + write("ui/src/unknown-query.test.ts", 'import text from "./leaf.ts?custom-loader"; console.log(text);'); write("ui/src/mocked.test.ts", 'import {vi} from "vitest"; vi.mock("./helper",()=>({read:()=>"mock"})); import {read} from "./helper"; read("input");'); write("ui/src/partial.test.ts", 'import {vi} from "vitest"; vi.mock("./helper",async (original)=>await original()); import {read} from "./helper"; read("input");'); write("ui/src/automock.test.ts", 'import {vi} from "vitest"; vi.mock("./helper"); import {read} from "./helper"; read("input");'); @@ -50,7 +52,11 @@ test("real graph retains direct and indirect filesystem, process and opaque depe execFileSync("git", ["add", "."], { cwd: directory }); const select = (changed) => selectProjectTests({ root: directory, project: "fixture", changed }); const result = await select(["ui/src/leaf.ts"]); - assert.deepEqual(result.related, ["ui/src/import.test.ts"]); + assert.ok(result.related.includes("ui/src/import.test.ts")); + assert.ok(result.inputRelated.includes("ui/src/raw.test.ts"), "raw text has an explicit repository input edge"); + assert.ok(!result.alwaysRun.includes("ui/src/raw.test.ts")); + assert.ok(!(await select(["ui/src/independent.test.ts"])).selected.includes("ui/src/raw.test.ts")); + assert.ok(result.alwaysRun.includes("ui/src/unknown-query.test.ts"), "unknown query loaders remain conservative"); assert.ok(result.inputRelated.includes("ui/src/reader.test.ts"), "literal file reads add the missing source dependency"); assert.ok(!result.alwaysRun.includes("ui/src/reader.test.ts")); for (const file of ["indirect", "process", "dynamic", "cjs", "builtin-loader", "opaque", "partial", "automock", "bypass", "spy", "remock", "arguments"]) { @@ -71,9 +77,11 @@ test("real graph retains direct and indirect filesystem, process and opaque depe assert.ok(broken.selected.includes("ui/src/broken.test.ts"), "an unresolved dependency must retain its test"); write("ui/src/generated.ts", "export const generated = 42;"); write("ui/src/generated.test.ts", 'import {generated} from "./generated"; console.log(generated);'); + write("ui/src/generated-raw.test.ts", 'import text from "./generated.ts?raw"; console.log(text);'); execFileSync("git", ["add", "ui/src/generated.test.ts"], { cwd: directory }); assert.ok((await select(["ui/src/leaf.ts"])).alwaysRun.includes("ui/src/generated.test.ts"), "generated inputs have no trustworthy source-to-output graph and retain their consumers"); + assert.ok((await select(["ui/src/leaf.ts"])).alwaysRun.includes("ui/src/generated-raw.test.ts")); write("ui/src/environment.ts", 'import {readFileSync} from "node:fs"; readFileSync(new URL("./leaf.ts", import.meta.url));'); const fileSetup = await select(["ui/src/leaf.ts"]); assert.deepEqual(fileSetup.selected, fileSetup.inventory, "an explicit setup file read affects the entire project"); diff --git a/scripts/ci/vitest-selection-graph.mjs b/scripts/ci/vitest-selection-graph.mjs index eb071bee156..78fcfb16665 100644 --- a/scripts/ci/vitest-selection-graph.mjs +++ b/scripts/ci/vitest-selection-graph.mjs @@ -252,6 +252,17 @@ export async function selectProjectTests({ root = process.cwd(), project, change && !coveredBuiltins.some(source => source.replace(/^node:/, "") === dependencyId.replace(/^node:/, ""))) reasons.push(`builtin ${dependencyId}`); continue; } + // Vite's exact ?raw form reads a file as text, not executable code. + // Record that input explicitly; do not treat arbitrary query loaders, + // generated files, or paths outside the checkout as independent. + if (dependencyId.endsWith("?raw") && dependencyId.indexOf("?") === dependencyId.length - 4) { + const input = dependencyPath(dependencyId.slice(0, -4)); + if (input && inside(root, input) && statSync(input).isFile() + && tracked.has(path.relative(root, input).split(path.sep).join("/"))) { + fileInputs.set(id, [...(fileInputs.get(id) ?? []), { path: input, directory: false }]); + continue; + } + } const dependency = dependencyPath(dependencyId); if (dependency && inside(root, dependency)) { local.push(dependency); diff --git a/scripts/run-vitest-stable.mjs b/scripts/run-vitest-stable.mjs index bb84c33ecf5..579037b06ae 100644 --- a/scripts/run-vitest-stable.mjs +++ b/scripts/run-vitest-stable.mjs @@ -513,9 +513,9 @@ async function runGeneralGroup(routeTests, groupName, shardIndex = null, shardCo await runVitest( ["--project", "@paperclipai/server", // Each file owns its database, home, and processes. Keep cases serial, - // but overlap long real lease windows in two isolated fork workers. + // but overlap long real lease windows in four isolated fork workers. ...(hasSharedTestDatabase ? serializedServerVitestArgs - : ["--pool=forks", "--isolate", "--fileParallelism", "--maxWorkers=2"]), + : ["--pool=forks", "--isolate", "--fileParallelism", "--maxWorkers=4"]), ...nativeRunnerSuites], "native runner vertical-slice suite", ); @@ -543,11 +543,11 @@ async function runGeneralGroup(routeTests, groupName, shardIndex = null, shardCo "--project", "@paperclipai/server", // Files have separate fork processes and isolated homes/configs. - // Keep each file's cases serial; overlap only two independent files. + // Keep each file's cases serial; overlap three independent files. // An explicit shared test database cannot be isolated this way. ...(hasSharedTestDatabase ? serializedServerVitestArgs - : ["--pool=forks", "--isolate", "--fileParallelism", "--maxWorkers=2"]), + : ["--pool=forks", "--isolate", "--fileParallelism", "--maxWorkers=3"]), ...shardFiles, ], `${groupName} shard ${shardIndex + 1}/${shardCount}`, @@ -724,7 +724,7 @@ try { const serverArgs = project !== "@paperclipai/server" ? [] : options.mode === serializedModeName || hasSharedTestDatabase ? ["--pool=forks", "--isolate", ...serializedServerVitestArgs] - : ["--pool=forks", "--isolate", "--fileParallelism", "--maxWorkers=2"]; + : ["--pool=forks", "--isolate", "--fileParallelism", "--maxWorkers=3"]; await runVitest(["--project", project, ...serverArgs, ...files], `selected ${project}: ${files.length} files`, null, true); } diff --git a/scripts/runner-debug-crate-cache.mjs b/scripts/runner-debug-crate-cache.mjs new file mode 100644 index 00000000000..dff1e16be0b --- /dev/null +++ b/scripts/runner-debug-crate-cache.mjs @@ -0,0 +1,79 @@ +import { execFileSync } from "node:child_process"; +import { createHash } from "node:crypto"; +import { appendFileSync, existsSync, globSync, readFileSync, statSync } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import { computeScope, HASHED_INPUTS, markCurrent, readRustcVersion, RUNNER_PACKAGE } from "./runner-release-crate-cache.mjs"; + +const directory = `${RUNNER_PACKAGE}/runner/target/debug`; +export const DEBUG_INPUTS = [...HASHED_INPUTS, `${RUNNER_PACKAGE}/package.json`, ".cargo/**", `${RUNNER_PACKAGE}/.cargo/**`, + `${RUNNER_PACKAGE}/runner/.cargo/**`, "scripts/runner-debug-crate-cache.mjs", + "scripts/runner-release-crate-cache.mjs"]; + +// Derive all bin units from Cargo, including automatically discovered targets. +// Refuse an expanded workspace until its library/cache contract is reviewed. +function readMetadata(root) { + return JSON.parse(execFileSync("cargo", ["metadata", "--offline", "--no-deps", + "--format-version=1", "--manifest-path", "runner/Cargo.toml"], { + cwd: path.join(root, RUNNER_PACKAGE), encoding: "utf8", + })); +} + +export function debugContract(root, metadata = readMetadata(root)) { + const packages = metadata.packages.filter(pkg => metadata.workspace_members.includes(pkg.id)); + if (packages.length !== 1 || packages[0].name !== "paperclip-runner-core") { + throw new Error("Review the debug artifact contract for the changed Runner workspace"); + } + const bins = packages[0].targets.filter(target => target.kind.includes("bin")).map(target => target.name).sort(); + if (!bins.includes("paperclip-runnerd")) throw new Error("Missing Runner binary target"); + const fingerprint = `${directory}/.fingerprint/paperclip-runner-core-*`; + const cachedPaths = [fingerprint, `${directory}/deps/libpaperclip_runner_core-*`, + `${directory}/deps/paperclip_runner_core-*.d`, ...bins.flatMap(bin => [ + `${directory}/deps/${bin.replaceAll("-", "_")}-*`, `${directory}/${bin}`, `${directory}/${bin}.d`, + ])]; + const required = [ + { label: "lib fingerprint", glob: `${fingerprint}/lib-paperclip_runner_core` }, + { label: "lib dep-info", glob: `${fingerprint}/dep-lib-paperclip_runner_core` }, + { label: "lib rlib", glob: `${directory}/deps/libpaperclip_runner_core-*.rlib` }, + { label: "lib rmeta", glob: `${directory}/deps/libpaperclip_runner_core-*.rmeta` }, + ...bins.flatMap(bin => [ + { label: `${bin} fingerprint`, glob: `${fingerprint}/bin-${bin}` }, + { label: `${bin} dep-info`, glob: `${fingerprint}/dep-bin-${bin}` }, + { label: `${bin} executable`, glob: `${directory}/deps/${bin.replaceAll("-", "_")}-*`, exclude: /\.(d|dSYM)$/ }, + ]), + ]; + return { bins, cachedPaths, required }; +} + +export function debugKey(root, env = process.env) { + const hash = createHash("sha256"); + const files = [...globSync(DEBUG_INPUTS, { cwd: root })].filter(file => statSync(path.join(root, file)).isFile()).sort(); + // Include path names, generated inputs, and local Cargo configuration. A + // changed compiler flag or target must never reuse an mtime-adjusted unit. + for (const file of [...files, path.join(env.CARGO_HOME || path.join(os.homedir(), ".cargo"), "config"), + path.join(env.CARGO_HOME || path.join(os.homedir(), ".cargo"), "config.toml")]) { + hash.update(file).update("\0").update(existsSync(path.resolve(root, file)) + ? readFileSync(path.resolve(root, file)) : "missing").update("\0"); + } + const buildEnv = Object.entries(env).filter(([name]) => + /^(?:RUSTFLAGS|RUSTDOCFLAGS|RUSTC|RUSTC_WRAPPER|RUSTC_WORKSPACE_WRAPPER|RUSTC_BOOTSTRAP|RUSTUP_TOOLCHAIN|RUST_TARGET_PATH|CC|CXX|CFLAGS|CXXFLAGS|LDFLAGS)$/.test(name) + || /^CARGO_(BUILD|PROFILE|TARGET|ENCODED)/.test(name)).sort(); + hash.update(JSON.stringify(buildEnv)); + return `runner-debug-crate-v1-${computeScope({ workspace: root, + rustcVersion: readRustcVersion(env, path.join(root, RUNNER_PACKAGE)) })}-${hash.digest("hex")}`; +} + +function main() { + const root = process.env.GITHUB_WORKSPACE || process.cwd(); + const contract = debugContract(root); + if (process.argv[2] === "mark-current") { + console.log(JSON.stringify(markCurrent({ root, ...contract, schema: "paperclip.runner-debug-crate.v1" }))); + } else if (process.argv[2] === "plan") { + const key = debugKey(root); + if (process.env.GITHUB_OUTPUT) appendFileSync(process.env.GITHUB_OUTPUT, + `key=${key}\npaths<"); +} +if (process.argv[1] && path.resolve(process.argv[1]) === fileURLToPath(import.meta.url)) main(); diff --git a/scripts/runner-release-crate-cache.mjs b/scripts/runner-release-crate-cache.mjs index 941bcf80824..3bc01a2b6ae 100644 --- a/scripts/runner-release-crate-cache.mjs +++ b/scripts/runner-release-crate-cache.mjs @@ -103,10 +103,10 @@ export function computeScope({ workspace = process.env.GITHUB_WORKSPACE ?? proce return `${keySafe(os.platform())}-${keySafe(os.arch())}-${keySafe(rustcVersion)}-${root}`; } -function readRustcVersion(env = process.env) { +export function readRustcVersion(env = process.env, cwd = process.cwd()) { let output; try { - output = execFileSync("rustc", ["--version", "--verbose"], { encoding: "utf8", env }); + output = execFileSync("rustc", ["--version", "--verbose"], { encoding: "utf8", env, cwd }); } catch { return "no-rustc"; } @@ -156,12 +156,13 @@ function walk(target, visit) { * older than the build. When the set is incomplete, remove whatever was * restored and report it: cargo then compiles the crate as on a miss. */ -export function markCurrent({ root = process.env.GITHUB_WORKSPACE ?? process.cwd(), now = new Date() } = {}) { - const missing = REQUIRED.find(({ glob, exclude }) => expandGlob(root, glob).filter((entry) => !exclude?.test(entry)).length === 0); - const matched = CACHED_PATHS.flatMap((pattern) => expandGlob(root, pattern)); +export function markCurrent({ root = process.env.GITHUB_WORKSPACE ?? process.cwd(), now = new Date(), + required = REQUIRED, cachedPaths = CACHED_PATHS, schema = SCHEMA } = {}) { + const missing = required.find(({ glob, exclude }) => expandGlob(root, glob).filter((entry) => !exclude?.test(entry)).length === 0); + const matched = cachedPaths.flatMap((pattern) => expandGlob(root, pattern)); if (missing) { for (const target of matched) rmSync(target, { recursive: true, force: true }); - return { schema: SCHEMA, action: "discarded", reason: `incomplete: ${missing.label} is missing`, removed: matched.length }; + return { schema, action: "discarded", reason: `incomplete: ${missing.label} is missing`, removed: matched.length }; } let files = 0; let bytes = 0; @@ -181,7 +182,7 @@ export function markCurrent({ root = process.env.GITHUB_WORKSPACE ?? process.cwd } }); } - return { schema: SCHEMA, action: "marked", files, bytes }; + return { schema, action: "marked", files, bytes }; } function main(argv) { diff --git a/server/src/__tests__/attention-service.test.ts b/server/src/__tests__/attention-service.test.ts index ca18a4a2d3c..00cf31058eb 100644 --- a/server/src/__tests__/attention-service.test.ts +++ b/server/src/__tests__/attention-service.test.ts @@ -66,9 +66,12 @@ describeEmbeddedPostgres("attention service", () => { }, 30_000); afterEach(async () => { - - - await db.execute(sql`TRUNCATE TABLE companies CASCADE`); + await db.transaction(async (tx) => { + // CASCADE emits a notice for every related table; keep cleanup quiet while + // retaining warnings/errors and the same complete database reset. + await tx.execute(sql`SET LOCAL client_min_messages = warning`); + await tx.execute(sql`TRUNCATE TABLE companies CASCADE`); + }); }); afterAll(async () => { @@ -1208,6 +1211,9 @@ describeEmbeddedPostgres("attention service", () => { contextSnapshot: { wakeReason: "timer" }, createdAt: laterAt, updatedAt: laterAt, finishedAt: laterAt, }); + // Bulk-loaded fixtures need planner statistics before exercising production queries. + await db.execute(sql`ANALYZE heartbeat_runs, heartbeat_run_events`); + const events: Array> = [], now = Date.now(); const capture = createFullDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event)); installDatabaseResponseCapture(db, { heartbeatRuns }, capture); @@ -1251,6 +1257,9 @@ describeEmbeddedPostgres("attention service", () => { contextSnapshot: { issueId }, createdAt: laterAt, updatedAt: laterAt, finishedAt: laterAt, }); + // Do not race autovacuum to collect statistics for the 16,400 new rows. + await db.execute(sql`ANALYZE heartbeat_runs, heartbeat_run_events`); + const feed = await attentionService(db).list(companyId, { includeDismissed: true, all: true, allowUnscopedAll: true, }); diff --git a/server/src/__tests__/heartbeat-run-dispatch-work.test.ts b/server/src/__tests__/heartbeat-run-dispatch-work.test.ts index 1c0e5005a04..30a1509e0f8 100644 --- a/server/src/__tests__/heartbeat-run-dispatch-work.test.ts +++ b/server/src/__tests__/heartbeat-run-dispatch-work.test.ts @@ -72,7 +72,10 @@ describe.skipIf(!support.supported)("run dispatch production wiring (controlled } async function quiet() { await heartbeat.drainActiveRunExecutions(); - await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBeNull(), { timeout: 5_000 }); + await vi.waitFor(() => { + expect(coordinator!.nextWakeAt()).toBeNull(); + expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 5_000 }); const count = reads.mock.calls.length; vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); try { diff --git a/server/src/__tests__/tool-access-service.test.ts b/server/src/__tests__/tool-access-service.test.ts index aa5afc03b1a..e728d320924 100644 --- a/server/src/__tests__/tool-access-service.test.ts +++ b/server/src/__tests__/tool-access-service.test.ts @@ -101,6 +101,7 @@ import { toolAccessRoutes } from "../routes/tool-access.js"; import { errorHandler } from "../middleware/index.js"; import * as sentry from "../sentry.js"; import { HttpError } from "../errors.js"; +import { githubConnectionEventService } from "../services/github-connection-events.js"; import type { VercelConnectClient } from "../services/vercel-connect.js"; import { invalidatePaperclipCloudConnectorCapabilities, PaperclipCloudConnectorError, type PaperclipCloudConnector } from "../services/paperclip-cloud-connector.js"; @@ -8150,7 +8151,7 @@ describeEmbeddedPostgres("tool access service", () => { ).toBe(false); }); - it.each(["none", "event", "same-time-refresh", "one-conflict"])( + it.each(["none", "event", "same-time-refresh", "one-conflict", "pr-webhook", "conflict-then-pr-webhook", "installation-webhook"])( "binds a managed GitHub identity and protects refresh from concurrent access changes (%s)", async (concurrentChange) => { const company = await createCompany(db); @@ -8170,6 +8171,7 @@ describeEmbeddedPostgres("tool access service", () => { platform_shared: true, }; let beforeRepositoryResponse = async () => {}; + let installationSuspended = false; vi.spyOn(globalThis, "fetch").mockImplementation(async (url) => { const href = String(url); if (href === "https://api.github.com/user") { @@ -8187,6 +8189,7 @@ describeEmbeddedPostgres("tool access service", () => { repository_selection: "selected", html_url: "https://github.com/settings/installations/101", account: { login: "paperclipai" }, + ...(installationSuspended ? { suspended_at: new Date().toISOString() } : {}), }, ], }); @@ -8341,6 +8344,55 @@ describeEmbeddedPostgres("tool access service", () => { ), ).resolves.toHaveLength(1); vi.mocked(connector.setWebhookBinding).mockClear(); + if (["pr-webhook", "conflict-then-pr-webhook", "installation-webhook"].includes(concurrentChange)) { + let repositoryResponses = 0; + let webhookAt: Date; + beforeRepositoryResponse = async () => { + repositoryResponses += 1; + if (concurrentChange === "conflict-then-pr-webhook" && repositoryResponses === 1) { + await db.update(connectionGrants).set({ + providerTenant: sql`jsonb_set(${connectionGrants.providerTenant}, '{github,accessRevision}', ${JSON.stringify(randomUUID())}::jsonb)`, + }).where(eq(connectionGrants.id, grant!.id)); + return; + } + webhookAt = new Date(Date.now() + repositoryResponses); + vi.mocked(connector.leaseEvents).mockResolvedValueOnce({ + leaseId: randomUUID(), + events: [{ + id: randomUUID(), provider: "github", + event: concurrentChange === "installation-webhook" ? "installation" : "pull_request", + action: concurrentChange === "installation-webhook" ? "suspend" : "edited", + installationId: "101", repositoryId: "1", createdAt: webhookAt.toISOString(), + bindingIds: [`${grant!.id}_101`], + payload: { repository: "paperclipai/repo-1", number: 1, state: "open", merged: false, + updatedAt: webhookAt.toISOString(), url: "https://github.com/paperclipai/repo-1/pull/1" }, + }], + }); + await expect(githubConnectionEventService(db, { connector, now: () => webhookAt }) + .pollOnce({ forDelivery: true })).resolves.toMatchObject({ processed: 1, failed: 0 }); + if (concurrentChange === "installation-webhook") installationSuspended = true; + }; + if (concurrentChange === "installation-webhook") { + await expect(service.checkHealth(connected.connectionId, actor)) + .rejects.toThrow("GitHub installation is suspended"); + expect(repositoryResponses).toBe(1); + const [latest] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, grant!.id)); + expect(latest?.providerTenant?.github).toMatchObject({ installationCount: 0, repositoryCount: 0 }); + expect(latest?.providerTenant?.github?.repositories).toBeUndefined(); + expect(connector.setWebhookBinding).not.toHaveBeenCalled(); + return; + } + await expect(service.checkHealth(connected.connectionId, actor)) + .resolves.toMatchObject({ connection: { healthStatus: "ok" } }); + expect(repositoryResponses).toBe(concurrentChange === "pr-webhook" ? 1 : 2); + const [latest] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, grant!.id)); + expect(latest?.providerTenant?.github).toMatchObject({ + repositoryCount: 3, lastWebhookAt: webhookAt!.toISOString(), webhookHealth: "healthy", + }); + expect(latest?.providerTenant?.github?.repositories).toHaveLength(3); + expect(connector.setWebhookBinding).toHaveBeenCalled(); + return; + } if (concurrentChange !== "none") { beforeRepositoryResponse = async () => { if (concurrentChange === "one-conflict") diff --git a/server/src/services/cancelled-native-startup.ts b/server/src/services/cancelled-native-startup.ts index 810c4d5939d..8467918f2ec 100644 --- a/server/src/services/cancelled-native-startup.ts +++ b/server/src/services/cancelled-native-startup.ts @@ -1,6 +1,6 @@ import { isPreDispatchReviewWait } from "./pre-dispatch-review-wait.js"; import { and, eq, inArray, isNotNull, or } from "drizzle-orm"; -import { environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; import { claimedAdapterType } from "./conversation-continuation.js"; import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED } from "./native-local-process-stop.js"; import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; @@ -15,6 +15,25 @@ type Run = Pick; type Coordinator = typeof nativeRunFinalizations.$inferSelect; +export async function isFailedNativePreparation(db: Db, run: Run, coordinator: Coordinator | undefined) { + const recovery = run.resultJson?.executionRecovery as Record | undefined; + if (run.status !== "failed" || !run.finishedAt || run.runtimeMode !== "legacy" || run.runtimeModeResolvedAt + || run.errorCode !== "setup_failed" || run.executionStage !== "preparing" + || claimedAdapterType(run) !== "paperclip_runner" || coordinator + || recovery?.kind !== "bootstrap" || recovery.providerWorkStarted !== false + || run.processPid || run.processGroupId || run.processStartedAt || run.sessionIdAfter + || run.nativeIssueId || run.nativeSessionId || run.runnerProfileJson?.nativeExecutionInput != null + || run.runnerProfileJson?.nativeWorkspaceSync != null || run.runnerProfileJson?.remoteWorkspace != null + || run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > new Date()) return false; + const [lease] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and( + eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), + )).limit(1); + const [result] = await db.select({ id: nativeRunResults.id }).from(nativeRunResults).where(and( + eq(nativeRunResults.companyId, run.companyId), eq(nativeRunResults.runId, run.id), + )).limit(1); + return !lease && !result && !await nativeStartupHasExecutionEvidence(db, run); +} + /** Candidate for explicit user Retry only, never automatic replay after Stop. */ export async function canRetryStoppedRun(db: Db, run: Run): Promise { if (run.errorCode === "native_session_cleanup_quarantined") return false; @@ -66,6 +85,10 @@ export async function isCancelledNativeStartup(db: Db, run: Run, coordinator: Co : !hasRemoteTerminationReceipt(lease))) return false; // Reject contradictory retained evidence, including a crash after a launch // request but before the PID callback. Provider events never certify a stop. + return !await nativeStartupHasExecutionEvidence(db, run); +} + +async function nativeStartupHasExecutionEvidence(db: Db, run: Run) { const [execution] = await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and( eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), or(isNotNull(heartbeatRunEvents.sourceEventId), @@ -73,5 +96,5 @@ export async function isCancelledNativeStartup(db: Db, run: Run, coordinator: Co "harness.ready", "session.started", "session.resumed", "session.updated", "turn.started", "provider.event", "provider.rpc_result", "tool.execution.started"])), )).limit(1); - return !execution; + return Boolean(execution); } diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index c93ec6b8681..b45214afd1c 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -6,12 +6,12 @@ import { readTaskRetryIntent } from "./task-retry.js"; import { readSystemRecovery } from "./system-recovery-state.js"; import { hasRequiredWorkspaceRecovery } from "./workspace-restore-recovery-state.js"; import { hasVerifiedWorkspaceRepair } from "./workspace-recovery-receipt.js"; -import { preserveNativeFailedWorkspace } from "./native-runtime/native-failed-workspace-recovery.js"; +import { preserveNativeFailedWorkspace, preserveTaskRemoteWorkspaceWriter } from "./native-runtime/native-failed-workspace-recovery.js"; import { createHash } from "node:crypto"; import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; import { canContinueCancelledRun } from "./run-cancellation.js"; import { readQueuedInteractionResponse } from "./queued-interaction-response.js"; -import { isCancelledNativeStartup } from "./cancelled-native-startup.js"; +import { isCancelledNativeStartup, isFailedNativePreparation } from "./cancelled-native-startup.js"; import { hasNativeLocalProcessStop, hasHistoricalSuspendedNativeSession, recordedLocalExecutionStopped } from "./native-local-process-stop.js"; import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js"; import { hasRemoteTerminationReceipt, remoteLeaseCleanupScope } from "./remote-execution-termination.js"; @@ -214,6 +214,7 @@ export async function admitExplicitNativeContinuation(input: { return blocked("send_intent_missing", "Send the saved message again to continue."); const authorizedAt = nativeQueuedSend ? interruptRequestedAt! : response?.comment.createdAt ?? comment?.createdAt ?? retryWake?.requestedAt ?? new Date(); + if (!input.dryRun) await preserveTaskRemoteWorkspaceWriter(db, companyId, issueId); const blocker = await getExecutionBlocker(db, companyId, issueId); const [boundary] = await db.select({ createdAt: issueComments.createdAt }).from(issues).innerJoin(issueComments, and( eq(issueComments.companyId, issues.companyId), eq(issueComments.issueId, issues.id), @@ -280,6 +281,16 @@ export async function admitExplicitNativeContinuation(input: { !run.finishedAt) return blocked("source_unavailable", "The previous execution has not finished or its owner changed. Your message is saved."); if (!queuedInterrupt && !queuedRequest && authorizedAt <= run.finishedAt) return blocked("message_predates_stop", "This message arrived before the previous run stopped. Send a new message to continue."); if (adapterExecutionControls.has(run.id)) return blocked("execution_settling", "Waiting for the previous run to stop. Your message will start automatically."); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, run.id), + )).for("update"); + const [lockedRun] = await db.select(continuationRunFields).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, run.id), + )).for("update"); + if (!lockedRun || lockedRun.status !== run.status || lockedRun.agentId !== run.agentId || + lockedRun.finishedAt?.getTime() !== run.finishedAt.getTime()) return null; + run = lockedRun; + const failedPreparation = await isFailedNativePreparation(db, run, coordinator); const unusedAdmission = run.status === "cancelled" && !run.startedAt && run.errorCode === "execution_reconciliation_required" && !run.processPid && !run.processGroupId && !run.nativeSessionId; @@ -295,7 +306,7 @@ export async function admitExplicitNativeContinuation(input: { const freshIntent = await canSelectFreshWorkspace({ db, companyId, issueId, agentId, actorId, sourceRunId: run.id, evidence: strategyEvidence, taskRetryRequestId: input.taskRetryRequestId, queuedRequestId: input.queuedCommentRequestId ?? savedChatRequestId }); - const legacyUserTurn = run.runtimeMode === "legacy" && + const legacyUserTurn = failedPreparation || run.runtimeMode === "legacy" && action.cause === "legacy_execution_requires_reconciliation" && (isConversationAdapter(agent.adapterType) || (agent.adapterType === "paperclip_runner" && (freshIntent || await hasVerifiedWorkspaceRepair(db, run)))); @@ -303,27 +314,13 @@ export async function admitExplicitNativeContinuation(input: { // operator Stop or redeliver a message already consumed by this run. if (queuedRequest && !queuedInterrupt && (run.contextSnapshot?.wakeCommentId === commentId || (Array.isArray(run.contextSnapshot?.wakeCommentIds) && run.contextSnapshot.wakeCommentIds.includes(commentId)))) return null; - if (legacyUserTurn) { + if (legacyUserTurn && !failedPreparation) { const historicalAdapter = await historicalAdapterType(db, run); // A settings change never converts a known process/webhook execution into // a conversation. Those adapters retain their reconciliation contract. if (historicalAdapter && !isConversationAdapter(historicalAdapter)) return null; if (!historicalAdapter && agent.adapterType === "paperclip_runner") return null; } - // For pre-upgrade rows without adapter evidence, only a new explicit user - // turn is allowed, after the termination proofs below. This does not infer - // an old adapter type, certify old outcomes, or authorize automatic replay. - const [coordinator] = await db.select().from(nativeRunFinalizations).where(and( - eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, run.id), - )).for("update"); - // Same lock order as the native claim. Re-read the run while holding both - // locks before accepting the never-claimed startup proof. - const [lockedRun] = await db.select(continuationRunFields).from(heartbeatRuns).where(and( - eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, run.id), - )).for("update"); - if (!lockedRun || lockedRun.status !== run.status || lockedRun.agentId !== run.agentId || - lockedRun.finishedAt?.getTime() !== run.finishedAt.getTime()) return null; - run = lockedRun; const interruptedByThisClick = response?.source.requiresFreshSession === true && interruptIntent?.targetRunId === run.id && run.resultJson?.operatorInterrupted === true && run.resultJson?.interruptionSource === "issue_comment_interrupt" && run.resultJson?.interruptedIssueId === issueId && @@ -351,7 +348,7 @@ export async function admitExplicitNativeContinuation(input: { return blocked("workspace_repair_required", "Paperclip is recovering saved work before continuing. Your message is saved."); } const cancelledStartup = await isCancelledNativeStartup(db, run, coordinator); - if (partiallyDeliveredQueue && run.runtimeMode !== "native" && !unusedAdmission && !cancelledStartup) return null; + if (partiallyDeliveredQueue && run.runtimeMode !== "native" && !unusedAdmission && !cancelledStartup && !failedPreparation) return null; if ((queuedInterrupt || queuedRequest) && !legacyUserTurn && !unusedAdmission && !(queuedRequest && run.runtimeMode === "native" && (run.status !== "cancelled" || authorizedAt > run.finishedAt!)) && @@ -406,7 +403,7 @@ export async function admitExplicitNativeContinuation(input: { }))) return null; } else { if (leases.some(lease => !lease.releasedAt || lease.cleanupStatus === "failed")) return blocked("local_cleanup", "Waiting for the previous environment to finish cleanup. Your message will start automatically."); - if (!unusedAdmission && !cancelledStartup) { + if (!unusedAdmission && !cancelledStartup && !failedPreparation) { if (freshIntent && (run.processPid || run.processGroupId) && !await recordedLocalExecutionStopped(db, run)) return blocked("process_identity_missing", "Paperclip is verifying that the previous session stopped."); // A missing process identity is not evidence that a provider exited. @@ -435,6 +432,17 @@ export async function admitExplicitNativeContinuation(input: { } sources.push(run); } + if (taskRetry && !sources.some(run => run.id === taskRetry.sourceRunId)) { + const [coordinator] = await db.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, taskRetry.sourceRunId), + )).for("update"); + const [source] = await db.select(continuationRunFields).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, taskRetry.sourceRunId), eq(heartbeatRuns.agentId, agentId), + )).for("update"); + if (!source || (source.nativeIssueId ?? source.issueId ?? source.contextSnapshot?.issueId) !== issueId + || adapterExecutionControls.has(source.id) || !await isFailedNativePreparation(db, source, coordinator)) return null; + sources.push(source); + } const nativeSources = sources.filter(run => run.runtimeMode === "native"); const executedSources = sources.filter(run => run.runtimeMode === "native" || run.errorCode !== "execution_reconciliation_required" || run.startedAt); if (!executedSources.length || (retry && !sources.some(run => run.id === input.failedRunId)) || diff --git a/server/src/services/fresh-workspace-continuation.ts b/server/src/services/fresh-workspace-continuation.ts index 25ac8556975..6789a12b928 100644 --- a/server/src/services/fresh-workspace-continuation.ts +++ b/server/src/services/fresh-workspace-continuation.ts @@ -9,6 +9,7 @@ import { adapterExecutionControls } from "./adapter-execution-control.js"; import { hasPendingRunCleanup } from "./execution-blocker.js"; import { toExecutionWorkspace } from "./execution-workspaces.js"; import { isBoundIsolatedTaskDirectory } from "./isolated-task-directory.js"; +import { readRemoteWorkspaceAuthority, remoteWorkspaceRunReferenceSchema } from "@paperclipai/shared"; export type ContinuationReadiness = | { mode: "resume_preserved_workspace"; preservationReceiptId: string } @@ -167,6 +168,33 @@ export async function readFreshWorkspaceReadiness(db: Db, run: Pick>`jsonb_build_object('issueId', ${heartbeatRuns.contextSnapshot}->'issueId')`, + runnerProfileJson: sql>`jsonb_build_object('remoteWorkspace', ${heartbeatRuns.runnerProfileJson}->'remoteWorkspace', + 'nativeWorkspaceSync', ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync')`, + }).from(issueRecoveryActions) + .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.companyId, issueRecoveryActions.companyId), + sql`${heartbeatRuns.id}::text = ${issueRecoveryActions.evidence}->>'runId'`)) + .where(and(eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, issueId), + eq(heartbeatRuns.agentId, run.agentId), + sql`${issueRecoveryActions.evidence}#>>'{explicitUserContinuation,runId}' = ${run.id}`, + sql`${heartbeatRuns.runnerProfileJson}#>>'{remoteWorkspace,workspaceId}' = ${bound.id}`, + sql`${heartbeatRuns.runnerProfileJson}#>>'{remoteWorkspace,snapshotId}' = ${authority.snapshotId}`)).limit(1); + for (const writer of recovered) { + const remote = remoteWorkspaceRunReferenceSchema.safeParse(writer.runnerProfileJson?.remoteWorkspace); + if (!remote.success || remote.data.state !== "saved" || remote.data.workspaceId !== bound.id + || remote.data.epoch !== authority.epoch || remote.data.snapshotId !== authority.snapshotId) continue; + const receipt = await verifiedWorkspaceRepairReceipt(db, writer); + if (receipt) return { mode: "resume_preserved_workspace", preservationReceiptId: receipt }; + } + } const workspace = run.contextSnapshot?.previousRunId === source.id ? await reusableDirectoryAfterPreparationFailure(db, source, issueId) : null; if (workspace?.cwd) { diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index d6902a99374..d11eee87ec9 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -62,6 +62,7 @@ import { } from "../native-runtime/native-review-participant.js"; import { claimQueuedNativeReviewRun } from "../native-runtime/native-review-dispatch.js"; import { adapterExecutionControls } from "../adapter-execution-control.js"; +import { preserveTaskRemoteWorkspaceWriter } from "../native-runtime/native-failed-workspace-recovery.js"; import { executionFailureRetryCount } from "../execution-recovery-attempt.js"; import { slackMentionAllowsRunStart, @@ -3486,6 +3487,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende )).limit(1) : []; const verifiedConversationReset = Boolean(resetComment && isConversationReset(resetComment.body)); + if (wakeCommentId && opts.requestedByActorType === "user") + await preserveTaskRemoteWorkspaceWriter(tx as unknown as Db, issue.companyId, issue.id); const executionBlocker = await getExecutionBlocker( tx as unknown as Db, issue.companyId, issue.id, { conversationResetCommentId: verifiedConversationReset ? wakeCommentId : null }, diff --git a/server/src/services/legacy-workspace-restore-recovery.ts b/server/src/services/legacy-workspace-restore-recovery.ts index 47ce54f6f84..45e451b0cc3 100644 --- a/server/src/services/legacy-workspace-restore-recovery.ts +++ b/server/src/services/legacy-workspace-restore-recovery.ts @@ -3,6 +3,7 @@ import { and, eq, inArray, isNotNull, ne, or, sql } from "drizzle-orm"; import { environmentLeases, environments, heartbeatRuns, type Db } from "@paperclipai/db"; import { hasWorkspaceRestoreFailure, WORKSPACE_RESTORE_FAILURE_CODES } from "@paperclipai/shared"; import { hasConfirmedSandboxStopAndRetain, prepareSandboxStopAndRetain } from "./sandbox-stop-and-retain.js"; +import { preparedRemoteWorkspaceSource } from "./remote-workspace-lifecycle.js"; import { hasRequiredWorkspaceRecovery, LEGACY_WORKSPACE_RECOVERY_SCHEMA } from "./workspace-restore-recovery-state.js"; @@ -77,6 +78,11 @@ export async function preserveLegacyWorkspaceRestoreSources(db: Db, run: Run, so if (recovery?.schema === LEGACY_WORKSPACE_RECOVERY_SCHEMA && recovery.runId === run.id) { retained.push(lease.id); continue; } + if (source && run.runtimeMode === "native" && hasConfirmedSandboxStopAndRetain(lease) + && ["reuse_by_environment", "reuse_by_execution_workspace"].includes(lease.leasePolicy) + && await preparedRemoteWorkspaceSource(db, lease)) { + retained.push(lease.id); continue; + } // Cancellation can finish the dedicated stop before copy-back reports its // failure. Only the host's original source and two matching receipts can // turn that released row into a file-repair hold. diff --git a/server/src/services/native-runtime/native-failed-workspace-recovery.ts b/server/src/services/native-runtime/native-failed-workspace-recovery.ts index a3e0216da61..8683cac4dc4 100644 --- a/server/src/services/native-runtime/native-failed-workspace-recovery.ts +++ b/server/src/services/native-runtime/native-failed-workspace-recovery.ts @@ -1,10 +1,10 @@ -import { remoteWorkspaceRunReferenceSchema } from "@paperclipai/shared"; +import { readRemoteWorkspaceAuthority, remoteWorkspaceRunReferenceSchema } from "@paperclipai/shared"; import type { CancelledWorkspacePreservationScope } from "../cancelled-workspace-preservation-notifications.js"; import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; import { notifyExecutionRunWork } from "../execution-control-notifications.js"; import { and, asc, eq, gt, inArray, isNull, or, sql } from "drizzle-orm"; -import { environmentLeases, heartbeatRuns, workspaceRestoreFailureCodesSql, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { environmentLeases, executionWorkspaces, heartbeatRuns, workspaceRestoreFailureCodesSql, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; import { recordRequiredWorkspaceRestoreFailure } from "../legacy-execution-recovery.js"; import { hasRequiredWorkspaceRecovery, nativeRunHasNoWorkspaceAccess } from "../workspace-restore-recovery-state.js"; import { logActivity } from "../activity-log.js"; @@ -22,6 +22,32 @@ function workspaceRecoveryReference(profile: Record | null | un return readNativeWorkspaceSyncReference(profile?.nativeWorkspaceSync) ?? (remote.success ? remote.data : null); } +export async function preserveTaskRemoteWorkspaceWriter(db: Db, companyId: string, issueId: string) { + const [bound] = await db.select({ id: executionWorkspaces.id, metadata: executionWorkspaces.metadata }).from(issues) + .innerJoin(executionWorkspaces, and(eq(executionWorkspaces.id, issues.executionWorkspaceId), + eq(executionWorkspaces.companyId, issues.companyId))) + .where(and(eq(issues.companyId, companyId), eq(issues.id, issueId))); + if (!bound) return; + const authority = readRemoteWorkspaceAuthority(bound.metadata); + if (!authority?.dirty || !authority.writerRunId) return; + const [run] = await db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, + nativeIssueId: heartbeatRuns.nativeIssueId, issueId: heartbeatRuns.issueId, + contextSnapshot: sql>`jsonb_build_object('issueId', ${heartbeatRuns.contextSnapshot}->'issueId')`, + runnerProfileJson: sql>`jsonb_build_object('remoteWorkspace', ${heartbeatRuns.runnerProfileJson}->'remoteWorkspace', + 'nativeWorkspaceSync', ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync')`, + }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, authority.writerRunId), + eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.status, "failed"), + sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.issueId}::text, + ${heartbeatRuns.contextSnapshot}->>'issueId') = ${issueId}`, + )); + const reference = remoteWorkspaceRunReferenceSchema.safeParse(run?.runnerProfileJson.remoteWorkspace); + if (!run || run.runnerProfileJson.nativeWorkspaceSync != null || !reference.success || reference.data.state !== "prepared" + || reference.data.workspaceId !== bound.id || reference.data.epoch !== authority.epoch) return; + try { return await preserveNativeFailedWorkspace(db, run); } + catch (error) { if (!(error instanceof NativeWorkspaceFinalizationBusyError)) throw error; } +} + export async function preserveNativeFailedWorkspace(db: Db, selected: Pick, options: { allowPreterminal?: boolean; requireCancelledStop?: boolean; restartRecovery?: NativeRestartRecoveryClaim } = {}) { diff --git a/server/src/services/native-runtime/native-workspace-finalization-state.ts b/server/src/services/native-runtime/native-workspace-finalization-state.ts index 3701c1fe0b6..00b783fe21c 100644 --- a/server/src/services/native-runtime/native-workspace-finalization-state.ts +++ b/server/src/services/native-runtime/native-workspace-finalization-state.ts @@ -1,49 +1,10 @@ import { environmentLeases, executionWorkspaces, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; import { and, eq, exists, sql } from "drizzle-orm"; -import { readRemoteWorkspaceAuthority, remoteWorkspaceRunReferenceSchema } from "@paperclipai/shared"; +import { remoteWorkspaceRunReferenceSchema } from "@paperclipai/shared"; import { hasRequiredWorkspaceRecovery, nativeRunHasNoWorkspaceAccess } from "../workspace-restore-recovery-state.js"; import { recordRequiredWorkspaceRestoreFailure } from "../legacy-execution-recovery.js"; import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js"; - -type WorkspaceSourceLease = Pick; - -/** Preparation publishes file authority before native execution is selected. - * Neither runtimeMode nor a later execution-input projection owns these files. */ -async function preparedRemoteWorkspaceSource(db: Db, lease: WorkspaceSourceLease) { - if (!lease.heartbeatRunId || !lease.provider || lease.provider === "local" || !lease.providerLeaseId) return null; - const [bound] = await db.select({ lease: environmentLeases, run: { - id: heartbeatRuns.id, runtimeMode: heartbeatRuns.runtimeMode, - controllerBootId: heartbeatRuns.controllerBootId, - issueId: sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.issueId}::text, - ${heartbeatRuns.contextSnapshot}->>'issueId')`, - reference: sql`${heartbeatRuns.runnerProfileJson}->'remoteWorkspace'`, - legacyReference: sql`${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'`, - } }).from(environmentLeases) - .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, environmentLeases.heartbeatRunId), - eq(heartbeatRuns.companyId, environmentLeases.companyId))) - .where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, lease.companyId), - eq(environmentLeases.heartbeatRunId, lease.heartbeatRunId), eq(environmentLeases.provider, lease.provider), - eq(environmentLeases.providerLeaseId, lease.providerLeaseId), - sql`${heartbeatRuns.runnerProfileJson}->'remoteWorkspace'->>'state' = 'prepared'`)).limit(1); - if (!bound) return null; - const parsed = remoteWorkspaceRunReferenceSchema.safeParse(bound.run.reference); - if (!parsed.success || parsed.data.state !== "prepared" || bound.run.legacyReference != null) return null; - const reference = parsed.data; - const issueId = bound.run.issueId; - if (typeof issueId !== "string" || bound.lease.issueId !== issueId || reference.leaseId !== bound.lease.id - || reference.providerLeaseId !== bound.lease.providerLeaseId || reference.remoteCwd !== bound.lease.metadata?.remoteCwd - || bound.lease.executionWorkspaceId && bound.lease.executionWorkspaceId !== reference.workspaceId) return null; - const [workspace] = await db.select().from(executionWorkspaces).where(and( - eq(executionWorkspaces.companyId, lease.companyId), eq(executionWorkspaces.id, reference.workspaceId))).limit(1); - const authority = readRemoteWorkspaceAuthority(workspace?.metadata); - if (!workspace || workspace.sourceIssueId !== issueId && workspace.mode !== "shared_workspace" - || !authority || !authority.dirty || authority.writerRunId !== bound.run.id || authority.epoch !== reference.epoch - || authority.snapshotId !== reference.snapshotId || !authority.location - || authority.location.environmentId !== bound.lease.environmentId || authority.location.leaseId !== bound.lease.id - || authority.location.providerLeaseId !== bound.lease.providerLeaseId || authority.location.remoteCwd !== reference.remoteCwd) return null; - return { ...bound, workspace: workspace!, reference, issueId }; -} +import { preparedRemoteWorkspaceSource, type WorkspaceSourceLease } from "../remote-workspace-lifecycle.js"; /** File-only recovery for setup that never reached native admission. The live * preparer must still own its controller; orphan repair accepts terminal runs diff --git a/server/src/services/remote-workspace-lifecycle.ts b/server/src/services/remote-workspace-lifecycle.ts index 7896d5abada..aa4d4097086 100644 --- a/server/src/services/remote-workspace-lifecycle.ts +++ b/server/src/services/remote-workspace-lifecycle.ts @@ -1,6 +1,6 @@ -import { and, eq } from "drizzle-orm"; -import { executionWorkspaces, executionWorkspaceSnapshots, type Db } from "@paperclipai/db"; -import { readRemoteWorkspaceAuthority, type EnvironmentLease } from "@paperclipai/shared"; +import { and, eq, sql } from "drizzle-orm"; +import { environmentLeases, heartbeatRuns, executionWorkspaces, executionWorkspaceSnapshots, type Db } from "@paperclipai/db"; +import { readRemoteWorkspaceAuthority, remoteWorkspaceRunReferenceSchema, type EnvironmentLease } from "@paperclipai/shared"; import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js"; type Workspace = Pick; @@ -9,6 +9,44 @@ export type DestroyRemoteWorkspaceLease = (input: { companyId: string; workspaceId: string; leaseId: string; environmentId: string; providerLeaseId: string; }) => Promise; +export type WorkspaceSourceLease = Pick; + +export async function preparedRemoteWorkspaceSource(db: Db, lease: WorkspaceSourceLease) { + if (!lease.heartbeatRunId || !lease.provider || lease.provider === "local" || !lease.providerLeaseId) return null; + const [bound] = await db.select({ lease: environmentLeases, run: { + id: heartbeatRuns.id, runtimeMode: heartbeatRuns.runtimeMode, + controllerBootId: heartbeatRuns.controllerBootId, + issueId: sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.issueId}::text, + ${heartbeatRuns.contextSnapshot}->>'issueId')`, + reference: sql`${heartbeatRuns.runnerProfileJson}->'remoteWorkspace'`, + legacyReference: sql`${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'`, + } }).from(environmentLeases) + .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, environmentLeases.heartbeatRunId), + eq(heartbeatRuns.companyId, environmentLeases.companyId))) + .where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, lease.companyId), + eq(environmentLeases.heartbeatRunId, lease.heartbeatRunId), eq(environmentLeases.provider, lease.provider), + eq(environmentLeases.providerLeaseId, lease.providerLeaseId), + sql`${heartbeatRuns.runnerProfileJson}->'remoteWorkspace'->>'state' = 'prepared'`)).limit(1); + if (!bound) return null; + const parsed = remoteWorkspaceRunReferenceSchema.safeParse(bound.run.reference); + if (!parsed.success || parsed.data.state !== "prepared" || bound.run.legacyReference != null) return null; + const reference = parsed.data; + const issueId = bound.run.issueId; + if (typeof issueId !== "string" || bound.lease.issueId !== issueId || reference.leaseId !== bound.lease.id + || reference.providerLeaseId !== bound.lease.providerLeaseId || reference.remoteCwd !== bound.lease.metadata?.remoteCwd + || bound.lease.executionWorkspaceId && bound.lease.executionWorkspaceId !== reference.workspaceId) return null; + const [workspace] = await db.select().from(executionWorkspaces).where(and( + eq(executionWorkspaces.companyId, lease.companyId), eq(executionWorkspaces.id, reference.workspaceId))).limit(1); + const authority = readRemoteWorkspaceAuthority(workspace?.metadata); + if (!workspace || workspace.sourceIssueId !== issueId && workspace.mode !== "shared_workspace" + || !authority || !authority.dirty || authority.writerRunId !== bound.run.id || authority.epoch !== reference.epoch + || authority.snapshotId !== reference.snapshotId || !authority.location + || authority.location.environmentId !== bound.lease.environmentId || authority.location.leaseId !== bound.lease.id + || authority.location.providerLeaseId !== bound.lease.providerLeaseId || authority.location.remoteCwd !== reference.remoteCwd) return null; + return { ...bound, workspace: workspace!, reference, issueId }; +} + export async function inspectRemoteWorkspacePersistence(db: Db | Transaction, workspace: Workspace) { const authority = readRemoteWorkspaceAuthority(workspace.metadata); if (!authority || authority.dirty || authority.writerRunId || !authority.snapshotId) { diff --git a/server/src/services/task-retry.test.ts b/server/src/services/task-retry.test.ts index d760071806b..363663258be 100644 --- a/server/src/services/task-retry.test.ts +++ b/server/src/services/task-retry.test.ts @@ -1,7 +1,7 @@ import { randomUUID } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; -import { agents, agentWakeupRequests, companies, createDb, environmentLeases, environments, heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, toolApplications, toolConnections, toolInvocations, applyPendingMigrations } from "@paperclipai/db"; +import { agents, agentWakeupRequests, companies, createDb, environmentLeases, environments, executionWorkspaces, heartbeatRuns, issueComments, issues, issueRecoveryActions, nativeRunFinalizations, toolApplications, toolConnections, toolInvocations, applyPendingMigrations } from "@paperclipai/db"; import { startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; import { priorRecoveryApiReceipt, unresolvedRecoveryToolInvocation } from "./recovery-effect-receipts.js"; import { requestTaskRetry } from "./task-retry.js"; @@ -11,6 +11,7 @@ import { createSystemRecovery, failSystemRecovery, readSystemRecovery, SystemRec import { instanceSettingsService } from "./instance-settings.js"; import { heartbeatService } from "./heartbeat.js"; import { taskChatRetryCanAwaitRecovery, taskChatRetryExecutionStopped } from "./chat-channels/task-retry-authority.js"; +import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js"; describe("durable task Retry admission", () => { let postgres: Awaited>; @@ -57,6 +58,90 @@ describe("durable task Retry admission", () => { expect(await requestTaskRetry(input)).toMatchObject({ accepted: false, stale: true }); expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, input.companyId))).toHaveLength(0); }); + it.each(["stopped", "running", "owned", "changed_workspace", "changed_source", "missing_stop", "destroyed", "foreign_task"].flatMap(state => + ["retry", "message"].map(entry => ({ state, entry }))).concat([{ state: "stopped", entry: "preview" }]))( + "preserves the actual remote writer behind a newer setup failure only with exact stopped authority: $entry $state", async ({ state, entry }) => { + const input = await seed(); + const [latest] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.observedRunId)); + const writerId = randomUUID(), workspaceId = randomUUID(), pluginId = randomUUID(); + const sourceIssueId = state === "foreign_task" ? randomUUID() : input.issueId; + if (sourceIssueId !== input.issueId) await db.insert(issues).values({ id: sourceIssueId, + companyId: input.companyId, title: "Another task", status: "blocked", assigneeAgentId: latest.agentId }); + await db.update(heartbeatRuns).set({ errorCode: "setup_failed", resultJson: { + executionRecovery: { kind: "bootstrap", providerWorkStarted: false }, + } }).where(eq(heartbeatRuns.id, latest.id)); + await db.insert(heartbeatRuns).values({ id: writerId, companyId: input.companyId, agentId: latest.agentId, + issueId: sourceIssueId, nativeIssueId: sourceIssueId, runtimeMode: "native", + status: state === "running" ? "running" : "failed", finishedAt: state === "running" ? null : new Date(), + createdAt: new Date(latest.createdAt.getTime() - 60_000), + contextSnapshot: { issueId: sourceIssueId, executionWorkspaceId: workspaceId }, + resultJson: { workspaceRestoreFailure: "restore_failed", originalDiagnostic: true }, + }); + await db.insert(nativeRunFinalizations).values({ companyId: input.companyId, issueId: sourceIssueId, + runId: writerId, phase: "terminal_failure", leaseOwner: state === "owned" ? "live-controller" : null }); + const [environment] = await db.insert(environments).values({ name: `Stopped source ${writerId}`, driver: "sandbox" }).returning(); + const [lease] = await db.insert(environmentLeases).values({ companyId: input.companyId, issueId: sourceIssueId, + heartbeatRunId: writerId, environmentId: environment.id, provider: "daytona", providerLeaseId: randomUUID(), + status: "released", cleanupStatus: "success", leasePolicy: "reuse_by_environment", acquiredAt: new Date(Date.now() - 60_000), + releasedAt: new Date(), + }).returning(); + const confirmedAt = lease.releasedAt!.toISOString(); + const identity = { companyId: input.companyId, runId: writerId, leaseId: lease.id, + provider: lease.provider, providerLeaseId: lease.providerLeaseId, confirmedAt }; + const metadata = { driver: "sandbox", pluginId, sandboxProviderPlugin: true, remoteCwd: "/work", + ...(state === "missing_stop" ? {} : { + remoteExecutionTermination: { ...identity, schema: "paperclip.remote-termination.v1", state: state === "destroyed" ? "destroyed" : "stopped" }, + sandboxStopAndRetainReceipt: { ...identity, schema: "paperclip.sandbox-stop-and-retain-receipt.v1", + pluginId, method: "environmentStopLease", requestId: randomUUID() }, + }), + }; + await db.update(environmentLeases).set({ metadata }).where(eq(environmentLeases.id, lease.id)); + const reference = { schema: "paperclip.remote-workspace-run.v1", workspaceId, epoch: 3, + snapshotId: randomUUID(), leaseId: lease.id, providerLeaseId: lease.providerLeaseId, remoteCwd: "/work", state: "prepared" }; + await db.update(heartbeatRuns).set({ runnerProfileJson: { remoteWorkspace: { + ...reference, ...(state === "changed_source" ? { providerLeaseId: randomUUID() } : {}), + } } }).where(eq(heartbeatRuns.id, writerId)); + const authority = { schema: "paperclip.remote-workspace.v1", epoch: state === "changed_workspace" ? 4 : 3, + snapshotId: reference.snapshotId, source: { kind: "empty" }, writerRunId: writerId, dirty: true, + location: { environmentId: environment.id, leaseId: lease.id, providerLeaseId: lease.providerLeaseId, remoteCwd: "/work" } }; + await db.insert(executionWorkspaces).values({ id: workspaceId, companyId: input.companyId, sourceIssueId, + name: "Unfinished remote writer", mode: "isolated_workspace", strategyType: "project_primary", cwd: "/work", + providerType: "cloud_sandbox", metadata: { remoteWorkspace: authority } }); + await db.update(environmentLeases).set({ executionWorkspaceId: workspaceId }).where(eq(environmentLeases.id, lease.id)); + await db.update(issues).set({ executionWorkspaceId: workspaceId }).where(eq(issues.id, input.issueId)); + + if (entry === "retry") { + const receipt = await requestTaskRetry(input); + expect((await requestTaskRetry(input)).requestId).toBe(receipt.requestId); + } else { + const [comment] = await db.insert(issueComments).values({ companyId: input.companyId, issueId: input.issueId, + authorType: "user", authorUserId: input.actorId, body: "Continue from saved work.", + createdAt: new Date() }).returning(); + for (let attempt = 0; attempt < 2; attempt++) await db.transaction(tx => admitExplicitNativeContinuation({ + db: tx as unknown as typeof db, companyId: input.companyId, issueId: input.issueId, agentId: latest.agentId, + actorType: "user", actorId: input.actorId, commentId: comment.id, successorRunId: randomUUID(), + reason: "issue_commented", dryRun: entry === "preview", + })); + } + const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, writerId)); + const repairs = (await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, input.issueId))) + .filter(action => action.evidence.workspaceRestoreRecovery); + if (state === "stopped" && entry !== "preview") { + expect(source.resultJson).toMatchObject({ originalDiagnostic: true, + workspaceRestoreRecovery: { leaseIds: [lease.id] } }); + expect(repairs).toHaveLength(1); + expect(repairs[0].evidence.runId).toBe(writerId); + } else { + expect(source.resultJson).not.toHaveProperty("workspaceRestoreRecovery"); + expect(repairs).toHaveLength(0); + } + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id)))[0]) + .toMatchObject({ status: "released", leasePolicy: "reuse_by_environment", metadata }); + expect((await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspaceId)))[0].metadata?.remoteWorkspace) + .toEqual(authority); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, input.companyId))).toHaveLength(2); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, input.companyId))).toHaveLength(entry === "retry" ? 1 : 0); + }); it.each(["prepared", "mismatched", "owned"] as const)("reconstructs missing native preservation on deliberate Retry only for an exact unowned source: %s", async state => { const input = await seed(); const [issue] = await db.select().from(issues).where(eq(issues.id, input.issueId)); diff --git a/server/src/services/task-retry.ts b/server/src/services/task-retry.ts index 71b6b12c292..d1fb8fd9b6e 100644 --- a/server/src/services/task-retry.ts +++ b/server/src/services/task-retry.ts @@ -13,7 +13,7 @@ import { adoptSystemRecovery, updateSystemRecovery } from "./system-recovery.js" import { instanceSettingsService } from "./instance-settings.js"; import { notifyChatActionWork } from "./chat-work-notifications.js"; import { authorizationService } from "./authorization.js"; -import { preserveNativeFailedWorkspace } from "./native-runtime/native-failed-workspace-recovery.js"; +import { preserveNativeFailedWorkspace, preserveTaskRemoteWorkspaceWriter } from "./native-runtime/native-failed-workspace-recovery.js"; import { NativeWorkspaceFinalizationBusyError } from "./native-runtime/native-workspace-finalization-ownership.js"; import { adapterExecutionControls } from "./adapter-execution-control.js"; import { hasRequiredWorkspaceRecovery } from "./workspace-restore-recovery-state.js"; @@ -256,6 +256,7 @@ export async function requestTaskRetry(input: { } const busy = ["queued", "running", "scheduled_retry"].includes(latest.status); if (!busy) { + await preserveTaskRemoteWorkspaceWriter(db, issue.companyId, issue.id); // Historical deadline failures can predate their preservation record. // Register the exact prepared source; this proves neither a stopped // executor nor permission to replay its work. diff --git a/server/src/services/tool-access/github-grant-continuity.ts b/server/src/services/tool-access/github-grant-continuity.ts index 13aaec986e7..ec5842e6aa9 100644 --- a/server/src/services/tool-access/github-grant-continuity.ts +++ b/server/src/services/tool-access/github-grant-continuity.ts @@ -163,11 +163,11 @@ export function createGitHubGrantContinuity(db: Db, { now, currentCloudConnector throw notFound("GitHub authorization not found"); const previousGitHub = currentGrant.providerTenant?.github; const initialGitHub = grant.providerTenant?.github; - // No lock is held during provider requests. Reject a snapshot if another - // refresh or webhook changed access while those requests were in flight. + // Access revisions fence installation changes, and refresh timestamps also + // protect legacy snapshots. Ordinary PR webhooks update delivery health + // without changing access; preserve that health when this refresh commits. if ( previousGitHub?.accessRevision !== initialGitHub?.accessRevision || - previousGitHub?.lastWebhookAt !== initialGitHub?.lastWebhookAt || previousGitHub?.lastAccessRefreshAt !== initialGitHub?.lastAccessRefreshAt ) { diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index a086715dbc8..3768ae1528b 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -338,12 +338,17 @@ describe("automatic retained workspace recovery", () => { }, 30_000); it.each([ - { status: "failed", runtimeMode: "native", uploadFailure: "none" }, - { status: "cancelled", runtimeMode: "native", uploadFailure: "none" }, - { status: "failed", runtimeMode: "legacy", uploadFailure: "final" }, - { status: "failed", runtimeMode: "legacy", uploadFailure: "initial" }, - ] as const)("saves the authoritative remote workspace after $runtimeMode $status without replaying a turn or certifying external effects ($uploadFailure upload failure)", async ({ status, runtimeMode, uploadFailure }) => { + { status: "failed", runtimeMode: "native", scenario: "none" }, + { status: "failed", runtimeMode: "native", scenario: "registration" }, + { status: "failed", runtimeMode: "native", scenario: "registration_after_dispatch" }, + { status: "failed", runtimeMode: "native", scenario: "registration_existing_hold" }, + { status: "failed", runtimeMode: "native", scenario: "registration_existing_hold_after_dispatch" }, + { status: "cancelled", runtimeMode: "native", scenario: "none" }, + { status: "failed", runtimeMode: "legacy", scenario: "final" }, + { status: "failed", runtimeMode: "legacy", scenario: "initial" }, + ] as const)("saves the authoritative remote workspace after $runtimeMode $status without replaying a turn or certifying external effects ($scenario)", async ({ status, runtimeMode, scenario }) => { const f = await seed(runtimeMode, false); + let retryRequestId: string | undefined; const remoteCwd = await realpath(f.remote); targets.get(f.lease.id)!.remoteCwd = remoteCwd; f.runtime.resumeRunLease.mockImplementation(async () => ({ providerLeaseId: f.providerLeaseId, metadata: { remoteCwd } })); @@ -361,7 +366,7 @@ describe("automatic retained workspace recovery", () => { metadata: { ...boundLease.metadata, remoteCwd, agentId: (await run(f.runId)).agentId } }) .where(eq(environmentLeases.id, f.lease.id)); const before = await run(f.runId); - if (uploadFailure === "final") { + if (scenario === "final") { // A reused allocation can fail during preparation, before native input or // its finalization coordinator exists. Keep a real saved predecessor and // the ordinary exact lease handoff; neither grants a new provider turn. @@ -415,7 +420,7 @@ describe("automatic retained workspace recovery", () => { expect((await recoverySources.inspectWorkspaceRecoverySource(db, f.input)).sources).toHaveLength(1); const failedUpload = vi.spyOn(getStorageService(), "putFile").mockRejectedValueOnce(new Error("Synthetic saved-copy outage")); try { - if (uploadFailure === "initial") { + if (scenario === "initial") { // Exercise the original setup boundary, not a failed final save that // retries the same operation. The initial ledger reservation survives. await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)); @@ -453,7 +458,48 @@ describe("automatic retained workspace recovery", () => { epoch: authority.epoch, snapshotId: authority.snapshotId }); expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(false); expect(await readFile(path.join(f.remote, "saved.txt"), "utf8")).toBe("finished work\n"); - if (uploadFailure === "final") await makeRetryDue(f.actionId); + if (scenario === "final") await makeRetryDue(f.actionId); + } + if (scenario.startsWith("registration")) { + await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.actionId)); + await db.update(heartbeatRuns).set({ resultJson: { workspaceRestoreFailure: "restore_failed", originalDiagnostic: true }, + runnerProfileJson: { ...(await run(f.runId)).runnerProfileJson, nativeResourceDisposition: "keep_running" }, + }).where(eq(heartbeatRuns.id, f.runId)); + const [stopped] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)); + const termination = stopped.metadata!.remoteExecutionTermination as Record; + await db.update(environmentLeases).set({ leasePolicy: "reuse_by_environment", releasedAt: new Date(String(termination.confirmedAt)), metadata: { ...stopped.metadata, + sandboxStopAndRetainReceipt: { ...termination, schema: "paperclip.sandbox-stop-and-retain-receipt.v1", + pluginId: stopped.metadata!.pluginId, method: "environmentStopLease", requestId: randomUUID() }, + } }).where(eq(environmentLeases.id, f.lease.id)); + const [later] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: before.agentId, + issueId: f.issueId, runtimeMode: "legacy", status: "failed", errorCode: "setup_failed", finishedAt: new Date(), + executionStage: "preparing", runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } }, + createdAt: new Date(before.createdAt.getTime() + 1), contextSnapshot: { issueId: f.issueId }, + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } }, + }).returning(); + if (scenario.endsWith("after_dispatch")) await appendHeartbeatRunEvent(db, { + companyId: f.companyId, runId: later.id, agentId: before.agentId, eventType: "adapter.invoke", stream: "system", + level: "info", message: "Execution started before the contradictory bootstrap result.", payload: null, + }); + const existingHold = scenario.includes("existing_hold"); + await issueRecoveryActionService(db).upsertSourceScoped({ companyId: f.companyId, sourceIssueId: f.issueId, + kind: existingHold ? "active_run_watchdog" : "stranded_assigned_issue", + cause: existingHold ? "legacy_execution_requires_reconciliation" : "stranded_assigned_issue", fingerprint: `latest-setup:${later.id}`, + ownerType: "board", nextAction: "Inspect the failed setup before continuing.", + evidence: existingHold ? { runId: later.id, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } } + : { latestRunId: later.id, latestRunStatus: "failed", latestRunErrorCode: "setup_failed" } }); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: before.agentId, status: "running" }); + const app = boardRecoveryApp(), retryIntent = { observedRunId: later.id, idempotencyKey: randomUUID() }; + const retry = (await request(app).post(`/api/issues/${f.issueId}/retry`).send(retryIntent).expect(202)).body; + retryRequestId = retry.requestId; + expect((await request(app).post(`/api/issues/${f.issueId}/retry`).send(retryIntent).expect(202)).body.requestId).toBe(retryRequestId); + await heartbeatService(db).resumeTaskRetry(f.companyId, retryRequestId!); + const [registered] = await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, f.companyId), + sql`${issueRecoveryActions.evidence}->>'runId' = ${f.runId}`)); + expect(registered).toBeDefined(); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(3); + f.actionId = f.input.actionId = registered.id; + await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, async () => {}) ?? []); } await recoverRetainedWorkspace(f.input); const recoveredAction = await action(f.actionId); @@ -474,24 +520,44 @@ describe("automatic retained workspace recovery", () => { } finally { await extracted.cleanup(); } expect(await readFile(path.join(f.host, "newer.txt"), "utf8")).toBe("newer local work\n"); - expect(await run(f.runId)).toMatchObject({ status, errorCode: uploadFailure === "initial" ? "setup_failed" : "execution_interrupted", + expect(await run(f.runId)).toMatchObject({ status, errorCode: scenario === "initial" ? "setup_failed" : "execution_interrupted", runnerProfileJson: { remoteWorkspace: { state: "saved", snapshotId: saved!.snapshotId } } }); - expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(0); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))) + .toHaveLength(retryRequestId ? 1 : 0); expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId))).toHaveLength(0); expect(await db.select().from(costEvents).where(eq(costEvents.companyId, f.companyId))).toHaveLength(0); expect(await db.select().from(workspaceOperations).where(and(eq(workspaceOperations.heartbeatRunId, f.runId), eq(workspaceOperations.phase, "workspace_recovery_snapshot")))).toHaveLength(0); - expect(f.runtime.resumeRunLease).toHaveBeenCalledTimes(uploadFailure === "final" ? 2 : 1); + expect(f.runtime.resumeRunLease).toHaveBeenCalledTimes(scenario === "final" ? 2 : 1); expect(f.runtime.retryPendingSandboxTeardown).toHaveBeenCalledTimes(runtimeMode === "legacy" ? 2 : 1); expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(true); await scheduleRecoveredWorkspaceCleanup(db, await run(f.runId)); const [settledLease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)); + if (retryRequestId) { + expect(settledLease).toMatchObject({ status: "released", leasePolicy: "reuse_by_environment" }); + await heartbeatService(db).resumeTaskRetry(f.companyId, retryRequestId); + await heartbeatService(db).resumeTaskRetry(f.companyId, retryRequestId); + const queued = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued"))); + if (scenario.endsWith("after_dispatch")) { + expect(queued).toHaveLength(0); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, retryRequestId)))[0]) + .toMatchObject({ status: "deferred_issue_execution", runId: null }); + return; + } + expect(queued).toHaveLength(1); + expect(await readFreshWorkspaceReadiness(db, queued[0])) + .toMatchObject({ mode: "resume_preserved_workspace", preservationReceiptId: expect.any(String) }); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toMatchObject({ executionWorkspaceId: workspaceId }); + expect((await run(f.runId)).resultJson).toMatchObject({ originalDiagnostic: true }); + expect(f.runtime.resumeRunLease).toHaveBeenCalledTimes(1); + return; + } if (runtimeMode === "legacy") { expect(settledLease).toMatchObject({ status: "released", - leasePolicy: uploadFailure === "initial" ? "retain_on_failure" : "reuse_by_execution_workspace" }); + leasePolicy: scenario === "initial" ? "retain_on_failure" : "reuse_by_execution_workspace" }); expect((await run(f.runId)).runtimeMode).toBe("legacy"); expect(await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId))).toEqual([]); - if (uploadFailure === "final") expect(snapshots.find(snapshot => snapshot.id === authority.snapshotId)).toMatchObject({ status: "superseded" }); + if (scenario === "final") expect(snapshots.find(snapshot => snapshot.id === authority.snapshotId)).toMatchObject({ status: "superseded" }); else expect(snapshots.find(snapshot => snapshot.generation === 1)).toMatchObject({ status: "pending" }); // Explicit Retry may admit the preserved files. Saturate agent capacity // so this tests admission without dispatching a new agent turn. @@ -504,7 +570,7 @@ describe("automatic retained workspace recovery", () => { expect(await readFreshWorkspaceReadiness(db, successor)) .toMatchObject({ mode: "resume_preserved_workspace", preservationReceiptId: expect.any(String) }); expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, successor.id))).toEqual([]); - expect(f.runtime.resumeRunLease).toHaveBeenCalledTimes(uploadFailure === "final" ? 2 : 1); + expect(f.runtime.resumeRunLease).toHaveBeenCalledTimes(scenario === "final" ? 2 : 1); return; } if (status === "failed") {