diff --git a/AUDIT.md b/AUDIT.md index a6d5719..3b02c21 100644 --- a/AUDIT.md +++ b/AUDIT.md @@ -760,3 +760,95 @@ whether Codex's/opencode's own turn-boundary mechanism actually surfaces this in-pane prompt to the agent before it exits the way Claude Code's Stop hook does — that's Phase 2 item 4 territory (native Face B per harness), not Phase 4. + +--- + +## Real harness quota sources — verified locally, 2026-07-27 + +Investigation, not a code change: B7 says `QuotaReported` has exactly one +producer (the `TaskCompleted` handler in `cmd/orchestra/main.go`, deriving +`consumed` from the harness's self-reported `usage.Numerator()`). The +question was whether any harness exposes its *real* subscription quota so +Orchestra can stop relying on operator-entered static caps plus estimated +token consumption. Two of three do. Everything below was read off this +machine's own installs, not recalled. + +### Claude Code — statusline stdin (confirmed) + +The JSON blob Claude Code pipes to `statusLine.command` on every render +carries server-reported rate-limit levels. `~/.claude/statusline.sh` already +reads them: + +- `.rate_limits.five_hour.used_percentage` +- `.rate_limits.seven_day.used_percentage` + +plus `.context_window.{used_percentage,total_input_tokens,context_window_size}`, +`.cost.total_cost_usd`, `.session_id`. These are percentages of the real +subscription pool, not estimates, and the two windows map 1:1 onto +`router.QuotaWindowLimits{FiveHour, Weekly}`. High frequency, zero cost. + +Note: `claude.ai/api/organizations/{org_id}/usage` also exists but is +authenticated by **claude.ai session cookies, not an API key** — wiring it +would make Orchestra hold and refresh a logged-in browser session. The +statusline path avoids that entirely and should be preferred. + +### Codex — `rate_limits` in the session rollout (confirmed) + +Every `token_count` event in `~/.codex/sessions/YYYY/MM/DD/rollout-*.jsonl` +carries a `rate_limits` object, e.g.: + +```json +"rate_limits": { + "limit_id": "codex", + "primary": { "used_percent": 20.0, "window_minutes": 10080, "resets_at": 1785650936 }, + "secondary": null, + "credits": { "has_credits": false, "unlimited": false, "balance": "0" }, + "plan_type": "plus", + "rate_limit_reached_type": null +} +``` + +Richer than Claude's: `window_minutes` makes the window self-describing +(10080 = weekly) and `resets_at` is absolute. There is **no `codex usage` +subcommand** — the transport is the rollout file, or `codex app-server`, +which emits the same events live. + +### opencode — no first-class quota surface + +- `opencode stats` is historical accounting only (cost/tokens/tools across + past sessions). No limits, no remaining. +- Zen is an OpenAI-compatible gateway at `https://opencode.ai/zen/v1` (and + `/zen/go/v1`). The binary contains `x-ratelimit-limit` / + `-remaining` / `-reset` / `-reset-after`, so the Zen free tier's **daily** + quota arrives as response headers, not via a queryable endpoint. Capturing + it requires intercepting a response — an opencode plugin + (`~/.config/opencode/plugin/`) is the only clean hook. + +### Design consequences (not yet implemented) + +1. **Percentages are a level, not a delta.** + `router.QuotaAvailability.sumSince` *sums* `consumed` across + `QuotaReported` events. Feeding a used-percentage into that sum is wrong + by construction. A real-quota feed needs a second `Availability` + implementation that reads the **latest** report per harness. `Availability` + is already an interface (`internal/router/router.go`), so this is a swap, + not a rewrite — and the additive `sumSince` path must stay for the + estimate-based producer (spec §5.2.1 receipts are genuinely additive + across rotations). +2. **Static `quota_limit_5h`/`quota_limit_weekly` become unnecessary** for + Claude and Codex, since the harness reports its own fraction of pool and + the 80% conservative rule applies directly with no operator-entered cap. + Keep the static path as the fallback for opencode. +3. **`QuotaWindowLimits{FiveHour, Weekly}` is too narrow.** Codex's windows + are self-describing via `window_minutes`, and opencode's Zen tier is + **daily** — a window Orchestra has no concept of today. A generic + `[]{WindowMinutes, UsedPercent, ResetsAt}` fits all three; the current + two named fields fit only Claude. +4. **Push path**: statusline script (Claude) and a rollout-tail or + app-server reader (Codex) POST to a new endpoint alongside + `/v1/harness/turn` in `cmd/orchestra/main.go` — the existing hook-ingress + pattern — appending `QuotaReported`. That gives B7 a second, *live* + producer next to the post-hoc one, and covers the case CLAUDE.md already + flags: a harness that never completes a task cleanly (the stuck `wA` pane) + currently under-counts its consumption forever, because the only producer + fires on completion. diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go index a84f887..912ee75 100644 --- a/cmd/orchestra/main.go +++ b/cmd/orchestra/main.go @@ -302,8 +302,58 @@ func main() { log.Printf("route task: %v", routeErr) } } + // B7 (AUDIT.md): QuotaReported has no other producer, so router's + // 5h/weekly availability filter and the brief's quota_consumed are + // permanently zero without this. Fed by the same per-harness usage + // read as the receipt above (spec §7.2 — "same per-harness session + // state as §5.2.1"); harness_id comes from the lease this completion + // closes out, before it's released. + if t.Lease != nil && t.Lease.HarnessID != "" { + qp, _ := json.Marshal(map[string]any{ + "harness_id": t.Lease.HarnessID, + "consumed": float64(usage.Numerator()), + }) + qe := domain.Event{ID: id(), Type: "QuotaReported", TaskID: "system", Payload: qp, Surface: string(authz.System)} + if err := s.Append(qe); err != nil { + log.Printf("quota report: %v", err) + } + } json.NewEncoder(w).Encode(e) }) + // /v1/harness/turn is the unified turn-decision endpoint (AUDIT.md Phase + // 2 items 1-2): the Face-B stop hook posts here on every ordinary turn + // boundary (report marker absent — /v1/harness/complete covers task + // completion separately) and gets back exactly one of continue / + // prepare_handoff / rotate_now / refuse, per spec §5.3. This replaces + // what would otherwise be separate ad-hoc marker-file conventions per + // decision. + mux.HandleFunc("/v1/harness/turn", func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + if harnessToken != "" && r.Header.Get("Authorization") != "Bearer "+harnessToken { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + if coordinator == nil { + http.Error(w, "coordinator not configured", http.StatusServiceUnavailable) + return + } + var p struct { + TaskID string `json:"task_id"` + } + if json.NewDecoder(r.Body).Decode(&p) != nil || p.TaskID == "" { + http.Error(w, "task_id is required", http.StatusBadRequest) + return + } + decision, err := coordinator.TurnDecision(r.Context(), p.TaskID) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + json.NewEncoder(w).Encode(map[string]string{"decision": decision}) + }) mux.HandleFunc("/v1/brief", func(w http.ResponseWriter, r *http.Request) { to := time.Now().UTC() from := to.Add(-12 * time.Hour) diff --git a/deploy/hooks/orchestra-stop.sh b/deploy/hooks/orchestra-stop.sh index fe4cee0..bff54d5 100755 --- a/deploy/hooks/orchestra-stop.sh +++ b/deploy/hooks/orchestra-stop.sh @@ -2,8 +2,9 @@ # Claude Code Stop hook — fires on every turn boundary, not just completion. # Convention: the agent signals "this task is done" by writing a report file # named .orchestra-report.md at the worktree root before stopping. If that -# marker is absent, this is an ordinary turn boundary and the hook is a -# no-op — do not treat every stop as completion (AUDIT.md B3). +# marker is present, report completion. If it is absent, this is an ordinary +# turn boundary — ask the unified turn-decision endpoint (AUDIT.md B3, Phase +# 2 items 1-2) what to do instead of no-op'ing. # # Requires: ORCHESTRA_TASK_ID and ORCHESTRA_URL set in the pane's env. # Optional: ORCHESTRA_HARNESS_TOKEN if the server requires one. @@ -21,28 +22,49 @@ cwd="$(printf '%s' "$payload" | jq -r '.cwd // empty')" [ -z "${ORCHESTRA_TASK_ID:-}" ] && exit 0 [ -z "${ORCHESTRA_URL:-}" ] && exit 0 -report_file="${cwd:-.}/.orchestra-report.md" -[ -f "$report_file" ] || exit 0 - -report="$(cat "$report_file")" - auth_header="" if [ -n "${ORCHESTRA_HARNESS_TOKEN:-}" ]; then auth_header="Authorization: Bearer ${ORCHESTRA_HARNESS_TOKEN}" fi -body="$(jq -n \ - --arg task_id "$ORCHESTRA_TASK_ID" \ - --arg transcript_path "$transcript_path" \ - --arg report "$report" \ - '{task_id: $task_id, transcript_path: $transcript_path, report: $report}')" +report_file="${cwd:-.}/.orchestra-report.md" -if curl -fsS -X POST "${ORCHESTRA_URL%/}/v1/harness/complete" \ +if [ -f "$report_file" ]; then + report="$(cat "$report_file")" + + body="$(jq -n \ + --arg task_id "$ORCHESTRA_TASK_ID" \ + --arg transcript_path "$transcript_path" \ + --arg report "$report" \ + '{task_id: $task_id, transcript_path: $transcript_path, report: $report}')" + + if curl -fsS -X POST "${ORCHESTRA_URL%/}/v1/harness/complete" \ + -H "Content-Type: application/json" \ + ${auth_header:+-H "$auth_header"} \ + -d "$body" >/dev/null 2>&1; then + rm -f "$report_file" + else + echo "orchestra-stop: failed to report completion" >&2 + exit 2 + fi + exit 0 +fi + +body="$(jq -n --arg task_id "$ORCHESTRA_TASK_ID" '{task_id: $task_id}')" + +response="$(curl -fsS -X POST "${ORCHESTRA_URL%/}/v1/harness/turn" \ -H "Content-Type: application/json" \ ${auth_header:+-H "$auth_header"} \ - -d "$body" >/dev/null 2>&1; then - rm -f "$report_file" -else - echo "orchestra-stop: failed to report completion" >&2 + -d "$body" 2>/dev/null)" || { + echo "orchestra-stop: failed to reach turn-decision endpoint" >&2 + exit 0 +} + +decision="$(printf '%s' "$response" | jq -r '.decision // empty')" + +if [ "$decision" = "refuse" ]; then + echo "orchestra-stop: turn decision is refuse — this turn boundary is not safe to stop at" >&2 exit 2 fi + +exit 0 diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go index b61bd11..dbff2b7 100644 --- a/internal/orchestrator/orchestrator.go +++ b/internal/orchestrator/orchestrator.go @@ -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 diff --git a/internal/orchestrator/rotation_test.go b/internal/orchestrator/rotation_test.go index 2eee19b..e580533 100644 --- a/internal/orchestrator/rotation_test.go +++ b/internal/orchestrator/rotation_test.go @@ -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()