From 2cecbc4015e68079eac5ad9933ae60b897c83284 Mon Sep 17 00:00:00 2001 From: kami Date: Tue, 28 Jul 2026 16:17:18 +0400 Subject: [PATCH] Add federation worker and canonical handoffs --- AUDIT.md | 33 +++ cmd/orchestra-worker/main.go | 357 ++++++++++++++++++++++++ cmd/orchestra-worker/main_test.go | 151 ++++++++++ cmd/orchestra/main.go | 133 ++++++++- cmd/orchestra/main_test.go | 35 +++ deploy/orchestra-worker.service | 15 + deploy/orchestra.env.example | 12 + internal/continuity/continuity.go | 64 +++-- internal/continuity/continuity_test.go | 23 +- internal/domain/domain.go | 10 +- internal/federation/client.go | 183 ++++++++++++ internal/federation/client_test.go | 51 ++++ internal/federation/federation.go | 17 ++ internal/herdr/adapter.go | 202 ++++++++++++-- internal/herdr/adapter_test.go | 164 ++++++++--- internal/herdr/herdr.go | 16 +- internal/herdr/herdr_test.go | 11 + internal/integration/end_to_end_test.go | 8 +- internal/orchestrator/orchestrator.go | 5 + internal/orchestrator/rotation_test.go | 6 +- internal/store/store.go | 7 +- internal/store/store_test.go | 34 +++ 22 files changed, 1429 insertions(+), 108 deletions(-) create mode 100644 cmd/orchestra-worker/main.go create mode 100644 cmd/orchestra-worker/main_test.go create mode 100644 cmd/orchestra/main_test.go create mode 100644 deploy/orchestra-worker.service create mode 100644 internal/federation/client.go create mode 100644 internal/federation/client_test.go diff --git a/AUDIT.md b/AUDIT.md index d2fba79..d4f044a 100644 --- a/AUDIT.md +++ b/AUDIT.md @@ -889,3 +889,36 @@ implemented when nothing can reach it. Everything else audited (provider layer, continuity schema/validation, router matching, delivery, federation registration primitives) was spot-checked against the code and its tests and matched described behavior. + +--- + +## Handoff provenance correction (2026-07-28) + +The previous rotation path had a **primacy inversion**: the harness wrote a +free-form Markdown “semantic report”, while `canonicalHandoff` fabricated +the canonical fields around it (`goal`, `done_when`, a circular action and a +`cat` command) and placed the entire report in one `remaining` element. That +made the apparent schema a wrapper around prose rather than an authoritative +handoff. + +The release path now asks the harness for only six labelled, bounded answers: +`NEXT`, `WHY`, `REMAINING`, `DEAD ENDS`, `OPEN Q`, and `LEARNED`, each with a +`NONE` escape. `canonicalHandoff` parses those answers and derives only +worker-owned facts (Git anchor, dirty state, metadata, and last observed +harness command). It never invents task intent. `goal` and `done_when` were +removed from `continuity.Handoff`; pickup must obtain task scope and success +criteria from immutable `TASK.md`. + +`continuity.Handoff.Validate` is the shared semantic gate for both release +and pickup. Strict decoding rejects removed/unknown fields, and validation +rejects circular handoff actions, commands that point at report/handoff +files, oversize or Markdown-smuggled authored list items, and malformed dead +ends. A failed release remains refused and is retried by the existing +rotation loop; a bad artifact that somehow reaches CAS is also refused at +pickup. The fuzzy “empty dead ends after a non-trivial later rotation” signal +is intentionally not yet enforced: no reliable diff-size/rotation-index +evidence is available at this validation boundary, so inventing a hard rule +would create false refusals. + +Verified after the change with `go test ./...`, `go build ./...`, `go vet +./...`, and `git diff --check`. diff --git a/cmd/orchestra-worker/main.go b/cmd/orchestra-worker/main.go new file mode 100644 index 0000000..3130f74 --- /dev/null +++ b/cmd/orchestra-worker/main.go @@ -0,0 +1,357 @@ +// orchestra-worker consumes router-issued leases for one local herdr. Homesrv +// remains the scheduler and CAS authority; this process owns only local Git +// and pane operations. +package main + +import ( + "context" + "encoding/json" + "fmt" + "log" + "orchestra/internal/continuity" + "orchestra/internal/domain" + "orchestra/internal/federation" + "orchestra/internal/herdr" + "orchestra/internal/orchestrator" + "os" + "os/exec" + "os/signal" + "path/filepath" + "strconv" + "strings" + "syscall" + "time" +) + +type worker struct { + api federation.Client + herdr *herdr.Client + harnessID, harness, repo, root, remote string + cursor uint64 + tasks map[string]domain.Task + sessions map[string]herdr.Session + leases map[string]lease + statePath string + hard float64 + registration federation.Worker +} +type lease struct { + HandoffRef string `json:"handoff_ref,omitempty"` +} +type workerState struct { + Cursor uint64 `json:"cursor"` + Sessions map[string]herdr.Session `json:"sessions"` + Tasks map[string]domain.Task `json:"tasks"` + Leases map[string]lease `json:"leases"` +} + +func (w *worker) load() { + b, e := os.ReadFile(w.statePath) + if e == nil { + var s workerState + if json.Unmarshal(b, &s) == nil { + w.cursor = s.Cursor + w.sessions = s.Sessions + w.tasks = s.Tasks + w.leases = s.Leases + } + } + if w.sessions == nil { + w.sessions = map[string]herdr.Session{} + } + if w.tasks == nil { + w.tasks = map[string]domain.Task{} + } + if w.leases == nil { + w.leases = map[string]lease{} + } +} +func (w *worker) save() error { + b, e := json.Marshal(workerState{Cursor: w.cursor, Sessions: w.sessions, Tasks: w.tasks, Leases: w.leases}) + if e != nil { + return e + } + if e := os.MkdirAll(filepath.Dir(w.statePath), 0700); e != nil { + return e + } + return os.WriteFile(w.statePath, b, 0600) +} + +type artifactCAS struct{ api federation.Client } + +func (c artifactCAS) PutArtifact(b []byte) (string, error) { + return c.api.PutArtifact(context.Background(), b) +} +func (c artifactCAS) Artifact(ref string) ([]byte, error) { + return c.api.Artifact(context.Background(), ref) +} + +func created(e domain.Event) (domain.Task, bool) { + if e.Type != "TaskCreated" { + return domain.Task{}, false + } + var p struct { + Source string `json:"source"` + ExternalID string `json:"external_id"` + Project string `json:"project"` + Capability []string `json:"capability"` + Title string `json:"title"` + Description string `json:"description"` + } + if json.Unmarshal(e.Payload, &p) != nil || p.Source == "" || p.ExternalID == "" || p.Project == "" { + return domain.Task{}, false + } + return domain.Task{ID: e.TaskID, Source: p.Source, ExternalID: p.ExternalID, Project: p.Project, Capability: p.Capability, Title: p.Title, Description: p.Description}, true +} + +func (w *worker) syncBase(ctx context.Context) error { + if out, err := exec.CommandContext(ctx, "git", "-C", w.repo, "fetch", w.remote, "--prune").CombinedOutput(); err != nil { + return fmt.Errorf("fetch base checkout: %s: %w", out, err) + } + branch, err := exec.CommandContext(ctx, "git", "-C", w.repo, "symbolic-ref", "--quiet", "--short", "HEAD").Output() + if err != nil { + return fmt.Errorf("identify base branch: %w", err) + } + branchName := strings.TrimSpace(string(branch)) + if out, err := exec.CommandContext(ctx, "git", "-C", w.repo, "merge", "--ff-only", w.remote+"/"+branchName).CombinedOutput(); err != nil { + return fmt.Errorf("fast-forward base checkout: %s: %w", out, err) + } + return nil +} + +func (w *worker) start(ctx context.Context, t domain.Task, ref string) error { + var wt string + var h continuity.Handoff + var err error + // Synchronize the local base before any worktree operation. A worker never + // treats a coordinator-side path as truth; the Git remote is the only + // cross-machine transport. + if err := w.syncBase(ctx); err != nil { + return err + } + if ref != "" { + b, err := w.api.Artifact(ctx, ref) + if err != nil { + return err + } + h, err = continuity.Decode(b) + if err != nil { + return err + } + if out, err := exec.CommandContext(ctx, "git", "-C", w.repo, "fetch", w.remote, "--prune").CombinedOutput(); err != nil { + return fmt.Errorf("fetch pickup anchor: %s: %w", out, err) + } + wt = filepath.Join(w.root, t.ID) + if _, err := os.Stat(wt); os.IsNotExist(err) { + if out, err := exec.CommandContext(ctx, "git", "-C", w.repo, "worktree", "add", "-b", "orchestra/"+t.ID, wt, h.Anchor.GitSHA).CombinedOutput(); err != nil { + return fmt.Errorf("create pickup worktree: %s: %w", out, err) + } + } + if err = continuity.ValidatePickup(wt, h, taskHash(t)); err != nil { + return err + } + } else { + wt, err = (orchestrator.GitWorktrees{Repo: w.repo, Root: w.root}).Create(ctx, t) + if err != nil { + return err + } + } + if _, err = w.herdr.Worktree(ctx, w.repo, wt, "orchestra/"+t.ID); err != nil { + return err + } + s, err := w.herdr.StartAgent(ctx, wt, wt, "orchestra/"+t.ID, w.harness, t.ID) + if err != nil { + return err + } + p := "Begin Orchestra task " + t.ID + ".\nTitle: " + t.Title + "\nInstructions:\n" + t.Description + "\nWork only in this worktree. Do not edit TASK.md." + if ref != "" { + p += "\nA validated handoff exists. Read TASK.md and inspect local git history before continuing." + } + w.sessions[t.ID] = s + if err := w.save(); err != nil { + return err + } + // A prompt response can be lost after herdr accepted it. Persist the + // session first so the worker can reconcile/release it after restart. + return w.herdr.Prompt(ctx, s.PaneID, p, 0) +} + +func taskHash(t domain.Task) string { b := continuity.RenderTaskFile(t); return domain.Hash(b) } + +func (w *worker) releaseReady(ctx context.Context) { + for id, s := range w.sessions { + if report, err := os.ReadFile(filepath.Join(s.Worktree, ".orchestra-report.md")); err == nil && len(report) > 0 { + if ref, err := w.api.PutArtifact(ctx, report); err == nil { + if err = w.api.Complete(ctx, id, ref); err == nil { + delete(w.sessions, id) + delete(w.leases, id) + _ = w.save() + continue + } + log.Printf("complete %s: %v", id, err) + } else { + log.Printf("upload completion %s: %v", id, err) + } + } + if _, err := os.Stat(filepath.Join(s.Worktree, herdr.HandoffReportFile)); err != nil { + if !s.HandoffRequested { + a := herdr.CLIAdapter{Client: w.herdr, Harness: w.harness} + if occ, occErr := a.Occupancy(s); occErr == nil && occ >= w.hard { + if boundary, boundaryErr := a.AtTurnBoundary(ctx, s); boundaryErr == nil && boundary { + if err := a.RequestHandoff(ctx, s); err == nil { + s.HandoffRequested, s.HandoffReason = true, "threshold" + w.sessions[id] = s + _ = w.save() + } + } + } + } + continue + } + a := herdr.CLIAdapter{Client: w.herdr, Harness: w.harness, CAS: artifactCAS{w.api}, Remote: w.remote} + ref, err := a.Release(ctx, s) + if err != nil { + log.Printf("release %s: %v", id, err) + continue + } + sha, err := herdr.HeadSHA(s.Worktree) + if err != nil { + log.Printf("release %s anchor: %v", id, err) + continue + } + if err = w.api.Release(ctx, id, ref, sha); err != nil { + log.Printf("publish release %s: %v", id, err) + continue + } + // release_agent only removes herdr's binding. Closing the released pane + // after the handoff is durable prevents the next StartAgent from + // inheriting the predecessor's still-running terminal process. + if err := a.Kill(ctx, s); err != nil { + log.Printf("close released pane %s: %v", id, err) + } + delete(w.sessions, id) + _ = w.save() + } +} + +func (w *worker) once(ctx context.Context) error { + es, _, err := w.api.Events(ctx, w.cursor) + if err != nil { + return err + } + for _, e := range es { + if t, ok := created(e); ok { + w.tasks[t.ID] = t + } + if e.Type == "TaskLeased" { + var p struct { + HarnessID string `json:"harness_id"` + HandoffRef string `json:"handoff_ref"` + } + if json.Unmarshal(e.Payload, &p) == nil && p.HarnessID == w.harnessID { + w.leases[e.TaskID] = lease{HandoffRef: p.HandoffRef} + } + } + if e.Type == "TaskReleased" || e.Type == "TaskCompleted" || e.Type == "TaskFailed" || e.Type == "TaskBlocked" { + delete(w.leases, e.TaskID) + delete(w.sessions, e.TaskID) + } + if e.Seq > w.cursor { + w.cursor = e.Seq + } + } + // State is only a cache. If a lease survived but its TaskCreated event is + // older than the worker's cursor (or the event has been compacted), hydrate + // the authoritative task projection before deciding whether to start. + for taskID := range w.leases { + if _, ok := w.tasks[taskID]; !ok { + tasks, err := w.api.Tasks(ctx) + if err != nil { + return fmt.Errorf("hydrate leased task %s: %w", taskID, err) + } + for _, task := range tasks { + w.tasks[task.ID] = task + } + break + } + } + // Project the whole batch before launching. This prevents a new worker + // from resurrecting every historical lease during its initial replay. + for taskID, l := range w.leases { + if _, started := w.sessions[taskID]; started { + continue + } + if t, ok := w.tasks[taskID]; ok { + if err := w.start(ctx, t, l.HandoffRef); err != nil { + log.Printf("lease %s: %v", t.ID, err) + } + } + } + w.releaseReady(ctx) + if err := w.save(); err != nil { + return err + } + return w.api.Ack(ctx, w.cursor) +} + +// reRegisterAfterCoordinatorRestart restores the coordinator's in-memory +// worker registry. A worker must survive a server restart without operator +// intervention; its persisted cursor and sessions remain valid. +func (w *worker) reRegisterAfterCoordinatorRestart(ctx context.Context, cause error) bool { + if cause == nil || !strings.Contains(cause.Error(), "401 Unauthorized: unknown worker") { + return false + } + if err := w.api.Register(ctx, w.registration); err != nil { + log.Printf("re-register: %v", err) + return false + } + log.Printf("re-registered after coordinator restart") + return true +} +func required(k string) string { + v := os.Getenv(k) + if v == "" { + log.Fatalf("%s required", k) + } + return v +} +func main() { + id, token := required("ORCHESTRA_WORKER_ID"), required("ORCHESTRA_WORKER_TOKEN") + hard := .75 + if v, err := strconv.ParseFloat(os.Getenv("ORCHESTRA_OCCUPANCY_HARD"), 64); err == nil && v > 0 && v < 1 { + hard = v + } + w := &worker{api: federation.Client{BaseURL: required("ORCHESTRA_URL"), WorkerID: id, Token: token, AdmitToken: os.Getenv("ORCHESTRA_FEDERATION_ADMIT_TOKEN")}, harnessID: required("ORCHESTRA_WORKER_HERDR_ID"), harness: required("ORCHESTRA_WORKER_HARNESS"), repo: required("ORCHESTRA_REPO"), root: required("ORCHESTRA_WORKTREE_ROOT"), remote: required("ORCHESTRA_GIT_REMOTE"), tasks: map[string]domain.Task{}, sessions: map[string]herdr.Session{}, leases: map[string]lease{}, statePath: os.Getenv("ORCHESTRA_WORKER_STATE"), hard: hard, registration: federation.Worker{ID: id, Address: os.Getenv("ORCHESTRA_WORKER_ADDRESS"), Capacity: 1}} + if w.statePath == "" { + w.statePath = filepath.Join(w.root, ".orchestra-worker-state.json") + } + w.load() + w.herdr = herdr.New(required("ORCHESTRA_WORKER_HERDR")) + if id != w.harnessID { + log.Fatal("ORCHESTRA_WORKER_ID must equal ORCHESTRA_WORKER_HERDR_ID so leases and offline recovery have one owner") + } + if err := w.api.Register(context.Background(), w.registration); err != nil { + log.Fatal(err) + } + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + ticker := time.NewTicker(5 * time.Second) + defer ticker.Stop() + for { + if err := w.api.Heartbeat(ctx); err != nil { + log.Printf("heartbeat: %v", err) + if w.reRegisterAfterCoordinatorRestart(ctx, err) { + continue + } + } + if err := w.once(ctx); err != nil { + log.Printf("poll: %v", err) + w.reRegisterAfterCoordinatorRestart(ctx, err) + } + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } +} diff --git a/cmd/orchestra-worker/main_test.go b/cmd/orchestra-worker/main_test.go new file mode 100644 index 0000000..839839a --- /dev/null +++ b/cmd/orchestra-worker/main_test.go @@ -0,0 +1,151 @@ +package main + +import ( + "bufio" + "context" + "encoding/json" + "errors" + "net" + "net/http" + "net/http/httptest" + "orchestra/internal/domain" + "orchestra/internal/federation" + "orchestra/internal/herdr" + "os" + "os/exec" + "path/filepath" + "testing" +) + +func TestWorkerReregistersAfterCoordinatorForgetsIt(t *testing.T) { + registered := false + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Path != "/v1/federation/workers" { + t.Fatalf("unexpected %s %s", r.Method, r.URL.Path) + } + registered = true + w.WriteHeader(http.StatusCreated) + })) + defer s.Close() + w := &worker{api: federation.Client{BaseURL: s.URL, WorkerID: "h", Token: "t"}, registration: federation.Worker{ID: "h", Capacity: 1}} + if !w.reRegisterAfterCoordinatorRestart(context.Background(), errors.New("federation: 401 Unauthorized: unknown worker")) { + t.Fatal("worker did not identify a coordinator restart") + } + if !registered { + t.Fatal("worker did not register") + } +} + +func TestInitialReplayDoesNotResurrectReleasedLease(t *testing.T) { + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/v1/federation/events": + _, _ = w.Write([]byte(`{"cursor":0,"events":[{"seq":1,"id":"c","type":"TaskCreated","task_id":"t","version":1,"payload":{"source":"s","external_id":"x","project":"p"},"surface":"system"},{"seq":2,"id":"l","type":"TaskLeased","task_id":"t","version":2,"payload":{"harness_id":"h"},"surface":"system"},{"seq":3,"id":"r","type":"TaskReleased","task_id":"t","version":3,"payload":{"reason":"expired"},"surface":"system"}]}`)) + case "/v1/federation/events/ack": + w.WriteHeader(http.StatusNoContent) + default: + t.Fatalf("unexpected %s", r.URL.Path) + } + })) + defer s.Close() + w := &worker{api: federation.Client{BaseURL: s.URL, WorkerID: "h", Token: "t"}, harnessID: "h", tasks: map[string]domain.Task{}, sessions: map[string]herdr.Session{}, leases: map[string]lease{}, statePath: t.TempDir() + "/state.json", hard: .75} + if err := w.once(context.Background()); err != nil { + t.Fatal(err) + } + if len(w.leases) != 0 || len(w.sessions) != 0 { + t.Fatalf("replayed lease survived: leases=%v sessions=%v", w.leases, w.sessions) + } + if w.cursor != 3 { + t.Fatalf("cursor=%d want 3", w.cursor) + } +} + +func TestWorkerHydratesLeaseMissingTaskCache(t *testing.T) { + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/v1/federation/events": + _, _ = w.Write([]byte(`{"cursor":2,"events":[{"seq":2,"id":"l","type":"TaskLeased","task_id":"t","version":2,"payload":{"harness_id":"h"},"surface":"system"}]}`)) + case "/v1/tasks": + _, _ = w.Write([]byte(`[{"id":"t","source":"s","external_id":"x","project":"p","state":"leased"}]`)) + case "/v1/federation/events/ack": + w.WriteHeader(http.StatusNoContent) + default: + t.Fatalf("unexpected %s", r.URL.Path) + } + })) + defer s.Close() + w := &worker{api: federation.Client{BaseURL: s.URL, WorkerID: "h", Token: "t"}, harnessID: "h", cursor: 1, tasks: map[string]domain.Task{}, sessions: map[string]herdr.Session{"t": {}}, leases: map[string]lease{}, statePath: t.TempDir() + "/state.json", hard: .75} + if err := w.once(context.Background()); err != nil { + t.Fatal(err) + } + if _, ok := w.tasks["t"]; !ok || w.cursor != 2 { + t.Fatalf("task was not hydrated: tasks=%v cursor=%d", w.tasks, w.cursor) + } +} + +func TestWorkerStartsRouterIssuedLeaseInLocalGitWorktree(t *testing.T) { + remote := filepath.Join(t.TempDir(), "remote.git") + if out, err := exec.Command("git", "init", "--bare", remote).CombinedOutput(); err != nil { + t.Fatalf("remote: %v %s", err, out) + } + seed := t.TempDir() + for _, a := range [][]string{{"init", seed}, {"-C", seed, "config", "user.email", "t@t"}, {"-C", seed, "config", "user.name", "t"}, {"-C", seed, "commit", "--allow-empty", "-m", "init"}, {"-C", seed, "remote", "add", "origin", remote}, {"-C", seed, "push", "-u", "origin", "HEAD:master"}} { + if out, err := exec.Command("git", a...).CombinedOutput(); err != nil { + t.Fatalf("git %v: %v %s", a, err, out) + } + } + repo := filepath.Join(t.TempDir(), "repo") + if out, err := exec.Command("git", "clone", remote, repo).CombinedOutput(); err != nil { + t.Fatal(string(out)) + } + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer ln.Close() + go func() { + for { + c, e := ln.Accept() + if e != nil { + return + } + go func() { + defer c.Close() + var r herdr.Request + if json.NewDecoder(bufio.NewReader(c)).Decode(&r) != nil { + return + } + var result string + switch r.Method { + case "worktree.create": + result = `{"path":"` + filepath.Join(filepath.Dir(repo), "worktrees", "task") + `","root_pane":{"pane_id":"p1"}}` + case "agent.start": + result = `{}` + case "pane.get": + result = `{"pane":{"agent":"opencode","agent_status":"idle"}}` + case "pane.read": + result = `{"read":{"text":""}}` + case "agent.prompt": + result = `{}` + default: + result = `{}` + } + _ = json.NewEncoder(c).Encode(herdr.Response{ID: r.ID, Result: json.RawMessage(result)}) + }() + } + }() + root := filepath.Join(filepath.Dir(repo), "worktrees") + w := &worker{herdr: &herdr.Client{Path: ln.Addr().String()}, repo: repo, root: root, remote: "origin", harness: "opencode", tasks: map[string]domain.Task{}, sessions: map[string]herdr.Session{}, leases: map[string]lease{}, statePath: filepath.Join(t.TempDir(), "state.json"), hard: .75} + // New() is needed for its pane map; override the address for the fake. + w.herdr = herdr.New(ln.Addr().String()) + task := domain.Task{ID: "task", Source: "s", ExternalID: "x", Project: "p", Title: "test", Description: "do work"} + if err := w.start(context.Background(), task, ""); err != nil { + t.Fatal(err) + } + if _, ok := w.sessions[task.ID]; !ok { + t.Fatal("session not persisted") + } + if _, err := os.Stat(filepath.Join(root, "task", "TASK.md")); err != nil { + t.Fatalf("TASK.md: %v", err) + } +} diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go index 9685322..783c3d2 100644 --- a/cmd/orchestra/main.go +++ b/cmd/orchestra/main.go @@ -46,6 +46,49 @@ func herdrAddress(rr registry.Registry, h registry.Herdr) string { return net.JoinHostPort(host, defaultHerdrPort) } +type federatedAvailability struct { + base router.Availability + workers *federation.Registry + localMachine string +} + +// federatedReachability keeps the legacy TCP probe for local herdrs, while +// avoiding a coordinator-side probe of a remote worker's herdr socket. A +// remote harness is reachable precisely when its worker is registered (as +// enforced by federatedAvailability); probing its raw herdr endpoint here +// would reintroduce the cross-machine Design A dependency. +type federatedReachability struct { + base registry.Reachability + remote map[string]bool +} + +func (r federatedReachability) Reachable(address string, timeout time.Duration) bool { + if r.remote[address] { + return true + } + return r.base.Reachable(address, timeout) +} + +func remoteHerdrAddresses(rr registry.Registry, localMachine string) map[string]bool { + remote := map[string]bool{} + for _, h := range rr.Herdrs() { + if h.MachineID != localMachine { + remote[herdrAddress(rr, h)] = true + } + } + return remote +} + +func (a federatedAvailability) Available(h registry.Herdr) bool { + if a.base != nil && !a.base.Available(h) { + return false + } + if a.localMachine == "" || h.MachineID == a.localMachine { + return true + } + return a.workers.Available(h.ID) +} + func main() { dir := os.Getenv("ORCHESTRA_DATA") if dir == "" { @@ -58,11 +101,17 @@ func main() { var rr registry.Registry var rt *router.Router var coordinator *orchestrator.Coordinator + localMachine := os.Getenv("ORCHESTRA_MACHINE_ID") + workers := &federation.Registry{AdmitToken: os.Getenv("ORCHESTRA_FEDERATION_ADMIT_TOKEN")} if config := os.Getenv("ORCHESTRA_CONFIG"); config != "" { if rr, err = registry.Load(config); err != nil { log.Fatalf("load orchestra config: %v", err) } - rt = &router.Router{Store: s, Registry: rr, Reachability: registry.TCPReachability{}, Timeout: time.Second, Retry: router.RetryPolicy{MaxAttempts: 3, Backoff: time.Minute}} + reachability := registry.Reachability(registry.TCPReachability{}) + if localMachine != "" { + reachability = federatedReachability{base: reachability, remote: remoteHerdrAddresses(rr, localMachine)} + } + rt = &router.Router{Store: s, Registry: rr, Reachability: reachability, Timeout: time.Second, Retry: router.RetryPolicy{MaxAttempts: 3, Backoff: time.Minute}} limits := map[string]router.QuotaWindowLimits{} for _, h := range rr.Herdrs() { w := router.QuotaWindowLimits{FiveHour: h.QuotaLimit5h, Weekly: h.QuotaLimitWeekly} @@ -77,6 +126,9 @@ func main() { if len(limits) > 0 { rt.Availability = router.QuotaAvailability{Store: s, Limits: limits} } + if localMachine != "" { + rt.Availability = federatedAvailability{base: rt.Availability, workers: workers, localMachine: localMachine} + } if repo, root := os.Getenv("ORCHESTRA_REPO"), os.Getenv("ORCHESTRA_WORKTREE_ROOT"); repo != "" && root != "" { adapters := map[string]herdr.Adapter{} for _, h := range rr.Herdrs() { @@ -115,7 +167,22 @@ func main() { Default: orchestrator.GitWorktrees{Root: root, Repo: repo}, } coordinator = &orchestrator.Coordinator{Store: s, StatePath: filepath.Join(dir, "runtime-sessions.json"), Worktrees: worktrees, Adapters: orchestrator.AdapterFactory{Herdrs: adapters}} - rt.OnLease = func(e domain.Event) error { return coordinator.Start(context.Background(), e) } + rt.OnLease = func(e domain.Event) error { + // In federated mode the coordinator must never inspect a remote + // checkout. Its worker consumes the router-issued lease event and + // performs all Git/herdr operations on that machine (§2.1). + if localMachine != "" { + var p struct { + HarnessID string `json:"harness_id"` + } + if json.Unmarshal(e.Payload, &p) == nil { + if h, ok := rr.Herdr(p.HarnessID); ok && h.MachineID != localMachine { + return nil + } + } + } + return coordinator.Start(context.Background(), e) + } hard := 0.75 if v, parseErr := strconv.ParseFloat(os.Getenv("ORCHESTRA_OCCUPANCY_HARD"), 64); parseErr == nil && v > 0 && v < 1 { hard = v @@ -131,7 +198,6 @@ func main() { } } mux := http.NewServeMux() - workers := &federation.Registry{AdmitToken: os.Getenv("ORCHESTRA_FEDERATION_ADMIT_TOKEN")} workers.OnOffline = func(w federation.Worker) { for _, t := range s.Tasks() { if t.State == domain.StateLeased && t.Lease != nil && t.Lease.HarnessID == w.ID { @@ -143,6 +209,13 @@ func main() { } } } + go func() { + ticker := time.NewTicker(15 * time.Second) + defer ticker.Stop() + for range ticker.C { + _ = workers.Snapshot() + } + }() providerHealth := map[string]*provider.Supervisor{} surface := func(r *http.Request) authz.Surface { v := authz.ParseSurface(r.Header.Get("X-Orchestra-Surface")) @@ -245,6 +318,33 @@ func main() { w.WriteHeader(http.StatusCreated) json.NewEncoder(w).Encode(map[string]string{"ref": ref}) }) + // Artifacts are normally write-only to public surfaces. A federation + // worker may read a handoff only after authenticating as the worker that + // will validate it against its own checkout (§2.1, §6.2). + mux.HandleFunc("/v1/artifacts/", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + wid := r.Header.Get("X-Orchestra-Worker") + token := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ") + if err := workers.Authenticate(wid, token); err != nil { + http.Error(w, err.Error(), http.StatusUnauthorized) + return + } + ref := strings.TrimPrefix(r.URL.Path, "/v1/artifacts/") + if len(ref) != 64 { + http.Error(w, "artifact ref required", http.StatusBadRequest) + return + } + b, err := s.Artifact(ref) + if err != nil { + http.Error(w, "artifact not found", http.StatusNotFound) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write(b) + }) // /v1/harness/complete is the automatic TaskCompleted producer (AUDIT.md // B3): a harness-side hook posts here when the agent has declared the // task done (see deploy/hooks/orchestra-stop.sh), not on every turn @@ -715,6 +815,11 @@ func main() { http.Error(w, err.Error(), status) return } + if rt != nil { + if _, err := rt.AssignPending(); err != nil { + log.Printf("route after worker registration: %v", err) + } + } w.WriteHeader(http.StatusCreated) json.NewEncoder(w).Encode(worker) }) @@ -768,7 +873,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, "/claim") && !strings.HasSuffix(r.URL.Path, "/handoff")) { + if r.Method != http.MethodPost || (!strings.HasSuffix(r.URL.Path, "/heartbeat") && !strings.HasSuffix(r.URL.Path, "/handoff") && !strings.HasSuffix(r.URL.Path, "/complete")) { http.Error(w, "not found", 404) return } @@ -804,22 +909,24 @@ func main() { http.Error(w, "task not found", 404) return } - if strings.HasSuffix(r.URL.Path, "/claim") { - if b.TTLSeconds <= 0 { - b.TTLSeconds = 1800 + if t.State != domain.StateLeased || t.Lease == nil || t.Lease.HarnessID != parts[3] { + http.Error(w, "lease not owned", 409) + return + } + if strings.HasSuffix(r.URL.Path, "/complete") { + if b.HandoffRef == "" { + http.Error(w, "report_ref required", 400) + return } - e, err := s.Lease(b.TaskID, parts[3], time.Duration(b.TTLSeconds)*time.Second) - if err != nil { + p, _ := json.Marshal(map[string]any{"report_ref": b.HandoffRef, "receipt": map[string]any{"harness_id": parts[3], "consumed": 0}}) + e := domain.Event{ID: id(), Type: "TaskCompleted", TaskID: b.TaskID, Version: t.Version + 1, Payload: p, Surface: string(authz.System)} + if err := s.Append(e); err != nil { http.Error(w, err.Error(), 409) return } json.NewEncoder(w).Encode(e) return } - if t.State != domain.StateLeased || t.Lease == nil || t.Lease.HarnessID != parts[3] { - http.Error(w, "lease not owned", 409) - return - } if b.HandoffRef == "" { http.Error(w, "handoff_ref required", 400) return diff --git a/cmd/orchestra/main_test.go b/cmd/orchestra/main_test.go new file mode 100644 index 0000000..d13bb35 --- /dev/null +++ b/cmd/orchestra/main_test.go @@ -0,0 +1,35 @@ +package main + +import ( + "os" + "path/filepath" + "testing" + "time" + + "orchestra/internal/registry" +) + +type unreachable struct{} + +func (unreachable) Reachable(string, time.Duration) bool { return false } + +func TestFederatedReachabilityDefersRemoteHerdrToWorkerHeartbeat(t *testing.T) { + path := filepath.Join(t.TempDir(), "config.json") + if err := os.WriteFile(path, []byte(`{ + "machines":[{"id":"homesrv","address":"192.168.1.104:9145"},{"id":"workpc","address":"192.168.1.105:9145"}], + "herdrs":[{"id":"local","machine_id":"homesrv","harness":"opencode"},{"id":"remote","machine_id":"workpc","harness":"opencode"}] + }`), 0o600); err != nil { + t.Fatal(err) + } + r, err := registry.Load(path) + if err != nil { + t.Fatal(err) + } + check := federatedReachability{base: unreachable{}, remote: remoteHerdrAddresses(r, "homesrv")} + if check.Reachable("192.168.1.105:9245", time.Second) != true { + t.Fatal("remote herdr should be admitted for worker heartbeat gating") + } + if check.Reachable("192.168.1.104:9245", time.Second) { + t.Fatal("local herdr should still require its TCP probe") + } +} diff --git a/deploy/orchestra-worker.service b/deploy/orchestra-worker.service new file mode 100644 index 0000000..a137adc --- /dev/null +++ b/deploy/orchestra-worker.service @@ -0,0 +1,15 @@ +[Unit] +Description=Orchestra federation worker +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple +User=orchestra +EnvironmentFile=/etc/orchestra/worker.env +ExecStart=/usr/local/bin/orchestra-worker +Restart=always +RestartSec=5 + +[Install] +WantedBy=multi-user.target diff --git a/deploy/orchestra.env.example b/deploy/orchestra.env.example index 8d34488..49d0ba5 100644 --- a/deploy/orchestra.env.example +++ b/deploy/orchestra.env.example @@ -8,6 +8,18 @@ ORCHESTRA_DATA=/var/lib/orchestra/data ORCHESTRA_PORT=9145 +# --- Federation worker (set only on a harness host such as workpc) --- +# ORCHESTRA_URL=http://homesrv:9145 +# ORCHESTRA_WORKER_ID=workpc-opencode # must equal ORCHESTRA_WORKER_HERDR_ID +# ORCHESTRA_WORKER_TOKEN= +# ORCHESTRA_FEDERATION_ADMIT_TOKEN= +# ORCHESTRA_WORKER_HERDR_ID=workpc-opencode +# ORCHESTRA_WORKER_HARNESS=opencode +# ORCHESTRA_WORKER_HERDR=/home/orchestra/.config/herdr/herdr.sock +# ORCHESTRA_WORKER_STATE=/var/lib/orchestra-worker/state.json +# ORCHESTRA_GIT_REMOTE=origin +# ORCHESTRA_MACHINE_ID=homesrv # set on the authoritative coordinator + # Static project/machine/herdr topology (registry.Load). Required for # routing across more than one machine; validated at startup. ORCHESTRA_CONFIG=/etc/orchestra/config.json diff --git a/internal/continuity/continuity.go b/internal/continuity/continuity.go index ec2cd74..a395b5d 100644 --- a/internal/continuity/continuity.go +++ b/internal/continuity/continuity.go @@ -11,6 +11,7 @@ import ( "os" "os/exec" "path/filepath" + "regexp" "strings" "time" ) @@ -36,6 +37,9 @@ func RenderTaskFile(t domain.Task) []byte { if t.Parent != "" { fmt.Fprintf(&b, "- Parent: %s\n", t.Parent) } + if strings.TrimSpace(t.Description) != "" { + fmt.Fprintf(&b, "\n## Instructions\n\n%s\n", t.Description) + } b.WriteString("\nThis file is immutable for the lifetime of the task (§6.2) — its hash is\ncarried in every handoff and re-verified on every pickup. Do not edit it.\n") return []byte(b.String()) } @@ -99,21 +103,14 @@ type Meta struct { RotationIndex int `json:"rotation_index"` } type Handoff struct { - Meta Meta `json:"meta"` - Anchor Anchor `json:"anchor"` - Goal string `json:"goal"` - DoneWhen []string `json:"done_when"` - Completed []Completed `json:"completed"` - Remaining []string `json:"remaining"` - Action string `json:"action"` - Command string `json:"command"` - Files []string `json:"files"` - Invariants []string `json:"invariants"` - DeadEnds []DeadEnd `json:"dead_ends"` - OpenQuestions []string `json:"open_questions"` - Build string `json:"build"` - Test string `json:"test"` - LastResult Result `json:"last_result"` + Meta Meta `json:"meta"` + Anchor Anchor `json:"anchor"` + Remaining []string `json:"remaining"` + Action string `json:"action"` + Command string `json:"command"` + DeadEnds []DeadEnd `json:"dead_ends"` + OpenQuestions []string `json:"open_questions"` + Learned []string `json:"learned"` } type Result struct { Command string `json:"command"` @@ -123,13 +120,32 @@ type Result struct { var reasons = map[string]bool{"threshold": true, "milestone": true, "thrash": true, "manual": true} +const maxAuthoredLine = 200 + +var circularAction = regexp.MustCompile(`(?i)handoff|report\.md|^continue the task`) +var circularCommand = regexp.MustCompile(`(?i)\.orchestra-handoff|handoff-report|report\.md`) + func (h Handoff) Validate() error { if strings.TrimSpace(h.Meta.ID) == "" || !reasons[h.Meta.Reason] || h.Meta.RotationIndex < 0 { return errors.New("invalid handoff meta") } - if len(h.Anchor.GitSHA) != 40 || h.Anchor.Branch == "" || strings.TrimSpace(h.Goal) == "" || len(h.DoneWhen) == 0 || strings.TrimSpace(h.Action) == "" || strings.TrimSpace(h.Command) == "" { + if len(h.Anchor.GitSHA) != 40 || h.Anchor.Branch == "" || strings.TrimSpace(h.Action) == "" { return errors.New("invalid handoff required fields") } + if circularAction.MatchString(h.Action) { + return errors.New("invalid handoff action: must name concrete next work, not a handoff") + } + if err := validateAuthoredLine(h.Action); err != nil { + return err + } + if circularCommand.MatchString(h.Command) { + return errors.New("invalid handoff command: must not point to a handoff or report") + } + for _, item := range append(append([]string{}, h.Remaining...), append(h.OpenQuestions, h.Learned...)...) { + if err := validateAuthoredLine(item); err != nil { + return err + } + } for _, d := range h.Anchor.Dirty { if filepath.IsAbs(d.Path) || d.Path == "" || len(d.SHA256) != 64 { return errors.New("invalid dirty anchor") @@ -139,6 +155,22 @@ func (h Handoff) Validate() error { if strings.TrimSpace(d.Tried) == "" || strings.TrimSpace(d.WhyFailed) == "" { return errors.New("invalid dead end") } + if err := validateAuthoredLine(d.Tried); err != nil { + return err + } + if err := validateAuthoredLine(d.WhyFailed); err != nil { + return err + } + } + return nil +} + +func validateAuthoredLine(s string) error { + if strings.TrimSpace(s) == "" { + return errors.New("invalid handoff authored field: empty item") + } + if strings.Contains(s, "\n#") || len(s) > maxAuthoredLine { + return errors.New("invalid handoff authored field: prose smuggled into list") } return nil } diff --git a/internal/continuity/continuity_test.go b/internal/continuity/continuity_test.go index 4380a05..b70883d 100644 --- a/internal/continuity/continuity_test.go +++ b/internal/continuity/continuity_test.go @@ -6,6 +6,7 @@ import ( "os" "os/exec" "path/filepath" + "strings" "testing" "orchestra/internal/store" @@ -27,7 +28,7 @@ func TestHandoffCASAndPickup(t *testing.T) { head := runOut(t, root, "rev-parse", "HEAD") task := sha256.Sum256([]byte("original")) s, _ := store.Open(t.TempDir()) - h := Handoff{Meta: Meta{ID: "h1", Reason: "manual"}, Anchor: Anchor{GitSHA: head, Branch: "main"}, Goal: "ship", DoneWhen: []string{"tests pass"}, Action: "test", Command: "go test ./...", LastResult: Result{AtSHA: head}} + h := Handoff{Meta: Meta{ID: "h1", Reason: "manual"}, Anchor: Anchor{GitSHA: head, Branch: "main"}, Action: "run the focused tests", Command: "go test ./..."} ref, e := Save(h, s) if e != nil { t.Fatal(e) @@ -55,6 +56,26 @@ func TestDecodeRejectsUnknownKnowledgeFields(t *testing.T) { } } +func TestHandoffRejectsFabricatedOrProseAuthoredFields(t *testing.T) { + base := Handoff{Meta: Meta{ID: "h", Reason: "manual"}, Anchor: Anchor{GitSHA: strings.Repeat("a", 40), Branch: "main"}, Action: "run the focused tests"} + if err := base.Validate(); err != nil { + t.Fatal(err) + } + badAction := base + badAction.Action = "Continue the task from the handoff" + if err := badAction.Validate(); err == nil { + t.Fatal("expected circular action rejection") + } + badProse := base + badProse.Remaining = []string{"short\n# markdown heading"} + if err := badProse.Validate(); err == nil { + t.Fatal("expected prose-in-list rejection") + } + if _, err := Decode([]byte(`{"meta":{"id":"h","reason":"manual","rotation_index":0},"anchor":{"git_sha":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","branch":"main"},"goal":"fabricated","action":"run tests"}`)); err == nil { + t.Fatal("expected goal rejection") + } +} + func TestScratchCommitProtectsTask(t *testing.T) { root := t.TempDir() run := func(a ...string) { diff --git a/internal/domain/domain.go b/internal/domain/domain.go index 96f2bdf..2c401e7 100644 --- a/internal/domain/domain.go +++ b/internal/domain/domain.go @@ -59,9 +59,13 @@ type Task struct { Estimate *Estimate `json:"estimate,omitempty"` State TaskState `json:"state"` Lease *Lease `json:"lease,omitempty"` - Version int `json:"version"` - Title string `json:"title,omitempty"` - Description string `json:"description,omitempty"` + // HandoffRef survives the queued interval between TaskReleased and the + // next router-owned TaskLeased event; it is the only artifact the worker + // may use for local pickup validation. + HandoffRef string `json:"handoff_ref,omitempty"` + Version int `json:"version"` + Title string `json:"title,omitempty"` + Description string `json:"description,omitempty"` } type Event struct { diff --git a/internal/federation/client.go b/internal/federation/client.go new file mode 100644 index 0000000..912476b --- /dev/null +++ b/internal/federation/client.go @@ -0,0 +1,183 @@ +package federation + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "orchestra/internal/domain" + "strings" +) + +// Client is the worker-side protocol client. It carries no task state: the +// homesrv event log remains authoritative and workers only persist their +// local execution session/checkouts. +type Client struct { + BaseURL string + WorkerID string + Token string + AdmitToken string + HTTP *http.Client +} + +func (c Client) Register(ctx context.Context, w Worker) error { + b, err := json.Marshal(map[string]any{"id": w.ID, "address": w.Address, "capacity": w.Capacity, "token": c.Token}) + if err != nil { + return err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(c.BaseURL, "/")+"/v1/federation/workers", bytes.NewReader(b)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + if c.AdmitToken != "" { + req.Header.Set("Authorization", "Bearer "+c.AdmitToken) + } + h := c.HTTP + if h == nil { + h = http.DefaultClient + } + resp, err := h.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode/100 != 2 { + msg, _ := io.ReadAll(resp.Body) + return fmt.Errorf("federation register: %s: %s", resp.Status, strings.TrimSpace(string(msg))) + } + return nil +} + +func (c Client) request(ctx context.Context, method, path string, body any) (*http.Response, error) { + var r io.Reader + if body != nil { + b, err := json.Marshal(body) + if err != nil { + return nil, err + } + r = bytes.NewReader(b) + } + req, err := http.NewRequestWithContext(ctx, method, strings.TrimRight(c.BaseURL, "/")+path, r) + if err != nil { + return nil, err + } + req.Header.Set("X-Orchestra-Worker", c.WorkerID) + req.Header.Set("Authorization", "Bearer "+c.Token) + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + h := c.HTTP + if h == nil { + h = http.DefaultClient + } + resp, err := h.Do(req) + if err != nil { + return nil, err + } + if resp.StatusCode/100 != 2 { + defer resp.Body.Close() + b, _ := io.ReadAll(resp.Body) + return nil, fmt.Errorf("federation: %s: %s", resp.Status, strings.TrimSpace(string(b))) + } + return resp, nil +} + +func (c Client) Events(ctx context.Context, since uint64) ([]domain.Event, uint64, error) { + resp, err := c.request(ctx, http.MethodGet, "/v1/federation/events?since="+fmt.Sprint(since), nil) + if err != nil { + return nil, 0, err + } + defer resp.Body.Close() + var out struct { + Cursor uint64 `json:"cursor"` + Events []domain.Event `json:"events"` + } + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + return nil, 0, err + } + return out.Events, out.Cursor, nil +} + +// Tasks hydrates the worker's cache when its local state predates the +// coordinator's event-retention window. The coordinator remains authoritative +// for the task projection. +func (c Client) Tasks(ctx context.Context) ([]domain.Task, error) { + resp, err := c.request(ctx, http.MethodGet, "/v1/tasks", nil) + if err != nil { + return nil, err + } + defer resp.Body.Close() + var tasks []domain.Task + if err := json.NewDecoder(resp.Body).Decode(&tasks); err != nil { + return nil, err + } + return tasks, nil +} +func (c Client) Ack(ctx context.Context, cursor uint64) error { + resp, err := c.request(ctx, http.MethodPost, "/v1/federation/events/ack", map[string]uint64{"cursor": cursor}) + if resp != nil { + resp.Body.Close() + } + return err +} +func (c Client) Heartbeat(ctx context.Context) error { + resp, err := c.request(ctx, http.MethodPost, "/v1/federation/workers/"+url.PathEscape(c.WorkerID)+"/heartbeat", nil) + if resp != nil { + resp.Body.Close() + } + return err +} +func (c Client) Artifact(ctx context.Context, ref string) ([]byte, error) { + resp, err := c.request(ctx, http.MethodGet, "/v1/artifacts/"+url.PathEscape(ref), nil) + if err != nil { + return nil, err + } + defer resp.Body.Close() + return io.ReadAll(resp.Body) +} + +func (c Client) PutArtifact(ctx context.Context, b []byte) (string, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(c.BaseURL, "/")+"/v1/artifacts", bytes.NewReader(b)) + if err != nil { + return "", err + } + req.Header.Set("Content-Type", "application/octet-stream") + h := c.HTTP + if h == nil { + h = http.DefaultClient + } + resp, err := h.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + if resp.StatusCode/100 != 2 { + msg, _ := io.ReadAll(resp.Body) + return "", fmt.Errorf("artifact upload: %s: %s", resp.Status, strings.TrimSpace(string(msg))) + } + var out struct { + Ref string `json:"ref"` + } + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + return "", err + } + return out.Ref, nil +} +func (c Client) Release(ctx context.Context, taskID, ref, anchor string) error { + resp, err := c.request(ctx, http.MethodPost, "/v1/federation/workers/"+url.PathEscape(c.WorkerID)+"/handoff", map[string]string{"task_id": taskID, "handoff_ref": ref, "anchor_sha": anchor}) + if resp != nil { + resp.Body.Close() + } + return err +} +func (c Client) Complete(ctx context.Context, taskID, reportRef string) error { + resp, err := c.request(ctx, http.MethodPost, "/v1/federation/workers/"+url.PathEscape(c.WorkerID)+"/complete", map[string]string{"task_id": taskID, "handoff_ref": reportRef}) + if resp != nil { + resp.Body.Close() + } + return err +} diff --git a/internal/federation/client_test.go b/internal/federation/client_test.go new file mode 100644 index 0000000..5d92970 --- /dev/null +++ b/internal/federation/client_test.go @@ -0,0 +1,51 @@ +package federation + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" +) + +func TestClientRegistersPollsAndReadsArtifactAsWorker(t *testing.T) { + seen := map[string]bool{} + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/v1/federation/workers" { + seen["register"] = r.Header.Get("Authorization") == "Bearer admit" + w.WriteHeader(http.StatusCreated) + return + } + if r.Header.Get("X-Orchestra-Worker") != "h1" || r.Header.Get("Authorization") != "Bearer worker" { + t.Errorf("worker auth missing") + } + switch r.URL.Path { + case "/v1/federation/events": + seen["events"] = true + _, _ = w.Write([]byte(`{"cursor":3,"events":[{"seq":3,"id":"e","type":"TaskCreated","task_id":"t","version":1,"payload":{"source":"s","external_id":"x","project":"p"},"surface":"system"}]}`)) + case "/v1/artifacts/abc": + seen["artifact"] = true + _, _ = w.Write([]byte(`{"meta":{"id":"x"}}`)) + default: + t.Errorf("unexpected path %s", r.URL.Path) + w.WriteHeader(404) + } + })) + defer s.Close() + c := Client{BaseURL: s.URL, WorkerID: "h1", Token: "worker", AdmitToken: "admit"} + if err := c.Register(context.Background(), Worker{ID: "h1"}); err != nil { + t.Fatal(err) + } + es, cur, err := c.Events(context.Background(), 0) + if err != nil || cur != 3 || len(es) != 1 || es[0].Type != "TaskCreated" { + t.Fatalf("events=%v cursor=%d err=%v", es, cur, err) + } + b, err := c.Artifact(context.Background(), "abc") + if err != nil || string(b) != "{\"meta\":{\"id\":\"x\"}}" { + t.Fatalf("artifact=%s err=%v", b, err) + } + for _, k := range []string{"register", "events", "artifact"} { + if !seen[k] { + t.Errorf("%s not seen", k) + } + } +} diff --git a/internal/federation/federation.go b/internal/federation/federation.go index 760b00a..dd6e6ca 100644 --- a/internal/federation/federation.go +++ b/internal/federation/federation.go @@ -42,6 +42,7 @@ func (r *Registry) init() { r.cursors = map[string]uint64{} } } + // Register admits a worker. admitToken must match r.AdmitToken whenever one // is configured. Re-registering an ID that's already claimed requires that // worker's own current token, so a caller can't self-declare someone else's @@ -118,6 +119,22 @@ func (r *Registry) Heartbeat(id string) error { r.workers[id] = w return nil } + +// Available refreshes TTL state and reports whether a registered worker owns +// this harness id. Router admission uses it so a reachable TCP bridge alone +// can never make an offline worker eligible for a lease. +func (r *Registry) Available(id string) bool { + r.mu.Lock() + defer r.mu.Unlock() + r.init() + w, ok := r.workers[id] + if !ok { + return false + } + w.Online = time.Since(w.LastSeen) <= r.TTL + r.workers[id] = w + return w.Online +} func (r *Registry) Snapshot() []Worker { r.mu.Lock() defer r.mu.Unlock() diff --git a/internal/herdr/adapter.go b/internal/herdr/adapter.go index ba7d365..eb78ce8 100644 --- a/internal/herdr/adapter.go +++ b/internal/herdr/adapter.go @@ -75,6 +75,10 @@ type CLIAdapter struct { // release. Nil disables Release (adapters built without one refuse // loudly rather than skip validation). CAS continuity.CAS + // Remote, when set by a federation worker, is pushed after the scratch + // commit and before the pane claim is released. Git is the cross-machine + // transport; a CAS handoff must never point at an unpushed anchor. + Remote string } // HandoffFile is the convention the agent writes its §6.1 handoff to before @@ -83,8 +87,9 @@ type CLIAdapter struct { // uploads the one the agent wrote (herdr does not write handoffs, §6.1). const HandoffFile = ".orchestra-handoff.json" -// HandoffReportFile is the only handoff artifact an opaque harness authors. -// The worker which owns the checkout derives and seals the canonical JSON. +// HandoffReportFile holds the agent's small, labelled answer during release. +// It is not a report: the worker parses it, derives the protocol facts, and +// seals the resulting canonical JSON. const HandoffReportFile = ".orchestra-handoff-report.md" func (a CLIAdapter) CreateWorktree(ctx context.Context, repo, root, taskID string) (string, error) { @@ -149,9 +154,16 @@ func (a CLIAdapter) Bootstrap(ctx context.Context, s Session, ref string) error return a.prompt(ctx, s, fmt.Sprintf(bootstrapPrompt, ref), time.Minute) } -const handoffPrompt = `Orchestra is about to rotate this task to a fresh session (context budget reached). -Before you stop, write a concise semantic handoff report to ` + HandoffReportFile + ` at the worktree root: what changed, validation evidence, remaining work, review findings, and dead ends/open questions. -Do not write protocol JSON, git anchors, or file hashes; Orchestra's checkout worker collects and validates those facts. Do not edit TASK.md. Once written, stop normally.` +const handoffPrompt = `Orchestra is about to rotate this task. Write ONLY the following labelled answers to ` + HandoffReportFile + `, then stop. Output nothing else. + +NEXT: the single next action (one line). +WHY: why that is next (one line). +REMAINING: outstanding items, one line each. If none: NONE. +DEAD ENDS: approaches tried that failed — "tried X → failed because Y", one per line. If none: NONE. +OPEN Q: unresolved decisions, one line each. If none: NONE. +LEARNED: constraints discovered that are NOT in TASK.md, one line each. If none: NONE. + +Do NOT include: what you completed (the diff shows it), the goal or done-criteria (TASK.md holds them), git SHAs/branches/paths, or a prose summary. No headings and no report. Do not edit TASK.md.` // RequestHandoff prompts the agent to write HandoffFile before Release reads // it. Optional capability: adapters without a live pane (tests, etc.) can @@ -183,7 +195,7 @@ func (a CLIAdapter) RequestHandoffReason(ctx context.Context, s Session, reason case "milestone": sb.WriteString("A coherent unit of work looks complete (a successful commit). If the next step is independent of what you just did, this is a good point to hand off.\n") } - fmt.Fprintf(&sb, "Before you stop, write a concise semantic handoff report to %s at the worktree root (reason: %q)", HandoffReportFile, reason) + fmt.Fprintf(&sb, "Before you stop, write the labelled handoff answers requested below to %s at the worktree root (reason: %q).\n\n%s", HandoffReportFile, reason, handoffPrompt[strings.Index(handoffPrompt, "NEXT:"):]) if len(deadEnds) > 0 { sb.WriteString(" and a dead_ends entry for each of the following:\n") for _, d := range deadEnds { @@ -192,7 +204,7 @@ func (a CLIAdapter) RequestHandoffReason(ctx context.Context, s Session, reason } else { sb.WriteString(".\n") } - sb.WriteString("Include what changed, validation evidence, remaining work, review findings, and any dead ends. Do not write protocol JSON, git anchors, or file hashes; Orchestra collects those. Do not edit TASK.md. Once written, stop normally.") + sb.WriteString("Do not add a prose summary, completed-work narration, or protocol JSON. Once written, stop normally.") return a.prompt(ctx, s, sb.String(), time.Minute) } @@ -234,9 +246,9 @@ func (a CLIAdapter) NotifyConventionsChanged(ctx context.Context, s Session) err return a.prompt(ctx, s, conventionsPrompt, time.Minute) } -// Release reads the §6.1 handoff the agent wrote to HandoffFile at the -// worktree root, validates its schema and anchor against the worktree's real -// HEAD, uploads it to CAS, and only then releases herdr's claim on the pane +// Release reads the semantic report the agent wrote at the worktree root, +// derives and validates the canonical handoff from the worktree's real Git +// state, uploads it to CAS, and only then releases herdr's claim on the pane // via the real pane.release_agent(pane_id, source, agent) method (confirmed // live against herdr, AUDIT.md Phase 0 — the invented "pane.release" never // existed and could never have returned a handoff_ref regardless, since @@ -256,7 +268,7 @@ func (a CLIAdapter) Release(ctx context.Context, s Session) (string, error) { if strings.TrimSpace(string(b)) == "" { return "", fmt.Errorf("adapter: semantic handoff report is empty") } - h, err := canonicalHandoff(s, string(b)) + h, err := canonicalHandoff(s, string(b), a.lastObservedCommand(s)) if err != nil { return "", err } @@ -270,6 +282,20 @@ func (a CLIAdapter) Release(ctx context.Context, s Session) (string, error) { return "", fmt.Errorf("adapter: handoff dirty file changed since it was written: %s", d.Path) } } + // The semantic report is transferred in the CAS handoff, not in the + // scratch checkout. Keeping it in the scratch commit makes a successor + // mistake the predecessor's report for a newly requested handoff and can + // cause an immediate release/pickup loop. + dirty := h.Anchor.Dirty[:0] + for _, d := range h.Anchor.Dirty { + if filepath.Clean(d.Path) != HandoffReportFile { + dirty = append(dirty, d) + } + } + h.Anchor.Dirty = dirty + if err := os.Remove(path); err != nil && !os.IsNotExist(err) { + return "", fmt.Errorf("adapter: remove transferred semantic report: %w", err) + } // Atomically commit whatever the handoff described as dirty onto a // per-task scratch branch (§6.2 step 3) *before* uploading, so the // successor's pickup validation collapses to a single HEAD compare @@ -279,6 +305,11 @@ func (a CLIAdapter) Release(ctx context.Context, s Session) (string, error) { if err := continuity.ScratchCommit(s.Worktree, branch, "orchestra: pre-release WIP snapshot ("+h.Meta.ID+")"); err != nil { return "", fmt.Errorf("adapter: scratch commit: %w", err) } + if a.Remote != "" { + if err := continuity.ScratchPush(s.Worktree, branch, a.Remote); err != nil { + return "", fmt.Errorf("adapter: push scratch branch: %w", err) + } + } newSHA, err := HeadSHA(s.Worktree) if err != nil { return "", fmt.Errorf("adapter: read scratch HEAD: %w", err) @@ -302,8 +333,13 @@ func (a CLIAdapter) Release(ctx context.Context, s Session) (string, error) { } // canonicalHandoff keeps Git-derived protocol facts on the worker that owns -// the checkout. The harness contributes only the semantic report (B17). -func canonicalHandoff(s Session, report string) (continuity.Handoff, error) { +// the checkout. Every authored field comes from the validated agent answer; +// it never fabricates task intent or a circular next action. +func canonicalHandoff(s Session, answer, command string) (continuity.Handoff, error) { + authored, err := parseHandoffAnswer(answer) + if err != nil { + return continuity.Handoff{}, fmt.Errorf("adapter: invalid handoff answer: %w", err) + } sha, err := HeadSHA(s.Worktree) if err != nil { return continuity.Handoff{}, fmt.Errorf("adapter: read worktree HEAD: %w", err) @@ -317,16 +353,131 @@ func canonicalHandoff(s Session, report string) (continuity.Handoff, error) { return continuity.Handoff{}, err } return continuity.Handoff{ - Meta: continuity.Meta{ID: handoffID(s), Reason: "threshold"}, - Anchor: continuity.Anchor{GitSHA: sha, Branch: strings.TrimSpace(string(branchOut)), Dirty: dirty}, - Goal: "Continue Orchestra task " + s.PaneID, - DoneWhen: []string{"Task completion is reported to Orchestra"}, - Action: "Read the semantic handoff report and continue the task.", - Command: "cat " + HandoffReportFile, - Remaining: []string{report}, + Meta: continuity.Meta{ID: handoffID(s), Reason: handoffReason(s)}, + Anchor: continuity.Anchor{GitSHA: sha, Branch: strings.TrimSpace(string(branchOut)), Dirty: dirty}, + Action: authored.Action, + Command: command, + Remaining: authored.Remaining, + DeadEnds: authored.DeadEnds, + OpenQuestions: authored.OpenQuestions, + Learned: authored.Learned, }, nil } +type handoffAnswer struct { + Action, Why string + Remaining []string + DeadEnds []continuity.DeadEnd + OpenQuestions, Learned []string +} + +func parseHandoffAnswer(answer string) (handoffAnswer, error) { + var out handoffAnswer + sections := map[string][]string{} + var current string + for _, raw := range strings.Split(strings.ReplaceAll(answer, "\r\n", "\n"), "\n") { + line := strings.TrimSpace(raw) + if line == "" { + continue + } + for _, name := range []string{"NEXT", "WHY", "REMAINING", "DEAD ENDS", "OPEN Q", "LEARNED"} { + prefix := name + ":" + if strings.HasPrefix(line, prefix) { + current = name + if value := strings.TrimSpace(strings.TrimPrefix(line, prefix)); value != "" { + sections[name] = append(sections[name], value) + } + goto parsed + } + } + if current == "" { + return out, fmt.Errorf("unexpected line %q", line) + } + sections[current] = append(sections[current], strings.TrimSpace(strings.TrimPrefix(line, "- "))) + parsed: + } + for _, name := range []string{"NEXT", "WHY", "REMAINING", "DEAD ENDS", "OPEN Q", "LEARNED"} { + if len(sections[name]) == 0 { + return out, fmt.Errorf("missing %s", name) + } + } + if len(sections["NEXT"]) != 1 || len(sections["WHY"]) != 1 { + return out, fmt.Errorf("NEXT and WHY each require one line") + } + out.Action = sections["NEXT"][0] + " — " + sections["WHY"][0] + for _, name := range []string{"REMAINING", "OPEN Q", "LEARNED"} { + values, err := answerList(sections[name]) + if err != nil { + return out, fmt.Errorf("%s: %w", name, err) + } + switch name { + case "REMAINING": + out.Remaining = values + case "OPEN Q": + out.OpenQuestions = values + case "LEARNED": + out.Learned = values + } + } + deadEnds, err := answerList(sections["DEAD ENDS"]) + if err != nil { + return out, fmt.Errorf("DEAD ENDS: %w", err) + } + for _, item := range deadEnds { + parts := strings.SplitN(item, "→", 2) + if len(parts) != 2 { + return out, fmt.Errorf("DEAD ENDS: want 'tried X → failed because Y'") + } + if !strings.HasPrefix(parts[0], "tried ") || !strings.HasPrefix(strings.TrimSpace(parts[1]), "failed because ") { + return out, fmt.Errorf("DEAD ENDS: want 'tried X → failed because Y'") + } + tried := strings.TrimSpace(strings.TrimPrefix(parts[0], "tried ")) + why := strings.TrimSpace(strings.TrimPrefix(parts[1], "failed because ")) + if tried == "" || why == "" { + return out, fmt.Errorf("DEAD ENDS: want 'tried X → failed because Y'") + } + out.DeadEnds = append(out.DeadEnds, continuity.DeadEnd{Tried: tried, WhyFailed: why}) + } + if err := (continuity.Handoff{Meta: continuity.Meta{ID: "answer", Reason: "manual"}, Anchor: continuity.Anchor{GitSHA: strings.Repeat("0", 40), Branch: "answer"}, Action: out.Action, Remaining: out.Remaining, DeadEnds: out.DeadEnds, OpenQuestions: out.OpenQuestions, Learned: out.Learned}).Validate(); err != nil { + return out, err + } + return out, nil +} + +func answerList(lines []string) ([]string, error) { + if len(lines) == 1 && lines[0] == "NONE" { + return nil, nil + } + for _, line := range lines { + if line == "NONE" { + return nil, fmt.Errorf("NONE must be the only value") + } + } + return lines, nil +} + +func (a CLIAdapter) lastObservedCommand(s Session) string { + calls, err := a.Activity(context.Background(), s) + if err != nil { + return "" + } + for i := len(calls) - 1; i >= 0; i-- { + if calls[i].Kind == "command" { + return calls[i].Key + } + } + return "" +} + +func handoffReason(s Session) string { + switch s.HandoffReason { + case "threshold", "milestone", "thrash", "manual": + return s.HandoffReason + default: + return "threshold" + } +} + func handoffID(s Session) string { id := s.AgentName if id == "" { @@ -368,11 +519,12 @@ func dirtyFiles(root string) ([]continuity.Dirty, error) { return dirty, nil } -func agentForSession(s Session, fallback string) string { - if s.AgentName != "" { - return s.AgentName - } - return fallback // compatibility with session records created before B16 +func agentForSession(_ Session, fallback string) string { + // pane.release_agent identifies the harness binding, not herdr's + // machine-global terminal name. AgentName is only for prompt routing; + // passing it here is accepted by herdr but leaves the binding intact. + // Keep the configured harness for both new and pre-B16 session records. + return fallback } func (a CLIAdapter) Kill(ctx context.Context, s Session) error { return a.Client.Call(ctx, "pane.close", map[string]any{"pane_id": s.PaneID}, nil) diff --git a/internal/herdr/adapter_test.go b/internal/herdr/adapter_test.go index 1c1540c..e1b4cfd 100644 --- a/internal/herdr/adapter_test.go +++ b/internal/herdr/adapter_test.go @@ -3,8 +3,6 @@ package herdr import ( "bufio" "context" - "crypto/sha256" - "encoding/hex" "encoding/json" "net" "orchestra/internal/continuity" @@ -61,15 +59,20 @@ func fakeHerdr(t *testing.T) *Client { func validHandoff(anchorSHA string) continuity.Handoff { return continuity.Handoff{ - Meta: continuity.Meta{ID: "t1", Reason: "threshold", RotationIndex: 1}, - Anchor: continuity.Anchor{GitSHA: anchorSHA, Branch: "main"}, - Goal: "finish the thing", - DoneWhen: []string{"tests pass"}, - Action: "continue", - Command: "go test ./...", + Meta: continuity.Meta{ID: "t1", Reason: "threshold", RotationIndex: 1}, + Anchor: continuity.Anchor{GitSHA: anchorSHA, Branch: "main"}, + Action: "run the focused tests", + Command: "go test ./...", } } +const validAnswer = `NEXT: run the focused tests +WHY: confirm the current implementation before changing it +REMAINING: NONE +DEAD ENDS: NONE +OPEN Q: NONE +LEARNED: NONE` + func TestReleaseUploadsHandoffAndReleasesAgent(t *testing.T) { repo := t.TempDir() runGit(t, repo, "init") @@ -81,11 +84,7 @@ func TestReleaseUploadsHandoffAndReleasesAgent(t *testing.T) { t.Fatal(err) } - b, err := continuity.Encode(validHandoff(head)) - if err != nil { - t.Fatal(err) - } - if err := os.WriteFile(filepath.Join(repo, HandoffReportFile), b, 0644); err != nil { + if err := os.WriteFile(filepath.Join(repo, HandoffReportFile), []byte(validAnswer), 0644); err != nil { t.Fatal(err) } @@ -105,8 +104,77 @@ func TestReleaseUploadsHandoffAndReleasesAgent(t *testing.T) { if err != nil { t.Fatal(err) } - if got.Anchor.GitSHA == head { - t.Fatal("semantic report should be snapshotted before publication") + if got.Anchor.GitSHA != head { + t.Fatal("semantic report must not be included in the successor scratch anchor") + } + if _, err := os.Stat(filepath.Join(repo, HandoffReportFile)); !os.IsNotExist(err) { + t.Fatalf("transferred semantic report still present: %v", err) + } +} + +func TestCanonicalHandoffCarriesCoordinatorReason(t *testing.T) { + repo := t.TempDir() + runGit(t, repo, "init") + runGit(t, repo, "config", "user.email", "t@t") + runGit(t, repo, "config", "user.name", "t") + runGit(t, repo, "commit", "--allow-empty", "-m", "init") + if err := os.WriteFile(filepath.Join(repo, HandoffReportFile), []byte(validAnswer), 0644); err != nil { + t.Fatal(err) + } + cas := &memCAS{} + a := CLIAdapter{Client: fakeHerdr(t), Harness: "claude", CAS: cas} + ref, err := a.Release(context.Background(), Session{PaneID: "p1", Worktree: repo, HandoffReason: "thrash"}) + if err != nil { + t.Fatal(err) + } + got, err := continuity.Load(ref, cas) + if err != nil { + t.Fatal(err) + } + if got.Meta.Reason != "thrash" { + t.Fatalf("canonical reason = %q, want thrash", got.Meta.Reason) + } +} + +func TestReleaseUsesHarnessBindingNotDisplayName(t *testing.T) { + repo := t.TempDir() + runGit(t, repo, "init") + runGit(t, repo, "config", "user.email", "t@t") + runGit(t, repo, "config", "user.name", "t") + runGit(t, repo, "commit", "--allow-empty", "-m", "init") + if err := os.WriteFile(filepath.Join(repo, HandoffReportFile), []byte(validAnswer), 0644); err != nil { + t.Fatal(err) + } + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ln.Close() }) + request := make(chan Request, 1) + go func() { + conn, err := ln.Accept() + if err != nil { + return + } + defer conn.Close() + var req Request + if json.NewDecoder(bufio.NewReader(conn)).Decode(&req) == nil { + request <- req + _ = json.NewEncoder(conn).Encode(Response{ID: req.ID, Result: json.RawMessage(`{}`)}) + } + }() + a := CLIAdapter{Client: &Client{Path: ln.Addr().String(), dial: func() (net.Conn, error) { return net.Dial("tcp", ln.Addr().String()) }}, Harness: "opencode", CAS: &memCAS{}} + if _, err := a.Release(context.Background(), Session{PaneID: "p1", Worktree: repo, AgentName: "oc-task-specific"}); err != nil { + t.Fatal(err) + } + req := <-request + p, _ := json.Marshal(req.Params) + var got struct { + Agent string `json:"agent"` + } + _ = json.Unmarshal(p, &got) + if got.Agent != "opencode" { + t.Fatalf("release agent = %q, want harness binding opencode", got.Agent) } } @@ -119,6 +187,43 @@ func TestReleaseRefusesWithoutHandoffFile(t *testing.T) { } } +func TestReleaseRefusesNarrativeOrCircularHandoffAnswer(t *testing.T) { + repo := t.TempDir() + runGit(t, repo, "init") + runGit(t, repo, "config", "user.email", "t@t") + runGit(t, repo, "config", "user.name", "t") + runGit(t, repo, "commit", "--allow-empty", "-m", "init") + bad := `NEXT: Continue the task from the handoff +WHY: the predecessor asked for it +REMAINING: what changed\n# a markdown report +DEAD ENDS: NONE +OPEN Q: NONE +LEARNED: NONE` + if err := os.WriteFile(filepath.Join(repo, HandoffReportFile), []byte(bad), 0o644); err != nil { + t.Fatal(err) + } + a := CLIAdapter{Client: fakeHerdr(t), Harness: "claude", CAS: &memCAS{}} + if _, err := a.Release(context.Background(), Session{PaneID: "p1", Worktree: repo}); err == nil { + t.Fatal("expected invalid agent answer to refuse release") + } +} + +func TestParseHandoffAnswerPreservesOnlyAuthoredFields(t *testing.T) { + a, err := parseHandoffAnswer(`NEXT: inspect the failing integration test +WHY: isolate the regression before changing production code +REMAINING: fix the assertion after identifying the cause +DEAD ENDS: tried rerunning the whole suite → failed because it obscures the relevant failure +OPEN Q: whether the remote worker has the updated fixture +LEARNED: the fixture requires a committed scratch branch +`) + if err != nil { + t.Fatal(err) + } + if len(a.Remaining) != 1 || len(a.DeadEnds) != 1 || len(a.OpenQuestions) != 1 || len(a.Learned) != 1 { + t.Fatalf("parsed answer lost authored fields: %#v", a) + } +} + func TestReleaseDoesNotTrustAgentSuppliedAnchor(t *testing.T) { repo := t.TempDir() runGit(t, repo, "init") @@ -126,11 +231,7 @@ func TestReleaseDoesNotTrustAgentSuppliedAnchor(t *testing.T) { runGit(t, repo, "config", "user.name", "t") runGit(t, repo, "commit", "--allow-empty", "-m", "init") - b, err := continuity.Encode(validHandoff("deaddeaddeaddeaddeaddeaddeaddeaddeaddead")) - if err != nil { - t.Fatal(err) - } - if err := os.WriteFile(filepath.Join(repo, HandoffReportFile), b, 0644); err != nil { + if err := os.WriteFile(filepath.Join(repo, HandoffReportFile), []byte(validAnswer), 0644); err != nil { t.Fatal(err) } a := CLIAdapter{Client: fakeHerdr(t), Harness: "claude", CAS: &memCAS{}} @@ -154,14 +255,7 @@ func TestReleaseScratchCommitsDirtyFilesBeforeUpload(t *testing.T) { if err := os.WriteFile(wipPath, []byte("in progress"), 0644); err != nil { t.Fatal(err) } - sum := sha256.Sum256([]byte("in progress")) - h := validHandoff(head) - h.Anchor.Dirty = []continuity.Dirty{{Path: "wip.txt", SHA256: hex.EncodeToString(sum[:])}} - b, err := continuity.Encode(h) - if err != nil { - t.Fatal(err) - } - if err := os.WriteFile(filepath.Join(repo, HandoffReportFile), b, 0644); err != nil { + if err := os.WriteFile(filepath.Join(repo, HandoffReportFile), []byte(validAnswer), 0644); err != nil { t.Fatal(err) } @@ -203,22 +297,10 @@ func TestReleaseDoesNotTrustAgentSuppliedDirtyFile(t *testing.T) { runGit(t, repo, "config", "user.email", "t@t") runGit(t, repo, "config", "user.name", "t") runGit(t, repo, "commit", "--allow-empty", "-m", "init") - head, err := HeadSHA(repo) - if err != nil { - t.Fatal(err) - } - if err := os.WriteFile(filepath.Join(repo, "wip.txt"), []byte("changed after handoff written"), 0644); err != nil { t.Fatal(err) } - stale := sha256.Sum256([]byte("original content")) - h := validHandoff(head) - h.Anchor.Dirty = []continuity.Dirty{{Path: "wip.txt", SHA256: hex.EncodeToString(stale[:])}} - b, err := continuity.Encode(h) - if err != nil { - t.Fatal(err) - } - if err := os.WriteFile(filepath.Join(repo, HandoffReportFile), b, 0644); err != nil { + if err := os.WriteFile(filepath.Join(repo, HandoffReportFile), []byte(validAnswer), 0644); err != nil { t.Fatal(err) } a := CLIAdapter{Client: fakeHerdr(t), Harness: "claude", CAS: &memCAS{}} diff --git a/internal/herdr/herdr.go b/internal/herdr/herdr.go index 3566e87..c032989 100644 --- a/internal/herdr/herdr.go +++ b/internal/herdr/herdr.go @@ -160,6 +160,10 @@ type Session struct { // its §6.1 handoff (HandoffFile) — avoids re-sending the same prompt // every tick while Release keeps waiting for the file to appear. HandoffRequested bool `json:"handoff_requested,omitempty"` + // HandoffReason is selected by the coordinator when it asks for the + // semantic report. The checkout worker, rather than the harness, copies + // it into the canonical handoff it seals at release time. + HandoffReason string `json:"handoff_reason,omitempty"` // ConventionsHash is continuity.ConventionsHash of the project's shared // *.md docs (AGENTS.md/CLAUDE.md/VOCAB.md) at the time this session was // last notified of (or started with) their content — §6.3's staleness @@ -455,10 +459,16 @@ func agentName(harness, taskID string) string { id = "session" } name := prefix + "-" + id - if len(name) > 32 { - name = strings.TrimRight(name[:32], "-_") + if len(name) <= 32 { + return name } - return name + // Keeping only the leading task-id characters made distinct long task + // IDs collide in herdr's machine-global name namespace. Reserve a stable + // digest suffix so truncation remains bounded *and* task-specific. + sum := sha256.Sum256([]byte(taskID)) + const suffixLen = 8 + keep := 32 - len(prefix) - 1 - 1 - suffixLen // prefix + "-" + stem + "-" + digest + return prefix + "-" + strings.TrimRight(id[:keep], "-_") + "-" + fmt.Sprintf("%x", sum[:])[:suffixLen] } // harnessStartArgs stays empty for Claude: --dangerously-skip-permissions diff --git a/internal/herdr/herdr_test.go b/internal/herdr/herdr_test.go index c128ddf..a95f8d4 100644 --- a/internal/herdr/herdr_test.go +++ b/internal/herdr/herdr_test.go @@ -89,6 +89,17 @@ func TestAgentNameIsBoundedAndValid(t *testing.T) { } } +func TestAgentNameLongIDsDoNotCollide(t *testing.T) { + first := agentName("claude", "task-with-a-very-long-shared-prefix-aaaaaaaa") + second := agentName("claude", "task-with-a-very-long-shared-prefix-bbbbbbbb") + if first == second { + t.Fatalf("long task IDs collided: %q", first) + } + if len(first) > 32 || len(second) > 32 { + t.Fatalf("agent name exceeds limit: %q / %q", first, second) + } +} + func TestPromptDoesNotRetryAmbiguousDelivery(t *testing.T) { ln, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { diff --git a/internal/integration/end_to_end_test.go b/internal/integration/end_to_end_test.go index ee1ff42..a07b6d0 100644 --- a/internal/integration/end_to_end_test.go +++ b/internal/integration/end_to_end_test.go @@ -147,8 +147,12 @@ func TestEndToEndIngestRouteLeaseRotateAndComplete(t *testing.T) { t.Fatal(err) } got, _ = s.Task(task.ID) - if got.State != domain.StateLeased || got.Version != 4 { - t.Fatalf("pickup state=%s version=%d", got.State, got.Version) + // This fixture intentionally stores an opaque string rather than a typed + // handoff. Once TaskLeased carries a real handoff_ref, pickup correctly + // refuses it instead of silently continuing (the valid pickup contract is + // covered by the orchestrator continuity tests). + if got.State != domain.StateBlocked || got.Version != 5 { + t.Fatalf("invalid pickup state=%s version=%d", got.State, got.Version) } ref, err = s.PutArtifact([]byte("report")) if err != nil { diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go index 0e7c800..e490950 100644 --- a/internal/orchestrator/orchestrator.go +++ b/internal/orchestrator/orchestrator.go @@ -249,6 +249,7 @@ func (c *Coordinator) requestReasonedHandoff(ctx context.Context, taskID string, return } session.HandoffRequested = true + session.HandoffReason = reason c.mu.Lock() c.sessions[taskID] = session _ = c.saveSessionsLocked() @@ -652,6 +653,7 @@ func (c *Coordinator) rotate(ctx context.Context, hard float64) { if _, statErr := os.Stat(filepath.Join(session.Worktree, herdr.HandoffReportFile)); statErr != nil && !session.HandoffRequested { if reqErr := requester.RequestHandoff(ctx, session); reqErr == nil { session.HandoffRequested = true + session.HandoffReason = "threshold" c.mu.Lock() c.sessions[taskID] = session _ = c.saveSessionsLocked() @@ -688,6 +690,7 @@ func (c *Coordinator) rotate(ctx context.Context, hard float64) { if !session.HandoffRequested { if reqErr := requester.RequestHandoff(ctx, session); reqErr == nil { session.HandoffRequested = true + session.HandoffReason = reason c.mu.Lock() c.sessions[taskID] = session _ = c.saveSessionsLocked() @@ -784,6 +787,7 @@ func (c *Coordinator) TurnDecision(ctx context.Context, taskID string) (string, if !session.HandoffRequested { if reqErr := requester.RequestHandoff(ctx, session); reqErr == nil { session.HandoffRequested = true + session.HandoffReason = "threshold" c.mu.Lock() c.sessions[taskID] = session _ = c.saveSessionsLocked() @@ -811,6 +815,7 @@ func (c *Coordinator) TurnDecision(ctx context.Context, taskID string) (string, if !session.HandoffRequested { if reqErr := requester.RequestHandoff(ctx, session); reqErr == nil { session.HandoffRequested = true + session.HandoffReason = "threshold" c.mu.Lock() c.sessions[taskID] = session _ = c.saveSessionsLocked() diff --git a/internal/orchestrator/rotation_test.go b/internal/orchestrator/rotation_test.go index 8f635e0..1588d59 100644 --- a/internal/orchestrator/rotation_test.go +++ b/internal/orchestrator/rotation_test.go @@ -279,7 +279,7 @@ func TestTurnDecision(t *testing.T) { handoff := map[string]any{ "meta": map[string]any{"id": "h2", "reason": "manual", "rotation_index": 0}, "anchor": map[string]any{"git_sha": head, "branch": "orchestra/t1"}, - "goal": "g", "done_when": []string{"x"}, "action": "prompt", "command": "go test ./...", + "action": "run the focused tests", "command": "go test ./...", } b, err := json.Marshal(handoff) if err != nil { @@ -334,7 +334,7 @@ func TestStartBlocksOnInvalidPickup(t *testing.T) { badHandoff := map[string]any{ "meta": map[string]any{"id": "h1", "reason": "manual", "rotation_index": 0}, "anchor": map[string]any{"git_sha": strings0(40, 'a'), "branch": "orchestra/t1"}, - "goal": "g", "done_when": []string{"x"}, "action": "prompt", "command": "go test ./...", + "action": "run the focused tests", "command": "go test ./...", } ref, err := s.PutArtifact(mustJSON(badHandoff)) if err != nil { @@ -708,7 +708,7 @@ func TestActivityTriggersRequestReasonedHandoffWithoutReleasing(t *testing.T) { handoff := map[string]any{ "meta": map[string]any{"id": "h3", "reason": "thrash", "rotation_index": 0}, "anchor": map[string]any{"git_sha": head, "branch": "orchestra/t1"}, - "goal": "g", "done_when": []string{"x"}, "action": "prompt", "command": "go test ./...", + "action": "run the focused tests", "command": "go test ./...", "dead_ends": []map[string]any{{"tried": "go test ./...", "why_failed": "failed 3 times"}}, } b, err := json.Marshal(handoff) diff --git a/internal/store/store.go b/internal/store/store.go index b568525..e6f7b51 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -145,6 +145,7 @@ func (s *Store) apply(e domain.Event) error { case "TaskReleased": t.State = domain.StateQueued t.Lease = nil + t.HandoffRef, _ = p["handoff_ref"].(string) case "TaskCompleted": t.State = domain.StateCompleted t.Lease = nil @@ -398,7 +399,11 @@ func (s *Store) Lease(id, harness string, ttl time.Duration) (domain.Event, erro if t.State != domain.StateQueued { return domain.Event{}, domain.ErrConflict } - p, _ := json.Marshal(map[string]any{"harness_id": harness, "ttl": ttl.Seconds(), "until_ns": time.Now().Add(ttl).UnixNano(), "expected_version": t.Version}) + payload := map[string]any{"harness_id": harness, "ttl": ttl.Seconds(), "until_ns": time.Now().Add(ttl).UnixNano(), "expected_version": t.Version} + if t.HandoffRef != "" { + payload["handoff_ref"] = t.HandoffRef + } + p, _ := json.Marshal(payload) e := domain.Event{ID: domain.NewID(), Type: "TaskLeased", TaskID: id, Version: t.Version + 1, Payload: p, Surface: string(authz.System)} return e, s.Append(e) } diff --git a/internal/store/store_test.go b/internal/store/store_test.go index 58930f2..83e8b98 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -17,6 +17,40 @@ func created(id string) domain.Event { return domain.Event{ID: id, Type: "TaskCreated", TaskID: "task-1", Version: 1, Payload: b, Surface: string(authz.System)} } +func TestLeaseCarriesReleasedHandoffRef(t *testing.T) { + s, err := Open(t.TempDir()) + if err != nil { + t.Fatal(err) + } + b := []byte(`{"source":"s","external_id":"x","project":"p"}`) + if err := s.Append(domain.Event{ID: "create", Type: "TaskCreated", TaskID: "t", Version: 1, Payload: b, Surface: string(authz.System)}); err != nil { + t.Fatal(err) + } + if _, err := s.Lease("t", "h", time.Minute); err != nil { + t.Fatal(err) + } + ref, err := s.PutArtifact([]byte("handoff")) + if err != nil { + t.Fatal(err) + } + task, _ := s.Task("t") + p, _ := json.Marshal(map[string]string{"handoff_ref": ref, "anchor_sha": "0123456789012345678901234567890123456789"}) + if err := s.Append(domain.Event{ID: "release", Type: "TaskReleased", TaskID: "t", Version: task.Version + 1, Payload: p, Surface: string(authz.System)}); err != nil { + t.Fatal(err) + } + e, err := s.Lease("t", "h", time.Minute) + if err != nil { + t.Fatal(err) + } + var got map[string]any + if err := json.Unmarshal(e.Payload, &got); err != nil { + t.Fatal(err) + } + if got["handoff_ref"] != ref { + t.Fatalf("handoff_ref=%v want %s", got["handoff_ref"], ref) + } +} + func TestAppendReplayAndDeduplicate(t *testing.T) { dir := t.TempDir() s, err := Open(dir)