diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go index b5a0f84..8ebf61b 100644 --- a/cmd/orchestra/main.go +++ b/cmd/orchestra/main.go @@ -1,6 +1,7 @@ package main import ( + "context" "encoding/json" "fmt" "log" @@ -8,6 +9,7 @@ import ( "orchestra/internal/authz" "orchestra/internal/domain" "orchestra/internal/operations" + "orchestra/internal/provider" "orchestra/internal/registry" "orchestra/internal/router" "orchestra/internal/store" @@ -173,7 +175,31 @@ func main() { return } types := map[string]string{"release": "TaskReleased", "complete": "TaskCompleted", "block": "TaskBlocked"} - e = domain.Event{ID: id(), Type: types[action], TaskID: taskID, Version: t.Version + 1, Payload: json.RawMessage(`{"source":"api"}`)} + var p map[string]any + if r.Body != nil { + if json.NewDecoder(r.Body).Decode(&p) != nil { + p = map[string]any{} + } + } + if p == nil { + p = map[string]any{} + } + if action == "release" && p["reason"] == nil && p["handoff_ref"] == nil { + p["reason"] = "released_by_api" + } + if action == "block" && p["blocker"] == nil { + p["blocker"] = "blocked_by_api" + } + if action == "complete" && p["report_ref"] == nil { + ref, putErr := s.PutArtifact([]byte("completed without an attached report\n")) + if putErr != nil { + http.Error(w, putErr.Error(), 500) + return + } + p["report_ref"] = ref + } + ePayload, _ := json.Marshal(p) + e = domain.Event{ID: id(), Type: types[action], TaskID: taskID, Version: t.Version + 1, Payload: ePayload} err = s.Append(e) default: http.Error(w, "unknown action", 404) @@ -204,6 +230,21 @@ func main() { }() } mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.Write([]byte("ok\n")) }) + if base := os.Getenv("ORCHESTRA_GITEA_URL"); base != "" { + g := provider.Gitea{BaseURL: base, Token: os.Getenv("ORCHESTRA_GITEA_TOKEN"), WebhookSecret: os.Getenv("ORCHESTRA_GITEA_WEBHOOK_SECRET"), Owner: os.Getenv("ORCHESTRA_GITEA_OWNER"), Repo: os.Getenv("ORCHESTRA_GITEA_REPO")} + mux.Handle("/v1/providers/gitea/webhook", g.WebhookHandler(s)) + go func() { + for { + ctx, cancel := context.WithTimeout(context.Background(), time.Minute) + _, e := g.Poll(ctx, s) + cancel() + if e != nil { + log.Printf("gitea poll: %v", e) + } + time.Sleep(time.Minute) + } + }() + } port := os.Getenv("ORCHESTRA_PORT") if port == "" { port = "9145" diff --git a/internal/store/store.go b/internal/store/store.go index e9aec55..1d8d665 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -20,6 +20,7 @@ type Store struct { tasks map[string]domain.Task external map[string]string snapshot string + seq uint64 } func Open(dir string) (*Store, error) { @@ -30,6 +31,23 @@ func Open(dir string) (*Store, error) { if err := os.MkdirAll(s.cas, 0755); err != nil { return nil, err } + var snapshotSeq uint64 + if b, readErr := os.ReadFile(s.snapshot); readErr == nil { + var snap struct { + Seq uint64 `json:"seq"` + Tasks []domain.Task `json:"tasks"` + } + if json.Unmarshal(b, &snap) != nil { + return nil, fmt.Errorf("invalid snapshot") + } + for _, t := range snap.Tasks { + s.tasks[t.ID] = t + s.external[t.Source+"\x00"+t.ExternalID] = t.ID + } + snapshotSeq = snap.Seq + } else if !errors.Is(readErr, os.ErrNotExist) { + return nil, readErr + } f, err := os.Open(s.path) if os.IsNotExist(err) { return s, nil @@ -39,13 +57,31 @@ func Open(dir string) (*Store, error) { } defer f.Close() sc := bufio.NewScanner(f) + var expected uint64 = snapshotSeq + 1 for sc.Scan() { var e domain.Event if err := json.Unmarshal(sc.Bytes(), &e); err == nil { + if err := domain.ValidateEvent(e); err != nil { + return nil, err + } + if e.Seq < expected { + continue + } + if e.Seq != expected { + return nil, fmt.Errorf("event sequence gap: got %d, want %d", e.Seq, expected) + } + if t, ok := s.tasks[e.TaskID]; ok && e.Version != t.Version+1 { + return nil, domain.ErrConflict + } + if _, ok := s.tasks[e.TaskID]; !ok && e.Type != "TaskCreated" { + return nil, domain.ErrNotFound + } s.events = append(s.events, e) + s.seq = e.Seq if err := s.apply(e); err != nil { return nil, err } + expected++ } else { return nil, err } @@ -64,6 +100,23 @@ func (s *Store) apply(e domain.Event) error { return err } t = domain.Task{ID: e.TaskID, Source: p["source"].(string), ExternalID: p["external_id"].(string), Project: p["project"].(string), State: domain.StateQueued} + if v, ok := p["parent"].(string); ok { + t.Parent = v + } + if v, ok := p["inherent_priority"].(float64); ok { + t.InherentPriority = int(v) + } + if v, ok := p["due"].(string); ok { + if d, err := time.Parse(time.RFC3339, v); err == nil { + t.Due = &d + } + } + if v, ok := p["estimate"].(map[string]any); ok { + t.Estimate = &domain.Estimate{} + t.Estimate.Value, _ = v["value"].(float64) + t.Estimate.Who, _ = v["who"].(string) + t.Estimate.Confidence, _ = v["confidence"].(float64) + } if v, ok := p["capability"].([]any); ok { for _, x := range v { if z, ok := x.(string); ok { @@ -111,7 +164,7 @@ func (s *Store) Append(e domain.Event) error { e.At = time.Now().UTC() } if e.Seq == 0 { - e.Seq = uint64(len(s.events) + 1) + e.Seq = s.seq + 1 } if e.Type == "TaskCreated" { var p map[string]any @@ -133,7 +186,7 @@ func (s *Store) Append(e domain.Event) error { _ = json.Unmarshal(e.Payload, &p) for _, k := range []string{"handoff_ref", "report_ref"} { if ref, ok := p[k].(string); ok { - if _, err := os.Stat(filepath.Join(s.cas, ref)); err != nil { + if _, err := s.Artifact(ref); err != nil { return fmt.Errorf("%w: missing artifact %s", domain.ErrInvalid, ref) } } @@ -155,6 +208,7 @@ func (s *Store) Append(e domain.Event) error { return err } s.events = append(s.events, e) + s.seq = e.Seq if err := s.writeSnapshot(); err != nil { return err } @@ -168,7 +222,7 @@ func (s *Store) writeSnapshot() error { b, err := json.Marshal(struct { Seq uint64 `json:"seq"` Tasks []domain.Task `json:"tasks"` - }{uint64(len(s.events)), tasks}) + }{s.seq, tasks}) if err != nil { return err } diff --git a/progress.md b/progress.md index cdfdb41..1e3bae3 100644 --- a/progress.md +++ b/progress.md @@ -17,14 +17,28 @@ Updated: 2026-07-26 - **Occupancy support is incomplete relative to the spec.** Codex active-session discovery and opencode's server/SSE plus fallback path are not implemented. - **Authorization is only partially enforced.** HTTP method restrictions exist, but handlers do not consistently call `AuthorizeEvent`; an absent surface defaults to full-control Web. +The first pass closed the store/API defects (lifecycle defaults, CAS content verification, validated replay, snapshot loading, and projection of task metadata) and added optional Gitea webhook/poll wiring. The remaining server-side gaps are below. + +### Remaining server-side gaps + +- **The orchestration coordinator is still absent.** A `TaskLeased` event only changes the projection. The server does not resolve a worktree, invoke `herdr.Adapter.Lease`, bootstrap the session, or translate adapter/session failures into lifecycle events. There is no durable session/lease-to-pane mapping. +- **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.** Gitea is available when its environment is configured, but JSONL watching is not started by `main.go`; there is no provider lifecycle management, cancellation, or error health projection. Gitea terminal-state reflection and provider fan-out are not connected to server routes or background workers. +- **Lifecycle event contracts remain incomplete.** Validation does not enforce the spec's `expected_version`, `ttl`, `anchor_sha`, `receipt`, or optional `handoff_ref` relationships, and the HTTP API does not validate actor/surface authorization at the event construction site. Completion without a report currently creates a generated placeholder artifact rather than requiring the stop-hook/wrapper receipt described by the spec. +- **Quota and standup scheduling are not implemented.** The event types and brief fields are accepted, but there is no per-harness/window quota projection, conservative availability filter, 3am safety behavior, or scheduled standup advisory producer. +- **Brief delivery and provider reflection are not implemented.** `/v1/brief` is read-only and computes local git state, but no Telegram/ntfy delivery, Gitea terminal reflection, Maven subscription, or cross-surface approval subscriber is started by the server. +- **Federated worker behavior is not complete.** Machine affinity filtering is implemented, but there is no worker registration/heartbeat protocol, remote event transport, cross-machine worktree coordination, or server-side synchronization status beyond local git inspection. +- **The server API is narrower than the spec.** There are no explicit task amendment/report/handoff upload endpoints, event subscription/streaming endpoint, health/readiness detail for providers and herdrs, or administrative endpoints for project/machine/herdr status. + Recommended order: -1. Fix lifecycle payloads and add endpoint tests. -2. Add the orchestration coordinator: lease → worktree → harness session → bootstrap → lifecycle events. -3. Implement rotation and turn-boundary monitoring. -4. Wire provider webhook and polling integrations. -5. Harden replay and CAS verification. -6. Reconcile the status/checklist sections below with the actual implementation. +1. Add the orchestration coordinator: lease → worktree → harness session → bootstrap → lifecycle events. +2. Implement rotation and turn-boundary monitoring. +3. Wire provider lifecycle management, JSONL startup, terminal reflection, and delivery integrations. +4. Add quota/standup projections and conservative availability filtering. +5. Add federated worker health/synchronization and the remaining control-plane API surface. +6. Add endpoint/contract tests for the coordinator and lifecycle receipts. ## Server implementation checklist @@ -115,7 +129,7 @@ Item 1 (task schema + provider port + JSONL adapter) is implemented as the basel ## Important limitations -- This is still a Layer 1 prototype. No harness adapters, herdr socket integration, rotation, handoff validation, approvals, TUI/web, quota projection, or morning brief exists yet. +- This is still a Layer 1/2 prototype. Harness adapter and continuity primitives exist, but unattended orchestration, rotation, quota accounting, and delivery integrations are not server-wired. - Surface authorization is enforced by the shared HTTP/bus policy; set `ORCHESTRA_*_TOKEN` variables to require bearer authentication per surface. - Event payload validation currently checks required fields and primitive types; replace the remaining map-based application logic with typed payload structs before exposing the API beyond the homelab. - Router retry counts/backoff and terminal `TaskFailed` are implemented; retry policy is currently configured in server wiring.