orchestra/internal/orchestrator/orchestrator.go

// Package orchestrator connects router lease events to an opaque herdr
// session. It is deliberately small: scheduling remains in router and the
// adapter remains the only component that knows how to drive a harness.
package orchestrator

import (
	"context"
	"encoding/json"
	"fmt"
	"orchestra/internal/authz"
	"orchestra/internal/continuity"
	"orchestra/internal/domain"
	"orchestra/internal/herdr"
	"orchestra/internal/store"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync"
	"time"
)

type Worktrees interface {
	Create(context.Context, domain.Task) (string, error)
}
type WorktreeSpec interface {
	Spec(domain.Task) (string, string, bool)
}
type WorktreeCleaner interface {
	Remove(context.Context, domain.Task, string) error
}
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
	TaskFileSHA string
}

func (w GitWorktrees) Spec(domain.Task) (string, string, bool) {
	return w.Repo, w.Root, w.Repo != "" && w.Root != ""
}

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 {
		if w.TaskFileSHA != "" {
			if err := continuity.VerifyTaskFile(p, w.TaskFileSHA); err != nil {
				return "", err
			}
		}
		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)
	}
	if w.TaskFileSHA != "" {
		if err := continuity.VerifyTaskFile(p, w.TaskFileSHA); err != nil {
			return "", err
		}
	}
	return p, nil
}

func (w GitWorktrees) Remove(ctx context.Context, _ domain.Task, path string) error {
	if path == "" {
		return fmt.Errorf("worktree: path required")
	}
	cmd := exec.CommandContext(ctx, "git", "-C", w.Repo, "worktree", "remove", "--force", path)
	if out, err := cmd.CombinedOutput(); err != nil {
		return fmt.Errorf("%s: %w", string(out), err)
	}
	return nil
}

// ProjectRepo is the minimal shape PerProjectGitWorktrees needs from a
// project's registry entry — kept local (not importing internal/registry)
// so orchestrator does not depend on registry's config-loading concerns.
type ProjectRepo struct {
	Repo         string
	WorktreeRoot string
}

// PerProjectGitWorktrees resolves a task's repo/root by its project (spec
// §2.2: each project is first-class and may have its own checkout), falling
// back to Default for any project not present in Projects — this keeps
// single-repo deployments working unchanged.
type PerProjectGitWorktrees struct {
	Projects    map[string]ProjectRepo
	Default     GitWorktrees
	TaskFileSHA string
}

func (w PerProjectGitWorktrees) Spec(t domain.Task) (string, string, bool) {
	g := w.Default
	if p, ok := w.Projects[t.Project]; ok && p.Repo != "" && p.WorktreeRoot != "" {
		g = GitWorktrees{Root: p.WorktreeRoot, Repo: p.Repo}
	}
	return g.Spec(t)
}

func (w PerProjectGitWorktrees) Create(ctx context.Context, t domain.Task) (string, error) {
	g := w.Default
	if p, ok := w.Projects[t.Project]; ok && p.Repo != "" && p.WorktreeRoot != "" {
		g = GitWorktrees{Root: p.WorktreeRoot, Repo: p.Repo, TaskFileSHA: w.TaskFileSHA}
	}
	if g.TaskFileSHA == "" {
		g.TaskFileSHA = w.TaskFileSHA
	}
	return g.Create(ctx, t)
}

func (w PerProjectGitWorktrees) Remove(ctx context.Context, t domain.Task, path string) error {
	g := w.Default
	if p, ok := w.Projects[t.Project]; ok && p.Repo != "" && p.WorktreeRoot != "" {
		g = GitWorktrees{Root: p.WorktreeRoot, Repo: p.Repo}
	}
	return g.Remove(ctx, t, path)
}

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
	Adapters  Adapters
	StatePath string
	mu        sync.Mutex
	sessions  map[string]herdr.Session
	loaded    bool
	healthMu  sync.RWMutex
	health    MonitorHealth
}

type MonitorHealth struct {
	Running   bool      `json:"running"`
	LastRun   time.Time `json:"last_run"`
	LastError string    `json:"last_error,omitempty"`
	Expired   int       `json:"expired"`
	// TurnBoundaryDegraded counts rotation ticks where Face B (spec §5.2,
	// §5.3 — "the Stop-hook/Face-B decides rotation, not the router") could
	// not be consulted, so occupancy-only thresholding is standing in. This
	// must stay observable rather than a silent fallback: an operator (or
	// the brief) can see when a deployment's rotation safety is degraded.
	TurnBoundaryDegraded int                      `json:"turn_boundary_degraded"`
	Sessions             map[string]SessionHealth `json:"sessions,omitempty"`
}

type SessionHealth struct {
	Status             string    `json:"status,omitempty"`
	WaitingForApproval bool      `json:"waiting_for_approval"`
	Blocker            string    `json:"blocker,omitempty"`
	UpdatedAt          time.Time `json:"updated_at"`
	LastError          string    `json:"last_error,omitempty"`
	// Occupancy and OccupancyError make the number rotation actually decides
	// on observable (spec §5.2.1: verify this against a live session before
	// trusting it). A resolution/read failure is recorded here rather than
	// silently treated as "not time to rotate yet" by a bare continue.
	Occupancy      float64 `json:"occupancy,omitempty"`
	OccupancyError string  `json:"occupancy_error,omitempty"`
}

// adapterFor resolves the herdr adapter for a session. Session.HerdrID (the
// registered herdr instance id, e.g. "homesrv-claude") is authoritative;
// Session.Harness (the harness kind, e.g. "claude") is only a fallback for
// sessions persisted before HerdrID was tracked. Adapters are keyed by
// instance id, so falling back to the lease's harness id (recorded on the
// task) rather than the kind keeps this resolvable even then.
func (c *Coordinator) adapterFor(taskID string, session herdr.Session) (herdr.Adapter, error) {
	id := session.HerdrID
	if id == "" {
		if task, ok := c.Store.Task(taskID); ok && task.Lease != nil {
			id = task.Lease.HarnessID
		}
	}
	if id == "" {
		id = session.Harness
	}
	return c.Adapters.Adapter(id)
}

func (c *Coordinator) MonitorHealth() MonitorHealth {
	c.healthMu.RLock()
	defer c.healthMu.RUnlock()
	return c.health
}
func (c *Coordinator) setMonitorHealth(err error, expired int) {
	c.healthMu.Lock()
	defer c.healthMu.Unlock()
	c.health.Running = err == nil
	c.health.LastRun = time.Now().UTC()
	c.health.Expired += expired
	if err != nil {
		c.health.LastError = err.Error()
	} else {
		c.health.LastError = ""
	}
}

func (c *Coordinator) recordTurnBoundaryDegraded() {
	c.healthMu.Lock()
	defer c.healthMu.Unlock()
	c.health.TurnBoundaryDegraded++
}

func waitingForApproval(status string) bool {
	s := strings.ToLower(strings.ReplaceAll(strings.ReplaceAll(status, "-", "_"), " ", "_"))
	return s == "waiting_for_approval" || s == "awaiting_approval" || s == "approval_required"
}

func (c *Coordinator) refreshSessionHealth(ctx context.Context) {
	c.loadSessions()
	c.mu.Lock()
	sessions := make(map[string]herdr.Session, len(c.sessions))
	for id, s := range c.sessions {
		sessions[id] = s
	}
	c.mu.Unlock()
	c.healthMu.Lock()
	if c.health.Sessions == nil {
		c.health.Sessions = map[string]SessionHealth{}
	}
	c.healthMu.Unlock()
	for taskID, session := range sessions {
		a, err := c.adapterFor(taskID, session)
		if err != nil {
			continue
		}
		var h SessionHealth
		h.UpdatedAt = time.Now().UTC()
		if occ, occErr := a.Occupancy(session); occErr != nil {
			h.OccupancyError = occErr.Error()
		} else {
			h.Occupancy = occ
		}
		p, ok := a.(herdr.AgentStatus)
		if !ok {
			c.healthMu.Lock()
			c.health.Sessions[taskID] = h
			c.healthMu.Unlock()
			continue
		}
		status, err := p.AgentStatus(ctx, session)
		h.Status = status
		h.WaitingForApproval = waitingForApproval(status)
		if err != nil {
			h.LastError = err.Error()
		} else if blocker, ok := a.(herdr.AgentBlocker); ok && strings.EqualFold(status, "blocked") {
			h.Blocker, _ = blocker.AgentBlocker(ctx, session)
		}
		c.healthMu.Lock()
		c.health.Sessions[taskID] = h
		c.healthMu.Unlock()
	}
}

func (c *Coordinator) loadSessions() {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.loaded {
		return
	}
	c.loaded = true
	c.sessions = map[string]herdr.Session{}
	if c.StatePath == "" {
		return
	}
	b, err := os.ReadFile(c.StatePath)
	if err != nil {
		return
	}
	_ = json.Unmarshal(b, &c.sessions)
}

func (c *Coordinator) saveSessionsLocked() error {
	if c.StatePath == "" {
		return nil
	}
	b, err := json.Marshal(c.sessions)
	if err != nil {
		return err
	}
	tmp := c.StatePath + ".tmp"
	if err = os.WriteFile(tmp, b, 0600); err != nil {
		return err
	}
	return os.Rename(tmp, c.StatePath)
}

// Reconcile drops mappings whose task lease did not survive restart and kills
// their recoverable herdr sessions so an orphan cannot keep consuming a slot.
func (c *Coordinator) Reconcile(ctx context.Context) error {
	c.loadSessions()
	c.mu.Lock()
	for taskID, session := range c.sessions {
		t, ok := c.Store.Task(taskID)
		if ok && t.State == domain.StateLeased {
			continue
		}
		if a, err := c.adapterFor(taskID, session); err == nil {
			_ = a.Kill(ctx, session)
		}
		delete(c.sessions, taskID)
	}
	err := c.saveSessionsLocked()
	c.mu.Unlock()
	if err != nil {
		return err
	}
	// Session health is derived state. Rebuild it immediately from the
	// durable session mappings so a restart does not hide an outstanding
	// approval until the first periodic monitor tick.
	c.refreshSessionHealth(ctx)
	return nil
}

// Monitor performs conservative hard-threshold rotation. The adapter owns
// the handoff creation; the coordinator only publishes its content address
// and frees the lease for pickup by the router.
func (c *Coordinator) Monitor(ctx context.Context, hard float64, interval time.Duration) error {
	if err := c.Reconcile(ctx); err != nil {
		c.setMonitorHealth(err, 0)
		return err
	}
	if interval <= 0 {
		interval = 30 * time.Second
	}
	t := time.NewTicker(interval)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			c.healthMu.Lock()
			c.health.Running = false
			c.healthMu.Unlock()
			return ctx.Err()
		case <-t.C:
			c.refreshSessionHealth(ctx)
			c.cleanupCompleted(ctx)
			expired, err := c.expire(ctx)
			c.setMonitorHealth(err, len(expired))
			if err != nil {
				continue
			}
			c.rotate(ctx, hard)
		}
	}
}

func (c *Coordinator) cleanupCompleted(ctx context.Context) {
	cleaner, ok := c.Worktrees.(WorktreeCleaner)
	if !ok {
		return
	}
	c.loadSessions()
	c.mu.Lock()
	defer c.mu.Unlock()
	changed := false
	for taskID, session := range c.sessions {
		t, exists := c.Store.Task(taskID)
		if !exists || t.State != domain.StateCompleted {
			continue
		}
		if err := cleaner.Remove(ctx, t, session.Worktree); err != nil {
			c.healthMu.Lock()
			c.health.LastError = "worktree cleanup: " + err.Error()
			c.healthMu.Unlock()
			continue
		}
		delete(c.sessions, taskID)
		changed = true
	}
	if changed {
		_ = c.saveSessionsLocked()
	}
}

func (c *Coordinator) expire(ctx context.Context) ([]domain.Event, error) {
	// pane.exited is the low-latency path; lease expiry below remains the
	// authoritative backstop when herdr misses an exit notification.
	c.loadSessions()
	c.mu.Lock()
	for taskID, s := range c.sessions {
		if t, ok := c.Store.Task(taskID); ok && t.State == domain.StateLeased {
			if a, ae := c.adapterFor(taskID, s); ae == nil {
				if p, ok := a.(herdr.PaneExit); ok {
					if exited, ee := p.PaneExited(ctx, s); ee == nil && exited {
						b, _ := json.Marshal(map[string]string{"reason": "pane_exited", "harness_id": s.Harness})
						_ = c.Store.Append(domain.Event{ID: domain.NewID(), Type: "TaskReleased", TaskID: taskID, Version: t.Version + 1, Payload: b, Surface: string(authz.System)})
					}
				}
			}
		}
	}
	c.mu.Unlock()
	events, err := c.Store.ExpireLeases(time.Now())
	if err != nil {
		return events, err
	}
	for _, e := range events {
		c.loadSessions()
		c.mu.Lock()
		s, ok := c.sessions[e.TaskID]
		delete(c.sessions, e.TaskID)
		if ok {
			if a, ae := c.adapterFor(e.TaskID, s); ae == nil {
				_ = a.Kill(ctx, s)
			}
		}
		_ = c.saveSessionsLocked()
		c.mu.Unlock()
	}
	return events, nil
}
func (c *Coordinator) rotate(ctx context.Context, hard float64) {
	c.loadSessions()
	c.mu.Lock()
	sessions := make(map[string]herdr.Session, len(c.sessions))
	for id, s := range c.sessions {
		sessions[id] = s
	}
	c.mu.Unlock()
	for taskID, session := range sessions {
		task, ok := c.Store.Task(taskID)
		if !ok || task.State != domain.StateLeased {
			continue
		}
		a, err := c.adapterFor(taskID, session)
		if err != nil {
			continue
		}
		reason := "threshold"
		if signal, ok := a.(herdr.RotationSignal); ok {
			if r, signalErr := signal.RotationSignal(ctx, session); signalErr == nil && r != "" {
				reason = r
			}
		}
		occupancy, err := a.Occupancy(session)
		if err != nil || (occupancy < hard && reason == "threshold") {
			continue
		}
		// Face B is treated as required, not best-effort (spec §5.2/§5.3):
		// an adapter that supports the turn-boundary probe but fails to
		// answer it blocks this tick's release rather than silently
		// proceeding as if mid-turn interruption were safe. Only an
		// adapter that genuinely does not implement TurnBoundary at all
		// falls back to occupancy-only thresholding, and that fallback is
		// recorded so it is observable (MonitorHealth.TurnBoundaryDegraded)
		// instead of invisible.
		if boundary, ok := a.(herdr.TurnBoundary); ok {
			atBoundary, boundaryErr := boundary.AtTurnBoundary(ctx, session)
			if boundaryErr != nil {
				c.recordTurnBoundaryDegraded()
				continue
			}
			if !atBoundary {
				continue
			}
		} else {
			c.recordTurnBoundaryDegraded()
		}
		ref, err := a.Release(ctx, session)
		if err != nil {
			continue
		}
		if ref == "" {
			continue
		}
		anchorSHA, err := herdr.HeadSHA(session.Worktree)
		if err != nil {
			// Cannot certify the anchor: do not release with an invalid
			// TaskReleased payload (it would fail validation and strand
			// the lease/session). Leave the lease intact for the next
			// tick or TTL expiry to reclaim.
			continue
		}
		b, _ := json.Marshal(map[string]string{"handoff_ref": ref, "reason": reason, "anchor_sha": anchorSHA})
		e := domain.Event{ID: domain.NewID(), Type: "TaskReleased", TaskID: taskID, Version: task.Version + 1, Payload: b, Surface: string(authz.System)}
		if c.Store.Append(e) == nil {
			c.mu.Lock()
			delete(c.sessions, taskID)
			_ = c.saveSessionsLocked()
			c.mu.Unlock()
		}
	}
}

func (c *Coordinator) Start(ctx context.Context, e domain.Event) error {
	if e.Type != "TaskLeased" {
		return nil
	}
	if c.Store == nil || c.Worktrees == nil || c.Adapters == nil {
		return fmt.Errorf("orchestrator: dependencies required")
	}
	c.loadSessions()
	t, ok := c.Store.Task(e.TaskID)
	if !ok {
		return domain.ErrNotFound
	}
	var p struct {
		HarnessID  string `json:"harness_id"`
		HandoffRef string `json:"handoff_ref"`
	}
	if err := json.Unmarshal(e.Payload, &p); err != nil || p.HarnessID == "" {
		return fmt.Errorf("orchestrator: invalid lease")
	}
	a, err := c.Adapters.Adapter(p.HarnessID)
	if err != nil {
		return c.block(t, "adapter: "+err.Error())
	}
	var w string
	if creator, ok := a.(herdr.WorktreeCreator); ok {
		planner, planned := c.Worktrees.(WorktreeSpec)
		if !planned {
			return c.block(t, "worktree: repository specification unavailable")
		}
		repo, root, valid := planner.Spec(t)
		if !valid {
			return c.block(t, "worktree: repository and root required")
		}
		w, err = creator.CreateWorktree(ctx, repo, root, t.ID)
	} else {
		w, err = c.Worktrees.Create(ctx, t)
	}
	if err != nil {
		return c.block(t, "worktree: "+err.Error())
	}
	s, err := a.Lease(ctx, t.ID, w)
	if err != nil {
		return c.block(t, "lease: "+err.Error())
	}
	if p.HandoffRef != "" {
		if err = a.Bootstrap(ctx, s, p.HandoffRef); err != nil {
			_ = a.Kill(ctx, s)
			return c.block(t, "bootstrap: "+err.Error())
		}
	}
	s.HerdrID = p.HarnessID
	c.mu.Lock()
	if c.sessions == nil {
		c.sessions = map[string]herdr.Session{}
	}
	c.sessions[t.ID] = s
	err = c.saveSessionsLocked()
	c.mu.Unlock()
	c.healthMu.Lock()
	if c.health.Sessions == nil {
		c.health.Sessions = map[string]SessionHealth{}
	}
	c.health.Sessions[t.ID] = SessionHealth{Status: "running", UpdatedAt: time.Now().UTC()}
	c.healthMu.Unlock()
	return err
}

func (c *Coordinator) block(t domain.Task, reason string) error {
	b, _ := json.Marshal(map[string]string{"blocker": reason})
	return c.Store.Append(domain.Event{ID: domain.NewID(), Type: "TaskBlocked", TaskID: t.ID, Version: t.Version + 1, Payload: b, Surface: string(authz.System)})
}

func (c *Coordinator) Session(taskID string) (herdr.Session, bool) {
	c.loadSessions()
	c.mu.Lock()
	defer c.mu.Unlock()
	s, ok := c.sessions[taskID]
	return s, ok
}

func (c *Coordinator) Capture(ctx context.Context, taskID, source string) (string, error) {
	s, ok := c.Session(taskID)
	if !ok {
		return "", domain.ErrNotFound
	}
	id := s.HerdrID
	if id == "" {
		if t, ok := c.Store.Task(taskID); ok && t.Lease != nil {
			id = t.Lease.HarnessID
		}
	}
	a, err := c.Adapters.Adapter(id)
	if err != nil {
		return "", err
	}
	p, ok := a.(herdr.PaneCapture)
	if !ok {
		return "", fmt.Errorf("pane capture unsupported")
	}
	return p.PaneCapture(ctx, s, source)
}