kami
4b320809bd
Delivery merges the task branch, and that branch carries the "orchestra: TASK.md" commit. Master therefore ends up holding the previous task's TASK.md, and the next worktree branches from it. writeTaskFile returned early on os.Stat, so it left that inherited file in place. The worker then hashed the current task and every immutability check failed with "TASK.md changed" against a hash for a task nobody was running. Releases failed, rotation never relaunched, phase requests were never read, and the task died on retry_limit without leaving research. Presence is not identity. Compare content, and rewrite when it differs. Found live in run 7, the first task to start after run 6's pull request merged. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CVbaKucEYBjMqVeUgJUsc1
1439 lines
49 KiB
Go
1439 lines
49 KiB
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 (
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"orchestra/internal/agentctx"
|
||
"orchestra/internal/authz"
|
||
"orchestra/internal/continuity"
|
||
"orchestra/internal/domain"
|
||
"orchestra/internal/herdr"
|
||
"orchestra/internal/operations"
|
||
"orchestra/internal/store"
|
||
"orchestra/internal/workphase"
|
||
"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 err := writeTaskFile(ctx, p, t); err != nil {
|
||
return "", err
|
||
}
|
||
if w.TaskFileSHA != "" {
|
||
if err := continuity.VerifyTaskFile(p, w.TaskFileSHA); err != nil {
|
||
return "", err
|
||
}
|
||
}
|
||
return p, nil
|
||
}
|
||
|
||
// writeTaskFile commits the §6.2 immutable TASK.md into a freshly created
|
||
// worktree. It must be committed, not left dirty, so ScratchCommit's
|
||
// "TASK.md is immutable" check (which inspects `git status`) sees it as
|
||
// clean, and so its hash survives independent of any later scratch commits.
|
||
func writeTaskFile(ctx context.Context, worktree string, t domain.Task) error {
|
||
path := filepath.Join(worktree, "TASK.md")
|
||
want := continuity.RenderTaskFile(t)
|
||
// Presence is not identity. Delivery merges the task branch, and the
|
||
// branch carries this commit, so a later task's worktree branches from a
|
||
// master that already holds the *previous* task's TASK.md. Returning early
|
||
// on os.Stat left that file in place, and every immutability check then
|
||
// failed with "TASK.md changed" against a hash for a task nobody was
|
||
// running. Found live: run 7 died on retry_limit without ever leaving
|
||
// research.
|
||
if got, err := os.ReadFile(path); err == nil && bytes.Equal(got, want) {
|
||
return nil
|
||
}
|
||
if err := os.WriteFile(path, want, 0644); err != nil {
|
||
return err
|
||
}
|
||
for _, args := range [][]string{{"add", "TASK.md"}, {"commit", "-m", "orchestra: TASK.md"}} {
|
||
cmd := exec.CommandContext(ctx, "git", append([]string{"-C", worktree}, args...)...)
|
||
if out, err := cmd.CombinedOutput(); err != nil {
|
||
return fmt.Errorf("%s: %w", string(out), err)
|
||
}
|
||
}
|
||
return 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
|
||
// LocalHerdr, when set, is the coordinator's machine-ownership boundary.
|
||
// A coordinator must never operate a pane or checkout owned by another
|
||
// machine; federation workers own those operations locally.
|
||
LocalHerdr func(string) bool
|
||
mu sync.Mutex
|
||
sessions map[string]herdr.Session
|
||
loaded bool
|
||
healthMu sync.RWMutex
|
||
health MonitorHealth
|
||
// Hard is the occupancy threshold Monitor's periodic rotate() runs
|
||
// against, mirrored here so TurnDecision (the synchronous, per-turn
|
||
// counterpart driven by the Face-B stop hook) evaluates the same
|
||
// threshold rather than needing its own copy passed in by the caller.
|
||
Hard float64
|
||
// Soft is the advisory occupancy threshold (spec §5.3: "soft ~55%
|
||
// threshold") at which TurnDecision starts asking the agent to prepare a
|
||
// handoff — non-blocking, doesn't require a turn boundary — well before
|
||
// Hard forces one. Zero means "use the package default" (see
|
||
// defaultSoft), so existing callers that never set this field keep
|
||
// working unchanged.
|
||
Soft float64
|
||
// Reconcile imports newer human input for one task. Store.PreLease covers
|
||
// the moment ownership begins; this field covers the other half, a
|
||
// correction written while a lease is already live. It runs only at a
|
||
// verified turn boundary, so nothing preempts a running tool call.
|
||
ReconcileHumanInput func(ctx context.Context, taskID string) error
|
||
// Thrash tunes DetectThrash's three circuit breakers (§5.3). Zero-value
|
||
// fields fall back to herdr's own defaults, so leaving this unset works.
|
||
Thrash herdr.ThrashConfig
|
||
// ReconcileFailureHandoff is how many *consecutive* failed turn-boundary
|
||
// reconciles escalate to prepare_handoff. One failure is transient and
|
||
// continuing is right; a streak means Orchestra can no longer promise
|
||
// that the newest human input outranks this session's intent, so the
|
||
// honest move is to hand the task to a successor whose Store.PreLease
|
||
// reconcile fails closed while the source is down. Zero means the package
|
||
// default (defaultReconcileFailureHandoff).
|
||
ReconcileFailureHandoff int
|
||
// reconcileFailures is the streak per task, fenced on the lease epoch so
|
||
// a successor never inherits its predecessor's count and no release path
|
||
// needs a cleanup hook. Guarded by healthMu.
|
||
reconcileFailures map[string]reconcileStreak
|
||
}
|
||
|
||
type reconcileStreak struct {
|
||
Epoch string
|
||
N int
|
||
}
|
||
|
||
// defaultReconcileFailureHandoff is used whenever ReconcileFailureHandoff is
|
||
// unset. Three consecutive verified boundaries is long enough to ride out a
|
||
// restart or a brief network fault, short enough that a stuck source does not
|
||
// let a session run indefinitely on intent Orchestra cannot refresh.
|
||
const defaultReconcileFailureHandoff = 3
|
||
|
||
func (c *Coordinator) reconcileFailureThreshold() int {
|
||
if c.ReconcileFailureHandoff > 0 {
|
||
return c.ReconcileFailureHandoff
|
||
}
|
||
return defaultReconcileFailureHandoff
|
||
}
|
||
|
||
// noteReconcileResult records one turn boundary's reconcile outcome and reports
|
||
// whether this session has reached the escalation threshold. Success resets the
|
||
// streak, so two failures followed by a success escalate nothing.
|
||
func (c *Coordinator) noteReconcileResult(taskID, epoch string, err error) bool {
|
||
c.healthMu.Lock()
|
||
defer c.healthMu.Unlock()
|
||
if c.reconcileFailures == nil {
|
||
c.reconcileFailures = map[string]reconcileStreak{}
|
||
}
|
||
if err == nil {
|
||
delete(c.reconcileFailures, taskID)
|
||
return false
|
||
}
|
||
streak := c.reconcileFailures[taskID]
|
||
if streak.Epoch != epoch {
|
||
// A different owner: count this session's failures, not the previous
|
||
// lease's.
|
||
streak = reconcileStreak{Epoch: epoch}
|
||
}
|
||
streak.N++
|
||
c.reconcileFailures[taskID] = streak
|
||
c.recordSessionErrorLocked(taskID, fmt.Sprintf("reconcile human input (%d consecutive): %v", streak.N, err))
|
||
return streak.N >= c.reconcileFailureThreshold()
|
||
}
|
||
|
||
// defaultSoft is used whenever Coordinator.Soft is unset (zero value).
|
||
const defaultSoft = 0.55
|
||
|
||
func (c *Coordinator) soft() float64 {
|
||
if c.Soft > 0 {
|
||
return c.Soft
|
||
}
|
||
return defaultSoft
|
||
}
|
||
|
||
// checkActivityTriggers is S11's milestone/thrash pair: given an adapter that
|
||
// implements herdr.ActivityReader, read its tool-call history and evaluate
|
||
// both detectors. thrash takes priority (a circuit breaker overrides a
|
||
// coherent-looking commit), same as the caller would want either way since
|
||
// only one handoff request happens per tick. Returns the reason to request
|
||
// ("thrash"/"milestone") and its dead ends, or "" if neither fired or the
|
||
// adapter has no activity source at all — the latter is not degraded-and-
|
||
// recorded the way TurnBoundary's absence is, since these two triggers are
|
||
// additive on top of threshold/manual rotation, not a required safety gate.
|
||
func checkActivityTriggers(ctx context.Context, a herdr.Adapter, session herdr.Session, cfg herdr.ThrashConfig) (reason string, deadEnds []continuity.DeadEnd) {
|
||
reader, ok := a.(herdr.ActivityReader)
|
||
if !ok {
|
||
return "", nil
|
||
}
|
||
calls, err := reader.Activity(ctx, session)
|
||
if err != nil {
|
||
return "", nil
|
||
}
|
||
if thrash, de := herdr.DetectThrash(calls, cfg); thrash {
|
||
return "thrash", de
|
||
}
|
||
if herdr.DetectMilestone(calls) {
|
||
return "milestone", nil
|
||
}
|
||
return "", nil
|
||
}
|
||
|
||
// requestReasonedHandoff is the shared "ask once, remember we asked" wiring
|
||
// checkActivityTriggers' two callers (rotate, TurnDecision) both need — same
|
||
// HandoffRequested guard the occupancy-driven HandoffRequester path already
|
||
// uses, so a repeated thrash/milestone detection on later ticks doesn't
|
||
// reprompt every time before the agent has finished writing the file.
|
||
func (c *Coordinator) requestReasonedHandoff(ctx context.Context, taskID string, session herdr.Session, a herdr.Adapter, reason string, deadEnds []continuity.DeadEnd) {
|
||
if session.HandoffRequested {
|
||
return
|
||
}
|
||
if _, statErr := os.Stat(filepath.Join(session.Worktree, herdr.HandoffReportFile)); statErr == nil {
|
||
return
|
||
}
|
||
requester, ok := a.(herdr.ReasonedHandoffRequester)
|
||
if !ok {
|
||
return
|
||
}
|
||
if err := requester.RequestHandoffReason(ctx, session, reason, deadEnds); err != nil {
|
||
return
|
||
}
|
||
session.HandoffRequested = true
|
||
session.HandoffReason = reason
|
||
c.mu.Lock()
|
||
c.sessions[taskID] = session
|
||
_ = c.saveSessionsLocked()
|
||
c.mu.Unlock()
|
||
}
|
||
|
||
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"`
|
||
// Observed distinguishes a health entry this coordinator actually read
|
||
// from a live adapter this pass from one it could not reach (remote-owned
|
||
// session, unresolvable adapter) or has merely seeded at lease time. A
|
||
// consumer must not read Status as current unless Observed is true.
|
||
Observed bool `json:"observed"`
|
||
}
|
||
|
||
// 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
|
||
}
|
||
if c.LocalHerdr != nil && !c.LocalHerdr(id) {
|
||
return nil, fmt.Errorf("session %s is owned by non-local herdr %s", taskID, id)
|
||
}
|
||
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++
|
||
}
|
||
|
||
// recordSessionError keeps a non-fatal failure observable instead of letting
|
||
// a bare continue hide it, which is this codebase's recurring bug shape.
|
||
func (c *Coordinator) recordSessionError(taskID, msg string) {
|
||
c.healthMu.Lock()
|
||
defer c.healthMu.Unlock()
|
||
c.recordSessionErrorLocked(taskID, msg)
|
||
}
|
||
|
||
// recordSessionErrorLocked is recordSessionError for callers already holding
|
||
// healthMu.
|
||
func (c *Coordinator) recordSessionErrorLocked(taskID, msg string) {
|
||
if c.health.Sessions == nil {
|
||
c.health.Sessions = map[string]SessionHealth{}
|
||
}
|
||
h := c.health.Sessions[taskID]
|
||
h.LastError = msg
|
||
h.UpdatedAt = time.Now().UTC()
|
||
c.health.Sessions[taskID] = h
|
||
}
|
||
|
||
// deliverDecisions sends the human decisions this session has not been shown
|
||
// yet. It runs only at a verified turn boundary, and only when the turn
|
||
// verdict is continue, so it never interrupts a running tool call and never
|
||
// competes with a rotation that is about to hand the task to a successor.
|
||
func (c *Coordinator) deliverDecisions(ctx context.Context, taskID string, session herdr.Session, a herdr.Adapter) {
|
||
notifier, ok := a.(herdr.DecisionNotifier)
|
||
if !ok {
|
||
return
|
||
}
|
||
intent, err := c.Store.EffectiveIntent(taskID)
|
||
if err != nil {
|
||
c.recordSessionError(taskID, "effective intent: "+err.Error())
|
||
return
|
||
}
|
||
seen := make(map[string]bool, len(session.DeliveredDecisions))
|
||
for _, id := range session.DeliveredDecisions {
|
||
seen[id] = true
|
||
}
|
||
var fresh []domain.HumanDecision
|
||
for _, d := range intent.Decisions {
|
||
if !seen[d.ID] {
|
||
fresh = append(fresh, d)
|
||
}
|
||
}
|
||
if len(fresh) == 0 {
|
||
return
|
||
}
|
||
if err := notifier.NotifyDecisions(ctx, session, agentctx.DecisionNotice(fresh)); err != nil {
|
||
// Not recorded as delivered, so the next boundary retries.
|
||
c.recordSessionError(taskID, "deliver decisions: "+err.Error())
|
||
return
|
||
}
|
||
for _, d := range fresh {
|
||
session.DeliveredDecisions = append(session.DeliveredDecisions, d.ID)
|
||
}
|
||
c.mu.Lock()
|
||
c.sessions[taskID] = session
|
||
_ = c.saveSessionsLocked()
|
||
c.mu.Unlock()
|
||
}
|
||
|
||
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 {
|
||
// Adapter resolution failing is the normal case for a session
|
||
// owned by a remote worker's herdr, and the abnormal case for a
|
||
// misregistered local one. Either way the previous entry — often
|
||
// the Status:"running" rememberSession writes at lease time — must
|
||
// not be left standing as if it were freshly observed, or
|
||
// /v1/tasks/<id>/health reports a dead pane as running forever.
|
||
// Record the resolution failure so it is observable, per the
|
||
// contract SessionHealth documents.
|
||
c.healthMu.Lock()
|
||
prev := c.health.Sessions[taskID]
|
||
prev.LastError = err.Error()
|
||
prev.UpdatedAt = time.Now().UTC()
|
||
prev.Observed = false
|
||
c.health.Sessions[taskID] = prev
|
||
c.healthMu.Unlock()
|
||
continue
|
||
}
|
||
var h SessionHealth
|
||
h.Observed = true
|
||
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"
|
||
f, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0600)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if _, err = f.Write(b); err == nil {
|
||
err = f.Sync()
|
||
}
|
||
if closeErr := f.Close(); err == nil {
|
||
err = closeErr
|
||
}
|
||
if err != nil {
|
||
_ = os.Remove(tmp)
|
||
return err
|
||
}
|
||
if err = os.Rename(tmp, c.StatePath); err != nil {
|
||
return err
|
||
}
|
||
dir, err := os.Open(filepath.Dir(c.StatePath))
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer dir.Close()
|
||
return dir.Sync()
|
||
}
|
||
|
||
// 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 || t.State == domain.StateNeedsAttention) {
|
||
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
|
||
}
|
||
c.Hard = hard
|
||
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)
|
||
c.checkConventions(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()
|
||
}
|
||
}
|
||
|
||
// checkConventions is §6.3: "on update, the orchestra injects a notice to
|
||
// agents whose current task is adjacent" — adjacency here is "same project's
|
||
// base repo," and staleness is tracked by comparing each session's own
|
||
// last-known continuity.ConventionsHash against the base repo's current one,
|
||
// never by trusting the agent to notice on its own.
|
||
func (c *Coordinator) checkConventions(ctx context.Context) {
|
||
spec, ok := c.Worktrees.(WorktreeSpec)
|
||
if !ok {
|
||
return
|
||
}
|
||
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()
|
||
changed := false
|
||
for taskID, session := range sessions {
|
||
t, ok := c.Store.Task(taskID)
|
||
if !ok || t.State != domain.StateLeased {
|
||
continue
|
||
}
|
||
repo, _, valid := spec.Spec(t)
|
||
if !valid {
|
||
continue
|
||
}
|
||
hash, err := continuity.ConventionsHash(repo)
|
||
if err != nil || hash == session.ConventionsHash {
|
||
continue
|
||
}
|
||
a, err := c.adapterFor(taskID, session)
|
||
if err != nil {
|
||
continue
|
||
}
|
||
notifier, ok := a.(herdr.ConventionsNotifier)
|
||
if !ok {
|
||
continue
|
||
}
|
||
if err := notifier.NotifyConventionsChanged(ctx, session); err != nil {
|
||
continue
|
||
}
|
||
session.ConventionsHash = hash
|
||
c.mu.Lock()
|
||
c.sessions[taskID] = session
|
||
c.mu.Unlock()
|
||
changed = true
|
||
}
|
||
if changed {
|
||
c.mu.Lock()
|
||
_ = c.saveSessionsLocked()
|
||
c.mu.Unlock()
|
||
}
|
||
}
|
||
|
||
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 || t.State == domain.StateNeedsAttention) {
|
||
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]any{"reason": "pane_exited", "harness_id": s.Harness, "lease_epoch": t.Lease.Epoch, "expected_version": t.Version})
|
||
_ = 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()
|
||
var events []domain.Event
|
||
var firstErr error
|
||
for _, task := range c.Store.Tasks() {
|
||
if (task.State != domain.StateLeased && task.State != domain.StateNeedsAttention) || task.Lease == nil || task.Lease.Until.After(time.Now()) {
|
||
continue
|
||
}
|
||
// Stop a local predecessor before making its lease eligible for a
|
||
// successor. If this cannot be done, keep both the mapping and the
|
||
// lease: safety beats reclaim speed.
|
||
c.mu.Lock()
|
||
s, local := c.sessions[task.ID]
|
||
c.mu.Unlock()
|
||
if local {
|
||
a, adapterErr := c.adapterFor(task.ID, s)
|
||
if adapterErr != nil {
|
||
if firstErr == nil {
|
||
firstErr = fmt.Errorf("expire %s: resolve old pane: %w", task.ID, adapterErr)
|
||
}
|
||
continue
|
||
}
|
||
if killErr := a.Kill(ctx, s); killErr != nil {
|
||
if firstErr == nil {
|
||
firstErr = fmt.Errorf("expire %s: quarantine old pane: %w", task.ID, killErr)
|
||
}
|
||
continue
|
||
}
|
||
c.mu.Lock()
|
||
delete(c.sessions, task.ID)
|
||
_ = c.saveSessionsLocked()
|
||
c.mu.Unlock()
|
||
}
|
||
e, expireErr := c.Store.ExpireLease(task.ID, time.Now())
|
||
if expireErr != nil {
|
||
if !errors.Is(expireErr, domain.ErrConflict) && firstErr == nil {
|
||
firstErr = expireErr
|
||
}
|
||
continue
|
||
}
|
||
events = append(events, e)
|
||
}
|
||
return events, firstErr
|
||
}
|
||
|
||
// handoffReason reads HandoffFile from the worktree, if present, and returns
|
||
// its meta.reason ("threshold|milestone|thrash|manual" per §6.1). An unread
|
||
// able or invalid file returns "" — callers treat that as "no signal yet",
|
||
// never as "manual".
|
||
func handoffReason(worktree string) string {
|
||
b, err := os.ReadFile(filepath.Join(worktree, herdr.HandoffFile))
|
||
if err != nil {
|
||
return ""
|
||
}
|
||
h, err := continuity.Decode(b)
|
||
if err != nil {
|
||
return ""
|
||
}
|
||
return h.Meta.Reason
|
||
}
|
||
|
||
// reasonReconcileFailure is the handoff reason for a session released because
|
||
// human input could not be reconciled at repeated verified turn boundaries.
|
||
const reasonReconcileFailure = "reconcile_failure"
|
||
|
||
// bypassReason reports whether a handoff already carrying this reason is
|
||
// itself the boundary signal, so occupancy and the turn-boundary probe are
|
||
// skipped and the session is released immediately. reconcile_failure joins the
|
||
// list because Orchestra, not the context window, asked for that handoff.
|
||
func bypassReason(r string) bool {
|
||
return r == "manual" || r == "milestone" || r == "thrash" || r == reasonReconcileFailure
|
||
}
|
||
|
||
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"
|
||
// Agent-initiated ROTATE, and the two orchestrator-detected triggers
|
||
// (§5.3: manual / milestone / thrash) all short-circuit the same way
|
||
// once a handoff carrying that reason already exists: the boundary
|
||
// question has already been answered, so skip occupancy and the
|
||
// turn-boundary probe and go straight to release.
|
||
existingReason := handoffReason(session.Worktree)
|
||
bypass := bypassReason(existingReason)
|
||
if bypass {
|
||
reason = existingReason
|
||
} else {
|
||
if trigger, deadEnds := checkActivityTriggers(ctx, a, session, c.Thrash); trigger != "" {
|
||
c.requestReasonedHandoff(ctx, taskID, session, a, trigger, deadEnds)
|
||
continue
|
||
}
|
||
occupancy, err := a.Occupancy(session)
|
||
if err != nil || occupancy < c.soft() {
|
||
continue
|
||
}
|
||
if occupancy < hard {
|
||
// Soft threshold (§5.3): request a handoff early, advisory
|
||
// only — no release, no turn-boundary requirement.
|
||
if requester, ok := a.(herdr.HandoffRequester); ok {
|
||
if _, statErr := os.Stat(filepath.Join(session.Worktree, herdr.HandoffReportFile)); statErr != nil && !session.HandoffRequested {
|
||
if reqErr := requester.RequestHandoff(ctx, session); reqErr == nil {
|
||
session.HandoffRequested = true
|
||
session.HandoffReason = "threshold"
|
||
c.mu.Lock()
|
||
c.sessions[taskID] = session
|
||
_ = c.saveSessionsLocked()
|
||
c.mu.Unlock()
|
||
}
|
||
}
|
||
}
|
||
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()
|
||
}
|
||
}
|
||
if requester, ok := a.(herdr.HandoffRequester); ok {
|
||
if _, statErr := os.Stat(filepath.Join(session.Worktree, herdr.HandoffReportFile)); statErr != nil {
|
||
if !session.HandoffRequested {
|
||
if reqErr := requester.RequestHandoff(ctx, session); reqErr == nil {
|
||
session.HandoffRequested = true
|
||
session.HandoffReason = reason
|
||
c.mu.Lock()
|
||
c.sessions[taskID] = session
|
||
_ = c.saveSessionsLocked()
|
||
c.mu.Unlock()
|
||
}
|
||
}
|
||
continue
|
||
}
|
||
}
|
||
ref, err := a.Release(ctx, session)
|
||
if err != nil {
|
||
// A release failure is operational state, not a silent retry. Keep
|
||
// the fenced lease and pane for recovery while durably exposing the
|
||
// failed phase to the worker and operator.
|
||
_ = c.block(task, "rotation release: "+err.Error())
|
||
continue
|
||
}
|
||
if ref == "" {
|
||
_ = c.block(task, "rotation release: empty handoff reference")
|
||
continue
|
||
}
|
||
anchorSHA, err := herdr.HeadSHA(session.Worktree)
|
||
if err != nil {
|
||
// Cannot certify the anchor. Record the fault while retaining the
|
||
// owner; an unseen bare continue used to leave this state opaque.
|
||
_ = c.block(task, "rotation anchor: "+err.Error())
|
||
continue
|
||
}
|
||
b, _ := json.Marshal(map[string]any{"handoff_ref": ref, "reason": reason, "anchor_sha": anchorSHA, "harness_id": task.Lease.HarnessID, "lease_epoch": task.Lease.Epoch, "expected_version": task.Version})
|
||
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 {
|
||
// A release only transfers the lease; this local coordinator owns
|
||
// the predecessor pane until it has actually stopped it.
|
||
if err := a.Kill(ctx, session); err == nil {
|
||
c.mu.Lock()
|
||
delete(c.sessions, taskID)
|
||
_ = c.saveSessionsLocked()
|
||
c.mu.Unlock()
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// RemoteTurn is the federated half of a turn boundary. A worker owns the pane,
|
||
// so it evaluates rotation locally and reports the verdict it reached; the
|
||
// coordinator owns authority, so it reconciles human input here and answers
|
||
// with the decisions that session has not been shown yet.
|
||
//
|
||
// The split is deliberate. Duplicating the rotation state machine in the
|
||
// worker would give two answers to "should this session stop"; asking the
|
||
// coordinator to probe a remote pane would give it a checkout it cannot
|
||
// validate. Neither half is authoritative about the other's state.
|
||
//
|
||
// Decisions are returned only when the verdict is continue, matching the local
|
||
// path: a rotating session's successor picks them up at re-lease.
|
||
func (c *Coordinator) RemoteTurn(ctx context.Context, taskID, epoch, verdict string, delivered []string) (string, []domain.HumanDecision, error) {
|
||
if c.Store == nil {
|
||
return "", nil, fmt.Errorf("orchestrator: dependencies required")
|
||
}
|
||
t, ok := c.Store.Task(taskID)
|
||
if !ok {
|
||
return "", nil, domain.ErrNotFound
|
||
}
|
||
if t.State != domain.StateLeased && t.State != domain.StateNeedsAttention {
|
||
return "", nil, fmt.Errorf("orchestrator: task %q not leased", taskID)
|
||
}
|
||
// Fenced like every other worker-driven call: a worker whose lease was
|
||
// reassigned must not be handed the current session's decisions.
|
||
if t.Lease == nil || epoch == "" || t.Lease.Epoch != epoch {
|
||
return "", nil, domain.ErrConflict
|
||
}
|
||
escalate := false
|
||
if c.ReconcileHumanInput != nil {
|
||
// Same contract as the local boundary: one failure is observable, not
|
||
// fatal, because refusing would freeze a live remote session without
|
||
// making its current intent any less stale. A streak escalates, on the
|
||
// same threshold the local path uses.
|
||
escalate = c.noteReconcileResult(taskID, epoch, c.ReconcileHumanInput(ctx, taskID))
|
||
}
|
||
if verdict != TurnContinue {
|
||
// The worker already wants to stop. Answering with a second reason
|
||
// would manufacture a rotation trigger nothing needs.
|
||
return verdict, nil, nil
|
||
}
|
||
if escalate {
|
||
return TurnPrepareHandoff, nil, nil
|
||
}
|
||
intent, err := c.Store.EffectiveIntent(taskID)
|
||
if err != nil {
|
||
return "", nil, err
|
||
}
|
||
seen := make(map[string]bool, len(delivered))
|
||
for _, id := range delivered {
|
||
seen[id] = true
|
||
}
|
||
var fresh []domain.HumanDecision
|
||
for _, d := range intent.Decisions {
|
||
if !seen[d.ID] {
|
||
fresh = append(fresh, d)
|
||
}
|
||
}
|
||
return verdict, fresh, nil
|
||
}
|
||
|
||
// Turn decision verdicts (spec §5.3, AUDIT.md Phase 2 items 1-2). These are
|
||
// the only valid results of TurnDecision and the only values the
|
||
// POST /v1/harness/turn endpoint may return.
|
||
const (
|
||
TurnContinue = "continue"
|
||
TurnPrepareHandoff = "prepare_handoff"
|
||
TurnRotateNow = "rotate_now"
|
||
TurnRefuse = "refuse"
|
||
)
|
||
|
||
// TurnDecision evaluates a single leased task's rotation state synchronously,
|
||
// at a harness-reported turn boundary, and acts on the result. It mirrors
|
||
// rotate()'s per-task logic (occupancy → turn-boundary → handoff-file →
|
||
// release) but is invoked once per turn from the Face-B stop hook instead of
|
||
// on Monitor's ticker, so an agent that's about to stop gets an authoritative
|
||
// answer instead of waiting for the next tick. `refuse` covers every case
|
||
// where continuing to let the harness stop would be unsafe: the turn
|
||
// boundary can't be verified, or release/anchor certification failed.
|
||
func (c *Coordinator) TurnDecision(ctx context.Context, taskID string) (string, error) {
|
||
c.loadSessions()
|
||
c.mu.Lock()
|
||
session, ok := c.sessions[taskID]
|
||
c.mu.Unlock()
|
||
if !ok {
|
||
return "", fmt.Errorf("orchestrator: no session for task %q", taskID)
|
||
}
|
||
task, ok := c.Store.Task(taskID)
|
||
if !ok || task.State != domain.StateLeased {
|
||
return "", fmt.Errorf("orchestrator: task %q not leased", taskID)
|
||
}
|
||
a, err := c.adapterFor(taskID, session)
|
||
if err != nil {
|
||
return "", fmt.Errorf("orchestrator: adapter: %w", err)
|
||
}
|
||
// A turn boundary is the one point where the agent is verifiably between
|
||
// actions, so it is where newer human input is imported for a live lease.
|
||
// The task is re-read afterwards because a recorded decision bumps its
|
||
// version, and finishRelease below writes against that version.
|
||
escalate := false
|
||
if c.ReconcileHumanInput != nil {
|
||
// One failure is recorded and the turn continues: blocking would not
|
||
// remove stale intent from the running agent, and a source outage
|
||
// would freeze every live session. A streak is different, and acts on
|
||
// the continue path below.
|
||
escalate = c.noteReconcileResult(taskID, task.Lease.Epoch, c.ReconcileHumanInput(ctx, taskID))
|
||
if fresh, ok := c.Store.Task(taskID); ok {
|
||
task = fresh
|
||
}
|
||
}
|
||
// Agent-initiated ROTATE, and the two orchestrator-detected triggers
|
||
// (§5.3: manual / milestone / thrash): a handoff already written with one
|
||
// of these reasons is itself the boundary signal — skip occupancy and the
|
||
// turn-boundary probe and release immediately.
|
||
if existingReason := handoffReason(session.Worktree); bypassReason(existingReason) {
|
||
return c.finishRelease(ctx, taskID, task, session, a, existingReason)
|
||
}
|
||
d := (RotationStateMachine{Soft: c.soft(), Hard: c.Hard, Thrash: c.Thrash}).Evaluate(ctx, a, session)
|
||
if d.Degraded != nil {
|
||
return "", fmt.Errorf("orchestrator: rotation: %w", d.Degraded)
|
||
}
|
||
if d.Action == TurnContinue {
|
||
if escalate {
|
||
// Rotation has no reason of its own, so this is the one place the
|
||
// reconcile streak can act. Ask for a handoff; release runs through
|
||
// the ordinary bypass path once the agent writes it, and the
|
||
// successor's Store.PreLease reconcile fails closed while the
|
||
// source is still down.
|
||
c.requestReasonedHandoff(ctx, taskID, session, a, reasonReconcileFailure, nil)
|
||
return TurnPrepareHandoff, nil
|
||
}
|
||
// Only on continue. A rotating session's successor picks the decision
|
||
// up through Store.PreLease when it acquires the lease.
|
||
c.deliverDecisions(ctx, taskID, session, a)
|
||
return TurnContinue, nil
|
||
}
|
||
if d.Action == TurnRefuse {
|
||
return TurnRefuse, nil
|
||
}
|
||
if d.Reason == "milestone" || d.Reason == "thrash" {
|
||
c.requestReasonedHandoff(ctx, taskID, session, a, d.Reason, d.DeadEnds)
|
||
return TurnPrepareHandoff, nil
|
||
}
|
||
if d.Action == TurnPrepareHandoff {
|
||
// Soft threshold (§5.3): advisory only. Ask the agent to start
|
||
// preparing a handoff well before Hard forces one, but don't block
|
||
// the turn on a boundary check — the agent is free to keep working.
|
||
if requester, ok := a.(herdr.HandoffRequester); ok {
|
||
if _, statErr := os.Stat(filepath.Join(session.Worktree, herdr.HandoffReportFile)); statErr != nil {
|
||
if !session.HandoffRequested {
|
||
if reqErr := requester.RequestHandoff(ctx, session); reqErr == nil {
|
||
session.HandoffRequested = true
|
||
session.HandoffReason = "threshold"
|
||
c.mu.Lock()
|
||
c.sessions[taskID] = session
|
||
_ = c.saveSessionsLocked()
|
||
c.mu.Unlock()
|
||
}
|
||
}
|
||
}
|
||
}
|
||
return TurnPrepareHandoff, nil
|
||
}
|
||
if requester, ok := a.(herdr.HandoffRequester); ok {
|
||
if _, statErr := os.Stat(filepath.Join(session.Worktree, herdr.HandoffReportFile)); statErr != nil {
|
||
if !session.HandoffRequested {
|
||
if reqErr := requester.RequestHandoff(ctx, session); reqErr == nil {
|
||
session.HandoffRequested = true
|
||
session.HandoffReason = "threshold"
|
||
c.mu.Lock()
|
||
c.sessions[taskID] = session
|
||
_ = c.saveSessionsLocked()
|
||
c.mu.Unlock()
|
||
}
|
||
}
|
||
return TurnPrepareHandoff, nil
|
||
}
|
||
}
|
||
return c.finishRelease(ctx, taskID, task, session, a, "threshold")
|
||
}
|
||
|
||
// finishRelease runs the common release tail shared by TurnDecision's
|
||
// threshold path and its agent-initiated-ROTATE (reason=manual) shortcut:
|
||
// call Adapter.Release, certify the anchor against the real worktree HEAD,
|
||
// and emit TaskReleased. Any failure refuses rather than emitting a
|
||
// TaskReleased payload that would fail validation and strand the session.
|
||
func (c *Coordinator) finishRelease(ctx context.Context, taskID string, task domain.Task, session herdr.Session, a herdr.Adapter, reason string) (string, error) {
|
||
ref, err := a.Release(ctx, session)
|
||
if err != nil || ref == "" {
|
||
return TurnRefuse, nil
|
||
}
|
||
anchorSHA, err := herdr.HeadSHA(session.Worktree)
|
||
if err != nil {
|
||
// Cannot certify the anchor: refuse rather than release with an
|
||
// invalid TaskReleased payload, same as rotate()'s bare continue.
|
||
return TurnRefuse, nil
|
||
}
|
||
b, _ := json.Marshal(map[string]any{"handoff_ref": ref, "reason": reason, "anchor_sha": anchorSHA, "harness_id": task.Lease.HarnessID, "lease_epoch": task.Lease.Epoch, "expected_version": task.Version})
|
||
e := domain.Event{ID: domain.NewID(), Type: "TaskReleased", TaskID: taskID, Version: task.Version + 1, Payload: b, Surface: string(authz.System)}
|
||
if err := c.Store.Append(e); err != nil {
|
||
return TurnRefuse, nil
|
||
}
|
||
if err := a.Kill(ctx, session); err != nil {
|
||
return TurnRefuse, nil
|
||
}
|
||
c.mu.Lock()
|
||
delete(c.sessions, taskID)
|
||
_ = c.saveSessionsLocked()
|
||
c.mu.Unlock()
|
||
return TurnRotateNow, nil
|
||
}
|
||
|
||
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")
|
||
}
|
||
if c.LocalHerdr != nil && !c.LocalHerdr(p.HarnessID) {
|
||
return c.block(t, "remote herdr must be operated by its federation worker")
|
||
}
|
||
a, err := c.Adapters.Adapter(p.HarnessID)
|
||
if err != nil {
|
||
return c.block(t, "adapter: "+err.Error())
|
||
}
|
||
// Worktrees, including immutable TASK.md, are coordinator-local state.
|
||
// A remote herdr must be driven by its federation worker instead of being
|
||
// asked to create an opaque checkout that this coordinator cannot validate.
|
||
w, err := c.Worktrees.Create(ctx, t)
|
||
if err != nil {
|
||
return c.block(t, "worktree: "+err.Error())
|
||
}
|
||
taskFileSHA, _ := continuity.TaskFileHash(w)
|
||
// One renderer. The launch instruction is built by agentctx so a decision
|
||
// the human recorded while this task was queued is visible to the agent
|
||
// from its first turn, above anything it will later read as continuity.
|
||
intent, err := c.Store.EffectiveIntent(t.ID)
|
||
if err != nil {
|
||
return c.block(t, "effective intent: "+err.Error())
|
||
}
|
||
git := agentctx.GitState{Worktree: w, Branch: "orchestra/" + t.ID}
|
||
if sha, shaErr := herdr.HeadSHA(w); shaErr == nil {
|
||
git.HeadSHA = sha
|
||
}
|
||
// §6.2 pickup validation happens before the agent exists, not after: the
|
||
// handoff is part of the launch instruction now, so it must be trusted
|
||
// before it is rendered. A failure blocks the task without ever starting
|
||
// a session (this is the gap AUDIT.md's B6 named as unreached).
|
||
var handoff *continuity.Handoff
|
||
if p.HandoffRef != "" {
|
||
h, loadErr := continuity.Load(p.HandoffRef, c.Store)
|
||
if loadErr != nil {
|
||
return c.block(t, "handoff: "+loadErr.Error())
|
||
}
|
||
if err := continuity.ValidatePickup(w, h, taskFileSHA); err != nil {
|
||
return c.block(t, "pickup: "+err.Error())
|
||
}
|
||
handoff = &h
|
||
}
|
||
in := agentctx.Input{
|
||
Task: t, Intent: intent, Handoff: handoff, Git: git,
|
||
Phase: t.WorkPhase, RepoRules: agentctx.DiscoverRepoRules(w),
|
||
DecisionRequest: t.DecisionRequest,
|
||
}
|
||
if t.ResearchRef != "" {
|
||
r, refErr := c.research(t.ResearchRef)
|
||
if refErr != nil {
|
||
return c.block(t, "research artifact: "+refErr.Error())
|
||
}
|
||
in.Research = r
|
||
}
|
||
if t.PlanRef != "" {
|
||
pl, refErr := c.plan(t.PlanRef)
|
||
if refErr != nil {
|
||
return c.block(t, "plan artifact: "+refErr.Error())
|
||
}
|
||
in.Plan = pl
|
||
}
|
||
if t.Review != nil {
|
||
r, refErr := operations.TaskReview(c.Store, t)
|
||
if refErr != nil {
|
||
return c.block(t, "review artifact: "+refErr.Error())
|
||
}
|
||
in.Review = r
|
||
}
|
||
built, err := agentctx.Build(in)
|
||
if err != nil {
|
||
return c.block(t, "context: "+err.Error())
|
||
}
|
||
prompt := built.System + "\n\n" + built.Task
|
||
if writeErr := herdr.WriteLaunchContext(w, prompt); writeErr != nil {
|
||
c.recordSessionError(t.ID, "launch context: "+writeErr.Error())
|
||
}
|
||
var s herdr.Session
|
||
if promptLeaser, ok := a.(herdr.PromptLeaser); ok {
|
||
s, err = promptLeaser.LeasePrompt(ctx, t.ID, w, prompt)
|
||
} else {
|
||
s, err = a.Lease(ctx, t.ID, w)
|
||
}
|
||
if err != nil {
|
||
// A UI-changing prompt can time out after herdr accepted it. Keep the
|
||
// live pane mapped before recording TaskNeedsAttention so a later completion
|
||
// can reconcile the lifecycle instead of becoming an orphan (B15).
|
||
if s.PaneID != "" {
|
||
s.HerdrID = p.HarnessID
|
||
s.TaskFileSHA = taskFileSHA
|
||
s.ConventionsHash, _ = continuity.ConventionsHash(w)
|
||
_ = c.rememberSession(t.ID, s)
|
||
}
|
||
return c.block(t, "lease: "+err.Error())
|
||
}
|
||
s.HerdrID = p.HarnessID
|
||
s.TaskFileSHA = taskFileSHA
|
||
// The launch instruction carried these, so the first turn boundary must
|
||
// not re-deliver them as news.
|
||
for _, d := range intent.Decisions {
|
||
s.DeliveredDecisions = append(s.DeliveredDecisions, d.ID)
|
||
}
|
||
// Best-effort, same caveat as taskFileSHA above: only meaningful for a
|
||
// worktree this process can read locally. Snapshots the shared-docs
|
||
// state this session starts trusting; checkConventions notices drift
|
||
// from here, not from whatever the agent's own cached view is (§6.3).
|
||
s.ConventionsHash, _ = continuity.ConventionsHash(w)
|
||
err = c.rememberSession(t.ID, s)
|
||
c.healthMu.Lock()
|
||
if c.health.Sessions == nil {
|
||
c.health.Sessions = map[string]SessionHealth{}
|
||
}
|
||
// Seeded at lease time, not observed from the harness: Observed stays
|
||
// false until refreshSessionHealth reads a live adapter.
|
||
c.health.Sessions[t.ID] = SessionHealth{Status: "running", UpdatedAt: time.Now().UTC()}
|
||
c.healthMu.Unlock()
|
||
return err
|
||
}
|
||
|
||
// research and plan read a sealed phase artifact. A stored ref that will not
|
||
// decode is a blocked task, not a silently empty context section.
|
||
func (c *Coordinator) research(ref string) (*workphase.Research, error) {
|
||
b, err := c.Store.Artifact(ref)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
r, err := workphase.DecodeStoredResearch(b)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &r, nil
|
||
}
|
||
|
||
func (c *Coordinator) plan(ref string) (*workphase.PlanDoc, error) {
|
||
b, err := c.Store.Artifact(ref)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
p, err := workphase.DecodeStoredPlan(b)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &p, nil
|
||
}
|
||
|
||
func (c *Coordinator) rememberSession(taskID string, s herdr.Session) error {
|
||
c.mu.Lock()
|
||
defer c.mu.Unlock()
|
||
if c.sessions == nil {
|
||
c.sessions = map[string]herdr.Session{}
|
||
}
|
||
c.sessions[taskID] = s
|
||
return c.saveSessionsLocked()
|
||
}
|
||
|
||
func (c *Coordinator) block(t domain.Task, reason string) error {
|
||
p := map[string]any{"blocker": reason, "block_reason": string(domain.InferBlockReason(reason)), "lifecycle_phase": "needs_attention", "last_error": reason, "pane_state": "unknown", "session_evidence": domain.SessionEvidence{PaneState: "unknown", Source: "coordinator", CheckedAt: time.Now().UTC()}}
|
||
if s, ok := c.Session(t.ID); ok {
|
||
p["pane_id"] = s.PaneID
|
||
p["harness_id"] = s.HerdrID
|
||
p["pane_state"] = "open"
|
||
p["session_evidence"] = domain.SessionEvidence{PaneID: s.PaneID, HarnessID: s.HerdrID, PaneState: "open", Source: "coordinator", CheckedAt: time.Now().UTC()}
|
||
}
|
||
b, _ := json.Marshal(p)
|
||
if t.Lease != nil {
|
||
p["harness_id"] = t.Lease.HarnessID
|
||
p["lease_epoch"] = t.Lease.Epoch
|
||
p["expected_version"] = t.Version
|
||
b, _ = json.Marshal(p)
|
||
}
|
||
return c.Store.Append(domain.Event{ID: domain.NewID(), Type: "TaskNeedsAttention", 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
|
||
}
|
||
|
||
// RequestHandoff asks the live harness to prepare its agent-authored handoff.
|
||
// It deliberately does not release the pane: a later validated handoff is the
|
||
// only evidence that can make a rotation safe.
|
||
func (c *Coordinator) RequestHandoff(ctx context.Context, taskID string) error {
|
||
c.loadSessions()
|
||
c.mu.Lock()
|
||
s, ok := c.sessions[taskID]
|
||
c.mu.Unlock()
|
||
if !ok {
|
||
return fmt.Errorf("session not found for task %s", taskID)
|
||
}
|
||
if s.HandoffRequested {
|
||
return nil
|
||
}
|
||
a, err := c.adapterFor(taskID, s)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
req, ok := a.(herdr.HandoffRequester)
|
||
if !ok {
|
||
return fmt.Errorf("harness does not support handoff requests")
|
||
}
|
||
if err := req.RequestHandoff(ctx, s); err != nil {
|
||
return err
|
||
}
|
||
s.HandoffRequested = true
|
||
c.mu.Lock()
|
||
c.sessions[taskID] = s
|
||
err = c.saveSessionsLocked()
|
||
c.mu.Unlock()
|
||
return err
|
||
}
|
||
|
||
// RespondApproval is the local implementation of the same guarded command
|
||
// contract used by federation workers. It rechecks the displayed capture at
|
||
// the owning herdr immediately before input is sent.
|
||
func (c *Coordinator) RespondApproval(ctx context.Context, taskID string, grant bool, expectedCapture string) error {
|
||
c.loadSessions()
|
||
c.mu.Lock()
|
||
s, ok := c.sessions[taskID]
|
||
c.mu.Unlock()
|
||
if !ok {
|
||
return fmt.Errorf("session not found for task %s", taskID)
|
||
}
|
||
a, err := c.adapterFor(taskID, s)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
responder, ok := a.(herdr.ApprovalResponder)
|
||
if !ok {
|
||
return fmt.Errorf("harness does not support approval responses")
|
||
}
|
||
return responder.RespondApproval(ctx, s, grant, expectedCapture)
|
||
}
|
||
|
||
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)
|
||
}
|