kami
86cc0b9276
rotate() and TurnDecision now check the agent's own handoff for
meta.reason=="manual" before evaluating occupancy/turn-boundary — per
spec §5.3, that reason is itself the boundary signal ("a coherent unit
finished and the next is independent"), so it bypasses both checks and
releases immediately. Extracted the shared release-and-anchor-certify
tail into Coordinator.finishRelease so the manual path gets the same
anchor safety guarantee as the threshold path.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01W1rkJ2hBMybnJctPbcy4tT
912 lines
28 KiB
Go
912 lines
28 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 (
|
||
"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 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")
|
||
if _, err := os.Stat(path); err == nil {
|
||
return nil
|
||
}
|
||
if err := os.WriteFile(path, continuity.RenderTaskFile(t), 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
|
||
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
|
||
}
|
||
|
||
// 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
|
||
}
|
||
|
||
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
|
||
}
|
||
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 {
|
||
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
|
||
}
|
||
// 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
|
||
}
|
||
|
||
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 (§5.3): the agent itself wrote a handoff
|
||
// with reason=manual — "a coherent unit finished and the next is
|
||
// independent". That is the boundary signal in its own right; skip
|
||
// occupancy and the turn-boundary probe and go straight to release.
|
||
manual := handoffReason(session.Worktree) == "manual"
|
||
if manual {
|
||
reason = "manual"
|
||
} else {
|
||
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.HandoffFile)); statErr != nil && !session.HandoffRequested {
|
||
if reqErr := requester.RequestHandoff(ctx, session); reqErr == nil {
|
||
session.HandoffRequested = true
|
||
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.HandoffFile)); statErr != nil {
|
||
if !session.HandoffRequested {
|
||
if reqErr := requester.RequestHandoff(ctx, session); reqErr == nil {
|
||
session.HandoffRequested = true
|
||
c.mu.Lock()
|
||
c.sessions[taskID] = session
|
||
_ = c.saveSessionsLocked()
|
||
c.mu.Unlock()
|
||
}
|
||
}
|
||
continue
|
||
}
|
||
}
|
||
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()
|
||
}
|
||
}
|
||
}
|
||
|
||
// 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)
|
||
}
|
||
// Agent-initiated ROTATE (§5.3): a handoff already written with
|
||
// reason=manual is itself the boundary signal — skip occupancy and the
|
||
// turn-boundary probe and release immediately.
|
||
if handoffReason(session.Worktree) == "manual" {
|
||
return c.finishRelease(ctx, taskID, task, session, a, "manual")
|
||
}
|
||
occupancy, err := a.Occupancy(session)
|
||
if err != nil {
|
||
return "", fmt.Errorf("orchestrator: occupancy: %w", err)
|
||
}
|
||
if occupancy < c.soft() {
|
||
return TurnContinue, nil
|
||
}
|
||
if occupancy < c.Hard {
|
||
// 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.HandoffFile)); statErr != nil {
|
||
if !session.HandoffRequested {
|
||
if reqErr := requester.RequestHandoff(ctx, session); reqErr == nil {
|
||
session.HandoffRequested = true
|
||
c.mu.Lock()
|
||
c.sessions[taskID] = session
|
||
_ = c.saveSessionsLocked()
|
||
c.mu.Unlock()
|
||
}
|
||
}
|
||
}
|
||
}
|
||
return TurnPrepareHandoff, nil
|
||
}
|
||
if boundary, ok := a.(herdr.TurnBoundary); ok {
|
||
atBoundary, boundaryErr := boundary.AtTurnBoundary(ctx, session)
|
||
if boundaryErr != nil {
|
||
c.recordTurnBoundaryDegraded()
|
||
return TurnRefuse, nil
|
||
}
|
||
if !atBoundary {
|
||
return TurnRefuse, nil
|
||
}
|
||
} else {
|
||
c.recordTurnBoundaryDegraded()
|
||
}
|
||
if requester, ok := a.(herdr.HandoffRequester); ok {
|
||
if _, statErr := os.Stat(filepath.Join(session.Worktree, herdr.HandoffFile)); statErr != nil {
|
||
if !session.HandoffRequested {
|
||
if reqErr := requester.RequestHandoff(ctx, session); reqErr == nil {
|
||
session.HandoffRequested = true
|
||
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]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 err := c.Store.Append(e); 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")
|
||
}
|
||
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())
|
||
}
|
||
// Best-effort: TASK.md only exists for worktrees this process can read
|
||
// locally (the GitWorktrees path). A herdr-hosted worktree on a remote
|
||
// machine (WorktreeCreator path) is the same cross-host gap named in
|
||
// AUDIT.md's federation-fork section — not solved here.
|
||
taskFileSHA, _ := continuity.TaskFileHash(w)
|
||
s, err := a.Lease(ctx, t.ID, w)
|
||
if err != nil {
|
||
return c.block(t, "lease: "+err.Error())
|
||
}
|
||
if p.HandoffRef != "" {
|
||
// §6.2 pickup validation: never bootstrap a successor onto a handoff
|
||
// whose anchor/dirty-file/TASK.md hashes don't match what's actually
|
||
// in the worktree. A failure here blocks the task rather than
|
||
// silently trusting an unvalidated ref (this is the gap AUDIT.md's
|
||
// B6 named as unreached from the live path).
|
||
h, err := continuity.Load(p.HandoffRef, c.Store)
|
||
if err != nil {
|
||
_ = a.Kill(ctx, s)
|
||
return c.block(t, "handoff: "+err.Error())
|
||
}
|
||
if err := continuity.ValidatePickup(w, h, taskFileSHA); err != nil {
|
||
_ = a.Kill(ctx, s)
|
||
return c.block(t, "pickup: "+err.Error())
|
||
}
|
||
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
|
||
s.TaskFileSHA = taskFileSHA
|
||
// 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)
|
||
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)
|
||
}
|