supervise providers and reflect terminal state

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 == "" {

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) {

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.