From ae4d3a57bcf5d06cfaee927ab3deb5ece582432d Mon Sep 17 00:00:00 2001 From: kami Date: Sun, 26 Jul 2026 20:30:19 +0400 Subject: [PATCH] supervise providers and reflect terminal state --- cmd/orchestra/main.go | 45 ++++++++++++------- internal/provider/provider.go | 84 +++++++++++++++++++++++++++++++++++ progress.md | 2 + 3 files changed, 114 insertions(+), 17 deletions(-) diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go index 56b100c..2eb61f7 100644 --- a/cmd/orchestra/main.go +++ b/cmd/orchestra/main.go @@ -80,6 +80,7 @@ func main() { } } mux := http.NewServeMux() + providerHealth := map[string]*provider.Supervisor{} surface := func(r *http.Request) authz.Surface { v := authz.ParseSurface(r.Header.Get("X-Orchestra-Surface")) if v == "" { @@ -321,29 +322,39 @@ func main() { } json.NewEncoder(w).Encode(map[string]any{"ready": ready, "checks": checks}) }) + mux.HandleFunc("/v1/providers/health", func(w http.ResponseWriter, r *http.Request) { + out := map[string]provider.Health{} + for name, sup := range providerHealth { + out[name] = sup.Health() + } + json.NewEncoder(w).Encode(out) + }) 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) + reflecting := provider.ReflectingSink{Sink: s, Tasks: s, Reflector: g} + mux.Handle("/v1/providers/gitea/webhook", g.WebhookHandler(reflecting)) + sup := &provider.Supervisor{Name: "gitea", Run: func(ctx context.Context) error { + pollCtx, cancel := context.WithTimeout(ctx, time.Minute) + defer cancel() + _, err := g.Poll(pollCtx, reflecting) + if err == nil { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(time.Minute): } - time.Sleep(time.Minute) } - }() + return err + }} + sup.Start(context.Background()) + providerHealth["gitea"] = sup } if path := os.Getenv("ORCHESTRA_JSONL"); path != "" { - go func() { - ctx := context.Background() - err := (provider.JSONLWatcher{Path: path, Interval: time.Second, Provider: provider.JSONL{Source: "jsonl"}}).Run(ctx, s) - if err != nil { - log.Printf("jsonl provider: %v", err) - } - }() + sup := &provider.Supervisor{Name: "jsonl", Run: func(ctx context.Context) error { + return (provider.JSONLWatcher{Path: path, Interval: time.Second, Provider: provider.JSONL{Source: "jsonl"}}).Run(ctx, s) + }} + sup.Start(context.Background()) + providerHealth["jsonl"] = sup } port := os.Getenv("ORCHESTRA_PORT") if port == "" { diff --git a/internal/provider/provider.go b/internal/provider/provider.go index db6348f..c2e6557 100644 --- a/internal/provider/provider.go +++ b/internal/provider/provider.go @@ -15,6 +15,7 @@ import ( "os" "strconv" "strings" + "sync" "time" "orchestra/internal/domain" @@ -29,6 +30,89 @@ type TaskReflector interface { ReflectTask(domain.Task, domain.Event) error } +type TaskLookup interface { + Task(string) (domain.Task, bool) +} + +// ReflectingSink preserves the append-first rule while asynchronously +// reflecting terminal state to an external provider. Reflection failures are +// returned to the caller so the supervisor can retry and surface health. +type ReflectingSink struct { + Sink Sink + Tasks TaskLookup + Reflector TaskReflector +} + +func (s ReflectingSink) Append(e domain.Event) error { + if err := s.Sink.Append(e); err != nil { + return err + } + if s.Reflector == nil || s.Tasks == nil { + return nil + } + if t, ok := s.Tasks.Task(e.TaskID); ok { + return s.Reflector.ReflectTask(t, e) + } + return nil +} + +type Health struct { + Name string `json:"name"` + Running bool `json:"running"` + LastError string `json:"last_error,omitempty"` + UpdatedAt time.Time `json:"updated_at"` +} + +// Supervise restarts a provider loop with bounded backoff and exposes its +// current health. The loop exits only when its context is cancelled. +type Supervisor struct { + Name string + Run func(context.Context) error + Backoff time.Duration + mu sync.RWMutex + health Health +} + +func (s *Supervisor) Start(ctx context.Context) { + if s.Backoff <= 0 { + s.Backoff = time.Second + } + s.mu.Lock() + s.health = Health{Name: s.Name, Running: true, UpdatedAt: time.Now().UTC()} + s.mu.Unlock() + go func() { + for { + err := s.Run(ctx) + if ctx.Err() != nil { + s.mu.Lock() + s.health.Running = false + s.health.UpdatedAt = time.Now().UTC() + s.mu.Unlock() + return + } + s.mu.Lock() + s.health.LastError = errString(err) + s.health.UpdatedAt = time.Now().UTC() + s.mu.Unlock() + t := time.NewTimer(s.Backoff) + select { + case <-ctx.Done(): + t.Stop() + return + case <-t.C: + } + } + }() +} + +func errString(err error) string { + if err == nil { + return "provider stopped" + } + return err.Error() +} +func (s *Supervisor) Health() Health { s.mu.RLock(); defer s.mu.RUnlock(); return s.health } + type JSONL struct{ Source string } func (j JSONL) Ingest(r io.Reader, sink Sink) (int, error) { diff --git a/progress.md b/progress.md index b12cc56..e17910e 100644 --- a/progress.md +++ b/progress.md @@ -61,6 +61,8 @@ Added bounded `POST /v1/artifacts` CAS upload support for report/handoff evidenc The coordinator now persists active task→herdr session mappings in an atomic runtime state file, reloads them after restart, reconciles them against durable task leases, kills stale recoverable sessions, and removes orphan mappings before monitoring begins. +Provider supervision/reflection is now wired: Gitea webhook and polling use append-first task reflection, JSONL and Gitea loops restart with bounded backoff, and `/v1/providers/health` exposes running/error state. Provider lifecycle cancellation remains tied to process shutdown until the server gains a root cancellation context. + Recommended order: 1. Add the orchestration coordinator: lease → worktree → harness session → bootstrap → lifecycle events.