diff --git a/AGENTS.md b/AGENTS.md index 4783d1f1..13891a34 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -21,6 +21,7 @@ For features with background jobs, polling, retries, cancellation, or shutdown: - Back off recurring external-status polling to a bounded cap. Reset the interval only after a meaningful lifecycle change or explicit user action. - When synchronous work moves behind an await (a worker thread, a subprocess, an async API), list every await it adds and apply the local-generation rule under "Guarded external actions" to each one. A local await must not follow the final remote validation: the last await before an irreversible external action is that validation. - Work that leaves a serializing path (the event loop, a lock, a single queue) needs its own concurrency bound. Queued work must honour cancellation and deadlines. +- Attach a rejection handler to a promise when it is created whenever anything is awaited before the promise is consumed. A rejection that settles in between reaches Node's process-wide unhandled-rejection handler, which by default exits the process. - Never compare clock readings across threads or processes. Pass a relative budget and let each side count it on its own monotonic clock. - A test that runs the server in the test's own process must wait for observable state before acting. Do not rely on the server blocking the test's event loop to order steps. @@ -50,6 +51,8 @@ Before opening or updating a PR for asynchronous behavior, test every applicable Every reproduced race requires a failing-before and passing-after regression. Assert both the visible result and the durable state when they can diverge. +A race regression must assert that the interleaving it names actually happened (for example which abort stopped which stage, or that a deadline fired before a release) before it uses the outcome as evidence. Otherwise machine load can turn it into a test of a different order that still passes. + ## Review readiness - Before requesting or re-requesting an automated Copilot review, self-review the full current diff, fix every issue found, and repeat the self-review and fix cycle until a complete pass finds no new issues. Re-run the relevant validation after fixes, then complete the mandatory independent-agent review loop below before requesting Copilot review. diff --git a/docs/implementation/guarded-merge.md b/docs/implementation/guarded-merge.md index fc56f535..00b0e189 100644 --- a/docs/implementation/guarded-merge.md +++ b/docs/implementation/guarded-merge.md @@ -29,9 +29,9 @@ Merging blocks when any plan item is unreviewed or stale, an attributed file is The GitHub adapter reads the current PR base/head, effective branch rulesets, classic branch protection, required check contexts and app identities, and issue cross-references. It ignores review records that are not required checks. The already-fixed check is the pre-PR check, `GhAlreadyFixedGateway` in `github/already-fixed.ts`, with the same rules (see “The check” in `pull-request-opening.md`): a closed issue, another open or merged PR that would close the issue or is linked manually, or a new commit on the base branch that mentions the issue. A PR that only mentions the issue does not count, in this repository or another; when the PR's base is not the default branch, a PR in this repository into that same base that references the issue counts too (decided 2026-09-30). The merge passes this PR, and every other PR the task records in this repository (#121), as the task's own PRs, matched by repository and number: each is excluded while open and counts once merged, as in the pre-PR check. Without a runner block only this PR is passed. The commit scan starts at the PR's base commit on its base branch. Merging requires that commit to equal the reviewed snapshot's base, which is the base the pre-PR check scanned from. The merge passes no own commits, because the task's commits reach the base branch only by this merge. Any result the check cannot complete is unknown and blocks merging, including a base branch name outside the pattern the check accepts. The check starts as soon as the PR is read, alongside the rule reads, and stops 0.65 seconds before the inspection's deadline (`MERGE_CHECK_SETTLE_MS` in `github/merge.ts`: both grace periods of the merge's `gh` runner, which the check also uses, plus a margin), so its processes settle before the inspection ends; a check stopped that way is unknown. If the rule reads fail, the check is stopped and awaited before the inspection rejects. When the PR read ends too late to leave that much time, the check is not started and is unknown. A check answer that arrives after the check was stopped is unknown, and an inspection whose caller aborted or whose deadline passed rejects even when its reads answer late. An injected check must name the merge repository. The blocker names what the check found (at most five closes, PRs and commits, by repository, number, state and short SHA, with a closing commit's SHA shortened too; never a commit message) or why it could not finish. Once this PR has merged, the check counts it as the fix, so the coordinator shows only the `pr-state` blocker, not an already-fixed one. Malformed pagination, rules without a type, malformed protection objects, missing or malformed check-app identities, and inconsistent check status/conclusion pairs fail closed, while an explicit `null` remains an unbound check. Display status is cached for five seconds and the combined inspection has a 12-second deadline. Each of the two sequential merge-validation inspections has a six-second deadline, keeping their combined validation budget below the 15-second serving request budget. A deadline aborts the underlying GitHub CLI process before releasing the shared in-flight request. A caller's own abort during the already-fixed check waits for its processes like any other `gh` call of the gateway, up to 0.4 seconds. An unreadable or incomplete rule source, a pending/missing/failing check, an already-fixed match or unknown result, a conflict, or a closed PR blocks merging. -Automatic merging also requires strict required status checks as a server-enforced current-base policy. A client-side fetch cannot supply that guarantee. Merge-queue branches were blocked in this increment because `gh pr merge` can report a successful enqueue before the pull request is merged. Queue support has since landed (#24; see `merge-queue.md`): a queue-capable adapter tracks each attempt until GitHub confirms the merge, removal or failure, and queue rules count as the server-side current-base guard. An adapter without queue inspection still fails closed with a `merge-queue` blocker. The coordinator performs the complete gate twice without using the display cache, verifies the same base/head pair, then re-reads the local review generation immediately before invoking `gh pr merge --match-head-commit` with literal argv. Each `gh` process gets only the allowlisted variables in `github/gh-env.ts`, plus fixed settings that turn off prompts, the pager, colour and update checks. No other server variables are passed on. A timeout or abort sends the `gh` process SIGTERM, then SIGKILL after a quarter of a second, and the call returns only after that process has exited (or 0.15 s after it exits, if a process it started keeps its output open). So a merge click aborted at its 14-second deadline settles by 14.4 s, inside the 14.5-second shutdown drain and below the 15-second request budget. Stopping `gh` cannot recall a merge request it had already sent: GitHub can still apply it, so a cancelled merge is reported as unknown and its attempt stays in flight, with the action disabled. A direct attempt is settled only when GitHub shows the pull request merged; a queued attempt also settles when GitHub reports it removed or failed (`merge-queue.md`). The adapter invalidates its pre-action cache before the command and after either success or refusal. A generation guard prevents inspections started before or during the command from restoring stale cache entries afterward. After command success, the server returns “merge submitted” even if its best-effort status reload fails, and the browser keeps the action disabled until Refresh confirms GitHub state. GitHub's command refusal text is returned to the local UI, and a failed attempt also remains disabled until Refresh loads fresh state. There is no “merge anyway” path around missing atomic protection. Demo mode never constructs the coordinator, including when a gateway is injected. +Automatic merging also requires strict required status checks as a server-enforced current-base policy. A client-side fetch cannot supply that guarantee. Merge-queue branches were blocked in this increment because `gh pr merge` can report a successful enqueue before the pull request is merged. Queue support has since landed (#24; see `merge-queue.md`): a queue-capable adapter tracks each attempt until GitHub confirms the merge, removal or failure, and queue rules count as the server-side current-base guard. An adapter without queue inspection still fails closed with a `merge-queue` blocker. The coordinator performs the complete gate twice without using the display cache, verifies the same base/head pair, then re-reads the local review generation immediately before invoking `gh pr merge --match-head-commit` with literal argv. Each `gh` process gets only the allowlisted variables in `github/gh-env.ts`, plus fixed settings that turn off prompts, the pager, colour and update checks. No other server variables are passed on. A timeout or abort sends the `gh` process SIGTERM, then SIGKILL after a quarter of a second, and the call returns only after that process has exited (or 0.15 s after it exits, if a process it started keeps its output open). So a merge call aborted at its 14-second deadline settles by 14.4 s. *Later (#48, #170):* a click can also be waiting on its issue-trust read, whose stop wait is 0.75 s, or on a review load, whose history worker gets 1 s after the abort before it is terminated; so it settles about 0.75 s after its deadline, or about 1 s or a little more after it when a review load was running, past the 14.5-second shutdown drain (traced, not tested); see [merge-queue.md](merge-queue.md). Stopping `gh` cannot recall a merge request it had already sent: GitHub can still apply it, so a cancelled merge is reported as unknown and its attempt stays in flight, with the action disabled. A direct attempt is settled only when GitHub shows the pull request merged; a queued attempt also settles when GitHub reports it removed or failed (`merge-queue.md`). The adapter invalidates its pre-action cache before the command and after either success or refusal. A generation guard prevents inspections started before or during the command from restoring stale cache entries afterward. After command success, the server returns “merge submitted” even if its best-effort status reload fails, and the browser keeps the action disabled until Refresh confirms GitHub state. GitHub's command refusal text is returned to the local UI, and a failed attempt also remains disabled until Refresh loads fresh state. There is no “merge anyway” path around missing atomic protection. Demo mode never constructs the coordinator, including when a gateway is injected. -Shutdown sets a terminal admission flag before inspecting active work and checks it again after partial request bodies are received. It then stops HTTP admission, aborts and awaits an active merge CLI process, drains requests, and closes the review service. A request admitted before shutdown cannot start a new merge afterward, and a cancelled command cannot outlive the local state that authorized it. +Shutdown sets a terminal admission flag before inspecting active work and checks it again after partial request bodies are received. It then stops HTTP admission, drains admitted requests, closes the pre-merge lifecycle and the Store write gate, and only then aborts and awaits an active merge CLI process (`web/server.ts`), so a refusal gh gave before the click was admitted, if it is handled only after the write gate closed, cannot be saved and answers a resendable 503 with the refusal's text; after admission, settlement keeps its shutdown capability and saves GitHub's refusal (#48). A request admitted before shutdown cannot start a new merge afterward, and a cancelled command cannot outlive the local state that authorized it. ## Deferred runner work diff --git a/docs/implementation/merge-queue.md b/docs/implementation/merge-queue.md index 5990c90d..0117761d 100644 --- a/docs/implementation/merge-queue.md +++ b/docs/implementation/merge-queue.md @@ -26,6 +26,29 @@ Transient or incomplete GitHub observations leave the attempt active, surface an Both pre-action validation reads must agree on whether merge queues apply. Queue mode performs one final fresh validation after capturing its timeline cursor, immediately before durable ownership and `gh pr merge`. Any mode change aborts before the command; the coordinator never decides whether to create durable queue ownership from an earlier observation. +## Stop reasons and answers (#48) + +A GitHub read that stops reports the first reason it stopped for, and the merge path and the server's trust read keep a timeout or a gh answer (`CommandFailed`, gh's own exit) that came before a later abort, as AGENTS.md requires: the original reason is kept and an actionable error is not replaced with cancellation text. Only what the abort itself stopped is answered with the abort (503 before admission, a 409 unknown outcome after it, or the queue poll's 409 shutdown error). Each path is listed below, with its exceptions. + +- **Gateway reads.** A GitHub read that runs out of its own time throws `RemoteTimeout` (`github/timeout.ts`): the merge-state inspection, the queue watermark, the queue inspection and the issue read. When the caller aborts later, while gh is still stopping, the read still reports its timeout. A failure gh reported before either abort (its own exit, `CommandFailed`) is reported as itself, not replaced by the abort, except in the rules read, which treats any failure as unknown rules (traced, not tested). When one inspection stage's gh exit stops the other stages, the inspection waits for every stage to settle and then reports that exit, even when its own deadline fired meanwhile (tested). The already-fixed check's result is handled from the moment it starts, so a check that a failed stage stopped and that settles first is not an unhandled rejection, which Node would raise process-wide (found by the #48 tests; it predated #48; tested); a successful answer that arrives after the abort is still discarded, and so is an error the gateway raised while checking a successful answer (for example a different issue returned; traced, not tested). An inspection stage's own validation error that comes before the abort, while the other stages are still settling, is also answered with the abort: resendable, never a wrong saved answer (traced, not tested). The issue gateway's rule also covers execution's issue text and the issue board's refresh; the board's prioritizer still answers a refresh stopped by shutdown with the shutdown reason (traced, not tested). +- **Issue-trust reads.** The server wraps a failed trust read as an upstream failure and keeps the timeout as its cause. On the merge path, a trust read that timed out, or that gh answered, before the caller's abort still reports that; the runner actions keep a trust read's own timeout too, even when shutdown started while gh stopped (tested), and answer any other failed trust read as shutting down once shutdown has started (the server is stopping, the runner is closing, or the request was aborted, which only shutdown does). Either way nothing is admitted. + - The trust read that gates the runner actions (start, resume, approve-continuation, publish, prepare-merge) is not saved as the action's outcome when it times out (tested for start; the other actions share the same branch): the action answers 503, and the same action ID reads trust again; an outcome already saved under that ID is still the answer, also when shutdown started while gh stopped (tested). + - The trust read inside merge preparation (`runner/pre-merge.ts`) is different: a timeout or gh answer there settles the preparation as failed, which is saved and replayed, and a shutdown that arrives after it is reported as that timeout or answer (tested: the preparation layer in `test/pre-merge.test.ts`, with the error shape the server's trust read builds, and that read itself through publishing in `test/runner-publishing.test.ts`); a cancel takes the same path (traced, not tested). The rebase conflict resolver's trust read (`runner/rebase-conflict.ts`) is not covered: `Rebaser` (`runner/rebase.ts`) reports a cancel, shutdown or expired rebase deadline that is pending when the resolver's error is handled in its place. This is unchanged by #48 and left as a follow-up (#174). + - A publish whose trust read timed out, or that gh answered, before shutdown or cancellation is recorded as failed with that reason, not as stopped (tested for shutdown, `test/runner-publishing.test.ts`); cancellation takes the same path, both leave the publish owed, and a cancelled task's PRs are still closed (traced, not tested). + - Execution's issue read reports its own timeout when a cancel arrives after it; the runner decides the attempt's outcome from its recorded cancel, not from that error (traced, not tested). + - Out of this change: the issue board's trust toggle still saves a timed-out author read as that action's failure, and planning's issue reads (`web/server.ts`, `web/planning.ts`) still answer a timeout as an upstream failure without that cause. +- **Before admission.** A `RemoteTimeout` (even wrapped), the merge deadline or shutdown applied nothing. The click answers 503 and may be resent; it is never saved as the click's definite refusal. An outcome another coordinator saved under the same click meanwhile is the answer instead, also when this coordinator's own abort followed the timeout (tested). A replay after the click's abort loads no review, starts or joins no GitHub read, and asks for a refresh (tested). A replay before it reads GitHub fresh under the click's own signal; if that read fails, the saved outcome is answered with a refresh blocker instead of the page status's own blocker (traced, not tested). A gh answer that came before a later abort, and that the read reports as its error, is handled as it would be without the abort: it is saved as the click's refusal. A result built from a gh failure, such as the already-fixed check's unknown blocker, that arrives after the abort is discarded for the abort instead. That holds for the merge deadline (tested through the server's trust refresh just before admission; the first trust read settles within 12.75 s of its start: its own 12 s plus gh's stop wait; it starts after the task and review are loaded, so on a slow load it can still be running when the click's 14 s deadline fires, which takes the same path). At shutdown the server closes the Store's write gate before it aborts the coordinator. A refusal handled during the drain is still saved; one handled after the write gate closed cannot be: the click answers 503 with gh's text, and nothing is saved. Any refusal that cannot be saved now keeps its text in that 503. Clicks are refused once shutdown starts. The drain waits up to 14.5 s, while a click settles only after the step it was in has stopped: about 14.75 s when that is a gh call (its 14 s deadline plus the slowest gh stop wait, the trust read's 0.75 s; the already-fixed check runs on the merge gateway's runner, with 0.4 s), and about 15 s or a little more when it is a review load, whose history worker gets 1 s after the abort before it is terminated (#170; traced, not tested), and the drain ends early when the server has no open connections (a click whose browser went away). When the drain times out, the click's own deadline has already aborted it (clicks are refused once shutdown starts, so its 14 s ends before the 14.5 s drain does); gh's answer, which can surface up to 0.75 s after that abort, is saved if it is handled before the write gate closes and answers 503 otherwise. When the drain ends early, the coordinator's shutdown aborts the click after the write gate closed; the coordinator test that closes writes before shutdown models that path. gh's own 30-second timeout is a plain error, but every read before admission has a shorter `RemoteTimeout` (traced, not tested). + - The already-fixed check reports its timeout as a blocker instead (`alreadyFixed: 'unknown'`), and a blocker is saved and replayed like a refusal: it says the merge is not ready on the state read for this click, which a new click reads again. Treating it as resendable is a possible follow-up. + - A rules read cannot report a timeout that way: the inspection's own deadline fires first. A gh failure inside the inspection (for example the PR read or a published target's PR list) that came before its deadline is reported as itself and saved like a refusal with gh's text (traced, not tested). The already-fixed check turns its own gh failure into an `alreadyFixed: 'unknown'` blocker, also saved; its own timer stops it before the inspection's deadline. When a caller's abort comes after that failure, while the check is still settling, the blocker arrives after the abort and is discarded like any other answer: the inspection reports the abort, which may be resent (tested). After an abort, every other error passes through the check, including the abort's own reason. When that reason is a failed stage's exit, the inspection reports the failed stage's own rejection, the same exit (tested); for the caller's reason or an error that is not gh's, it answers with the abort (traced, not tested). +- **gh's exit.** Once gh's exit has been observed, its exit status is the answer (`github/run-with-input.ts`). A deadline or shutdown that fires while its output pipes drain does not turn GitHub's refusal, or its success, into an unknown outcome. The rule holds for every gh call: a publish whose gh command had already failed when it was cancelled or stopped is recorded as failed or refused with gh's message rather than as stopped, and still stays owed; one whose gh command had already succeeded goes on, and its next external step (a gh call or a git push) stops at once on the aborted signal (traced, not tested). +- **After admission.** GitHub's refusal is the click's answer even when the coordinator's deadline or shutdown came later, including a refusal whose failed state could not be saved (then answered as an unknown outcome with GitHub's text). Any other gh answer is reported with its text as an unknown outcome; a failure the abort itself caused reports the abort, as an unknown outcome. +- **Queue reads.** A queue read that timed out, or that gh answered, before shutdown answers with that as an observation error (200), not with the shutdown error (409). A queue read that fails answers from the attempt on record, which may have become queued or been replaced by a retry while GitHub was read; the observation error is attached only to the attempt that was read. Reading that attempt is a store read; if it fails, the poll fails (503) instead of answering with the observation error. +- **A merge click's answer.** It takes its readiness from the attempt on record, so after a submission it does not offer Merge again. Its GitHub readiness was read before the command, so it never offers an action from it: if a queue poll ended the attempt while the server reloaded the review, the answer is not ready and asks for a refresh (tested), and a retry is offered only after a fresh GitHub read (traced, not tested). A coordinator built without a store (only tests do this; the demo server has no merge coordinator) has no attempt on record, so its answer asks for a refresh the same way (traced, not tested). + +The status cache in `GhMergeGateway` is cleared before and after every merge command, successful or refused, and a read that started before the merge cannot fill it afterwards. Admission always reads GitHub fresh; only the read-only page status uses the cache. Publishing (push, ready, close) does not clear it, so the page may show the status from before publishing for up to five seconds; no merge decision uses that status. The clearing and the refused stale fill are tested; that admission always reads fresh and the publishing staleness were traced in the #48 investigation, not tested. + ## Acceptance evidence Controlled unit and browser regressions cover enqueue success, delayed merge, queue removal, unmergeable failure, head replacement, retry, stale observation publication, active-attempt retry refusal, shutdown settlement, durable restart recovery, and preservation of UI input while polling. Final evidence is recorded against the exact pushed PR head. + +#48 regressions: `test/merge-stop-reasons.test.ts` (stop reasons and resendable timeouts; a saved success answered after the click's own abort with no further load or GitHub read; output past the limit while the pipes drain still fails the command; an outcome saved meanwhile answered after a timeout and this click's own abort; gh's answer kept when shutdown finds writes closed; one inspection stage's gh exit reported after its stopped siblings settle, including past the deadline, with no unhandled rejection; a caller abort answered in place of the already-fixed check's own earlier gh failure), `test/merge-queue-poll.test.ts` (a failed, timed-out or gh-answered queue read answers from the current attempt; GitHub's refusal survives a later shutdown, saved or not; the click's answer is inactive, also when a poll ends the attempt before the answer is built; an issue-trust timeout answers 503, alone or before the merge deadline, and a gh answer before the deadline is saved and replayed), `test/merge-status-cache.test.ts` (cache evidence; these passed before the fixes), `test/pre-merge.test.ts` and `test/runner-publishing.test.ts` (a preparation's or a publish's trust read that timed out or that gh answered before shutdown is saved or recorded as failed with that answer, not as the shutdown), and `test/runner-start.test.ts` (a runner action's timed-out trust read is not saved and answers 503, also when shutdown started while gh stopped; a saved outcome still wins). The browser's duplicate merge polls are fixed separately on top of F6b, which rewrites that polling. diff --git a/docs/implementation/read-only-review.md b/docs/implementation/read-only-review.md index bab7c8cf..ee92c3cb 100644 --- a/docs/implementation/read-only-review.md +++ b/docs/implementation/read-only-review.md @@ -143,5 +143,5 @@ Review loading used to run history reading and linking on the server's event loo - **Cancellation.** An abort (a request ending at shutdown, a merge or preparation deadline) sends the worker a cancel. The worker kills its running Git process, waits for it to exit, and replies. A worker that has not replied within one second is terminated: one in synchronous work (linking, blob decoding, the object storage walk), which runs no Git process, or, rarely, one whose killed Git process is stuck in the kernel, which then outlives the load. The load rejects with the abort's own reason. - **Staleness.** The plan, snapshot, review version and ledger ownership are read before the worker starts. Every ledger write records a new snapshot, so an unchanged review version, plan revision and snapshot after the worker mean its inputs are current. If they changed (another load recorded a new HEAD, or a write landed), the load reads the history again from the current state, in at most three reads in all within its deadline, and is refused as stale only after that. - **Review commands.** A command applies, synchronously, to a review loaded just before it (inside the recorded action when it has an action ID). Its writes are guarded by that review's state. A load failure before the command is not recorded under its action ID, so the same ID can be resent. If the reload after the command was applied fails, or shutdown stops it, the refusal says `outcomeUnknown`, and a question saved during shutdown still gets its retryable "Server stopped" answer. -- **Merge and preparation.** Every merge load takes the click's operation signal; a replayed click's status refresh takes the coordinator's shutdown signal, and shutdown waits for it. Every preparation load takes the preparation's signal. A review load awaits a worker, so neither runs one after its final remote read: merge admission inspects the pull request once more after its final review load, and preparation inspects it after its last review load. Both then check local trust and the review's generations in the same turn as the write. Preparation also checks its generations after reading the old history it rebases and before it records a moved pull request head. Each write is either in the same turn as those checks, or (a rebase, after its authorization read) guarded again by the Store in its own transaction. +- **Merge and preparation.** Every merge load takes the click's operation signal; a replayed click's status refresh takes the coordinator's shutdown signal, and shutdown waits for it; a replay reached while the click is still running also takes the click's own signal, and a click already aborted reads nothing (#48). Every preparation load takes the preparation's signal. A review load awaits a worker, so neither runs one after its final remote read: merge admission inspects the pull request once more after its final review load, and preparation inspects it after its last review load. Both then check local trust and the review's generations in the same turn as the write. Preparation also checks its generations after reading the old history it rebases and before it records a moved pull request head. Each write is either in the same turn as those checks, or (a rebase, after its authorization read) guarded again by the Store in its own transaction. diff --git a/docs/implementation/runner-lifecycle.md b/docs/implementation/runner-lifecycle.md index 42a58bc6..afe01249 100644 --- a/docs/implementation/runner-lifecycle.md +++ b/docs/implementation/runner-lifecycle.md @@ -368,13 +368,13 @@ F1 owns the production planning endpoints that G4 needs. They wrap E3's coordina | `POST /api/plan/drafts/:id/cancel` | as suggestion cancel | request is a draft; a `pending` or `ready` one is cancelled, any other state is returned unchanged | | `POST /api/plan/drafts/:id/apply` | `Store.applyDraft`: imports the draft as revision + 1 with every import check | request is a draft, `ready` and bound to the current revision and snapshot | -Live planning runs for a non-demo review with a `github` block (#117): `web/planning.ts` reads the issue from GitHub before each start, and `runner/planning.ts` runs the request in planning's worker. E3's suggestion timer and the request share one ten-minute budget (`PLANNING_BUDGET_MS`). At shutdown E3 aborts every request; one that has not settled within 20 seconds (`PLANNING_SHUTDOWN_GRACE_MS`, as Ask) is ended by abandoning planning's worker. A start replays its recorded outcome before reading GitHub; a GitHub failure is a 502 and is not recorded. Without a `github` block, or in a demo, the server returns "Planning agent not available yet". Drafts follow the same rules (#124). A draft and a suggestion are both rows of `requests`, told apart by `mode` (schema v11); each kind's routes refuse the other's IDs. At startup, once the single-runner lock is verified to name the database (`web/planning.ts`), every request an earlier process left `pending` is failed ("The server stopped before this request finished. Ask again."): its provider ran in that process's planning worker, so nothing can complete it. +Live planning runs for a non-demo review with a `github` block (#117): `web/planning.ts` reads the issue from GitHub before each start, and `runner/planning.ts` runs the request in planning's worker. E3's suggestion timer and the request share one ten-minute budget (`PLANNING_BUDGET_MS`). At shutdown E3 aborts every request; one that has not settled within 20 seconds (`PLANNING_SHUTDOWN_GRACE_MS`, as Ask) is ended by abandoning planning's worker. A start replays its recorded outcome before reading GitHub; a GitHub failure is a 502 and is recorded for replay like a refusal (`web/server.ts`; `test/planning-production.test.ts`). Without a `github` block, or in a demo, the server returns "Planning agent not available yet". Drafts follow the same rules (#124). A draft and a suggestion are both rows of `requests`, told apart by `mode` (schema v11); each kind's routes refuse the other's IDs. At startup, once the single-runner lock is verified to name the database (`web/planning.ts`), every request an earlier process left `pending` is failed ("The server stopped before this request finished. Ask again."): its provider ran in that process's planning worker, so nothing can complete it. ## Feedback-event contract (for lane J) **What becomes an event.** Only feedback the user wrote or chose. Issue text, issue comments and agent output never become events. -`actionId` identifies the one user action (or confirmed external outcome) that caused the event. **For user actions it is an idempotency key the UI generates** (a UUID) when the user acts. The server validates it before opening any transaction: it must be a lowercase UUID v4 (the same regular expression as attempt IDs, 36 characters), or the request is refused with HTTP 400 and nothing is stored. The UI generates it and resends unchanged if it retries after a lost response. Every writing user action goes through one `user_actions` table, keyed by `(planKey, actionId)`. That covers notes, choice changes, finding acceptance, rejection, cancel, retry, "run again", start and resume, **starting a merge**, and every planning endpoint that writes: plan import, suggestion start, suggestion cancel, suggestion apply, draft start, draft cancel and draft apply (#124). Each planning request carries an `actionId`; its `user_actions` row is inserted in the same transaction as the `Store` call it wraps, and the saved response includes the request ID or plan revision it produced. For a merge, the `user_actions` row is inserted in the same transaction as `beginMergeAttempt`, and it saves the merge attempt ID with the response "in progress". When the merge coordinator records the outcome, the same transaction updates that saved response. A replayed merge click therefore returns the same attempt's current outcome and never starts a second attempt. The row holds the action kind, a hash of the request body and the bounded response. **The first definite outcome is recorded, including a refusal.** If the action applies, its row is inserted in the same transaction. If a guard refuses it (for example HTTP 409 for a stale state version), the refusal and its response are recorded in their own transaction. Only outcomes that applied nothing for a passing reason are not recorded, so the UI may resend them: HTTP 503 during shutdown, HTTP 502 when the issue cannot be read from GitHub before a suggestion or draft starts (#117), and a storage error. A replay is handled before any other guard: +`actionId` identifies the one user action (or confirmed external outcome) that caused the event. **For user actions it is an idempotency key the UI generates** (a UUID) when the user acts. The server validates it before opening any transaction: it must be a lowercase UUID v4 (the same regular expression as attempt IDs, 36 characters), or the request is refused with HTTP 400 and nothing is stored. The UI generates it and resends unchanged if it retries after a lost response. Every writing user action goes through one `user_actions` table, keyed by `(planKey, actionId)`. That covers notes, choice changes, finding acceptance, rejection, cancel, retry, "run again", start and resume, **starting a merge**, and every planning endpoint that writes: plan import, suggestion start, suggestion cancel, suggestion apply, draft start, draft cancel and draft apply (#124). Each planning request carries an `actionId`; its `user_actions` row is inserted in the same transaction as the `Store` call it wraps, and the saved response includes the request ID or plan revision it produced. For a merge, the `user_actions` row is inserted in the same transaction as `beginMergeAttempt`, and it saves the merge attempt ID with the response "in progress". When the merge coordinator records the outcome, the same transaction updates that saved response. A replayed merge click therefore returns the same attempt's current outcome and never starts a second attempt. The row holds the action kind, a hash of the request body and the bounded response. **The first definite outcome is recorded, including a refusal.** If the action applies, its row is inserted in the same transaction. If a guard refuses it (for example HTTP 409 for a stale state version), the refusal and its response are recorded in their own transaction. Only outcomes that applied nothing for a passing reason are not recorded, so the UI may resend them: HTTP 503 during shutdown, HTTP 503 when a runner action's issue-trust read at the gate or a merge click's GitHub read before admission ran out of time (#48; a timed-out trust read inside merge preparation settles the preparation as failed, which is recorded; the already-fixed check's timeout is a saved blocker instead, see [merge-queue.md](merge-queue.md)), HTTP 503 for a merge click's refusal from before admission that is handled only after shutdown closed the write gate (#48, same document), and a storage error. A replay is handled before any other guard: - the same `actionId` with the same request hash returns the saved response and applies nothing, even though the state version has since moved on; - the same `actionId` with a different request hash (including a different action kind) is refused with HTTP 409 "Action ID already used" and applies nothing. diff --git a/github/already-fixed.ts b/github/already-fixed.ts index ee882239..4439d1a7 100644 --- a/github/already-fixed.ts +++ b/github/already-fixed.ts @@ -131,7 +131,8 @@ export class GhAlreadyFixedGateway implements AlreadyFixedGateway { if (!SHA.test(input.taskBase)) throw new Error('Invalid task base commit.'); if (!BRANCH.test(input.baseBranch)) throw new Error('Invalid base branch name.'); if (input.ownPullRequests.some(number => !Number.isSafeInteger(number) || number < 1)) throw new Error('Invalid pull request number.'); - // Every stage shares one deadline; reaching it aborts the running `gh` call and makes the check unknown. + // Every stage shares one deadline; reaching it aborts the running `gh` call and makes the check unknown. A `gh` call + // that exited before the deadline keeps its answer (#48), so a check whose stages all answered in time still returns it. // The two stages are independent and run together. The first failure stops the other, and the check still waits for // both to settle, so it never returns while a `gh` process it started is running. const deadline = AbortSignal.timeout(this.deadlineMs), failed = new AbortController(); diff --git a/github/issues.ts b/github/issues.ts index 174e70ab..218d2fea 100644 --- a/github/issues.ts +++ b/github/issues.ts @@ -1,6 +1,7 @@ import { ghEnvironment } from './gh-env.ts'; import { dataJSON, MAX_PROMPT_BYTES } from '../core/planning-author.ts'; -import { runWithInput } from './run-with-input.ts'; +import { ghAnswered, runWithInput } from './run-with-input.ts'; +import { RemoteTimeout } from './timeout.ts'; const PAGE_SIZE = 100; const MAX_PAGES = 10; @@ -25,7 +26,7 @@ export async function withIssueReadDeadline(signal: AbortSignal, signal.throwIfAborted(); const deadline = new AbortController(); const shared = AbortSignal.any([signal, deadline.signal]); - const timer = setTimeout(() => deadline.abort(new Error('Issue retrieval timed out.')), ISSUE_READ_ACTIVE_MS); + const timer = setTimeout(() => deadline.abort(new RemoteTimeout('Issue retrieval timed out.')), ISSUE_READ_ACTIVE_MS); try { const value = await read(shared, ISSUE_READ_ACTIVE_MS); shared.throwIfAborted(); @@ -373,14 +374,15 @@ export class GhIssueGateway implements IssueGateway { const controller = new AbortController(); const relay = () => controller.abort(options.signal?.reason); options.signal?.addEventListener('abort', relay, { once: true }); - const timer = setTimeout(() => controller.abort(new Error('Issue retrieval timed out.')), timeoutMs); + const timer = setTimeout(() => controller.abort(new RemoteTimeout('Issue retrieval timed out.')), timeoutMs); try { const value = await load(controller.signal); controller.signal.throwIfAborted(); return value; } catch (error) { - if (options.signal?.aborted) throw options.signal.reason; - if (controller.signal.aborted) throw new Error('Issue retrieval timed out.'); + // The controller keeps the reason of whichever abort came first, the caller's or the timeout's. A CommandFailed is + // gh's own answer, given before the abort, and is reported as itself (#48). + if (controller.signal.aborted && !ghAnswered(error)) throw controller.signal.reason; throw error; } finally { clearTimeout(timer); diff --git a/github/merge.ts b/github/merge.ts index 95893a34..a113cda2 100644 --- a/github/merge.ts +++ b/github/merge.ts @@ -1,14 +1,18 @@ import { GhAlreadyFixedGateway, type AlreadyFixedGateway, type AlreadyFixedMatch } from './already-fixed.ts'; import { ghEnvironment } from './gh-env.ts'; import { markerOf } from './pull-requests.ts'; -import { runWithInput } from './run-with-input.ts'; +import { ghAnswered, runWithInput } from './run-with-input.ts'; +import { RemoteTimeout } from './timeout.ts'; import { BRANCH, REPOSITORY, SHA } from './validate.ts'; /** * How long a stopped `gh` gets after SIGTERM before SIGKILL, and how long its inherited output pipes may stay open after * it exits. Every gh call of this gateway settles only when gh has stopped, so a call aborted at a deadline settles up - * to 0.4 s later. That keeps a merge click (14 s deadline in runner/merge.ts) within 14.4 s, inside the 14.5 s shutdown - * drain and below the 15-second serving request budget, and an inspection (at most 12 s) within 12.4 s. + * to 0.4 s later, and an inspection (at most 12 s) within 12.4 s. A merge click (14 s deadline in runner/merge.ts) can + * also be waiting on its issue-trust read, whose stop wait is 0.75 s (github/issues.ts), or on a review load, whose + * history worker gets 1 s after the abort before it is terminated (runner/history-job.ts). So it settles about 0.75 s + * after its deadline, or about 1 s or a little more when a review load was running: past the 14.5 s shutdown drain + * (docs/implementation/merge-queue.md). */ export const MERGE_KILL_GRACE_MS = 250, MERGE_PIPE_GRACE_MS = 150; /** @@ -215,7 +219,11 @@ export class GhMergeGateway implements MergeGateway, MergeQueueGateway { if (based && result.outcome === 'found' && Array.isArray(result.matches) && result.matches.length) return { alreadyFixed: 'found', alreadyFixedDetail: describeMatches(result.matches) }; return { alreadyFixed: 'unknown', alreadyFixedDetail: result.outcome === 'unknown' && typeof result.reason === 'string' ? result.reason.slice(0, 300) : 'The check returned an invalid result.' }; } catch (error) { - if (signal?.aborted) throw error; + // After an abort, the check's own gh failure stays its result, as without the abort: an unknown blocker, which the + // inspection then discards for the abort (#48). Any other error is rethrown, including the abort's own reason; a + // failed stage's error is that reason, so a check it stopped rejects with that error, which changes no answer: the + // failed stage's own rejection is checked first. + if (signal?.aborted && (error === signal.reason || !ghAnswered(error))) throw error; return { alreadyFixed: 'unknown', alreadyFixedDetail: stop.signal.aborted ? 'The check did not finish in time.' : 'GitHub could not be read.' }; } finally { clearTimeout(timer); } } @@ -234,12 +242,15 @@ export class GhMergeGateway implements MergeGateway, MergeQueueGateway { // process they started outlives the inspection. const failed = new AbortController(), reads = signal ? AbortSignal.any([signal, failed.signal]) : failed.signal; const stopOthers = (error: unknown) => { failed.abort(error); throw error; }; - const alreadyFixed = this.#alreadyFixed(target.ownPullRequests, pr.baseRefName, base, deadlineAt, reads); + // Handled from the start: a stopped check can settle before the other stages, and an unhandled rejection meanwhile + // would reach Node's process-wide handler (#48). + const alreadyFixed = this.#alreadyFixed(target.ownPullRequests, pr.baseRefName, base, deadlineAt, reads) + .catch((error: unknown) => error instanceof Error ? error : new Error(String(error))); const [ruleRead, listRead] = await Promise.allSettled([ this.#rules(pr.baseRefName, pr.statusCheckRollup as Array>, reads).catch(stopOthers), published ? this.#branchOpen(target.headBranch!, reads).catch(stopOthers) : undefined, ]); - const fixed = await alreadyFixed.catch(error => error instanceof Error ? error : new Error(String(error))); + const fixed = await alreadyFixed; for (const read of [ruleRead, listRead]) if (read.status === 'rejected') throw read.reason; if (fixed instanceof Error) throw fixed; const rules = (ruleRead as PromiseFulfilledResult).value; @@ -373,12 +384,15 @@ export class GhMergeGateway implements MergeGateway, MergeQueueGateway { const generation = this.#generation; if (!options.fresh && this.#inflight?.generation === generation && this.#inflight.key === key) return this.#inflight.promise; const timeout = new AbortController(), deadlineAt = Date.now() + timeoutMs; - const timer = setTimeout(() => timeout.abort(new Error('GitHub merge-state inspection timed out.')), timeoutMs); + const timer = setTimeout(() => timeout.abort(new RemoteTimeout('GitHub merge-state inspection timed out.')), timeoutMs); const signal = options.signal ? AbortSignal.any([options.signal, timeout.signal]) : timeout.signal; const attempt = this.#inspectNow(target, deadlineAt, signal).catch(error => { // The combined signal keeps the reason of whichever abort came first, the caller's or the deadline's; a stopped gh - // call can settle after both have fired. - if (signal.aborted) throw signal.reason; + // call can settle after both have fired. A CommandFailed is a gh exit, reported as itself (#48): gh exited before + // the abort. When one stage's gh exit stops the other stages, the inspection waits for every stage to settle and + // then reports that exit first, whatever the stopped stages settled with, even when its deadline fired meanwhile + // (test/merge-stop-reasons.test.ts). A gh call that the abort stopped fails with the abort's reason instead. + if (signal.aborted && !ghAnswered(error)) throw signal.reason; throw error; }).then(state => { if (this.#generation === generation) this.#cache = { key, expiresAt: Date.now() + 5_000, state }; @@ -396,7 +410,7 @@ export class GhMergeGateway implements MergeGateway, MergeQueueGateway { const [owner, name] = this.config.repository.split('/') as [string, string]; const query = `query($owner:String!,$name:String!,$number:Int!){repository(owner:$owner,name:$name){pullRequest(number:$number){number headRefOid timelineItems(last:1,itemTypes:[ADDED_TO_MERGE_QUEUE_EVENT,REMOVED_FROM_MERGE_QUEUE_EVENT]){edges{cursor node{id}}}}}}`; const timeout = new AbortController(); - const timer = setTimeout(() => timeout.abort(new Error('GitHub merge-queue watermark timed out.')), timeoutMs); + const timer = setTimeout(() => timeout.abort(new RemoteTimeout('GitHub merge-queue watermark timed out.')), timeoutMs); const signal = options.signal ? AbortSignal.any([options.signal, timeout.signal]) : timeout.signal; try { if (options.signal?.aborted) throw options.signal.reason; @@ -419,8 +433,9 @@ export class GhMergeGateway implements MergeGateway, MergeQueueGateway { throw new Error('GitHub returned an invalid merge-queue watermark.'); return (edge as { cursor: string }).cursor; } catch (error) { - if (options.signal?.aborted) throw options.signal.reason; - if (timeout.signal.aborted && !options.signal?.aborted) throw new Error('GitHub merge-queue watermark timed out.'); + // The combined signal keeps the reason of whichever abort came first, the caller's or the deadline's. A CommandFailed + // is gh's own answer, given before the abort, and is reported as itself (#48). + if (signal.aborted && !ghAnswered(error)) throw signal.reason; throw error; } finally { clearTimeout(timer); } } @@ -436,7 +451,7 @@ export class GhMergeGateway implements MergeGateway, MergeQueueGateway { const [owner, name] = this.config.repository.split('/') as [string, string]; const query = `query($owner:String!,$name:String!,$number:Int!,$after:String){repository(owner:$owner,name:$name){pullRequest(number:$number){number headRefOid state mergedAt mergeQueueEntry{id state position enqueuedAt headCommit{oid} pullRequest{number headRefOid}} timelineItems(first:100,after:$after,itemTypes:[ADDED_TO_MERGE_QUEUE_EVENT,REMOVED_FROM_MERGE_QUEUE_EVENT]){edges{cursor node{id __typename ... on AddedToMergeQueueEvent{createdAt} ... on RemovedFromMergeQueueEvent{createdAt reason beforeCommit{oid}}}} pageInfo{hasNextPage endCursor}}}}}`; const timeout = new AbortController(); - const timer = setTimeout(() => timeout.abort(new Error('GitHub merge-queue inspection timed out.')), timeoutMs); + const timer = setTimeout(() => timeout.abort(new RemoteTimeout('GitHub merge-queue inspection timed out.')), timeoutMs); const signal = options.signal ? AbortSignal.any([options.signal, timeout.signal]) : timeout.signal; try { if (options.signal?.aborted) throw options.signal.reason; @@ -516,8 +531,9 @@ export class GhMergeGateway implements MergeGateway, MergeQueueGateway { if (correlated && (attemptEvents.length !== 2 || attemptEvents[1]?.type !== 'RemovedFromMergeQueueEvent')) throw new Error('GitHub returned an ambiguous terminal event sequence for the current enqueue attempt.'); return { state: 'removed', reviewedHead, removedAt: last.createdAt, reason: last.reason! }; } catch (error) { - if (options.signal?.aborted) throw options.signal.reason; - if (timeout.signal.aborted && !options.signal?.aborted) throw new Error('GitHub merge-queue inspection timed out.'); + // The combined signal keeps the reason of whichever abort came first, the caller's or the deadline's. A CommandFailed + // is gh's own answer, given before the abort, and is reported as itself (#48). + if (signal.aborted && !ghAnswered(error)) throw signal.reason; throw error; } finally { clearTimeout(timer); } } @@ -535,7 +551,10 @@ export class GhMergeGateway implements MergeGateway, MergeQueueGateway { } catch (error) { if (error instanceof MergeSubmissionError) throw error; const message = error instanceof Error ? error.message : 'GitHub merge submission failed with an unknown outcome.'; - const outcome = !options.signal?.aborted && confirmedMergeRefusal(message) ? 'refused' : 'unknown'; + // GitHub's refusal stays definite though an abort arrived after gh had already answered (a CommandFailed is gh's + // own exit); an error from the abort itself is never a refusal (#48). + const answered = !options.signal?.aborted || ghAnswered(error); + const outcome = answered && confirmedMergeRefusal(message) ? 'refused' : 'unknown'; throw new MergeSubmissionError(message, outcome, { cause: error }); } finally { this.#generation++; this.#cache = null; } } diff --git a/github/run-with-input.ts b/github/run-with-input.ts index f4f04666..acdbcf7c 100644 --- a/github/run-with-input.ts +++ b/github/run-with-input.ts @@ -9,12 +9,24 @@ export class CommandFailed extends Error { constructor(message: string, stderr: string, stdout: string) { super(message); this.stderr = stderr; this.stdout = stdout; } } +/** + * Whether `error`, or an error it was caused by, is gh's own exit (a CommandFailed): an answer to report as it is, never + * to replace with a later abort's reason (#48). + */ +export function ghAnswered(error: unknown): boolean { + for (let current = error, depth = 0; current instanceof Error && depth < 8; current = current.cause, depth++) + if (current instanceof CommandFailed) return true; + return false; +} + /** * Runs a command with literal argv and writes `input` to its stdin, so large text never travels as an OS argument * (Linux limits one argument to 128 KiB, and a NUL cannot be passed at all). Settles only after the process closed: * a timeout, an abort or an output past `maxBuffer` sends SIGTERM, then SIGKILL after `killGraceMs` if the process is * still running, and the promise rejects once it has exited. If a process it started keeps the output pipes open after - * it exits, the pipes are closed after `pipeGraceMs` so the promise still settles. + * it exits, the pipes are closed after `pipeGraceMs` so the promise still settles. Once its exit has been observed, the + * exit status is the answer: a timeout or abort that fires while the pipes drain does not replace it (#48). Output past + * `maxBuffer` still fails it (tested), and so does a late process error (traced, not tested). */ export function runWithInput(command: string, args: readonly string[], options: { input?: string; signal?: AbortSignal; env?: NodeJS.ProcessEnv; timeout?: number; maxBuffer?: number; killGraceMs?: number; pipeGraceMs?: number; @@ -23,15 +35,15 @@ export function runWithInput(command: string, args: readonly string[], options: if (options.signal?.aborted) { reject(options.signal.reason ?? new Error('Aborted.')); return; } const child = spawn(command, [...args], { env: options.env, stdio: ['pipe', 'pipe', 'pipe'] }); const out: Buffer[] = [], err: Buffer[] = []; - let size = 0, failure: Error | null = null; + let size = 0, failure: Error | null = null, exited = false; let escalation: NodeJS.Timeout | null = null; const stop = (error: Error) => { failure ??= error; child.kill('SIGTERM'); escalation ??= setTimeout(() => child.kill('SIGKILL'), options.killGraceMs ?? 5_000); }; - const timer = options.timeout ? setTimeout(() => stop(new Error(`${command} timed out.`)), options.timeout) : null; - const onAbort = () => stop(options.signal?.reason instanceof Error ? options.signal.reason : new Error('Aborted.')); + const timer = options.timeout ? setTimeout(() => { if (!exited) stop(new Error(`${command} timed out.`)); }, options.timeout) : null; + const onAbort = () => { if (!exited) stop(options.signal?.reason instanceof Error ? options.signal.reason : new Error('Aborted.')); }; options.signal?.addEventListener('abort', onAbort, { once: true }); const collect = (chunks: Buffer[]) => (chunk: Buffer) => { size += chunk.length; @@ -61,6 +73,7 @@ export function runWithInput(command: string, args: readonly string[], options: child.on('close', finish); // 'close' also waits for the pipes; a grandchild that inherited them can hold them open after this process exits. child.on('exit', (code, signal) => { + exited = true; pipes = setTimeout(() => { child.stdout.destroy(); child.stderr.destroy(); finish(code, signal); }, options.pipeGraceMs ?? 1_000); }); }); diff --git a/github/timeout.ts b/github/timeout.ts new file mode 100644 index 00000000..620d25d6 --- /dev/null +++ b/github/timeout.ts @@ -0,0 +1,14 @@ +/** + * A remote read that ran out of its own time. Nothing it started was applied, so a caller may try again: unlike a + * refusal, a timeout is never a definite answer to record (#48). + */ +export class RemoteTimeout extends Error { + override readonly name = 'RemoteTimeout'; +} + +/** Whether `error`, or an error it was caused by, is a RemoteTimeout: wrappers such as an upstream failure keep it. */ +export function timedOut(error: unknown): boolean { + for (let current = error, depth = 0; current instanceof Error && depth < 8; current = current.cause, depth++) + if (current instanceof RemoteTimeout) return true; + return false; +} diff --git a/runner/lifecycle.ts b/runner/lifecycle.ts index cafcd7e6..a45a5962 100644 --- a/runner/lifecycle.ts +++ b/runner/lifecycle.ts @@ -39,7 +39,11 @@ export function assertUuidV4(value: unknown, name: string): asserts value is str /** A guard refused the action. Refusals are definite outcomes and are recorded for replay. */ export class GuardRefusal extends Error {} -/** A definite failure from a read-only external dependency. Saved action replays preserve its HTTP 502 classification. */ +/** + * A failure from a read-only external dependency, answered with HTTP 502 and preserved as 502 by saved action replays. + * One caused by a `RemoteTimeout` is not definite: the server answers it with 503, and the runner-action gate and merge + * admission do not save it; merge preparation and publishing record it in their own outcomes (#48). + */ export class UpstreamFailure extends Error {} /** A malformed request, refused before any transaction and never recorded. The server maps it to HTTP 400. */ export class BadRequest extends GuardRefusal {} diff --git a/runner/merge.ts b/runner/merge.ts index 71f7b6a5..c3d5a1a1 100644 --- a/runner/merge.ts +++ b/runner/merge.ts @@ -3,13 +3,18 @@ import { mergeActionResponse, type MergeAttempt } from './store.ts'; import { ActionIdReused, GuardRefusal, MERGEABLE_STATUSES, ShuttingDownError, assertUuidV4, settleWith, type ShutdownCapability } from './lifecycle.ts'; import { MERGE_INSPECTION_TIMEOUT_MS, MergeSubmissionError, type MergeGateway, type MergeQueueGateway, type MergeQueueObservation, type MergeResult, type MergeTarget, type RemoteMergeState } from '../github/merge.ts'; import { openingMarker } from '../github/pull-requests.ts'; +import { ghAnswered } from '../github/run-with-input.ts'; +import { timedOut } from '../github/timeout.ts'; type QueueGateway = MergeGateway & MergeQueueGateway; export interface MergeBlocker { code: string; message: string; } -/** The longest a merge click may run before it is aborted; its last gh call's stop wait comes on top (github/merge.ts). */ +/** The longest a merge click may run before it is aborted; the stop wait of the step it was in comes on top (a gh call: 0.4 s in github/merge.ts, 0.75 s for the issue-trust read in github/issues.ts; a review load: its history worker's 1 s grace, then termination, in runner/history-job.ts). */ export const MERGE_OPERATION_TIMEOUT_MS = 14_000; const storageError = (error: unknown) => (error as { code?: string } | null)?.code === 'ERR_SQLITE_ERROR'; -/** The merge was not applied for a passing reason (deadline, shutdown); the same click may be sent again. */ +/** The blocker codes `#attemptBlocker` produces: what the merge attempt on record says. The Record type forces every code to be listed. */ +type AttemptBlockerCode = 'queue-active' | 'queue-merged' | 'queue-head'; +const ATTEMPT_BLOCKERS: Readonly> = { 'queue-active': true, 'queue-merged': true, 'queue-head': true }; +/** The merge was not applied for a passing reason (deadline, shutdown, a remote read's own timeout); the same click may be sent again. */ export class MergeNotApplied extends Error {} /** A concurrent poll already committed this attempt's terminal failure; it is the answer, passed through as is. */ class CommittedFailure extends Error {} @@ -239,17 +244,36 @@ export class MergeCoordinator { this.service.store.finishMergeAttempt(this.service.config.identity, attempt.id, { state: 'merged', ...(remote.url ? { url: remote.url } : {}) }); attempt = this.#attempt(); } + return { status: this.#withAttempt(attempt, blockers, remote), resolved }; + } + + /** The attempt on record decides readiness: an active or merged attempt blocks, and an ended one may offer a retry. */ + #withAttempt(attempt: MergeAttempt | null, blockers: MergeBlocker[], remote: RemoteMergeState): MergeStatus { const queue = this.#queueStatus(attempt); const attemptBlocker = this.#attemptBlocker(attempt); - if (attemptBlocker) blockers.unshift(attemptBlocker); + const all = attemptBlocker ? [attemptBlocker, ...blockers] : blockers; const active = attempt?.state === 'submitting' || attempt?.state === 'queued' || attempt?.state === 'merged'; - const retry = !!queue?.retryable; - const ready = !active && blockers.length === 0; - return { status: { available: true, ready, action: ready ? (retry ? 'retry' : 'merge') : null, blockers, remote, queue }, resolved }; + const ready = !active && all.length === 0; + return { available: true, ready, action: ready ? (queue?.retryable ? 'retry' : 'merge') : null, blockers: all, remote, queue }; + } + + /** + * The status to answer a merge click with. The click's status was read before the command, so its readiness is + * recomputed from the attempt on record, which is active or merged after a submission (#48). That GitHub readiness + * is stale once the command ran: if the attempt has already ended (a queue poll during the server's review reload), + * the answer offers no action and asks for a refresh, so a retry is only offered from a fresh GitHub read. + */ + statusAfterMerge(status: MergeStatus | MergeUnavailableStatus): MergeStatus | MergeUnavailableStatus { + // An unavailable status already offers nothing. + if (status.remote === null) return status; + const current = this.#withAttempt(this.#attempt(), status.blockers.filter(blocker => !Object.hasOwn(ATTEMPT_BLOCKERS, blocker.code)), status.remote); + if (!current.ready) return current; + return { ...current, ready: false, action: null, + blockers: [{ code: 'refresh', message: 'Refresh to read GitHub\'s current state before merging again.' }, ...current.blockers] }; } /** What the merge attempt on record says, shown first. */ - #attemptBlocker(attempt: MergeAttempt | null): MergeBlocker | null { + #attemptBlocker(attempt: MergeAttempt | null): (MergeBlocker & { code: AttemptBlockerCode }) | null { if (attempt?.state === 'submitting' || attempt?.state === 'queued') return { code: 'queue-active', message: attempt.state === 'submitting' ? attempt.reason ? `The merge submission outcome is unknown. ${attempt.reason} Waiting for GitHub reconciliation.` : attempt.kind === 'queue' ? 'The reviewed head is being submitted to the merge queue.' : 'The reviewed head is being submitted for direct merge.' : 'The reviewed head is queued. Waiting for GitHub to confirm the outcome.' }; @@ -311,30 +335,34 @@ export class MergeCoordinator { /** * Save a definite refusal as this click's outcome, so a resend replays it instead of being evaluated again. * Returns undefined when this refusal was saved (the caller answers with it). If another coordinator saved the - * click first, its outcome is the answer: a saved success comes back as a replay, a saved refusal or a reused key - * is thrown. + * click first, its outcome is the answer: a saved success comes back as a replay (its status read bounded by + * `signal` when one is given), a saved refusal or a reused key is thrown. A refusal that cannot be saved (a storage + * error, or the write gate closed at shutdown) throws MergeNotApplied, keeping the refusal's text. */ - #recordRefusal(token: string, actionId: string, refusal: unknown): ReturnType | undefined { + #recordRefusal(token: string, actionId: string, refusal: unknown, signal?: AbortSignal): ReturnType | undefined { if (!this.service.store || !this.service.config) return undefined; const action = { actionId, kind: 'merge', request: { token } }; try { // Returning normally means userAction replayed a success saved concurrently: answer from that record. this.service.store.userAction(this.service.config.identity, action, () => { throw refusal; }); - return this.#replay(token, actionId); + return this.#replay(token, actionId, signal); } catch (error) { + // An unsaved refusal keeps its own text in the resend answer (#48; AGENTS.md: do not replace an actionable error). + const reason = refusal instanceof Error ? ` ${refusal.message}` : ''; // A storage error, whether it was the refusal itself or the failed save, recorded nothing: resend. - if (storageError(refusal) || storageError(error)) throw new MergeNotApplied('The merge outcome could not be saved. Try again.', { cause: error }); + if (storageError(refusal) || storageError(error)) throw new MergeNotApplied(`The merge outcome could not be saved. Try again.${reason}`, { cause: error }); // userAction re-raised this refusal after saving it. if (error === refusal) return undefined; // A refusal saved concurrently under this key, or the key reused for another request: that is the answer. if (error instanceof GuardRefusal) throw error; - // Anything else left the refusal unsaved: nothing was applied, so resend. - throw new MergeNotApplied('The merge refusal could not be saved. Try again.', { cause: error }); + // Anything else (such as the write gate closed at shutdown) left the refusal unsaved: nothing was applied, so + // resend. + throw new MergeNotApplied(`The merge refusal could not be saved. Try again.${reason}`, { cause: error }); } } /** The saved outcome of this click, if it has one. Validation never runs again for a replay. */ - #replay(token: string, actionId: string): Promise<{ status: MergeStatus | MergeUnavailableStatus; result: MergeResult }> | undefined { + #replay(token: string, actionId: string, signal?: AbortSignal): Promise<{ status: MergeStatus | MergeUnavailableStatus; result: MergeResult }> | undefined { if (!this.service.store || !this.service.config) return undefined; const { store, config } = this.service; const read = () => { @@ -363,10 +391,19 @@ export class MergeCoordinator { // The status refresh may reconcile the attempt (merged URL, queue failure), so answer from the record re-read after // it. A failed re-read is not answered with the older record: read() turns a storage error into "resend" (503). const current = () => read()?.response ?? saved.response; - const signal = this.#replayAbort.signal; + // A click that is already aborted loads no review and starts or joins no GitHub read: it answers the saved outcome and + // asks for a refresh (#48; test/merge-stop-reasons.test.ts). + if (signal?.aborted) return Promise.resolve().then(() => answer(current(), unavailable(undefined))); + // Every replay's read stops at shutdown, which awaits it. A click's replay also reads fresh under its own signal, so + // its deadline bounds the read and no shared read (a page load's) inherits that signal (#48). Other replays keep the + // shared display read. + const bound = signal ? AbortSignal.any([signal, this.#replayAbort.signal]) : this.#replayAbort.signal; const replayed = (async () => { let status: MergeStatus | MergeUnavailableStatus; - try { status = await this.displayStatus(await this.service.load({ signal }), signal); } catch (error) { status = unavailable(error); } + try { + const view = await this.service.load({ signal: bound }); + status = signal ? await this.status(view, true, bound) : await this.displayStatus(view, bound); + } catch (error) { status = unavailable(error); } return answer(current(), status); })(); const tracked = replayed.then(() => undefined, () => undefined); @@ -503,20 +540,36 @@ export class MergeCoordinator { }); } else this.service.store.recordMergeAttemptDiagnostic(this.service.config.identity, attempt.id, message); }); } catch {} - // A definite refusal before admission is this click's outcome; a resend replays it. Aborts, the deadline and + // Why the work stopped, if it stopped rather than being refused. A remote read that timed out (as itself, or + // wrapped as an upstream failure) reports its timeout: had this coordinator's abort come first, the read would + // have thrown that reason instead (#48). gh's own 30 s timeout (set by the merge gateway's runner; the issue + // gateway's runner sets none) is a plain error, but every read before admission has a shorter RemoteTimeout, so it + // cannot come first. + // gh's own answer (a CommandFailed) that came before the abort is not a stop: it is handled as it would be without + // the abort (AGENTS.md: keep the original reason; do not replace an actionable error with cancellation text). + const readTimedOut = timedOut(error), abortWins = signal.aborted && !readTimedOut && !ghAnswered(error); + const ownAbort = signal.reason instanceof Error ? signal.reason : new Error('Merge request stopped.', { cause: signal.reason }); + const stopped = readTimedOut ? error as Error : abortWins ? ownAbort : null; + // A definite refusal before admission is this click's outcome; a resend replays it. A timeout, the deadline and // shutdown applied nothing for a passing reason, so the same click may be sent again. - if (!queueAttempt && !signal.aborted && action) { - // If another coordinator saved this click meanwhile, its outcome is the answer, not this refusal. - const replay = this.#replay(token, action.actionId) ?? this.#recordRefusal(token, action.actionId, error); + // (A coordinator without a store, which only tests build, saves nothing: a gh answer then passes through as is.) + if (!queueAttempt && action) { + // If another coordinator saved this click meanwhile, its outcome is the answer, not this refusal, timeout or + // abort. For an aborted click the replay reads nothing from GitHub. Only a refusal is recorded; a stop never is. + const replay = this.#replay(token, action.actionId, signal) ?? (stopped ? undefined : this.#recordRefusal(token, action.actionId, error, signal)); if (replay) return await replay; } - // Before admission an abort applied nothing, so the click may be resent (503). After admission the GitHub outcome - // is unknown: the durable attempt stays in flight and reconciles on refresh, so report the original error. - if (signal.aborted && !queueAttempt) throw new MergeNotApplied(signal.reason instanceof Error ? signal.reason.message : 'Merge request stopped.', { cause: signal.reason }); + // Before admission an abort or a remote timeout applied nothing, so the click may be resent (503). After admission + // the attempt stays in flight unless GitHub's refusal was saved, and reconciles on refresh. + if (stopped && !queueAttempt) throw new MergeNotApplied(stopped.message, { cause: stopped }); // After admission only GitHub's confirmed refusal, durably recorded, is definite (the attempt is now failed); // anything else, including a refusal whose failed state could not be saved, leaves the attempt in flight. const refused = error instanceof MergeSubmissionError && error.outcome === 'refused' && refusalSaved; - const failure = signal.aborted && signal.reason instanceof Error ? signal.reason : error; + // After admission, GitHub's refusal is its answer, though this coordinator's deadline or shutdown came later: the + // durable attempt records it (#48), and an unsaved one still carries GitHub's text, as an unknown outcome. Any + // other failure while aborted reports the abort, as an unknown outcome. + const ghRefused = error instanceof MergeSubmissionError && error.outcome === 'refused'; + const failure = signal.aborted && signal.reason instanceof Error && !ghRefused && !ghAnswered(error) ? signal.reason : error; if (queueAttempt && !refused) throw new MergeOutcomeUnknown(failure instanceof Error ? failure.message : 'The merge outcome is unknown.', { cause: failure }); throw failure; } @@ -557,13 +610,17 @@ export class MergeCoordinator { return this.#queueStatus(); } catch (error) { if (error instanceof ShuttingDownError) throw error; - if (signal.aborted) throw signal.reason; + // A queue read that timed out on its own, or that gh answered, keeps that reason, though shutdown came after (#48). + if (signal.aborted && !timedOut(error) && !ghAnswered(error)) throw signal.reason; const message = error instanceof Error ? error.message : 'Could not read the merge queue.'; if (/head changed after review/i.test(message)) { this.service.store.finishMergeAttempt(this.service.config.identity, attempt.id, { state: 'failed', reason: message, requiresFreshReview: true }); return this.#queueStatus(); } - return this.#queueStatus(attempt, message); + // Answer from the attempt on record, which may have moved on while GitHub was read (queued, or a retry). The + // failed read describes only the attempt it inspected (#48). + const current = this.#attempt(); + return this.#queueStatus(current, current?.id === attempt.id ? message : undefined); } } diff --git a/test/gh-env.test.ts b/test/gh-env.test.ts index 5d58b173..d5d2045e 100644 --- a/test/gh-env.test.ts +++ b/test/gh-env.test.ts @@ -64,7 +64,8 @@ const settleBudgetMs: Record = { describe('gh stop waits', () => { it('keep each deadline plus its stop wait below the 15-second serving request budget', () => { - // A merge click admitted just before shutdown must also settle inside the shutdown drain. + // A merge click stopped in one of the merge gateway's gh calls settles inside the shutdown drain. One stopped in its + // issue-trust read or a review load can settle just after it (docs/implementation/merge-queue.md, #48). expect(MERGE_OPERATION_TIMEOUT_MS + MERGE_KILL_GRACE_MS + MERGE_PIPE_GRACE_MS).toBeLessThan(MAX_SHUTDOWN_DRAIN_MS); expect(MAX_SHUTDOWN_DRAIN_MS).toBeLessThan(15_000); // A merge-state or queue inspection, including the polls a browser makes. diff --git a/test/merge-queue-poll.test.ts b/test/merge-queue-poll.test.ts new file mode 100644 index 00000000..02ca6c60 --- /dev/null +++ b/test/merge-queue-poll.test.ts @@ -0,0 +1,410 @@ +import { randomUUID } from 'node:crypto'; +import { chmodSync, mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { expect, it, vi } from 'vitest'; +import { ReviewService } from '../runner/review.ts'; +import { MergeCoordinator, MergeOutcomeUnknown } from '../runner/merge.ts'; +import { MergeSubmissionError, type MergeGateway, type MergeQueueGateway, type MergeQueueObservation, type MergeResult, type RemoteMergeState } from '../github/merge.ts'; +import { Store } from '../runner/store.ts'; +import { createDemo } from '../scripts/demo.ts'; +import { startServer } from '../web/server.ts'; +import { fixtureGit } from './fixtures/git.ts'; +import { RemoteTimeout } from '../github/timeout.ts'; +import { CommandFailed } from '../github/run-with-input.ts'; + +// Issue #48: a queue inspection that started before a newer lifecycle transition must not have its result (or its +// failure) reported as the attempt's current state after that transition. Also: a refusal or timeout keeps its own +// reason across a later shutdown, a merge click's answer is inactive after a submission, and a timed-out trust read +// answers 503. + +type ReviewView = Awaited>; +const sha = (digit: string) => digit.repeat(40); + +function deferred() { + let resolve!: (value: T) => void, reject!: (error: unknown) => void; + const promise = new Promise((res, rej) => { resolve = res; reject = rej; }); + return { promise, resolve, reject }; +} + +function harness() { + const identity = { repositoryId: 'repo', taskId: 'task', planId: 'plan' }; + const store = new Store(':memory:'); + const plan = { schema_version: 1 as const, revision: 1, issue: 24, summary: 'Queue', questions: [], items: [{ id: 'P1', title: 'Queue', intent: 'Queue safely', files: [{ path: 'a', kind: 'edit' as const, renamed_from: null, change: 'Change' }], acceptance: [{ type: 'check' as const, text: 'Works' }], depends_on: [] }] }; + const context = { identity, issue: 24, baseEntries: [{ path: 'a', kind: 'file' as const }], pathKey: (path: string) => path, allowedCommands: [] }; + store.createPlan(JSON.stringify(plan), 'json', context, sha('a'), sha('b')); + const view = { + items: [{ id: 'P1', state: 'approved', outside: [], acceptance: [{ type: 'check', text: 'Works' }], checks: { tests: '– No tests defined' } }], + plan: { revision: 1 }, segments: [], notes: [], snapshot: { id: store.getSnapshot(identity).id, base: sha('a'), head: sha('b') }, token: 'review-token', + expected: { revision: 1, snapshotId: store.getSnapshot(identity).id, reviewVersion: store.reviewVersion(identity) }, + } as unknown as ReviewView; + const service = { store, config: { identity }, load: vi.fn(() => view) } as unknown as ReviewService; + const remote = (): RemoteMergeState => ({ base: sha('a'), head: sha('b'), pullRequestState: 'OPEN', mergeable: 'MERGEABLE', rulesKnown: true, atomicBaseGuard: true, + mergeQueue: true, requiredChecks: [], alreadyFixed: 'clear', pullRequest: 7, draft: false }); + // Each enqueue command and each queue inspection is held until the test settles it. + const commands: Array>> = []; + const inspections: Array>> = []; + const signals: Array = []; + let commandStarted = deferred(), inspectionStarted = deferred(); + const client: MergeGateway & MergeQueueGateway = { + inspect: vi.fn(async () => remote()), + queueWatermark: vi.fn(async () => 'CURSOR_before'), + merge: vi.fn((_head: string, options?: { signal?: AbortSignal }) => { + const command = deferred(); commands.push(command); signals.push(options?.signal); commandStarted.resolve(); return command.promise; + }), + inspectQueue: vi.fn(() => { const inspection = deferred(); inspections.push(inspection); inspectionStarted.resolve(); return inspection.promise; }), + }; + return { + store, identity, view, client, commands, inspections, signals, coordinator: new MergeCoordinator(service, client), + async commandRunning() { await commandStarted.promise; commandStarted = deferred(); }, + async inspectionRunning() { await inspectionStarted.promise; inspectionStarted = deferred(); }, + }; +} + +it('reports the durable queued state when an inspection started during enqueue fails after the enqueue settled', async () => { + const h = harness(); + try { + const clickId = randomUUID(); + const merging = h.coordinator.merge(h.view.token, clickId); + await h.commandRunning(); + expect(h.store.getMergeAttempt(h.identity)).toMatchObject({ state: 'submitting', actionId: clickId }); + + // A poll (another tab, or the browser after a reload) starts its inspection while the enqueue is in flight. + const earlyPoll = h.coordinator.pollQueue(); + await h.inspectionRunning(); + + // The enqueue command completes: the attempt is durably queued. + h.commands[0]!.resolve({ url: 'https://github.example/pr/7' }); + await merging; + expect(h.store.getMergeAttempt(h.identity)).toMatchObject({ state: 'queued', url: 'https://github.example/pr/7' }); + + // A poll that arrives after the transition joins the inspection that started before it. + const latePoll = h.coordinator.pollQueue(); + h.inspections[0]!.reject(new Error('GitHub queue status unavailable.')); + + const durable = h.store.getMergeAttempt(h.identity)!; + expect(durable.state).toBe('queued'); + // Neither answer may regress the attempt to the pre-transition "submitting" snapshot the inspection captured. + expect(await latePoll).toMatchObject({ state: durable.state, url: durable.url, observationError: 'GitHub queue status unavailable.' }); + expect(await earlyPoll).toMatchObject({ state: durable.state, url: durable.url }); + } finally { await h.coordinator.close(); h.store.close(); } +}); + +it('reports the retry attempt, not the superseded attempt, when an inspection started before the retry fails', async () => { + const h = harness(); + try { + const firstClick = randomUUID(), retryClick = randomUUID(); + const first = h.coordinator.merge(h.view.token, firstClick); + await h.commandRunning(); + + // An inspection for the first attempt starts while its enqueue command is in flight. + const earlyPoll = h.coordinator.pollQueue(); + await h.inspectionRunning(); + + // GitHub definitely refuses the first enqueue: the attempt fails and a retry becomes available. + h.commands[0]!.reject(new MergeSubmissionError('Required review is missing.', 'refused', + { cause: new CommandFailed('gh failed (exit 1): Required review is missing.', 'Required review is missing.', '') })); + await expect(first).rejects.toThrow('Required review is missing.'); + expect(h.store.getMergeAttempt(h.identity)).toMatchObject({ state: 'failed', actionId: firstClick }); + expect((await h.coordinator.status(h.view)).action).toBe('retry'); + + // The retry starts a new attempt and GitHub accepts it. + const retry = h.coordinator.merge(h.view.token, retryClick); + await h.commandRunning(); + h.commands[1]!.resolve({ url: 'https://github.example/pr/7' }); + await retry; + const current = h.store.getMergeAttempt(h.identity)!; + expect(current).toMatchObject({ state: 'queued', actionId: retryClick }); + + // A poll after the retry joins the first attempt's still-running inspection, which then fails. + const latePoll = h.coordinator.pollQueue(); + h.inspections[0]!.reject(new Error('GitHub queue status unavailable.')); + + // The answer must describe the current attempt, not resurrect the superseded one as "submitting". + // The failed read was of the superseded attempt, so its error is not attached to the retry. + for (const answer of [await latePoll, await earlyPoll]) { + expect(answer).toMatchObject({ state: 'queued', actionId: retryClick }); + expect(answer).not.toHaveProperty('observationError'); + } + } finally { await h.coordinator.close(); h.store.close(); } +}); + +it('control: a successful inspection started before the retry does not write to or report the superseded attempt', async () => { + const h = harness(); + try { + const firstClick = randomUUID(), retryClick = randomUUID(); + const first = h.coordinator.merge(h.view.token, firstClick); + await h.commandRunning(); + const earlyPoll = h.coordinator.pollQueue(); + await h.inspectionRunning(); + h.commands[0]!.reject(new MergeSubmissionError('Required review is missing.', 'refused', + { cause: new CommandFailed('gh failed (exit 1): Required review is missing.', 'Required review is missing.', '') })); + await expect(first).rejects.toThrow(); + const retry = h.coordinator.merge(h.view.token, retryClick); + await h.commandRunning(); + h.commands[1]!.resolve({ url: 'https://github.example/pr/7' }); + await retry; + // The old inspection reports a removal; the store's id/state compare-and-set must refuse it for the new attempt. + h.inspections[0]!.resolve({ state: 'removed', reviewedHead: sha('b'), removedAt: '2026-09-24T08:05:00Z', reason: 'Old removal.' }); + expect(await earlyPoll).toMatchObject({ state: 'queued', actionId: retryClick, reason: null }); + expect(h.store.getMergeAttempt(h.identity)).toMatchObject({ state: 'queued', actionId: retryClick, reason: null }); + } finally { await h.coordinator.close(); h.store.close(); } +}); + +it('answers a refusal that gh printed before a later shutdown with GitHub\'s reason, as the attempt records', async () => { + const h = harness(); + try { + const merging = h.coordinator.merge(h.view.token, randomUUID()).then(() => null, (error: unknown) => error as Error); + await h.commandRunning(); + const closing = h.coordinator.close(); // Shutdown aborts after gh has already answered. + h.commands[0]!.reject(new MergeSubmissionError('GraphQL: Head branch was modified. Review and try the merge again.', 'refused', + { cause: new CommandFailed('gh failed (exit 1): GraphQL: Head branch was modified. Review and try the merge again.', '', '') })); + const error = await merging; + await closing; + expect(error?.message).toMatch(/Head branch was modified/); + expect(h.store.getMergeAttempt(h.identity)).toMatchObject({ state: 'failed', reason: expect.stringMatching(/Head branch was modified/) }); + } finally { h.store.close(); } +}); + +it('answers a gh failure after admission that came before a later shutdown with gh\'s text, as an unknown outcome', async () => { + const h = harness(); + try { + const merging = h.coordinator.merge(h.view.token, randomUUID()).then(() => null, (error: unknown) => error as Error); + await h.commandRunning(); + const closing = h.coordinator.close(); + // gh exited with a failure that is not a refusal; its answer is delivered after the shutdown abort. + h.commands[0]!.reject(new MergeSubmissionError('gh failed (exit 1): HTTP 502: Bad Gateway', 'unknown', + { cause: new CommandFailed('gh failed (exit 1): HTTP 502: Bad Gateway', '', '') })); + const error = await merging; + await closing; + expect(error).toBeInstanceOf(MergeOutcomeUnknown); + expect(error?.message).toBe('gh failed (exit 1): HTTP 502: Bad Gateway'); + expect(h.store.getMergeAttempt(h.identity)).toMatchObject({ state: 'submitting' }); + } finally { h.store.close(); } +}); + +// Control: the old code also reported the abort here. It fails if GitHub's refusal exception is widened to every error. +it('reports the shutdown that stopped an admitted gh command, as an unknown outcome', async () => { + const h = harness(); + try { + const merging = h.coordinator.merge(h.view.token, randomUUID()).then(() => null, (error: unknown) => error as Error); + await h.commandRunning(); + const closing = h.coordinator.close(); + // gh was stopped by the coordinator's own abort and failed without a refusal: the abort is reported, not gh's text. + // The message is synthetic on purpose: the real gateway would carry the abort's own text, which could not tell the two apart. + h.commands[0]!.reject(new MergeSubmissionError('gh was stopped.', 'unknown', { cause: h.signals[0]!.reason })); + const error = await merging; + await closing; + expect(error).toBeInstanceOf(MergeOutcomeUnknown); + expect(error?.message).toBe('Merge cancelled during shutdown.'); + expect(h.store.getMergeAttempt(h.identity)).toMatchObject({ state: 'submitting' }); + } finally { h.store.close(); } +}); + +it('answers an unsaved refusal that gh printed before a later shutdown with GitHub\'s text, as an unknown outcome', async () => { + const h = harness(); + try { + const merging = h.coordinator.merge(h.view.token, randomUUID()).then(() => null, (error: unknown) => error as Error); + await h.commandRunning(); + vi.spyOn(h.store, 'finishMergeAttempt').mockImplementation(() => { throw new Error('disk full'); }); + const closing = h.coordinator.close(); + h.commands[0]!.reject(new MergeSubmissionError('GraphQL: Head branch was modified. Review and try the merge again.', 'refused', + { cause: new CommandFailed('gh failed (exit 1): GraphQL: Head branch was modified. Review and try the merge again.', '', '') })); + const error = await merging; + await closing; + // Not saved, so not definite: the attempt stays in flight, but the answer carries what GitHub said. + expect(error).toBeInstanceOf(MergeOutcomeUnknown); + expect(error?.message).toMatch(/Head branch was modified/); + } finally { vi.restoreAllMocks(); h.store.close(); } +}); + +it('passes an unavailable status through a merge click\'s answer unchanged', () => { + const h = harness(); + try { + const unavailable = { available: true as const, ready: false as const, action: null, remote: null, queue: null, + blockers: [{ code: 'refresh', message: 'Merge was submitted. Refresh to confirm GitHub state.' }] }; + expect(h.coordinator.statusAfterMerge(unavailable)).toBe(unavailable); + } finally { h.store.close(); } +}); + +it('a queue read that gh answered keeps that answer when shutdown aborts it while the answer is delivered', async () => { + const h = harness(); + try { + const merging = h.coordinator.merge(h.view.token, randomUUID()); + await h.commandRunning(); + h.commands[0]!.resolve({ url: 'https://github.example/pr/7' }); + await merging; + const poll = h.coordinator.pollQueue().then(value => ({ value }), (error: unknown) => ({ error })); + await h.inspectionRunning(); + const closing = h.coordinator.close(); + h.inspections[0]!.reject(new CommandFailed('gh failed (exit 1): HTTP 502: Bad Gateway', '', '')); + const outcome = await poll; + await closing; + expect(outcome).toMatchObject({ value: { state: 'queued', observationError: 'gh failed (exit 1): HTTP 502: Bad Gateway' } }); + } finally { await h.coordinator.close(); h.store.close(); } +}); + +it('a queue read that timed out keeps its own reason when shutdown aborts it while gh stops', async () => { + const h = harness(); + try { + const merging = h.coordinator.merge(h.view.token, randomUUID()); + await h.commandRunning(); + h.commands[0]!.resolve({ url: 'https://github.example/pr/7' }); + await merging; + const poll = h.coordinator.pollQueue().then(value => ({ value }), (error: unknown) => ({ error })); + await h.inspectionRunning(); + // Shutdown aborts the poll; the inspection had already timed out and settles with its own reason afterwards. + const closing = h.coordinator.close(); + h.inspections[0]!.reject(new RemoteTimeout('GitHub merge-queue inspection timed out.')); + const outcome = await poll; + await closing; + expect(outcome).toMatchObject({ value: { state: 'queued', observationError: 'GitHub merge-queue inspection timed out.' } }); + } finally { await h.coordinator.close(); h.store.close(); } +}); + +/** A server over the demo task with every item approved, and one merge click ready to send (and resend). */ +async function serveApproved(gateway: (app: () => Awaited>) => MergeGateway, options: { github?: boolean; issues?: unknown } = {}) { + const root = mkdtempSync(join(tmpdir(), 'codeboost-merge-')); + const config = { ...createDemo(join(root, 'demo')), demo: false, ...options.github ? { github: { repository: 'owner/repo', issue: 3, pullRequest: 7 } } : {} }; + chmodSync(join(config.repository, 'run.sh'), 0o644); fixtureGit(config.repository, 'commit', '-am', 'Restore declared scope'); + let app!: Awaited>; + app = await startServer(config, 0, undefined, gateway(() => app), undefined, options.issues as never); + let view = await app.service.load(); + for (const segment of view.segments.filter(value => value.row === 'Unplanned' || value.row === 'Ambiguous')) view = await app.service.act({ action: 'accept', key: segment.key, token: view.token }); + for (const item of view.items) view = await app.service.act({ action: 'approve', item: item.id, confirmNoChange: item.count === 0, token: view.token }); + const click = JSON.stringify({ action: 'merge', token: view.token, actionId: randomUUID() }); + const send = () => fetch(new URL('/api/action', app.url), { method: 'POST', headers: { 'x-codeboost-token': app.token, 'Content-Type': 'application/json' }, body: click }); + return { app, send, remove: () => rmSync(root, { recursive: true, force: true }) }; +} +// The recorded snapshot, read from the Store: a review load here would run a history worker inside every inspection. +const remoteFor = (app: Awaited>, mergeQueue: boolean): RemoteMergeState => { + const snapshot = app.service.store.getSnapshot(app.service.config.identity); + return { base: snapshot.base, head: snapshot.head, pullRequestState: 'OPEN', mergeable: 'MERGEABLE', rulesKnown: true, atomicBaseGuard: true, mergeQueue, + requiredChecks: [], alreadyFixed: 'clear', pullRequest: 7, draft: false }; +}; + +// Copilot (126b07f): a queue poll can end the attempt while the server reloads the review after a successful enqueue. +// The click's GitHub readiness was read before the command, so its answer must not offer a retry from it. +it('answers a successful enqueue inactively when a poll ends the attempt before the answer is built', async () => { + const h = harness(); + try { + const before = await h.coordinator.status(h.view); // The pre-command readiness the server holds for the click. + expect(before).toMatchObject({ ready: true, action: 'merge' }); + const merging = h.coordinator.merge(h.view.token, randomUUID()); + await h.commandRunning(); + h.commands[0]!.resolve({ url: 'https://github.example/pr/7' }); + await merging; + // During the server's post-command review load, a poll sees GitHub remove the entry. + const polling = h.coordinator.pollQueue(); + await h.inspectionRunning(); + h.inspections[0]!.resolve({ state: 'removed', reviewedHead: sha('b'), removedAt: '2026-10-14T08:05:00Z', reason: 'A required check failed.' }); + await polling; + expect(h.store.getMergeAttempt(h.identity)?.state).not.toMatch(/^(submitting|queued|merged)$/); + const answer = h.coordinator.statusAfterMerge(before); + expect(answer).toMatchObject({ ready: false, action: null }); + expect(answer.blockers.map(blocker => blocker.code)).toContain('refresh'); + } finally { await h.coordinator.close(); h.store.close(); } +}); + +// Copilot's fifth concern ("Return inactive lifecycle status after successful enqueue", web/server.ts; fix 6 in the docs): +// a successful merge response reports the pre-command readiness next to the now-active attempt. +it.each([['queue enqueue', true, 'queued'], ['direct merge', false, 'merged']] as const)('answers a successful %s with an inactive lifecycle status', async (_kind, mergeQueue, state) => { + const { app, send, remove } = await serveApproved(app => ({ + inspect: async () => remoteFor(app(), mergeQueue), + queueWatermark: async () => null, + merge: async () => ({ url: 'https://github.com/example/repo/pull/24' }), + inspectQueue: async (head: string) => ({ state: 'queued', reviewedHead: head, entryId: 'MQE_1', phase: 'QUEUED', position: 1, enqueuedAt: '2026-09-24T08:00:00Z', queueHead: head }), + }) as MergeGateway & MergeQueueGateway); + try { + type Body = { merge: { ready: boolean; action: string | null; blockers: Array<{ code: string }>; queue: { state: string } | null }; mergeQueue: { state: string } | null }; + const attemptBlockers = (body: Body) => body.merge.blockers.filter(blocker => ['queue-active', 'queue-merged', 'queue-head'].includes(blocker.code)); + const response = await send(); + const body = await response.json() as Body; + expect(response.status).toBe(200); + expect(body.mergeQueue?.state).toBe(state); + expect(body.merge.queue?.state).toBe(state); + // The attempt is active or merged: the same response must not also offer another merge. + expect({ ready: body.merge.ready, action: body.merge.action }).toEqual({ ready: false, action: null }); + expect(attemptBlockers(body)).toHaveLength(1); + // Control: a replay already answered from the current status before #48. It stays inactive, and the attempt's + // blocker appears once (the filter in statusAfterMerge keeps it from appearing twice). + const replay = await send(); + const again = await replay.json() as Body; + expect(replay.status).toBe(200); + expect({ ready: again.merge.ready, action: again.merge.action }).toEqual({ ready: false, action: null }); + expect(attemptBlockers(again)).toHaveLength(1); + } finally { await app.close(); remove(); } +}, 60_000); + +// The click's 14 s deadline aborts the trust refresh just before admission (here the first trust read answers at once; in +// production it settles within 12.75 s of its start); gh had already answered, and the read surfaces that answer after the abort. +it('answers a merge click whose issue-trust read gh answered before the merge deadline with that answer, saved', async () => { + let readStarted!: () => void, reads = 0, slowNext = false; + const reading = new Promise(resolve => { readStarted = resolve; }); + const issues = { + repository: 'owner/repo', + async fetch() { return { repository: 'owner/repo', retrievedAt: new Date().toISOString(), issues: [] }; }, + // gh answered; the answer surfaces only after the caller's abort (the merge deadline). + async issueAccess(number: number, options: { signal?: AbortSignal } = {}) { + // The first read admits the click; the refresh before admission is held until the deadline aborts it. + if (++reads === 1) { slowNext = true; return { number, authorLogin: 'member', collaborator: true }; } + readStarted(); await new Promise(resolve => options.signal?.addEventListener('abort', resolve, { once: true })); + throw new CommandFailed('gh failed (exit 1): HTTP 404: Not Found', '', ''); + }, + async issueText(number: number) { return { number, title: '', body: '', comments: [] }; }, + }; + const { app, send, remove } = await serveApproved(app => ({ + // The first inspection after the click's first trust read takes 2.5 s (plus the click's review loads), so the refresh + // starts late enough that its own 12 s would end after the 14 s deadline: the deadline is what stops it, after gh had + // answered. + inspect: async () => { if (slowNext) { slowNext = false; await new Promise(resolve => setTimeout(resolve, 2_500)); } return remoteFor(app(), false); }, + merge: async () => { throw new Error('The merge command must not run.'); }, + }), { github: true, issues }); + try { + const answer = async (response: Response) => ({ status: response.status, error: (await response.json() as { error: string }).error }); + const expected = { status: 502, error: 'Issue trust could not be verified: gh failed (exit 1): HTTP 404: Not Found' }; + expect(await answer(await send())).toEqual(expected); + await reading; + // Saved: resending the same click replays the answer without reading trust again. + expect(await answer(await send())).toEqual(expected); + expect(reads).toBe(2); + } finally { await app.close(); remove(); } +}, 60_000); + +// The merge path reads the issue's trust before admission. A read that runs out of time is wrapped as an upstream +// failure; it must still answer "resend" (503) and leave nothing saved under the click's key. +it.each(['alone', 'before the merge deadline'] as const)('answers a merge click whose issue-trust read timed out (%s) with a resendable 503', async when => { + let readStarted!: () => void, reads = 0, slowNext = false; + const reading = new Promise(resolve => { readStarted = resolve; }); + const issues = { + repository: 'owner/repo', + async fetch() { return { repository: 'owner/repo', retrievedAt: new Date().toISOString(), issues: [] }; }, + // Alone: the first read times out. Deadline case: the first read admits the click, and the refresh before admission + // timed out on its own; that surfaces only after the click's 14 s deadline abort. + async issueAccess(number: number, options: { signal?: AbortSignal } = {}) { + if (when === 'before the merge deadline') { + if (++reads % 2 === 1) { slowNext = true; return { number, authorLogin: 'member', collaborator: true }; } + readStarted(); await new Promise(resolve => options.signal?.addEventListener('abort', resolve, { once: true })); + } + throw new RemoteTimeout('Issue retrieval timed out.'); + }, + async issueText(number: number) { return { number, title: '', body: '', comments: [] }; }, + }; + const { app, send, remove } = await serveApproved(app => ({ + // In the deadline case the first inspection after each click's first trust read takes 1.6 s, so in production the + // refresh's own 12 s would end at about 13.6 s plus the click's review loads, near the 14 s deadline, with gh still + // stopping when the deadline fires. This mock ignores the read's own timeout and answers RemoteTimeout once aborted, + // so the answer does not depend on which of the two fires first. + inspect: async () => { + if (slowNext) { slowNext = false; await new Promise(resolve => setTimeout(resolve, 1_600)); } + return remoteFor(app(), false); + }, + merge: async () => { throw new Error('The merge command must not run.'); }, + }), { github: true, issues }); + try { + const expected = { status: 503, error: 'Issue trust could not be verified: Issue retrieval timed out.' }; + const answer = async (response: Response) => ({ status: response.status, error: (await response.json() as { error: string }).error }); + for (const response of [await send(), await send()]) expect(await answer(response)).toEqual(expected); + if (when !== 'alone') await reading; + } finally { await app.close(); remove(); } +}, 60_000); diff --git a/test/merge-status-cache.test.ts b/test/merge-status-cache.test.ts new file mode 100644 index 00000000..d53f388b --- /dev/null +++ b/test/merge-status-cache.test.ts @@ -0,0 +1,121 @@ +import { expect, it } from 'vitest'; +import { GhMergeGateway, MergeSubmissionError, type RemoteMergeState } from '../github/merge.ts'; +import type { AlreadyFixedGateway } from '../github/already-fixed.ts'; +import { CommandFailed } from '../github/run-with-input.ts'; + +// #48: "invalidate status caches after successful and refused irreversible mutations, and reject stale in-flight +// cache publication". GhMergeGateway keeps the only server-side merge status cache (5 s, keyed by target) and an +// in-flight inspection shared by non-fresh readers. Its only mutation is merge() (direct merge or enqueue). +const sha = (c: string) => c.repeat(40); +type Deferred = { promise: Promise; resolve: (value: T) => void; reject: (error: unknown) => void }; +function deferred(): Deferred { + let resolve!: (value: T) => void, reject!: (error: unknown) => void; + const promise = new Promise((res, rej) => { resolve = res; reject = rej; }); + return { promise, resolve, reject }; +} + +const checks: AlreadyFixedGateway = { repository: 'owner/repo', check: async () => ({ outcome: 'clear', baseHead: sha('9') }) }; + +/** + * A fake gh. `pr view` answers with the PR state GitHub had when the call started, released by the test's gate, so + * a read that began before a mutation returns pre-mutation state. `pr merge` waits on its own gate. + */ +function fakeGh() { + let state: 'OPEN' | 'MERGED' = 'OPEN', mergeable: 'MERGEABLE' | 'CONFLICTING' = 'MERGEABLE'; + const views: Array> = []; + let mergeGate: Deferred | null = null; + const run = async (args: readonly string[], _options?: { signal?: AbortSignal }) => { + const joined = args.join(' '); + if (joined.startsWith('pr view')) { + const observed = state, observedMergeable = mergeable, gate = deferred(); + views.push(gate); + await gate.promise; + return JSON.stringify({ number: 7, isDraft: false, baseRefName: 'main', baseRefOid: sha('a'), headRefName: 'feature', headRefOid: sha('b'), + state: observed, mergeable: observedMergeable, statusCheckRollup: [] }); + } + if (joined.startsWith('pr merge')) { + if (mergeGate) await mergeGate.promise; + return ''; + } + if (joined.includes('/rules/branches/')) return JSON.stringify([[]]); + if (/branches\/main$/.test(joined)) return JSON.stringify({ protected: false }); + if (joined.endsWith('/protection')) throw new Error('HTTP 404: Not Found'); + throw new Error(`Unexpected gh call: ${joined}`); + }; + return { + run, views, + setState: (next: 'OPEN' | 'MERGED') => { state = next; }, + setMergeable: (next: 'MERGEABLE' | 'CONFLICTING') => { mergeable = next; }, + gateMerge: () => (mergeGate = deferred()), + }; +} +const flush = () => new Promise(resolve => setTimeout(resolve, 0)); +const client = (gh: ReturnType) => new GhMergeGateway({ repository: 'owner/repo', pullRequest: 7, issue: 21 }, gh.run, checks); + +for (const outcome of ['merged', 'refused'] as const) { + const gatewayFor = (gh: ReturnType) => outcome === 'merged' ? client(gh) + : new GhMergeGateway({ repository: 'owner/repo', pullRequest: 7, issue: 21 }, async (args, options) => { + if (args[0] === 'pr' && args[1] === 'merge') { gh.setMergeable('CONFLICTING'); throw new CommandFailed('gh failed (exit 1): GraphQL: Pull Request is not mergeable', 'GraphQL: Pull Request is not mergeable', ''); } + return gh.run(args, options); + }, checks); + const mutate = async (gh: ReturnType, gateway: GhMergeGateway) => { + if (outcome === 'merged') { await gateway.merge(sha('b')); gh.setState('MERGED'); return; } + const refusal = await gateway.merge(sha('b')).catch(error => error); + expect(refusal).toBeInstanceOf(MergeSubmissionError); + expect((refusal as MergeSubmissionError).outcome).toBe('refused'); + }; + + // What changed on GitHub with the mutation: the PR merged, or (refused) GitHub now reports the conflict that refused it. + const observed = (state: RemoteMergeState) => outcome === 'merged' ? state.pullRequestState : state.mergeable; + const expected = outcome === 'merged' ? 'MERGED' : 'CONFLICTING'; + + it(`does not serve the status cached before a ${outcome} merge after it`, async () => { + const gh = fakeGh(), gateway = gatewayFor(gh); + const cached = gateway.inspect(); await flush(); gh.views[0]!.resolve(); + expect((await cached).pullRequestState).toBe('OPEN'); + // Still within the 5 s lifetime: served from the cache before the mutation. + expect((await gateway.inspect()).pullRequestState).toBe('OPEN'); + expect(gh.views).toHaveLength(1); + await mutate(gh, gateway); + const after = gateway.inspect(); await flush(); + expect(gh.views).toHaveLength(2); + gh.views[1]!.resolve(); + expect(observed(await after)).toBe(expected); + }); + + it(`neither joins nor publishes a read in flight across a ${outcome} merge`, async () => { + const gh = fakeGh(), gateway = gatewayFor(gh); + const stale = gateway.inspect(); await flush(); + expect(gh.views).toHaveLength(1); + await mutate(gh, gateway); + // A reader after the mutation does not join the pre-mutation read. + const after = gateway.inspect(); await flush(); + expect(gh.views).toHaveLength(2); + // The pre-mutation read settles last; it must not overwrite the newer answer in the cache. + gh.views[1]!.resolve(); await after; + gh.views[0]!.resolve(); + expect((await stale).pullRequestState).toBe('OPEN'); + expect(observed(await gateway.inspect())).toBe(expected); + expect(gh.views).toHaveLength(2); + }); +} + +it('a read that starts while the merge command runs and ends after it is neither joined nor published', async () => { + const gh = fakeGh(), gateway = client(gh); + const merge = gh.gateMerge(); + const merging = gateway.merge(sha('b')); await flush(); + // Started during the merge: GitHub still shows the PR open. + const during = gateway.inspect(); await flush(); + expect(gh.views).toHaveLength(1); + merge.resolve(); await merging; + gh.setState('MERGED'); + // A reader arriving after the merge must not join the read that started during it. + const joined = gateway.inspect(); await flush(); + expect(gh.views).toHaveLength(2); + // The read that started during the merge settles last; it must not overwrite the newer answer in the cache. + gh.views[1]!.resolve(); expect((await joined).pullRequestState).toBe('MERGED'); + gh.views[0]!.resolve(); expect((await during).pullRequestState).toBe('OPEN'); + // And the stale read did not land in the cache afterwards. + expect((await gateway.inspect()).pullRequestState).toBe('MERGED'); + expect(gh.views).toHaveLength(2); +}); diff --git a/test/merge-stop-reasons.test.ts b/test/merge-stop-reasons.test.ts new file mode 100644 index 00000000..f19350b1 --- /dev/null +++ b/test/merge-stop-reasons.test.ts @@ -0,0 +1,485 @@ +// Issue #48: the original timeout/cancellation reason must survive gateway, coordinator, durable attempt and response. +// Interleavings are ordered by deferred gates or observed process state. The two real-process tests also need the +// command to start and exit well inside their timers (1.5 s and a 3 s pipe grace); under heavy load they can fail, not pass. +// The parallelStages() tests use real 1 s inspection deadlines; the deadline test checks the order it relies on and fails, +// not passes, if heavy load delays its stage failure past the deadline. +import { randomUUID } from 'node:crypto'; +import { mkdtempSync, readFileSync, existsSync, writeFileSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterEach, expect, it, vi } from 'vitest'; +import { ReviewService } from '../runner/review.ts'; +import { MergeCoordinator, MergeNotApplied } from '../runner/merge.ts'; +import { GhMergeGateway, MergeSubmissionError } from '../github/merge.ts'; +import { CommandFailed, runWithInput } from '../github/run-with-input.ts'; +import { GhIssueGateway, ISSUE_READ_ACTIVE_MS, withIssueReadDeadline } from '../github/issues.ts'; +import { RemoteTimeout } from '../github/timeout.ts'; +import type { AlreadyFixedGateway } from '../github/already-fixed.ts'; +import { Store } from '../runner/store.ts'; + +type ReviewView = Awaited>; +type Run = ConstructorParameters[1]; +const sha = (digit: string) => digit.repeat(40); +const config = { repository: 'owner/repo', pullRequest: 7, issue: 24 }; +function deferred() { + let resolve!: (value: T) => void; + const promise = new Promise(done => { resolve = done; }); + return { promise, resolve }; +} +/** A gh call that runs until its signal aborts, then settles (with that signal's reason) only when `release` opens. */ +function stoppingRun() { + const started = deferred(), stopped = deferred(), release = deferred(); + const run: Run = (_args, options) => new Promise((_, reject) => { + started.resolve(); + options?.signal?.addEventListener('abort', () => { stopped.resolve(); void release.promise.then(() => reject(options.signal!.reason)); }, { once: true }); + }); + return { run, started: started.promise, stopped: stopped.promise, release: () => release.resolve() }; +} +function readyView(extra: Record = {}): ReviewView { + return { + items: [{ id: 'P1', state: 'approved', outside: [], acceptance: [{ type: 'check', text: 'Works' }], checks: { tests: '– No tests defined' } }], + plan: { revision: 1 }, segments: [], notes: [], snapshot: { id: 'snapshot', base: sha('a'), head: sha('b') }, token: 'review-token', ...extra, + } as unknown as ReviewView; +} + +afterEach(() => { vi.useRealTimers(); }); + +it('gateway queue watermark keeps its own earlier timeout when the caller aborts during the gh stop wait', async () => { + const gh = stoppingRun(), caller = new AbortController(); + const gateway = new GhMergeGateway(config, gh.run); + const watermark = gateway.queueWatermark(sha('b'), { signal: caller.signal, timeoutMs: 1 }).catch((error: unknown) => error); + await gh.stopped; // The watermark's own 1 ms deadline fired first and is stopping gh. + caller.abort(new Error('Merge cancelled during shutdown.')); + gh.release(); + expect(((await watermark) as Error).message).toBe('GitHub merge-queue watermark timed out.'); +}); + +it('gateway queue inspection keeps its own earlier timeout when the caller aborts during the gh stop wait', async () => { + const gh = stoppingRun(), caller = new AbortController(); + const gateway = new GhMergeGateway(config, gh.run); + const inspection = gateway.inspectQueue(sha('b'), { signal: caller.signal, timeoutMs: 1, afterCursor: null }).catch((error: unknown) => error); + await gh.stopped; + caller.abort(new Error('Merge-queue inspection cancelled during shutdown.')); + gh.release(); + expect(((await inspection) as Error).message).toBe('GitHub merge-queue inspection timed out.'); +}); + +it('an issue read keeps its own earlier timeout when the caller aborts during the gh stop wait', async () => { + const gh = stoppingRun(), caller = new AbortController(); + const reading = new GhIssueGateway('owner/repo', gh.run).issueAccess(24, { signal: caller.signal, timeoutMs: 1 }).catch((error: unknown) => error); + await gh.stopped; + caller.abort(new Error('Merge cancelled during shutdown.')); + gh.release(); + const error = await reading; + expect(error).toBeInstanceOf(RemoteTimeout); + expect((error as Error).message).toBe('Issue retrieval timed out.'); +}); + +it('gateway merge never reports its own abort as GitHub\'s refusal, even when the abort reason reads like one', async () => { + const gh = stoppingRun(), caller = new AbortController(); + const merging = new GhMergeGateway(config, gh.run).merge(sha('b'), { signal: caller.signal }).catch((error: unknown) => error); + await gh.started; + caller.abort(new Error('GraphQL: Head branch was modified. Review and try the merge again.')); + gh.release(); + const error = await merging as MergeSubmissionError; + expect(error).toBeInstanceOf(MergeSubmissionError); + expect(error.outcome).toBe('unknown'); +}); + +/** A gh call whose answer (gh's own failure) is delivered only after its signal aborted: it answered before the abort. */ +function lateAnswerRun(answer: Error) { + const started = deferred(), stopped = deferred(), release = deferred(); + const run: Run = (_args, options) => new Promise((_, reject) => { + started.resolve(); + options?.signal?.addEventListener('abort', () => { stopped.resolve(); void release.promise.then(() => reject(answer)); }, { once: true }); + }); + return { run, started: started.promise, stopped: stopped.promise, release: () => release.resolve() }; +} +it.each([ + ['queue watermark', (run: Run, signal: AbortSignal) => new GhMergeGateway(config, run).queueWatermark(sha('b'), { signal, timeoutMs: 1 })], + ['queue inspection', (run: Run, signal: AbortSignal) => new GhMergeGateway(config, run).inspectQueue(sha('b'), { signal, timeoutMs: 1, afterCursor: null })], + ['merge-state inspection', (run: Run, signal: AbortSignal) => new GhMergeGateway(config, run).inspect({ fresh: true, signal, timeoutMs: 1 })], + ['issue read', (run: Run, signal: AbortSignal) => new GhIssueGateway('owner/repo', run).issueAccess(24, { signal, timeoutMs: 1 })], +] as const)('%s reports gh\'s own answer when its deadline fires after gh answered', async (_read, read) => { + // gh exited with this answer before the deadline fired; runWithInput reports it as a CommandFailed. + const answer = new CommandFailed('gh failed (exit 1): GraphQL: Could not resolve to a PullRequest with the number of 7.', '', ''); + const gh = lateAnswerRun(answer); + const reading = read(gh.run, new AbortController().signal).catch((error: unknown) => error); + await gh.stopped; // The read's own 1 ms deadline fired; gh's answer is already on its way. + gh.release(); + expect(await reading).toBe(answer); +}); + +it('coordinator keeps gh\'s answer before admission when shutdown comes after it and writes are already closed', async () => { + const identity = { repositoryId: 'repo', taskId: 'task', planId: 'plan' }; + const store = new Store(':memory:'); + const plan = { schema_version: 1 as const, revision: 1, issue: 24, summary: 'Merge', questions: [], items: [{ id: 'P1', title: 'Merge', intent: 'Merge safely', files: [{ path: 'a', kind: 'edit' as const, renamed_from: null, change: 'Change' }], acceptance: [{ type: 'check' as const, text: 'Works' }], depends_on: [] }] }; + store.createPlan(JSON.stringify(plan), 'json', { identity, issue: 24, baseEntries: [{ path: 'a', kind: 'file' as const }], pathKey: (path: string) => path, allowedCommands: [] }, sha('a'), sha('b')); + const view = readyView({ expected: { revision: 1, snapshotId: store.getSnapshot(identity).id, reviewVersion: store.reviewVersion(identity) } }); + // gh exited with this answer; its delivery is held until after the coordinator's own shutdown abort. + const answer = new CommandFailed('gh failed (exit 1): GraphQL: Could not resolve to a PullRequest with the number of 7.', '', ''); + const gh = lateAnswerRun(answer); + const coordinator = new MergeCoordinator({ store, config: { identity }, load: vi.fn(() => view) } as unknown as ReviewService, new GhMergeGateway(config, gh.run)); + const actionId = randomUUID(); + try { + const merging = coordinator.merge(view.token, actionId).catch((error: unknown) => error); + await gh.started; // Shutdown finds the click's gh call running. + // The server's shutdown order: the Store's write gate closes, then the merge coordinator is aborted. + store.closeWrites(); + const closing = coordinator.close(); + await gh.stopped; + gh.release(); + const error = await merging as Error; + await closing; + // Not saved (writes are closed), so resendable; but gh's answer is the reason given, not the shutdown. + expect(error).toBeInstanceOf(MergeNotApplied); + expect(error.message).toMatch(/Could not resolve to a PullRequest/); + expect(error.message).not.toMatch(/shutdown/); + expect(store.savedAction(identity, { actionId, kind: 'merge', request: { token: view.token } })).toBeUndefined(); + } finally { store.close(); } +}); + +it('keeps a refusal\'s text when a storage error stops it from being saved', async () => { + const identity = { repositoryId: 'repo', taskId: 'task', planId: 'plan' }; + const store = new Store(':memory:'); + const plan = { schema_version: 1 as const, revision: 1, issue: 24, summary: 'Merge', questions: [], items: [{ id: 'P1', title: 'Merge', intent: 'Merge safely', files: [{ path: 'a', kind: 'edit' as const, renamed_from: null, change: 'Change' }], acceptance: [{ type: 'check' as const, text: 'Works' }], depends_on: [] }] }; + store.createPlan(JSON.stringify(plan), 'json', { identity, issue: 24, baseEntries: [{ path: 'a', kind: 'file' as const }], pathKey: (path: string) => path, allowedCommands: [] }, sha('a'), sha('b')); + const view = readyView({ expected: { revision: 1, snapshotId: store.getSnapshot(identity).id, reviewVersion: store.reviewVersion(identity) } }); + const answer = new CommandFailed('gh failed (exit 1): GraphQL: Could not resolve to a PullRequest with the number of 7.', '', ''); + const run: Run = async () => { throw answer; }; + const coordinator = new MergeCoordinator({ store, config: { identity }, load: vi.fn(() => view) } as unknown as ReviewService, new GhMergeGateway(config, run)); + vi.spyOn(store, 'userAction').mockImplementation(() => { throw Object.assign(new Error('database is locked'), { code: 'ERR_SQLITE_ERROR' }); }); + try { + const error = await coordinator.merge(view.token, randomUUID()).catch((error: unknown) => error) as Error; + expect(error).toBeInstanceOf(MergeNotApplied); + expect(error.message).toMatch(/could not be saved.*Could not resolve to a PullRequest/); + } finally { vi.restoreAllMocks(); await coordinator.close(); store.close(); } +}); + +it('coordinator reports the inspection timeout that stopped validation, not a shutdown that arrived while gh was stopping', async () => { + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }); + const gh = stoppingRun(); + const view = readyView(); + const coordinator = new MergeCoordinator({ load: vi.fn(() => view) } as unknown as ReviewService, new GhMergeGateway(config, gh.run)); + const merging = coordinator.merge(view.token).catch((error: unknown) => error); + await gh.started; + await vi.advanceTimersByTimeAsync(6_000); // The fresh inspection's 6 s deadline fires; the 14 s merge deadline has not. + await gh.stopped; + const closing = coordinator.close(); // Shutdown aborts the merge while the timed-out gh call is still stopping. + gh.release(); + const error = await merging as Error; + await closing; + expect(error).toBeInstanceOf(MergeNotApplied); + expect(error.message).toBe('GitHub merge-state inspection timed out.'); +}); + +it.each([false, true])('a gateway inspection timeout before admission is resendable, not saved as the click\'s definite refusal (saved meanwhile by another coordinator: %s)', async savedMeanwhile => { + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }); + const identity = { repositoryId: 'repo', taskId: 'task', planId: 'plan' }; + const store = new Store(':memory:'); + const plan = { schema_version: 1 as const, revision: 1, issue: 24, summary: 'Merge', questions: [], items: [{ id: 'P1', title: 'Merge', intent: 'Merge safely', files: [{ path: 'a', kind: 'edit' as const, renamed_from: null, change: 'Change' }], acceptance: [{ type: 'check' as const, text: 'Works' }], depends_on: [] }] }; + store.createPlan(JSON.stringify(plan), 'json', { identity, issue: 24, baseEntries: [{ path: 'a', kind: 'file' as const }], pathKey: (path: string) => path, allowedCommands: [] }, sha('a'), sha('b')); + const view = readyView({ expected: { revision: 1, snapshotId: store.getSnapshot(identity).id, reviewVersion: store.reviewVersion(identity) } }); + const gh = stoppingRun(); + const coordinator = new MergeCoordinator({ store, config: { identity }, load: vi.fn(() => view) } as unknown as ReviewService, new GhMergeGateway(config, gh.run)); + const actionId = randomUUID(); + try { + const merging = coordinator.merge(view.token, actionId).catch((error: unknown) => error); + await gh.started; + // Another coordinator on this database settles the same click with a refusal while this one's read runs. + if (savedMeanwhile) store.userAction(identity, { actionId, kind: 'merge', request: { token: view.token } }, + () => ({ attemptId: 'other', state: 'failed', reason: 'Required review is missing.', url: null })); + await vi.advanceTimersByTimeAsync(6_000); + gh.release(); + const error = await merging as Error; + // The saved outcome is the answer; without one, the timeout is. + if (savedMeanwhile) { expect(error).not.toBeInstanceOf(MergeNotApplied); expect(error.message).toBe('Required review is missing.'); return; } + expect(error.message).toBe('GitHub merge-state inspection timed out.'); + // A timeout applied nothing: the click must stay resendable (503), with nothing saved under its key. + expect(error).toBeInstanceOf(MergeNotApplied); + // Nothing is saved under the click's key: a saved refusal would be thrown here instead. + expect(store.savedAction(identity, { actionId, kind: 'merge', request: { token: view.token } })).toBeUndefined(); + } finally { await coordinator.close(); store.close(); } +}); + +it('answers an outcome another coordinator saved meanwhile, even when a timeout is followed by this click\'s own abort', async () => { + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }); + const identity = { repositoryId: 'repo', taskId: 'task', planId: 'plan' }; + const store = new Store(':memory:'); + const plan = { schema_version: 1 as const, revision: 1, issue: 24, summary: 'Merge', questions: [], items: [{ id: 'P1', title: 'Merge', intent: 'Merge safely', files: [{ path: 'a', kind: 'edit' as const, renamed_from: null, change: 'Change' }], acceptance: [{ type: 'check' as const, text: 'Works' }], depends_on: [] }] }; + store.createPlan(JSON.stringify(plan), 'json', { identity, issue: 24, baseEntries: [{ path: 'a', kind: 'file' as const }], pathKey: (path: string) => path, allowedCommands: [] }, sha('a'), sha('b')); + const view = readyView({ expected: { revision: 1, snapshotId: store.getSnapshot(identity).id, reviewVersion: store.reviewVersion(identity) } }); + const gh = stoppingRun(); + const coordinator = new MergeCoordinator({ store, config: { identity }, load: vi.fn(() => view) } as unknown as ReviewService, new GhMergeGateway(config, gh.run)); + const actionId = randomUUID(); + try { + const merging = coordinator.merge(view.token, actionId).catch((error: unknown) => error); + await gh.started; + store.userAction(identity, { actionId, kind: 'merge', request: { token: view.token } }, + () => ({ attemptId: 'other', state: 'failed', reason: 'Required review is missing.', url: null })); + await vi.advanceTimersByTimeAsync(6_000); + await gh.stopped; // Before close(), only the inspection's own timeout can have stopped gh. + const closing = coordinator.close(); // Then this coordinator's shutdown aborts the click, before gh settles. + gh.release(); + const error = await merging as Error; + await closing; + // The durable outcome, not a resendable 503 for the timeout. + expect(error).not.toBeInstanceOf(MergeNotApplied); + expect(error.message).toBe('Required review is missing.'); + } finally { store.close(); } +}); + +it('answers a success saved meanwhile after the click\'s own abort without loading the review or reading GitHub again', async () => { + const identity = { repositoryId: 'repo', taskId: 'task', planId: 'plan' }; + const store = new Store(':memory:'); + const plan = { schema_version: 1 as const, revision: 1, issue: 24, summary: 'Merge', questions: [], items: [{ id: 'P1', title: 'Merge', intent: 'Merge safely', files: [{ path: 'a', kind: 'edit' as const, renamed_from: null, change: 'Change' }], acceptance: [{ type: 'check' as const, text: 'Works' }], depends_on: [] }] }; + store.createPlan(JSON.stringify(plan), 'json', { identity, issue: 24, baseEntries: [{ path: 'a', kind: 'file' as const }], pathKey: (path: string) => path, allowedCommands: [] }, sha('a'), sha('b')); + const view = readyView({ expected: { revision: 1, snapshotId: store.getSnapshot(identity).id, reviewVersion: store.reviewVersion(identity) } }); + const gh = stoppingRun(); + let laterCalls = 0, first = true; + // The click's first gh call is held until shutdown stops it; any later call (a replay's GitHub read) is counted. + const run: Run = (args, options) => { + if (first) { first = false; return gh.run(args, options); } + laterCalls++; + return Promise.reject(options?.signal?.reason ?? new Error('Unexpected gh call.')); + }; + const load = vi.fn(() => view); + const coordinator = new MergeCoordinator({ store, config: { identity }, load } as unknown as ReviewService, new GhMergeGateway(config, run)); + const actionId = randomUUID(); + try { + const merging = coordinator.merge(view.token, actionId); + await gh.started; + // Another coordinator on this database merged the same click meanwhile. + store.userAction(identity, { actionId, kind: 'merge', request: { token: view.token } }, + () => ({ attemptId: 'other', state: 'merged', reason: null, url: 'https://github.com/owner/repo/pull/7' })); + const loadsBefore = load.mock.calls.length; + const closing = coordinator.close(); // Shutdown aborts the click while its gh call runs. + await gh.stopped; + gh.release(); + const answer = await merging; + await closing; + expect(answer.result).toEqual({ url: 'https://github.com/owner/repo/pull/7' }); + expect(answer.status.blockers).toEqual([expect.objectContaining({ code: 'refresh' })]); // It asks for a refresh instead. + expect(load.mock.calls.length).toBe(loadsBefore); + expect(laterCalls).toBe(0); + } finally { store.close(); } +}); + +it('fails a command whose output passes its limit while its pipes drain after it exited', async () => { + const dir = mkdtempSync(join(tmpdir(), 'merge-stop-buffer-')); + const script = join(dir, 'gh.cjs'); + // The command exits 0 at once; a grandchild that inherited its stdout writes past the limit 0.3 s later. + writeFileSync(script, ` + const { spawn } = require('node:child_process'); + spawn(process.execPath, ['-e', 'setTimeout(() => { process.stdout.write("x".repeat(4096)); }, 300)'], { stdio: ['ignore', 'inherit', 'inherit'], detached: true }).unref(); + `); + try { + const started = Date.now(); + const error = await runWithInput(process.execPath, [script], { maxBuffer: 1024, killGraceMs: 100, pipeGraceMs: 3_000 }).then(() => null, (failure: unknown) => failure); + expect((error as Error | null)?.message).toMatch(/exceeded its limit/); + expect(Date.now() - started).toBeGreaterThanOrEqual(250); // The output came during the drain, after the exit. + } finally { rmSync(dir, { recursive: true, force: true }); } +}, 10_000); + +it.each([ + ['refusal', 1], ['success', 0], +] as const)('gateway merge keeps GitHub\'s %s when the deadline fires after gh exited but before its pipes closed', async (answer, exitCode) => { + const dir = mkdtempSync(join(tmpdir(), 'issue-48-reason-')); + const script = join(dir, 'gh.cjs'); + // gh exits with its answer (1 after printing a refusal, or 0); a grandchild that inherited stdout/stderr keeps the + // pipes open, as runWithInput allows. Only the exit status decides: a success ignores the refusal text on stderr. + writeFileSync(script, ` + const { spawn } = require('node:child_process'); const fs = require('node:fs'); const path = require('node:path'); + const dir = process.argv[2]; + const g = spawn(process.execPath, ['-e', 'setTimeout(() => {}, 20000)'], { stdio: ['ignore', 'inherit', 'inherit'], detached: true }); + fs.writeFileSync(path.join(dir, 'grandchild'), String(g.pid)); g.unref(); + process.stderr.write('GraphQL: Head branch was modified. Review and try the merge again. (mergePullRequest)\\n'); + fs.writeFileSync(path.join(dir, 'child'), String(process.pid)); + process.exitCode = ${exitCode}; + `); + const run: Run = (_args, options) => runWithInput(process.execPath, [script, dir], { signal: options?.signal, killGraceMs: 250, pipeGraceMs: 3_000 }); + const abort = new AbortController(); + try { + let settled = false; + const merging = new GhMergeGateway(config, run).merge(sha('b'), { signal: abort.signal }).catch((error: unknown) => error) + .finally(() => { settled = true; }); + // Wait until gh has been reaped: Node has then emitted its 'exit' and runWithInput is in its pipe grace period. + const gone = (pid: number) => { try { process.kill(pid, 0); return false; } catch { return true; } }; + for (let i = 0; i < 1_000 && !(existsSync(join(dir, 'child')) && readFileSync(join(dir, 'child'), 'utf8') && gone(Number(readFileSync(join(dir, 'child'), 'utf8')))); i++) + await new Promise(resolve => setTimeout(resolve, 5)); + expect(gone(Number(readFileSync(join(dir, 'child'), 'utf8')))).toBe(true); + // Still draining the pipes: the abort below lands after exit and before the answer, the window under test. + expect(settled).toBe(false); + abort.abort(new Error('Merge request deadline exceeded.')); + const outcome = await merging; + if (answer === 'success') expect(outcome).toEqual({ url: 'https://github.com/owner/repo/pull/7' }); + else { + expect(outcome).toBeInstanceOf(MergeSubmissionError); + expect((outcome as MergeSubmissionError).message).toMatch(/Head branch was modified/); + expect((outcome as MergeSubmissionError).outcome).toBe('refused'); + } + } finally { + try { process.kill(Number(readFileSync(join(dir, 'grandchild'), 'utf8'))); } catch {} + rmSync(dir, { recursive: true, force: true }); + } +}, 15_000); + +it('an issue read that runs out of time reports a RemoteTimeout', async () => { + // A consistency check, not a behaviour guard: execution and planning issue reads share this deadline, and nothing on + // those paths reads the class yet. (The merge path's trust read is tested through the server in merge-queue-poll.) + vi.useFakeTimers(); + try { + const read = withIssueReadDeadline(new AbortController().signal, signal => new Promise((_resolve, reject) => + signal.addEventListener('abort', () => reject(signal.reason), { once: true }))); + const outcome = read.catch((error: unknown) => error); + await vi.advanceTimersByTimeAsync(ISSUE_READ_ACTIVE_MS); + const error = await outcome; + expect(error).toBeInstanceOf(RemoteTimeout); + expect((error as Error).message).toBe('Issue retrieval timed out.'); + } finally { vi.useRealTimers(); } +}); + +it('keeps a command\'s exit status when its own timeout fires after it exited, while its pipes drain', async () => { + const dir = mkdtempSync(join(tmpdir(), 'merge-stop-timeout-')); + const script = join(dir, 'gh.cjs'); + // Exits at once with its answer; a grandchild keeps the pipes open past the command's 1.5 s timeout. + writeFileSync(script, ` + const { spawn } = require('node:child_process'); const fs = require('node:fs'); const path = require('node:path'); + const g = spawn(process.execPath, ['-e', 'setTimeout(() => {}, 20000)'], { stdio: ['ignore', 'inherit', 'inherit'], detached: true }); + fs.writeFileSync(path.join(process.argv[2], 'grandchild'), String(g.pid)); g.unref(); + process.stdout.write('answered'); + `); + try { + expect(await runWithInput(process.execPath, [script, dir], { timeout: 1_500, killGraceMs: 250, pipeGraceMs: 3_000 })).toBe('answered'); + } finally { + try { process.kill(Number(readFileSync(join(dir, 'grandchild'), 'utf8'))); } catch {} + rmSync(dir, { recursive: true, force: true }); + } +}, 15_000); + +// Parallel inspection stages (#48): the branch's PR list fails with gh's own exit, which stops the other stages still +// running (the rule read in flight, and the already-fixed check unless its own timer stopped it first). The inspection +// settles only after every stage has settled, and it reports gh's exit. In production the check is stopped 0.65 s before +// the inspection's deadline (0.35 s into the 1 s used here), and a stopped gh call settles within the runner's 0.4 s. +function parallelStages(listFailsAt: 'once-running' | number) { + const answer = new CommandFailed('gh failed (exit 1): HTTP 502: Bad Gateway', 'HTTP 502: Bad Gateway', ''); + const bothRunning = deferred(), bothStopped = deferred(), releaseRule = deferred(), releaseCheck = deferred(); + let running = 0, stopped = 0; + const observed: { listFailedAfterMs?: number; ruleStopReason?: unknown; checkStopReason?: unknown } = {}; + // A stage that runs until its signal aborts, then settles with that signal's reason when the test releases it. + const held = (release: Promise, signal?: AbortSignal, onStop?: () => void) => new Promise((_, reject) => { + if (++running === 2) bothRunning.resolve(); + signal?.addEventListener('abort', () => { + onStop?.(); + if (++stopped === 2) bothStopped.resolve(); + void release.then(() => reject(signal.reason)); + }, { once: true }); + }); + const pause = (ms: number) => new Promise(resolve => setTimeout(resolve, ms)); + let started = 0; // Set once inspect() has armed its deadline timer, so that timer is due at or before started + 1 s. + const run: Run = async (args, options) => { + const joined = args.join(' '); + if (joined.startsWith('pr view')) return JSON.stringify({ number: 7, isDraft: false, baseRefName: 'main', baseRefOid: sha('a'), headRefName: 'task', + headRefOid: sha('b'), state: 'OPEN', mergeable: 'MERGEABLE', statusCheckRollup: [], body: '', isCrossRepository: false }); + if (joined.includes('/pulls?')) { + // The already-fixed check starts only within the first 0.35 s (before its own stop); under extreme load it may not. + // Checked at 0.5 s, before the 1 s deadline, so the inspection reports this error rather than the deadline. + let premise: NodeJS.Timeout | undefined; + await Promise.race([bothRunning.promise, new Promise((_, reject) => { + premise = setTimeout(() => { + releaseRule.resolve(); releaseCheck.resolve(); // Let the stopped stages settle so the inspection reports this. + reject(new Error('Test premise failed: the rule read and the check did not both start.')); + }, 500); + })]).finally(() => clearTimeout(premise)); + if (listFailsAt !== 'once-running') await pause(listFailsAt - (Date.now() - started)); + observed.listFailedAfterMs = Date.now() - started; + throw answer; + } + // #rules reads one at a time; only the first starts. + return held(releaseRule.promise, options?.signal, () => { observed.ruleStopReason = options?.signal?.reason; }); + }; + const checks: AlreadyFixedGateway = { repository: 'owner/repo', check: (_input, signal) => held(releaseCheck.promise, signal, () => { observed.checkStopReason = signal?.reason; }) }; + const state = { settled: false }; + const inspection = new GhMergeGateway(config, run, checks).inspect({ fresh: true, timeoutMs: 1_000, target: { pullRequest: 7, headBranch: 'task' } }) + .then(() => null, (error: unknown) => error).finally(() => { state.settled = true; }); + started = Date.now(); + // Both stages stopped; an inspection that settles first (for example with the premise error above) fails the test with it. + const whenStopped = () => Promise.race([bothStopped.promise, inspection.then(error => { + throw error instanceof Error ? error : new Error('The inspection settled before both stages stopped.'); + })]); + return { answer, whenStopped, releaseRule: () => releaseRule.resolve(), releaseCheck: () => releaseCheck.resolve(), + state, inspection, pause, started, observed }; +} + +it('merge-state inspection waits for a stopped already-fixed check before reporting a stage\'s gh exit', async () => { + const stages = parallelStages('once-running'); + await stages.whenStopped(); + expect(stages.observed.checkStopReason).toBe(stages.answer); // The list's exit, not the check's own timer, stopped it. + stages.releaseRule(); + await stages.pause(50); + expect(stages.state.settled).toBe(false); // The already-fixed check is still settling. + stages.releaseCheck(); + expect(await stages.inspection).toBe(stages.answer); +}); + +it('merge-state inspection handles a stopped already-fixed check that settles before the other stages', async () => { + const unhandled: unknown[] = []; + const record = (reason: unknown) => { unhandled.push(reason); }; + process.on('unhandledRejection', record); + try { + const stages = parallelStages('once-running'); + await stages.whenStopped(); + expect(stages.observed.checkStopReason).toBe(stages.answer); // The list's exit, not the check's own timer, stopped it. + stages.releaseCheck(); // The check rejects with the list's exit while the rule read is still stopping. + await stages.pause(50); + expect(stages.state.settled).toBe(false); + stages.releaseRule(); + expect(await stages.inspection).toBe(stages.answer); + await stages.pause(0); + expect(unhandled).toEqual([]); + } finally { process.off('unhandledRejection', record); } +}); + +it('merge-state inspection reports a stage\'s gh exit, not its deadline, when a stopped rule read settles after the deadline', async () => { + // The list fails 0.75 s in: after the check's own timer stopped it (0.35 s), within 0.4 s of the 1 s deadline. + const stages = parallelStages(750); + stages.releaseCheck(); + await stages.whenStopped(); + await stages.pause(1_050 - (Date.now() - stages.started)); + // The order this test claims: the list failed before the deadline and stopped the rule read; then the deadline fired. + // (Under heavy load the list can fail late; these checks then fail rather than pass on a different interleaving.) + expect(stages.observed.listFailedAfterMs).toBeLessThan(1_000); + expect(stages.observed.ruleStopReason).toBe(stages.answer); + expect((stages.observed.checkStopReason as Error).message).toBe('The already-fixed check did not finish in time.'); // Its own timer. + expect(Date.now() - stages.started).toBeGreaterThanOrEqual(1_000); // The deadline has fired. + expect(stages.state.settled).toBe(false); // The rule read, stopped 0.3 s ago, is still settling. + stages.releaseRule(); + expect(await stages.inspection).toBe(stages.answer); +}, 10_000); + +it('merge-state inspection answers a caller abort, not the already-fixed check\'s own earlier gh failure', async () => { + // The check's gh call failed before the caller aborted; the failure surfaces only after the abort, while gh stops. + const own = new CommandFailed('gh failed (exit 1): HTTP 502: Bad Gateway', 'HTTP 502: Bad Gateway', ''); + const caller = new AbortController(), reason = new Error('Merge cancelled during shutdown.'); + const bothRunning = deferred(); + let running = 0; + const afterAbort = (signal: AbortSignal | undefined, settle: (reject: (error: unknown) => void) => void) => new Promise((_, reject) => { + if (++running === 2) bothRunning.resolve(); + signal?.addEventListener('abort', () => settle(reject), { once: true }); + }); + const run: Run = async (args, options) => { + if (args.join(' ').startsWith('pr view')) return JSON.stringify({ number: 7, isDraft: false, baseRefName: 'main', baseRefOid: sha('a'), + headRefName: 'feature', headRefOid: sha('b'), state: 'OPEN', mergeable: 'MERGEABLE', statusCheckRollup: [] }); + return afterAbort(options?.signal, reject => reject(options!.signal!.reason)); // The rule read in flight. + }; + let checkStopReason: unknown; + const checks: AlreadyFixedGateway = { repository: 'owner/repo', check: (_input, signal) => afterAbort(signal, reject => { checkStopReason = signal!.reason; reject(own); }) }; + // The default 12 s inspection: the check's own stop timer (0.65 s before that) cannot fire first. + const inspection = new GhMergeGateway(config, run, checks).inspect({ fresh: true, signal: caller.signal }) + .then(() => null, (error: unknown) => error); + await bothRunning.promise; + caller.abort(reason); + expect(checkStopReason).toBe(reason); // The caller's abort, not the check's own stop timer, stopped the check. + // Not the check's gh text: a blocker result that arrives after the abort is discarded, so the click may be resent. + expect(await inspection).toBe(reason); +}); diff --git a/test/pre-merge.test.ts b/test/pre-merge.test.ts index 10e7e27d..806c37e9 100644 --- a/test/pre-merge.test.ts +++ b/test/pre-merge.test.ts @@ -14,6 +14,9 @@ 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 { UpstreamFailure } from '../runner/lifecycle.ts'; +import { RemoteTimeout } from '../github/timeout.ts'; +import { CommandFailed } from '../github/run-with-input.ts'; // A passthrough, so a test can act while a preparation's worker history read is in flight. const historyReads = vi.hoisted(() => ({ during: undefined as (() => void) | undefined })); @@ -70,8 +73,11 @@ async function rebaseFixture(rebasedText: string, commandExit?: number, options: rebaseCleanupFailure?: Error; runnerCommitted?: boolean; operationTimeoutMs?: number; - authorize?: (signal: AbortSignal) => Promise; + /** The second argument lets an authorization stop the preparation while its read is settling. */ + authorize?: (signal: AbortSignal, fixture: { coordinator: PreMergeCoordinator }) => Promise; reserves?: { processMs?: number; commandMs?: number }; + /** Saves the preparation's outcome under this action ID, as the server's prepare-merge action does. */ + actionId?: string; } = {}) { const root = mkdtempSync(join(tmpdir(), 'codeboost-pre-merge-base-')); roots.push(root); const repository = join(root, 'repo'); fixtureGit(root, 'init', '-q', '-b', 'main', repository); @@ -176,7 +182,7 @@ async function rebaseFixture(rebasedText: string, commandExit?: number, options: }, context.allowedCommands, 'a'.repeat(32))); let remoteReads = 0, remotePair = { base: onto, head }; let authorizationChecks = 0; - const authorize = options.authorize ?? (options.duringAuthorize ? async () => ({ + const authorize = options.authorize ? (signal: AbortSignal) => options.authorize!(signal, { coordinator }) : (options.duringAuthorize ? async () => ({ refresh: async () => undefined, validate: () => { remotePair = options.duringAuthorize!(service, ++authorizationChecks) ?? remotePair; }, }) : undefined); @@ -200,9 +206,12 @@ async function rebaseFixture(rebasedText: string, commandExit?: number, options: }) as RunnerCoordinator['start']; } const task = service.store.getTask(identity); + const actionRequest = { attemptId: undefined, expectedStateVersion: task.stateVersion, expectedReviewVersion: view.expected.reviewVersion }; + if (options.actionId) service.store.userAction(identity, { actionId: options.actionId, kind: 'prepare-merge', request: actionRequest }, () => ({ outcome: 'preparing' })); 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, + reviewVersion: view.expected.reviewVersion!, snapshotId: view.snapshot.id, base: view.snapshot.base, head: view.snapshot.head, + ...(options.actionId ? { actionId: options.actionId } : {}) }); + return { service, coordinator, runner, result, rebased, checkedHeads, cancelReasons, actionRequest, rebaseRuns: () => rebaseRuns, rebaseAborts: () => rebaseAborts, rebaseBudgets, abortBudgets }; } @@ -269,6 +278,28 @@ it('keeps a failed rebase bound to the review that started it when that review c await fixture.coordinator.close(); await fixture.runner.close(); }); +it.each([ + ['timed out', () => new RemoteTimeout('Issue retrieval timed out.')], + ['was answered by gh', () => new CommandFailed('gh failed (exit 1): HTTP 404: Not Found', 'HTTP 404: Not Found', '')], +] as const)('saves a preparation whose issue-trust read %s before shutdown with that answer, not the shutdown (#48)', async (_case, answer) => { + const actionId = randomUUID(); + let stoppedFirst = false; + const cause = answer(), expected = `Issue trust could not be verified: ${cause.message}`; + const fixture = await rebaseFixture('feature\n', undefined, { actionId, authorize: async (signal, { coordinator }) => { + // The read answered (or ran out of time) before shutdown, and surfaces while shutdown stops the preparation. This is + // the error the server's readIssueAccess builds then: its own answer, wrapped, with the answer as the cause. + void coordinator.close(); + stoppedFirst = signal.aborted; + throw new UpstreamFailure(expected, { cause }); + } }); + expect(stoppedFirst).toBe(true); + expect(fixture.result).toMatchObject({ state: 'failed', reason: expected }); + expect(fixture.service.store.savedAction(fixture.service.config.identity, { actionId, kind: 'prepare-merge', request: fixture.actionRequest })?.response) + .toMatchObject({ outcome: 'failed', reason: expected }); + expect(fixture.rebaseRuns()).toBe(0); + await fixture.runner.close(); +}); + it('revalidates authorization before starting the delayed local rebase', async () => { const calls: string[] = []; const fixture = await rebaseFixture('feature\n', undefined, { authorize: async () => { diff --git a/test/runner-publishing.test.ts b/test/runner-publishing.test.ts index 07ca51a8..4ba9f8d3 100644 --- a/test/runner-publishing.test.ts +++ b/test/runner-publishing.test.ts @@ -24,6 +24,8 @@ import type { PlanIdentity } from '../core/identity.ts'; import { approveItem } from '../core/approvals.ts'; import { GhIssueGateway, type IssueTrustGateway } from '../github/issues.ts'; import type { PreMergeCoordinator } from '../runner/pre-merge.ts'; +import { RemoteTimeout } from '../github/timeout.ts'; +import { CommandFailed } from '../github/run-with-input.ts'; vi.setConfig({ testTimeout: 30_000 }); const roots: string[] = [], cleanups: (() => Promise | void)[] = []; @@ -609,6 +611,31 @@ describe('publishing a finished task (#103)', () => { expect(w.github.calls).toEqual([]); }); + it.each([ + ['timed out', () => new RemoteTimeout('Issue retrieval timed out.'), /could not be verified: Issue retrieval timed out/], + ['was answered by gh', () => new CommandFailed('gh failed (exit 1): HTTP 404: Not Found', 'HTTP 404: Not Found', ''), /could not be verified: gh failed \(exit 1\): HTTP 404/], + ] as const)('records a publish whose issue-trust read %s before shutdown as failed with that answer, not stopped (#48)', async (_case, answer, expected) => { + const w = world(), reading = Promise.withResolvers(), release = Promise.withResolvers(); + let accessSignal: AbortSignal | undefined; + const issueGateway: IssueTrustGateway = { + repository: REPO, + async fetch() { return { repository: REPO, retrievedAt: new Date().toISOString(), issues: [] }; }, + // The read answered (or ran out of time) before shutdown; its answer surfaces only after shutdown aborted it. + async issueAccess(_number, options = {}) { accessSignal = options.signal; reading.resolve(); await release.promise; throw answer(); }, + async issueText(number) { return { number, title: '', body: '', comments: [] }; }, + }; + const { app, store, close } = await serve(w, { startup: false, issueGateway, before: completeAll }); + const records = vi.spyOn(store, 'recordPublish'); + app.publishOwed(() => undefined); + await reading.promise; + const closing = close(); + await vi.waitFor(() => expect(accessSignal?.aborted).toBe(true)); + release.resolve(); + await closing; + expect(records.mock.calls.at(-1)?.[1]).toMatchObject({ outcome: 'failed', message: expect.stringMatching(expected) }); + expect(w.github.calls).toEqual([]); + }); + it('closes the Store only after a publish stopped during its push has settled, and the next start publishes once', async () => { const w = world(); let closedWhileBusy: boolean | undefined; diff --git a/test/runner-start.test.ts b/test/runner-start.test.ts index 3bc16610..fabe3bb9 100644 --- a/test/runner-start.test.ts +++ b/test/runner-start.test.ts @@ -14,6 +14,7 @@ import type { AttemptKind } from '../runner/lifecycle.ts'; import { approveItem } from '../core/approvals.ts'; import type { IssueTrustGateway } from '../github/issues.ts'; import { fixtureGit } from './fixtures/git.ts'; +import { RemoteTimeout } from '../github/timeout.ts'; vi.setConfig({ testTimeout: 20_000 }); const roots: string[] = [], cleanups: (() => Promise | void)[] = []; @@ -131,7 +132,20 @@ describe('start (#91 part 2)', () => { body: { error: expect.stringMatching(/could not be verified.*incomplete collaborator page/i) } }); expect(store.getAttempts(identity)).toEqual([]); }); - it('returns the first durable success when an identical concurrent access read fails later', async () => { + it('does not record a collaborator read that timed out, so the same action ID reads trust again', async () => { + let result: ReturnType[0]> = new RemoteTimeout('Issue retrieval timed out.'); + const { app, identity, store } = await serve({ issueGateway: trustGateway(() => result) }); + const actionId = randomUUID(); + expect(await act(app, 'start', { actionId })).toMatchObject({ status: 503, + body: { error: 'Issue trust could not be verified: Issue retrieval timed out.' } }); + expect(store.getAttempts(identity)).toEqual([]); + result = { authorLogin: 'member', collaborator: true }; + expect((await act(app, 'start', { actionId })).body.result).toMatchObject({ outcome: 'started', item: 'P1' }); + }); + it.each([ + ['fails', () => new Error('later GitHub failure')], + ['times out', () => new RemoteTimeout('Issue retrieval timed out.')], + ])('returns the first durable success when an identical concurrent access read %s later', async (_case, failure) => { const secondStarted = Promise.withResolvers(), releaseFailure = Promise.withResolvers(); let reads = 0; const gateway: IssueTrustGateway = { @@ -139,7 +153,7 @@ describe('start (#91 part 2)', () => { async fetch() { return { repository: 'owner/repo', retrievedAt: new Date().toISOString(), issues: [] }; }, async issueAccess(number) { if (++reads === 1) { await secondStarted.promise; return { number, authorLogin: 'member', collaborator: true }; } - secondStarted.resolve(); await releaseFailure.promise; throw new Error('later GitHub failure'); + secondStarted.resolve(); await releaseFailure.promise; throw failure(); }, async issueText(number) { return { number, title: '', body: '', comments: [] }; }, }; @@ -153,6 +167,31 @@ describe('start (#91 part 2)', () => { expect(await second).toMatchObject({ status: 200, body: { result: accepted.body.result } }); expect(store.getAttempts(identity)).toHaveLength(1); }); + it.each([false, true])('keeps a trust read\'s timeout when shutdown starts while gh stops (outcome saved meanwhile: %s)', async savedMeanwhile => { + const readStarted = Promise.withResolvers(), releaseTimeout = Promise.withResolvers(); + const gateway: IssueTrustGateway = { + repository: 'owner/repo', + async fetch() { return { repository: 'owner/repo', retrievedAt: new Date().toISOString(), issues: [] }; }, + // The read's own timeout fired; gh is still stopping when shutdown starts, and the timeout surfaces after that. + async issueAccess() { readStarted.resolve(); await releaseTimeout.promise; throw new RemoteTimeout('Issue retrieval timed out.'); }, + async issueText(number) { return { number, title: '', body: '', comments: [] }; }, + }; + const { app, close, identity, store } = await serve({ issueGateway: gateway }); + const { stateVersion, reviewVersion } = await view(app); + const actionId = randomUUID(); + const response = act(app, 'start', { actionId, expectedStateVersion: stateVersion, expectedReviewVersion: reviewVersion }); + await readStarted.promise; + // Distinct from what an admitted start answers, so the answer itself shows nothing was admitted. + const saved = { outcome: 'saved meanwhile' }; + if (savedMeanwhile) store.userAction(identity, { actionId, kind: 'start', request: { expectedStateVersion: stateVersion, expectedReviewVersion: reviewVersion } }, () => saved); + const closing = close(); + await new Promise(resolve => setTimeout(resolve, 50)); // Shutdown has started and is draining this request. + releaseTimeout.resolve(); + const answer = await response; + await closing; + if (savedMeanwhile) expect(answer).toMatchObject({ status: 200, body: { result: saved } }); + else expect(answer).toMatchObject({ status: 503, body: { error: 'Issue trust could not be verified: Issue retrieval timed out.' } }); + }); it('requires every item of the current plan revision to be approved before start', async () => { const { app, identity, store } = await serve({ approve: false }); expect(await view(app)).toMatchObject({ startable: false }); diff --git a/web/server.ts b/web/server.ts index f1a4dc85..9ab4c464 100644 --- a/web/server.ts +++ b/web/server.ts @@ -11,6 +11,8 @@ import { RunnerCoordinator, type RunnerDeps } from '../runner/coordinator.ts'; import { ItemExecutor } from '../runner/execution.ts'; import { OWED_REFUSAL, TaskPublishing } from '../runner/publishing.ts'; import type { RunnerAssembly } from '../runner/production.ts'; +import { ghAnswered } from '../github/run-with-input.ts'; +import { timedOut } from '../github/timeout.ts'; import { baseBranch } from '../github/validate.ts'; import type { ShutdownCapability } from '../runner/lifecycle.ts'; import { ActionIdReused, BadRequest, GuardRefusal, ShuttingDownError, UpstreamFailure, assertUuidV4, isUuidV4, sameContext } from '../runner/lifecycle.ts'; @@ -157,8 +159,9 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge if (!trustGateway || !config.github) throw new GuardRefusal('Issue trust admission is not configured.'); try { return await trustGateway.issueAccess(config.github.issue, { signal, timeoutMs: 12_000 }); } catch (error) { - if (signal.aborted) throw signal.reason; - throw new UpstreamFailure(`Issue trust could not be verified: ${error instanceof Error ? error.message : String(error)}`); + // A read that timed out, or that gh answered, before the caller's abort reports that, not the abort (#48). + if (signal.aborted && !timedOut(error) && !ghAnswered(error)) throw signal.reason; + throw new UpstreamFailure(`Issue trust could not be verified: ${error instanceof Error ? error.message : String(error)}`, { cause: error }); } }; authorizePublish = async (_publishIdentity, signal) => { @@ -341,6 +344,14 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge if (stopping || runner?.closing) throw new ShuttingDownError(); try { access = await readIssueAccess(signal); } catch (error) { + // A trust read that ran out of time answered nothing: it is not this action's outcome, so it is not saved, the + // answer is 503, and the same action ID reads trust again (#48). An outcome saved meanwhile is still the answer. + // Its timeout came first, so a shutdown that started while gh stopped does not replace it; nothing is admitted. + if (timedOut(error)) { + const saved = service.store.savedAction(identity, { actionId: actionId as string, kind: action as string, request }); + if (saved) return saved.response; + throw error; + } if (stopping || runner?.closing || signal.aborted) throw new ShuttingDownError(); return service.store.userAction(identity, { actionId: actionId as string, kind: action as string, request }, () => { throw error; }).response; } @@ -687,7 +698,8 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge if (error instanceof MergeOutcomeUnknown) { json(409,{error:error.message,outcomeUnknown:true}); return; } throw error; } - try { const mergeQueue=merges.queueSnapshot();json(200,{...await loadReview(requestAbort.signal),merge:{...merged.status,queue:mergeQueue},mergeResult:merged.result,mergeQueue,mergeRefreshRequired:false}); } + // The queue snapshot and the attempt's status are read together, after the review load, so they agree. + try { const review=await loadReview(requestAbort.signal),mergeQueue=merges.queueSnapshot();json(200,{...review,merge:{...merges.statusAfterMerge(merged.status),queue:mergeQueue},mergeResult:merged.result,mergeQueue,mergeRefreshRequired:false}); } catch { json(200,{mergeResult:merged.result,mergeQueue:null,mergeRefreshRequired:true}); } return; } @@ -748,6 +760,10 @@ export async function startServer(config: ReviewConfig, port = 4318, questionAge json(503, { error: error instanceof Error ? error.message : 'Storage failed.', outcomeUnknown: true }); return; } if (error instanceof BadRequest) { json(400, { error: error.message }); return; } + // A GitHub read that ran out of time decided nothing: resend with the same key (#48). Precondition: no caller saves + // an UpstreamFailure caused by a RemoteTimeout as an HTTP action's answer (only readIssueAccess builds one; merge + // preparation and publishing keep it in their own background records). + if (error instanceof UpstreamFailure && timedOut(error)) { json(503, { error: error.message }); return; } if (error instanceof UpstreamFailure) { json(502, { error: error.message }); return; } json(409, { error: error instanceof Error ? error.message : 'Review failed.', ...unknown }); }