From be9b79543045251a9975311958abdf903c54504a Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 27 Sep 2026 15:42:48 +0000 Subject: [PATCH] Wait for green CI before a job is accepted (publish.ci) A definition can now opt into `publish: {ci: {wait: true, timeout: 30m}}`. After push, pull request, and proof, publish runs a fourth fenced step: it polls the check runs and commit statuses on the branch's current remote head via `gh api`, and records `ci` on the ledger only when every signal is green. Red fails fast, pending past the timeout fails, a head with no checks passes after a two-minute registration grace, three consecutive read errors end the wait, and a cancel stops it. Every failure ends the job accepted_unpublished with the branch and pull request in place and the red checks named in the result. The control plane enforces the rule itself: CompleteAttempt refuses `accepted` without a ci record when the run's frozen definition waits for CI (publish_ci_required), mirroring the proof requirement. The publish-only retry now carries the run snapshot, so it re-applies the same policy and judges whatever the branch head is by then, which lets a person's fix count. Migration 004 rebuilds publish_records to widen its step CHECK with 'ci'. factory.yaml opts in; definitions without `ci` are unchanged. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LwarQhEaL1ygzyMLvaXvZH --- README.md | 14 + docs/quickstart.md | 6 + examples/definitions/factory.yaml | 9 + internal/controlplane/migrate_test.go | 62 +++ internal/controlplane/publish_ledger.go | 10 +- internal/controlplane/publish_ledger_test.go | 114 ++++++ internal/controlplane/store.go | 42 +++ internal/protocol/definition.go | 61 ++- internal/protocol/definition_test.go | 36 +- internal/protocol/publish.go | 24 +- internal/worker/publish.go | 369 ++++++++++++++++-- internal/worker/publish_ci_test.go | 377 +++++++++++++++++++ internal/worker/publish_test.go | 18 +- internal/worker/worker_integration_test.go | 10 +- migrations/004_publish_ci_step.sql | 24 ++ 15 files changed, 1137 insertions(+), 39 deletions(-) create mode 100644 internal/worker/publish_ci_test.go create mode 100644 migrations/004_publish_ci_step.sql diff --git a/README.md b/README.md index b4ad9ea..c097615 100644 --- a/README.md +++ b/README.md @@ -165,6 +165,7 @@ phases: acceptance: [all_phases_passed, diff_matches_claims] publish: hold_when: "risk == high" # accepted but held for a person; publish retry releases it + ci: {wait: true, timeout: 30m} # accepted only once CI on the PR head is green ``` Rules worth knowing before you write one: @@ -221,6 +222,19 @@ Rules worth knowing before you write one: access to the control plane: jig has one trusted operator and no authentication (see *Loopback only*), so the operator's retry is the approval, with no separate approver identity. +- **Publish can wait for CI.** `publish: {ci: {wait: true, timeout: 30m}}` + adds a fourth publish step after proof: the worker polls the check runs + and commit statuses on the branch's head through `gh api`, and the job is + `accepted` only when every one is green — the control plane enforces this + from the frozen definition, not the worker's word. A red check fails fast; + checks still pending at the timeout (default 30m, 1m–6h) fail too; a head + with no checks at all passes after two minutes, since that repository runs + no CI. Any of these ends the job `accepted_unpublished` with the branch and + pull request in place and the red checks named in the result. Fix the + branch — push to it yourself — and the publish-only retry judges CI on + whatever its head is by then. Neutral and skipped checks count as green. + The wait holds a worker slot, and a cancel during it ends the job the same + way. Omit `ci` and publish does not wait, exactly as before. - **Validation happens before anything runs.** `jig def validate ` is the same check the store applies at save time, offline. diff --git a/docs/quickstart.md b/docs/quickstart.md index 28f9f98..a6f56b7 100644 --- a/docs/quickstart.md +++ b/docs/quickstart.md @@ -193,6 +193,12 @@ purpose: `factory.yaml` scores every change for risk and holds high-risk work with `publish: held` in its result. Read the risk reviewer's briefing in the trace, then the publish retry is your sign-off. +`factory.yaml` also sets `publish.ci.wait`, so a published job is +`accepted` only once CI on its pull request is green. Red CI lands it in +`accepted_unpublished` with the failing checks in its result; push a fix to +the `jig//` branch and the publish retry re-checks CI on the +new head. + `Ctrl-C` stops the worker in order: no new claims, in-flight attempts cancelled, terminal states recorded, ephemeral scratch destroyed. diff --git a/examples/definitions/factory.yaml b/examples/definitions/factory.yaml index d802d36..1a2956c 100644 --- a/examples/definitions/factory.yaml +++ b/examples/definitions/factory.yaml @@ -28,6 +28,12 @@ # `risk == high` so they fail closed: work the classifier never scored — a # missing or unexpected `risk` — is held, not shipped. # +# THE CI GATE. `publish.ci.wait` makes green CI part of `accepted`: after the +# pull request exists, the worker polls its head's checks for up to 30 +# minutes. Red or timed-out CI ends the job accepted_unpublished with the +# failing checks named; push a fix to the branch and the publish retry +# re-judges CI on the new head. +# # THE ROSTER. Builders and reviewers are `opus`; the planner and the extra # high-risk reviewer are `sonnet`. `effort` is medium for building to a plan # and high for reviewers hunting what the build missed. `budget_usd` caps what @@ -352,3 +358,6 @@ phases: acceptance: [all_phases_passed, verdict_consistent, diff_matches_claims] publish: hold_when: "risk != low" + ci: + wait: true + timeout: 30m diff --git a/internal/controlplane/migrate_test.go b/internal/controlplane/migrate_test.go index ee1eb33..b9d5841 100644 --- a/internal/controlplane/migrate_test.go +++ b/internal/controlplane/migrate_test.go @@ -169,3 +169,65 @@ func TestDuplicateAttemptEventSequenceInsertIsIgnored(t *testing.T) { t.Fatalf("replay overwrote the original payload: %q", payload) } } + +// 004 rebuilds publish_records to widen its step vocabulary with `ci`. A +// rebuild is only safe if it carries every existing row across, so this +// applies 001–003 by hand, records a proven step under the old schema, then +// applies 004 and checks the row survived and the new vocabulary holds. +func TestPublishCIStepMigrationPreservesRecordsAndWidensTheStepVocabulary(t *testing.T) { + path := filepath.Join(t.TempDir(), "jig.db") + db, err := sql.Open("sqlite", "file:"+path+"?_pragma=foreign_keys(1)") + if err != nil { + t.Fatalf("open database: %v", err) + } + t.Cleanup(func() { db.Close() }) + apply := func(name string) { + t.Helper() + body, err := migrations.Files.ReadFile(name) + if err != nil { + t.Fatalf("read %s: %v", name, err) + } + if _, err := db.Exec(string(body)); err != nil { + t.Fatalf("apply %s: %v", name, err) + } + } + for _, name := range []string{"001_core.sql", "002_job_cancellation.sql", "003_admission_triggers.sql"} { + apply(name) + } + jobID := seedJob(t, db) + if err := insertAttempt(db, "attempt-1", jobID, 1, "running"); err != nil { + t.Fatalf("insert attempt: %v", err) + } + insertStep := func(step string) error { + _, err := db.Exec(`INSERT INTO publish_records(attempt_id, step, branch, remote_ref, completed_at) + VALUES ('attempt-1', ?, 'jig/job-1/1', '1111111111111111111111111111111111111111', 7)`, step) + return err + } + if err := insertStep("proof"); err != nil { + t.Fatalf("record proof under the 003 schema: %v", err) + } + if err := insertStep("ci"); err == nil { + t.Fatal("the 003 schema accepted a ci step; the migration under test would be a no-op") + } + + apply("004_publish_ci_step.sql") + + var branch, ref string + var completedAt int64 + if err := db.QueryRow(`SELECT branch, remote_ref, completed_at FROM publish_records + WHERE attempt_id = 'attempt-1' AND step = 'proof'`).Scan(&branch, &ref, &completedAt); err != nil { + t.Fatalf("the proof record did not survive the rebuild: %v", err) + } + if branch != "jig/job-1/1" || completedAt != 7 { + t.Fatalf("proof record = %s %s %d after the rebuild, want it unchanged", branch, ref, completedAt) + } + if err := insertStep("ci"); err != nil { + t.Fatalf("record ci after 004: %v", err) + } + if err := insertStep("deploy"); err == nil { + t.Fatal("the rebuilt table accepted a step outside its vocabulary") + } + if err := insertStep("ci"); err == nil { + t.Fatal("the rebuilt table lost its (attempt_id, step) primary key") + } +} diff --git a/internal/controlplane/publish_ledger.go b/internal/controlplane/publish_ledger.go index 82b86b1..a06650e 100644 --- a/internal/controlplane/publish_ledger.go +++ b/internal/controlplane/publish_ledger.go @@ -61,7 +61,7 @@ func loadPublishContext(ctx context.Context, tx *sql.Tx, attemptID string) (publ func validatePublishStep(step, branch, expectedBranch string) error { if !protocol.ValidPublishStep(step) { return invalid("invalid_publish_step", - "step must be one of push, pull_request, proof") + "step must be one of push, pull_request, proof, ci") } if len(branch) > protocol.MaxPublishBranchBytes { return invalid("invalid_publish_branch", "branch exceeds its storage limit") @@ -193,10 +193,10 @@ func (s *Store) RecordPublishStep( return record, invalid("invalid_publish_ref", "remote_ref exceeds its storage limit") } switch input.Step { - case protocol.PublishStepPush, protocol.PublishStepProof: + case protocol.PublishStepPush, protocol.PublishStepProof, protocol.PublishStepCI: if !publishRefPattern.MatchString(input.RemoteRef) { return record, invalid("invalid_publish_ref", - "push and proof records must carry the pushed commit SHA as remote_ref") + "push, proof, and ci records must carry a commit SHA as remote_ref") } case protocol.PublishStepPullRequest: if !strings.HasPrefix(input.PullRequestURL, "https://") { @@ -424,6 +424,10 @@ func (s *Store) RetryPublish( if err != nil { return retry, err } + if err := s.db.QueryRowContext(ctx, + `SELECT snapshot FROM runs WHERE id = ?`, job.RunID).Scan(&retry.Snapshot); err != nil { + return retry, unavailable(err) + } retry.Attempt = attempt retry.Job = job retry.Records = records diff --git a/internal/controlplane/publish_ledger_test.go b/internal/controlplane/publish_ledger_test.go index 16c7f5a..6835fa4 100644 --- a/internal/controlplane/publish_ledger_test.go +++ b/internal/controlplane/publish_ledger_test.go @@ -313,3 +313,117 @@ func TestPublishRetryIsRefusedForJobsThatAreNotAcceptedUnpublished(t *testing.T) t.Fatalf("retry of a failed job: err=%v, want publish_retry_not_allowed", err) } } + +// ciFixtureSnapshot is fixtureSnapshot with publish.ci opted in. +const ciFixtureSnapshot = fixtureSnapshot + `publish: + ci: + wait: true +` + +// optInToCI swaps the run's frozen definition for one whose publish waits +// for CI — the fixture every CI-gate scenario starts from. +func optInToCI(t *testing.T, store *Store, runID string) { + t.Helper() + if _, err := store.db.Exec(`UPDATE runs SET snapshot = ? WHERE id = ?`, ciFixtureSnapshot, runID); err != nil { + t.Fatalf("opt run %s into CI: %v", runID, err) + } +} + +// A definition that waits for CI extends R12's "accepted means published" +// to "accepted means published AND green". Like proof, that is checked by +// the store against its own ledger and the run's frozen definition — never +// taken from the worker's word — and the ci step sits after proof in the +// fenced step order. +func TestAcceptedRequiresARecordedCIStepWhenTheDefinitionWaitsForCI(t *testing.T) { + store, _ := newTestStore(t) + registerTestWorker(t, store, 1) + seedRun(t, store, "run-1", protocol.RunTarget{Repository: repoA, BaseSHA: shaA}) + optInToCI(t, store, "run-1") + claim := claimAndStart(t, store, "run-1", repoA) + ctx := context.Background() + + if _, err := recordStep(store, claim, tokenA, protocol.PublishStepPush, shaB, ""); err != nil { + t.Fatalf("record push: %v", err) + } + if _, err := recordStep(store, claim, tokenA, protocol.PublishStepCI, shaB, ""); err == nil || + serviceCode(t, err) != "publish_step_out_of_order" { + t.Fatalf("ci before proof: err=%v, want publish_step_out_of_order", err) + } + if _, err := recordStep(store, claim, tokenA, protocol.PublishStepPullRequest, "", + "https://github.com/example/repo-a/pull/3"); err != nil { + t.Fatalf("record pull request: %v", err) + } + if _, err := recordStep(store, claim, tokenA, protocol.PublishStepProof, shaB, ""); err != nil { + t.Fatalf("record proof: %v", err) + } + + _, err := store.CompleteAttempt(ctx, claim.Attempt.ID, protocol.CompleteAttemptRequest{ + LeaseToken: tokenA, State: protocol.AttemptAccepted, + }) + if err == nil || serviceCode(t, err) != "publish_ci_required" { + t.Fatalf("accepted with proof but no ci record: err=%v, want publish_ci_required", err) + } + + if _, err := recordStep(store, claim, tokenA, protocol.PublishStepCI, "green", ""); err == nil || + serviceCode(t, err) != "invalid_publish_ref" { + t.Fatalf("ci with a non-SHA ref: err=%v, want invalid_publish_ref", err) + } + // CI may be judged on a later head than jig pushed (a person's fix on the + // branch), so the ci ref need not equal the push ref. + if _, err := recordStep(store, claim, tokenA, protocol.PublishStepCI, shaA, ""); err != nil { + t.Fatalf("record ci: %v", err) + } + accepted, err := store.CompleteAttempt(ctx, claim.Attempt.ID, protocol.CompleteAttemptRequest{ + LeaseToken: tokenA, State: protocol.AttemptAccepted, + }) + if err != nil { + t.Fatalf("complete accepted with a ci record: %v", err) + } + if accepted.State != protocol.AttemptAccepted { + t.Fatalf("attempt state = %q, want accepted", accepted.State) + } +} + +// A definition that does not opt in is unchanged: proof alone is enough. +func TestAcceptedNeedsNoCIStepWhenTheDefinitionDoesNotWait(t *testing.T) { + store, _ := newTestStore(t) + registerTestWorker(t, store, 1) + seedRun(t, store, "run-1", protocol.RunTarget{Repository: repoA, BaseSHA: shaA}) + claim := claimAndStart(t, store, "run-1", repoA) + for _, step := range []struct{ name, ref, url string }{ + {protocol.PublishStepPush, shaB, ""}, + {protocol.PublishStepPullRequest, "", "https://github.com/example/repo-a/pull/4"}, + {protocol.PublishStepProof, shaB, ""}, + } { + if _, err := recordStep(store, claim, tokenA, step.name, step.ref, step.url); err != nil { + t.Fatalf("record %s: %v", step.name, err) + } + } + if _, err := store.CompleteAttempt(context.Background(), claim.Attempt.ID, protocol.CompleteAttemptRequest{ + LeaseToken: tokenA, State: protocol.AttemptAccepted, + }); err != nil { + t.Fatalf("complete accepted without CI opt-in: %v", err) + } +} + +// The publish-only retry hands the worker the run's frozen definition, so a +// retry re-applies the same CI policy the first publish ran under. +func TestPublishRetryCarriesTheRunSnapshot(t *testing.T) { + store, _ := newTestStore(t) + registerTestWorker(t, store, 1) + seedRun(t, store, "run-1", protocol.RunTarget{Repository: repoA, BaseSHA: shaA}) + optInToCI(t, store, "run-1") + claim := claimAndStart(t, store, "run-1", repoA) + if _, err := store.CompleteAttempt(context.Background(), claim.Attempt.ID, + protocol.CompleteAttemptRequest{LeaseToken: tokenA, State: protocol.AttemptAcceptedUnpublished}); err != nil { + t.Fatalf("complete accepted_unpublished: %v", err) + } + retry, err := store.RetryPublish(context.Background(), claim.Job.ID, + protocol.PublishRetryRequest{WorkerID: workerA, LeaseToken: tokenB}) + if err != nil { + t.Fatalf("publish retry: %v", err) + } + if retry.Snapshot != ciFixtureSnapshot { + t.Fatalf("retry snapshot = %q, want the run's frozen definition", retry.Snapshot) + } +} diff --git a/internal/controlplane/store.go b/internal/controlplane/store.go index d26caad..0341740 100644 --- a/internal/controlplane/store.go +++ b/internal/controlplane/store.go @@ -195,6 +195,28 @@ 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) { + var snapshot string + if err := tx.QueryRowContext(ctx, ` + SELECT r.snapshot FROM attempts a + JOIN jobs j ON j.id = a.job_id + JOIN runs r ON r.id = j.run_id + WHERE a.id = ? + `, attemptID).Scan(&snapshot); err != nil { + return false, unavailable(err) + } + spec, err := protocol.ParseDefinition([]byte(snapshot)) + if err != nil { + return false, conflict("snapshot_unparseable", + "the run's frozen definition no longer parses: "+err.Error()) + } + return spec.WaitsForCI(), nil +} + func loadLease(ctx context.Context, tx *sql.Tx, attemptID string) (leaseState, error) { var value leaseState var cancel int @@ -527,6 +549,26 @@ func (s *Store) CompleteAttempt(ctx context.Context, attemptID string, input pro "an attempt is `accepted` only with a recorded proof-of-publish step (R12); "+ "complete as accepted_unpublished when publish did not finish") } + // 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) + if err != nil { + return protocol.Attempt{}, err + } + if waits { + 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 protocol.Attempt{}, unavailable(err) + } + if green == 0 { + return protocol.Attempt{}, conflict("publish_ci_required", + "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 := tx.ExecContext(ctx, ` UPDATE attempts SET state = ?, result = ?, error = ?, completed_at = ? diff --git a/internal/protocol/definition.go b/internal/protocol/definition.go index 0ce426d..09e096c 100644 --- a/internal/protocol/definition.go +++ b/internal/protocol/definition.go @@ -5,6 +5,7 @@ import ( "fmt" "sort" "strings" + "time" "gopkg.in/yaml.v3" ) @@ -72,9 +73,44 @@ type DefinitionSpec struct { // attempt ends accepted_unpublished with a recorded hold instead of a branch // and pull request, and the operator's publish retry is the human review // step that releases it. A change a classifier scored `risk == high` waits -// for a person; everything else ships. +// for a person; everything else ships. CI, when it waits, makes a green CI +// run on the pull request's head part of what `accepted` means. type PublishSpec struct { - HoldWhen string `yaml:"hold_when"` + HoldWhen string `yaml:"hold_when"` + CI *CISpec `yaml:"ci"` +} + +// CISpec opts a definition into waiting for the pull request's CI after +// publish. Timeout is a Go duration ("30m"); empty means DefaultCITimeout. +type CISpec struct { + Wait bool `yaml:"wait"` + Timeout string `yaml:"timeout"` +} + +// CI wait bounds. The timeout covers the whole wait for one head commit; the +// ceiling keeps a typo like "30h" from parking a worker slot for a day. +const ( + DefaultCITimeout = 30 * time.Minute + MinCITimeout = time.Minute + MaxCITimeout = 6 * time.Hour +) + +// WaitsForCI reports whether the definition gates acceptance on CI. +func (spec *DefinitionSpec) WaitsForCI() bool { + return spec.Publish != nil && spec.Publish.CI != nil && spec.Publish.CI.Wait +} + +// CITimeout is the validated CI wait budget, DefaultCITimeout when unset. +// Call it only on a spec that passed Validate. +func (spec *DefinitionSpec) CITimeout() time.Duration { + if !spec.WaitsForCI() || strings.TrimSpace(spec.Publish.CI.Timeout) == "" { + return DefaultCITimeout + } + timeout, err := time.ParseDuration(spec.Publish.CI.Timeout) + if err != nil { + return DefaultCITimeout + } + return timeout } // RoleSpec is one roster entry: the model, prompts, and allowlists an agent @@ -238,11 +274,24 @@ func (spec *DefinitionSpec) validatePublish() error { if spec.Publish == nil { return nil } - if strings.TrimSpace(spec.Publish.HoldWhen) == "" { - return fmt.Errorf("publish: hold_when is required when publish is declared") + hold := strings.TrimSpace(spec.Publish.HoldWhen) + if hold == "" && spec.Publish.CI == nil { + return fmt.Errorf("publish: declare hold_when, ci, or both") } - if _, err := ParsePredicate(spec.Publish.HoldWhen); err != nil { - return fmt.Errorf("publish: hold_when: %w", err) + if hold != "" { + if _, err := ParsePredicate(spec.Publish.HoldWhen); err != nil { + return fmt.Errorf("publish: hold_when: %w", err) + } + } + if ci := spec.Publish.CI; ci != nil && strings.TrimSpace(ci.Timeout) != "" { + timeout, err := time.ParseDuration(ci.Timeout) + if err != nil { + return fmt.Errorf("publish: ci: timeout %q is not a duration like \"30m\"", ci.Timeout) + } + if timeout < MinCITimeout || timeout > MaxCITimeout { + return fmt.Errorf("publish: ci: timeout %s is outside %s..%s", + timeout, MinCITimeout, MaxCITimeout) + } } return nil } diff --git a/internal/protocol/definition_test.go b/internal/protocol/definition_test.go index d7e89f6..47200d4 100644 --- a/internal/protocol/definition_test.go +++ b/internal/protocol/definition_test.go @@ -3,6 +3,7 @@ package protocol import ( "strings" "testing" + "time" ) // simpleSDLC is the plan's canonical shape (KTD2): an agent-phase repair @@ -439,7 +440,40 @@ roster: phases: - {name: build, kind: agent, owner: builder} publish: {} -`, "hold_when is required") +`, "declare hold_when, ci, or both") +} + +func TestPublishCIWaitIsOptInWithABoundedTimeout(t *testing.T) { + const base = ` +name: ci +roster: + builder: {model: opus, system_prompt: s, user_prompt: u} +phases: + - {name: build, kind: agent, owner: builder} +` + if spec := mustParse(t, base); spec.WaitsForCI() { + t.Fatal("a definition with no publish block waits for CI; CI wait is opt-in") + } + if spec := mustParse(t, base+"publish: {hold_when: \"risk == high\"}\n"); spec.WaitsForCI() { + t.Fatal("a definition with only hold_when waits for CI") + } + if spec := mustParse(t, base+"publish: {ci: {wait: false}}\n"); spec.WaitsForCI() { + t.Fatal("ci.wait: false waits for CI") + } + + spec := mustParse(t, base+"publish: {ci: {wait: true}}\n") + if !spec.WaitsForCI() || spec.CITimeout() != DefaultCITimeout { + t.Fatalf("waits=%v timeout=%s, want a wait with the default %s", + spec.WaitsForCI(), spec.CITimeout(), DefaultCITimeout) + } + spec = mustParse(t, base+"publish: {hold_when: \"risk != low\", ci: {wait: true, timeout: 45m}}\n") + if !spec.WaitsForCI() || spec.CITimeout() != 45*time.Minute || spec.Publish.HoldWhen != "risk != low" { + t.Fatalf("publish = %+v timeout=%s, want hold_when and a 45m CI wait", spec.Publish, spec.CITimeout()) + } + + mustReject(t, base+"publish: {ci: {wait: true, timeout: soon}}\n", "publish: ci: timeout", "not a duration") + mustReject(t, base+"publish: {ci: {wait: true, timeout: 30s}}\n", "publish: ci: timeout", "outside") + mustReject(t, base+"publish: {ci: {wait: true, timeout: 7h}}\n", "publish: ci: timeout", "outside") } func TestNegativeRoleBudgetIsRejected(t *testing.T) { diff --git a/internal/protocol/publish.go b/internal/protocol/publish.go index 7089533..5f56d7c 100644 --- a/internal/protocol/publish.go +++ b/internal/protocol/publish.go @@ -19,10 +19,15 @@ const ( PublishStepPush = "push" PublishStepPullRequest = "pull_request" PublishStepProof = "proof" + // PublishStepCI records a green CI run on the pull request's head. It + // runs only for definitions whose publish.ci waits, and its remote_ref is + // the head CI passed on — which may be a later commit than proof's when a + // person pushed a fix before the publish retry. + PublishStepCI = "ci" ) // PublishSteps is the pipeline in execution order. -var PublishSteps = []string{PublishStepPush, PublishStepPullRequest, PublishStepProof} +var PublishSteps = []string{PublishStepPush, PublishStepPullRequest, PublishStepProof, PublishStepCI} // PublishStepPrerequisites names the steps that must already be recorded // before a step may be authorized or recorded. It is enforced @@ -32,6 +37,7 @@ var PublishStepPrerequisites = map[string][]string{ PublishStepPush: nil, PublishStepPullRequest: {PublishStepPush}, PublishStepProof: {PublishStepPush, PublishStepPullRequest}, + PublishStepCI: {PublishStepPush, PublishStepPullRequest, PublishStepProof}, } // ValidPublishStep reports whether value names a publish step. @@ -73,6 +79,19 @@ const ( MaxPublishStdoutBytes = 4 << 20 MaxPublishStderrBytes = 64 << 10 + // CIPollInterval is how often the ci step re-reads check state while it + // waits. GitHub's own UI refreshes on a similar cadence. + CIPollInterval = 30 * time.Second + + // CIRegistrationGrace is how long a head with no checks at all counts as + // "CI has not registered yet" before it counts as "this repository runs + // no CI". Workflows usually attach within seconds of a push. + CIRegistrationGrace = 2 * time.Minute + + // MaxCITransientFailures is how many consecutive failed check reads the + // ci step tolerates before it gives up with a diagnostic. + MaxCITransientFailures = 3 + // MaxPublishDiagnosticBytes caps the diagnostic text carried out of a // failed publish step into the attempt result. MaxPublishDiagnosticBytes = 4 << 10 @@ -169,4 +188,7 @@ type PublishRetry struct { Attempt Attempt `json:"attempt"` Job Job `json:"job"` Records []PublishRecord `json:"records"` + // Snapshot is the run's frozen definition, so the retrying worker applies + // the same publish policy (a CI wait, say) the original attempt ran under. + Snapshot string `json:"snapshot"` } diff --git a/internal/worker/publish.go b/internal/worker/publish.go index e7ee8fc..e89cb9b 100644 --- a/internal/worker/publish.go +++ b/internal/worker/publish.go @@ -8,7 +8,7 @@ // on a context that cannot be cancelled and never consults the attempt's // cancellation signal: cancel is a no-op inside publish. // -// Three idempotent steps, each authorized before its side effect and recorded +// Three idempotent steps (four with CI), each authorized before its side effect and recorded // after it, both under the lease token (R6): // // push the attempt-scoped branch jig//, staging @@ -20,6 +20,12 @@ // proof the REMOTE ref is read back and must equal the pushed // commit; a local ref or reflog entry proves nothing, and it is // this proof that later unblocks worktree cleanup (R16) +// ci only when the definition sets publish.ci.wait: the branch's +// current remote head must go green before the job may be +// `accepted`. This step has no side effect, so it is the one +// place publish honours cancellation — a cancel while waiting +// ends it as ci_wait_cancelled, with the branch and pull +// request left in place // // A push that succeeds and a pull request that fails leaves the job // `accepted_unpublished` with every proven step recorded, and the publish-only @@ -47,10 +53,12 @@ import ( "os" "os/exec" "path/filepath" + "regexp" "sort" "strconv" "strings" "sync" + "time" "github.com/StructuPath/jig/internal/protocol" ) @@ -107,14 +115,18 @@ const ( // says it reused push and performed pull_request is a retry that provably did // not push twice. type PublishSummary struct { - State string `json:"state"` - Code string `json:"code,omitempty"` - Detail string `json:"detail,omitempty"` - Branch string `json:"branch,omitempty"` - RemoteRef string `json:"remote_ref,omitempty"` - PullRequestURL string `json:"pr_url,omitempty"` - Performed []string `json:"performed_steps,omitempty"` - Reused []string `json:"reused_steps,omitempty"` + State string `json:"state"` + Code string `json:"code,omitempty"` + Detail string `json:"detail,omitempty"` + Branch string `json:"branch,omitempty"` + RemoteRef string `json:"remote_ref,omitempty"` + PullRequestURL string `json:"pr_url,omitempty"` + // CIRef is the head commit CI was judged on; CIFailures names the checks + // that were red (or still pending at the timeout), bounded. + CIRef string `json:"ci_ref,omitempty"` + CIFailures []CICheck `json:"ci_failures,omitempty"` + Performed []string `json:"performed_steps,omitempty"` + Reused []string `json:"reused_steps,omitempty"` } // Published reports whether the pipeline completed with remote proof — the @@ -149,6 +161,28 @@ type PullRequestRequest struct { // request. Tests substitute a fake; no unit test reaches GitHub. type PullRequestGateway interface { FindOrCreatePullRequest(ctx context.Context, request PullRequestRequest) (PullRequest, error) + // CommitChecks reports every CI signal attached to one commit: check runs + // and legacy commit statuses alike, each normalized to pending, pass, or + // fail. + CommitChecks(ctx context.Context, repository, sha string) ([]CICheck, error) +} + +// CI check verdicts, normalized across check runs and commit statuses. +const ( + CIPending = "pending" + CIPass = "pass" + CIFail = "fail" +) + +// CICheck is one CI signal on a commit. +type CICheck struct { + Name string `json:"name"` + // Verdict is CIPending, CIPass, or CIFail. + Verdict string `json:"verdict"` + // Conclusion is GitHub's own word (failure, timed_out, error, …), kept + // for the diagnostic. + Conclusion string `json:"conclusion,omitempty"` + URL string `json:"url,omitempty"` } // ---- publish options ------------------------------------------------------- @@ -164,6 +198,43 @@ type PublishOptions struct { // BaseBranch is the pull request's target branch; empty resolves the // repository's default branch through the gateway. BaseBranch string + // CIPollInterval and CIRegistrationGrace override the protocol defaults; + // zero means the default. Tests shrink them; operators need not. + CIPollInterval time.Duration + CIRegistrationGrace time.Duration +} + +func (o PublishOptions) ciPollInterval() time.Duration { + if o.CIPollInterval > 0 { + return o.CIPollInterval + } + return protocol.CIPollInterval +} + +func (o PublishOptions) ciRegistrationGrace() time.Duration { + if o.CIRegistrationGrace > 0 { + return o.CIRegistrationGrace + } + return protocol.CIRegistrationGrace +} + +// ciPolicy is what the frozen definition says about CI: whether publish +// waits for it, and for how long. +type ciPolicy struct { + wait bool + timeout time.Duration +} + +// ciPolicyFor reads the policy out of a run snapshot. A snapshot that does +// not parse could not have run; publish does not wait on it, and the control +// plane — which parses the same snapshot — refuses to accept the attempt, so +// a parse failure can never turn into an unchecked `accepted`. +func ciPolicyFor(snapshot string) ciPolicy { + spec, err := protocol.ParseDefinition([]byte(snapshot)) + if err != nil { + return ciPolicy{} + } + return ciPolicy{wait: spec.WaitsForCI(), timeout: spec.CITimeout()} } func (o PublishOptions) authorName() string { @@ -264,7 +335,7 @@ func (r *PublishingRunner) Run(ctx context.Context, prepared *PreparedAttempt) O // cancellation signal deliberately unread. Every command inside still // carries its own timeout, so "uncancellable" is not "unbounded". summary := worker.publish(context.WithoutCancel(ctx), r.gateway, r.options, - target, changedPathsFromResult(outcome.Result)) + target, changedPathsFromResult(outcome.Result), ciPolicyFor(prepared.Claim.Snapshot)) return publishedOutcome(outcome, summary) } @@ -312,6 +383,7 @@ func (w *Worker) publish( options PublishOptions, target publishTarget, changedPaths []string, + ci ciPolicy, ) PublishSummary { summary := PublishSummary{State: PublishStateFailed, Branch: target.branch} if target.branch != protocol.PublishBranch(target.jobID, target.attemptNumber) { @@ -379,6 +451,23 @@ func (w *Worker) publish( }); err != nil { return summary } + if ci.wait { + green, err := w.publishStep(ctx, target, protocol.PublishStepCI, proven, &summary, + func(authorization protocol.PublishAuthorization) (protocol.PublishStepRequest, error) { + head, failures, err := w.awaitCI(ctx, gateway, options, target, authorization.Branch, ci.timeout) + summary.CIRef = head + summary.CIFailures = failures + if err != nil { + return protocol.PublishStepRequest{}, err + } + return protocol.PublishStepRequest{Step: protocol.PublishStepCI, + Branch: authorization.Branch, RemoteRef: head}, nil + }) + if err != nil { + return summary + } + summary.CIRef = green.RemoteRef + } summary.State = PublishStatePublished summary.Code = "" summary.Detail = "" @@ -551,35 +640,173 @@ func (w *Worker) commitChangedPaths( // ref this worker wrote itself all prove nothing — only the origin's answer // does, and it is this answer that later lets cleanup delete the worktree. func (w *Worker) verifyRemoteRef(ctx context.Context, target publishTarget, branch, expected string) error { + observed, err := w.remoteHead(ctx, target, branch) + if err != nil { + return err + } + if observed == "" { + return publishFailure("publish_proof_missing", + "the remote has no ref refs/heads/%s; the push did not reach it", branch) + } + if observed != expected { + return publishFailure("publish_proof_mismatch", + "the remote's refs/heads/%s is %s, not the pushed commit %s", branch, observed, expected) + } + return nil +} + +// remoteHead reads the branch's current commit on the remote itself (never a +// local ref), or "" when the remote has no such branch. +func (w *Worker) remoteHead(ctx context.Context, target publishTarget, branch string) (string, error) { entry, err := w.cache.entry(ctx, target.repository) if err != nil { - return publishFailure("publish_repository_unavailable", + return "", publishFailure("publish_repository_unavailable", "repository cache entry unavailable: %s", err) } entry.mutex.Lock() defer entry.mutex.Unlock() stdout, err := runGit(ctx, entry.dir, "ls-remote", "origin", "refs/heads/"+branch) if err != nil { - return publishFailure("publish_proof_unavailable", "%s", err.Error()) + return "", publishFailure("publish_proof_unavailable", "%s", err.Error()) } - observed := "" for _, line := range strings.Split(stdout, "\n") { sha, ref, found := strings.Cut(strings.TrimSpace(line), "\t") - if !found || strings.TrimSpace(ref) != "refs/heads/"+branch { - continue + if found && strings.TrimSpace(ref) == "refs/heads/"+branch { + return strings.TrimSpace(sha), nil } - observed = strings.TrimSpace(sha) - break } - if observed == "" { - return publishFailure("publish_proof_missing", - "the remote has no ref refs/heads/%s; the push did not reach it", branch) + return "", nil +} + +// ---- CI wait --------------------------------------------------------------- + +// awaitCI waits for CI on the branch's CURRENT remote head — not only the +// commit jig pushed, so a fix a person pushed before a publish retry is the +// one judged. It returns the head CI passed on, or a diagnostic: +// +// - ci_failed: a check concluded red. It fails fast on the first red check +// rather than waiting for the rest — red is red. +// - ci_timeout: checks were still pending when the definition's timeout ran +// out. +// - ci_wait_cancelled: the operator cancelled the job while it waited. +// - ci_unavailable: check state could not be read MaxCITransientFailures +// times in a row. +// +// A head with no checks at all passes once CIRegistrationGrace has elapsed: +// that repository runs no CI on this branch, and waiting the full timeout +// would only park a worker slot. A head that moves mid-wait restarts the +// grace clock but not the timeout. +func (w *Worker) awaitCI( + ctx context.Context, gateway PullRequestGateway, options PublishOptions, + target publishTarget, branch string, timeout time.Duration, +) (string, []CICheck, error) { + start := time.Now() + deadline := start.Add(timeout) + head, headSeen := "", start + transient := 0 + var lastErr error + var pending []CICheck + for { + current, err := w.remoteHead(ctx, target, branch) + if err == nil && current == "" { + err = publishFailure("publish_proof_missing", + "the remote has no ref refs/heads/%s to run CI on", branch) + } + var checks []CICheck + if err == nil { + if current != head { + head, headSeen = current, time.Now() + } + checks, err = gateway.CommitChecks(ctx, target.repository, head) + } + if err != nil { + transient++ + lastErr = err + if transient >= protocol.MaxCITransientFailures { + return head, nil, publishFailure("ci_unavailable", + "CI state could not be read %d times in a row: %s", + transient, boundedText(lastErr.Error(), protocol.MaxPublishDiagnosticBytes)) + } + } else { + transient = 0 + var failed []CICheck + pending = pending[:0] + for _, check := range checks { + switch check.Verdict { + case CIFail: + failed = append(failed, check) + case CIPending: + pending = append(pending, check) + } + } + if len(failed) > 0 { + return head, boundedChecks(failed), publishFailure("ci_failed", + "%d CI check(s) failed on %s: %s", len(failed), shortSHA(head), describeChecks(failed)) + } + if len(checks) > 0 && len(pending) == 0 { + return head, nil, nil + } + if len(checks) == 0 && time.Since(headSeen) >= options.ciRegistrationGrace() { + return head, nil, nil + } + } + if !time.Now().Before(deadline) { + waiting := "no checks had registered" + if len(pending) > 0 { + waiting = "still pending: " + describeChecks(pending) + } + return head, boundedChecks(pending), publishFailure("ci_timeout", + "CI did not finish on %s within %s; %s", shortSHA(head), timeout, waiting) + } + wait := options.ciPollInterval() + if remaining := time.Until(deadline); remaining < wait { + wait = remaining + } + timer := time.NewTimer(wait) + select { + case <-target.lease.cancelled: + timer.Stop() + return head, nil, publishFailure("ci_wait_cancelled", + "the job was cancelled while waiting for CI on %s", shortSHA(head)) + case <-ctx.Done(): + timer.Stop() + return head, nil, publishFailure("ci_wait_cancelled", "the CI wait was interrupted: %s", ctx.Err()) + case <-timer.C: + } } - if observed != expected { - return publishFailure("publish_proof_mismatch", - "the remote's refs/heads/%s is %s, not the pushed commit %s", branch, observed, expected) +} + +// maxReportedCIChecks bounds how many checks a summary names. +const maxReportedCIChecks = 20 + +func boundedChecks(checks []CICheck) []CICheck { + if len(checks) > maxReportedCIChecks { + checks = checks[:maxReportedCIChecks] } - return nil + return append([]CICheck(nil), checks...) +} + +func describeChecks(checks []CICheck) string { + parts := make([]string, 0, len(checks)) + for i, check := range checks { + if i == maxReportedCIChecks { + parts = append(parts, fmt.Sprintf("and %d more", len(checks)-i)) + break + } + label := check.Name + if check.Conclusion != "" && check.Verdict == CIFail { + label += " (" + check.Conclusion + ")" + } + parts = append(parts, label) + } + return strings.Join(parts, ", ") +} + +func shortSHA(sha string) string { + if len(sha) > 12 { + return sha[:12] + } + return sha } // disabledHooksPath is an empty directory jig points git at for its own @@ -807,7 +1034,7 @@ func (w *Worker) RetryPublish( // The critical section again: a retry that has just pushed must not be // abandoned before its pull request exists. summary := w.publish(context.WithoutCancel(ctx), gateway, options, target, - changedPathsFromResult(retry.Attempt.Result)) + changedPathsFromResult(retry.Attempt.Result), ciPolicyFor(retry.Snapshot)) outcome := publishedOutcome(Outcome{ State: protocol.AttemptAcceptedUnpublished, Result: retry.Attempt.Result, }, summary) @@ -1160,6 +1387,98 @@ func (g *GitHubCLIGateway) findPullRequest(ctx context.Context, project, head st return PullRequest{}, false, nil } +// ciSHAPattern is a full commit SHA; nothing else goes into an API path. +var ciSHAPattern = regexp.MustCompile(`^[0-9a-f]{40}([0-9a-f]{24})?$`) + +// CommitChecks reads both GitHub CI surfaces for one commit: check runs +// (Actions and most apps) and the combined commit status (older integrations). +// Each is read with --jq down to one compact JSON object per line, so the +// parse never depends on how gh concatenates paginated pages. +func (g *GitHubCLIGateway) CommitChecks(ctx context.Context, repository, sha string) ([]CICheck, error) { + project, err := githubProject(repository) + if err != nil { + return nil, err + } + if !ciSHAPattern.MatchString(sha) { + return nil, publishFailure("ci_invalid_ref", "%q is not a commit SHA", sha) + } + if _, err := g.lookPath()("gh"); err != nil { + return nil, publishFailure("gh_not_found", + "the GitHub CLI (gh) was not found on PATH. Install gh, then run `gh auth login`.") + } + runs, err := g.apiLines(ctx, "gh api check-runs", + "repos/"+project+"/commits/"+sha+"/check-runs?per_page=100", + `.check_runs[] | {name: .name, status: .status, conclusion: (.conclusion // ""), url: (.html_url // "")}`) + if err != nil { + return nil, err + } + statuses, err := g.apiLines(ctx, "gh api status", + "repos/"+project+"/commits/"+sha+"/status?per_page=100", + `.statuses[] | {name: .context, state: .state, url: (.target_url // "")}`) + if err != nil { + return nil, err + } + checks := make([]CICheck, 0, len(runs)+len(statuses)) + for _, line := range runs { + var run struct{ Name, Status, Conclusion, URL string } + if err := json.Unmarshal(line, &run); err != nil { + return nil, publishFailure("gh_malformed_output", "gh api check-runs returned unreadable JSON") + } + checks = append(checks, CICheck{Name: run.Name, Verdict: checkRunVerdict(run.Status, run.Conclusion), + Conclusion: run.Conclusion, URL: run.URL}) + } + for _, line := range statuses { + var status struct{ Name, State, URL string } + if err := json.Unmarshal(line, &status); err != nil { + return nil, publishFailure("gh_malformed_output", "gh api status returned unreadable JSON") + } + checks = append(checks, CICheck{Name: status.Name, Verdict: commitStatusVerdict(status.State), + Conclusion: status.State, URL: status.URL}) + } + return checks, nil +} + +func (g *GitHubCLIGateway) apiLines(ctx context.Context, label, path, filter string) ([][]byte, error) { + stdout, stderr, stdoutTooLarge, stderrTooLarge, err := g.run()(ctx, "gh", + "api", "--paginate", "-H", "Accept: application/vnd.github+json", path, "--jq", filter) + if err != nil { + return nil, ghDiagnostic(label, err, stderr, stdoutTooLarge, stderrTooLarge) + } + var lines [][]byte + for _, line := range bytes.Split(stdout, []byte("\n")) { + if line = bytes.TrimSpace(line); len(line) > 0 { + lines = append(lines, line) + } + } + return lines, nil +} + +// checkRunVerdict maps a GitHub check run. Neutral and skipped conclusions +// are not failures; anything else that completed without success is. +func checkRunVerdict(status, conclusion string) string { + if status != "completed" { + return CIPending + } + switch conclusion { + case "success", "neutral", "skipped": + return CIPass + case "": + return CIPending + } + return CIFail +} + +// commitStatusVerdict maps a legacy commit status state. +func commitStatusVerdict(state string) string { + switch state { + case "success": + return CIPass + case "failure", "error": + return CIFail + } + return CIPending +} + func (g *GitHubCLIGateway) defaultBranch(ctx context.Context, project string) (string, error) { // `gh repo view` takes the project positionally — it has no --repo flag, // unlike every other gh subcommand publish uses. githubProject has already diff --git a/internal/worker/publish_ci_test.go b/internal/worker/publish_ci_test.go new file mode 100644 index 0000000..3c68eb8 --- /dev/null +++ b/internal/worker/publish_ci_test.go @@ -0,0 +1,377 @@ +// publish_ci_test.go — the opt-in CI wait: a definition with publish.ci.wait +// reaches `accepted` only once the branch's head is green, ends +// `accepted_unpublished` with the red checks named when it is not, and a +// publish-only retry judges whatever the branch's head is by then. The remote +// is a real bare repository; CI itself is the scripted fake gateway. +package worker + +import ( + "context" + "errors" + "os" + "path/filepath" + "slices" + "strings" + "testing" + "time" + + "github.com/StructuPath/jig/internal/protocol" +) + +// ciSnapshot is integrationSnapshot opted into the CI wait. +const ciSnapshot = integrationSnapshot + `publish: + ci: + wait: true +` + +// fastCIOptions shrinks the poll interval and registration grace so a CI +// scenario takes milliseconds. +func fastCIOptions() PublishOptions { + return PublishOptions{ + CommitAuthorName: "jig-test", CommitAuthorEmail: "jig@test", + CIPollInterval: 5 * time.Millisecond, CIRegistrationGrace: 20 * time.Millisecond, + } +} + +func newCIWorker( + t *testing.T, h *harness, inner AttemptRunner, gateway PullRequestGateway, +) *Worker { + t.Helper() + runner := NewPublishingRunner(inner, gateway, fastCIOptions()) + w := newTestWorker(t, h, filepath.Join(t.TempDir(), "worker"), 1, runner) + runner.Bind(w) + return w +} + +func check(name, verdict string) CICheck { + conclusion := "" + if verdict == CIFail { + conclusion = "failure" + } + return CICheck{Name: name, Verdict: verdict, Conclusion: conclusion} +} + +func recordedSteps(t *testing.T, w *Worker, attemptID string) []string { + t.Helper() + records, err := w.client.AttemptPublishRecords(context.Background(), attemptID) + if err != nil { + t.Fatalf("read publish records: %v", err) + } + steps := make([]string, 0, len(records)) + for _, record := range records { + steps = append(steps, record.Step) + } + slices.Sort(steps) + return steps +} + +// pushToBranch commits a file onto the attempt branch on the remote the way +// a person fixing CI would, and returns the new head. +func pushToBranch(t *testing.T, originDir, branch string) string { + t.Helper() + clone := filepath.Join(t.TempDir(), "human") + gitRun(t, "", "clone", "--branch", branch, originDir, clone) + if err := os.WriteFile(filepath.Join(clone, "fix.txt"), []byte("the fix\n"), 0o644); err != nil { + t.Fatalf("write fix: %v", err) + } + gitRun(t, clone, "add", "fix.txt") + gitRun(t, clone, "-c", "user.name=human", "-c", "user.email=human@test", "commit", "-m", "fix CI") + gitRun(t, clone, "push", "origin", branch) + return gitRun(t, clone, "rev-parse", "HEAD") +} + +// ---- scenario: green --------------------------------------------------------- + +// CI that is pending and then green lets the attempt reach `accepted`, with +// the ci step on the ledger at the commit CI judged. +func TestAWaitingDefinitionIsAcceptedOnceCIOnTheHeadIsGreen(t *testing.T) { + h := newHarness(t) + originDir, head, identity := newOriginRepo(t) + h.seedRunWithSnapshot("run-1", ciSnapshot, protocol.RunTarget{Repository: identity, BaseSHA: head}) + job := h.enqueue("run-1", identity) + + gateway := newFakeGateway() + gateway.checks = func(_ string, poll int) ([]CICheck, error) { + if poll < 3 { + return []CICheck{check("test", CIPending), check("lint", CIPass)}, nil + } + return []CICheck{check("test", CIPass), check("lint", CIPass)}, nil + } + w := newCIWorker(t, h, writeAndDeclare(t, map[string]string{"work.txt": "work\n"}), gateway) + + attempt, err := w.ClaimOnce(context.Background()) + if err != nil { + t.Fatalf("claim: %v", err) + } + if attempt == nil || attempt.State != protocol.AttemptAccepted { + t.Fatalf("attempt = %+v, want accepted", attempt) + } + pushed := remoteBranches(t, originDir)[protocol.PublishBranch(job.ID, 1)] + summary := publishSummaryOf(t, attempt.Result) + if !summary.Published() || summary.CIRef != pushed || len(summary.CIFailures) != 0 { + t.Fatalf("publish summary = %+v, want published with ci_ref %s", summary, pushed) + } + if !slices.Contains(summary.Performed, protocol.PublishStepCI) { + t.Fatalf("performed steps = %v, want ci among them", summary.Performed) + } + if steps := recordedSteps(t, w, attempt.ID); !slices.Equal(steps, + []string{"ci", "proof", "pull_request", "push"}) { + t.Fatalf("ledger steps = %v, want all four", steps) + } + if gateway.checkPolls != 3 { + t.Fatalf("CI was polled %d times, want 3 (pending, pending, green)", gateway.checkPolls) + } +} + +// ---- scenario: red, then a person's fix, then the retry ---------------------- + +// Red CI ends the attempt `accepted_unpublished` — branch and pull request in +// place, the red checks named — and the publish-only retry judges the +// branch's CURRENT head, so a fix a person pushed is what turns it green. +func TestRedCIEndsAcceptedUnpublishedAndTheRetryJudgesTheCurrentHead(t *testing.T) { + h := newHarness(t) + originDir, head, identity := newOriginRepo(t) + h.seedRunWithSnapshot("run-1", ciSnapshot, protocol.RunTarget{Repository: identity, BaseSHA: head}) + job := h.enqueue("run-1", identity) + branch := protocol.PublishBranch(job.ID, 1) + + var redSHA string + gateway := newFakeGateway() + gateway.checks = func(sha string, _ int) ([]CICheck, error) { + if redSHA == "" || sha == redSHA { + redSHA = sha + return []CICheck{check("lint", CIPass), check("test", CIFail), check("e2e", CIPending)}, nil + } + return []CICheck{check("lint", CIPass), check("test", CIPass), check("e2e", CIPass)}, nil + } + w := newCIWorker(t, h, writeAndDeclare(t, map[string]string{"work.txt": "work\n"}), gateway) + + attempt, err := w.ClaimOnce(context.Background()) + if err != nil { + t.Fatalf("claim: %v", err) + } + if attempt == nil || attempt.State != protocol.AttemptAcceptedUnpublished { + t.Fatalf("attempt = %+v, want accepted_unpublished", attempt) + } + summary := publishSummaryOf(t, attempt.Result) + if summary.Published() || summary.Code != "ci_failed" { + t.Fatalf("publish summary = %+v, want ci_failed", summary) + } + if len(summary.CIFailures) != 1 || summary.CIFailures[0].Name != "test" || + !strings.Contains(summary.Detail, "test (failure)") { + t.Fatalf("ci failures = %+v detail %q, want exactly the red test check", summary.CIFailures, summary.Detail) + } + if gateway.checkPolls != 1 { + t.Fatalf("CI was polled %d times; red is red, the wait fails fast", gateway.checkPolls) + } + if summary.PullRequestURL == "" || remoteBranches(t, originDir)[branch] != redSHA { + t.Fatalf("summary = %+v, want the branch and pull request left in place", summary) + } + if steps := recordedSteps(t, w, attempt.ID); slices.Contains(steps, "ci") { + t.Fatalf("ledger steps = %v, a red run must not record ci", steps) + } + jobAfter, err := h.store.Job(context.Background(), job.ID) + if err != nil { + t.Fatalf("read job: %v", err) + } + if jobAfter.State != protocol.JobAcceptedUnpublished { + t.Fatalf("job state = %q, want accepted_unpublished", jobAfter.State) + } + + fixed := pushToBranch(t, originDir, branch) + retried, err := w.RetryPublish(context.Background(), job.ID, gateway, fastCIOptions()) + if err != nil { + t.Fatalf("publish retry: %v", err) + } + if retried.State != protocol.AttemptAccepted { + t.Fatalf("retried attempt state = %q (result %s), want accepted", retried.State, retried.Result) + } + retrySummary := publishSummaryOf(t, retried.Result) + if retrySummary.CIRef != fixed { + t.Fatalf("retry judged %s, want the person's fix %s", retrySummary.CIRef, fixed) + } + if !slices.Equal(retrySummary.Performed, []string{protocol.PublishStepCI}) { + t.Fatalf("retry performed %v, want only ci — push, pull request, and proof were proven", retrySummary.Performed) + } + if created, _ := gateway.counts(); created != 1 { + t.Fatalf("gateway created %d pull requests, want 1", created) + } +} + +// ---- scenario: no CI configured --------------------------------------------- + +// A head with no checks at all passes after the registration grace instead +// of parking a worker slot for the full timeout. +func TestAHeadWithNoChecksPassesAfterTheRegistrationGrace(t *testing.T) { + h := newHarness(t) + _, head, identity := newOriginRepo(t) + h.seedRunWithSnapshot("run-1", ciSnapshot, protocol.RunTarget{Repository: identity, BaseSHA: head}) + h.enqueue("run-1", identity) + + gateway := newFakeGateway() // checks == nil: the repository runs no CI + w := newCIWorker(t, h, writeAndDeclare(t, map[string]string{"work.txt": "work\n"}), gateway) + started := time.Now() + attempt, err := w.ClaimOnce(context.Background()) + if err != nil { + t.Fatalf("claim: %v", err) + } + if attempt == nil || attempt.State != protocol.AttemptAccepted { + t.Fatalf("attempt = %+v, want accepted", attempt) + } + if gateway.checkPolls < 2 || time.Since(started) < fastCIOptions().CIRegistrationGrace { + t.Fatalf("passed after %d polls; an empty head must wait out the grace first", gateway.checkPolls) + } +} + +// ---- awaitCI edges ----------------------------------------------------------- + +// publishedWorker runs one ordinary (non-waiting) publish so the worker's +// repository cache and the remote branch exist, and returns what awaitCI +// needs to be driven directly. +func publishedWorker(t *testing.T) (*Worker, *fakeGateway, publishTarget, string) { + t.Helper() + h := newHarness(t) + originDir, head, identity := newOriginRepo(t) + h.seedRun("run-1", protocol.RunTarget{Repository: identity, BaseSHA: head}) + job := h.enqueue("run-1", identity) + gateway := newFakeGateway() + w := newCIWorker(t, h, writeAndDeclare(t, map[string]string{"work.txt": "work\n"}), gateway) + attempt, err := w.ClaimOnce(context.Background()) + if err != nil || attempt == nil || attempt.State != protocol.AttemptAccepted { + t.Fatalf("setup publish: attempt=%+v err=%v", attempt, err) + } + if gateway.checkPolls != 0 { + t.Fatalf("a definition without publish.ci polled CI %d times", gateway.checkPolls) + } + branch := protocol.PublishBranch(job.ID, 1) + target := publishTarget{ + attemptID: attempt.ID, jobID: job.ID, attemptNumber: 1, + repository: identity, branch: branch, baseSHA: head, + lease: newAttemptLease(w.client, attempt.ID, "unused"), + } + return w, gateway, target, remoteBranches(t, originDir)[branch] +} + +func TestTheCIWaitTimesOutNamingThePendingChecks(t *testing.T) { + w, gateway, target, pushed := publishedWorker(t) + gateway.checks = func(string, int) ([]CICheck, error) { + return []CICheck{check("lint", CIPass), check("slow-e2e", CIPending)}, nil + } + head, failures, err := w.awaitCI(context.Background(), gateway, fastCIOptions(), target, target.branch, + 40*time.Millisecond) + if publishCode(err) != "ci_timeout" || !strings.Contains(err.Error(), "slow-e2e") { + t.Fatalf("err = %v, want ci_timeout naming slow-e2e", err) + } + if head != pushed || len(failures) != 1 || failures[0].Name != "slow-e2e" { + t.Fatalf("head=%s failures=%+v, want %s with the pending check", head, failures, pushed) + } +} + +func TestTheCIWaitStopsWhenTheJobIsCancelled(t *testing.T) { + w, gateway, target, _ := publishedWorker(t) + gateway.checks = func(string, int) ([]CICheck, error) { + return []CICheck{check("test", CIPending)}, nil + } + target.lease.cancelOnce.Do(func() { close(target.lease.cancelled) }) + _, _, err := w.awaitCI(context.Background(), gateway, fastCIOptions(), target, target.branch, time.Hour) + if publishCode(err) != "ci_wait_cancelled" { + t.Fatalf("err = %v, want ci_wait_cancelled", err) + } +} + +func TestRepeatedCIReadFailuresEndTheWaitAsUnavailable(t *testing.T) { + w, gateway, target, _ := publishedWorker(t) + gateway.checks = func(string, int) ([]CICheck, error) { + return nil, errors.New("HTTP 502 from api.github.com") + } + _, _, err := w.awaitCI(context.Background(), gateway, fastCIOptions(), target, target.branch, time.Hour) + if publishCode(err) != "ci_unavailable" || !strings.Contains(err.Error(), "502") { + t.Fatalf("err = %v, want ci_unavailable carrying the cause", err) + } + if gateway.checkPolls != protocol.MaxCITransientFailures { + t.Fatalf("polled %d times, want %d", gateway.checkPolls, protocol.MaxCITransientFailures) + } +} + +// One transient read failure is not a verdict: the next poll decides. +func TestASingleCIReadFailureIsRetried(t *testing.T) { + w, gateway, target, pushed := publishedWorker(t) + gateway.checks = func(_ string, poll int) ([]CICheck, error) { + if poll == 1 { + return nil, errors.New("connection reset") + } + return []CICheck{check("test", CIPass)}, nil + } + head, _, err := w.awaitCI(context.Background(), gateway, fastCIOptions(), target, target.branch, time.Hour) + if err != nil || head != pushed { + t.Fatalf("head=%s err=%v, want green on %s", head, err, pushed) + } +} + +// ---- the gh gateway ---------------------------------------------------------- + +// CommitChecks reads check runs and commit statuses through `gh api`, and +// the normalization is what every verdict above rests on. +func TestGitHubGatewayNormalizesCheckRunsAndCommitStatuses(t *testing.T) { + const sha = "0123456789abcdef0123456789abcdef01234567" + var calls [][]string + gateway := &GitHubCLIGateway{ + LookPath: func(string) (string, error) { return "/usr/bin/gh", nil }, + Run: func(_ context.Context, name string, arguments ...string) ([]byte, []byte, bool, bool, error) { + calls = append(calls, append([]string{name}, arguments...)) + path := strings.Join(arguments, " ") + switch { + case strings.Contains(path, "/check-runs"): + return []byte(strings.Join([]string{ + `{"name":"build","status":"completed","conclusion":"success","url":"https://ci/1"}`, + `{"name":"docs","status":"completed","conclusion":"skipped","url":""}`, + `{"name":"test","status":"completed","conclusion":"timed_out","url":"https://ci/3"}`, + `{"name":"e2e","status":"in_progress","conclusion":"","url":""}`, + "", + }, "\n")), nil, false, false, nil + case strings.Contains(path, "/status"): + return []byte(`{"name":"legacy/ci","state":"error","url":"https://legacy"}` + "\n" + + `{"name":"legacy/cov","state":"pending","url":""}` + "\n"), nil, false, false, nil + } + t.Fatalf("unexpected gh call %v", arguments) + return nil, nil, false, false, nil + }, + } + checks, err := gateway.CommitChecks(context.Background(), "github.com/example/repo", sha) + if err != nil { + t.Fatalf("commit checks: %v", err) + } + got := map[string]string{} + for _, c := range checks { + got[c.Name] = c.Verdict + } + want := map[string]string{ + "build": CIPass, "docs": CIPass, "test": CIFail, "e2e": CIPending, + "legacy/ci": CIFail, "legacy/cov": CIPending, + } + if len(got) != len(want) { + t.Fatalf("checks = %+v, want %v", checks, want) + } + for name, verdict := range want { + if got[name] != verdict { + t.Fatalf("%s verdict = %q, want %q (all: %+v)", name, got[name], verdict, checks) + } + } + if len(calls) != 2 { + t.Fatalf("gh was called %d times, want check-runs and status", len(calls)) + } + for _, call := range calls { + joined := strings.Join(call, " ") + if !strings.Contains(joined, "--paginate") || !strings.Contains(joined, "--jq") || + !strings.Contains(joined, "repos/example/repo/commits/"+sha+"/") { + t.Fatalf("gh call %v does not read the commit's CI with pagination", call) + } + } + + if _, err := gateway.CommitChecks(context.Background(), "github.com/example/repo", "main;rm -rf"); publishCode(err) != "ci_invalid_ref" { + t.Fatalf("a non-SHA ref: err = %v, want ci_invalid_ref", err) + } + if len(calls) != 2 { + t.Fatal("an invalid ref reached gh") + } +} diff --git a/internal/worker/publish_test.go b/internal/worker/publish_test.go index cc79507..4d18c8e 100644 --- a/internal/worker/publish_test.go +++ b/internal/worker/publish_test.go @@ -34,6 +34,11 @@ type fakeGateway struct { failCreate error byHead map[string]PullRequest requests []PullRequestRequest + // checks scripts CI: it is called once per poll with the commit being + // judged and the 1-based poll count. Nil reports no checks at all. + checks func(sha string, poll int) ([]CICheck, error) + checkPolls int + checkedSHA []string } func newFakeGateway() *fakeGateway { @@ -61,6 +66,17 @@ func (g *fakeGateway) FindOrCreatePullRequest(_ context.Context, request PullReq return pullRequest, nil } +func (g *fakeGateway) CommitChecks(_ context.Context, _ string, sha string) ([]CICheck, error) { + g.mutex.Lock() + defer g.mutex.Unlock() + g.checkPolls++ + g.checkedSHA = append(g.checkedSHA, sha) + if g.checks == nil { + return nil, nil + } + return g.checks(sha, g.checkPolls) +} + func (g *fakeGateway) counts() (created, adopted int) { g.mutex.Lock() defer g.mutex.Unlock() @@ -614,7 +630,7 @@ func TestRepublishingAProvenPushCreatesNoDuplicateBranchOrPullRequest(t *testing baseSHA: head, lease: newAttemptLease(w.client, attempt.ID, "unused-token-for-a-ledger-that-answers-first"), } - summary := w.publish(context.Background(), gateway, PublishOptions{}, target, []string{"work.txt"}) + summary := w.publish(context.Background(), gateway, PublishOptions{}, target, []string{"work.txt"}, ciPolicy{}) if !summary.Published() { t.Fatalf("re-publish summary = %+v, want published from the ledger alone", summary) } diff --git a/internal/worker/worker_integration_test.go b/internal/worker/worker_integration_test.go index fd198a6..41d980a 100644 --- a/internal/worker/worker_integration_test.go +++ b/internal/worker/worker_integration_test.go @@ -76,6 +76,12 @@ func newHarness(t *testing.T) *harness { } func (h *harness) seedRun(runID string, targets ...protocol.RunTarget) { + h.t.Helper() + h.seedRunWithSnapshot(runID, integrationSnapshot, targets...) +} + +// seedRunWithSnapshot seeds a run frozen from the given definition source. +func (h *harness) seedRunWithSnapshot(runID, snapshot string, targets ...protocol.RunTarget) { h.t.Helper() targetsJSON, err := json.Marshal(targets) if err != nil { @@ -85,13 +91,13 @@ func (h *harness) seedRun(runID string, targets ...protocol.RunTarget) { if _, err := h.db.Exec(` INSERT INTO definitions(id, name, generation, source, created_at, updated_at) VALUES (?, ?, 1, ?, ?, ?) - `, "def-"+runID, "fixture-"+runID, integrationSnapshot, now, now); err != nil { + `, "def-"+runID, "fixture-"+runID, snapshot, now, now); err != nil { h.t.Fatalf("seed definition: %v", err) } if _, err := h.db.Exec(` INSERT INTO runs(id, definition_id, definition_generation, snapshot, parameters, targets, state, created_at, updated_at) VALUES (?, ?, 1, ?, '{}', ?, 'active', ?, ?) - `, runID, "def-"+runID, integrationSnapshot, string(targetsJSON), now, now); err != nil { + `, runID, "def-"+runID, snapshot, string(targetsJSON), now, now); err != nil { h.t.Fatalf("seed run: %v", err) } } diff --git a/migrations/004_publish_ci_step.sql b/migrations/004_publish_ci_step.sql new file mode 100644 index 0000000..6d66911 --- /dev/null +++ b/migrations/004_publish_ci_step.sql @@ -0,0 +1,24 @@ +-- 004_publish_ci_step.sql — the publish ledger learns the `ci` step. +-- +-- A definition whose publish.ci waits is `accepted` only once CI is green on +-- the pull request's head, and that fact is a ledger row like every other +-- publish step. SQLite cannot alter a CHECK constraint in place, so the table +-- is rebuilt with the widened vocabulary. Nothing references publish_records, +-- so the rebuild needs no foreign-key choreography. + +CREATE TABLE publish_records_v4 ( + attempt_id TEXT NOT NULL REFERENCES attempts(id), + step TEXT NOT NULL CHECK (step IN ('push', 'pull_request', 'proof', 'ci')), + branch TEXT NOT NULL, + remote_ref TEXT NOT NULL DEFAULT '', + pr_url TEXT NOT NULL DEFAULT '', + completed_at INTEGER NOT NULL, + PRIMARY KEY (attempt_id, step) +); + +INSERT INTO publish_records_v4(attempt_id, step, branch, remote_ref, pr_url, completed_at) +SELECT attempt_id, step, branch, remote_ref, pr_url, completed_at FROM publish_records; + +DROP TABLE publish_records; + +ALTER TABLE publish_records_v4 RENAME TO publish_records;