diff --git a/docs/plans/2026-09-27-001-feat-ci-repair-plan.md b/docs/plans/2026-09-27-001-feat-ci-repair-plan.md index 1cb67f5..c7cea36 100644 --- a/docs/plans/2026-09-27-001-feat-ci-repair-plan.md +++ b/docs/plans/2026-09-27-001-feat-ci-repair-plan.md @@ -31,7 +31,7 @@ The outside reference is OpenAI's account of Codex babysitting a PR until CI is - R3. A round sends the failing checks to the `run` phase as its input envelope, then runs every phase after `run` to the end of the chain, with every `if:` guard, gate and `on_fail` edge live. Phases before `run` do not re-run; their envelope fields carry over. - R4. After the chain, acceptance and `publish.hold_when` are judged again over the merged field view. If acceptance fails or the hold now holds, nothing is pushed. - R5. A fix is pushed to the same attempt-scoped branch without force, and the push is fenced and recorded as a numbered round on the control plane before and after the side effect. -- R6. The control plane enforces the round budget against the frozen snapshot, and refuses `accepted` unless the `ci` record was written after the last recorded round and names none of the heads any round repaired (each of those was judged red). It does not require the green head to equal the last round's `head_after`: a person may push a fix after the rounds run out, and the publish-only retry must still be able to accept it. +- R6. The control plane enforces the round budget against the frozen snapshot, refuses any round once `ci` is recorded green (so a `ci` record always postdates every round), and refuses `accepted` when the `ci` record names any head a round repaired (each of those was judged red). It does not require the green head to equal the last round's `head_after`: a person may push a fix after the rounds run out, and the publish-only retry must still be able to accept it. - R7. Engine wall-clock and prompt sends are shared with the attempt: rounds draw from the same `AttemptCeiling` and `MaxAttemptSends`, so a repair cannot buy an attempt more budget than it was admitted with. - R8. Cancellation stops a round at the next phase boundary. The engine part of a round is never inside the publish critical section, which runs under `context.WithoutCancel` (`internal/worker/publish.go:337`). - R9. If the branch head on the remote is not the head jig last pushed, a person has taken over the branch: no round runs, and the result says so. @@ -54,7 +54,7 @@ The outside reference is OpenAI's account of Codex babysitting a PR until CI is - **KTD6 — Every failure degrades to today's red-CI outcome.** A round that fails acceptance, trips the hold, makes no change, exhausts the budget, breaches a boundary, loses its lease, or finds the head moved pushes nothing further and ends `accepted_unpublished` with the pull request at the last pushed head. The worst case of opting in is therefore today's behaviour plus fix commits that each passed the full downstream chain before they were pushed. -- **KTD7 — A person's push ends automation for that attempt.** Before a round, the worker compares the red head with the head it last pushed (the `push` record or the last round's `head_after`). If they differ, a person has pushed, and jig does not stack an agent's commits on top. The existing retry path already re-judges whatever a person pushed. The non-force push is the second line of defence if a person pushes mid-round. +- **KTD7 — A person's push ends automation for that attempt.** Before a round, the worker compares the red head with the head it last pushed (the `push` record or the last round's `head_after`). If they differ, a person has pushed, and jig does not stack an agent's commits on top. The existing retry path already re-judges whatever a person pushed. The control plane enforces the same rule on the ledger: a round's `head_before` must be the head jig last pushed (the proof ref for round 1, the previous round's `head_after` after that), so a worker that skipped the check is refused before it pushes. The non-force push is the third line if a person pushes mid-round. --- @@ -111,8 +111,8 @@ Round state lives in the worker's `PublishingRunner.Run` loop. The engine owns c ### U2. Control plane: round ledger and the accepted rule - **Files:** `internal/protocol/publish.go` (round record, authorization and record request types), `migrations/005_publish_ci_repairs.sql`, `internal/controlplane/publish_ledger.go`, `internal/controlplane/store.go`, `internal/controlplane/http.go`, `internal/worker/client.go`, tests beside each. -- **Approach:** authorize round N only when the lease is valid, the snapshot declares `on_fail`, N = last round + 1, N ≤ budget, and `proof` is recorded. Record with idempotent replay (same values return the stored row; different values conflict). Extend `CompleteAttempt` (`store.go:555`) so that when rounds exist, `accepted` needs a `ci` record completed after the last round whose `remote_ref` equals no round's `head_before` (R6). -- **Tests:** over-budget authorization refused; zombie lease refused; replay idempotence; `accepted` refused when `ci` predates the last round; migration from 004 keeps existing rows. +- **Approach:** authorize and record round N only when the lease is valid, the branch is the attempt's own, the frozen snapshot declares `on_fail`, `proof` is recorded, no `ci` record exists yet, N = last round + 1, N ≤ budget, and `head_before` is the head jig last pushed (KTD7). Record with idempotent replay (same values return the stored row; different values conflict). `head_after` must differ from `head_before`, and at most `MaxCIRepairFailedChecks` (20) check names of at most 200 bytes each are stored. `RecordPublishStep` refuses a `ci` record whose `remote_ref` equals any round's `head_before` (R6). That has to happen when the row is written, because a `ci` row is permanent and no round is allowed once one exists, so a bad row would leave the job unable ever to be accepted. `CompleteAttempt` (`store.go:555`) repeats the rule as a backstop. The snapshot reader `attemptWaitsForCI` becomes `attemptDefinition`, returning the parsed spec, since both rules read it. +- **Tests:** the chain from proof through two rounds; out-of-order, head-mismatch and over-budget rounds refused at both halves; foreign and expired leases refused; replay idempotence and conflicts; no rounds without `on_fail` or after green; malformed rounds refused; `accepted` refused on a repaired head but allowed on the pushed fix or a person's later fix; the client methods over the real HTTP handler. ### U3. Engine: continuation and `RepairCI` @@ -129,7 +129,7 @@ Round state lives in the worker's `PublishingRunner.Run` loop. The engine owns c ### U5. Worker: the round loop - **Files:** `internal/worker/publish.go`, new `internal/worker/publish_repair.go`, `internal/worker/publish_repair_test.go` -- **Approach:** in `PublishingRunner.Run`, when `publish` returns `ci_failed` and the continuation is present with budget left: the head-moved check (KTD7), authorize the round, `RepairCI` under the attempt's own cancellable context, then commit and non-force push in the critical section, record the round, and call `awaitCI` again as a fresh `ci` step. `PublishSummary` gains `CIRepairs []{Round, HeadBefore, HeadAfter, FailedChecks, Outcome}`. Events: `ci_repair_round_start`, `ci_repair_round_end`, `ci_repair_exhausted`. +- **Approach:** in `PublishingRunner.Run`, when `publish` returns `ci_failed` and the continuation is present with budget left: the head-moved check (KTD7), authorize the round, `RepairCI` under the attempt's own cancellable context, then commit and non-force push in the critical section, record the round, and call `awaitCI` again as a fresh `ci` step. `PublishSummary` gains `CIRepairs []{Round, HeadBefore, HeadAfter, FailedChecks, Outcome}`. Events: `ci_repair_round_start`, `ci_repair_round_end`, `ci_repair_exhausted`. Two traps from the U2 review: the head jig last pushed comes from `AttemptCIRepairs` (the last round's `head_after`, else the proof ref), because `RetryPublish` returns only `publish_records`; and an authorization's `Completed` is returned without comparing `head_before`, so the worker must check that the recorded round repaired the head it is holding before treating the round as done. - **Tests (fake gateway plus scripted runtime):** red → fix → green ends `accepted` with one round recorded; red twice at budget 1 ends `accepted_unpublished` naming both heads; a person's push before round 1 means no round; non-force push rejected mid-round ends cleanly; hold tripped in a round pushes nothing; cancel mid-round leaves the pushed head intact. ### U6. Docs and the stock factory diff --git a/internal/controlplane/ci_repair_ledger.go b/internal/controlplane/ci_repair_ledger.go new file mode 100644 index 0000000..f6e1e69 --- /dev/null +++ b/internal/controlplane/ci_repair_ledger.go @@ -0,0 +1,403 @@ +// ci_repair_ledger.go — the control-plane side of CI repair rounds +// (publish.ci.on_fail). A round is a fix pushed to the attempt's own branch +// after CI went red, and like every publish side effect it is authorized +// before the push and recorded after it, both under the lease token. +// +// The ledger enforces what makes rounds safe to trust: +// - only a definition that declares on_fail may have rounds, and only as +// many as its frozen budget allows; +// - rounds chain: round N repairs exactly the head jig last pushed (the +// proof ref for round 1, round N-1's head_after after that), so a person's +// push in between is refused rather than built on; +// - no round once CI is recorded green, so a ci record always postdates +// every round, and CompleteAttempt can refuse `accepted` on a green run +// that names a head a round already found red. +package controlplane + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "net/http" + + "github.com/StructuPath/jig/internal/protocol" +) + +// validateCIRepairRequest checks everything about a round request that needs +// no database: the token, the round number, the branch, and the head shape. +func validateCIRepairRequest(token string, round int, branch, headBefore, expectedBranch string) error { + if err := validateLeaseToken(token); err != nil { + return err + } + if round < 1 { + return invalid("invalid_ci_repair_round", "round must be 1 or more") + } + if len(branch) > protocol.MaxPublishBranchBytes { + return invalid("invalid_publish_branch", "branch exceeds its storage limit") + } + if branch != "" && branch != expectedBranch { + return invalid("publish_branch_mismatch", fmt.Sprintf( + "an attempt may only publish its own attempt-scoped branch %q", expectedBranch)) + } + if !publishRefPattern.MatchString(headBefore) { + return invalid("invalid_publish_ref", "head_before must be a commit SHA") + } + return nil +} + +// ciRepairBudget is the frozen definition's on_fail budget, or a conflict +// when the definition declares no repair. +func ciRepairBudget(ctx context.Context, tx *sql.Tx, attemptID string) (int, error) { + spec, err := attemptDefinition(ctx, tx, attemptID) + if err != nil { + return 0, err + } + if !spec.WaitsForCI() || spec.Publish.CI.OnFail == nil { + return 0, conflict("ci_repair_not_declared", + "this attempt's definition declares no publish.ci.on_fail, so it has no repair rounds") + } + return spec.Publish.CI.OnFail.Budget, nil +} + +// requireNextCIRepairRound enforces the chain for a round that is not yet +// recorded: CI not yet green, the round is the next one, within budget, and +// it repairs the head jig last pushed. +func requireNextCIRepairRound(ctx context.Context, tx *sql.Tx, attemptID string, round, budget int, headBefore string) error { + var green int + if err := tx.QueryRowContext(ctx, ` + SELECT COUNT(*) FROM publish_records WHERE attempt_id = ? AND step = ? + `, attemptID, protocol.PublishStepCI).Scan(&green); err != nil { + return unavailable(err) + } + if green > 0 { + return conflict("ci_repair_after_green", "CI is already recorded green; there is nothing to repair") + } + var lastRound int + var pushed string + err := tx.QueryRowContext(ctx, ` + SELECT round, head_after FROM publish_ci_repairs + WHERE attempt_id = ? ORDER BY round DESC LIMIT 1 + `, attemptID).Scan(&lastRound, &pushed) + if errors.Is(err, sql.ErrNoRows) { + proof, proven, readErr := readPublishRecord(ctx, tx, attemptID, protocol.PublishStepProof) + if readErr != nil { + return readErr + } + if !proven { + return conflict("publish_step_out_of_order", + "a CI repair round requires the proof step to be recorded first") + } + pushed = proof.RemoteRef + } else if err != nil { + return unavailable(err) + } + if round != lastRound+1 { + return conflict("ci_repair_out_of_order", fmt.Sprintf( + "the next CI repair round is %d, not %d", lastRound+1, round)) + } + if round > budget { + return conflict("ci_repair_budget_exhausted", fmt.Sprintf( + "the definition allows %d CI repair round(s)", budget)) + } + if headBefore != pushed { + return conflict("ci_repair_head_mismatch", fmt.Sprintf( + "a round repairs the head jig last pushed (%s), not %s; "+ + "a different head means someone else pushed to the branch", pushed, headBefore)) + } + return nil +} + +func readCIRepair(ctx context.Context, tx *sql.Tx, attemptID string, round int) (protocol.CIRepairRecord, bool, error) { + var record protocol.CIRepairRecord + var checks string + var completedAt int64 + err := tx.QueryRowContext(ctx, ` + SELECT r.attempt_id, a.job_id, a.attempt_number, r.round, r.branch, + r.head_before, r.head_after, r.failed_checks, r.completed_at + FROM publish_ci_repairs r JOIN attempts a ON a.id = r.attempt_id + WHERE r.attempt_id = ? AND r.round = ? + `, attemptID, round).Scan(&record.AttemptID, &record.JobID, &record.AttemptNumber, + &record.Round, &record.Branch, &record.HeadBefore, &record.HeadAfter, &checks, &completedAt) + if errors.Is(err, sql.ErrNoRows) { + return record, false, nil + } + if err != nil { + return record, false, unavailable(err) + } + if err := json.Unmarshal([]byte(checks), &record.FailedChecks); err != nil { + return record, false, unavailable(err) + } + record.CompletedAt = fromMillis(completedAt) + return record, true, nil +} + +// AuthorizeCIRepair is the pre-push fence for one repair round. It reports +// the budget, and a recorded round as Completed so a re-entering worker +// skips the push instead of repeating it. +func (s *Store) AuthorizeCIRepair( + ctx context.Context, attemptID string, input protocol.CIRepairAuthorizationRequest, +) (protocol.CIRepairAuthorization, error) { + var authorization protocol.CIRepairAuthorization + now := s.now().UnixMilli() + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return authorization, unavailable(err) + } + defer tx.Rollback() + value, err := loadPublishContext(ctx, tx, attemptID) + if err != nil { + return authorization, err + } + if err := validateCIRepairRequest(input.LeaseToken, input.Round, input.Branch, + input.HeadBefore, value.expectedBranch); err != nil { + return authorization, err + } + if err := verifyActiveLease(value.lease, input.LeaseToken, now); err != nil { + return authorization, err + } + budget, err := ciRepairBudget(ctx, tx, attemptID) + if err != nil { + return authorization, err + } + record, exists, err := readCIRepair(ctx, tx, attemptID, input.Round) + if err != nil { + return authorization, err + } + if !exists { + if err := requireNextCIRepairRound(ctx, tx, attemptID, input.Round, budget, input.HeadBefore); err != nil { + return authorization, err + } + } + if err := tx.Commit(); err != nil { + return authorization, unavailable(err) + } + authorization = protocol.CIRepairAuthorization{ + AttemptID: attemptID, Round: input.Round, Budget: budget, Branch: value.expectedBranch, + } + if exists { + authorization.Completed = &record + } + return authorization, nil +} + +// RecordCIRepair records one pushed repair round under the lease token. +// Replaying identical values returns the stored record; replaying different +// values for a recorded round is a conflict. +func (s *Store) RecordCIRepair( + ctx context.Context, attemptID string, input protocol.CIRepairRecordRequest, +) (protocol.CIRepairRecord, error) { + var record protocol.CIRepairRecord + if !publishRefPattern.MatchString(input.HeadAfter) { + return record, invalid("invalid_publish_ref", "head_after must be a commit SHA") + } + if input.HeadAfter == input.HeadBefore { + return record, invalid("invalid_ci_repair_head", "a round must push a new head") + } + if len(input.FailedChecks) > protocol.MaxCIRepairFailedChecks { + return record, invalid("invalid_ci_repair_checks", fmt.Sprintf( + "a round records at most %d failed check names", protocol.MaxCIRepairFailedChecks)) + } + for _, name := range input.FailedChecks { + if len(name) > protocol.MaxCIRepairCheckNameBytes { + return record, invalid("invalid_ci_repair_checks", "a failed check name exceeds its storage limit") + } + } + checks := input.FailedChecks + if checks == nil { + checks = []string{} + } + encodedChecks, err := json.Marshal(checks) + if err != nil { + return record, invalid("invalid_ci_repair_checks", err.Error()) + } + now := s.now().UnixMilli() + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return record, unavailable(err) + } + defer tx.Rollback() + value, err := loadPublishContext(ctx, tx, attemptID) + if err != nil { + return record, err + } + if err := validateCIRepairRequest(input.LeaseToken, input.Round, input.Branch, + input.HeadBefore, value.expectedBranch); err != nil { + return record, err + } + if err := verifyActiveLease(value.lease, input.LeaseToken, now); err != nil { + return record, err + } + budget, err := ciRepairBudget(ctx, tx, attemptID) + if err != nil { + return record, err + } + existing, exists, err := readCIRepair(ctx, tx, attemptID, input.Round) + if err != nil { + return record, err + } + if exists { + existingChecks, _ := json.Marshal(existing.FailedChecks) + if existing.HeadBefore != input.HeadBefore || existing.HeadAfter != input.HeadAfter || + string(existingChecks) != string(encodedChecks) { + return record, conflict("ci_repair_conflict", fmt.Sprintf( + "CI repair round %d is already recorded with a different result", input.Round)) + } + if err := tx.Commit(); err != nil { + return record, unavailable(err) + } + return existing, nil + } + if err := requireNextCIRepairRound(ctx, tx, attemptID, input.Round, budget, input.HeadBefore); err != nil { + return record, err + } + if _, err := tx.ExecContext(ctx, ` + INSERT INTO publish_ci_repairs(attempt_id, round, branch, head_before, head_after, failed_checks, completed_at) + VALUES (?, ?, ?, ?, ?, ?, ?) + `, attemptID, input.Round, value.expectedBranch, input.HeadBefore, input.HeadAfter, + string(encodedChecks), now); err != nil { + if isConstraintViolation(err) { + return record, conflict("ci_repair_conflict", "a concurrent publish already recorded this round") + } + return record, unavailable(err) + } + written, _, err := readCIRepair(ctx, tx, attemptID, input.Round) + if err != nil { + return record, err + } + if err := tx.Commit(); err != nil { + return record, unavailable(err) + } + return written, nil +} + +// AttemptCIRepairs lists one attempt's recorded repair rounds in order. +func (s *Store) AttemptCIRepairs(ctx context.Context, attemptID string) ([]protocol.CIRepairRecord, error) { + var exists int + if err := s.db.QueryRowContext(ctx, + `SELECT COUNT(*) FROM attempts WHERE id = ?`, attemptID).Scan(&exists); err != nil { + return nil, unavailable(err) + } + if exists == 0 { + return nil, ErrNotFound + } + rows, err := s.db.QueryContext(ctx, ` + SELECT r.attempt_id, a.job_id, a.attempt_number, r.round, r.branch, + r.head_before, r.head_after, r.failed_checks, r.completed_at + FROM publish_ci_repairs r JOIN attempts a ON a.id = r.attempt_id + WHERE r.attempt_id = ? ORDER BY r.round + `, attemptID) + if err != nil { + return nil, unavailable(err) + } + defer rows.Close() + records := []protocol.CIRepairRecord{} + for rows.Next() { + var record protocol.CIRepairRecord + var checks string + var completedAt int64 + if err := rows.Scan(&record.AttemptID, &record.JobID, &record.AttemptNumber, &record.Round, + &record.Branch, &record.HeadBefore, &record.HeadAfter, &checks, &completedAt); err != nil { + return nil, unavailable(err) + } + if err := json.Unmarshal([]byte(checks), &record.FailedChecks); err != nil { + return nil, unavailable(err) + } + record.CompletedAt = fromMillis(completedAt) + records = append(records, record) + } + if err := rows.Err(); err != nil { + return nil, unavailable(err) + } + return records, nil +} + +// refuseGreenOnRepairedHead is R6 at the moment a ci record is written: a +// green run may not name a head any repair round found red. It does not +// require the last pushed head, because a person may push a fix after the +// rounds run out and the publish-only retry must still be able to accept it. +func refuseGreenOnRepairedHead(ctx context.Context, tx *sql.Tx, attemptID, head string) error { + var repaired int + if err := tx.QueryRowContext(ctx, ` + SELECT COUNT(*) FROM publish_ci_repairs WHERE attempt_id = ? AND head_before = ? + `, attemptID, head).Scan(&repaired); err != nil { + return unavailable(err) + } + if repaired > 0 { + return conflict("publish_ci_on_repaired_head", fmt.Sprintf( + "CI cannot be recorded green on %s: a repair round already found that head red", head)) + } + return nil +} + +// requireCIGreenPastRepairs is CompleteAttempt's backstop for R6. The ci +// record is refused on a repaired head when it is written, so this can only +// fire on a ledger written before that rule existed or edited by hand; it +// keeps `accepted` honest either way. Ordering needs no check: the ledger +// refuses a round once ci is recorded, so a ci record always postdates +// every round. +func requireCIGreenPastRepairs(ctx context.Context, tx *sql.Tx, attemptID string) error { + var repaired int + if err := tx.QueryRowContext(ctx, ` + SELECT COUNT(*) FROM publish_ci_repairs r + JOIN publish_records p ON p.attempt_id = r.attempt_id AND p.step = ? + WHERE r.attempt_id = ? AND p.remote_ref = r.head_before + `, protocol.PublishStepCI, attemptID).Scan(&repaired); err != nil { + return unavailable(err) + } + if repaired > 0 { + return conflict("publish_ci_on_repaired_head", + "the recorded green CI names a head a repair round found red") + } + return nil +} + +// ---- HTTP surface ---------------------------------------------------------- + +func (a *API) registerCIRepairRoutes(mux *http.ServeMux) { + mux.HandleFunc("POST /api/attempts/{attempt_id}/publish/ci-repair/authorize", a.authorizeCIRepair) + mux.HandleFunc("POST /api/attempts/{attempt_id}/publish/ci-repair/record", a.recordCIRepair) + mux.HandleFunc("GET /api/attempts/{attempt_id}/publish/ci-repairs", a.attemptCIRepairs) +} + +func (a *API) authorizeCIRepair(w http.ResponseWriter, r *http.Request) { + if !a.prepareMutation(w, r) { + return + } + var input protocol.CIRepairAuthorizationRequest + if !decodeJSON(w, r, &input) { + return + } + authorization, err := a.store.AuthorizeCIRepair(r.Context(), r.PathValue("attempt_id"), input) + if err != nil { + writeError(w, err) + return + } + writeJSON(w, http.StatusOK, authorization) +} + +func (a *API) recordCIRepair(w http.ResponseWriter, r *http.Request) { + if !a.prepareMutation(w, r) { + return + } + var input protocol.CIRepairRecordRequest + if !decodeJSON(w, r, &input) { + return + } + record, err := a.store.RecordCIRepair(r.Context(), r.PathValue("attempt_id"), input) + if err != nil { + writeError(w, err) + return + } + writeJSON(w, http.StatusOK, record) +} + +func (a *API) attemptCIRepairs(w http.ResponseWriter, r *http.Request) { + records, err := a.store.AttemptCIRepairs(r.Context(), r.PathValue("attempt_id")) + if err != nil { + writeError(w, err) + return + } + writeJSON(w, http.StatusOK, records) +} diff --git a/internal/controlplane/ci_repair_ledger_test.go b/internal/controlplane/ci_repair_ledger_test.go new file mode 100644 index 0000000..5c460f6 --- /dev/null +++ b/internal/controlplane/ci_repair_ledger_test.go @@ -0,0 +1,329 @@ +// ci_repair_ledger_test.go — the control-plane half of CI repair (U2): the +// round fence, the chain from proof through every pushed fix, the frozen +// budget, and the rule that `accepted` never rests on a head a round found +// red. +package controlplane + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/StructuPath/jig/internal/protocol" +) + +const ( + shaC = "3333333333333333333333333333333333333333" + shaD = "4444444444444444444444444444444444444444" + shaE = "5555555555555555555555555555555555555555" +) + +// repairFixtureSnapshot waits for CI and repairs it: two rounds, the builder +// fixing and the reviewer after it judging. +const repairFixtureSnapshot = `name: fixture +roster: + builder: + model: claude-sonnet + system_prompt: build + user_prompt: build it + reviewer: + model: claude-sonnet + system_prompt: review + user_prompt: review it +phases: + - name: build + kind: agent + owner: builder + - name: review + kind: agent + owner: reviewer +publish: + ci: + wait: true + on_fail: {run: build, budget: 2} +` + +func useSnapshot(t *testing.T, store *Store, runID, snapshot string) { + t.Helper() + if _, err := store.db.Exec(`UPDATE runs SET snapshot = ? WHERE id = ?`, snapshot, runID); err != nil { + t.Fatalf("set run %s snapshot: %v", runID, err) + } +} + +// claimRepairable seeds a run under the given snapshot and claims it. +func claimRepairable(t *testing.T, snapshot string) (*Store, *testClock, *protocol.Claim) { + t.Helper() + store, clock := newTestStore(t) + registerTestWorker(t, store, 1) + seedRun(t, store, "run-1", protocol.RunTarget{Repository: repoA, BaseSHA: shaA}) + useSnapshot(t, store, "run-1", snapshot) + return store, clock, claimAndStart(t, store, "run-1", repoA) +} + +// publishThroughProof records push, pull request, and proof on proofRef. +func publishThroughProof(t *testing.T, store *Store, claim *protocol.Claim, proofRef string) { + t.Helper() + for _, step := range []struct{ name, ref, url string }{ + {protocol.PublishStepPush, proofRef, ""}, + {protocol.PublishStepPullRequest, "", "https://github.com/example/repo-a/pull/7"}, + {protocol.PublishStepProof, proofRef, ""}, + } { + if _, err := recordStep(store, claim, tokenA, step.name, step.ref, step.url); err != nil { + t.Fatalf("record %s: %v", step.name, err) + } + } +} + +func authorizeRound(store *Store, claim *protocol.Claim, token string, round int, before string) (protocol.CIRepairAuthorization, error) { + return store.AuthorizeCIRepair(context.Background(), claim.Attempt.ID, protocol.CIRepairAuthorizationRequest{ + LeaseToken: token, Round: round, Branch: publishBranchOf(claim), HeadBefore: before, + }) +} + +func recordRound(store *Store, claim *protocol.Claim, token string, round int, before, after string, checks ...string) (protocol.CIRepairRecord, error) { + return store.RecordCIRepair(context.Background(), claim.Attempt.ID, protocol.CIRepairRecordRequest{ + LeaseToken: token, Round: round, Branch: publishBranchOf(claim), + HeadBefore: before, HeadAfter: after, FailedChecks: checks, + }) +} + +func wantCode(t *testing.T, err error, code, what string) { + t.Helper() + if err == nil || serviceCode(t, err) != code { + t.Fatalf("%s: err=%v, want %s", what, err, code) + } +} + +// Rounds chain from the proof ref through every pushed fix, in order, within +// the budget the run froze — and each is authorized before its push and +// idempotent after it, exactly like a publish step. +func TestCIRepairRoundsChainFromProofWithinTheFrozenBudget(t *testing.T) { + store, _, claim := claimRepairable(t, repairFixtureSnapshot) + + _, err := authorizeRound(store, claim, tokenA, 1, shaB) + wantCode(t, err, "publish_step_out_of_order", "round before proof") + + publishThroughProof(t, store, claim, shaB) + + _, err = authorizeRound(store, claim, tokenA, 2, shaB) + wantCode(t, err, "ci_repair_out_of_order", "round 2 before round 1") + _, err = authorizeRound(store, claim, tokenA, 1, shaA) + wantCode(t, err, "ci_repair_head_mismatch", "round 1 over a head jig never pushed") + + granted, err := authorizeRound(store, claim, tokenA, 1, shaB) + if err != nil { + t.Fatalf("authorize round 1: %v", err) + } + if granted.Budget != 2 || granted.Completed != nil || granted.Branch != publishBranchOf(claim) { + t.Fatalf("authorization = %+v, want budget 2, not completed, the attempt's branch", granted) + } + first, err := recordRound(store, claim, tokenA, 1, shaB, shaC, "lint", "test (ubuntu)") + if err != nil { + t.Fatalf("record round 1: %v", err) + } + if first.Round != 1 || first.HeadBefore != shaB || first.HeadAfter != shaC || + len(first.FailedChecks) != 2 || first.FailedChecks[1] != "test (ubuntu)" { + t.Fatalf("round 1 = %+v", first) + } + + if replay, err := recordRound(store, claim, tokenA, 1, shaB, shaC, "lint", "test (ubuntu)"); err != nil || + replay.HeadAfter != shaC { + t.Fatalf("identical replay: record=%+v err=%v, want the stored round", replay, err) + } + if again, err := authorizeRound(store, claim, tokenA, 1, shaB); err != nil || + again.Completed == nil || again.Completed.HeadAfter != shaC { + t.Fatalf("re-authorize a recorded round: %+v err=%v, want Completed", again, err) + } + _, err = recordRound(store, claim, tokenA, 1, shaB, shaD, "lint", "test (ubuntu)") + wantCode(t, err, "ci_repair_conflict", "round 1 replayed with a different head") + _, err = recordRound(store, claim, tokenA, 1, shaB, shaC, "lint") + wantCode(t, err, "ci_repair_conflict", "round 1 replayed with different checks") + _, err = recordRound(store, claim, tokenA, 1, shaA, shaC, "lint", "test (ubuntu)") + wantCode(t, err, "ci_repair_conflict", "round 1 replayed over a different head_before") + + _, err = authorizeRound(store, claim, tokenA, 2, shaB) + wantCode(t, err, "ci_repair_head_mismatch", "round 2 over round 1's red head") + if _, err := authorizeRound(store, claim, tokenA, 2, shaC); err != nil { + t.Fatalf("authorize round 2: %v", err) + } + if _, err := recordRound(store, claim, tokenA, 2, shaC, shaD, "lint"); err != nil { + t.Fatalf("record round 2: %v", err) + } + + _, err = authorizeRound(store, claim, tokenA, 3, shaD) + wantCode(t, err, "ci_repair_budget_exhausted", "round 3 on a budget of 2") + _, err = recordRound(store, claim, tokenA, 3, shaD, shaE) + wantCode(t, err, "ci_repair_budget_exhausted", "recording round 3 without authorization") + + rounds, err := store.AttemptCIRepairs(context.Background(), claim.Attempt.ID) + if err != nil { + t.Fatalf("list rounds: %v", err) + } + if len(rounds) != 2 || rounds[0].HeadAfter != shaC || rounds[1].HeadBefore != shaC || + rounds[1].HeadAfter != shaD || len(rounds[1].FailedChecks) != 1 { + t.Fatalf("rounds = %+v, want the two chained rounds", rounds) + } +} + +// Only a definition that declares on_fail has rounds; one that merely waits +// for CI, or does not wait at all, is refused before anything is pushed. +func TestCIRepairIsRefusedWhenTheDefinitionDeclaresNoRepair(t *testing.T) { + for name, snapshot := range map[string]string{ + "waits without on_fail": ciFixtureSnapshot, + "does not wait": fixtureSnapshot, + } { + t.Run(name, func(t *testing.T) { + store, _, claim := claimRepairable(t, snapshot) + publishThroughProof(t, store, claim, shaB) + _, err := authorizeRound(store, claim, tokenA, 1, shaB) + wantCode(t, err, "ci_repair_not_declared", "authorize") + _, err = recordRound(store, claim, tokenA, 1, shaB, shaC) + wantCode(t, err, "ci_repair_not_declared", "record") + }) + } +} + +// A round is a fenced write: a foreign token or an expired lease is refused +// at both halves, and nothing is recorded. +func TestCIRepairRoundsValidateTheLeaseToken(t *testing.T) { + store, clock, claim := claimRepairable(t, repairFixtureSnapshot) + publishThroughProof(t, store, claim, shaB) + + _, err := authorizeRound(store, claim, tokenB, 1, shaB) + wantCode(t, err, "lease_not_owner", "authorize with a foreign token") + _, err = recordRound(store, claim, tokenB, 1, shaB, shaC) + wantCode(t, err, "lease_not_owner", "record with a foreign token") + + clock.Advance(protocol.LeaseDuration + time.Second) + _, err = authorizeRound(store, claim, tokenA, 1, shaB) + wantCode(t, err, "lease_not_owner", "authorize on an expired lease") + _, err = recordRound(store, claim, tokenA, 1, shaB, shaC) + wantCode(t, err, "lease_not_owner", "record on an expired lease") + + rounds, err := store.AttemptCIRepairs(context.Background(), claim.Attempt.ID) + if err != nil || len(rounds) != 0 { + t.Fatalf("rounds = %+v err=%v, want none from a fenced-out attempt", rounds, err) + } +} + +// Once CI is recorded green there is nothing to repair, which is also what +// guarantees a ci record always postdates every round. +func TestCIRepairIsRefusedOnceCIIsGreen(t *testing.T) { + store, _, claim := claimRepairable(t, repairFixtureSnapshot) + publishThroughProof(t, store, claim, shaB) + if _, err := recordStep(store, claim, tokenA, protocol.PublishStepCI, shaB, ""); err != nil { + t.Fatalf("record ci: %v", err) + } + _, err := authorizeRound(store, claim, tokenA, 1, shaB) + wantCode(t, err, "ci_repair_after_green", "authorize after green") + _, err = recordRound(store, claim, tokenA, 1, shaB, shaC) + wantCode(t, err, "ci_repair_after_green", "record after green") +} + +// Malformed rounds are refused before any transaction. +func TestCIRepairRecordsAreValidated(t *testing.T) { + store, _, claim := claimRepairable(t, repairFixtureSnapshot) + publishThroughProof(t, store, claim, shaB) + + _, err := recordRound(store, claim, tokenA, 1, shaB, shaB) + wantCode(t, err, "invalid_ci_repair_head", "a round that pushes no new head") + _, err = recordRound(store, claim, tokenA, 1, shaB, "fixed") + wantCode(t, err, "invalid_publish_ref", "a non-SHA head_after") + _, err = authorizeRound(store, claim, tokenA, 1, "main") + wantCode(t, err, "invalid_publish_ref", "a non-SHA head_before") + _, err = authorizeRound(store, claim, tokenA, 0, shaB) + wantCode(t, err, "invalid_ci_repair_round", "round 0") + + tooMany := make([]string, protocol.MaxCIRepairFailedChecks+1) + for i := range tooMany { + tooMany[i] = fmt.Sprintf("check-%d", i) + } + _, err = recordRound(store, claim, tokenA, 1, shaB, shaC, tooMany...) + wantCode(t, err, "invalid_ci_repair_checks", "too many failed checks") + _, err = recordRound(store, claim, tokenA, 1, shaB, shaC, string(make([]byte, protocol.MaxCIRepairCheckNameBytes+1))) + wantCode(t, err, "invalid_ci_repair_checks", "an oversized check name") + + _, err = store.AuthorizeCIRepair(context.Background(), claim.Attempt.ID, protocol.CIRepairAuthorizationRequest{ + LeaseToken: tokenA, Round: 1, Branch: "jig/someone-else/1", HeadBefore: shaB, + }) + wantCode(t, err, "publish_branch_mismatch", "another attempt's branch") +} + +// After rounds, CI may be recorded green only on a head no round found red +// (R6), and that is refused when the ci row is written: the row is +// permanent, so a bad one would leave the job unable ever to be accepted. +// The green head need not be the last pushed one: a person's later fix is +// acceptable. +func TestGreenCIOnAHeadARoundFoundRedIsRefusedWhenRecorded(t *testing.T) { + for _, tc := range []struct { + name, greenHead, wantCode string + }{ + {"green on the head round 1 repaired", shaB, "publish_ci_on_repaired_head"}, + {"green on the head round 2 repaired", shaC, "publish_ci_on_repaired_head"}, + {"green on the last pushed fix", shaD, ""}, + {"green on a person's later fix", shaE, ""}, + } { + t.Run(tc.name, func(t *testing.T) { + store, _, claim := claimRepairable(t, repairFixtureSnapshot) + publishThroughProof(t, store, claim, shaB) + if _, err := recordRound(store, claim, tokenA, 1, shaB, shaC, "lint"); err != nil { + t.Fatalf("record round 1: %v", err) + } + if _, err := recordRound(store, claim, tokenA, 2, shaC, shaD, "lint"); err != nil { + t.Fatalf("record round 2: %v", err) + } + _, err := recordStep(store, claim, tokenA, protocol.PublishStepCI, tc.greenHead, "") + if tc.wantCode != "" { + wantCode(t, err, tc.wantCode, "record ci") + return + } + if err != nil { + t.Fatalf("record ci: %v", err) + } + attempt, err := store.CompleteAttempt(context.Background(), claim.Attempt.ID, + protocol.CompleteAttemptRequest{LeaseToken: tokenA, State: protocol.AttemptAccepted}) + if err != nil || attempt.State != protocol.AttemptAccepted { + t.Fatalf("complete accepted: attempt=%+v err=%v", attempt, err) + } + }) + } +} + +// CompleteAttempt keeps R6 as a backstop for a ledger that holds a green +// record on a repaired head anyway (written before the record-time rule, or +// by hand): `accepted` is still refused. +func TestAcceptedIsRefusedOnALedgerWithGreenOnARepairedHead(t *testing.T) { + store, clock, claim := claimRepairable(t, repairFixtureSnapshot) + publishThroughProof(t, store, claim, shaB) + if _, err := recordRound(store, claim, tokenA, 1, shaB, shaC, "lint"); err != nil { + t.Fatalf("record round 1: %v", err) + } + // Two rounds, and the planted record names the FIRST round's red head: + // the backstop must check every round, not only the last. + if _, err := recordRound(store, claim, tokenA, 2, shaC, shaD, "lint"); err != nil { + t.Fatalf("record round 2: %v", err) + } + if _, err := store.db.Exec(` + INSERT INTO publish_records(attempt_id, step, branch, remote_ref, pr_url, completed_at) + VALUES (?, ?, ?, ?, '', ?) + `, claim.Attempt.ID, protocol.PublishStepCI, publishBranchOf(claim), shaB, clock.Now().UnixMilli()); err != nil { + t.Fatalf("plant a green record on the repaired head: %v", err) + } + _, err := store.CompleteAttempt(context.Background(), claim.Attempt.ID, + protocol.CompleteAttemptRequest{LeaseToken: tokenA, State: protocol.AttemptAccepted}) + wantCode(t, err, "publish_ci_on_repaired_head", "complete accepted") +} + +// Without rounds the CI gate is unchanged: green on the proof head accepts. +func TestAcceptedWithoutRepairRoundsIsUnchanged(t *testing.T) { + store, _, claim := claimRepairable(t, repairFixtureSnapshot) + publishThroughProof(t, store, claim, shaB) + if _, err := recordStep(store, claim, tokenA, protocol.PublishStepCI, shaB, ""); err != nil { + t.Fatalf("record ci: %v", err) + } + if _, err := store.CompleteAttempt(context.Background(), claim.Attempt.ID, + protocol.CompleteAttemptRequest{LeaseToken: tokenA, State: protocol.AttemptAccepted}); err != nil { + t.Fatalf("complete accepted with no rounds: %v", err) + } +} diff --git a/internal/controlplane/migrate_test.go b/internal/controlplane/migrate_test.go index b9d5841..7b4c886 100644 --- a/internal/controlplane/migrate_test.go +++ b/internal/controlplane/migrate_test.go @@ -82,6 +82,7 @@ func TestMigratingAnEmptyDatabaseAppliesAllMigrationsAndRecordsTheLedger(t *test for _, table := range []string{ "definitions", "runs", "jobs", "attempts", "claim_requests", "events", "envelopes", "gate_results", "workers", "retained_worktrees", "publish_records", + "publish_ci_repairs", } { var name string err := db.QueryRow( diff --git a/internal/controlplane/publish_ledger.go b/internal/controlplane/publish_ledger.go index a06650e..2acb1b6 100644 --- a/internal/controlplane/publish_ledger.go +++ b/internal/controlplane/publish_ledger.go @@ -239,6 +239,14 @@ func (s *Store) RecordPublishStep( if err := requirePublishPrerequisites(ctx, tx, attemptID, input.Step); err != nil { return record, err } + if input.Step == protocol.PublishStepCI { + // A ci row is permanent, so a green record on a head a repair round + // found red must be refused here: accepted would be refused on it + // forever, and no further round is allowed once ci exists. + if err := refuseGreenOnRepairedHead(ctx, tx, attemptID, input.RemoteRef); err != nil { + return record, err + } + } if _, err := tx.ExecContext(ctx, ` INSERT INTO publish_records(attempt_id, step, branch, remote_ref, pr_url, completed_at) VALUES (?, ?, ?, ?, ?, ?) @@ -445,6 +453,7 @@ func (a *API) registerPublishRoutes(mux *http.ServeMux) { mux.HandleFunc("GET /api/attempts/{attempt_id}/publish", a.attemptPublishRecords) mux.HandleFunc("GET /api/jobs/{job_id}/publish", a.jobPublishRecords) mux.HandleFunc("POST /api/jobs/{job_id}/publish-retry", a.retryPublish) + a.registerCIRepairRoutes(mux) } func (a *API) authorizePublishStep(w http.ResponseWriter, r *http.Request) { diff --git a/internal/controlplane/store.go b/internal/controlplane/store.go index 0341740..8f8822e 100644 --- a/internal/controlplane/store.go +++ b/internal/controlplane/store.go @@ -195,11 +195,11 @@ type leaseState struct { cancel bool } -// attemptWaitsForCI reads the attempt's run snapshot — the definition frozen -// at admission (R2) — and reports whether its publish.ci waits. A snapshot -// that no longer parses cannot have been executed, so it is an error, not a -// silent "no". -func attemptWaitsForCI(ctx context.Context, tx *sql.Tx, attemptID string) (bool, error) { +// attemptDefinition reads the attempt's run snapshot — the definition frozen +// at admission (R2) — so publish rules come from what the attempt actually +// ran under, never from the worker's word. A snapshot that no longer parses +// cannot have been executed, so it is an error, not a silent "no". +func attemptDefinition(ctx context.Context, tx *sql.Tx, attemptID string) (*protocol.DefinitionSpec, error) { var snapshot string if err := tx.QueryRowContext(ctx, ` SELECT r.snapshot FROM attempts a @@ -207,14 +207,14 @@ func attemptWaitsForCI(ctx context.Context, tx *sql.Tx, attemptID string) (bool, JOIN runs r ON r.id = j.run_id WHERE a.id = ? `, attemptID).Scan(&snapshot); err != nil { - return false, unavailable(err) + return nil, unavailable(err) } spec, err := protocol.ParseDefinition([]byte(snapshot)) if err != nil { - return false, conflict("snapshot_unparseable", + return nil, conflict("snapshot_unparseable", "the run's frozen definition no longer parses: "+err.Error()) } - return spec.WaitsForCI(), nil + return spec, nil } func loadLease(ctx context.Context, tx *sql.Tx, attemptID string) (leaseState, error) { @@ -552,11 +552,11 @@ func (s *Store) CompleteAttempt(ctx context.Context, attemptID string, input pro // The same rule for CI, read from the run's frozen definition rather // than from anything the worker says: a definition whose publish.ci // waits is not `accepted` until a green CI run is on the ledger. - waits, err := attemptWaitsForCI(ctx, tx, attemptID) + spec, err := attemptDefinition(ctx, tx, attemptID) if err != nil { return protocol.Attempt{}, err } - if waits { + if spec.WaitsForCI() { var green int if err := tx.QueryRowContext(ctx, ` SELECT COUNT(*) FROM publish_records WHERE attempt_id = ? AND step = ? @@ -568,6 +568,9 @@ func (s *Store) CompleteAttempt(ctx context.Context, attemptID string, input pro "this definition waits for CI, so the attempt is `accepted` only with a "+ "recorded ci step; complete as accepted_unpublished while CI is not green") } + if err := requireCIGreenPastRepairs(ctx, tx, attemptID); err != nil { + return protocol.Attempt{}, err + } } } if _, err := tx.ExecContext(ctx, ` diff --git a/internal/protocol/publish.go b/internal/protocol/publish.go index 5f56d7c..085330b 100644 --- a/internal/protocol/publish.go +++ b/internal/protocol/publish.go @@ -192,3 +192,64 @@ type PublishRetry struct { // the same publish policy (a CI wait, say) the original attempt ran under. Snapshot string `json:"snapshot"` } + +// MaxCIRepairFailedChecks caps how many failed check names one CI repair +// round records, and MaxCIRepairCheckNameBytes caps each name. The round +// record is evidence of what was red, not the log itself. +const ( + MaxCIRepairFailedChecks = 20 + MaxCIRepairCheckNameBytes = 200 +) + +// CIRepairRecord is one recorded CI repair round (publish.ci.on_fail): CI +// was red on HeadBefore, a repair ran the chain again, and HeadAfter is the +// fix it pushed to the same branch. Rounds chain: round 1's HeadBefore is +// the proof step's remote_ref, and every later round's HeadBefore is the +// previous round's HeadAfter. It is a record of its own rather than another +// publish step because a second push is a new fact, not a replay of the +// first. +type CIRepairRecord struct { + AttemptID string `json:"attempt_id"` + JobID string `json:"job_id"` + AttemptNumber int `json:"attempt_number"` + Round int `json:"round"` + Branch string `json:"branch"` + HeadBefore string `json:"head_before"` + HeadAfter string `json:"head_after"` + FailedChecks []string `json:"failed_checks"` + CompletedAt time.Time `json:"completed_at"` +} + +// CIRepairAuthorizationRequest asks whether this lease may push repair +// round Round over HeadBefore right now. Like a publish step, it is asked +// BEFORE the push, so a zombie or an over-budget round is refused before +// it touches the remote. +type CIRepairAuthorizationRequest struct { + LeaseToken string `json:"lease_token"` + Round int `json:"round"` + Branch string `json:"branch"` + HeadBefore string `json:"head_before"` +} + +// CIRepairAuthorization is a granted round. Budget is the frozen +// definition's on_fail budget. Completed is non-nil when the round is +// already recorded, and the worker must then skip the push. +type CIRepairAuthorization struct { + AttemptID string `json:"attempt_id"` + Round int `json:"round"` + Budget int `json:"budget"` + Branch string `json:"branch"` + Completed *CIRepairRecord `json:"completed,omitempty"` +} + +// CIRepairRecordRequest records one pushed repair round under the lease +// token. Replaying identical values returns the stored record; different +// values for a recorded round are a conflict. +type CIRepairRecordRequest struct { + LeaseToken string `json:"lease_token"` + Round int `json:"round"` + Branch string `json:"branch"` + HeadBefore string `json:"head_before"` + HeadAfter string `json:"head_after"` + FailedChecks []string `json:"failed_checks"` +} diff --git a/internal/worker/publish.go b/internal/worker/publish.go index e89cb9b..99899ed 100644 --- a/internal/worker/publish.go +++ b/internal/worker/publish.go @@ -1646,6 +1646,35 @@ func (c *Client) JobPublishRecords(ctx context.Context, jobID string) ([]protoco return records, err } +// AuthorizeCIRepair asks whether this lease may push one CI repair round +// (fenced, before the push). +func (c *Client) AuthorizeCIRepair( + ctx context.Context, attemptID string, request protocol.CIRepairAuthorizationRequest, +) (protocol.CIRepairAuthorization, error) { + var authorization protocol.CIRepairAuthorization + err := c.call(ctx, http.MethodPost, + "/api/attempts/"+url.PathEscape(attemptID)+"/publish/ci-repair/authorize", request, &authorization) + return authorization, err +} + +// RecordCIRepair records one pushed CI repair round (fenced, after the push). +func (c *Client) RecordCIRepair( + ctx context.Context, attemptID string, request protocol.CIRepairRecordRequest, +) (protocol.CIRepairRecord, error) { + var record protocol.CIRepairRecord + err := c.call(ctx, http.MethodPost, + "/api/attempts/"+url.PathEscape(attemptID)+"/publish/ci-repair/record", request, &record) + return record, err +} + +// AttemptCIRepairs reads one attempt's recorded CI repair rounds. +func (c *Client) AttemptCIRepairs(ctx context.Context, attemptID string) ([]protocol.CIRepairRecord, error) { + var records []protocol.CIRepairRecord + err := c.call(ctx, http.MethodGet, + "/api/attempts/"+url.PathEscape(attemptID)+"/publish/ci-repairs", nil, &records) + return records, err +} + // RetryPublish requests the publish-only retry of one accepted_unpublished // job (R14). func (c *Client) RetryPublish( diff --git a/internal/worker/publish_ci_test.go b/internal/worker/publish_ci_test.go index 3c68eb8..fb26949 100644 --- a/internal/worker/publish_ci_test.go +++ b/internal/worker/publish_ci_test.go @@ -375,3 +375,85 @@ func TestGitHubGatewayNormalizesCheckRunsAndCommitStatuses(t *testing.T) { t.Fatal("an invalid ref reached gh") } } + +// Heads for the round wire test; the ledger checks their shape, not git. +const ( + headOne = "1111111111111111111111111111111111111111" + headTwo = "2222222222222222222222222222222222222222" +) + +// repairSnapshot declares a CI repair, so the round routes accept it. +const repairSnapshot = integrationSnapshot + ` - name: review + kind: agent + owner: builder +publish: + ci: + wait: true + on_fail: {run: build, budget: 1} +` + +// The worker reaches the round ledger through its client over the real +// control-plane handler: authorize, record, replay as Completed, and list. +// The ledger's rules are the control plane's tests; this proves the wire. +func TestCIRepairRoundsTravelTheClientAndHTTPSurface(t *testing.T) { + h := newHarness(t) + const identity = "github.com/example/repair" + h.seedRunWithSnapshot("run-1", repairSnapshot, protocol.RunTarget{Repository: identity, BaseSHA: headOne}) + h.enqueue("run-1", identity) + w := newTestWorker(t, h, filepath.Join(t.TempDir(), "worker"), 1, nil) + ctx := context.Background() + + token, err := mintLeaseToken() + if err != nil { + t.Fatalf("mint lease token: %v", err) + } + claim, err := w.client.Claim(ctx, protocol.ClaimRequest{RequestID: "repair-wire", LeaseToken: token}) + if err != nil || claim == nil { + t.Fatalf("claim: claim=%v err=%v", claim, err) + } + if _, err := w.client.StartAttempt(ctx, claim.Attempt.ID, protocol.StartAttemptRequest{LeaseToken: token}); err != nil { + t.Fatalf("start attempt: %v", err) + } + branch := protocol.PublishBranch(claim.Job.ID, claim.Attempt.AttemptNumber) + for _, step := range []protocol.PublishStepRequest{ + {Step: protocol.PublishStepPush, RemoteRef: headOne}, + {Step: protocol.PublishStepPullRequest, PullRequestURL: "https://github.com/example/repair/pull/1"}, + {Step: protocol.PublishStepProof, RemoteRef: headOne}, + } { + step.LeaseToken, step.Branch = token, branch + if _, err := w.client.RecordPublishStep(ctx, claim.Attempt.ID, step); err != nil { + t.Fatalf("record %s: %v", step.Step, err) + } + } + + authorization, err := w.client.AuthorizeCIRepair(ctx, claim.Attempt.ID, protocol.CIRepairAuthorizationRequest{ + LeaseToken: token, Round: 1, Branch: branch, HeadBefore: headOne, + }) + if err != nil || authorization.Budget != 1 || authorization.Completed != nil { + t.Fatalf("authorize round 1: %+v err=%v", authorization, err) + } + record, err := w.client.RecordCIRepair(ctx, claim.Attempt.ID, protocol.CIRepairRecordRequest{ + LeaseToken: token, Round: 1, Branch: branch, HeadBefore: headOne, HeadAfter: headTwo, + FailedChecks: []string{"lint"}, + }) + if err != nil || record.HeadAfter != headTwo || record.JobID != claim.Job.ID { + t.Fatalf("record round 1: %+v err=%v", record, err) + } + again, err := w.client.AuthorizeCIRepair(ctx, claim.Attempt.ID, protocol.CIRepairAuthorizationRequest{ + LeaseToken: token, Round: 1, Branch: branch, HeadBefore: headOne, + }) + if err != nil || again.Completed == nil || again.Completed.HeadAfter != headTwo { + t.Fatalf("re-authorize a recorded round: %+v err=%v, want Completed", again, err) + } + _, err = w.client.AuthorizeCIRepair(ctx, claim.Attempt.ID, protocol.CIRepairAuthorizationRequest{ + LeaseToken: token, Round: 2, Branch: branch, HeadBefore: headTwo, + }) + var rejected *APIError + if !errors.As(err, &rejected) || rejected.Code != "ci_repair_budget_exhausted" { + t.Fatalf("round 2 on a budget of 1: err=%v, want ci_repair_budget_exhausted over the wire", err) + } + rounds, err := w.client.AttemptCIRepairs(ctx, claim.Attempt.ID) + if err != nil || len(rounds) != 1 || rounds[0].FailedChecks[0] != "lint" { + t.Fatalf("list rounds: %+v err=%v", rounds, err) + } +} diff --git a/migrations/005_publish_ci_repairs.sql b/migrations/005_publish_ci_repairs.sql new file mode 100644 index 0000000..806eb98 --- /dev/null +++ b/migrations/005_publish_ci_repairs.sql @@ -0,0 +1,19 @@ +-- 005_publish_ci_repairs.sql — the ledger of CI repair rounds. +-- +-- A definition with publish.ci.on_fail may push fixes to the same branch +-- after CI goes red. Each pushed fix is one round: the red head it repaired, +-- the head it pushed, and the checks that were red. publish_records keeps +-- one row per step by design, so rounds get their own table. Rounds chain +-- (round N's head_before is round N-1's head_after), which the control +-- plane enforces; the table enforces only what SQL can say cheaply. + +CREATE TABLE publish_ci_repairs ( + attempt_id TEXT NOT NULL REFERENCES attempts(id), + round INTEGER NOT NULL CHECK (round >= 1), + branch TEXT NOT NULL, + head_before TEXT NOT NULL, + head_after TEXT NOT NULL CHECK (head_after <> head_before), + failed_checks TEXT NOT NULL DEFAULT '[]', + completed_at INTEGER NOT NULL, + PRIMARY KEY (attempt_id, round) +);