diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go index 06ae1c9..7ee83ce 100644 --- a/cmd/orchestra/main.go +++ b/cmd/orchestra/main.go @@ -8,7 +8,9 @@ import ( "net/http" "orchestra/internal/authz" "orchestra/internal/domain" + "orchestra/internal/herdr" "orchestra/internal/operations" + "orchestra/internal/orchestrator" "orchestra/internal/provider" "orchestra/internal/registry" "orchestra/internal/router" @@ -36,6 +38,18 @@ func main() { 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}} + if repo, root := os.Getenv("ORCHESTRA_REPO"), os.Getenv("ORCHESTRA_WORKTREE_ROOT"); repo != "" && root != "" { + adapters := map[string]herdr.Adapter{} + for _, h := range rr.Herdrs() { + if h.Address == "" { + continue + } + client := herdr.New(h.Address) + adapters[h.ID] = herdr.Codex(client, 200000) + } + coordinator := &orchestrator.Coordinator{Store: s, Worktrees: orchestrator.GitWorktrees{Root: root, Repo: repo}, Adapters: orchestrator.AdapterFactory{Herdrs: adapters}} + rt.OnLease = func(e domain.Event) error { return coordinator.Start(context.Background(), e) } + } } mux := http.NewServeMux() surface := func(r *http.Request) authz.Surface { diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go index 7d4c87f..a897dc2 100644 --- a/internal/orchestrator/orchestrator.go +++ b/internal/orchestrator/orchestrator.go @@ -10,6 +10,9 @@ import ( "orchestra/internal/domain" "orchestra/internal/herdr" "orchestra/internal/store" + "os" + "os/exec" + "path/filepath" "sync" ) @@ -20,6 +23,43 @@ type Adapters interface { Adapter(string) (herdr.Adapter, error) } +// GitWorktrees creates one isolated checkout per task. The root is expected +// to be a clone containing the project's remote; callers may set a separate +// root per deployment. +type GitWorktrees struct { + Root string + Repo string +} + +func (w GitWorktrees) Create(ctx context.Context, t domain.Task) (string, error) { + if w.Root == "" || w.Repo == "" { + return "", fmt.Errorf("worktree: root and repo required") + } + if err := os.MkdirAll(w.Root, 0755); err != nil { + return "", err + } + p := filepath.Join(w.Root, t.ID) + if _, err := os.Stat(p); err == nil { + return p, nil + } + branch := "orchestra/" + t.ID + cmd := exec.CommandContext(ctx, "git", "-C", w.Repo, "worktree", "add", "-b", branch, p, "HEAD") + if out, err := cmd.CombinedOutput(); err != nil { + return "", fmt.Errorf("%s: %w", string(out), err) + } + return p, nil +} + +type AdapterFactory struct{ Herdrs map[string]herdr.Adapter } + +func (f AdapterFactory) Adapter(id string) (herdr.Adapter, error) { + a, ok := f.Herdrs[id] + if !ok { + return nil, fmt.Errorf("adapter %q not registered", id) + } + return a, nil +} + type Coordinator struct { Store *store.Store Worktrees Worktrees diff --git a/internal/registry/registry.go b/internal/registry/registry.go index 4702c80..eae749b 100644 --- a/internal/registry/registry.go +++ b/internal/registry/registry.go @@ -109,7 +109,15 @@ func putID[T any](m map[string]T, id, kind string) error { func (r Registry) Project(id string) (Project, bool) { p, ok := r.projects[id]; return p, ok } func (r Registry) Machine(id string) (Machine, bool) { m, ok := r.machines[id]; return m, ok } func (r Registry) Herdr(id string) (Herdr, bool) { h, ok := r.herdrs[id]; return h, ok } -func (r Registry) Projects() []Project { return projects(r.projects) } +func (r Registry) Herdrs() []Herdr { + out := make([]Herdr, 0, len(r.herdrs)) + for _, h := range r.herdrs { + out = append(out, h) + } + sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID }) + return out +} +func (r Registry) Projects() []Project { return projects(r.projects) } func projects(m map[string]Project) []Project { out := make([]Project, 0, len(m)) for _, v := range m { diff --git a/internal/router/router.go b/internal/router/router.go index 84eaafe..8d75fb6 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -31,6 +31,7 @@ type Router struct { Now func() time.Time backoff map[string]time.Time attempts map[string]int + OnLease func(domain.Event) error } func (r *Router) init() { @@ -99,6 +100,11 @@ func (r *Router) AssignPending() ([]domain.Event, error) { } r.attempts[t.ID]++ out = append(out, e) + if r.OnLease != nil { + if err := r.OnLease(e); err != nil { + return out, err + } + } break } } diff --git a/progress.md b/progress.md index 62f1955..f54839b 100644 --- a/progress.md +++ b/progress.md @@ -6,7 +6,7 @@ Updated: 2026-07-26 `go test ./...` passes, but the implementation is still a tested substrate/router prototype rather than a functioning unattended multi-harness orchestra. The following gaps were verified against `orchestra-spec (1).md` and the current code: -- **Harness execution is only partially wired.** `internal/orchestrator` now provides lease → worktree → adapter session → optional bootstrap coordination, but `main.go` does not yet construct concrete worktree/adapter registries or run monitoring callbacks. +- **Harness execution is wired for configured Git/Codex deployments.** The router invokes the coordinator; `ORCHESTRA_REPO` + `ORCHESTRA_WORKTREE_ROOT` enable Git worktree creation, configured herdr socket adapters, session creation, and optional bootstrap. Other harness types and monitoring callbacks remain pending. - **Rotation is not implemented.** There is no turn-boundary callback, herdr event subscription, occupancy-triggered rotation, milestone/thrash trigger, or split-then-close coordinator. - **Lifecycle API payloads are invalid.** The release, complete, and block endpoints all emit `{"source":"api"}`, while validation requires `handoff_ref` or `reason`, `report_ref`, and `blocker` respectively. The documented lifecycle endpoints therefore cannot complete successfully. - **Provider integrations are partially wired.** JSONL watching and optional Gitea webhook/poll loops now start from environment configuration, but terminal reflection, provider health, cancellation, and delivery fan-out remain absent. @@ -21,7 +21,7 @@ The first pass closed the store/API defects (lifecycle defaults, CAS content ver ### Remaining server-side gaps -- **The orchestration coordinator is not fully operational.** A reusable coordinator now resolves a worktree, invokes `herdr.Adapter.Lease`, bootstraps handoffs, and blocks failed starts. It is not yet constructed from deployment configuration, persisted, or connected to turn/lifecycle monitoring in `main.go`. +- **The orchestration coordinator is operational but incomplete.** It is constructed from deployment configuration, resolves a worktree, invokes `herdr.Adapter.Lease`, bootstraps handoffs, and blocks failed starts. Session mappings are in-memory and turn/lifecycle monitoring is still pending. - **Rotation is still absent.** No adapter turn-boundary callback, occupancy trigger, milestone/thrash trigger, handoff save/validate flow, split-then-close sequence, or lease transfer coordinator is wired into the server. The existing occupancy readers and anchor validator are standalone library primitives. - **Harness discovery and registration are not operational.** Static herdr configuration and socket clients exist, but startup does not create adapters, ping configured herdrs, discover active Codex sessions, subscribe to opencode SSE, or run the required fallback/TTL monitoring loop. - **Provider ingestion is only partially wired.** JSONL and Gitea are available when configured, but there is no provider lifecycle management, cancellation, error health projection, terminal-state reflection, or provider fan-out.