docs: update server implementation gaps

2026-07-26 19:47:08 +04:00
parent 073d5c69b0
commit aa6719ca16
3 changed files with 120 additions and 11 deletions
@@ -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"
@@ -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
 	}
@@ -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.