diff --git a/cmd/jig/ci_repair_test.go b/cmd/jig/ci_repair_test.go new file mode 100644 index 0000000..0f6256a --- /dev/null +++ b/cmd/jig/ci_repair_test.go @@ -0,0 +1,191 @@ +// ci_repair_test.go — one CI repair round through the real `jig worker` +// wiring (plan U5's required end-to-end test). Everything is production +// except the runtime (scripted) and the GitHub gateway (a fake): a live +// `jig serve`, workerAttemptRunner with its per-attempt trace stream and +// scratch, the publishing runner, and a real git remote. It proves what the +// U3 review found broken: the continuation's scratch (handoff notes +// included) and trace stream must outlive Execute until the round is done. +package main + +import ( + "context" + "fmt" + "io" + "log/slog" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/StructuPath/jig/internal/controlplane" + "github.com/StructuPath/jig/internal/protocol" + "github.com/StructuPath/jig/internal/worker" +) + +const ciRepairDefinition = `name: ci-repair-e2e +roster: + builder: + model: test-model + system_prompt: Build. + user_prompt: "Build." + env: [PATH] + writes: ["jig-note.txt", "jig-fix.txt"] +phases: + - {name: build, kind: agent, owner: builder} + - name: verify + kind: code + command: "test -s jig-note.txt" +acceptance: [all_phases_passed] +publish: + ci: + wait: true + on_fail: {run: build, budget: 1} +` + +// e2eGateway is the fake GitHub: CI is red on the first head it judges and +// green on the next, and every poll runs the test's probe first. +type e2eGateway struct { + mutex sync.Mutex + first string + probe func(sha string, red bool) + polled []string +} + +func (g *e2eGateway) FindOrCreatePullRequest(context.Context, worker.PullRequestRequest) (worker.PullRequest, error) { + return worker.PullRequest{Number: 1, URL: "https://github.com/example/repo/pull/1", State: "open"}, nil +} + +func (g *e2eGateway) CommitChecks(_ context.Context, _ string, sha string) ([]worker.CICheck, error) { + g.mutex.Lock() + defer g.mutex.Unlock() + g.polled = append(g.polled, sha) + if g.first == "" { + g.first = sha + } + red := sha == g.first + g.probe(sha, red) + if red { + return []worker.CICheck{{Name: "lint", Verdict: worker.CIFail, Conclusion: "failure"}}, nil + } + return []worker.CICheck{{Name: "lint", Verdict: worker.CIPass, Conclusion: "success"}}, nil +} + +func (g *e2eGateway) FailedCheckLogs(_ context.Context, _ string, checks []worker.CICheck) []worker.CICheck { + return checks +} + +func TestACIRepairRoundRunsThroughTheRealWorkerWiring(t *testing.T) { + repo := initRepo(t) + serverData, workerData := t.TempDir(), t.TempDir() + scriptRuntime(t, + map[string]any{"files": map[string]any{"jig-note.txt": "note\n"}, + "text": envelopeJSON(t, map[string]any{"status": "success", "summary": "wrote the note"})}, + map[string]any{"files": map[string]any{"jig-fix.txt": "fixed\n"}, + "text": envelopeJSON(t, map[string]any{"status": "success", "summary": "fixed lint"})}, + ) + base, stopServe := startServe(t, serverData, "--no-github-poll") + defer stopServe() + + var definition protocol.Definition + if err := apiCall(context.Background(), base, "POST", "/api/definitions", + controlplane.DefinitionInput{Source: ciRepairDefinition}, &definition); err != nil { + t.Fatal(err) + } + var view controlplane.RunView + if err := apiCall(context.Background(), base, "POST", "/api/runs", controlplane.RunInvocation{ + DefinitionID: definition.ID, Instructions: "write the note", + Targets: []controlplane.InvocationTarget{{Repository: repo}}, + }, &view); err != nil { + t.Fatal(err) + } + + // The handoff note is planted while CI is red — after Execute returned, + // before the round — and must still be there when CI judges the fix. + var probeErrors []string + note := "" + gateway := &e2eGateway{probe: func(sha string, red bool) { + handoffs, _ := filepath.Glob(filepath.Join(workerData, "scratch", "*", "handoff")) + if len(handoffs) != 1 { + probeErrors = append(probeErrors, fmt.Sprintf( + "CI poll on %.8s (red=%v) found %d scratch handoff dirs; the continuation's scratch is gone", + sha, red, len(handoffs))) + return + } + if red && note == "" { + note = filepath.Join(handoffs[0], "plan.md") + if err := os.WriteFile(note, []byte("the plan\n"), 0o600); err != nil { + probeErrors = append(probeErrors, err.Error()) + } + return + } + if !red { + if _, err := os.Stat(note); err != nil { + probeErrors = append(probeErrors, "the handoff note did not survive the round: "+err.Error()) + } + } + }} + + ctx := context.Background() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + selected, err := selectRuntime(ctx, "claude-code", true) + if err != nil { + t.Fatal(err) + } + var host *worker.Worker + publisher := worker.NewPublishingRunner( + workerAttemptRunner(ctx, workerData, selected, logger, func() *worker.Worker { return host }), + gateway, worker.PublishOptions{ + CommitAuthorName: "jig-test", CommitAuthorEmail: "jig@test", + CIPollInterval: 5 * time.Millisecond, CIRegistrationGrace: 20 * time.Millisecond, + }) + host, err = worker.New(worker.Config{ + ServerURL: base, DataDir: workerData, Capacity: 1, + Probes: []worker.RuntimeProbe{selected.Runtime}, Runner: publisher, Logger: logger, + }) + if err != nil { + t.Fatal(err) + } + publisher.Bind(host) + if _, err := host.RegisterOnce(ctx); err != nil { + t.Fatal(err) + } + + attempt, err := host.ClaimOnce(ctx) + if err != nil || attempt == nil { + t.Fatalf("claim: attempt=%v err=%v", attempt, err) + } + for _, problem := range probeErrors { + t.Error(problem) + } + if attempt.State != protocol.AttemptAccepted { + t.Fatalf("attempt = %s (%s)\n%s, want accepted after one repair round", attempt.State, attempt.Error, attempt.Result) + } + if len(gateway.polled) < 2 || note == "" { + t.Fatalf("CI polls = %d, note planted = %v: the round never ran", len(gateway.polled), note != "") + } + + // The round's trace reached the control plane: the stream was still open. + var page struct { + Events []protocol.Event `json:"events"` + } + if err := apiCall(ctx, base, "GET", "/api/attempts/"+attempt.ID+"/events?limit=500", nil, &page); err != nil { + t.Fatal(err) + } + repairStarted, acceptances := false, 0 + for _, event := range page.Events { + repairStarted = repairStarted || event.Name == "ci_repair_start" + if event.Name == "acceptance" { + acceptances++ + } + } + if !repairStarted || acceptances != 2 { + t.Fatalf("control-plane trace has ci_repair_start=%v and %d acceptance event(s), want the round's events too", + repairStarted, acceptances) + } + + // And once the attempt is over, its scratch is gone (KTD11). + if entries, _ := os.ReadDir(filepath.Join(workerData, "scratch")); len(entries) != 0 { + t.Fatalf("attempt scratch survived the attempt: %v", entries) + } +} 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 3666f2a..ebda315 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 @@ -128,10 +128,19 @@ 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`. 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. -- **Required end-to-end test:** one round must run through the real `jig worker` wiring (`cmd/jig/worker_test.go`'s scripted-runtime harness), asserting that the round's trace events reach the control plane and that the handoff notes were present during the round. U3 could only unit-test `DeferToContinuation`; this is the first place the host wiring in `cmd/jig/worker.go` can be proven. -- **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. +- **Files:** `internal/worker/publish.go` (`ciPolicy.repairBudget`, `PublishSummary.CIRepairs`, the hook in `PublishingRunner.Run`), new `internal/worker/publish_repair.go`, `internal/worker/publish_repair_test.go`, new `cmd/jig/ci_repair_test.go`. +- **Approach (as built):** `PublishingRunner.Run` enters `repairCI` only when publish ended `ci_failed`, the outcome carries a continuation, and the frozen definition declares a repair budget. Each round: + 1. reads the rounds from the ledger (`AttemptCIRepairs`), so the head jig last pushed is the last round's `head_after`, else the proof ref, never memory; + 2. stops on `ci_repair_exhausted` past the budget, and on `ci_repair_head_moved` when CI was red on a head jig did not push (KTD7); + 3. authorizes round N under a freshened lease, and refuses a `Completed` it did not list (`ci_repair_conflict`); + 4. attaches log tails (`FailedCheckLogs`) and calls `RepairCI`; + 5. stops without pushing on a round that is cancelled (`ci_repair_cancelled`), not accepted (`ci_repair_round_failed`), held (`ci_repair_held`), or unchanged (`ci_repair_no_change`); + 6. **fences again** (freshen, then re-authorize) right before the push, because a round can run for tens of minutes and a lease lost meanwhile must never push; refuses a round whose own commits carry a path it never changed (`publish_staging_escape`, the committed twin of the staging check, e.g. a definition's `git add -A` sweeping in the chain's test output); then commits and pushes the round's paths non-force, proves the remote head, records the round, and moves `remote_ref` to it; + 7. runs a fresh `ci` step on the new head, where green publishes and red loops. + + Rounds run on the same uncancelled context as publish: cancellation reaches the engine through the attempt's cancel channel and the CI wait through the lease, and git and ledger writes always finish. `PublishSummary.CIRepairs` lists every round with its heads, failed checks, changed paths, acceptance evidence, and outcome, because the attempt's top-level result describes the chain, and a pushed round is described here. Trace events come from the engine (`ci_repair_start`, then the round's phase, gate, and acceptance events); the worker has no trace sink of its own, so the per-round outcome lives in the publish summary rather than a worker event. The publish-only retry calls `publish` directly and so never repairs (R11). +- **End-to-end test (done):** `cmd/jig/ci_repair_test.go` runs one round through a live `jig serve`, `workerAttemptRunner`, the real publishing runner, a scripted runtime, and a fake gateway. It plants a handoff note while CI is red and asserts that it survives the round, that the round's events reach the control plane, and that the scratch is gone once the attempt ends. Reintroducing the U3 wiring bug fails it ("scratch is gone"). +- **Tests:** one round to green; two rounds chained through the ledger; budget exhausted; a person's push before a round; a person's push mid-round (the non-force push is rejected and nothing is recorded); held, failed, cancelled and no-change rounds push nothing; no continuation, or a continuation without `on_fail`, means plain `ci_failed`; the publish retry never repairs. ### U6. Docs and the stock factory diff --git a/internal/worker/publish.go b/internal/worker/publish.go index 42f34c6..1778f43 100644 --- a/internal/worker/publish.go +++ b/internal/worker/publish.go @@ -125,8 +125,10 @@ type PublishSummary struct { // 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"` + // CIRepairs lists the CI repair rounds this publish ran, in order. + CIRepairs []CIRepairSummary `json:"ci_repairs,omitempty"` + Performed []string `json:"performed_steps,omitempty"` + Reused []string `json:"reused_steps,omitempty"` } // Published reports whether the pipeline completed with remote proof — the @@ -235,6 +237,9 @@ func (o PublishOptions) ciRegistrationGrace() time.Duration { type ciPolicy struct { wait bool timeout time.Duration + // repairBudget is publish.ci.on_fail's budget; zero means red CI ends + // the attempt, as it always has. + repairBudget int } // ciPolicyFor reads the policy out of a run snapshot. A snapshot that does @@ -246,7 +251,11 @@ func ciPolicyFor(snapshot string) ciPolicy { if err != nil { return ciPolicy{} } - return ciPolicy{wait: spec.WaitsForCI(), timeout: spec.CITimeout()} + policy := ciPolicy{wait: spec.WaitsForCI(), timeout: spec.CITimeout()} + if policy.wait && spec.Publish.CI.OnFail != nil { + policy.repairBudget = spec.Publish.CI.OnFail.Budget + } + return policy } func (o PublishOptions) authorName() string { @@ -349,8 +358,16 @@ func (r *PublishingRunner) Run(ctx context.Context, prepared *PreparedAttempt) O // The critical section: an uncancellable context, and the attempt's // cancellation signal deliberately unread. Every command inside still // carries its own timeout, so "uncancellable" is not "unbounded". + ci := ciPolicyFor(prepared.Claim.Snapshot) summary := worker.publish(context.WithoutCancel(ctx), r.gateway, r.options, - target, changedPathsFromResult(outcome.Result), ciPolicyFor(prepared.Claim.Snapshot)) + target, changedPathsFromResult(outcome.Result), ci) + if summary.Code == "ci_failed" && outcome.Continuation != nil && ci.repairBudget > 0 { + // Rounds run on the same uncancelled context: cancellation reaches + // the engine through the attempt's cancel channel and the CI wait + // through the lease, while git and the ledger always finish. + summary = worker.repairCI(context.WithoutCancel(ctx), r.gateway, r.options, + target, outcome.Continuation, summary, ci) + } return publishedOutcome(outcome, summary) } diff --git a/internal/worker/publish_repair.go b/internal/worker/publish_repair.go new file mode 100644 index 0000000..79abae3 --- /dev/null +++ b/internal/worker/publish_repair.go @@ -0,0 +1,261 @@ +// publish_repair.go — the CI repair round loop (publish.ci.on_fail, plan +// U5). When publish ends red (`ci_failed`) and the engine kept its chain +// alive, the publishing runner repairs inside the attempt, one round at a +// time, until CI is green, the declared budget runs out, or a round ends any +// other way: +// +// red CI → head still the one jig pushed? → authorize round N (fenced) +// → RepairCI: the on_fail phase and every phase after it, judged +// again by acceptance and the publish hold +// → commit and push the round's paths, non-force (critical section) +// → record round N → prove the remote head → wait for CI again +// +// Every way a round can stop short of green leaves the attempt exactly where +// red CI leaves it today: accepted_unpublished, the pull request at the last +// head jig pushed, and the reason in the publish summary. The publish-only +// retry never enters this loop (R11): it judges CI and nothing more. +package worker + +import ( + "context" + "encoding/json" + "fmt" + "strings" + + "github.com/StructuPath/jig/internal/protocol" +) + +// CIRepairSummary is one repair round as the publish summary reports it. +type CIRepairSummary struct { + Round int `json:"round"` + HeadBefore string `json:"head_before"` + HeadAfter string `json:"head_after,omitempty"` + FailedChecks []string `json:"failed_checks,omitempty"` + // ChangedPaths and Acceptance are the round's own: what it changed and + // the acceptance evidence it was pushed on. The attempt's top-level + // result describes the chain; a pushed round is described here. + ChangedPaths []string `json:"changed_paths,omitempty"` + Acceptance json.RawMessage `json:"acceptance,omitempty"` + // Outcome is pushed, or the code the round stopped on. + Outcome string `json:"outcome"` + Detail string `json:"detail,omitempty"` +} + +// ciRepairCodes are the publish summary codes a repair can end on, beyond +// the CI wait's own (ci_failed, ci_timeout, ci_unavailable, ...). +const ( + ciRepairPushed = "pushed" + ciRepairExhausted = "ci_repair_exhausted" + ciRepairHeadMoved = "ci_repair_head_moved" + ciRepairFailed = "ci_repair_round_failed" + ciRepairCancelled = "ci_repair_cancelled" + ciRepairHeld = "ci_repair_held" + ciRepairNoChange = "ci_repair_no_change" +) + +// repairCI runs rounds while CI is red and budget remains. summary is the +// publish summary that ended on ci_failed; the returned summary is final. +// ctx must not be cancelled out from under git: cancellation reaches the +// engine through the attempt's cancel channel, exactly as it reaches the +// chain. +func (w *Worker) repairCI( + ctx context.Context, + gateway PullRequestGateway, + options PublishOptions, + target publishTarget, + continuation Continuation, + summary PublishSummary, + ci ciPolicy, +) PublishSummary { + for summary.Code == "ci_failed" { + rounds, err := w.client.AttemptCIRepairs(ctx, target.attemptID) + if err != nil { + return failSummary(summary, publishCode(err), "CI repair ledger unavailable: "+err.Error()) + } + // The head jig last pushed comes from the ledger, not from memory: + // the last round's head_after, else the proof ref (U2 review). + pushed := summary.RemoteRef + if len(rounds) > 0 { + pushed = rounds[len(rounds)-1].HeadAfter + } + round := len(rounds) + 1 + if round > ci.repairBudget { + return failSummary(summary, ciRepairExhausted, fmt.Sprintf( + "CI is still red on %s after %d repair round(s), the definition's budget; %s", + shortSHA(summary.CIRef), ci.repairBudget, summary.Detail)) + } + red := summary.CIRef + if red != pushed { + // KTD7: someone else pushed to the branch. jig does not stack an + // agent's commits on a person's; the publish-only retry judges + // whatever they pushed. + return failSummary(summary, ciRepairHeadMoved, fmt.Sprintf( + "CI is red on %s, but jig last pushed %s: someone else pushed to the branch, "+ + "so no repair round runs", shortSHA(red), shortSHA(pushed))) + } + names := checkNames(summary.CIFailures) + entry := CIRepairSummary{Round: round, HeadBefore: red, FailedChecks: names} + + if err := w.authorizeRound(ctx, target, round, red); err != nil { + return failRound(summary, entry, publishCode(err), err.Error()) + } + + failure := CIFailure{Head: red, Checks: gateway.FailedCheckLogs(ctx, target.repository, summary.CIFailures)} + repaired := continuation.RepairCI(ctx, failure) + entry.ChangedPaths = changedPathsFromResult(repaired.Result) + entry.Acceptance = acceptanceFromResult(repaired.Result) + switch { + case repaired.State == protocol.AttemptCancelled: + return failRound(summary, entry, ciRepairCancelled, fmt.Sprintf( + "the job was cancelled during round %d; nothing was pushed", round)) + case repaired.State != protocol.AttemptAcceptedUnpublished: + return failRound(summary, entry, ciRepairFailed, fmt.Sprintf( + "round %d ended %s: %s", round, repaired.State, repaired.Error)) + case repaired.PublishHold != "": + return failRound(summary, entry, ciRepairHeld, fmt.Sprintf( + "round %d's fix tripped the publish hold (%s); nothing was pushed", round, repaired.PublishHold)) + case repaired.Error != "": + return failRound(summary, entry, ciRepairNoChange, repaired.Error) + } + + // The round may have run for a long time: fence again right before + // the irreversible push, so an attempt whose lease was lost (and whose + // job may have a successor by now) never pushes (R6, KTD6). + if err := w.authorizeRound(ctx, target, round, red); err != nil { + return failRound(summary, entry, publishCode(err), err.Error()) + } + if err := w.requireRoundCommitsDeclared(ctx, target, red, entry.ChangedPaths); err != nil { + return failRound(summary, entry, publishCode(err), err.Error()) + } + head, err := w.pushAttemptBranch(ctx, options, target, entry.ChangedPaths) + if err == nil && head == red { + err = publishFailure(ciRepairNoChange, "round %d produced no new commit", round) + } + if err != nil { + return failRound(summary, entry, publishCode(err), err.Error()) + } + entry.HeadAfter = head + if err := w.verifyRemoteRef(ctx, target, target.branch, head); err != nil { + return failRound(summary, entry, publishCode(err), err.Error()) + } + if _, err := w.client.RecordCIRepair(ctx, target.attemptID, protocol.CIRepairRecordRequest{ + LeaseToken: target.lease.token, Round: round, Branch: target.branch, + HeadBefore: red, HeadAfter: head, FailedChecks: names, + }); err != nil { + return failRound(summary, entry, publishCode(err), err.Error()) + } + entry.Outcome = ciRepairPushed + summary.CIRepairs = append(summary.CIRepairs, entry) + // The remote now holds the round's head: that is what was pushed. + summary.RemoteRef = head + + // A fresh ci step on the new head: green is recorded and publishes; + // red loops; anything else ends the attempt as the CI wait does. + summary.CIFailures = nil + green, err := w.publishStep(ctx, target, protocol.PublishStepCI, map[string]protocol.PublishRecord{}, &summary, + func(authorization protocol.PublishAuthorization) (protocol.PublishStepRequest, error) { + ciHead, failures, err := w.awaitCI(ctx, gateway, options, target, authorization.Branch, ci.timeout) + summary.CIRef = ciHead + summary.CIFailures = failures + if err != nil { + return protocol.PublishStepRequest{}, err + } + return protocol.PublishStepRequest{Step: protocol.PublishStepCI, + Branch: authorization.Branch, RemoteRef: ciHead}, nil + }) + if err == nil { + summary.CIRef = green.RemoteRef + summary.State, summary.Code, summary.Detail = PublishStatePublished, "", "" + return summary + } + } + return summary +} + +// authorizeRound fences one round under a freshened lease: the heartbeat +// first, because only it can revive an expired-but-unswept lease (R5), then +// the ledger's authorization. It runs before the round and again right +// before the push. Either way the round must not be on the ledger yet: the +// loop listed no round N, so one appearing is another publisher's, never +// ours to adopt blind. +func (w *Worker) authorizeRound(ctx context.Context, target publishTarget, round int, red string) error { + if err := target.lease.freshen(ctx); err != nil { + return fmt.Errorf("lease could not be freshened: %w", err) + } + authorization, err := w.client.AuthorizeCIRepair(ctx, target.attemptID, protocol.CIRepairAuthorizationRequest{ + LeaseToken: target.lease.token, Round: round, Branch: target.branch, HeadBefore: red, + }) + if err != nil { + return err + } + if authorization.Completed != nil { + return publishFailure("ci_repair_conflict", "round %d was recorded concurrently", round) + } + return nil +} + +// requireRoundCommitsDeclared refuses a round whose own commits (a +// definition's commit phase runs `git add -A`) carry a path the round never +// changed — a test artifact the chain left untracked, say. Uncommitted +// paths get the same check from commitChangedPaths; this is its twin for +// what the round committed itself. +func (w *Worker) requireRoundCommitsDeclared(ctx context.Context, target publishTarget, red string, declared []string) error { + committed, err := runGit(ctx, target.worktreePath, "diff", "--name-only", "-z", red, "HEAD") + if err != nil { + return publishFailure("publish_staging_unreadable", "%s", err.Error()) + } + allowed := make(map[string]bool, len(declared)) + for _, path := range declared { + allowed[path] = true + } + var escaped []string + for _, path := range strings.Split(committed, "\x00") { + if path != "" && !allowed[path] { + escaped = append(escaped, path) + } + } + if len(escaped) > 0 { + return publishFailure("publish_staging_escape", + "the repair round committed %d path(s) it never changed (%s); publish refuses to push them", + len(escaped), strings.Join(escaped, ", ")) + } + return nil +} + +// acceptanceFromResult is the acceptance evidence in an engine result, or +// nil. +func acceptanceFromResult(result string) json.RawMessage { + var parsed struct { + Acceptance json.RawMessage `json:"acceptance"` + } + if json.Unmarshal([]byte(result), &parsed) != nil { + return nil + } + return parsed.Acceptance +} + +func failSummary(summary PublishSummary, code, detail string) PublishSummary { + summary.State = PublishStateFailed + summary.Code = code + summary.Detail = boundedText(detail, protocol.MaxPublishDiagnosticBytes) + return summary +} + +// failRound records the round that stopped and ends the summary on it. +func failRound(summary PublishSummary, entry CIRepairSummary, code, detail string) PublishSummary { + entry.Outcome = code + entry.Detail = boundedText(detail, protocol.MaxPublishDiagnosticBytes) + summary.CIRepairs = append(summary.CIRepairs, entry) + return failSummary(summary, code, detail) +} + +func checkNames(checks []CICheck) []string { + names := make([]string, 0, len(checks)) + for _, check := range checks { + if len(names) == protocol.MaxCIRepairFailedChecks { + break + } + names = append(names, boundedText(check.Name, protocol.MaxCIRepairCheckNameBytes)) + } + return names +} diff --git a/internal/worker/publish_repair_test.go b/internal/worker/publish_repair_test.go new file mode 100644 index 0000000..fe439cc --- /dev/null +++ b/internal/worker/publish_repair_test.go @@ -0,0 +1,385 @@ +// publish_repair_test.go — the CI repair round loop (plan U5) through the +// real worker, control plane, and git remotes, with a scripted continuation +// standing in for the engine and a fake gh for CI. +package worker + +import ( + "context" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/StructuPath/jig/internal/protocol" +) + +// repairLoopSnapshot waits for CI and repairs it with up to `budget` rounds. +func repairLoopSnapshot(budget int) string { + return integrationSnapshot + fmt.Sprintf(` - name: review + kind: agent + owner: builder +publish: + ci: + wait: true + on_fail: {run: build, budget: %d} +`, budget) +} + +// roundScript is what one scripted repair round does: files to write into +// the worktree, an outcome override, and an action to run first (a person +// pushing mid-round, say). +type roundScript struct { + files map[string]string + before func() + result func(paths []string) Outcome +} + +type scriptedContinuation struct { + t *testing.T + worktree string + rounds []roundScript + failures []CIFailure + released int +} + +func (c *scriptedContinuation) RepairCI(_ context.Context, failure CIFailure) Outcome { + c.failures = append(c.failures, failure) + if len(c.failures) > len(c.rounds) { + c.t.Errorf("repair round %d was not scripted", len(c.failures)) + return Outcome{State: protocol.AttemptFailed, Error: "unscripted round"} + } + script := c.rounds[len(c.failures)-1] + if script.before != nil { + script.before() + } + paths := make([]string, 0, len(script.files)) + for name, body := range script.files { + if err := os.WriteFile(filepath.Join(c.worktree, name), []byte(body), 0o644); err != nil { + c.t.Errorf("write %s: %v", name, err) + } + paths = append(paths, name) + } + if script.result != nil { + return script.result(paths) + } + return Outcome{State: protocol.AttemptAcceptedUnpublished, Result: engineResult(c.t, paths)} +} + +func (c *scriptedContinuation) Release() { c.released++ } + +// repairRunner is the inner runner: it writes the chain's work, and hands +// back the scripted continuation bound to this attempt's worktree. +func repairRunner(t *testing.T, continuation *scriptedContinuation) AttemptRunner { + return RunnerFunc(func(_ context.Context, attempt *PreparedAttempt) Outcome { + continuation.worktree = attempt.WorktreePath + if err := os.WriteFile(filepath.Join(attempt.WorktreePath, "work.txt"), []byte("work\n"), 0o644); err != nil { + t.Errorf("write work: %v", err) + } + return Outcome{State: protocol.AttemptAcceptedUnpublished, + Result: engineResult(t, []string{"work.txt"}), Continuation: continuation} + }) +} + +// redUntil scripts CI: every head in red fails "test", anything else is +// green. The first head judged is added to red when redFirst is set, so the +// chain's own push is red. +type ciScript struct { + red map[string]bool + judged []string +} + +func (c *ciScript) gateway(redPushes int) *fakeGateway { + gateway := newFakeGateway() + gateway.checks = func(sha string, _ int) ([]CICheck, error) { + if len(c.judged) == 0 || c.judged[len(c.judged)-1] != sha { + c.judged = append(c.judged, sha) + if len(c.judged) <= redPushes { + c.red[sha] = true + } + } + if c.red[sha] { + return []CICheck{check("lint", CIPass), check("test", CIFail)}, nil + } + return []CICheck{check("lint", CIPass), check("test", CIPass)}, nil + } + return gateway +} + +type repairScenario struct { + h *harness + originDir string + job protocol.Job + branch string + w *Worker + gateway *fakeGateway + ci *ciScript + continuation *scriptedContinuation +} + +func newRepairScenario(t *testing.T, budget, redPushes int, rounds ...roundScript) *repairScenario { + t.Helper() + s := &repairScenario{h: newHarness(t), ci: &ciScript{red: map[string]bool{}}} + var head, identity string + s.originDir, head, identity = newOriginRepo(t) + s.h.seedRunWithSnapshot("run-1", repairLoopSnapshot(budget), protocol.RunTarget{Repository: identity, BaseSHA: head}) + s.job = s.h.enqueue("run-1", identity) + s.branch = protocol.PublishBranch(s.job.ID, 1) + s.continuation = &scriptedContinuation{t: t, rounds: rounds} + s.gateway = s.ci.gateway(redPushes) + s.w = newCIWorker(t, s.h, repairRunner(t, s.continuation), s.gateway) + return s +} + +func (s *repairScenario) run(t *testing.T) (*protocol.Attempt, PublishSummary) { + t.Helper() + attempt, err := s.w.ClaimOnce(context.Background()) + if err != nil || attempt == nil { + t.Fatalf("claim: attempt=%v err=%v", attempt, err) + } + return attempt, publishSummaryOf(t, attempt.Result) +} + +func (s *repairScenario) rounds(t *testing.T, attemptID string) []protocol.CIRepairRecord { + t.Helper() + rounds, err := s.w.client.AttemptCIRepairs(context.Background(), attemptID) + if err != nil { + t.Fatalf("read rounds: %v", err) + } + return rounds +} + +func fixRound() roundScript { return roundScript{files: map[string]string{"fix.txt": "fixed\n"}} } + +// Red, one round, green: the attempt is accepted, the round is on the ledger +// from the proof head to the fix, and CI was recorded green on the fix. +func TestRedCIIsRepairedInOneRoundAndTheAttemptIsAccepted(t *testing.T) { + s := newRepairScenario(t, 2, 1, fixRound()) + attempt, summary := s.run(t) + if attempt.State != protocol.AttemptAccepted || !summary.Published() { + t.Fatalf("attempt = %s summary = %+v, want accepted and published", attempt.State, summary) + } + fixed := remoteBranches(t, s.originDir)[s.branch] + proof := s.ci.judged[0] + rounds := s.rounds(t, attempt.ID) + if len(rounds) != 1 || rounds[0].HeadBefore != proof || rounds[0].HeadAfter != fixed || + len(rounds[0].FailedChecks) != 1 || rounds[0].FailedChecks[0] != "test" { + t.Fatalf("rounds = %+v, want one from the proof head %s to the fix %s failing [test]", rounds, proof, fixed) + } + if summary.CIRef != fixed || summary.RemoteRef != fixed || len(summary.CIRepairs) != 1 || + summary.CIRepairs[0].Outcome != "pushed" { + t.Fatalf("summary = %+v, want green on the fix, remote_ref at the fix, one pushed round", summary) + } + if paths := summary.CIRepairs[0].ChangedPaths; len(paths) != 1 || paths[0] != "fix.txt" || + !strings.Contains(string(summary.CIRepairs[0].Acceptance), `"passed":true`) { + t.Fatalf("round evidence = %+v, want its changed paths and acceptance", summary.CIRepairs[0]) + } + if len(s.continuation.failures) != 1 || s.continuation.failures[0].Head != proof || + s.continuation.failures[0].Checks[0].Name != "test" { + t.Fatalf("the round was handed %+v, want the red head and the red check", s.continuation.failures) + } + if steps := recordedSteps(t, s.w, attempt.ID); !strings.Contains(strings.Join(steps, ","), "ci") { + t.Fatalf("ledger steps = %v, want ci recorded", steps) + } + if created, _ := s.gateway.counts(); created != 1 { + t.Fatalf("pull requests created = %d, want the one pull request, fixed in place", created) + } +} + +// Red after every round until the budget runs out: accepted_unpublished, +// every round on the ledger, the pull request at the last fix. +func TestRedCIPastTheBudgetEndsAcceptedUnpublished(t *testing.T) { + s := newRepairScenario(t, 1, 2, fixRound()) + attempt, summary := s.run(t) + if attempt.State != protocol.AttemptAcceptedUnpublished || summary.Code != "ci_repair_exhausted" { + t.Fatalf("attempt = %s summary = %+v, want accepted_unpublished on ci_repair_exhausted", attempt.State, summary) + } + rounds := s.rounds(t, attempt.ID) + fixed := remoteBranches(t, s.originDir)[s.branch] + if len(rounds) != 1 || rounds[0].HeadAfter != fixed || summary.CIRef != fixed { + t.Fatalf("rounds = %+v ci_ref = %s, want one round and red CI judged on its fix %s", rounds, summary.CIRef, fixed) + } + if !strings.Contains(summary.Detail, "1 repair round") { + t.Fatalf("detail = %q, want the budget named", summary.Detail) + } +} + +// Someone else's push before the first round means no round at all (KTD7). +func TestAPersonsPushBeforeARoundMeansNoRound(t *testing.T) { + s := newRepairScenario(t, 2, 2) + var personal string + checks := s.gateway.checks + s.gateway.checks = func(sha string, poll int) ([]CICheck, error) { + if poll == 1 { + personal = pushToBranch(t, s.originDir, s.branch) + return []CICheck{check("test", CIPending)}, nil + } + return checks(sha, poll) + } + attempt, summary := s.run(t) + if attempt.State != protocol.AttemptAcceptedUnpublished || summary.Code != "ci_repair_head_moved" { + t.Fatalf("attempt = %s summary = %+v, want ci_repair_head_moved", attempt.State, summary) + } + if len(s.continuation.failures) != 0 || len(s.rounds(t, attempt.ID)) != 0 { + t.Fatal("a round ran on top of a person's push") + } + if remoteBranches(t, s.originDir)[s.branch] != personal { + t.Fatal("the person's push was overwritten") + } +} + +// A person pushing while a round runs makes jig's non-force push fail: the +// round ends without a record, and the person's commit stands. +func TestAPushRejectedMidRoundEndsCleanly(t *testing.T) { + var personal string + s := newRepairScenario(t, 2, 1) + s.continuation.rounds = []roundScript{{files: map[string]string{"fix.txt": "fixed\n"}, + before: func() { personal = pushToBranch(t, s.originDir, s.branch) }}} + attempt, summary := s.run(t) + if attempt.State != protocol.AttemptAcceptedUnpublished || summary.Code != "publish_push_failed" { + t.Fatalf("attempt = %s summary = %+v, want publish_push_failed", attempt.State, summary) + } + if len(s.rounds(t, attempt.ID)) != 0 || remoteBranches(t, s.originDir)[s.branch] != personal { + t.Fatal("a rejected push was recorded, or the person's commit was overwritten") + } + if len(summary.CIRepairs) != 1 || summary.CIRepairs[0].Outcome != "publish_push_failed" { + t.Fatalf("ci_repairs = %+v, want the round that stopped on the push", summary.CIRepairs) + } +} + +// A round that ends held, failed, cancelled, or with no change pushes +// nothing: the pull request stays at the head CI found red. +func TestARoundThatDoesNotEndAcceptedPushesNothing(t *testing.T) { + for _, tc := range []struct { + name, code string + result func(paths []string) Outcome + }{ + {"held", "ci_repair_held", func([]string) Outcome { + return Outcome{State: protocol.AttemptAcceptedUnpublished, PublishHold: "risk = high"} + }}, + {"failed", "ci_repair_round_failed", func([]string) Outcome { + return Outcome{State: protocol.AttemptFailed, Error: "acceptance predicate failed"} + }}, + {"cancelled", "ci_repair_cancelled", func([]string) Outcome { + return Outcome{State: protocol.AttemptCancelled, Error: "cancelled between phases"} + }}, + {"no change", "ci_repair_no_change", func([]string) Outcome { + return Outcome{State: protocol.AttemptAcceptedUnpublished, + Error: "ci_repair_no_change: the CI repair round changed no files"} + }}, + } { + t.Run(tc.name, func(t *testing.T) { + s := newRepairScenario(t, 2, 1, roundScript{files: map[string]string{"fix.txt": "x\n"}, result: tc.result}) + attempt, summary := s.run(t) + if attempt.State != protocol.AttemptAcceptedUnpublished || summary.Code != tc.code { + t.Fatalf("attempt = %s summary = %+v, want %s", attempt.State, summary, tc.code) + } + if remoteBranches(t, s.originDir)[s.branch] != summary.RemoteRef || len(s.rounds(t, attempt.ID)) != 0 { + t.Fatal("a round that did not end accepted moved the branch or was recorded") + } + }) + } +} + +// Without a continuation (the engine could not keep the chain alive), red +// CI ends the attempt exactly as it did before repair existed. +func TestRedCIWithoutAContinuationEndsAsItAlwaysHas(t *testing.T) { + s := newRepairScenario(t, 2, 1) + s.w.config.Runner.(*PublishingRunner).inner = writeAndDeclare(t, map[string]string{"work.txt": "work\n"}) + attempt, summary := s.run(t) + if attempt.State != protocol.AttemptAcceptedUnpublished || summary.Code != "ci_failed" || len(summary.CIRepairs) != 0 { + t.Fatalf("attempt = %s summary = %+v, want plain ci_failed", attempt.State, summary) + } +} + +// The publish-only retry judges CI and never repairs (R11), even after the +// rounds ran out. +func TestThePublishRetryNeverRepairs(t *testing.T) { + s := newRepairScenario(t, 1, 3, fixRound()) + attempt, _ := s.run(t) + if attempt.State != protocol.AttemptAcceptedUnpublished { + t.Fatalf("attempt = %s, want accepted_unpublished", attempt.State) + } + retried, err := s.w.RetryPublish(context.Background(), s.job.ID, s.gateway, fastCIOptions()) + if err != nil { + t.Fatalf("publish retry: %v", err) + } + summary := publishSummaryOf(t, retried.Result) + if retried.State != protocol.AttemptAcceptedUnpublished || summary.Code != "ci_failed" || + len(summary.CIRepairs) != 0 || len(s.continuation.failures) != 1 { + t.Fatalf("retry = %s summary = %+v rounds run = %d, want red judged and no new round", + retried.State, summary, len(s.continuation.failures)) + } +} + +// Two rounds chain: round 2 repairs round 1's fix, which the worker learns +// from the ledger, and the second fix goes green. +func TestTwoRoundsChainOnTheLedgerUntilGreen(t *testing.T) { + s := newRepairScenario(t, 2, 2, fixRound(), roundScript{files: map[string]string{"fix2.txt": "better\n"}}) + attempt, summary := s.run(t) + if attempt.State != protocol.AttemptAccepted { + t.Fatalf("attempt = %s summary = %+v, want accepted after two rounds", attempt.State, summary) + } + rounds := s.rounds(t, attempt.ID) + if len(rounds) != 2 || rounds[1].HeadBefore != rounds[0].HeadAfter || + rounds[1].HeadAfter != remoteBranches(t, s.originDir)[s.branch] { + t.Fatalf("rounds = %+v, want round 2 to repair round 1's fix and end at the remote head", rounds) + } + if len(s.continuation.failures) != 2 || s.continuation.failures[1].Head != rounds[0].HeadAfter { + t.Fatalf("round 2 was handed %+v, want round 1's red fix", s.continuation.failures) + } +} + +// A continuation under a definition that declares no repair is never used: +// red CI ends as it always has. +func TestAContinuationWithoutOnFailIsNeverUsed(t *testing.T) { + s := newRepairScenario(t, 1, 1, fixRound()) + if _, err := s.h.db.Exec(`UPDATE runs SET snapshot = ? WHERE id = 'run-1'`, ciSnapshot); err != nil { + t.Fatal(err) + } + attempt, summary := s.run(t) + if attempt.State != protocol.AttemptAcceptedUnpublished || summary.Code != "ci_failed" || len(s.continuation.failures) != 0 { + t.Fatalf("attempt = %s summary = %+v rounds = %d, want plain ci_failed and no round", + attempt.State, summary, len(s.continuation.failures)) + } +} + +// A round whose lease is lost while it runs (the job swept, a successor +// leased) never pushes: the fence runs again right before the push. +func TestARoundThatLosesItsLeaseNeverPushes(t *testing.T) { + s := newRepairScenario(t, 2, 1) + s.continuation.rounds = []roundScript{{files: map[string]string{"fix.txt": "fixed\n"}, before: func() { + if _, err := s.h.db.Exec(`UPDATE attempts SET lease_digest = x'00'`); err != nil { + t.Errorf("supersede the lease: %v", err) + } + }}} + attempt, err := s.w.ClaimOnce(context.Background()) + if err == nil && attempt != nil && attempt.State == protocol.AttemptAccepted { + t.Fatalf("attempt = %+v, a zombie was accepted", attempt) + } + if head := remoteBranches(t, s.originDir)[s.branch]; head != s.ci.judged[0] { + t.Fatalf("remote head = %s, want the red head %s: a zombie round pushed", head, s.ci.judged[0]) + } +} + +// A round that commits a path it never changed (a definition's commit +// phase sweeping in the chain's test output) is refused before the push. +func TestARoundCommittingUndeclaredPathsIsNotPushed(t *testing.T) { + s := newRepairScenario(t, 2, 1) + s.continuation.rounds = []roundScript{{files: map[string]string{"fix.txt": "fixed\n"}, before: func() { + worktree := s.continuation.worktree + if err := os.WriteFile(filepath.Join(worktree, "coverage.out"), []byte("mode: set\n"), 0o644); err != nil { + t.Errorf("write artifact: %v", err) + } + gitRun(t, worktree, "add", "coverage.out") + gitRun(t, worktree, "-c", "user.name=jig", "-c", "user.email=jig@test", "commit", "-q", "-m", "sweep") + }}} + attempt, summary := s.run(t) + if attempt.State != protocol.AttemptAcceptedUnpublished || summary.Code != "publish_staging_escape" || + !strings.Contains(summary.Detail, "coverage.out") { + t.Fatalf("attempt = %s summary = %+v, want publish_staging_escape naming coverage.out", attempt.State, summary) + } + if remoteBranches(t, s.originDir)[s.branch] != s.ci.judged[0] || len(s.rounds(t, attempt.ID)) != 0 { + t.Fatal("a round with an undeclared committed path was pushed or recorded") + } +}