diff --git a/agents/process-group.ts b/agents/process-group.ts index bd460e57..f7bc914b 100644 --- a/agents/process-group.ts +++ b/agents/process-group.ts @@ -49,6 +49,13 @@ export interface ProcessGroupOptions { * Callers without durable ownership wait fail-closed instead of returning while unknown work is still live. */ readonly allowUnsettledReturn?: boolean; + /** + * Bound only the wait for the output pipes once the whole group is gone: a process that left the group but still holds + * them makes the call return failed (`ESTDIOHELD`) after the pipe drain, instead of waiting for it. Members of the + * group are still waited for, fail-closed. For callers whose command changes nothing (a read), so a process that + * escaped the group cannot affect anything they decide. + */ + readonly boundEscapedOutput?: boolean; } const DEFAULT_GRACE_MS = 5_000; @@ -94,7 +101,8 @@ export const drainProcessGroup = (group: ProcessGroup): Promise => drai * returns control to the event loop. On abort or at the deadline the whole group gets SIGTERM, then SIGKILL after the * grace period, so a child that ignores SIGTERM is still bounded. The promise settles only after the leader has * exited and every other member of its group is gone, unless a caller with durable recovery ownership explicitly - * allows an unsettled return. Never rejects: failures are in the outcome, shaped like `runDocker`'s, so `status` is a + * allows an unsettled return. `boundEscapedOutput` bounds only the wait for a process outside the group that holds the + * output pipes. Never rejects: failures are in the outcome, shaped like `runDocker`'s, so `status` is a * number only when the command ran to completion. */ export function runInProcessGroup(file: string, args: readonly string[], @@ -175,7 +183,7 @@ export function runInProcessGroup(file: string, args: readonly string[], options.signal?.removeEventListener('abort', onAbort); void drainGroup(group, options.allowUnsettledReturn === true).then(async drained => { let stdioTimer: ReturnType | undefined; - if (options.allowUnsettledReturn) { + if (options.allowUnsettledReturn || options.boundEscapedOutput) { await Promise.race([closed, new Promise(done => { stdioTimer = setTimeout(done, STDIO_CLOSE_MS); })]); } else await closed; // Cleared so a finished call never keeps the process alive. diff --git a/docs/architecture.md b/docs/architecture.md index a107e91d..542d7ddb 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -42,9 +42,9 @@ Writing standard: plain language, ISO 24495-1:2023 - **Build state:** - **In use:** plan and linking library, SQLite store, review screen, plan display/import/authoring screen, Ask (questions to a Claude agent), guarded merge with merge-queue support, ranked Issues screen, agent isolation boundary, and the single-runner lock. - **In use for a non-demo review with a `github` block:** planning. `/api/plan/suggestions` and `/api/plan/drafts` ask Claude, through lane D, for suggestion cards or a whole next plan revision (#117, #124). Nothing becomes a revision until you apply it. - - **In use with an opt-in `runner` block in `review.json`:** the whole loop after planning. `start` and `resume` on `/api/runner` run a task's plan item by item with a Claude agent (#91). They require current plan-item approvals, bind both the task state and review version, recheck approvals before each later item, and refuse when a completed prefix no longer ends at the task's current head (#107). After an out-of-scope pause, an approved amended plan can reconcile the completed prefix and resume only the unfinished suffix (#88, PR #134). When a run ends, codeboost publishes the task's pull request, a draft if the task needs a person (#103). `prepare-merge` refreshes the published PR's exact base/head, rebases through the durable F3/F4 path, recomputes attribution and approvals, and runs allowlisted `cmd:` checks in a credential-free read-only container bound to the resulting head (#22). Approve & merge merges that pull request (#121). Cancelling a task closes its pull requests (#111). These are API actions; the screen has no buttons for them yet. - - **Partly built for pre-merge automation:** the trusted local rebase engine records one-to-one commit mappings, verifies the resulting checkout byte for byte, owns its subprocesses durably, and recovers interrupted work (#22, PR #136). Its production conflict path gives a sandboxed agent only the bounded conflict files and imports only audited results for both foreign and plan-owned commits (PRs #138, #140 and #144). `prepare-merge` now invokes that path after current pull-request base/head reads, then refreshes attribution and approvals and runs exact-head `cmd:` checks; required-check refresh, the already-fixed rerun, rewritten-head push, and guarded merge handoff remain for F6. - - **Not built:** screens for the runner actions, remote push/check refresh, review rounds, and the Queue and Learning screens. + - **In use with an opt-in `runner` block in `review.json`:** the whole loop after planning. `start` and `resume` on `/api/runner` run a task's plan item by item with a Claude agent (#91). They require current plan-item approvals, bind both the task state and review version, recheck approvals before each later item, and refuse when a completed prefix no longer ends at the task's current head (#107). After an out-of-scope pause, an approved amended plan can reconcile the completed prefix and resume only the unfinished suffix (#88, PR #134). When a run ends, codeboost publishes the task's pull request, a draft if the task needs a person (#103). `prepare-merge` refreshes the published PR's exact base/head, rebases through the durable F3/F4 path, recomputes attribution and approvals, pushes a rewritten head whose review is still clear over exactly the PR head it rewrote, and runs allowlisted `cmd:` checks in a credential-free read-only container bound to the resulting head (#22). Approve & merge merges that pull request (#121). Cancelling a task closes its pull requests (#111). These are API actions; the screen has no buttons for them yet. + - **Partly built for pre-merge automation:** the trusted local rebase engine records one-to-one commit mappings, verifies the resulting checkout byte for byte, owns its subprocesses durably, and recovers interrupted work (#22, PR #136). Its production conflict path gives a sandboxed agent only the bounded conflict files and imports only audited results for both foreign and plan-owned commits (PRs #138, #140 and #144). `prepare-merge` now invokes that path after current pull-request base/head reads, refreshes attribution and approvals, pushes the rewritten head behind a durable push marker (F6a), and runs exact-head `cmd:` checks; required-check refresh, the already-fixed rerun, and the one-click guarded merge handoff remain for F6b. + - **Not built:** screens for the runner actions, the required-check refresh and one-click merge handoff (F6b), review rounds, and the Queue and Learning screens. ## Terms used @@ -357,7 +357,7 @@ The diagram shows the designed transitions. With the `runner` block, running, ne ### Stored data -All state is in one SQLite file, opened with WAL and full synchronization. Each write takes an immediate write lock. The schema version is `PRAGMA user_version`, and migrations run in explicit steps (version 17 today); an unknown version fails. +All state is in one SQLite file, opened with WAL and full synchronization. Each write takes an immediate write lock. The schema version is `PRAGMA user_version`, and migrations run in explicit steps (version 18 today); an unknown version fails. | Group | Tables | Notes | |---|---|---| @@ -410,7 +410,7 @@ These are the decisions that shape the structure. The full list, with the reason | C, K | Guarded merge, merge queue | Done | | D | Agent isolation boundary | Done. Since the isolation gate: bounded diff export (#76), task-change inspection and the runner's commit as a bounded `git bundle` (#66, PRs #85 and #90), owner-scoped recovery for Ask (#95), the plan schema for Claude (#97), gitlink mounts and the pre-launch tree check (#81, #99, PR #100), and Codex refused in every phase (#93, PR #106). The runner-contract follow-ups (#51) are closed: the last one tests export and removal through a recovery handle after a real restart (PR #109). | | E | Planning library and suggestions | Done. E4 (PR #45) added the acceptance fixtures and recorded Claude output that replays through validation and the Store. | -| F | Runner | Merged: F1 (lifecycle store, coordinator, shutdown wiring, startup recovery and the lock, planning API, feedback events, and stops that wait for their commit: #53, #74, #57, #59, #60, #96); F2a–F2b (execute prompts, audit, per-item execution, real workspace, durable safety findings: #67, #68, #92, #94, #128); F2d (already-fixed check, PR opening, branch pusher, production publishing, closing PRs on cancel: #84, #101, #110, #115, #116, #119, #123); production wiring and `start`/`resume` (#91, PRs #102, #105); planning through lane D (#117, #124, PRs #118, #120, #125); merging the task's own PR (#121, PR #126); start/resume approval, review-version, recovery and completed-prefix guards (#107, PR #130); continuation after a scope pause (#88, PR #134); the F3 durable trusted local-rebase foundation (#22, PR #136); the F4 foreign-conflict engine plus production sandbox adapter (#22, PRs #138 and #140); and sandboxed owned-commit conflict resolution (#22, PR #144). This branch adds F5: current base/head refresh, rebase admission, post-rebase attribution and approval refresh, and head-bound `cmd:` execution. F6 owns rewritten-head push, fresh required-check refresh, the already-fixed rerun, and exact-pair guarded-merge handoff. | +| F | Runner | Merged: F1 (lifecycle store, coordinator, shutdown wiring, startup recovery and the lock, planning API, feedback events, and stops that wait for their commit: #53, #74, #57, #59, #60, #96); F2a–F2b (execute prompts, audit, per-item execution, real workspace, durable safety findings: #67, #68, #92, #94, #128); F2d (already-fixed check, PR opening, branch pusher, production publishing, closing PRs on cancel: #84, #101, #110, #115, #116, #119, #123); production wiring and `start`/`resume` (#91, PRs #102, #105); planning through lane D (#117, #124, PRs #118, #120, #125); merging the task's own PR (#121, PR #126); start/resume approval, review-version, recovery and completed-prefix guards (#107, PR #130); continuation after a scope pause (#88, PR #134); the F3 durable trusted local-rebase foundation (#22, PR #136); the F4 foreign-conflict engine plus production sandbox adapter (#22, PRs #138 and #140); and sandboxed owned-commit conflict resolution (#22, PR #144). F5 (PR #149) added current base/head refresh, rebase admission, post-rebase attribution and approval refresh, and head-bound `cmd:` execution. This branch adds F6a: the durable push of the rewritten head. F6b owns fresh required-check refresh, the already-fixed rerun, and the one-click exact-pair guarded-merge handoff; F6c owns common CI. | | Hardening | Allowlisted subprocess environments | Done: every Git command (#82, #83) and every `gh` runner (#84, #86). | | Tests | Test suite reliability | Done. The full suite runs ordinary tests first and Docker-backed files serially; CI and the real-Docker gate passed on PR #132 (#129). | | G | Planning screen | Done (#145). The screen displays/imports the current plan, requests and renders a next-revision draft or suggestion cards, and applies results with revision-bound exact ambiguous replay. G4 composes the browser with the production planning setup, D-backed provider boundary, F endpoints and SQLite persistence; focused adapter, recorded-output and exact-head real-Docker planning suites cover the lower boundary. | @@ -419,7 +419,7 @@ These are the decisions that shape the structure. The full list, with the reason ## Known limits -- `prepare-merge` invokes the trusted rebase and sandboxed conflict machinery, refreshes the review, and runs head-bound `cmd:` checks. Until F6 of #22 lands, it does not push a rewritten head, wait for fresh required checks, rerun the already-fixed check, or hand the exact pair to guarded merge. +- `prepare-merge` invokes the trusted rebase and sandboxed conflict machinery, refreshes the review, pushes the rewritten head, and runs head-bound `cmd:` checks. Until F6b of #22 lands, it does not wait for fresh required checks, rerun the already-fixed check, or hand the exact pair to guarded merge; a person clicks Merge once GitHub's checks on the pushed head pass. - Linking is line-based. It cannot see an unrelated edit inside a declared file; only the review agent and you can catch that. - The container's guarantees are those of Docker. The agent can always reach its own vendor account. - `cmd:` acceptance checks run during `prepare-merge`, and their passing evidence is bound to the exact reviewed head. A plan item without current passing evidence still blocks merging. diff --git a/docs/implementation/pre-merge-rebase.md b/docs/implementation/pre-merge-rebase.md index 157f6732..7a2acd12 100644 --- a/docs/implementation/pre-merge-rebase.md +++ b/docs/implementation/pre-merge-rebase.md @@ -1,6 +1,6 @@ # Pre-merge rebase and sandboxed conflict boundary -**Issue and lane:** #22, lanes F3-F4. This is the trusted local rewrite, recovery, and sandboxed conflict engine boundary. The `rebase-fix` child adapter is assembled inside the production rebaser through lane D with durable resource ownership and bounded exact-path copy-in/copy-out. No production request starts that rebaser yet: F5-F6 must own admission plus the current base/head refresh before invoking it, then push the rewritten head, run plan `cmd:` checks, wait for GitHub checks, and hand the validated pair to the merge coordinator. Exposing the raw rebaser earlier would bypass those required guards. +**Issue and lane:** #22, lanes F3-F4. This is the trusted local rewrite, recovery, and sandboxed conflict engine boundary. The `rebase-fix` child adapter is assembled inside the production rebaser through lane D with durable resource ownership and bounded exact-path copy-in/copy-out. `prepare-merge` starts it after admission and the current base/head refresh (F5); the rewritten head is then pushed (F6a) and plan `cmd:` checks run. Waiting for GitHub's checks and handing the validated pair to the merge coordinator follow in F6b. Exposing the raw rebaser earlier would bypass those required guards. ## Ownership and lifecycle @@ -30,7 +30,23 @@ All other conflicts are a review outcome (`RebaseConflict`). Cancellation and fa Startup first validates every stored marker and child identity. A recovered `spawning` marker is ambiguous because the runner may have crashed before the synchronous spawn callback recorded the child. A recovered exact process group is also ambiguous even after its leader or numeric group disappears: the runner may have crashed before an escaped output-pipe holder was durably marked `unsettled`, and that process need not use the workspace. Recovery requests identity-bound termination for an exact live group when its process controller can provide it; the host controller instead fails closed on a live reusable numeric group. Once no live exact group blocks the transition, recovery changes either ambiguous owner to `unsettled` and keeps admission closed. The operator must stop any unknown process and run `--release-rebase-process ` under the runner lock. Release revalidates the complete runner/owner/rebases chain as owner-only plain directories before it inspects the attempt leaf, and it refuses any visible workspace user; preparation release applies the same check to the attempts chain. Lane D then removes this runner's labelled agent containers, proxies, and networks and returns authenticated storage handles. The production recovery adapter accepts only the configured plan key, then removes the matching worktree. It deletes a result ref only with Git's exact-old-value guard using the durably recorded `resultHead`; absence is already-clean success. A `prepared` marker preserves a different pre-existing value, while a `ready` marker fails closed if its ref changed. A legacy, pre-result, or reused-attempt marker without an intended value preserves any existing ref. An exact child storage identity is removed after the parent workspace and before either marker clears; a partial match fails closed. A live run whose child cleanup is uncertain retains the parent workspace instead of racing those resources. Only after cleanup succeeds does `Store.abortRebase` compare-and-swap the same attempt ID to clear the marker. A missing adapter, malformed marker, wrong plan, recovered or unknown process, cleanup failure, unsafe ancestor, workspace user, or changed marker blocks startup and retains the durable ownership evidence. -This ordering is intentionally local-only. The later push integration must extend the durable lifecycle before its first external write and reconcile an ambiguous push outcome; it must not reuse this local completion point as proof that GitHub moved. +This ordering is intentionally local-only. The push below extends the durable lifecycle before its first external write and reconciles an ambiguous outcome by reading the branch; it never treats this local completion point as proof that GitHub moved. + +## Push of the rewritten head (F6a) + +`prepare-merge` pushes a rewritten head only when the refreshed review has no blocker: every item approved, no ambiguous or unplanned change, no open change request. A rewrite that made an approval stale stays local until a person reviews it; a later preparation recognizes the unpushed rewrite through the durable rewrite lineage. + +| Step | What happens | +|---|---| +| Admission | The task state version is captured, then issue trust is refreshed. `Store.beginPrePush` then records `{ attemptId, branch, from, to, snapshotId, startedAt }` in `tasks.push_in_progress` (schema v18) and advances the task state version. It requires that captured state version (a reassignment during the trust read refuses), a mergeable status, no cancel, no active attempt, merge, rebase or push, `to` equal to the snapshot head, and `from` a durable rewrite predecessor of `to`. | +| Last local check | Run by the pusher as its synchronous finalizer, as Git starts: trust is revalidated and the task state version, review version and snapshot must be unchanged since admission. After a push, the same three must still be exactly what the push left before any `cmd:` check runs, so a plan revision or review edit made while the push was in flight fails the preparation. The same check runs before and after the new head is fetched for review (after a refusal, a `moved` outcome, or a pull request that moved after the push): a change made during the push, the wait or the fetch fails the preparation as stale instead of being reviewed. | +| Push | `GitBranchPusher` pushes `to` to the task PR's `codeboost/` branch with a lease pinned to exactly `from`, the PR head read before the rebase. Any other value at the tip refuses with nothing pushed, even a commit codeboost made. The branch is read back after the push. | +| Refused | Nothing was pushed. The marker clears. Once GitHub's PR API reports the branch's new head (the wait below), that head is reviewed and preparation returns `review-required`. | +| Visible on the PR | After a push or a refusal, GitHub's pull-request API reports the branch's new head asynchronously. Preparation inspects the PR with backoff until it reports a head other than the pre-push head. It starts no inspection after 30 seconds, so the wait lasts at most 30 seconds plus one inspection (a 6-second deadline in production, plus up to 6 seconds for a stopped `gh` call to settle), within the preparation deadline. A base change alone does not end the wait. The pushed head continues to the `cmd:` checks; any other head or base is reviewed and returns `review-required`; a PR still at the pre-push head fails the preparation without touching the review. | +| Unknown outcome | A failure, a timeout or a cancel during the push reads the branch once, unless the preparation's own deadline ended the push (see the next row). The read and Git's settlement after an abort together fit within 30 seconds and what is left of the preparation's deadline: the read is aborted the 16-second process-settlement reserve early, and it does not start when less than that reserve is left. A process that left Git's group and still holds its output does not hold the read either: the read changes nothing, so after the 1-second pipe drain it returns failed rather than waiting, and the marker stays. A process that stays in Git's group after Git exits is waited for, as for every Git call: it cannot be signalled safely once Git has exited, because its group ID may be reused, so that one case is not bounded. Its answer: `to` means pushed, `from` means not pushed yet, anything else means moved. The marker then clears. After `moved`, the preparation reviews that head as after a refusal: it waits for the pull request to show it, refreshes attribution and approvals, and returns `review-required`. After `pushed` or `not pushed`, the push's error fails the preparation. Only shutdown aborts that read; a cancel requested during the push lets it finish and closes the task when the marker clears. | +| Unreadable, deadline or shutdown | The marker stays when the branch cannot be read, when the preparation's own deadline ended the push (less than the settlement reserve is left, so no read starts), or on shutdown. Merge admission, runner admission, status changes, reassignment, another rebase and another push all refuse while it exists, and merge status names it. Publishing cannot overlap it either: it runs only from `running` or `needs human`, and a push requires a mergeable status. After a restart, a tracked job started under the verified runner lock reads the branch and clears it; a cancel requested while no preparation runs starts the same job, and repeating that cancel starts it again if its read failed; preparation requests are refused as already running until it ends, and a task it closes as cancelled has its PRs closed. Otherwise the next preparation reads the branch, clears the marker, and asks for a reload. | + +A push that was killed (by a cancel, a deadline or a crash) after GitHub received it may still land after the settlement read saw `from`. Its lease still names `from`, so it overwrites nothing newer, and a later preparation reviews the branch as it finds it. ## Regression evidence diff --git a/runner/branch-push.ts b/runner/branch-push.ts index 32fccb48..b54161b8 100644 --- a/runner/branch-push.ts +++ b/runner/branch-push.ts @@ -10,7 +10,10 @@ import type { GitCallOptions, RunnerRepository } from './runner-repository.ts'; * user's checkout is never read or written, and neither the user's Git config nor their credential helpers are used. */ -/** The branch holds a commit codeboost did not make, or moved while the push ran. Nothing was pushed; a person decides. */ +/** + * Nothing was pushed: the branch holds a commit the push may not overwrite (one codeboost did not make, or, with an + * `expected` head, anything else), or it moved while the push ran. A person, or a fresh review, decides. + */ export class BranchPushRefused extends Error {} export interface GitBranchPusherConfig extends Pick { @@ -21,8 +24,8 @@ export interface GitBranchPusherConfig extends Pick Iterable; /** The push URL. Default `https:///.git`. A local path is allowed, for tests. */ @@ -113,7 +116,7 @@ export class GitBranchPusher implements BranchPusher { const unique = [...new Set(commits)]; await this.#git(['fetch', '--no-tags', '--no-write-fetch-head', '--no-recurse-submodules', '--', this.#url, ...unique], signal, true); const found = await this.#git(['cat-file', '--batch-check=%(objectname) %(objecttype)'], signal, false, - Buffer.from(`${unique.join('\n')}\n`)); + Buffer.from(`${unique.join('\n')}\n`), true); const records = found.split('\n'); if (records.length !== unique.length || records.some((record, index) => record !== `${unique[index]} commit`)) throw new Error('The refreshed remote base or head is not a commit in the runner repository.'); @@ -123,22 +126,31 @@ export class GitBranchPusher implements BranchPusher { await this.#git(['update-ref', `refs/codeboost/remote-commits/${commit}`, commit], signal); } - async push(identity: PlanIdentity, input: { head: string; branch: string; beforePush?: MutationBoundary }, signal?: AbortSignal): Promise { + async push(identity: PlanIdentity, input: { head: string; branch: string; expected?: string; beforePush?: MutationBoundary }, + signal?: AbortSignal): Promise { signal?.throwIfAborted(); if (identity.repositoryId !== this.#config.repositoryId) throw new Error('The task belongs to another repository.'); if (!COMMIT_ID.test(input.head)) throw new Error('A full commit ID is required.'); - if (!BRANCH.test(input.branch) || input.branch.length > 255) throw new Error('Only a codeboost/ task branch can be pushed.'); - const ref = `refs/heads/${input.branch}`; + if (input.expected !== undefined && !COMMIT_ID.test(input.expected)) throw new Error('A full commit ID is required.'); + const ref = this.#ref(input.branch); // Git answers " missing" for an absent object; any other failure (no repository, permissions) keeps Git's error. - const found = await this.#git(['cat-file', '--batch-check=%(objectname) %(objecttype)'], signal, false, Buffer.from(`${input.head}\n`)); + const found = await this.#git(['cat-file', '--batch-check=%(objectname) %(objecttype)'], signal, false, Buffer.from(`${input.head}\n`), true); if (found === `${input.head} missing`) throw new Error(`The runner repository has no commit ${input.head}.`); if (found !== `${input.head} commit`) throw new Error(`${input.head} is not a commit in the runner repository.`); const remote = await this.#read(ref, signal); // Already there: a push whose outcome was lost, or a retry. Nothing to do. if (remote === input.head) return; - const owned = new Set(this.#config.ownedCommits(identity)); - if (remote !== null && !owned.has(remote)) - throw new BranchPushRefused(`The branch ${input.branch} holds commit ${remote}, which codeboost did not make. Nothing was pushed.`); + // `expected` is the exact head a pre-merge rebase rewrote (F6 of #22). Its commits, a collaborator's included, live on + // in the rewritten head, so it alone may be overwritten: any other value, even a commit codeboost made, refuses. + if (input.expected !== undefined) { + if (remote !== input.expected) + throw new BranchPushRefused(`The branch ${input.branch} holds ${remote ?? 'nothing'}, not the rebased head ${input.expected}. Nothing was pushed.`); + } else { + // Read after the remote branch, every time, so a ledger written meanwhile is seen. + const owned = new Set(this.#config.ownedCommits(identity)); + if (remote !== null && !owned.has(remote)) + throw new BranchPushRefused(`The branch ${input.branch} holds commit ${remote}, which codeboost did not make. Nothing was pushed.`); + } signal?.throwIfAborted(); const finalize = await input.beforePush?.(); finalize?.(); @@ -155,21 +167,39 @@ export class GitBranchPusher implements BranchPusher { if (after !== input.head) throw new Error(`After the push, the branch ${input.branch} is at ${after ?? 'nothing'}, not ${input.head}.`); } + /** The commit the task branch points at on the remote, or null when it does not exist. */ + async readBranch(identity: PlanIdentity, branch: string, signal?: AbortSignal): Promise { + if (identity.repositoryId !== this.#config.repositoryId) throw new Error('The task belongs to another repository.'); + return this.#read(this.#ref(branch), signal); + } + + #ref(branch: string): string { + if (!BRANCH.test(branch) || branch.length > 255) throw new Error('Only a codeboost/ task branch can be pushed.'); + return `refs/heads/${branch}`; + } + /** The commit `ref` points at on the remote, or null when it does not exist. */ async #read(ref: string, signal?: AbortSignal): Promise { - const out = await this.#git(['ls-remote', '--refs', '--', this.#url, ref], signal, true); + // A read changes nothing, so it need not wait for a process that escaped Git's group and holds its output: it returns + // failed (the output may be incomplete) after the pipe drain. The caller then knows nothing: a push marker stays. + // Members still in Git's group are waited for, as for every Git call. + const out = await this.#git(['ls-remote', '--refs', '--', this.#url, ref], signal, true, undefined, true); // ls-remote matches patterns by their tail, so `refs/heads/x/` would match too: keep only the exact name. const matches = out.split('\n').filter(Boolean).map(line => line.split('\t')).filter(([, name]) => name === ref); if (matches.length > 1 || (matches[0] && !COMMIT_ID.test(matches[0][0]!))) throw new Error(`The remote's answer for ${ref} is malformed.`); return matches[0]?.[0] ?? null; } - async #git(args: readonly string[], signal?: AbortSignal, remote = false, input?: Buffer): Promise { + /** + * `readOnly` (ls-remote, cat-file: commands that change nothing) bounds the wait for a process that left Git's group + * but holds its output: the call fails after the pipe drain instead. Members of Git's group are always waited for. + */ + async #git(args: readonly string[], signal?: AbortSignal, remote = false, input?: Buffer, readOnly = false): Promise { const protocol = this.#url.startsWith('/') ? 'file' : 'https'; const outcome = await runInProcessGroup('git', [...GIT_OPTIONS, '-c', 'gc.auto=0', '-c', 'maintenance.auto=false', ...remote ? ['-c', `protocol.${protocol}.allow=always`, ...CREDENTIAL_HELPER] : [], ...args], { cwd: this.#config.repository.path, env: pushEnvironment(this.#config.env), timeoutMs: this.#config.timeoutMs ?? 120_000, - signal, onProcessGroup: this.#config.onProcessGroup, maxBuffer: 1024 * 1024, input }); + signal, onProcessGroup: this.#config.onProcessGroup, maxBuffer: 1024 * 1024, input, boundEscapedOutput: readOnly }); // Aborted before or while Git ran: the caller's reason (such as shutdown) is what leaves, not Git's failure. signal?.throwIfAborted(); if (outcome.status !== 0) throw gitFailure(args[0]!, outcome.status, [outcome.stderr.trim(), outcome.stdout.trim()].filter(Boolean).join('\n'), diff --git a/runner/merge.ts b/runner/merge.ts index e46f0e7e..e1eae8d6 100644 --- a/runner/merge.ts +++ b/runner/merge.ts @@ -162,8 +162,11 @@ export class MergeCoordinator { return (await this.#status(view, fresh, signal)).status; } - /** Fresh exact base/head pair for pre-merge preparation, using the same task-PR resolution as merge admission. */ - async remotePair(signal?: AbortSignal): Promise<{ base: string; head: string }> { + /** + * Fresh exact base/head pair for pre-merge preparation, using the same task-PR resolution as merge admission. `branch` + * is the task PR's head branch when the runner published it; only then may preparation push a rewritten head. + */ + async remotePair(signal?: AbortSignal): Promise<{ base: string; head: string; branch?: string }> { const resolved = this.#resolve(this.#attempt()); const remote = await this.gateway.inspect({ fresh: true, timeoutMs: 6_000, signal, ...(resolved.target ? { target: resolved.target } : {}) }); @@ -178,7 +181,7 @@ export class MergeCoordinator { const blockers = this.#publishedBlockers(remote, resolved); if (blockers.length) throw new GuardRefusal(blockers[0]!.message); } - return { base: remote.base, head: remote.head }; + return { base: remote.base, head: remote.head, ...(resolved.headBranch !== undefined ? { branch: resolved.headBranch } : {}) }; } async #status(view: ReviewView, fresh: boolean, signal?: AbortSignal): Promise<{ status: MergeStatus; resolved: ResolvedTarget }> { @@ -198,7 +201,9 @@ export class MergeCoordinator { if (this.service.store && this.service.config) { const task = this.service.store.getTask(this.service.config.identity); if (task.status !== 'merged' && !MERGEABLE_STATUSES.includes(task.status)) blockers.push({ code: 'task', message: `The task is ${task.status}; merge it from review.` }); - if (this.published?.requiresPreparation && view.expected.reviewVersion !== undefined + if (task.pushInProgress) blockers.push({ code: 'preparation', + message: 'A push of the rewritten head has not settled. Prepare the merge again to read its outcome.' }); + else if (this.published?.requiresPreparation && view.expected.reviewVersion !== undefined && !this.service.store.preMergeReady(this.service.config.identity, { stateVersion: task.stateVersion, reviewVersion: view.expected.reviewVersion, snapshotId: view.snapshot.id, base: view.snapshot.base, head: view.snapshot.head, diff --git a/runner/pre-merge.ts b/runner/pre-merge.ts index adb8bf8d..9c4e29c6 100644 --- a/runner/pre-merge.ts +++ b/runner/pre-merge.ts @@ -6,13 +6,29 @@ import { GuardRefusal, MERGEABLE_STATUSES, settleWith, type ShutdownCapability } import { MAX_REBASE_TIMEOUT_MS, MIN_REBASE_CLEANUP_TIMEOUT_MS, MIN_REBASE_TIMEOUT_MS, RebaseResourcesUnsettled, type GitRebaser } from './rebase.ts'; import type { ReviewService } from './review.ts'; -import type { PreMergeReadiness, RebaseMarker } from './store.ts'; +import { BranchPushRefused } from './branch-push.ts'; +import type { PreMergeReadiness, PushMarker, PushOutcome, RebaseMarker } from './store.ts'; -export interface RemotePair { base: string; head: string } +/** `branch` is the task PR's head branch, present when the runner published the PR. */ +export interface RemotePair { base: string; head: string; branch?: string } export interface PreMergeRemote { inspect(signal?: AbortSignal): Promise; fetch(pair: RemotePair, signal?: AbortSignal): Promise; + /** + * Make the task branch point at `to`, leased to exactly `from`. Throws `BranchPushRefused` when nothing was pushed + * because the branch holds something else; any other failure leaves the outcome unknown. + */ + push(input: { branch: string; from: string; to: string; beforePush: () => void }, signal?: AbortSignal): Promise; + /** The commit the task branch points at on GitHub, or null when it does not exist. */ + readBranch(branch: string, signal?: AbortSignal): Promise; } +/** One read of the branch, bounded as `#settlePush` says, settles a push whose outcome is unknown. */ +export const PUSH_SETTLEMENT_TIMEOUT_MS = 30_000; +/** + * GitHub's pull-request API reports a pushed head asynchronously. After a push, no new inspection starts once this long + * has passed; the last one may still take its own timeout to answer. + */ +export const PUSHED_HEAD_VISIBLE_TIMEOUT_MS = 30_000; export interface PreMergeAuthorization { /** Complete the last external authorization read before a later external operation. */ refresh(): Promise; @@ -28,12 +44,17 @@ type PreparedResult = PreMergeResult & { readiness?: PreMergeReadiness }; // D may spend 30 s on its first cleanup and 60 s retrying it; F may then spend 30 s releasing task storage. Keep all // of that ownership settlement inside the preparation's one overall deadline. export const COMMAND_CHECK_SETTLEMENT_RESERVE_MS = 120_000; -/** Git and GitHub subprocesses abort before the advertised operation deadline, leaving their bounded stop/drain time. */ +/** + * Git and GitHub subprocesses abort before the advertised operation deadline, leaving their stop/drain time. That time + * is bounded except in two cases, which are waited for, fail-closed: a process that stays in a subprocess's group after + * it exits, and, for Git calls that change something (fetch, update-ref, push), a process that left the group but holds + * its output (reads, ls-remote and cat-file, fail instead after the pipe drain). + */ export const PRE_MERGE_PROCESS_SETTLEMENT_RESERVE_MS = DEFAULT_PROCESS_SETTLEMENT_MS; /** Keep a full process-settlement window between live rebase expiry and its separately budgeted durable cleanup. */ const REBASE_CLEANUP_HANDOFF_RESERVE_MS = PRE_MERGE_PROCESS_SETTLEMENT_RESERVE_MS; -/** Production preparation before F6: refresh, rebase, re-review, then exact-head command checks. */ +/** Production preparation: refresh, rebase, re-review, push the rewritten head, then exact-head command checks. */ export class PreMergeCoordinator { readonly service: ReviewService; readonly runner: RunnerCoordinator; @@ -42,10 +63,13 @@ export class PreMergeCoordinator { readonly operationTimeoutMs: number; readonly processSettlementReserveMs: number; readonly commandSettlementReserveMs: number; + readonly pushVisibleTimeoutMs: number; readonly authorize?: (signal: AbortSignal) => Promise; readonly settle: (fn: () => T) => T; - #active: Promise | null = null; + #active: Promise | null = null; #abort: AbortController | null = null; + /** Aborted only by shutdown: a cancel must not stop the branch read that settles a push and closes the task. */ + readonly #shutdown = new AbortController(); #closing = false; #last: PreMergeResult | null = null; #lastBinding: { stateVersion: number; reviewVersion: number; snapshotId: string } | null = null; @@ -53,15 +77,17 @@ export class PreMergeCoordinator { constructor(service: ReviewService, runner: RunnerCoordinator, rebaser: GitRebaser, remote: PreMergeRemote, operationTimeoutMs = 10 * 60_000, capability?: ShutdownCapability, authorize?: (signal: AbortSignal) => Promise, - reserves: { processMs?: number; commandMs?: number } = {}) { + reserves: { processMs?: number; commandMs?: number; pushVisibleMs?: number } = {}) { if (!Number.isSafeInteger(operationTimeoutMs) || operationTimeoutMs < 1 || operationTimeoutMs > 60 * 60_000) throw new Error('Invalid pre-merge operation deadline.'); this.service = service; this.runner = runner; this.rebaser = rebaser; this.remote = remote; this.operationTimeoutMs = operationTimeoutMs; this.authorize = authorize; this.processSettlementReserveMs = reserves.processMs ?? PRE_MERGE_PROCESS_SETTLEMENT_RESERVE_MS; this.commandSettlementReserveMs = reserves.commandMs ?? COMMAND_CHECK_SETTLEMENT_RESERVE_MS; + this.pushVisibleTimeoutMs = reserves.pushVisibleMs ?? PUSHED_HEAD_VISIBLE_TIMEOUT_MS; if (!Number.isSafeInteger(this.processSettlementReserveMs) || this.processSettlementReserveMs < 0 - || !Number.isSafeInteger(this.commandSettlementReserveMs) || this.commandSettlementReserveMs < 0) + || !Number.isSafeInteger(this.commandSettlementReserveMs) || this.commandSettlementReserveMs < 0 + || !Number.isSafeInteger(this.pushVisibleTimeoutMs) || this.pushVisibleTimeoutMs < 0) throw new Error('Invalid pre-merge settlement reserve.'); this.settle = settleWith(capability); } @@ -136,7 +162,8 @@ export class PreMergeCoordinator { return active; } async close(): Promise { - this.#closing = true; this.#abort?.abort(new Error('Server shutdown.')); await this.#active; + this.#closing = true; this.#shutdown.abort(new Error('Server shutdown.')); this.#abort?.abort(new Error('Server shutdown.')); + await this.#active; } /** Cancel the task and promptly stop whichever rebase, remote read or command check this preparation owns. */ cancelTask(expectedStateVersion: number, actionId: string): 'closed' | 'stopping' { @@ -148,6 +175,47 @@ export class PreMergeCoordinator { return outcome; } + /** + * Settle a push whose outcome is unknown (a crash, shutdown or failure during the push) by reading the branch, then + * clear its marker. Returns null when no push is pending; throws when the branch could not be read. + */ + async settleInterruptedPush(): Promise { + const marker = this.service.store.getTask(this.service.config.identity).pushInProgress; + return marker ? (await this.#settlePush(marker)).outcome : null; + } + /** + * Settle a push left by a crash, shutdown or unreadable branch, as one tracked job that shutdown aborts and awaits: + * after startup recovery, and when a task is cancelled with no preparation running. While it runs, a preparation is + * refused as already running. Never throws: a failed read leaves the marker for the next preparation, and a server + * already closing or busy skips it. + */ + settlePendingPush(): Promise { + if (this.#closing || this.#active) return Promise.resolve(null); + const active = this.settleInterruptedPush().catch(() => null) + .finally(() => { if (this.#active === active) this.#active = null; }); + this.#active = active; + return active; + } + /** Clear the marker; returns the task state version just before and just after, read in the same turn. */ + #finishPush(marker: PushMarker): { before: number; after: number } { + const identity = this.service.config.identity, before = this.service.store.getTask(identity).stateVersion; + this.settle(() => this.service.store.finishPrePush(this.service.store.getTask(identity).planKey, marker.attemptId)); + return { before, after: this.service.store.getTask(identity).stateVersion }; + } + /** + * One branch read, all of it, Git's settlement after an abort included, within `limitMs` (at most + * PUSH_SETTLEMENT_TIMEOUT_MS): the read is aborted the process-settlement reserve before that, and does not start + * when less than the reserve is left. One case is not bounded: a process that stays in Git's group after Git exits + * is waited for, as for every Git call, since its group ID may be reused and it cannot be signalled safely. + */ + async #settlePush(marker: PushMarker, limitMs = PUSH_SETTLEMENT_TIMEOUT_MS): Promise<{ outcome: PushOutcome; before: number; after: number }> { + const readMs = Math.min(PUSH_SETTLEMENT_TIMEOUT_MS, limitMs) - this.processSettlementReserveMs; + if (readMs < 1) throw Object.assign(new Error('Too little time is left to read the branch and settle Git.'), { code: 'ETIMEDOUT' }); + const read = await this.remote.readBranch(marker.branch, AbortSignal.any([this.#shutdown.signal, AbortSignal.timeout(readMs)])); + const outcome: PushOutcome = read === marker.to ? 'pushed' : read === marker.from ? 'not-pushed' : 'moved'; + return { outcome, ...this.#finishPush(marker) }; + } + #refresh(pair: RemotePair, remaining: () => number, priorHead?: string) { const historyOptions = () => ({ maxDurationMs: Math.min(30_000, remaining()) }); const view = this.service.load(historyOptions()); @@ -201,6 +269,84 @@ export class PreMergeCoordinator { if (!this.settle(() => this.service.store.abortRebase(this.service.store.getTask(identity).planKey, marker.attemptId))) throw new GuardRefusal('This rebase attempt no longer owns its durable marker.'); } + /** + * Inspect the PR until GitHub reports a head other than the pre-push head, with backoff. No inspection starts after + * `pushVisibleTimeoutMs`, so the wait lasts at most that plus one inspection, within the preparation deadline. A PR + * still at the pre-push head after that fails the preparation without refreshing the review: the branch itself + * already holds another head. + */ + async #awaitHeadChange(before: RemotePair, signal: AbortSignal, remaining: () => number): Promise { + const until = performance.now() + this.pushVisibleTimeoutMs; + for (let delay = 250; ; delay = Math.min(delay * 2, 4_000)) { + const seen = await this.remote.inspect(signal); signal.throwIfAborted(); + // Only the head shows whether GitHub caught up: a base that moved meanwhile says nothing about the pushed head. + if (seen.head !== before.head) return seen; + const wait = Math.min(delay, until - performance.now(), remaining() - this.processSettlementReserveMs); + if (wait < 1) throw new GuardRefusal(`GitHub has not reported the pull request's new head yet; it still shows ${before.head}. Prepare the merge again shortly.`); + await new Promise((resolve, reject) => { + const timer = setTimeout(() => { signal.removeEventListener('abort', stop); resolve(); }, wait); + const stop = () => { clearTimeout(timer); reject(signal.reason); }; + signal.addEventListener('abort', stop, { once: true }); + }); + } + } + /** + * Push the reviewed, rewritten head over exactly the remote head it rewrote. The durable marker is written before the + * first external write and cleared only after a read of the branch settles the outcome; when that read is impossible + * (shutdown, or GitHub unreachable) the marker stays for the next preparation or startup. + */ + async #pushRewrite(remote: RemotePair, view: ReturnType, signal: AbortSignal, deadline: number, + authorize: () => Promise, + bind: (reviewed: { reviewVersion: number; snapshotId: string }, pair: RemotePair, stateVersion?: number) => void): + Promise<{ result: 'pushed' | 'refused' | 'moved'; stateVersion: number }> { + const identity = this.service.config.identity; + if (!remote.branch) throw new GuardRefusal('The pull request was not published by the runner, so its rewritten head cannot be pushed.'); + const reviewed = { revision: view.expected.revision, snapshotId: view.expected.snapshotId, + reviewVersion: view.expected.reviewVersion! }; + // Captured before the asynchronous authorization read: any task write that lands during it (a reassignment, a + // referenced-code change) must refuse the claim, not be adopted by it. + const admitted = this.service.store.getTask(identity).stateVersion; + const authorization = await authorize(); + signal.throwIfAborted(); + const marker = this.service.store.beginPrePush(identity, reviewed, admitted, + { branch: remote.branch, from: remote.head, to: view.snapshot.head }); + const binding = this.service.store.getTask(identity).stateVersion; + // The task state this push left, for binding a later failure: the version clearing the marker produced when nothing + // else wrote since the claim, else the claim's own. A write while the push or its read was in flight, such as a plan + // revision, must leave a later failure stale. + const owned = (cleared: { before: number; after: number }) => cleared.before === binding ? cleared.after : binding; + try { + await this.remote.push({ branch: marker.branch, from: marker.from, to: marker.to, beforePush: () => { + // The last local checks before the external write: trust, and no task or review change since the claim. + authorization?.validate(); + const task = this.service.store.getTask(identity); + if (task.stateVersion !== binding || task.cancelRequested !== null + || this.service.store.reviewVersion(identity) !== reviewed.reviewVersion + || this.service.store.getSnapshot(identity).id !== reviewed.snapshotId) + throw new GuardRefusal('The task or review changed before the rewritten head was pushed. Reload first.'); + signal.throwIfAborted(); + } }, signal); + } catch (error) { + if (error instanceof BranchPushRefused) return { result: 'refused', stateVersion: owned(this.#finishPush(marker)) }; + // Shutdown: no new GitHub read may start; startup settles the marker. Otherwise one read, within what is left of + // the preparation's deadline with Git's settlement reserve kept inside it (except for a process that stays in + // Git's group after Git exits, which is waited for; see #settlePush), settles it; a cancel does not stop that + // read. When the preparation's own deadline ended the push, less than that reserve is left, so no read starts and + // the marker stays, as for an unreadable branch. + let settled = binding, outcome: PushOutcome | null = null; + const left = Math.floor(deadline - performance.now()); + if (!this.#closing && left >= 1) { + try { const read = await this.#settlePush(marker, left); settled = owned(read); outcome = read.outcome; } + catch { /* The marker stays; the next preparation or startup reads the branch again. */ } + } + bind(reviewed, { base: view.snapshot.base, head: view.snapshot.head }, settled); + // The branch holds neither the pre-push head nor ours: someone else moved it while the push's outcome was unknown + // (for example, right after the push, before its read-back). That head is reviewed, as after a refusal. + if (outcome === 'moved') return { result: 'moved', stateVersion: settled }; + throw error; + } + return { result: 'pushed', stateVersion: owned(this.#finishPush(marker)) }; + } async #run(expected: { stateVersion: number; reviewVersion: number; snapshotId: string; actionId?: string }, signal: AbortSignal, deadline: number, onObserved: (value: { pair: RemotePair; binding: { stateVersion: number; reviewVersion: number; snapshotId: string } }) => void, onChecked: (checked: readonly string[]) => void): Promise { @@ -249,10 +395,10 @@ export class PreMergeCoordinator { // Rebase lifecycle writes advance task state without changing the review that admitted them. Bind a resulting // failure to that original review/snapshot while adopting only the task-state transition owned by the rebase. If a // read fails, the older binding remains safely stale. - const bindRebaseFailure = (reviewed: { reviewVersion: number; snapshotId: string }, pair: RemotePair) => { + const bindRebaseFailure = (reviewed: { reviewVersion: number; snapshotId: string }, pair: RemotePair, stateVersion?: number) => { try { onObserved({ pair, binding: { - stateVersion: this.service.store.getTask(identity).stateVersion, + stateVersion: stateVersion ?? this.service.store.getTask(identity).stateVersion, reviewVersion: reviewed.reviewVersion, snapshotId: reviewed.snapshotId, } }); } catch { /* An unbound failure is conservatively historical. */ } @@ -265,6 +411,18 @@ export class PreMergeCoordinator { throw new GuardRefusal('Runner cleanup is unresolved; restart and recover owned resources before preparing a merge.'); if (this.service.store.getTask(identity).rebaseInProgress !== null) throw new GuardRefusal('Rebase cleanup is unresolved; restart and recover the owned rebase before preparing a merge.'); + const pending = this.service.store.getTask(identity).pushInProgress; + if (pending) { + // Not this preparation's signal: a cancel must let the read finish, so the cleared marker can close the task. + try { await this.#settlePush(pending, remaining()); } + catch (error) { + signal.throwIfAborted(); + throw new GuardRefusal(`The outcome of an earlier push of the rewritten head is unknown, and the branch could not be read: ${error instanceof Error ? error.message : String(error)}`); + } + signal.throwIfAborted(); + // Settling advanced task state. The caller's binding is stale now; it reloads and starts again. + throw new GuardRefusal('An earlier push of the rewritten head was settled. Reload, then prepare the merge again.'); + } let view = load(), task = this.service.store.getTask(identity); if (task.stateVersion !== expected.stateVersion || view.expected.reviewVersion !== expected.reviewVersion || view.snapshot.id !== expected.snapshotId) throw new GuardRefusal('The review changed before preparation started. Reload first.'); @@ -282,10 +440,11 @@ export class PreMergeCoordinator { || this.service.store.getSnapshot(identity).id !== guard.snapshotId) throw new GuardRefusal('The task or review changed during preparation. Reload first.'); }; - const initial = await this.remote.inspect(signal); await this.remote.fetch(initial, signal); signal.throwIfAborted(); + let initial = await this.remote.inspect(signal); await this.remote.fetch(initial, signal); signal.throwIfAborted(); assertCurrent(expected); - // F6 will push the rewritten head. Until then, a retry must recognize the durable rewrite lineage instead of - // mistaking codeboost's still-remote predecessor for a collaborator push. + // A rewrite is pushed only once its review is clear. Until then (a review stop, or a failed or refused push), a retry + // must recognize the durable rewrite lineage instead of mistaking codeboost's still-remote predecessor for a + // collaborator push. const retainedRemoteHead = initial.head !== view.snapshot.head && this.service.store.isRewrittenHead(identity, initial.head, view.snapshot.head); if (initial.head !== view.snapshot.head && !retainedRemoteHead) { @@ -331,6 +490,38 @@ export class PreMergeCoordinator { const blocker = this.#reviewBlocker(view); if (blocker) return { state: 'review-required', base: view.snapshot.base, head: view.snapshot.head, checked: [], reason: blocker }; + if (initial.head !== view.snapshot.head) { + const { result: pushed, stateVersion: pushState } = await this.#pushRewrite(initial, view, signal, deadline, authorize, + bindRebaseFailure); + // Bind any later failure to the state the push (or refusal) left, so it is reported as current, not stale. + bindRebaseFailure({ reviewVersion: view.expected.reviewVersion!, snapshotId: view.snapshot.id }, + { base: view.snapshot.base, head: view.snapshot.head }, pushState); + // Either way the branch no longer holds the pre-push head: ours after a push, someone else's after a refusal. + // GitHub's PR API reports that asynchronously, so wait for it; its lag must not read as the old head. + const seen = await this.#awaitHeadChange(initial, signal, remaining); + if (pushed !== 'pushed' || seen.head !== view.snapshot.head || seen.base !== initial.base) { + // As on the pushed path below: a plan revision or review edit that landed during the wait (or the fetch) must not + // have this preparation's remote observation recorded into it. + const binding = { stateVersion: pushState, reviewVersion: view.expected.reviewVersion!, snapshotId: view.snapshot.id }; + assertCurrent(binding); + const moved = seen; await this.remote.fetch(moved, signal); signal.throwIfAborted(); + assertCurrent(binding); + // Traced from the reviewed head and its rewrite lineage, nearest first. That lineage holds the pushed head, every + // earlier rewrite codeboost may already have pushed, and the original head a collaborator may have built on. + view = track(this.#refresh(moved, reviewBudget, view.snapshot.head)); + return { state: 'review-required', base: view.snapshot.base, head: view.snapshot.head, checked: [], + reason: pushed === 'pushed' + ? 'The pull request moved after the rewritten head was pushed; attribution and approvals were refreshed.' + : pushed === 'moved' + ? 'The pull request branch moved while the rewritten head was being pushed; attribution and approvals were refreshed.' + : 'The pull request branch moved before the rewritten head could be pushed; attribution and approvals were refreshed.' }; + } + // Nothing but this push may have changed the task meanwhile: a plan revision or review edit applied while it was in + // flight would otherwise be adopted by the reload below and run checks under approvals it made stale. + assertCurrent({ stateVersion: pushState, reviewVersion: view.expected.reviewVersion!, snapshotId: view.snapshot.id }); + initial = seen; + view = load(); + } const checked: string[] = []; for (const item of view.items.filter(candidate => candidate.acceptance.some(check => check.type === 'cmd'))) { signal.throwIfAborted(); diff --git a/runner/production.ts b/runner/production.ts index 6280032d..2d307e60 100644 --- a/runner/production.ts +++ b/runner/production.ts @@ -317,8 +317,8 @@ export async function setUpRunner(o: { service: ReviewService; capability: Shutd runnerCommandLauncher({ imageId, runnerRoot: config.root, runnerOwner }), review.allowedCommands ?? [], runnerOwner)); const github = review.github; // Never a `url` here (#101 review, finding 6): the push goes to the configured repository on GH_HOST, from the runner's - // own repository. Only commits the ledger records as codeboost's may be overwritten. The F3 rebase foundation can - // write mapped foreign entries, but #22 does not wire their external push until its later durable-action slice. + // own repository. Publishing overwrites only commits the ledger records as codeboost's; pre-merge preparation overwrites + // only the exact head its durable rebase rewrote (F6 of #22). const pusher = new GitBranchPusher({ repository, repositoryId: identity.repositoryId, remote: github.repository, env: o.env as NodeJS.ProcessEnv, ownedCommits: requested => service.store.getLedger(requested).filter(entry => entry.origin === 'owned').map(entry => entry.sha) }); // The same environment as the push, so the PR calls go to the same GH_HOST with the same credentials. @@ -333,6 +333,12 @@ export async function setUpRunner(o: { service: ReviewService; capability: Shutd return new PreMergeCoordinator(service, runner, rebaser, { inspect, fetch: (pair, signal) => pusher.fetchCommits([pair.base, pair.head], signal), + // Leased to exactly the inspected head the rebase rewrote; the coordinator's durable marker brackets the write. + // The guard is returned as the boundary's finalizer, which the pusher runs synchronously as Git starts: nothing can + // land between the last check and the write. + push: (input, signal) => pusher.push(identity, { head: input.to, branch: input.branch, expected: input.from, + beforePush: async () => input.beforePush }, signal), + readBranch: (branch, signal) => pusher.readBranch(identity, branch, signal), }, undefined, o.capability, authorize); }; return { deps, sources, findings, recovery, publisher, env, preMerge }; diff --git a/runner/store.ts b/runner/store.ts index 3e359b50..632801ef 100644 --- a/runner/store.ts +++ b/runner/store.ts @@ -92,7 +92,7 @@ export function preMergeActionResponse(result: PreMergeActionResult) { } export interface TaskRecord { planKey: string; status: TaskStatus; stateVersion: number; contextGeneration: number; assignmentId: string; referencedCodeHash: string; - currentAttemptId: string | null; requeuePending: boolean; cancelRequested: string | null; rebaseInProgress: unknown; budgetDeadline: number | null; + currentAttemptId: string | null; requeuePending: boolean; cancelRequested: string | null; rebaseInProgress: unknown; pushInProgress: PushMarker | null; budgetDeadline: number | null; createdAt: string; updatedAt: string; } export interface RebaseMarker { @@ -111,6 +111,24 @@ export interface RebaseMarker { conflict: { attemptId: string; allocationId: string; networkAllocationId: string; source: string } | null; processGroup: { pgid: number; startedAt: number; identity: string | null } | 'spawning' | 'unsettled' | null; } +/** + * A pre-merge push of the rewritten head (F6 of #22), recorded before the push's first external write. While it exists, + * GitHub may or may not hold `to`: nothing irreversible starts until a read of the branch settles the outcome. + */ +export interface PushMarker { + attemptId: string; + /** The task's PR branch on GitHub (`codeboost/...`). */ + branch: string; + /** The exact remote head read before the push; the push is leased to it. */ + from: string; + /** The reviewed, rewritten head being pushed. */ + to: string; + /** The snapshot whose head is `to`. */ + snapshotId: string; + startedAt: number; +} +/** What a read of the branch showed after a push: `to`, still `from`, or anything else. */ +export type PushOutcome = 'pushed' | 'not-pushed' | 'moved'; export interface AttemptRecord { id: string; kind: AttemptKind; phase: string; item: string | null; state: AttemptState; context: InvocationContext; deadline: number; firstReason: FirstReason | null; stopReason: StopReason | null; exitCode: number | null; signal: string | null; result: unknown; @@ -188,8 +206,8 @@ export class Store { this.#db.exec('PRAGMA foreign_keys=ON; PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;'); this.#transaction(() => { const version = this.#get('PRAGMA user_version')!.user_version as number; - if (![0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17].includes(version)) throw new Error('Unsupported store schema version.'); - if (version === 17) return; + if (![0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18].includes(version)) throw new Error('Unsupported store schema version.'); + if (version === 18) return; if (version === 0) this.#db.exec(` CREATE TABLE plans (key TEXT PRIMARY KEY, issue INTEGER NOT NULL, revision INTEGER NOT NULL, snapshot_id TEXT); CREATE TABLE revisions (key TEXT NOT NULL REFERENCES plans(key), revision INTEGER NOT NULL, data TEXT NOT NULL, PRIMARY KEY(key,revision)); @@ -305,6 +323,12 @@ export class Store { this.#db.exec('ALTER TABLE attempts ADD COLUMN prompt_comments TEXT'); this.#db.exec('PRAGMA user_version=17'); } + // The durable marker of a pre-merge push of the rewritten head (F6 of #22). + if (version < 18) { + if (!this.#db.prepare('PRAGMA table_info(tasks)').all().some(column => column.name === 'push_in_progress')) + this.#db.exec('ALTER TABLE tasks ADD COLUMN push_in_progress TEXT'); + this.#db.exec('PRAGMA user_version=18'); + } }); } catch (error) { this.#db.close(); throw error; } } @@ -533,7 +557,7 @@ export class Store { const task = this.#task(key); if (this.#closed(task.status as TaskStatus)) throw new GuardRefusal(`The task is ${task.status}; it cannot be merged.`); if (task.cancel_requested !== null) throw new GuardRefusal('The task is being cancelled; it cannot be merged.'); - if (task.rebase_in_progress !== null) throw new GuardRefusal('A rebase is in progress for this task.'); + this.#assertNoRewriteInFlight(task); if (!MERGEABLE_STATUSES.includes(task.status as TaskStatus)) throw new GuardRefusal(`The task is ${task.status}; merge it from review.`); if (this.#activeAttempt(key)) throw new GuardRefusal('An attempt is still active for this task; it cannot be merged.'); if (expectedTaskStateVersion !== null && task.state_version !== expectedTaskStateVersion) throw new GuardRefusal('Stale task state. Reload before writing.'); @@ -760,6 +784,7 @@ export class Store { this.#expect(key, expected); this.#assertContextWritable(key); if (this.#task(key).rebase_in_progress !== null) throw new GuardRefusal('A claimed rebase is in progress; wait for it to settle.'); + if (this.#task(key).push_in_progress !== null) throw new GuardRefusal('A push of the rewritten head is in progress; wait for it to settle.'); return this.#applyRebase(identity, key, base, head, mappings); }); } @@ -791,6 +816,7 @@ export class Store { if (this.#activeAttempt(key)) throw new GuardRefusal('An attempt is still active for this task; it cannot be rebased.'); if (this.#activeMerge(key)) throw new GuardRefusal('A merge is in progress; wait for its outcome.'); if (task.rebase_in_progress !== null) throw new GuardRefusal('A rebase is already in progress for this task.'); + if (task.push_in_progress !== null) throw new GuardRefusal('A push of the rewritten head is in progress for this task.'); const snapshot = this.getSnapshot(identity); if (snapshot.base !== input.oldBase || snapshot.head !== input.oldHead) throw new GuardRefusal('The reviewed base or head changed before the rebase started.'); @@ -966,6 +992,62 @@ export class Store { return snapshot; }); } + /** + * Claim the push of the reviewed, rewritten head before its first external write. `from` is the exact remote head + * read for the lease; it must be a durable rewrite predecessor of `to`, and `to` must be the current snapshot head. + */ + beginPrePush(identity: PlanIdentity, expected: ReviewState & { reviewVersion: number }, expectedTaskStateVersion: number, + input: { branch: string; from: string; to: string; attemptId?: string; startedAt?: number }): PushMarker { + sha(input.from); sha(input.to); + if (!/^codeboost\/[a-z0-9]+(?:-[a-z0-9]+)*$/.test(input.branch) || input.branch.length > 255) + throw new Error('Only a codeboost/ task branch can be pushed.'); + if (!Number.isSafeInteger(expected.reviewVersion) || expected.reviewVersion < 0) throw new Error('A current review version is required for pushing.'); + if (!Number.isSafeInteger(expectedTaskStateVersion) || expectedTaskStateVersion < 0) throw new Error('Invalid expected task state version.'); + const attemptId = input.attemptId ?? randomUUID(), startedAt = input.startedAt ?? Date.now(); + assertUuidV4(attemptId, 'Push attempt ID'); + if (!Number.isSafeInteger(startedAt) || startedAt < 0) throw new Error('Invalid push start time.'); + const key = identityKey(identity); + return this.#transaction(() => { + this.#expect(key, expected); + const task = this.#task(key); + if (task.state_version !== expectedTaskStateVersion) throw new GuardRefusal('Stale task state. Reload before writing.'); + if (this.#closed(task.status as TaskStatus)) throw new GuardRefusal(`The task is ${task.status}; it cannot be pushed.`); + if (!MERGEABLE_STATUSES.includes(task.status as TaskStatus)) throw new GuardRefusal(`The task is ${task.status}; push it from review.`); + if (task.cancel_requested !== null) throw new GuardRefusal('The task is being cancelled; it cannot be pushed.'); + if (this.#activeAttempt(key)) throw new GuardRefusal('An attempt is still active for this task; it cannot be pushed.'); + if (this.#activeMerge(key)) throw new GuardRefusal('A merge is in progress; wait for its outcome.'); + this.#assertNoRewriteInFlight(task); + const snapshot = this.getSnapshot(identity); + if (snapshot.head !== input.to) throw new GuardRefusal('The reviewed head changed before the push started.'); + if (!this.isRewrittenHead(identity, input.from, input.to)) + throw new GuardRefusal('The pull request head is not a rewrite predecessor of the reviewed head.'); + const marker: PushMarker = { attemptId, branch: input.branch, from: input.from, to: input.to, snapshotId: snapshot.id, startedAt }; + this.#run('UPDATE tasks SET push_in_progress=? WHERE plan_key=?', encode(marker), key); + this.#touch(key); + return marker; + }); + } + /** + * Clear the matching push once a read of the branch settled its outcome. A cancel requested while the push ran closes + * the task now, as a settled rebase does. + */ + finishPrePush(planKey: string, attemptId: string): boolean { + assertUuidV4(attemptId, 'Push attempt ID'); + return this.#transaction(() => { + const task = this.#task(planKey); + const encoded = task.push_in_progress as string | null | undefined; + if (encoded === null || encoded === undefined || decode(encoded).attemptId !== attemptId) return false; + if (this.#run('UPDATE tasks SET push_in_progress=NULL WHERE plan_key=? AND push_in_progress=?', planKey, encoded).changes !== 1) return false; + if (task.cancel_requested !== null && !this.#closed(task.status as TaskStatus)) this.#closeTask(planKey, 'cancelled', task.cancel_requested as string); + else this.#touch(planKey); + return true; + }); + } + /** Pushes whose outcome no read has settled yet (a crash or shutdown during the push). */ + pushesInProgress(): { planKey: string; marker: PushMarker }[] { + return this.#db.prepare('SELECT plan_key, push_in_progress FROM tasks WHERE push_in_progress IS NOT NULL').all() + .map(row => ({ planKey: row.plan_key as string, marker: decode(row.push_in_progress) })); + } /** Clear only the matching rebase, after the live operation settles or recovery removes its resources. */ abortRebase(planKey: string, attemptId: string): boolean { assertUuidV4(attemptId, 'Rebase attempt ID'); @@ -1389,7 +1471,7 @@ export class Store { status TEXT NOT NULL CHECK (status IN (${list(TASK_STATUSES)})), state_version INTEGER NOT NULL DEFAULT 0, context_generation INTEGER NOT NULL DEFAULT 0, assignment_id TEXT NOT NULL, referenced_code_hash TEXT NOT NULL, - current_attempt_id TEXT, requeue_pending INTEGER NOT NULL DEFAULT 0, cancel_requested TEXT, rebase_in_progress TEXT, + current_attempt_id TEXT, requeue_pending INTEGER NOT NULL DEFAULT 0, cancel_requested TEXT, rebase_in_progress TEXT, push_in_progress TEXT, budget_deadline INTEGER, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, FOREIGN KEY (plan_key, current_attempt_id) REFERENCES attempts(plan_key, id)); CREATE TABLE IF NOT EXISTS attempts ( @@ -1436,6 +1518,7 @@ export class Store { referencedCodeHash: row.referenced_code_hash as string, currentAttemptId: row.current_attempt_id as string | null, requeuePending: row.requeue_pending === 1, cancelRequested: row.cancel_requested as string | null, rebaseInProgress: row.rebase_in_progress === null ? null : decode(row.rebase_in_progress), + pushInProgress: row.push_in_progress === null || row.push_in_progress === undefined ? null : decode(row.push_in_progress), budgetDeadline: row.budget_deadline as number | null, createdAt: row.created_at as string, updatedAt: row.updated_at as string, }; } @@ -1677,7 +1760,7 @@ export class Store { if (task.state_version !== expectedStateVersion) throw new GuardRefusal('Stale task state. Reload before writing.'); if (this.#closed(task.status as TaskStatus)) throw new GuardRefusal('A closed task never changes.'); if (this.#activeMerge(key)) throw new GuardRefusal('A merge is in progress; wait for its outcome.'); - if (task.rebase_in_progress !== null) throw new GuardRefusal('A rebase is in progress for this task.'); + this.#assertNoRewriteInFlight(task); this.#run('UPDATE tasks SET assignment_id=?, referenced_code_hash=? WHERE plan_key=?', assignmentId, referencedCodeHash, key); this.#bumpContext(key); }); @@ -1692,11 +1775,17 @@ export class Store { if (this.#closed(task.status)) throw new GuardRefusal('A closed task never changes.'); if (this.#activeAttempt(key)) throw new GuardRefusal('An attempt is still active for this task.'); if (this.#activeMerge(key)) throw new GuardRefusal('A merge is in progress; wait for its outcome.'); - if (task.rebase_in_progress !== null) throw new GuardRefusal('A rebase is in progress for this task.'); + this.#assertNoRewriteInFlight(task); this.#run('UPDATE tasks SET status=? WHERE plan_key=?', to, key); this.#touch(key); }); } + /** A rebase still owns local resources, or a push of its result may still move the PR head on GitHub. */ + #assertNoRewriteInFlight(task: Record): void { + if (task.rebase_in_progress !== null) throw new GuardRefusal('A rebase is in progress for this task.'); + if (task.push_in_progress !== null && task.push_in_progress !== undefined) + throw new GuardRefusal('A push of the rewritten head is in progress for this task.'); + } /** GitHub may still merge the reviewed head while the latest merge attempt is submitting or queued. */ #activeMerge(key: string): boolean { const row = this.#get('SELECT data FROM merge_attempts WHERE key=? ORDER BY rowid DESC LIMIT 1', key); @@ -1736,7 +1825,7 @@ export class Store { if (task.cancel_requested !== null) throw new GuardRefusal('The task is being cancelled.'); if (this.#activeAttempt(key)) throw new GuardRefusal('An attempt is already active for this task.'); if (this.#activeMerge(key)) throw new GuardRefusal('A merge is in progress; wait for its outcome.'); - if (task.rebase_in_progress !== null) throw new GuardRefusal('A rebase is in progress for this task.'); + this.#assertNoRewriteInFlight(task); // The code-writing task budget (null until execution starts) ends execution admission. Review checks have their // own exact-head deadline and must remain runnable after implementation ends. if (!reviewCheck && task.budget_deadline !== null && (task.budget_deadline as number) <= now) @@ -1870,7 +1959,7 @@ export class Store { if (merge && (merge.state === 'submitting' || merge.state === 'queued')) throw new GuardRefusal('A merge is in progress. Cancel the task after it finishes or fails.'); const active = this.#activeAttempt(key); - if (task.rebase_in_progress !== null) { + if (task.rebase_in_progress !== null || task.push_in_progress !== null) { if (task.cancel_requested !== null) throw new GuardRefusal('The task is already being cancelled.'); this.#run('UPDATE tasks SET cancel_requested=? WHERE plan_key=?', actionId, key); this.#touch(key); diff --git a/test/agent-process-group.test.ts b/test/agent-process-group.test.ts index 1a4edfb1..0cb718ff 100644 --- a/test/agent-process-group.test.ts +++ b/test/agent-process-group.test.ts @@ -260,6 +260,34 @@ describe('runInProcessGroup', () => { } }, 30_000); + it('with boundEscapedOutput, returns failed promptly while a process outside the group holds its output pipe', async () => { + const dir = mkdtempSync(join(tmpdir(), 'escaped-bounded-')), pidFile = join(dir, 'pid'); + const escape = `const c = require('node:child_process').spawn('sleep', ['60'], { detached: true, ` + + `stdio: ['ignore', 'inherit', 'inherit'] }); require('node:fs').writeFileSync(${JSON.stringify(pidFile)}, ` + + `String(c.pid)); c.unref(); console.log('leader done');`; + try { + const began = performance.now(); + const outcome = await runInProcessGroup(process.execPath, ['-e', escape], { env, timeoutMs: 30_000, boundEscapedOutput: true }); + expect(outcome).toMatchObject({ status: null, stdout: 'leader done\n' }); + expect((outcome.error as NodeJS.ErrnoException).code).toBe('ESTDIOHELD'); + expect(performance.now() - began).toBeLessThan(10_000); + } finally { + try { process.kill(Number(readFileSync(pidFile, 'utf8')), 'SIGKILL'); } catch { /* already gone */ } + rmSync(dir, { recursive: true, force: true }); + } + }, 30_000); + + it('with boundEscapedOutput, still waits for a member that stays in the group after the leader exits', async () => { + // The leader leaves a child in its own group (not detached, no inherited output) that outlives the 10 s bounded + // group drain an unsettled return would stop at. + const linger = `require('node:child_process').spawn('sleep', ['12'], { stdio: 'ignore' }).unref(); console.log('leader done');`; + const began = performance.now(); + const outcome = await runInProcessGroup(process.execPath, ['-e', linger], { env, timeoutMs: 30_000, boundEscapedOutput: true }); + expect(performance.now() - began).toBeGreaterThanOrEqual(11_500); + expect(outcome).toMatchObject({ status: 0, stdout: 'leader done\n' }); + expect(outcome.error).toBeUndefined(); + }, 60_000); + it('does not release an untracked caller while a process outside the group holds its output pipe', async () => { const dir = mkdtempSync(join(tmpdir(), 'escaped-untracked-')), pidFile = join(dir, 'pid'); const escape = `const c = require('node:child_process').spawn('sleep', ['60'], { detached: true, ` diff --git a/test/merge-task-pr.test.ts b/test/merge-task-pr.test.ts index 89f44a90..4fc4bed5 100644 --- a/test/merge-task-pr.test.ts +++ b/test/merge-task-pr.test.ts @@ -313,6 +313,25 @@ describe('the merge gate with a runner block (#121)', () => { expect(gh.merged).toEqual([]); }); + it('gives pre-merge preparation the task PR\'s branch, the only one it may push a rewritten head to', async () => { + const { store } = published(), gh = github(store); + const merges = new MergeCoordinator(service(store), gh.client, undefined, undefined, PUBLISHED); + await expect(merges.remotePair()).resolves.toEqual({ base: oid(1), head: oid(2), branch: BRANCH }); + }); + + it('names an unsettled push of the rewritten head as the preparation blocker', async () => { + const { store } = published(), gh = github(store); + const reviewed = () => ({ revision: 1, snapshotId: store.getSnapshot(identity).id, reviewVersion: store.reviewVersion(identity) }); + store.recordRebase(identity, reviewed(), oid(10), oid(12), [{ oldSha: oid(2), newSha: oid(12) }]); + store.beginPrePush(identity, reviewed(), store.getTask(identity).stateVersion, { branch: BRANCH, from: oid(2), to: oid(12) }); + const merges = new MergeCoordinator(service(store), gh.client, undefined, undefined, PREPARED_PUBLISHED); + const blockers = (await merges.status()).blockers.filter(blocker => blocker.code === 'preparation'); + expect(blockers).toEqual([{ code: 'preparation', + message: 'A push of the rewritten head has not settled. Prepare the merge again to read its outcome.' }]); + await expect(merges.merge('review-token')).rejects.toThrow(); + expect(gh.merged).toEqual([]); + }); + it('refuses pre-merge preparation for a pull request that is no longer open', async () => { const { store } = published(); const gh = github(store, () => ({ pullRequestState: 'CLOSED' })); diff --git a/test/planning-drafts.test.ts b/test/planning-drafts.test.ts index e7939170..56290fd1 100644 --- a/test/planning-drafts.test.ts +++ b/test/planning-drafts.test.ts @@ -170,7 +170,7 @@ it.each([9, 10])('migrates a v%i database to the current schema, reading its exi const migrated = open(); expect(migrated.getSuggestions(identity, id)).toMatchObject({ mode: 'suggest', state: 'pending' }); const db = new DatabaseSync(path); - try { expect(db.prepare('PRAGMA user_version').get()).toEqual({ user_version: 17 }); } finally { db.close(); } + try { expect(db.prepare('PRAGMA user_version').get()).toEqual({ user_version: 18 }); } finally { db.close(); } }); it('adds checkpoint bindings to v14 planning requests without losing their state', () => { const { store, path, open } = fixture(), id = store.beginSuggestions(identity, state(store), 'draft'); @@ -181,7 +181,7 @@ it('adds checkpoint bindings to v14 planning requests without losing their state expect(migrated.getDraft(identity, id).state).toBe('pending'); const db = new DatabaseSync(path); try { - expect(db.prepare('PRAGMA user_version').get()).toEqual({ user_version: 17 }); + expect(db.prepare('PRAGMA user_version').get()).toEqual({ user_version: 18 }); expect(db.prepare('PRAGMA table_info(requests)').all().some(column => column.name === 'continuation')).toBe(true); } finally { db.close(); } }); diff --git a/test/pre-merge.test.ts b/test/pre-merge.test.ts index e38b4a74..6be84a5e 100644 --- a/test/pre-merge.test.ts +++ b/test/pre-merge.test.ts @@ -14,10 +14,15 @@ import { RunnerCoordinator } from '../runner/coordinator.ts'; import type { InvocationHandle, InvocationResult } from '../agents/contract.ts'; import type { TaskWorkspace, WorkspaceRef } from '../runner/execution.ts'; import { MIN_REBASE_CLEANUP_TIMEOUT_MS, MIN_REBASE_TIMEOUT_MS, RebaseResourcesUnsettled } from '../runner/rebase.ts'; +import { BranchPushRefused } from '../runner/branch-push.ts'; const roots: string[] = [], services: ReviewService[] = []; +const BRANCH = 'codeboost/task-1'; +// A remote fake for tests whose path must never push or read the branch. +const unusedPush = { push: async () => { throw new Error('No push expected.'); }, + readBranch: async () => { throw new Error('No branch read expected.'); } }; vi.setConfig({ testTimeout: 15_000 }); -afterEach(() => { for (const service of services.splice(0)) service.close(); for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); }); +afterEach(() => { vi.restoreAllMocks(); for (const service of services.splice(0)) service.close(); for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); }); const markRunnerOwned = (service: ReviewService) => { const identity = service.config.identity; @@ -44,6 +49,10 @@ const markRunnerUnchanged = (service: ReviewService) => { async function rebaseFixture(rebasedText: string, commandExit?: number, options: { duringRebase?: (service: ReviewService, coordinator: PreMergeCoordinator, timeoutMs?: number) => void; duringFinalInspect?: (service: ReviewService, coordinator: PreMergeCoordinator, read: number) => void; + /** Runs on each inspection that still shows the pre-push pull request (the wait for the new head to show). */ + duringVisibilityWait?: (service: ReviewService) => void; + /** Runs inside each fetch after a push or refusal, while the fetch is in flight. */ + duringFetchAfterPush?: (service: ReviewService) => void; duringAuthorize?: (service: ReviewService, call: number) => { base: string; head: string } | void; closeDuringCommand?: boolean; closeAfterCommandAdmission?: boolean; @@ -56,7 +65,29 @@ async function rebaseFixture(rebasedText: string, commandExit?: number, options: runnerCommitted?: boolean; operationTimeoutMs?: number; authorize?: (signal: AbortSignal) => Promise; - reserves?: { processMs?: number; commandMs?: number }; + reserves?: { processMs?: number; commandMs?: number; pushVisibleMs?: number }; + /** The PR API keeps reporting the pre-push head for this many inspections after a push. */ + pushVisibleAfter?: number; + /** The base branch moves as the push lands: lagging and later inspections report this base. */ + movedBaseAfterPush?: (service: ReviewService) => string; + /** The push lands on the branch before it fails. */ + landBeforeFailure?: boolean; + /** A later rebase's result (the first rebase always yields the fixture's `rebased`); `undefined` keeps the default. */ + rebaseResult?: (service: ReviewService, input: { oldHead: string; onto: string }, run: number) => string | undefined; + /** Branch reads never answer; only their abort signal ends them. */ + branchReadHangs?: boolean; + /** Remote fetches fail once codeboost has pushed. */ + fetchFailsAfterPush?: boolean; + /** A collaborator pushes this commit to the branch just before codeboost's push, which is then refused. */ + pushedFirst?: (service: ReviewService) => string; + /** Runs once the push has landed and may move the branch again, as a collaborator push would. */ + afterPush?: (service: ReviewService, pushed: string) => string; + /** The push lands, but its read-back (as GitBranchPusher's) sees another head and throws. */ + readBackFails?: boolean; + pushFailure?: (service: ReviewService, input: { from: string; to: string }) => Error; + branchRead?: (current: string, signal?: AbortSignal) => string | null | Promise; + /** Runs while the push reads the remote, before its final local checks (the pusher's `beforePush`). */ + duringPushRead?: (service: ReviewService, coordinator: PreMergeCoordinator) => void; } = {}) { const root = mkdtempSync(join(tmpdir(), 'codeboost-pre-merge-base-')); roots.push(root); const repository = join(root, 'repo'); fixtureGit(root, 'init', '-q', '-b', 'main', repository); @@ -90,7 +121,7 @@ async function rebaseFixture(rebasedText: string, commandExit?: number, options: let coordinator!: PreMergeCoordinator, rebaseRuns = 0; let rebaseAborts = 0; const rebaseBudgets: Array = [], abortBudgets: Array = []; - const rebaser = { run: async (input: { attemptId: string; signal: AbortSignal; timeoutMs?: number }) => { + const rebaser = { run: async (input: { attemptId: string; signal: AbortSignal; timeoutMs?: number; oldHead: string; onto: string }) => { rebaseRuns++; rebaseBudgets.push(input.timeoutMs); options.duringRebase?.(service, coordinator, input.timeoutMs); @@ -98,6 +129,12 @@ async function rebaseFixture(rebasedText: string, commandExit?: number, options: if (options.unsettledRebase) throw new RebaseResourcesUnsettled('Conflict resources remain owned.'); if (options.rebaseFailure) throw options.rebaseFailure; const planKey = service.store.getTask(identity).planKey; + const custom = options.rebaseResult?.(service, input, rebaseRuns); + if (custom) { + service.store.prepareRebaseResult(planKey, input.attemptId, custom, [custom]); + service.store.completeRebaseResult(planKey, input.attemptId); + return { oldHead: input.oldHead, base: input.onto, head: custom, mappings: [{ oldSha: input.oldHead, newSha: custom }], resolvedConflicts: [] }; + } service.store.prepareRebaseResult(planKey, input.attemptId, rebased, [rebased]); service.store.completeRebaseResult(planKey, input.attemptId); return { oldHead: head, base: onto, head: rebased, @@ -159,7 +196,9 @@ async function rebaseFixture(rebasedText: string, commandExit?: number, options: settled: Promise.resolve({ attemptId: input.attemptId, context: input.context, exitCode: commandExit ?? 0, signal: null, stdout: '', stderr: commandExit ? 'failed' : '' }), cancel() {} } satisfies InvocationHandle; }, context.allowedCommands, 'a'.repeat(32))); - let remoteReads = 0, remotePair = { base: onto, head }; + let remoteReads = 0, remotePair: { base: string; head: string; branch?: string } = { base: onto, head, branch: BRANCH }; + const pushes: { branch: string; from: string; to: string }[] = []; + let lagging = 0, lagged = remotePair, branchReads = 0, fetchesAfterPush = 0; let authorizationChecks = 0; const authorize = options.authorize ?? (options.duringAuthorize ? async () => ({ refresh: async () => undefined, @@ -168,10 +207,46 @@ async function rebaseFixture(rebasedText: string, commandExit?: number, options: coordinator = new PreMergeCoordinator(service, runner, rebaser, { inspect: async () => { + if (lagging > 0) { lagging--; options.duringVisibilityWait?.(service); return lagged; } remoteReads++; - if (remoteReads > 1) options.duringFinalInspect?.(service, coordinator, remoteReads); + // Read 2 is the post-push visibility check; hooks keep counting the later inspections as before the push existed. + if (remoteReads > 2) options.duringFinalInspect?.(service, coordinator, remoteReads - 1); return remotePair; - }, fetch: async () => undefined }, options.operationTimeoutMs, undefined, authorize, options.reserves); + }, fetch: async () => { + if (pushes.length) fetchesAfterPush++; + if (pushes.length) options.duringFetchAfterPush?.(service); + if (options.fetchFailsAfterPush && pushes.length) throw new Error('fetch failed after the push'); + }, + push: async input => { + pushes.push({ branch: input.branch, from: input.from, to: input.to }); + options.duringPushRead?.(service, coordinator); + input.beforePush(); + if (options.landBeforeFailure && remotePair.head === input.from) { lagged = remotePair; remotePair = { ...remotePair, head: input.to }; } + if (options.pushedFirst) { + lagged = remotePair; lagging = options.pushVisibleAfter ?? 0; + remotePair = { ...remotePair, head: options.pushedFirst(service) }; + } + if (options.pushFailure) throw options.pushFailure(service, input); + if (remotePair.head !== input.from) throw new BranchPushRefused('moved'); + lagged = remotePair; lagging = options.pushVisibleAfter ?? 0; + remotePair = { ...remotePair, head: input.to }; + if (options.movedBaseAfterPush) { + const base = options.movedBaseAfterPush(service); + lagged = { ...lagged, base }; remotePair = { ...remotePair, base }; + } + if (options.afterPush) remotePair = { ...remotePair, head: options.afterPush(service, input.to) }; + if (options.readBackFails && remotePair.head !== input.to) + throw new Error(`After the push, the branch ${input.branch} is at ${remotePair.head}, not ${input.to}.`); + }, + // A real branch read is aborted by its signal; so is this one, so a cancel that stopped settlement would show. + readBranch: async (_branch, signal) => { + branchReads++; + signal?.throwIfAborted(); + if (options.branchReadHangs) return new Promise((_resolve, reject) => + signal?.addEventListener('abort', () => reject(signal.reason), { once: true })); + return options.branchRead ? options.branchRead(remotePair.head, signal) : remotePair.head; + }, + }, options.operationTimeoutMs, undefined, authorize, options.reserves); if (options.closeAfterCommandAdmission) { const start = runner.start.bind(runner); runner.start = ((...args: Parameters) => { @@ -183,7 +258,13 @@ async function rebaseFixture(rebasedText: string, commandExit?: number, options: const task = service.store.getTask(identity); const result = await coordinator.start({ stateVersion: task.stateVersion, reviewVersion: view.expected.reviewVersion!, snapshotId: view.snapshot.id, base: view.snapshot.base, head: view.snapshot.head }); - return { service, coordinator, runner, result, rebased, checkedHeads, cancelReasons, + return { service, coordinator, runner, result, rebased, checkedHeads, cancelReasons, pushes, remote: () => remotePair, + branchReads: () => branchReads, fetchesAfterPush: () => fetchesAfterPush, + /** The branch moves (as GitHub's Git data shows it); the PR API reports `shown` for the next `reads` inspections. */ + moveRemote: (pair: { base: string; head: string }, shown?: { base: string; head: string }, reads = 0) => { + remotePair = { ...remotePair, ...pair }; + if (shown) { lagged = { ...remotePair, ...shown }; lagging = reads; } else lagging = 0; + }, rebaseRuns: () => rebaseRuns, rebaseAborts: () => rebaseAborts, rebaseBudgets, abortBudgets }; } @@ -339,7 +420,7 @@ it('revalidates task and review versions after the final remote read', async () it('revalidates task and review versions after final authorization', async () => { const fixture = await rebaseFixture('feature\n', undefined, { duringAuthorize(service, call) { - if (call !== 2) return; + if (call !== 4) return; // validations: 1 rebase, 2 push read, 3 push boundary, 4 final const view = service.load(); service.store.addReviewNote(service.config.identity, view.expected, 'P1', 'change', 'Changed during authorization.'); } }); @@ -350,7 +431,7 @@ it('revalidates task and review versions after final authorization', async () => it('refreshes a pull request head that moves during final authorization', async () => { const fixture = await rebaseFixture('feature\n', undefined, { duringAuthorize(service, call) { - if (call !== 2) return; + if (call !== 4) return; // validations: 1 rebase, 2 push read, 3 push boundary, 4 final writeFileSync(join(service.config.repository, 'late.txt'), 'late collaborator push\n'); fixtureGit(service.config.repository, 'add', 'late.txt'); fixtureGit(service.config.repository, 'commit', '-qm', 'late push'); return { base: service.load().snapshot.base, head: fixtureGit(service.config.repository, 'rev-parse', 'HEAD') }; @@ -469,7 +550,7 @@ it('refreshes the moved head against its prior base when the PR base and head ad stop: () => false, isActive: () => false, status: () => ({ active: false, stopRequested: null, unresolved: null }), get unreleased() { return null; } } as never, { run: async () => { throw new Error('No rebase expected.'); }, abort: async () => undefined } as never, - { inspect: async () => moved, fetch: async () => undefined }); + { inspect: async () => moved, fetch: async () => undefined, ...unusedPush }); const task = service.store.getTask(config.identity); const result = await coordinator.start({ stateVersion: task.stateVersion, reviewVersion: view.expected.reviewVersion!, snapshotId: view.snapshot.id, base: view.snapshot.base, head: view.snapshot.head }); @@ -525,7 +606,7 @@ it('traces an unpushed rewrite back to the base of a collaborator head', async ( get unreleased() { return null; } } as never, { run: async () => { throw new Error('No rebase expected.'); }, abort: async () => undefined } as never, { inspect: async () => ++reads === 1 ? { base: newBase, head: rewrittenHead } : { base: newBase, head: collaboratorHead }, - fetch: async () => undefined }); + fetch: async () => undefined, ...unusedPush }); const task = service.store.getTask(identity); const result = await coordinator.start({ stateVersion: task.stateVersion, reviewVersion: view.expected.reviewVersion!, snapshotId: view.snapshot.id, base: view.snapshot.base, head: view.snapshot.head }); @@ -551,7 +632,7 @@ it('refuses preparation before review or remote work while runner cleanup is unr get unreleased() { return kind === 'resources' ? [{ kind: 'container', name: 'agent-x' }] : null; } } as never, { run: async () => { throw new Error('No rebase expected.'); }, abort: async () => undefined } as never, { inspect: async () => { inspected = true; return { base: view.snapshot.base, head: view.snapshot.head }; }, - fetch: async () => undefined }); + fetch: async () => undefined, ...unusedPush }); const result = await coordinator.start({ stateVersion: task.stateVersion, reviewVersion: view.expected.reviewVersion!, snapshotId: view.snapshot.id, base: view.snapshot.base, head: view.snapshot.head }); expect(result).toMatchObject({ state: 'failed', reason: expect.stringMatching(/restart|cleanup/i) }); @@ -579,7 +660,7 @@ it('refuses preparation before review or remote work while a durable rebase mark get unreleased() { return null; } } as never, { run: async () => { throw new Error('No rebase expected.'); }, abort: async () => undefined } as never, { inspect: async () => { inspected = true; return { base: view.snapshot.base, head: view.snapshot.head }; }, - fetch: async () => undefined }); + fetch: async () => undefined, ...unusedPush }); const result = await coordinator.start({ stateVersion: task.stateVersion, reviewVersion: view.expected.reviewVersion!, snapshotId: view.snapshot.id, base: view.snapshot.base, head: view.snapshot.head }); expect(result).toMatchObject({ state: 'failed', reason: expect.stringMatching(/rebase|recovery|cleanup/i) }); @@ -599,7 +680,7 @@ it('preserves an already-requested shutdown without loading the review', async ( stop: () => false, isActive: () => false, status: () => ({ active: false, stopRequested: null, unresolved: null }), get unreleased() { return null; } } as never, { run: async () => { throw new Error('No rebase expected.'); }, abort: async () => undefined } as never, - { inspect: async () => { throw new Error('No remote inspection expected.'); }, fetch: async () => undefined }); + { inspect: async () => { throw new Error('No remote inspection expected.'); }, fetch: async () => undefined, ...unusedPush }); const active = coordinator.start({ stateVersion: task.stateVersion, reviewVersion: view.expected.reviewVersion!, snapshotId: view.snapshot.id, base: view.snapshot.base, head: view.snapshot.head }); await coordinator.close(); @@ -621,7 +702,7 @@ it('settles an admitted action after shutdown aborts its remote refresh', async get unreleased() { return null; } } as never, { run: async () => { throw new Error('No rebase expected.'); }, abort: async () => undefined } as never, { inspect: signal => new Promise((_resolve, reject) => signal?.addEventListener('abort', () => reject(signal.reason), { once: true })), - fetch: async () => undefined }, 60_000, capability); + fetch: async () => undefined, ...unusedPush }, 60_000, capability); const active = coordinator.start({ stateVersion: task.stateVersion, reviewVersion: view.expected.reviewVersion!, snapshotId: view.snapshot.id, base: view.snapshot.base, head: view.snapshot.head, actionId }); await Promise.resolve(); @@ -642,7 +723,7 @@ it('applies one operation-wide deadline to remote work and later checks', async get unreleased() { return null; } } as never, { run: async () => { throw new Error('No rebase expected.'); }, abort: async () => undefined } as never, { inspect: signal => new Promise((_resolve, reject) => signal?.addEventListener('abort', () => reject(signal.reason), { once: true })), - fetch: async () => undefined }, 20); + fetch: async () => undefined, ...unusedPush }, 20); await expect(coordinator.start({ stateVersion: task.stateVersion, reviewVersion: view.expected.reviewVersion!, snapshotId: view.snapshot.id, base: view.snapshot.base, head: view.snapshot.head })).resolves.toMatchObject({ state: 'failed' }); await coordinator.close(); @@ -663,7 +744,7 @@ it('aborts remote work early enough to reserve bounded subprocess settlement', a { inspect: signal => new Promise((_resolve, reject) => signal?.addEventListener('abort', () => { abortedAt = performance.now(); setTimeout(() => reject(signal.reason), processReserve); - }, { once: true })), fetch: async () => undefined }, operationTimeout, undefined, undefined, + }, { once: true })), fetch: async () => undefined, ...unusedPush }, operationTimeout, undefined, undefined, { processMs: processReserve }); const startedAt = performance.now(); const result = await coordinator.start({ stateVersion: task.stateVersion, reviewVersion: view.expected.reviewVersion!, @@ -730,7 +811,7 @@ it('makes a preparation action resendable when its terminal storage write fails' stop: () => false, isActive: () => false, status: () => ({ active: false, stopRequested: null, unresolved: null }), get unreleased() { return null; } } as never, { run: async () => { throw new Error('No rebase expected.'); }, abort: async () => undefined } as never, - { inspect: async () => ({ base: view.snapshot.base, head: view.snapshot.head }), fetch: async () => undefined }); + { inspect: async () => ({ base: view.snapshot.base, head: view.snapshot.head }), fetch: async () => undefined, ...unusedPush }); const result = await coordinator.start({ stateVersion: task.stateVersion, reviewVersion: view.expected.reviewVersion!, snapshotId: view.snapshot.id, base: view.snapshot.base, head: view.snapshot.head, actionId }); expect(result).toMatchObject({ state: 'failed', reason: 'transient storage failure' }); @@ -758,7 +839,7 @@ it('settles a failed preparation even when the fallback snapshot read would fail stop: () => false, isActive: () => false, status: () => ({ active: false, stopRequested: null, unresolved: null }), get unreleased() { return null; } } as never, { run: async () => { throw new Error('No rebase expected.'); }, abort: async () => undefined } as never, - { inspect: async () => { throw new Error('remote inspection failed'); }, fetch: async () => undefined }); + { inspect: async () => { throw new Error('remote inspection failed'); }, fetch: async () => undefined, ...unusedPush }); await expect(coordinator.start({ stateVersion: task.stateVersion, reviewVersion: view.expected.reviewVersion!, snapshotId: view.snapshot.id, base: view.snapshot.base, head: view.snapshot.head, actionId })) .resolves.toMatchObject({ state: 'failed', reason: 'transient snapshot read failure' }); @@ -815,7 +896,7 @@ it('rechecks the remote after local preparation and refreshes a head that moved stop: () => false, isActive: () => false, status: () => ({ active: false, stopRequested: null, unresolved: null }), get unreleased() { return null; } } as never, { run: async () => { throw new Error('No rebase expected.'); }, abort: async () => undefined } as never, - { inspect: async () => ++reads === 1 ? { base, head } : moved, fetch: async () => undefined }); + { inspect: async () => ++reads === 1 ? { base, head } : moved, fetch: async () => undefined, ...unusedPush }); const task = service.store.getTask(identity); const result = await coordinator.start({ stateVersion: task.stateVersion, reviewVersion: view.expected.reviewVersion!, snapshotId: view.snapshot.id, base: view.snapshot.base, head: view.snapshot.head }); @@ -824,3 +905,564 @@ it('rechecks the remote after local preparation and refreshes a head that moved expect(service.load().segments.some(segment => segment.row === 'Unplanned')).toBe(true); await coordinator.close(); }); + +// F6a of #22: the reviewed, rewritten head is pushed over exactly the remote head it rewrote, bracketed by a durable marker. +const prepareAgain = (fixture: Awaited>) => { + const identity = fixture.service.config.identity, view = fixture.service.load(); + return fixture.coordinator.start({ stateVersion: fixture.service.store.getTask(identity).stateVersion, + reviewVersion: view.expected.reviewVersion!, snapshotId: view.snapshot.id, base: view.snapshot.base, head: view.snapshot.head }); +}; +const pushMarker = (fixture: Awaited>) => + fixture.service.store.getTask(fixture.service.config.identity).pushInProgress; + +it('pushes the rewritten head over the exact remote head before running command checks', async () => { + const fixture = await rebaseFixture('feature\n', 0); + expect(fixture.result).toMatchObject({ state: 'ready', head: fixture.rebased, checked: ['P1'] }); + expect(fixture.pushes).toEqual([{ branch: BRANCH, from: expect.any(String), to: fixture.rebased }]); + expect(fixture.pushes[0]!.from).not.toBe(fixture.rebased); + expect(fixture.remote().head).toBe(fixture.rebased); + expect(fixture.checkedHeads).toEqual([fixture.rebased]); + expect(pushMarker(fixture)).toBeNull(); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('does not push a rewrite whose approval became stale', async () => { + const fixture = await rebaseFixture('changed by rebase\n'); + expect(fixture.result).toMatchObject({ state: 'review-required', head: fixture.rebased }); + expect(fixture.pushes).toEqual([]); + expect(fixture.remote().head).not.toBe(fixture.rebased); + expect(pushMarker(fixture)).toBeNull(); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('returns to review without pushing when the branch moved off the rewritten head', async () => { + let collaborator = ''; + const fixture = await rebaseFixture('feature\n', undefined, { pushedFirst(service) { + writeFileSync(join(service.config.repository, 'late.txt'), 'collaborator\n'); + fixtureGit(service.config.repository, 'add', 'late.txt'); fixtureGit(service.config.repository, 'commit', '-qm', 'collaborator'); + collaborator = fixtureGit(service.config.repository, 'rev-parse', 'HEAD'); + return collaborator; + } }); + expect(fixture.result).toMatchObject({ state: 'review-required', head: collaborator }); + expect(fixture.result.reason).toMatch(/moved before the rewritten head could be pushed/); + expect(fixture.remote().head).toBe(collaborator); + expect(pushMarker(fixture)).toBeNull(); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('does not record a refused push\'s observation into a review edited during the wait for the new head', async () => { + let collaborator = '', edited = false; + const fixture = await rebaseFixture('feature\n', undefined, { pushVisibleAfter: 1, pushedFirst(service) { + writeFileSync(join(service.config.repository, 'late.txt'), 'collaborator\n'); + fixtureGit(service.config.repository, 'add', 'late.txt'); fixtureGit(service.config.repository, 'commit', '-qm', 'collaborator'); + collaborator = fixtureGit(service.config.repository, 'rev-parse', 'HEAD'); + return collaborator; + }, duringVisibilityWait(service) { + // A person edits the review while GitHub still shows the pre-push head. + if (edited) return; edited = true; + service.store.addReviewNote(service.config.identity, service.load().expected, 'P1', 'change', 'Changed during the wait.'); + } }); + expect(edited).toBe(true); + expect(fixture.result).toMatchObject({ state: 'failed', reason: expect.stringMatching(/changed during preparation/i) }); + expect(fixture.coordinator.last?.stale).toBe(true); + // The collaborator head was not recorded into the edited review. + expect(fixture.service.load().snapshot.head).not.toBe(collaborator); + // Caught before the collaborator head was even fetched. + expect(fixture.fetchesAfterPush()).toBe(0); + expect(pushMarker(fixture)).toBeNull(); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('does not record a refused push\'s observation into a review edited while the new head was fetched', async () => { + let collaborator = '', edited = false; + const fixture = await rebaseFixture('feature\n', undefined, { pushedFirst(service) { + writeFileSync(join(service.config.repository, 'late.txt'), 'collaborator\n'); + fixtureGit(service.config.repository, 'add', 'late.txt'); fixtureGit(service.config.repository, 'commit', '-qm', 'collaborator'); + collaborator = fixtureGit(service.config.repository, 'rev-parse', 'HEAD'); + return collaborator; + }, duringFetchAfterPush(service) { + // A person edits the review while the collaborator head is being fetched. + if (edited) return; edited = true; + service.store.addReviewNote(service.config.identity, service.load().expected, 'P1', 'change', 'Changed during the fetch.'); + } }); + expect(edited).toBe(true); + expect(fixture.result).toMatchObject({ state: 'failed', reason: expect.stringMatching(/changed during preparation/i) }); + expect(fixture.coordinator.last?.stale).toBe(true); + expect(fixture.service.load().snapshot.head).not.toBe(collaborator); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('waits for GitHub to report the collaborator head that refused the push before reviewing it', async () => { + let collaborator = ''; + const fixture = await rebaseFixture('feature\n', undefined, { pushVisibleAfter: 2, pushedFirst(service) { + writeFileSync(join(service.config.repository, 'late.txt'), 'collaborator\n'); + fixtureGit(service.config.repository, 'add', 'late.txt'); fixtureGit(service.config.repository, 'commit', '-qm', 'collaborator'); + collaborator = fixtureGit(service.config.repository, 'rev-parse', 'HEAD'); + return collaborator; + } }); + expect(fixture.result).toMatchObject({ state: 'review-required', head: collaborator }); + expect(fixture.service.load().snapshot.head).toBe(collaborator); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('settles an ambiguous push by reading the branch, then prepares again without pushing twice', async () => { + let fail = true; + const fixture = await rebaseFixture('feature\n', undefined, { landBeforeFailure: true, + pushFailure: () => fail ? new Error('connection reset after the push was sent') : new Error('No second push expected.') }); + expect(fixture.result).toMatchObject({ state: 'failed', reason: 'connection reset after the push was sent' }); + // The push claim and its settlement advanced task state; the failure is bound to that state, so it reads as current. + expect(fixture.coordinator.last).toMatchObject({ state: 'failed', stale: false }); + expect(pushMarker(fixture)).toBeNull(); + expect(fixture.remote().head).toBe(fixture.rebased); + fail = false; + await expect(prepareAgain(fixture)).resolves.toMatchObject({ state: 'ready', head: fixture.rebased }); + expect(fixture.pushes).toHaveLength(1); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('keeps the push marker when the branch cannot be read, blocks merging, and settles it on the next preparation', async () => { + let readable = false; + const fixture = await rebaseFixture('feature\n', undefined, { + pushFailure: () => new Error('timed out'), + branchRead: current => { if (!readable) throw new Error('GitHub unreachable'); return current; }, + }); + const identity = fixture.service.config.identity; + expect(fixture.result).toMatchObject({ state: 'failed', reason: 'timed out' }); + const marker = pushMarker(fixture); + expect(marker).toMatchObject({ branch: BRANCH, to: fixture.rebased }); + const view = fixture.service.load(); + expect(() => fixture.service.store.transitionTask(identity, fixture.service.store.getTask(identity).stateVersion, 'queued')) + .toThrow(/push of the rewritten head is in progress/); + expect(() => fixture.service.store.beginMergeAttempt(identity, { ...view.expected, reviewVersion: view.expected.reviewVersion! }, + fixture.rebased, null, 'direct', randomUUID(), fixture.service.store.getTask(identity).stateVersion)) + .toThrow(/push of the rewritten head is in progress/); + await expect(prepareAgain(fixture)).resolves.toMatchObject({ state: 'failed', + reason: expect.stringMatching(/outcome of an earlier push .* is unknown.*GitHub unreachable/) }); + expect(pushMarker(fixture)).toEqual(marker); + readable = true; + await expect(prepareAgain(fixture)).resolves.toMatchObject({ state: 'failed', reason: expect.stringMatching(/was settled/) }); + expect(pushMarker(fixture)).toBeNull(); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('pushes nothing when the review changes at the push boundary', async () => { + const fixture = await rebaseFixture('feature\n', undefined, { duringAuthorize(service, call) { + if (call !== 3) return; // validations: 1 rebase, 2 push read, 3 push boundary + const view = service.load(); + service.store.addReviewNote(service.config.identity, view.expected, 'P1', 'change', 'Changed at the push boundary.'); + } }); + expect(fixture.result).toMatchObject({ state: 'failed' }); + expect(fixture.result.reason).toMatch(/changed before the rewritten head was pushed/); + expect(fixture.remote().head).not.toBe(fixture.rebased); + expect(pushMarker(fixture)).toBeNull(); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('closes a task cancelled during its push once the push outcome is read', async () => { + const actionId = randomUUID(); + let coordinatorRef: PreMergeCoordinator | null = null; + const fixture = await rebaseFixture('feature\n', undefined, { pushFailure(service) { + const task = service.store.getTask(service.config.identity); + expect(coordinatorRef!.cancelTask(task.stateVersion, actionId)).toBe('stopping'); + return new Error('Task cancelled.'); + }, duringRebase(_service, coordinator) { coordinatorRef = coordinator; } }); + expect(fixture.result).toMatchObject({ state: 'failed', reason: 'Task cancelled.' }); + expect(pushMarker(fixture)).toBeNull(); + expect(fixture.service.store.getTask(fixture.service.config.identity).status).toBe('cancelled'); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('leaves the push marker for startup when shutdown interrupts the push, and startup settles it', async () => { + let coordinatorRef: PreMergeCoordinator | null = null; + const fixture = await rebaseFixture('feature\n', undefined, { pushFailure() { + void coordinatorRef!.close(); + return new Error('Server shutdown.'); + }, duringRebase(_service, coordinator) { coordinatorRef = coordinator; } }); + expect(fixture.result).toMatchObject({ state: 'failed', reason: 'Server shutdown.' }); + expect(pushMarker(fixture)).toMatchObject({ to: fixture.rebased }); + // Shutdown starts no new GitHub read: the marker waits for startup. + expect(fixture.branchReads()).toBe(0); + const restarted = new PreMergeCoordinator(fixture.service, fixture.runner, {} as never, { + inspect: async () => { throw new Error('not used'); }, fetch: async () => undefined, + push: async () => { throw new Error('not used'); }, readBranch: async () => fixture.remote().head }); + await expect(restarted.settlePendingPush()).resolves.toBe('not-pushed'); + expect(pushMarker(fixture)).toBeNull(); + await restarted.close(); await fixture.runner.close(); +}); + +it('waits for GitHub to report the pushed head before the later inspections compare against it', async () => { + const fixture = await rebaseFixture('feature\n', 0, { pushVisibleAfter: 2 }); + expect(fixture.result).toMatchObject({ state: 'ready', head: fixture.rebased, checked: ['P1'] }); + expect(fixture.service.load().snapshot.head).toBe(fixture.rebased); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('fails without refreshing the review when GitHub never reports the pushed head in time', async () => { + const fixture = await rebaseFixture('feature\n', 0, { pushVisibleAfter: 1_000, reserves: { pushVisibleMs: 600 } }); + expect(fixture.result).toMatchObject({ state: 'failed', checked: [] }); + expect(fixture.result.reason).toMatch(/has not reported the pull request's new head yet/); + expect(fixture.coordinator.last).toMatchObject({ state: 'failed', stale: false }); + expect(fixture.checkedHeads).toEqual([]); + expect(fixture.service.load().snapshot.head).toBe(fixture.rebased); + expect(pushMarker(fixture)).toBeNull(); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('lets a cancel during the startup push settlement finish the read and close the task', async () => { + let coordinatorRef: PreMergeCoordinator | null = null; + const fixture = await rebaseFixture('feature\n', undefined, { pushFailure() { + void coordinatorRef!.close(); + return new Error('Server shutdown.'); + }, duringRebase(_service, coordinator) { coordinatorRef = coordinator; } }); + const identity = fixture.service.config.identity; + expect(pushMarker(fixture)).not.toBeNull(); + const read = Promise.withResolvers(); + let readSignal: AbortSignal | undefined; + const restarted = new PreMergeCoordinator(fixture.service, fixture.runner, {} as never, { + inspect: async () => { throw new Error('not used'); }, fetch: async () => undefined, + push: async () => { throw new Error('not used'); }, + readBranch: async (_branch, signal) => { readSignal = signal; return read.promise; } }); + const settling = restarted.settlePendingPush(); + expect(restarted.active).toBe(true); + expect(restarted.cancelTask(fixture.service.store.getTask(identity).stateVersion, randomUUID())).toBe('stopping'); + expect(readSignal?.aborted).toBe(false); + read.resolve(fixture.remote().head); + await expect(settling).resolves.toBe('not-pushed'); + expect(pushMarker(fixture)).toBeNull(); + expect(fixture.service.store.getTask(identity).status).toBe('cancelled'); + await restarted.close(); await fixture.runner.close(); +}); + +it('reviews a collaborator push that lands on the rewritten head right after it was pushed', async () => { + let collaborator = ''; + const fixture = await rebaseFixture('feature\n', 0, { afterPush(service, pushed) { + const tree = fixtureGit(service.config.repository, 'rev-parse', `${pushed}^{tree}`); + collaborator = fixtureGit(service.config.repository, 'commit-tree', tree, '-p', pushed, '-m', 'collaborator on top'); + return collaborator; + } }); + expect(fixture.result).toMatchObject({ state: 'review-required', head: collaborator, checked: [] }); + expect(fixture.result.reason).toMatch(/moved after the rewritten head was pushed/); + expect(fixture.service.load().snapshot).toMatchObject({ base: fixture.remote().base, head: collaborator }); + expect(fixture.checkedHeads).toEqual([]); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('reviews a collaborator push that lands between the push and its read-back', async () => { + let collaborator = ''; + const fixture = await rebaseFixture('feature\n', 0, { readBackFails: true, afterPush(service, pushed) { + const tree = fixtureGit(service.config.repository, 'rev-parse', `${pushed}^{tree}`); + collaborator = fixtureGit(service.config.repository, 'commit-tree', tree, '-p', pushed, '-m', 'collaborator before the read-back'); + return collaborator; + } }); + // The push's own read-back failed; the settlement read found the collaborator's head, so it is reviewed, not failed. + expect(fixture.result).toMatchObject({ state: 'review-required', head: collaborator, checked: [] }); + expect(fixture.result.reason).toMatch(/moved while the rewritten head was being pushed/); + expect(fixture.service.store.getTask(fixture.service.config.identity).pushInProgress).toBeNull(); + expect(fixture.service.load().snapshot).toMatchObject({ base: fixture.remote().base, head: collaborator }); + expect(fixture.checkedHeads).toEqual([]); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('keeps waiting while GitHub still reports the pre-push head, even if the base moved meanwhile', async () => { + let movedBase = ''; + const fixture = await rebaseFixture('feature\n', 0, { pushVisibleAfter: 2, movedBaseAfterPush(service) { + const onto = service.store.getSnapshot(service.config.identity).base; + const tree = fixtureGit(service.config.repository, 'rev-parse', `${onto}^{tree}`); + movedBase = fixtureGit(service.config.repository, 'commit-tree', tree, '-p', onto, '-m', 'base moved again'); + return movedBase; + } }); + expect(fixture.result).toMatchObject({ state: 'review-required', head: fixture.rebased, checked: [] }); + expect(fixture.result.reason).toMatch(/moved after the rewritten head was pushed/); + // The review stays on the head the branch holds, against the base it descends from: never back on the pre-push head. + expect(fixture.service.load().snapshot.head).toBe(fixture.rebased); + expect(fixture.service.load().segments.filter(segment => segment.row === 'Unplanned')).toEqual([]); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('pushes nothing when a cancel is committed while the push reads the remote, and closes the task once the outcome is read', async () => { + const fixture = await rebaseFixture('feature\n', undefined, { duringPushRead(service, coordinator) { + // The production route: the coordinator cancels the task and aborts this preparation after the cancel commits. + expect(coordinator.cancelTask(service.store.getTask(service.config.identity).stateVersion, randomUUID())).toBe('stopping'); + } }); + expect(fixture.result).toMatchObject({ state: 'failed' }); + expect(fixture.remote().head).not.toBe(fixture.rebased); + expect(fixture.service.store.getTask(fixture.service.config.identity)).toMatchObject({ status: 'cancelled', pushInProgress: null }); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('pushes nothing when the reviewed snapshot changes at the push boundary', async () => { + const fixture = await rebaseFixture('feature\n', undefined, { duringAuthorize(service, call) { + if (call !== 3) return; // validations: 1 rebase, 2 push read, 3 push boundary + const identity = service.config.identity, view = service.load(), snapshot = service.store.getSnapshot(identity); + const tree = fixtureGit(service.config.repository, 'rev-parse', `${snapshot.head}^{tree}`); + const later = fixtureGit(service.config.repository, 'commit-tree', tree, '-p', snapshot.head, '-m', 'later'); + service.store.recordHistory(identity, view.expected, snapshot.base, later, []); + } }); + expect(fixture.result).toMatchObject({ state: 'failed' }); + expect(fixture.result.reason).toMatch(/changed before the rewritten head was pushed/); + expect(fixture.remote().head).not.toBe(fixture.rebased); + expect(fixture.service.store.getTask(fixture.service.config.identity).pushInProgress).toBeNull(); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('after a refused push, traces a lagging earlier push of codeboost\'s own rewrite from its own base, not the original one', async () => { + // 1. The push of R1 lands, but the PR API keeps showing the original head, so preparation fails. + let r2 = ''; + const fixture = await rebaseFixture('feature\n', undefined, { pushVisibleAfter: 1_000, reserves: { pushVisibleMs: 300 }, + rebaseResult(service, input, run) { + if (run !== 2) return undefined; + // 3. The second rebase rewrites R1 onto the moved base as R2. + const repository = service.config.repository; + fixtureGit(repository, 'switch', '-qc', 'second-rebase', input.onto); + writeFileSync(join(repository, 'a.txt'), 'feature\n'); fixtureGit(repository, 'commit', '-qam', 'rebased feature again'); + r2 = fixtureGit(repository, 'rev-parse', 'HEAD'); + fixtureGit(repository, 'switch', '-q', 'feature'); + return r2; + } }); + expect(fixture.result).toMatchObject({ state: 'failed' }); + const identity = fixture.service.config.identity, r1 = fixture.rebased, b1 = fixture.service.store.getSnapshot(identity).base; + expect(fixture.remote().head).toBe(r1); + // 2. The base branch moves again; the PR API still shows the original head for two more reads. + const repository = fixture.service.config.repository; + fixtureGit(repository, 'switch', '-qc', 'moved-again', b1); + writeFileSync(join(repository, 'c.txt'), 'later base\n'); fixtureGit(repository, 'add', 'c.txt'); fixtureGit(repository, 'commit', '-qm', 'move base again'); + const b2 = fixtureGit(repository, 'rev-parse', 'HEAD'); + fixtureGit(repository, 'switch', '-q', 'feature'); + const original = fixture.pushes[0]!.from; + fixture.moveRemote({ base: b2, head: r1 }, { base: b2, head: original }, 2); + // 4. The retry rebases R1 onto B2 and pushes R2 leased to the original head; the branch holds R1, so it is refused. + const retried = await prepareAgain(fixture); + expect(retried).toMatchObject({ state: 'review-required', head: r1 }); + expect(fixture.pushes.at(-1)).toMatchObject({ from: original, to: r2 }); + // R1 is reviewed against B1, the base it was rewritten onto: the upstream base commit is not attributed to the PR. + expect(fixture.service.store.getSnapshot(identity)).toMatchObject({ base: b1, head: r1 }); + expect(fixture.service.load().segments.filter(segment => segment.row === 'Unplanned')).toEqual([]); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('reports a failure after a successful push as current, not stale', async () => { + const fixture = await rebaseFixture('feature\n', 0, { fetchFailsAfterPush: true, afterPush(service, pushed) { + const tree = fixtureGit(service.config.repository, 'rev-parse', `${pushed}^{tree}`); + return fixtureGit(service.config.repository, 'commit-tree', tree, '-p', pushed, '-m', 'collaborator on top'); + } }); + expect(fixture.result).toMatchObject({ state: 'failed', reason: 'fetch failed after the push' }); + expect(fixture.coordinator.last).toMatchObject({ state: 'failed', stale: false }); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +/** Make the monotonic clock jump so only `leftMs` of a default ten-minute preparation deadline remains. */ +const leaveOfDeadline = (startedAt: number, leftMs: number) => { + const real = performance.now.bind(performance), offset = startedAt + 10 * 60_000 - leftMs - real(); + vi.spyOn(performance, 'now').mockImplementation(() => real() + offset); +}; + +it('pushes nothing when a plan revision is applied at the push boundary', async () => { + const fixture = await rebaseFixture('feature\n', undefined, { duringAuthorize(service, call) { + if (call !== 3) return; // validations: 1 rebase, 2 push read, 3 push boundary + const plan = service.store.getPlan(service.config.identity); + // A plan revision advances only the task state version: neither the review version nor the snapshot changes. + service.store.importRevision(JSON.stringify({ ...plan, summary: 'Amended at the push boundary' }), 'json', + service.planContext(), plan.revision); + } }); + expect(fixture.result).toMatchObject({ state: 'failed' }); + expect(fixture.result.reason).toMatch(/changed before the rewritten head was pushed/); + expect(fixture.remote().head).not.toBe(fixture.rebased); + expect(fixture.pushes).toHaveLength(1); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('ends the wait for the pushed head when the preparation deadline, not the visibility limit, runs out', async () => { + const startedAt = performance.now(), waitStarted: number[] = []; + const fixture = await rebaseFixture('feature\n', undefined, { pushVisibleAfter: 1_000_000, afterPush(_service, pushed) { + // The process-settlement reserve plus 1.5 s remain once the push lands; the 30 s visibility limit would outlast it. + leaveOfDeadline(startedAt, PRE_MERGE_PROCESS_SETTLEMENT_RESERVE_MS + 1_500); + waitStarted.push(Date.now()); + return pushed; + } }); + expect(fixture.result).toMatchObject({ state: 'failed' }); + expect(fixture.result.reason).toMatch(/has not reported the pull request's new head yet/); + expect(Date.now() - waitStarted[0]!).toBeLessThan(10_000); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('leaves the push marker without a branch read when the preparation deadline ends the push', async () => { + let startedAt = 0; + const fixture = await rebaseFixture('feature\n', undefined, { branchReadHangs: true, + duringRebase() { startedAt = performance.now(); }, + pushFailure() { + // The preparation's timer aborts its work the settlement reserve before the deadline; so does this push. + leaveOfDeadline(startedAt, PRE_MERGE_PROCESS_SETTLEMENT_RESERVE_MS - 50); + return Object.assign(new Error('Pre-merge preparation deadline exceeded.'), { code: 'ETIMEDOUT' }); + } }); + expect(fixture.result).toMatchObject({ state: 'failed', reason: 'Pre-merge preparation deadline exceeded.' }); + // Less than Git's settlement reserve is left, so no read starts: the outcome stays unknown and the marker stays. + expect(fixture.branchReads()).toBe(0); + expect(pushMarker(fixture)).toMatchObject({ to: fixture.rebased }); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('lets shutdown abort a startup push settlement whose branch read does not answer', async () => { + let coordinatorRef: PreMergeCoordinator | null = null; + const fixture = await rebaseFixture('feature\n', undefined, { pushFailure() { + void coordinatorRef!.close(); + return new Error('Server shutdown.'); + }, duringRebase(_service, coordinator) { coordinatorRef = coordinator; } }); + expect(pushMarker(fixture)).not.toBeNull(); + const restarted = new PreMergeCoordinator(fixture.service, fixture.runner, {} as never, { + inspect: async () => { throw new Error('not used'); }, fetch: async () => undefined, + push: async () => { throw new Error('not used'); }, + readBranch: async (_branch, signal) => new Promise((_resolve, reject) => + signal?.addEventListener('abort', () => reject(signal.reason), { once: true })) }); + const settling = restarted.settlePendingPush(), closedAt = Date.now(); + await restarted.close(); + expect(Date.now() - closedAt).toBeLessThan(5_000); + await expect(settling).resolves.toBeNull(); + expect(pushMarker(fixture)).not.toBeNull(); + await fixture.runner.close(); +}); + +it('lets a cancel during the next preparation\'s push settlement finish the read and close the task', async () => { + let readable = false; + const read = Promise.withResolvers(); + const fixture = await rebaseFixture('feature\n', undefined, { + pushFailure: () => new Error('timed out'), + branchRead: (current, signal) => { + if (!readable) throw new Error('GitHub unreachable'); + // A real read is aborted by its signal; a cancel must not be able to abort this one. + signal?.addEventListener('abort', () => read.reject(signal.reason), { once: true }); + return read.promise; + }, + }); + const identity = fixture.service.config.identity; + expect(pushMarker(fixture)).not.toBeNull(); + readable = true; + const preparing = prepareAgain(fixture); + await vi.waitFor(() => expect(fixture.branchReads()).toBe(2)); + expect(fixture.coordinator.cancelTask(fixture.service.store.getTask(identity).stateVersion, randomUUID())).toBe('stopping'); + read.resolve(fixture.remote().head); + await expect(preparing).resolves.toMatchObject({ state: 'failed', reason: 'Task cancelled.' }); + expect(fixture.service.store.getTask(identity)).toMatchObject({ status: 'cancelled', pushInProgress: null }); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('reports a failed push as stale when a plan revision lands while its outcome is read', async () => { + let serviceRef: ReviewService | null = null, revised = false; + const fixture = await rebaseFixture('feature\n', undefined, { landBeforeFailure: true, + duringRebase(service) { serviceRef = service; }, + pushFailure: () => new Error('connection reset'), + branchRead: (current) => { + // Another request applies a plan revision while the settlement read is in flight. + const service = serviceRef!, plan = service.store.getPlan(service.config.identity); + service.store.importRevision(JSON.stringify({ ...plan, summary: 'Amended during the read' }), 'json', + service.planContext(), plan.revision); + revised = true; + return current; + } }); + expect(fixture.result).toMatchObject({ state: 'failed', reason: 'connection reset' }); + expect(revised).toBe(true); + expect(pushMarker(fixture)).toBeNull(); + expect(fixture.coordinator.last).toMatchObject({ state: 'failed', stale: true }); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('reports a failure after a push as stale when a plan revision landed while the push was on the wire', async () => { + const revise = (service: ReviewService) => { + const plan = service.store.getPlan(service.config.identity); + service.store.importRevision(JSON.stringify({ ...plan, summary: 'Amended during the push' }), 'json', + service.planContext(), plan.revision); + }; + // A successful push, then a failure while reviewing the head a collaborator pushed on top of it. + const pushed = await rebaseFixture('feature\n', 0, { fetchFailsAfterPush: true, afterPush(service, head) { + revise(service); + const tree = fixtureGit(service.config.repository, 'rev-parse', `${head}^{tree}`); + return fixtureGit(service.config.repository, 'commit-tree', tree, '-p', head, '-m', 'collaborator on top'); + } }); + // The revision is caught before the collaborator head is fetched and recorded: the failure names the change. + expect(pushed.result).toMatchObject({ state: 'failed', reason: expect.stringMatching(/changed during preparation/i) }); + expect(pushed.coordinator.last).toMatchObject({ state: 'failed', stale: true }); + await pushed.coordinator.close(); await pushed.runner.close(); + // A refused push, then a failure while reviewing the collaborator head that refused it. + const refused = await rebaseFixture('feature\n', undefined, { fetchFailsAfterPush: true, pushedFirst(service) { + revise(service); + writeFileSync(join(service.config.repository, 'late.txt'), 'collaborator\n'); + fixtureGit(service.config.repository, 'add', 'late.txt'); fixtureGit(service.config.repository, 'commit', '-qm', 'collaborator'); + return fixtureGit(service.config.repository, 'rev-parse', 'HEAD'); + } }); + expect(refused.result).toMatchObject({ state: 'failed', reason: expect.stringMatching(/changed during preparation/i) }); + expect(refused.coordinator.last).toMatchObject({ state: 'failed', stale: true }); + await refused.coordinator.close(); await refused.runner.close(); +}); + +it('runs no command check when a plan revision landed while the push was on the wire', async () => { + const fixture = await rebaseFixture('feature\n', 0, { afterPush(service, head) { + const plan = service.store.getPlan(service.config.identity); + service.store.importRevision(JSON.stringify({ ...plan, summary: 'Amended during the push' }), 'json', + service.planContext(), plan.revision); + return head; + } }); + expect(fixture.result).toMatchObject({ state: 'failed', checked: [] }); + expect(fixture.result.reason).toMatch(/changed during preparation/); + expect(fixture.checkedHeads).toEqual([]); + expect(fixture.coordinator.last).toMatchObject({ state: 'failed', stale: true }); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('leaves a running preparation in charge when a push settlement is requested meanwhile', async () => { + let requested: Promise | null = null, stillActive: boolean | null = null; + const fixture = await rebaseFixture('feature\n', undefined, { + duringRebase(_service, coordinator) { requested = coordinator.settlePendingPush(); }, + // Later, after several awaits: the settlement request must not have taken the preparation's place. + duringPushRead(_service, coordinator) { stillActive = coordinator.active; }, + }); + await expect(requested).resolves.toBeNull(); + expect(stillActive).toBe(true); + expect(fixture.result).toMatchObject({ state: 'ready', head: fixture.rebased }); + expect(fixture.branchReads()).toBe(0); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('pushes nothing when the task is reassigned while the push authorization is read', async () => { + const fixture = await rebaseFixture('feature\n', undefined, { duringAuthorize(service, call) { + if (call !== 2) return; // validations: 1 rebase, 2 push authorization read + const identity = service.config.identity; + service.store.setAssignment(identity, service.store.getTask(identity).stateVersion, 'someone-else', 'c'.repeat(40)); + } }); + expect(fixture.result).toMatchObject({ state: 'failed' }); + expect(fixture.pushes).toEqual([]); + expect(fixture.remote().head).not.toBe(fixture.rebased); + expect(pushMarker(fixture)).toBeNull(); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('keeps the process-settlement reserve inside a settlement read\'s time limit', async () => { + // Reserve plus 2 s left: the read may run for at most about 2 s, leaving the reserve for Git to settle. + let startedAt = 0, failedAt = 0, pushedAt = 0; + const fixture = await rebaseFixture('feature\n', undefined, { branchReadHangs: true, + // Just after the preparation started, so `leaveOfDeadline` measures from (almost) its own start. + duringRebase() { startedAt = performance.now(); }, + pushFailure() { + leaveOfDeadline(startedAt, PRE_MERGE_PROCESS_SETTLEMENT_RESERVE_MS + 2_000); + pushedAt = Date.now(); + return new Error('connection reset'); + } }); + failedAt = Date.now(); + expect(fixture.result).toMatchObject({ state: 'failed', reason: 'connection reset' }); + expect(fixture.branchReads()).toBe(1); + expect(failedAt - pushedAt).toBeLessThan(8_000); + expect(pushMarker(fixture)).toMatchObject({ to: fixture.rebased }); + await fixture.coordinator.close(); await fixture.runner.close(); +}); + +it('starts no settlement read when less than the process-settlement reserve is left', async () => { + let startedAt = 0; + const fixture = await rebaseFixture('feature\n', undefined, { branchReadHangs: true, + duringRebase() { startedAt = performance.now(); }, + pushFailure() { + leaveOfDeadline(startedAt, PRE_MERGE_PROCESS_SETTLEMENT_RESERVE_MS - 1_000); + return new Error('connection reset'); + } }); + expect(fixture.result).toMatchObject({ state: 'failed', reason: 'connection reset' }); + expect(fixture.branchReads()).toBe(0); + expect(pushMarker(fixture)).toMatchObject({ to: fixture.rebased }); + await fixture.coordinator.close(); await fixture.runner.close(); +}); diff --git a/test/publish.test.ts b/test/publish.test.ts index 6d90569a..8322ffab 100644 --- a/test/publish.test.ts +++ b/test/publish.test.ts @@ -358,7 +358,7 @@ describe('schema v7', () => { expect(upgraded.taskPullRequests(identity)).toEqual([]); expect(upgraded.latestAlreadyFixed(identity)).toBeNull(); const db = new DatabaseSync(path, { readOnly: true }); - expect(db.prepare('PRAGMA user_version').get()).toEqual({ user_version: 17 }); + expect(db.prepare('PRAGMA user_version').get()).toEqual({ user_version: 18 }); const names = db.prepare("SELECT name FROM sqlite_master WHERE type IN ('table','index') AND (name LIKE '%pull_requests%' OR name LIKE 'already_fixed_checks%') AND name NOT LIKE 'sqlite_autoindex%' ORDER BY name").all().map(row => row.name); expect(names).toEqual(['already_fixed_checks', 'already_fixed_checks_task', 'task_pull_requests', 'task_pull_requests_number', 'task_pull_requests_opening', 'task_pull_requests_task']); db.close(); diff --git a/test/runner-branch-push.test.ts b/test/runner-branch-push.test.ts index f2b6e9be..cf09289b 100644 --- a/test/runner-branch-push.test.ts +++ b/test/runner-branch-push.test.ts @@ -119,6 +119,33 @@ describe('GitBranchPusher', () => { expect(remoteRefs(s)).toBe(`${s.base} ${REF}`); }); + it('with an expected head, overwrites exactly that commit, even one codeboost did not make', async () => { + const s = await setup(), head = await runnerCommit(s, 'one'); + personPushes(s, REF); + await pusher(s).instance.push(IDENTITY, { head, branch: BRANCH, expected: s.base }); + expect(remoteRefs(s)).toBe(`${head} ${REF}`); + }); + + it('with an expected head, refuses any other value, even a commit codeboost made, and pushes nothing', async () => { + const s = await setup(), first = await runnerCommit(s, 'one'), second = await runnerCommit(s, 'two'); + await pusher(s).instance.push(IDENTITY, { head: first, branch: BRANCH }); + const p = pusher(s, { owned: () => [first] }); + await expect(p.instance.push(IDENTITY, { head: second, branch: BRANCH, expected: s.base })) + .rejects.toThrow(`holds ${first}, not the rebased head ${s.base}`); + await expect(p.instance.push(IDENTITY, { head: second, branch: 'codeboost/absent-1', expected: s.base })) + .rejects.toThrow(BranchPushRefused); + expect(remoteRefs(s)).toBe(`${first} ${REF}`); + }); + + it('reads the task branch for push settlement, and null when it does not exist', async () => { + const s = await setup(), head = await runnerCommit(s, 'one'), p = pusher(s); + await expect(p.instance.readBranch(IDENTITY, BRANCH)).resolves.toBeNull(); + await p.instance.push(IDENTITY, { head, branch: BRANCH }); + await expect(p.instance.readBranch(IDENTITY, BRANCH)).resolves.toBe(head); + await expect(p.instance.readBranch(IDENTITY, 'main')).rejects.toThrow(/codeboost\/ task branch/); + await expect(p.instance.readBranch({ ...IDENTITY, repositoryId: 'other' }, BRANCH)).rejects.toThrow(/another repository/); + }); + it('refuses when the branch moves between the read and the push: the lease is the value read', async () => { const s = await setup(), first = await runnerCommit(s, 'one'), second = await runnerCommit(s, 'two'); await pusher(s).instance.push(IDENTITY, { head: first, branch: BRANCH }); @@ -284,6 +311,56 @@ describe('GitBranchPusher', () => { expect(remoteRefs(s)).toBe(''); }); + it('fails a branch read promptly when a process outside Git\'s group holds its output', async () => { + const s = await setup(), bin = join(s.root, 'fake-bin'), pidFile = join(s.root, 'escaped.pid'); + mkdirSync(bin); + // Git's place on PATH: `ls-remote` leaves a process in a new session holding the inherited output, then exits. + writeFileSync(join(bin, 'git'), `#!/bin/sh\nexec ${JSON.stringify(process.execPath)} -e ${JSON.stringify( + `const c = require('node:child_process').spawn('sleep', ['60'], { detached: true, stdio: ['ignore', 'inherit', 'inherit'] });` + + `require('node:fs').writeFileSync(${JSON.stringify(pidFile)}, String(c.pid)); c.unref();`)}\n`, { mode: 0o755 }); + try { + const began = performance.now(); + const read = pusher(s, { env: { ...process.env, PATH: `${bin}:${process.env.PATH}` } }).instance.readBranch(IDENTITY, BRANCH); + await expect(read).rejects.toThrow(/outside its group still holds its output/); + expect(performance.now() - began).toBeLessThan(20_000); + } finally { + try { process.kill(Number(readFileSync(pidFile, 'utf8')), 'SIGKILL'); } catch { /* already gone */ } + } + }, 40_000); + + it('fails a commit check promptly when a process outside Git\'s group holds its output', async () => { + const s = await setup(), bin = join(s.root, 'fake-bin-cat'), pidFile = join(s.root, 'escaped-cat.pid'); + mkdirSync(bin); + // Git's place on PATH: other commands succeed at once with no output; `cat-file` leaves a process in a new session + // holding its output. + const escape = `const c = require('node:child_process').spawn('sleep', ['60'], { detached: true, stdio: ['ignore', 'inherit', 'inherit'] });` + + `require('node:fs').writeFileSync(${JSON.stringify(pidFile)}, String(c.pid)); c.unref();`; + writeFileSync(join(bin, 'git'), `#!/bin/sh\ncase " $* " in *" cat-file "*) exec ${JSON.stringify(process.execPath)} -e ${JSON.stringify(escape)} ;; esac\nexit 0\n`, { mode: 0o755 }); + try { + const began = performance.now(); + const instance = pusher(s, { env: { ...process.env, PATH: `${bin}:${process.env.PATH}` } }).instance; + await expect(instance.fetchCommits([s.base])).rejects.toThrow(/outside its group still holds its output/); + // A push checks its head with cat-file before anything is written: that read is bounded the same way. + await expect(instance.push(IDENTITY, { head: s.base, branch: BRANCH })).rejects.toThrow(/outside its group still holds its output/); + expect(performance.now() - began).toBeLessThan(30_000); + } finally { + try { process.kill(Number(readFileSync(pidFile, 'utf8')), 'SIGKILL'); } catch { /* already gone */ } + } + }, 40_000); + + it('waits for a process that stays in Git\'s group before a branch read returns', async () => { + const s = await setup(), bin = join(s.root, 'fake-bin-group'); + mkdirSync(bin); + // Git's place on PATH: `ls-remote` answers nothing and exits, leaving a child in its own group for 12 s more, past + // the 10 s bounded group drain an unsettled return would stop at. + writeFileSync(join(bin, 'git'), `#!/bin/sh\nexec ${JSON.stringify(process.execPath)} -e ${JSON.stringify( + `require('node:child_process').spawn('sleep', ['12'], { stdio: 'ignore' }).unref();`)}\n`, { mode: 0o755 }); + const began = performance.now(); + await expect(pusher(s, { env: { ...process.env, PATH: `${bin}:${process.env.PATH}` } }).instance.readBranch(IDENTITY, BRANCH)).resolves.toBeNull(); + // A member of Git's group is never abandoned: the read returned only after it had gone. + expect(performance.now() - began).toBeGreaterThanOrEqual(11_500); + }, 60_000); + it('removes a configured enterprise token of any shape from Git\'s error text', async () => { const s = await setup(), head = await runnerCommit(s, 'one'), token = 'enterprise-secret-0123456789'; const missing = new GitBranchPusher({ repository: s.repository, repositoryId: 'repo-1', remote: 'acme/app', diff --git a/test/runner-lifecycle-store.test.ts b/test/runner-lifecycle-store.test.ts index 4fa406ff..21a4e9c0 100644 --- a/test/runner-lifecycle-store.test.ts +++ b/test/runner-lifecycle-store.test.ts @@ -105,7 +105,7 @@ describe('schema v12 review ordering', () => { migrated.saveReview(identity, { revision: 1, snapshotId: snapshot.id, reviewVersion: migratedVersion }, [{ item: 'P1', fingerprint: 'approved-after-upgrade' }], []); expect(migrated.unapprovedExecutionItems(identity, 1)).toEqual([]); - expect(new DatabaseSync(path).prepare('PRAGMA user_version').get()).toEqual({ user_version: 17 }); + expect(new DatabaseSync(path).prepare('PRAGMA user_version').get()).toEqual({ user_version: 18 }); }); }); @@ -783,7 +783,7 @@ describe('runner commits and preparation groups (#87)', () => { db.exec('ALTER TABLE attempts DROP COLUMN preparation_identity; PRAGMA user_version=15;'); db.close(); const reopened = open(path); expect(reopened.interruptedAttempts()).toEqual([expect.objectContaining({ id: attempt.id, preparationIdentity: null })]); - expect(new DatabaseSync(path).prepare('PRAGMA user_version').get()).toEqual({ user_version: 17 }); + expect(new DatabaseSync(path).prepare('PRAGMA user_version').get()).toEqual({ user_version: 18 }); }); }); @@ -917,7 +917,7 @@ describe('durable safety findings (#87 item 3)', () => { db.exec('ALTER TABLE attempts DROP COLUMN safety_finding; PRAGMA user_version=7;'); db.close(); const reopened = open(path); expect(reopened.getAttempt(identity, attempt.id).safetyFinding).toBeNull(); - expect(new DatabaseSync(path).prepare('PRAGMA user_version').get()).toEqual({ user_version: 17 }); + expect(new DatabaseSync(path).prepare('PRAGMA user_version').get()).toEqual({ user_version: 18 }); }); it('adds the durable owed marker to a version 12 database', () => { @@ -927,7 +927,7 @@ describe('durable safety findings (#87 item 3)', () => { db.exec('ALTER TABLE attempts DROP COLUMN safety_owed; PRAGMA user_version=12;'); db.close(); const reopened = open(path); expect(reopened.owedSafetyFindings()).toEqual([]); - expect(new DatabaseSync(path).prepare('PRAGMA user_version').get()).toEqual({ user_version: 17 }); + expect(new DatabaseSync(path).prepare('PRAGMA user_version').get()).toEqual({ user_version: 18 }); }); it('restores and clears an escalation owed behind a human gate across real store reopens', () => { @@ -967,7 +967,7 @@ describe('allocation baseline (#91)', () => { db.exec('ALTER TABLE attempts DROP COLUMN metadata_baseline; ALTER TABLE attempts DROP COLUMN storage_base; PRAGMA user_version=8;'); db.close(); const reopened = open(path); expect(reopened.interruptedAttempts()).toEqual([expect.objectContaining({ id: attempt.id, metadataBaseline: null, storageBase: null })]); - expect(new DatabaseSync(path).prepare('PRAGMA user_version').get()).toEqual({ user_version: 17 }); + expect(new DatabaseSync(path).prepare('PRAGMA user_version').get()).toEqual({ user_version: 18 }); }); }); describe('publish outcomes (#103)', () => { @@ -983,7 +983,7 @@ describe('publish outcomes (#103)', () => { reopened.recordPublish(identity, { outcome: 'opened', draft: false, message: 'Pull request #1 is open.', number: 1, url: 'https://github.com/o/r/pull/1' }); expect(reopened.lastPublish(identity)).toMatchObject({ outcome: 'opened', number: 1 }); expect(reopened.getTask(identity)).toEqual(before); - expect(new DatabaseSync(path).prepare('PRAGMA user_version').get()).toEqual({ user_version: 17 }); + expect(new DatabaseSync(path).prepare('PRAGMA user_version').get()).toEqual({ user_version: 18 }); }); it('stamps an outcome with the version the publish saw, and refuses one it cannot have seen (#114)', () => { const { store } = queued(), version = store.getTask(identity).stateVersion; @@ -1002,3 +1002,87 @@ describe('publish outcomes (#103)', () => { message: expect.stringMatching(/Use the publish action/) }); }); }); + +describe('pre-merge push marker (F6a of #22)', () => { + const BRANCH = 'codeboost/task-1'; + /** The fixture's head oid(2) rewritten to oid(12) on a moved base oid(10). */ + function rewritten() { + const f = fixture(); + const reviewed = () => ({ revision: 1, snapshotId: f.store.getSnapshot(identity).id, reviewVersion: f.store.reviewVersion(identity) }); + f.store.recordHistory(identity, reviewed(), oid(1), oid(2), [{ sha: oid(2), owner: 'P1', origin: 'owned', sourceSha: null }]); + f.store.recordRebase(identity, reviewed(), oid(10), oid(12), [{ oldSha: oid(2), newSha: oid(12) }]); + return { ...f, reviewed }; + } + const begin = (f: ReturnType, input: Partial<{ branch: string; from: string; to: string }> = {}) => + f.store.beginPrePush(identity, f.reviewed(), f.store.getTask(identity).stateVersion, + { branch: BRANCH, from: oid(2), to: oid(12), ...input }); + + it('migrates a v17 store and reopens it unchanged', () => { + const { store, path } = fixture(); + store.close(); stores.splice(stores.indexOf(store), 1); + const legacy = new DatabaseSync(path); + legacy.exec('ALTER TABLE tasks DROP COLUMN push_in_progress; PRAGMA user_version=17;'); legacy.close(); + expect(open(path).getTask(identity).pushInProgress).toBeNull(); + expect(new DatabaseSync(path).prepare('PRAGMA user_version').get()).toEqual({ user_version: 18 }); + }); + + it('claims only a push of the current head over a durable rewrite predecessor', () => { + const f = rewritten(); + expect(() => begin(f, { from: oid(5) })).toThrow(/not a rewrite predecessor/); + expect(() => begin(f, { to: oid(2), from: oid(12) })).toThrow(/reviewed head changed/); + expect(() => begin(f, { branch: 'main' })).toThrow(/codeboost\/ task branch/); + const version = f.store.getTask(identity).stateVersion; + const marker = begin(f); + expect(f.store.getTask(identity)).toMatchObject({ stateVersion: version + 1, + pushInProgress: { attemptId: marker.attemptId, branch: BRANCH, from: oid(2), to: oid(12), snapshotId: f.store.getSnapshot(identity).id } }); + expect(() => begin(f)).toThrow(/Stale task state|push of the rewritten head is in progress/); + }); + + it('refuses a push claim for a task that is not mergeable, is being cancelled, or has an attempt or merge in flight', () => { + const queuedTask = rewritten(); + queuedTask.store.transitionTask(identity, queuedTask.store.getTask(identity).stateVersion, 'queued'); + expect(() => begin(queuedTask)).toThrow(/The task is queued; push it from review/); + + const attempting = rewritten(); + admit(attempting.store, { kind: 'check' }); + expect(() => begin(attempting)).toThrow(/attempt is still active/); + + const cancelling = rewritten(); + admit(cancelling.store, { kind: 'check' }); + expect(cancelling.store.cancelTask(identity, cancelling.store.getTask(identity).stateVersion, randomUUID())).toBe('stopping'); + expect(() => begin(cancelling)).toThrow(/being cancelled/); + + const merging = rewritten(); + merging.store.beginMergeAttempt(identity, merging.reviewed(), oid(12), null, 'direct', randomUUID(), merging.store.getTask(identity).stateVersion); + expect(() => begin(merging)).toThrow(/merge is in progress/); + }); + + it('blocks every other task write, rebase and merge until the push settles, and survives a restart', () => { + const f = rewritten(), marker = begin(f), version = () => f.store.getTask(identity).stateVersion; + expect(() => f.store.transitionTask(identity, version(), 'queued')).toThrow(/push of the rewritten head is in progress/); + expect(() => f.store.beginMergeAttempt(identity, f.reviewed(), oid(12), null, 'direct', randomUUID(), version())) + .toThrow(/push of the rewritten head is in progress/); + expect(() => f.store.beginRebase(identity, f.reviewed(), version(), { oldBase: oid(10), oldHead: oid(12), onto: oid(20), oldHistory: [oid(12)] })) + .toThrow(/push of the rewritten head is in progress/); + expect(() => f.store.recordRebase(identity, f.reviewed(), oid(20), oid(22), [{ oldSha: oid(12), newSha: oid(22) }])) + .toThrow(/push of the rewritten head is in progress/); + expect(() => f.store.admitAttempt(identity, { expectedStateVersion: version(), kind: 'check', item: 'P1', + expectedContext: f.store.currentContext(identity), deadline: later() })).toThrow(/push of the rewritten head is in progress/); + expect(() => f.store.setAssignment(identity, version(), 'someone', 'b'.repeat(40))).toThrow(/push of the rewritten head is in progress/); + f.store.close(); stores.splice(stores.indexOf(f.store), 1); + const reopened = open(f.path); + expect(reopened.pushesInProgress()).toEqual([{ planKey: reopened.getTask(identity).planKey, marker }]); + expect(reopened.finishPrePush(reopened.getTask(identity).planKey, randomUUID())).toBe(false); + expect(reopened.finishPrePush(reopened.getTask(identity).planKey, marker.attemptId)).toBe(true); + expect(reopened.getTask(identity).pushInProgress).toBeNull(); + expect(reopened.pushesInProgress()).toEqual([]); + }); + + it('defers a cancel requested during the push until its outcome is read, then closes the task', () => { + const f = rewritten(), marker = begin(f), actionId = randomUUID(); + expect(f.store.cancelTask(identity, f.store.getTask(identity).stateVersion, actionId)).toBe('stopping'); + expect(f.store.getTask(identity).status).toBe('in review'); + expect(f.store.finishPrePush(f.store.getTask(identity).planKey, marker.attemptId)).toBe(true); + expect(f.store.getTask(identity).status).toBe('cancelled'); + }); +}); diff --git a/test/runner-production.test.ts b/test/runner-production.test.ts index a15f6da4..bd5502d0 100644 --- a/test/runner-production.test.ts +++ b/test/runner-production.test.ts @@ -17,6 +17,7 @@ import { exportTaskDiff, removeTaskFilesystemsAsync } from '../agents/container/ import { RunnerCoordinator, type RunnerDeps } from '../runner/coordinator.ts'; import { SafetyFindings, type ExecutionSources } from '../runner/execution.ts'; import type { GitRebaser } from '../runner/rebase.ts'; +import { GitBranchPusher } from '../runner/branch-push.ts'; vi.mock('../agents/recovery.ts', async original => ({ ...await original(), recoverLeftovers: vi.fn() })); const issueReads: number[] = []; @@ -121,6 +122,29 @@ describe('runner startup', () => { expect(await assembly.sources.vendor(service.config.identity)).toBe('claude'); expect(() => assembly.sources.planContext({ ...service.config.identity, planId: 'other' })).toThrow(/only its configured plan/); }); + it('pushes a prepared rewrite only under an exact lease on the PR head it rewrote', async () => { + const { root, service } = fixture(), calls: string[] = []; + const assembly = await setUpRunner({ service, capability, config: { root: join(root, 'runner'), committer }, lock: lock(calls), env: { CLAUDE_CODE_OAUTH_TOKEN: 't' }, + buildImage: () => 'sha256:x', recovery: () => recovery(calls) }); + const push = vi.spyOn(GitBranchPusher.prototype, 'push').mockResolvedValue(); + const read = vi.spyOn(GitBranchPusher.prototype, 'readBranch').mockResolvedValue(null); + try { + const coordinator = assembly.preMerge!({} as RunnerCoordinator, async () => ({ base: 'a'.repeat(40), head: 'a'.repeat(40) }), + async () => ({ refresh: async () => undefined, validate: () => undefined })); + const beforePush = vi.fn(), from = 'c'.repeat(40), to = 'd'.repeat(40); + await coordinator.remote.push({ branch: 'codeboost/task-1', from, to, beforePush }); + expect(push).toHaveBeenCalledWith(service.config.identity, + expect.objectContaining({ head: to, branch: 'codeboost/task-1', expected: from }), undefined); + // The boundary only hands back the guard; the pusher runs it as its finalizer, synchronously, as Git starts. + const finalize = await push.mock.calls[0]![1].beforePush!(); + expect(beforePush).not.toHaveBeenCalled(); + expect(finalize).toBeTypeOf('function'); + (finalize as () => void)(); + expect(beforePush).toHaveBeenCalledOnce(); + await expect(coordinator.remote.readBranch('codeboost/task-1')).resolves.toBeNull(); + expect(read).toHaveBeenCalledWith(service.config.identity, 'codeboost/task-1', undefined); + } finally { push.mockRestore(); read.mockRestore(); } + }); it('reads the issue once for a run of items, and again once the reuse window has passed', async () => { const { root, service } = fixture(); service.store.setIssueTrust({ repository: 'owner/repo', issue: service.store.getPlan(service.config.identity).issue, diff --git a/test/runner-publishing.test.ts b/test/runner-publishing.test.ts index 6e0bdc0b..2939ce91 100644 --- a/test/runner-publishing.test.ts +++ b/test/runner-publishing.test.ts @@ -1369,7 +1369,7 @@ describe('closing a cancelled task\'s pull requests (#111)', () => { }; let writesClosed: boolean | undefined; const { close } = await serve(w, { issueGateway, preMerge: service => ({ - get active() { return false; }, get last() { return null; }, assertStartable() {}, + get active() { return false; }, get last() { return null; }, assertStartable() {}, settlePendingPush: async () => null, start: async () => { throw new Error('not used'); }, cancelTask: () => 'closed' as const, close: async () => { writesClosed = service.store.writesClosed; }, } as unknown as PreMergeCoordinator) }); @@ -1387,7 +1387,7 @@ describe('closing a cancelled task\'s pull requests (#111)', () => { }; let active = false, cancelId: string | undefined; const { app, identity, store } = await serve(w, { before: completeAll, issueGateway, preMerge: service => ({ - get active() { return active; }, get last() { return null; }, assertStartable() {}, + get active() { return active; }, get last() { return null; }, assertStartable() {}, settlePendingPush: async () => null, start: () => { active = true; return settle.promise.then(() => { @@ -1409,6 +1409,87 @@ describe('closing a cancelled task\'s pull requests (#111)', () => { expect(w.github.prs[0]!.open).toBe(false); }); + it('settles an interrupted pre-merge push only under the verified lock, and closes the PRs of a task it closed', async () => { + const w = world(), calls: string[] = []; + const issueGateway: IssueTrustGateway = { + repository: REPO, + async fetch() { return { repository: REPO, retrievedAt: new Date().toISOString(), issues: [] }; }, + async issueAccess(number) { return { number, authorLogin: 'member', collaborator: true }; }, + async issueText(number) { return { number, title: '', body: '', comments: [] }; }, + }; + const { app, identity, store, branch } = await serve(w, { before: completeAll, issueGateway, preMerge: service => ({ + get active() { return false; }, get last() { return null; }, assertStartable() {}, + // As the real job does: read the branch, then clear the marker it finds. + settlePendingPush: async () => { + calls.push('settle'); + const marker = service.store.getTask(service.config.identity).pushInProgress; + if (!marker) return null; + service.store.finishPrePush(service.store.getTask(service.config.identity).planKey, marker.attemptId); + return 'not-pushed' as const; + }, + start: async () => { throw new Error('not used'); }, cancelTask: () => { throw new Error('not used'); }, close: async () => undefined, + } as unknown as PreMergeCoordinator) }); + // serve() ran startup as the CLI does, after its (passing) lock check; nothing was pending then. + expect(calls).toEqual(['settle']); + await publishSettled(app, identity); + // The state a crash leaves: a push of the rewritten head whose outcome is unknown, and a cancel requested meanwhile. + const reviewed = () => ({ revision: store.getPlan(identity).revision, snapshotId: store.getSnapshot(identity).id, reviewVersion: store.reviewVersion(identity) }); + const head = store.getSnapshot(identity).head, rewritten = 'e'.repeat(40); + store.recordRebase(identity, reviewed(), 'f'.repeat(40), rewritten, [{ oldSha: head, newSha: rewritten }]); + store.beginPrePush(identity, reviewed(), store.getTask(identity).stateVersion, { branch, from: head, to: rewritten }); + expect(store.cancelTask(identity, store.getTask(identity).stateVersion, randomUUID())).toBe('stopping'); + // A failed lock check starts nothing. + expect(() => app.publishOwed(() => { throw new Error('The database path changed.'); })).toThrow(/database path changed/); + expect(calls).toEqual(['settle']); + app.publishOwed(() => { calls.push('verify'); }); + expect(calls).toEqual(['settle', 'verify', 'settle']); + await vi.waitFor(() => expect(store.lastPublish(identity)).toMatchObject(closedRecord), { timeout: 20_000, interval: 20 }); + expect(store.getTask(identity)).toMatchObject({ status: 'cancelled', pushInProgress: null }); + expect(w.github.prs[0]!.open).toBe(false); + }); + + it('settles a leftover pre-merge push when the task is cancelled with no preparation running, retrying on a repeated cancel', async () => { + const w = world(), settled: string[] = []; + let readable = false; + const issueGateway: IssueTrustGateway = { + repository: REPO, + async fetch() { return { repository: REPO, retrievedAt: new Date().toISOString(), issues: [] }; }, + async issueAccess(number) { return { number, authorLogin: 'member', collaborator: true }; }, + async issueText(number) { return { number, title: '', body: '', comments: [] }; }, + }; + const { app, identity, store, branch } = await serve(w, { before: completeAll, issueGateway, preMerge: service => ({ + get active() { return false; }, get last() { return null; }, assertStartable() {}, + // Startup finds no marker yet; the cancel below must start a settlement of its own. + settlePendingPush: async () => { + const marker = service.store.getTask(service.config.identity).pushInProgress; + if (!marker) return null; + settled.push(marker.attemptId); + // As the real job does when GitHub cannot be read: the marker stays. + if (!readable) return null; + service.store.finishPrePush(service.store.getTask(service.config.identity).planKey, marker.attemptId); + return 'not-pushed' as const; + }, + start: async () => { throw new Error('not used'); }, cancelTask: () => { throw new Error('not routed while inactive'); }, + close: async () => undefined, + } as unknown as PreMergeCoordinator) }); + await publishSettled(app, identity); + // An earlier preparation rewrote the head and could not read the branch after its push: the marker stayed. + const reviewed = () => ({ revision: store.getPlan(identity).revision, snapshotId: store.getSnapshot(identity).id, reviewVersion: store.reviewVersion(identity) }); + const head = store.getSnapshot(identity).head, rewritten = 'e'.repeat(40); + store.recordRebase(identity, reviewed(), 'f'.repeat(40), rewritten, [{ oldSha: head, newSha: rewritten }]); + const marker = store.beginPrePush(identity, reviewed(), store.getTask(identity).stateVersion, { branch, from: head, to: rewritten }); + expect((await act(app, 'cancel-task')).body.result).toEqual({ outcome: 'stopping' }); + await vi.waitFor(() => expect(settled).toEqual([marker.attemptId]), { timeout: 20_000, interval: 20 }); + expect(store.getTask(identity)).toMatchObject({ status: 'in review', pushInProgress: marker }); + // GitHub is reachable again: repeating the cancel reads the branch again instead of refusing. + readable = true; + expect((await act(app, 'cancel-task')).body.result).toEqual({ outcome: 'stopping' }); + await vi.waitFor(() => expect(store.lastPublish(identity)).toMatchObject(closedRecord), { timeout: 20_000, interval: 20 }); + expect(settled).toEqual([marker.attemptId, marker.attemptId]); + expect(store.getTask(identity)).toMatchObject({ status: 'cancelled', pushInProgress: null }); + expect(w.github.prs[0]!.open).toBe(false); + }); + it('closes a ready PR when a task in review is cancelled, and keeps the branch', async () => { const w = world(); const { app, identity, store, branch, head } = await serve(w, { before: completeAll }); diff --git a/test/store.test.ts b/test/store.test.ts index fcf945fb..18cbf060 100644 --- a/test/store.test.ts +++ b/test/store.test.ts @@ -121,7 +121,7 @@ it('scopes issue trust to repository and current author, supports revoke, and mi legacy.exec('DROP TABLE issue_trust; ALTER TABLE attempts DROP COLUMN prompt_comments; PRAGMA user_version=16;'); legacy.close(); const migrated = open(path), db = new DatabaseSync(path); try { - expect(db.prepare('PRAGMA user_version').get()).toEqual({ user_version: 17 }); + expect(db.prepare('PRAGMA user_version').get()).toEqual({ user_version: 18 }); expect(db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name='issue_trust'").get()).toEqual({ name: 'issue_trust' }); expect(db.prepare('PRAGMA table_info(attempts)').all().some(column => column.name === 'prompt_comments')).toBe(true); expect(migrated.issueTrust('owner/a', 5)).toBeNull(); diff --git a/web/server.ts b/web/server.ts index 42d705bc..6012081c 100644 --- a/web/server.ts +++ b/web/server.ts @@ -400,10 +400,24 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge if (service.store.getTask(identity).stateVersion !== expectedStateVersion) throw new GuardRefusal('Stale task state. Reload before writing.'); if ((action === 'start' || action === 'resume' || action === 'approve-continuation' || action === 'prepare-merge') && service.store.reviewVersion(identity) !== expectedReviewVersion) throw new GuardRefusal('Stale review state. Reload before writing.'); - if (action === 'cancel-task') return { outcome: preMerge?.active - ? preMerge.cancelTask(expectedStateVersion as number, actionId as string) - : runner ? runner.cancelTask(identity, expectedStateVersion as number, actionId as string) - : service.store.cancelTask(identity, expectedStateVersion as number, actionId as string) }; + if (action === 'cancel-task') { + // A cancel already waits on a push whose outcome is unknown, and the read that would settle it failed. Read the + // branch again instead of refusing, so repeating the cancel can still close the task without a restart. + const pending = service.store.getTask(identity); + if (preMerge && !preMerge.active && !stopping && pending.pushInProgress !== null && pending.cancelRequested !== null) { + service.store.afterCommit(() => afterPreMerge(preMerge!.settlePendingPush())); + return { outcome: 'stopping' as const }; + } + const outcome = preMerge?.active + ? preMerge.cancelTask(expectedStateVersion as number, actionId as string) + : runner ? runner.cancelTask(identity, expectedStateVersion as number, actionId as string) + : service.store.cancelTask(identity, expectedStateVersion as number, actionId as string); + // A push of a rewritten head whose outcome is still unknown holds this cancel at 'stopping'. With no preparation + // to settle it, read its branch now; clearing the marker closes the task, and then its PRs. + if (outcome === 'stopping' && preMerge && !preMerge.active && !stopping && service.store.getTask(identity).pushInProgress !== null) + service.store.afterCommit(() => afterPreMerge(preMerge!.settlePendingPush())); + return { outcome }; + } if (!runner) throw new GuardRefusal(config.demo ? RUNNER_NOT_IN_DEMO : RUNNER_NOT_CONFIGURED); if (action === 'prepare-merge') { if (access) requireTrustedIssue(access); @@ -742,9 +756,12 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge * Startup (#103): a publish an earlier process owed (a lost opening, a run that ended before its publish completed) * runs once, in the background; recovery finds a lost opening by its marker. A publish pushes, so `verifyLock` * (the runner lock still names the database) runs first, here: if it throws, nothing starts and its error is thrown. + * Under the same lock, a push of a rewritten head left unsettled by a crash or shutdown (F6a of #22) is settled by a + * read of its branch. That can close a task cancelled meanwhile, so it ends like a preparation: its PRs are closed. */ publishOwed: (verifyLock: () => void): void | Promise => { verifyLock(); + if (preMerge && !stopping) afterPreMerge(preMerge.settlePendingPush()); if (!publishing) return; publishing.startup(identity); },