kami/orchestra
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Give the reviewed change a path to the human

Browse Source 
The completion tail ended at TaskCompleted with no pull request. Nothing in
the running system ever called the review or submission endpoints: the whole
event log holds zero ReviewRecorded and zero TaskSubmitted, so the merge
reflection, the publisher and the human trust boundary had no entry point.

Four links, in the order the tail needs them:

- finalize commits first and runs the quality gate against the committed
  tree, so the gate result is bound to the commit being submitted.
  CheckSubmission requires gate sha, review sha and head sha to be one
  commit, which a gate run on the pre-commit tree can never satisfy.
- The worker seals the reviewer's findings and submits, through a new
  /v1/federation/workers/<id>/submit. A blocking review returns the task to
  implementation instead; a project with no forge still completes directly.
- The reviewing session is told where findings go. The brief asked for
  findings and named no file, and it described a diff nobody supplied.
- GiteaPublisher.Push asks the forge what the branch holds before reaching
  for a local checkout. A worker-owned worktree is on another machine and
  has already pushed the commit; the coordinator has no such directory.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CVbaKucEYBjMqVeUgJUsc1
This commit is contained in:
kami
2026-08-28 10:28:50 +04:00
parent d9a5a61965
commit e8d04d719d
6 changed files with 259 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/orchestra-worker/main.go
+84 -21
View File
@@ -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/<task>" 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
cmd/orchestra/main.go
+50 -2
View File
@@ -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.
internal/federation/client.go
+35
View File
@@ -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 {
internal/provider/gitea_pr.go
+24 -1
View File
@@ -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"`
internal/provider/gitea_pr_test.go
+47
View File
@@ -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")
}
}
internal/review/review.go
+19 -2
View File
@@ -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.`