feat(orchestrator): wire turn-decision endpoint and QuotaReported producer

Closes Phase 2 items 1-2 (AUDIT.md): Coordinator.TurnDecision evaluates occupancy/turn-boundary/handoff state synchronously per turn and returns continue/prepare_handoff/rotate_now/refuse, exposed via POST /v1/harness/turn. The Claude Stop hook now calls it on ordinary turn boundaries instead of no-op'ing, and exits 2 on refuse. Also closes B7's post-hoc producer: /v1/harness/complete now appends a QuotaReported event from the completing lease's harness usage, so the router's quota-availability filter and the brief's quota_consumed stop evaluating against a permanent zero. Live per-harness push producers (Claude statusline, Codex rollout tail) remain unbuilt — investigation recorded in AUDIT.md. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01W1rkJ2hBMybnJctPbcy4tT
2026-07-27 23:17:21 +04:00
parent 0ca78243b9
commit 972845bd98
5 changed files with 382 additions and 21 deletions
@@ -175,6 +175,11 @@ type Coordinator struct {
 	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
 }

 type MonitorHealth struct {
@@ -367,6 +372,7 @@ func (c *Coordinator) Monitor(ctx context.Context, hard float64, interval time.D
 		c.setMonitorHealth(err, 0)
 		return err
 	}
+	c.Hard = hard
 	if interval <= 0 {
 		interval = 30 * time.Second
 	}
@@ -595,6 +601,95 @@ func (c *Coordinator) rotate(ctx context.Context, hard float64) {
 	}
 }

+// 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)
+	}
+	occupancy, err := a.Occupancy(session)
+	if err != nil {
+		return "", fmt.Errorf("orchestrator: occupancy: %w", err)
+	}
+	if occupancy < c.Hard {
+		return TurnContinue, 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
+		}
+	}
+	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": "threshold", "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
@@ -30,8 +30,8 @@ func (a *fakeAdapter) Release(context.Context, herdr.Session) (string, error) {
 	a.releases++
 	return a.ref, nil
 }
-func (a *fakeAdapter) Kill(context.Context, herdr.Session) error         { return nil }
-func (a *fakeAdapter) Occupancy(herdr.Session) (float64, error)         { return a.occupancy, nil }
+func (a *fakeAdapter) Kill(context.Context, herdr.Session) error { return nil }
+func (a *fakeAdapter) Occupancy(herdr.Session) (float64, error)  { return a.occupancy, nil }
 func (a *fakeAdapter) AtTurnBoundary(context.Context, herdr.Session) (bool, error) {
 	return a.boundary, nil
 }
@@ -137,6 +137,108 @@ func TestRotationEmitsValidReleaseWithAnchorSHA(t *testing.T) {

 func mustJSON(v any) []byte { b, _ := json.Marshal(v); return b }

+// TestTurnDecision guards AUDIT.md Phase 2 items 1-2: the synchronous,
+// per-turn counterpart to rotate() must return the same verdicts the
+// periodic ticker would compute, and rotate_now must actually perform the
+// release (not just report what rotate() would eventually do).
+func TestTurnDecision(t *testing.T) {
+	repo := t.TempDir()
+	run(t, repo, "init")
+	run(t, repo, "config", "user.email", "t@t")
+	run(t, repo, "config", "user.name", "t")
+	run(t, repo, "commit", "--allow-empty", "-m", "init")
+
+	newCoordinator := func(a *fakeAdapter) (*orchestrator.Coordinator, *store.Store, domain.Task) {
+		s, err := store.Open(t.TempDir())
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskCreated", TaskID: "t1", Surface: string(authz.System), Payload: mustJSON(map[string]any{
+			"source": "jsonl", "external_id": "1", "project": "p",
+		})}); err != nil {
+			t.Fatal(err)
+		}
+		task := s.Tasks()[0]
+		c := &orchestrator.Coordinator{Store: s, Worktrees: worktrees{path: repo}, Adapters: adapters{a}, StatePath: t.TempDir() + "/sessions.json", Hard: .8}
+		leaseEvt, err := s.Lease(task.ID, "h1", time.Minute)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := c.Start(context.Background(), leaseEvt); err != nil {
+			t.Fatal(err)
+		}
+		return c, s, task
+	}
+
+	t.Run("continue below threshold", func(t *testing.T) {
+		a := &fakeAdapter{occupancy: .5}
+		c, _, task := newCoordinator(a)
+		decision, err := c.TurnDecision(context.Background(), task.ID)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if decision != orchestrator.TurnContinue {
+			t.Fatalf("decision=%q want %q", decision, orchestrator.TurnContinue)
+		}
+	})
+
+	t.Run("refuse when not at turn boundary", func(t *testing.T) {
+		a := &fakeAdapter{occupancy: .95, boundary: false}
+		c, _, task := newCoordinator(a)
+		decision, err := c.TurnDecision(context.Background(), task.ID)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if decision != orchestrator.TurnRefuse {
+			t.Fatalf("decision=%q want %q", decision, orchestrator.TurnRefuse)
+		}
+	})
+
+	t.Run("rotate_now releases and emits a valid TaskReleased", func(t *testing.T) {
+		a := &fakeAdapter{occupancy: .95, boundary: true}
+		c, st, task := newCoordinator(a)
+		artifactRef, err := st.PutArtifact([]byte("handoff"))
+		if err != nil {
+			t.Fatal(err)
+		}
+		a.ref = artifactRef
+		decision, err := c.TurnDecision(context.Background(), task.ID)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if decision != orchestrator.TurnRotateNow {
+			t.Fatalf("decision=%q want %q releases=%d", decision, orchestrator.TurnRotateNow, a.releases)
+		}
+		if a.releases == 0 {
+			t.Fatal("adapter Release was never invoked")
+		}
+		got, ok := st.Task(task.ID)
+		if !ok || got.State != domain.StateQueued {
+			t.Fatalf("task state=%v ok=%v, want queued", got.State, ok)
+		}
+	})
+
+	t.Run("prepare_handoff requests handoff without releasing", func(t *testing.T) {
+		a := &handoffRequestingAdapter{fakeAdapter: fakeAdapter{occupancy: .95, boundary: true}}
+		c, st, task := newCoordinator(&a.fakeAdapter)
+		c.Adapters = adapters{a}
+		decision, err := c.TurnDecision(context.Background(), task.ID)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if decision != orchestrator.TurnPrepareHandoff {
+			t.Fatalf("decision=%q want %q", decision, orchestrator.TurnPrepareHandoff)
+		}
+		if a.releases != 0 {
+			t.Fatal("adapter Release was invoked, expected only a handoff request")
+		}
+		got, ok := st.Task(task.ID)
+		if !ok || got.State != domain.StateLeased {
+			t.Fatalf("task state=%v ok=%v, want leased", got.State, ok)
+		}
+	})
+}
+
 // TestStartBlocksOnInvalidPickup guards AUDIT.md's B6/Phase 4 item 4:
 // Coordinator.Start must run §6.2 pickup validation against the real
 // worktree before bootstrapping a successor onto a handoff_ref, and refuse
@@ -294,8 +396,8 @@ func (a *noBoundaryAdapter) Release(context.Context, herdr.Session) (string, err
 	a.releases++
 	return a.ref, nil
 }
-func (a *noBoundaryAdapter) Kill(context.Context, herdr.Session) error         { return nil }
-func (a *noBoundaryAdapter) Occupancy(herdr.Session) (float64, error)         { return a.occupancy, nil }
+func (a *noBoundaryAdapter) Kill(context.Context, herdr.Session) error { return nil }
+func (a *noBoundaryAdapter) Occupancy(herdr.Session) (float64, error)  { return a.occupancy, nil }

 func setupRotationTask(t *testing.T, repo string) (*store.Store, string, domain.Task, string) {
 	t.Helper()