diff --git a/cmd/orchestra-worker/main.go b/cmd/orchestra-worker/main.go index a350959..12a0ec4 100644 --- a/cmd/orchestra-worker/main.go +++ b/cmd/orchestra-worker/main.go @@ -146,8 +146,17 @@ type completionEvidence struct { Remote string `json:"remote"` QualityGate string `json:"quality_gate,omitempty"` GateExit int `json:"gate_exit"` + GateOutput string `json:"gate_output,omitempty"` CompletedAt time.Time `json:"completed_at"` } + +// tail keeps the end of a gate log, which is where the failure is. +func tail(s string, max int) string { + if len(s) <= max { + return s + } + return s[len(s)-max:] +} type workerState struct { Cursor uint64 `json:"cursor"` Sessions map[string]herdr.Session `json:"sessions"` @@ -568,11 +577,27 @@ func (w *worker) releaseReady(ctx context.Context) { log.Printf("upload completion %s: %v", id, err) continue } - if err = w.api.Complete(ctx, id, ref, evidence.ResultSHA, evidence.Branch, evidence.Remote, w.leases[id].Epoch, w.leases[id].Version, w.usageReceipt(s, w.leases[id]), w.sessionEvidence(ctx, id, s)); err != nil { - w.recordError(fmt.Errorf("complete %s: %w", id, err)) - log.Printf("complete %s: %v", id, err) + outcome, err := w.submit(ctx, id, s, evidence) + if err != nil { + w.recordError(fmt.Errorf("submit %s: %w", id, err)) + log.Printf("submit %s: %v", id, err) continue } + if outcome == federation.SubmitChangesRequested { + // The sealed review sent the work back. The task is in + // implement again, so the done marker is stale and the phase + // change rotates this session on the next tick. + _ = os.Remove(filepath.Join(s.Worktree, ".orchestra", "done")) + log.Printf("review returned %s to implementation", id) + continue + } + if outcome == federation.SubmitNoPublisher { + if err = w.api.Complete(ctx, id, ref, evidence.ResultSHA, evidence.Branch, evidence.Remote, w.leases[id].Epoch, w.leases[id].Version, w.usageReceipt(s, w.leases[id]), w.sessionEvidence(ctx, id, s)); err != nil { + w.recordError(fmt.Errorf("complete %s: %w", id, err)) + log.Printf("complete %s: %v", id, err) + continue + } + } // Completion is durable before closing the exact pane. If close // fails, retain the session mapping for a later explicit cleanup. a := herdr.CLIAdapter{Backend: w.executionBackend(), Harness: w.harness} @@ -946,19 +971,6 @@ func (w *worker) finalize(ctx context.Context, id string, s herdr.Session) (comp return completionEvidence{}, fmt.Errorf("base sha: %s: %w", base, err) } e := completionEvidence{TaskID: id, Project: t.Project, Worker: w.harnessID, Harness: w.harness, PaneID: s.PaneID, BaseSHA: strings.TrimSpace(string(base)), Remote: p.Remote, QualityGate: t.QualityGate, CompletedAt: time.Now().UTC()} - gateCommand := t.QualityGate - if gateCommand == "" { - gateCommand = p.QualityGate - } - e.QualityGate = gateCommand - if gateCommand != "" { - gate := exec.CommandContext(ctx, "sh", "-c", gateCommand) - gate.Dir = s.Worktree - if out, err := gate.CombinedOutput(); err != nil { - e.GateExit = 1 - return completionEvidence{}, fmt.Errorf("quality gate %q: %s: %w", gateCommand, out, err) - } - } if _, err := git(ctx, s.Worktree, "diff", "--quiet", "--", "TASK.md"); err != nil { return completionEvidence{}, errors.New("TASK.md was modified") } @@ -970,16 +982,34 @@ func (w *worker) finalize(ctx context.Context, id string, s herdr.Session) (comp return completionEvidence{}, fmt.Errorf("commit result: %s: %w", out, err) } } - branch, err := git(ctx, s.Worktree, "branch", "--show-current") - if err != nil || strings.TrimSpace(string(branch)) == "" { - return completionEvidence{}, fmt.Errorf("result branch: %s: %w", branch, err) - } - e.Branch = strings.TrimSpace(string(branch)) sha, err := git(ctx, s.Worktree, "rev-parse", "HEAD") if err != nil { return completionEvidence{}, fmt.Errorf("result sha: %s: %w", sha, err) } e.ResultSHA = strings.TrimSpace(string(sha)) + // The gate runs after the result commit, against a clean tree that is + // byte-for-byte the commit being submitted. Running it first bound the + // evidence to the pre-commit HEAD, which CheckSubmission rejects because + // gate sha, review sha and head sha must be one commit. + gateCommand := t.QualityGate + if gateCommand == "" { + gateCommand = p.QualityGate + } + e.QualityGate = gateCommand + if gateCommand != "" { + gate := exec.CommandContext(ctx, "sh", "-c", gateCommand) + gate.Dir = s.Worktree + out, gateErr := gate.CombinedOutput() + e.GateOutput = tail(string(out), review.MaxGateOutputBytes) + if gateErr != nil { + e.GateExit = 1 + return completionEvidence{}, fmt.Errorf("quality gate %q: %s: %w", gateCommand, out, gateErr) + } + } + // The submission branch is derived, not the worktree's local name: the + // coordinator computes "orchestra/" independently, and a pull + // request can only be opened for a branch both halves name the same way. + e.Branch = "orchestra/" + id if out, err := git(ctx, s.Worktree, "push", p.Remote, "HEAD:refs/heads/"+e.Branch); err != nil { return completionEvidence{}, fmt.Errorf("push result: %s: %w", out, err) } @@ -990,6 +1020,39 @@ func (w *worker) finalize(ctx context.Context, id string, s herdr.Session) (comp return e, nil } +// reviewFile is where the reviewing session leaves its findings. The reviewer +// supplies findings and nothing else: the commit they are bound to is the +// result commit the worker just made, which the agent cannot know and must not +// assert. +const reviewFile = "review.json" + +// submit seals the review and hands the result to the human through a pull +// request. It is the only path from a reviewed change to TaskSubmitted; the +// direct completion call remains for a project with no forge configured. +func (w *worker) submit(ctx context.Context, id string, s herdr.Session, e completionEvidence) (string, error) { + var result review.Result + b, err := os.ReadFile(filepath.Join(s.Worktree, ".orchestra", reviewFile)) + if err != nil { + // An absent review file is not a pass. Submission needs a sealed + // review, and inventing an empty one would launder "the reviewer wrote + // nothing" into "the reviewer found nothing". + return "", fmt.Errorf("review phase sealed no .orchestra/%s: %w", reviewFile, err) + } + if err := json.Unmarshal(b, &result); err != nil { + return "", fmt.Errorf(".orchestra/%s: %w", reviewFile, err) + } + result.ResultSHA = e.ResultSHA + if err := result.Validate(); err != nil { + return "", fmt.Errorf(".orchestra/%s: %w", reviewFile, err) + } + l := w.leases[id] + // The gate result is bound to the commit the gate ran against, which + // finalize guarantees is the commit being submitted. A project with no + // quality gate still produces a bound result: Passed() needs the sha. + gate := domain.GateResult{Command: e.QualityGate, ExitCode: e.GateExit, SHA: e.ResultSHA, Output: e.GateOutput} + return w.api.Submit(ctx, id, l.Epoch, l.Version, e.ResultSHA, e.Remote, result, gate) +} + // stageExclude keeps Orchestra's own control files out of the result commit. // It names the directory, not the marker inside it: .orchestra carries a // .gitignore of "*" (herdr/adapter.go), and naming an ignored file in a diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go index 70f8a60..55734d2 100644 --- a/cmd/orchestra/main.go +++ b/cmd/orchestra/main.go @@ -1378,7 +1378,7 @@ func main() { w.WriteHeader(http.StatusNoContent) }) mux.HandleFunc("/v1/federation/workers/", func(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodPost || (!strings.HasSuffix(r.URL.Path, "/heartbeat") && !strings.HasSuffix(r.URL.Path, "/renew") && !strings.HasSuffix(r.URL.Path, "/start") && !strings.HasSuffix(r.URL.Path, "/nack") && !strings.HasSuffix(r.URL.Path, "/handoff") && !strings.HasSuffix(r.URL.Path, "/pickup") && !strings.HasSuffix(r.URL.Path, "/complete") && !strings.HasSuffix(r.URL.Path, "/captures")) { + if r.Method != http.MethodPost || (!strings.HasSuffix(r.URL.Path, "/heartbeat") && !strings.HasSuffix(r.URL.Path, "/renew") && !strings.HasSuffix(r.URL.Path, "/start") && !strings.HasSuffix(r.URL.Path, "/nack") && !strings.HasSuffix(r.URL.Path, "/handoff") && !strings.HasSuffix(r.URL.Path, "/pickup") && !strings.HasSuffix(r.URL.Path, "/complete") && !strings.HasSuffix(r.URL.Path, "/submit") && !strings.HasSuffix(r.URL.Path, "/captures")) { http.Error(w, "not found", 404) return } @@ -1443,6 +1443,8 @@ func main() { FailureClass string `json:"failure_class"` LastError string `json:"last_error"` SessionEvidence domain.SessionEvidence `json:"session_evidence"` + Review review.Result `json:"review"` + Gate domain.GateResult `json:"gate"` } if json.NewDecoder(r.Body).Decode(&b) != nil || b.TaskID == "" { http.Error(w, "invalid lease body", 400) @@ -1465,10 +1467,56 @@ func main() { http.Error(w, "lease not owned", 409) return } - if strings.HasSuffix(r.URL.Path, "/complete") && b.ExpectedVersion != t.Version { + if (strings.HasSuffix(r.URL.Path, "/complete") || strings.HasSuffix(r.URL.Path, "/submit")) && b.ExpectedVersion != t.Version { http.Error(w, "lease version conflict", http.StatusConflict) return } + if strings.HasSuffix(r.URL.Path, "/submit") { + // Seal the review, then submit the commit it examined. Both events + // are Orchestra's; the worker supplies verified evidence and the + // coordinator decides what it means. + project, ok := rr.Project(t.Project) + if !ok { + http.Error(w, "unknown project "+t.Project, 409) + return + } + if len(b.ResultSHA) != 40 { + http.Error(w, "verified result_sha required", 400) + return + } + if !t.Submitted(b.ResultSHA) && (t.Review == nil || t.Review.ResultSHA != b.ResultSHA) { + b.Review.ResultSHA = b.ResultSHA + if _, err := operations.RecordReview(s, project, b.TaskID, b.Review); err != nil { + http.Error(w, err.Error(), http.StatusConflict) + return + } + if !b.Review.Accepted() { + // The review sent the work back to implementation. That is + // an outcome, not a failure, and there is nothing to submit. + json.NewEncoder(w).Encode(map[string]any{"status": federation.SubmitChangesRequested, "blocking": len(b.Review.Blocking())}) + return + } + } + if submissionPublisher == nil { + json.NewEncoder(w).Encode(map[string]string{"status": federation.SubmitNoPublisher}) + return + } + plan, err := operations.PrepareSubmission(s, project, b.TaskID, b.ResultSHA, b.Gate, operations.Notes{}) + if err != nil { + http.Error(w, err.Error(), http.StatusConflict) + return + } + if b.Remote != "" { + plan.Remote = b.Remote + } + e, err := operations.ExecuteSubmission(r.Context(), s, plan, submissionPublisher(plan), nil) + if err != nil { + http.Error(w, err.Error(), http.StatusConflict) + return + } + json.NewEncoder(w).Encode(map[string]any{"status": federation.SubmitSubmitted, "event": e}) + return + } if strings.HasSuffix(r.URL.Path, "/start") { // A lost response after append is an idempotent start ACK, not a // reason to strand the running pane behind a stale version. diff --git a/internal/federation/client.go b/internal/federation/client.go index 4d89572..b2b8caa 100644 --- a/internal/federation/client.go +++ b/internal/federation/client.go @@ -9,6 +9,7 @@ import ( "net/http" "net/url" "orchestra/internal/domain" + "orchestra/internal/review" "strings" ) @@ -291,6 +292,40 @@ func (c Client) Complete(ctx context.Context, taskID, reportRef, resultSHA, bran return err } +// Submit outcomes. Sealing a review and submitting are one call because a +// review that blocks has no submission to make, and splitting them would let a +// lost response leave a sealed review with nothing acting on it. +const ( + // SubmitSubmitted: the pull request is open and TaskSubmitted is recorded. + SubmitSubmitted = "submitted" + // SubmitChangesRequested: the review blocked and the task is back in + // implementation. + SubmitChangesRequested = "changes_requested" + // SubmitNoPublisher: the project has no forge, so the caller completes the + // task directly instead. + SubmitNoPublisher = "no_publisher" +) + +// Submit seals the review against resultSHA and submits that commit for human +// review. The coordinator owns both events; the worker supplies evidence. +func (c Client) Submit(ctx context.Context, taskID, epoch string, expectedVersion int, resultSHA, remote string, result review.Result, gate domain.GateResult) (string, error) { + resp, err := c.request(ctx, http.MethodPost, "/v1/federation/workers/"+url.PathEscape(c.WorkerID)+"/submit", map[string]any{ + "task_id": taskID, "lease_epoch": epoch, "expected_version": expectedVersion, + "result_sha": resultSHA, "remote": remote, "review": result, "gate": gate, + }) + if err != nil { + return "", err + } + defer resp.Body.Close() + var out struct { + Status string `json:"status"` + } + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + return "", err + } + return out.Status, nil +} + func (c Client) PublishCapture(ctx context.Context, capture Capture) (Capture, error) { resp, err := c.request(ctx, http.MethodPost, "/v1/federation/workers/"+url.PathEscape(c.WorkerID)+"/captures", capture) if err != nil { diff --git a/internal/provider/gitea_pr.go b/internal/provider/gitea_pr.go index 43990a8..dbeffbc 100644 --- a/internal/provider/gitea_pr.go +++ b/internal/provider/gitea_pr.go @@ -32,9 +32,18 @@ type GiteaPublisher struct { } func (g GiteaPublisher) Push(ctx context.Context, remote, branch, sha string) (string, error) { + // A worker-owned checkout lives on another machine, and it has already + // pushed the commit it is submitting. Ask the forge what the branch holds + // before reaching for a local repository the coordinator may not have: if + // the remote is already at the submitted commit, publishing is done. This + // is the same read the caller performs afterwards, so nothing is taken on + // trust that a push would have proven. + if head, err := g.branchHead(ctx, branch); err == nil && head == sha { + return head, nil + } root := g.Root if strings.TrimSpace(root) == "" { - return "", fmt.Errorf("gitea publisher: worktree root is required") + return "", fmt.Errorf("gitea publisher: %s does not hold %s and there is no local checkout to push from", branch, sha) } if out, err := exec.CommandContext(ctx, "git", "-C", root, "push", remote, sha+":refs/heads/"+branch).CombinedOutput(); err != nil { return "", fmt.Errorf("%s: %w", strings.TrimSpace(string(out)), err) @@ -52,6 +61,20 @@ func (g GiteaPublisher) Push(ctx context.Context, remote, branch, sha string) (s return fields[0], nil } +// branchHead reports the commit the forge holds for a branch, or an error if +// the branch does not exist. +func (g GiteaPublisher) branchHead(ctx context.Context, branch string) (string, error) { + var out struct { + Commit struct { + ID string `json:"id"` + } `json:"commit"` + } + if err := g.get(ctx, "/branches/"+branch, &out); err != nil { + return "", err + } + return out.Commit.ID, nil +} + type giteaPR struct { Number int `json:"number"` State string `json:"state"` diff --git a/internal/provider/gitea_pr_test.go b/internal/provider/gitea_pr_test.go new file mode 100644 index 0000000..0bb93d6 --- /dev/null +++ b/internal/provider/gitea_pr_test.go @@ -0,0 +1,47 @@ +package provider + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +// A worker-owned checkout lives on another machine and has already pushed the +// commit. The coordinator must publish by verifying the forge, not by running +// git in a directory it does not have. +func TestPushAcceptsACommitTheRemoteAlreadyHolds(t *testing.T) { + const sha = "1111111111111111111111111111111111111111" + var asked string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + asked = r.URL.Path + json.NewEncoder(w).Encode(map[string]any{"commit": map[string]string{"id": sha}}) + })) + defer srv.Close() + + // Root is deliberately empty: a push would have nowhere to run. + p := GiteaPublisher{Gitea: Gitea{BaseURL: srv.URL, Owner: "kami", Repo: "test-e2e"}} + got, err := p.Push(context.Background(), "origin", "orchestra/T1", sha) + if err != nil { + t.Fatalf("push: %v", err) + } + if got != sha { + t.Fatalf("got %q, want %q", got, sha) + } + if asked != "/api/v1/repos/kami/test-e2e/branches/orchestra/T1" { + t.Fatalf("unexpected branch read %q", asked) + } +} + +func TestPushRefusesWhenTheRemoteHoldsAnotherCommitAndThereIsNoCheckout(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + json.NewEncoder(w).Encode(map[string]any{"commit": map[string]string{"id": "2222222222222222222222222222222222222222"}}) + })) + defer srv.Close() + + p := GiteaPublisher{Gitea: Gitea{BaseURL: srv.URL, Owner: "kami", Repo: "test-e2e"}} + if _, err := p.Push(context.Background(), "origin", "orchestra/T1", "1111111111111111111111111111111111111111"); err == nil { + t.Fatal("expected a refusal, got a published commit") + } +} diff --git a/internal/review/review.go b/internal/review/review.go index ea36d57..90e836b 100644 --- a/internal/review/review.go +++ b/internal/review/review.go @@ -165,7 +165,11 @@ func Decode(b []byte) (Result, error) { // Instructions is the reviewer's whole brief. It is narrow on purpose: an open // invitation produces a list of ways the reviewer would have written it // instead, which is not review. -const Instructions = `Review the supplied diff against, in order: +// The diff is named rather than supplied: nothing populates agentctx.Evidence +// on the federated path, so a brief that said "the supplied diff" described +// material the reviewing session never received. +const Instructions = `Read the change under review with git in this worktree, +then review it against, in order: 1. the task goal and acceptance criteria 2. the human decisions and constraints 3. the repository rules @@ -180,4 +184,17 @@ defect or contradicts a decision, minor otherwise. Do not redesign the solution. Do not suggest optional refactors. Do not edit any file. Do not report style preferences unless they violate a repository -rule. You are not implementing this task and you do not decide its lifecycle.` +rule. You are not implementing this task and you do not decide its lifecycle. + +Write your findings to .orchestra/review.json before you finish, even when you +found nothing: + + {"findings": [ + {"id": "f1", "severity": "blocker", "file": "path/to/file.go", + "line": 42, "claim": "what is wrong", "evidence": "why, from the diff"} + ]} + +An empty findings list is how you report a clean review. Do not set a commit +sha: Orchestra binds the review to the commit it seals, because the commit is +not something you can observe. A review with no file is a review nobody can +act on, so the file is what finishes the phase, not the text in your pane.`