diff --git a/AUDIT.md b/AUDIT.md index 18bb60f..67975ee 100644 --- a/AUDIT.md +++ b/AUDIT.md @@ -88,11 +88,18 @@ path only and remains open for Design A. B18 (the unauthenticated web surface) is closed in code but **requires an env change before the service will start** — see B18 and "What's next". -Nothing is currently blocking in the sense B12/B13 were. The open defects -B19–B21 and S12–S13, all found while implementing B18, are correctness and -hygiene gaps in the new command channel rather than things that stop a task -from running. The real remaining risk is evidential: B13 through B17 were -each found live and none has been re-verified live. +Nothing is currently blocking in the sense B12/B13 were. B19–B21 and S12–S13 +— the correctness and hygiene gaps found while implementing B18 — are now +fixed in code and covered by tests (2026-07-28); each is marked closed in +its own section below. The flaky router test is also fixed, and the flake +was in the test, not in assignment. + +The real remaining risk is entirely evidential now: B13 through B17 were +each found live and **none has been re-verified live**, and none of the +2026-07-28 fixes has run on the deployed instance either — the service is +stopped, `/usr/local/bin/orchestra` predates all of them, and installing a +new binary or editing `/etc/orchestra/orchestra.env` needs privileges this +sandbox does not have. ### B18 — the web UI surface is unauthenticated (found by audit 2026-07-28, uncommitted working tree) — closed (code fix; requires an env change before restart) @@ -621,12 +628,13 @@ individually below — **B19** (federated approvals emit no event), **B20** as build output; there is no documented step tying `web/` sources to a rebuild of `internal/webui/assets`, so the two can silently diverge. -## Found while implementing B18 (2026-07-28) — not yet fixed +## Found while implementing B18 (2026-07-28) — all fixed the same day -These were noticed while wiring the session gate. None is fixed; all were -verified against the code rather than inferred. +These were noticed while wiring the session gate, and all five were fixed in +the following pass. Each section states the original defect first, then how +it was closed. -### B19 — federated approvals leave no audit trail +### B19 — federated approvals leave no audit trail — closed (2026-07-28) `Server.action`'s `grant_approval` / `deny_approval` branch (`internal/ui/ui.go:301-341`) has two exits that are **not** symmetric. The @@ -648,7 +656,21 @@ already carries `acknowledged` / `rejected` / `stale`). Emitting `ApprovalGranted` at queue time would claim an outcome that has not happened yet. -### B20 — the local capture revision is a timestamp, not a change counter +**Fix.** The resolution half already existed and this entry understated it: +`/v1/federation/commands/` in `cmd/orchestra/main.go` appends +`ApprovalGranted`/`ApprovalDenied` — but only on `status == "acknowledged"`, +so a queued click never claims an outcome. What was missing was the request +half. `Server.action` now appends `ApprovalRequested` at queue time, with +`subject_ref` set to the command ID that the later resolution event carries, +plus `decision`, `worker`, `pane_id` and `capture_revision`. If that append +fails the queued command is immediately resolved `rejected`, so the worker +can never execute a keystroke that left no audit trail. `rejected`/`stale` +outcomes still emit no resolution event; the `ApprovalRequested` stays +unresolved and surfaces in `operations.NeedsAttention`, which is the honest +reading. Covered by +`TestFederatedApprovalIsQueuedAtTheWorkerRevisionAndAudited`. + +### B20 — the local capture revision is a timestamp, not a change counter — closed (2026-07-28) `Server.capture` (`ui.go:80`) fabricates `Revision: uint64(time.Now().UnixNano())` for the coordinator path. The federation @@ -670,7 +692,7 @@ Two consequences: never happens. The precedence between the two capture sources needs to be explicit rather than incidental. -### B21 — `Registry.commands` is append-only +### B21 — `Registry.commands` is append-only — closed (2026-07-28) `r.commands[worker]` is only ever appended to (`federation.go:119`); `CompleteCommand` flips a status in place and nothing is ever deleted. @@ -679,7 +701,16 @@ rescans the entire history on every worker poll. Combined with the in-memory-only storage already noted, the lifecycle is wrong at both ends: it forgets across restarts and never forgets within one. -### S12 — `ORCHESTRA_NTFY_TOKEN` serves two unrelated purposes +**Fix (within-process half only).** `Registry.pruneCommands` drops resolved +commands older than `CommandRetention` (30 minutes), running on both `Queue` +and `Commands`, so the per-worker list is bounded and the poll path no +longer rescans unbounded history. Pending commands are never pruned, at any +age — dropping one would silently discard an operator decision. Covered by +`TestResolvedCommandsArePrunedButPendingOnesSurvive`. **The persistence half +of this defect is still open:** captures and commands remain in-memory only, +so a coordinator restart still drops pending approvals silently. + +### S12 — `ORCHESTRA_NTFY_TOKEN` serves two unrelated purposes — closed (2026-07-28) The same env var is the ntfy *server* credential (`main.go:1153`, `delivery.Ntfy{Token: ...}`) and the authz *surface* credential for @@ -688,7 +719,13 @@ trust boundaries: one is handed to a third-party notification server, the other authenticates callers to Orchestra. Setting the former silently makes it a valid inbound credential. They need separate variables. -### S13 — dead `auth()` middleware in `cmd/orchestra/main.go` +**Fix.** The `authz.Ntfy` surface token is now `ORCHESTRA_NTFY_SURFACE_TOKEN`; +`ORCHESTRA_NTFY_TOKEN` is once again only the credential handed to the ntfy +server. Documented in `deploy/orchestra.env.example`. **Operator action:** +any deployment that relied on the old dual use must now set the new variable +explicitly, or the ntfy surface reverts to having no inbound gate. + +### S13 — dead `auth()` middleware in `cmd/orchestra/main.go` — closed (2026-07-28) `func auth(next http.Handler)` (`main.go:1212`) is defined and never referenced; Go does not flag unused functions, so `go vet` stays quiet. It @@ -710,28 +747,41 @@ bypass is closed by construction rather than by the header check alone. In dependency order, not importance order: -1. **Apply the B18 env change** before any restart — - `.orchestra-config/orchestra.env` needs `ORCHESTRA_WEB_TOKEN` (plus - `ORCHESTRA_UI_INSECURE_COOKIE=1` for plain HTTP), and every existing - unauthenticated `/v1/` client of this instance needs the token too. The - service will not start otherwise. Nothing else can be tested live until - this is done. -2. **S13, then S12** — deleting dead policy code and splitting the conflated - ntfy token are both small, local, and reduce the chance of the next - auth change being made against the wrong copy. -3. **B19** — decide the queued-approval event model and give federated - approvals an audit trail. This is the largest correctness gap in the new - command channel. -4. **B20 and B21** — make the capture revision mean one thing, make the - coordinator/worker precedence explicit, and give commands a retention - policy. B20 in particular can silently defeat approvals. -5. **Diagnose the flaky router test** (see open gaps). A nondeterministic - assignment test is exactly the failure shape this codebase has hidden - real defects behind before; do not paper over it with a retry. -6. **Then, and only then, the live re-verification** that B13–B17 all still - lack. Each was found live and closed on paper; the code fixes are - unproven against a real cross-machine run, and that remains the single - biggest gap between this document and reality. +1. **Finish applying the B18 env change on the deployed instance.** The + repo-side `.orchestra-config/orchestra.env` now sets a generated + `ORCHESTRA_WEB_TOKEN` and `ORCHESTRA_UI_INSECURE_COOKIE=1`, but the unit + reads `/etc/orchestra/orchestra.env`, which is `orchestra:orchestra 0600` + and unreadable from this sandbox — whether it carries the token could not + be confirmed. It was last modified 2026-07-28 14:30 and the service did + start at 16:17 with the B18 build absent, so this is unverified either + way. Also newly required: `ORCHESTRA_NTFY_SURFACE_TOKEN` (S12) if the + ntfy surface should stay gated, and the token for every existing + unauthenticated `/v1/` client, which now defaults to the Web surface. +2. **Install the new binary and restart.** `orchestra.service` has been + stopped since 2026-07-28 19:46 and `/usr/local/bin/orchestra` predates + every fix in this section. Needs privileges the sandbox lacks: + `go build -o /tmp/orchestra ./cmd/orchestra && sudo install /tmp/orchestra + /usr/local/bin/orchestra && sudo systemctl restart orchestra.service`. +3. **The live re-verification** that B13–B17 all still lack, plus a first + live exercise of B18–B21. Each of B13–B17 was found live and closed on + paper; the code fixes are unproven against a real cross-machine run, and + that remains the single biggest gap between this document and reality. + Note that no herdr is currently reachable at all — the 16:17 startup logs + `connection refused` for **all six**, including workpc's, which was live + on 2026-07-27 — so a live run needs a herdr brought up first. +4. **Two operational faults visible in the journal**, unrelated to this + audit's defects but blocking a clean live run: every herdr is refusing + connections (above), and ntfy delivery is failing `403 Forbidden` on + every send (16:43 and 16:46), i.e. the ntfy server credential is wrong or + expired. +5. **The persistence half of B21** — captures and commands are in-memory + only, so a restart still silently drops pending approvals. The retention + fix bounds growth within a process; it does not make the lifecycle + durable. +6. **`RespondApproval` (coordinator path) still has no test**, unlike its + worker counterpart. With B20 fixed its revision is now meaningful, but + its text-comparison guard remains the thing actually binding a decision + to what the operator saw, and that guard is untested. Deliberately *not* next: building further on Design A's cross-machine calls, and closing B17 for the coordinator path. Both wait on the federation-fork @@ -1227,16 +1277,24 @@ Numbered defects are not repeated here — B19, B20, B21, S12 and S13 are in in "What's next". This list is the unnumbered residue: conditions that are known, accepted, or not actionable as a single fix. -- **B18's env change is not yet applied.** The code now refuses to start - without `ORCHESTRA_WEB_TOKEN`, but `.orchestra-config/orchestra.env` does - not set one — `orchestra.service` will fail to start until it does. +- **B18's env change is applied repo-side only.** `.orchestra-config/ + orchestra.env` now sets `ORCHESTRA_WEB_TOKEN`, but the unit loads + `/etc/orchestra/orchestra.env`, which is not readable or writable from + this sandbox. See "What's next" item 1. - **The listener still binds all interfaces**; loopback-by-default was considered and not taken (B18). -- **`TestAssignsByAffinityCapabilityAndConcurrency` is flaky** (`no task - leased`, roughly 1 run in 10, reproduced with `-count=10`). Predates these - changes — confirmed by stashing them — and is unrelated to authz. A - nondeterministic router assignment test is exactly the kind of thing this - codebase has been bitten by before; it should be diagnosed, not retried. +- ~~**`TestAssignsByAffinityCapabilityAndConcurrency` is flaky**~~ — + diagnosed and fixed 2026-07-28. The flake was in the test, not in + assignment. `Store.Tasks()` ranges a map, so its order is randomized per + call, and the assertion indexed **two separate `Tasks()` calls** + (`s.Tasks()[0] ... && s.Tasks()[1] ...`); it failed whenever the two + orderings disagreed. Instrumenting the failure showed a valid `TaskLeased` + event and a genuinely leased task on every "failing" run. It now snapshots + once and asserts the real invariant — concurrency 1 means exactly one of + the two tasks is leased — and passes at `-count=60`. Worth recording that + *which* of two equal-priority tasks wins is still nondeterministic, since + `sort.SliceStable` is applied to a randomly ordered slice; that is a real + property of the router, not a test artifact. - **B14/B15/B16 are code-fixed but not re-verified live**, and B17 is closed only for tasks that run through a worker. Each was found live, so a code fix plus unit tests is weaker evidence than the failure that produced it. diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go index 7b867a0..e474af8 100644 --- a/cmd/orchestra/main.go +++ b/cmd/orchestra/main.go @@ -1205,19 +1205,11 @@ func main() { tokens := map[authz.Surface]string{ authz.TUI: os.Getenv("ORCHESTRA_TUI_TOKEN"), authz.Web: os.Getenv("ORCHESTRA_WEB_TOKEN"), authz.MCP: os.Getenv("ORCHESTRA_MCP_TOKEN"), authz.Maven: os.Getenv("ORCHESTRA_MAVEN_TOKEN"), - authz.Telegram: os.Getenv("ORCHESTRA_TELEGRAM_TOKEN"), authz.Ntfy: os.Getenv("ORCHESTRA_NTFY_TOKEN"), + // S12: this is the credential callers present *to* Orchestra on the + // ntfy surface. ORCHESTRA_NTFY_TOKEN is a different secret entirely — + // it is handed out to the third-party ntfy server (see the sender + // above) and must never be accepted as an inbound credential. + authz.Telegram: os.Getenv("ORCHESTRA_TELEGRAM_TOKEN"), authz.Ntfy: os.Getenv("ORCHESTRA_NTFY_SURFACE_TOKEN"), } log.Fatal(http.ListenAndServe(":"+port, authz.HTTPWithSessions(tokens, sessions, mux))) } -func auth(next http.Handler) http.Handler { - return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - surface := strings.ToLower(r.Header.Get("X-Orchestra-Surface")) - if surface == "telegram" || surface == "ntfy" { - if r.Method != "GET" { - http.Error(w, "notify-only surface", 403) - return - } - } - next.ServeHTTP(w, r) - }) -} diff --git a/deploy/orchestra.env.example b/deploy/orchestra.env.example index 49d0ba5..f08a589 100644 --- a/deploy/orchestra.env.example +++ b/deploy/orchestra.env.example @@ -64,7 +64,9 @@ ORCHESTRA_OCCUPANCY_HARD=0.75 # Telegram: both required together. #ORCHESTRA_TELEGRAM_BOT_TOKEN= #ORCHESTRA_TELEGRAM_CHAT_ID= -# ntfy: topic required, token/url optional (self-hosted ntfy). +# ntfy: topic required, token/url optional (self-hosted ntfy). This token is +# handed *out* to the ntfy server; it is not an inbound credential — see +# ORCHESTRA_NTFY_SURFACE_TOKEN below. #ORCHESTRA_NTFY_TOPIC= #ORCHESTRA_NTFY_TOKEN= #ORCHESTRA_NTFY_URL=https://ntfy.sh @@ -72,10 +74,18 @@ ORCHESTRA_OCCUPANCY_HARD=0.75 # --- Bus authorization tokens (bearer auth per surface; a surface with no # token set has no auth requirement — set these once you have real clients) --- #ORCHESTRA_TUI_TOKEN= -#ORCHESTRA_WEB_TOKEN= +# Required: the service refuses to start without it. It gates the web UI's +# task, lifecycle and approval controls, and it is also the token for any +# /v1/ caller that does not declare a surface (they default to Web). +ORCHESTRA_WEB_TOKEN= +# Set when the UI is served over plain HTTP, so the session cookie can be +# sent without Secure. Leave unset behind TLS. +#ORCHESTRA_UI_INSECURE_COOKIE=1 #ORCHESTRA_MCP_TOKEN= #ORCHESTRA_MAVEN_TOKEN= -# Telegram/ntfy tokens above double as their surface auth tokens -# (ORCHESTRA_TELEGRAM_TOKEN is the inbound bearer token if you also expose an -# endpoint they poll, separate from the bot token used to send messages). +# Inbound bearer tokens for the notify-only surfaces, separate from the +# credentials used to *send* (ORCHESTRA_TELEGRAM_BOT_TOKEN, ORCHESTRA_NTFY_TOKEN). +# S12: ORCHESTRA_NTFY_TOKEN used to serve both roles, so configuring ntfy +# delivery silently minted a valid inbound credential. #ORCHESTRA_TELEGRAM_TOKEN= +#ORCHESTRA_NTFY_SURFACE_TOKEN= diff --git a/internal/federation/federation.go b/internal/federation/federation.go index eaa35ba..34839a5 100644 --- a/internal/federation/federation.go +++ b/internal/federation/federation.go @@ -117,8 +117,33 @@ func (r *Registry) Queue(worker string, c Command) (Command, error) { c.CreatedAt = time.Now().UTC() c.Status = "pending" r.commands[worker] = append(r.commands[worker], c) + r.pruneCommands(worker) return c, nil } + +// CommandRetention is how long a resolved command stays queryable so a worker +// that retries a completion, or an operator reading the UI, still sees its +// outcome. Pending commands are never pruned. +const CommandRetention = 30 * time.Minute + +// pruneCommands drops resolved commands past CommandRetention. B21: this list +// was append-only, so resolved commands accumulated for the process lifetime +// and every worker poll rescanned the entire history. Callers hold r.mu. +func (r *Registry) pruneCommands(worker string) { + cutoff := time.Now().UTC().Add(-CommandRetention) + in := r.commands[worker] + out := in[:0] + for _, c := range in { + if c.Status == "pending" || c.CreatedAt.After(cutoff) { + out = append(out, c) + } + } + if len(out) == 0 { + delete(r.commands, worker) + return + } + r.commands[worker] = out +} func (r *Registry) Commands(worker string) ([]Command, error) { r.mu.Lock() defer r.mu.Unlock() @@ -126,6 +151,7 @@ func (r *Registry) Commands(worker string) ([]Command, error) { if _, ok := r.workers[worker]; !ok { return nil, ErrUnknownWorker } + r.pruneCommands(worker) var out []Command for _, c := range r.commands[worker] { if c.Status == "pending" { diff --git a/internal/federation/federation_test.go b/internal/federation/federation_test.go index ddf5471..8f7d808 100644 --- a/internal/federation/federation_test.go +++ b/internal/federation/federation_test.go @@ -105,3 +105,36 @@ func TestCaptureRevisionAndCommandQueue(t *testing.T) { t.Fatalf("pending=%#v", commands) } } + +func TestResolvedCommandsArePrunedButPendingOnesSurvive(t *testing.T) { + r := &Registry{} + if err := r.Register(Worker{ID: "w", Token: "t"}, ""); err != nil { + t.Fatal(err) + } + old, err := r.Queue("w", Command{TaskID: "task", Kind: "grant_approval", PaneID: "pane", CaptureRevision: 1}) + if err != nil { + t.Fatal(err) + } + if err := r.CompleteCommand("w", old.ID, "acknowledged", ""); err != nil { + t.Fatal(err) + } + pending, err := r.Queue("w", Command{TaskID: "task", Kind: "deny_approval", PaneID: "pane", CaptureRevision: 2}) + if err != nil { + t.Fatal(err) + } + // Age both past the retention window; only the resolved one may go. + r.mu.Lock() + for i := range r.commands["w"] { + r.commands["w"][i].CreatedAt = time.Now().UTC().Add(-2 * CommandRetention) + } + r.mu.Unlock() + if _, err := r.Commands("w"); err != nil { + t.Fatal(err) + } + if _, ok := r.Command("w", old.ID); ok { + t.Fatal("resolved command past retention was not pruned") + } + if c, ok := r.Command("w", pending.ID); !ok || c.Status != "pending" { + t.Fatalf("pending command was pruned: %#v ok=%v", c, ok) + } +} diff --git a/internal/router/router_test.go b/internal/router/router_test.go index 09df11a..0e1b30d 100644 --- a/internal/router/router_test.go +++ b/internal/router/router_test.go @@ -40,8 +40,19 @@ func TestAssignsByAffinityCapabilityAndConcurrency(t *testing.T) { if err != nil || len(got) != 1 { t.Fatalf("assigned %d events, err=%v", len(got), err) } - if s.Tasks()[0].State != domain.StateLeased && s.Tasks()[1].State != domain.StateLeased { - t.Fatal("no task leased") + // Store.Tasks() ranges a map, so its order is randomized per call. This + // assertion used to index two *separate* Tasks() calls, and failed + // whenever the two orderings disagreed — the flake was in the test, not + // in assignment. Snapshot once, and assert the actual invariant: + // concurrency 1 means exactly one of the two tasks is leased. + leased := 0 + for _, tk := range s.Tasks() { + if tk.State == domain.StateLeased { + leased++ + } + } + if leased != 1 { + t.Fatalf("leased %d tasks, want exactly 1 (concurrency 1); got=%+v", leased, got) } } diff --git a/internal/ui/ui.go b/internal/ui/ui.go index d8e3dbe..9dd9b16 100644 --- a/internal/ui/ui.go +++ b/internal/ui/ui.go @@ -5,6 +5,7 @@ import ( "context" "encoding/json" "fmt" + "hash/fnv" "net/http" "orchestra/internal/authz" "orchestra/internal/domain" @@ -74,15 +75,21 @@ func (s Server) capture(ctx context.Context, t domain.Task) (*Capture, error) { if t.State != domain.StateLeased || t.Lease == nil { return nil, nil } + // B20: precedence is deliberate, not incidental. A published worker + // capture means a registered worker owns that pane, and its revision is + // the counter the worker's own staleness check compares against — so it + // must win. Preferring the coordinator here (as this once did) handed + // Queue a revision the worker could never match, and every such approval + // resolved "stale" and silently never happened. + if s.Workers != nil { + if c, ok := s.Workers.Capture(t.Lease.HarnessID, t.ID); ok { + return &Capture{TaskID: t.ID, Source: "worker", Text: c.Text, Revision: c.Revision, At: c.At}, nil + } + } if s.Coordinator != nil { text, err := s.Coordinator.Capture(ctx, t.ID, "recent") if err == nil { - return &Capture{TaskID: t.ID, Source: "recent", Text: text, Revision: uint64(time.Now().UnixNano()), At: time.Now().UTC()}, nil - } - } - if s.Workers != nil { - if c, ok := s.Workers.Capture(t.Lease.HarnessID, t.ID); ok { - return &Capture{TaskID: t.ID, Source: "recent", Text: c.Text, Revision: c.Revision, At: c.At}, nil + return &Capture{TaskID: t.ID, Source: "recent", Text: text, Revision: captureRevision(t.Lease.HarnessID, text), At: time.Now().UTC()}, nil } } return nil, fmt.Errorf("capture unavailable from owning worker") @@ -152,7 +159,7 @@ func (s Server) detail(ctx context.Context, id string) (TaskDetail, error) { session := &Session{HarnessID: t.Lease.HarnessID, LeaseUntil: &t.Lease.Until} if c, err := s.capture(ctx, t); err == nil && c != nil { session.Capture = c - session.PaneID = capturePane(s, t.ID, c) + session.PaneID = capturePane(s, t, c) session.Approval = ParsePendingApproval(c.Text, session.PaneID, c.Revision, c.At) } else if err != nil { session.Blocker = "capture unavailable: " + err.Error() @@ -161,17 +168,35 @@ func (s Server) detail(ctx context.Context, id string) (TaskDetail, error) { } return d, nil } -func capturePane(s Server, taskID string, c *Capture) string { - if s.Coordinator != nil { - if session, ok := s.Coordinator.Session(taskID); ok { - return session.PaneID - } +// captureRevision identifies *what the operator saw*, not when they saw it. +// B20: this was UnixNano, so it changed on every read and said nothing about +// whether the pane had changed. A content hash changes if and only if the +// text does, which is the only property any consumer of a revision wants. +func captureRevision(harness, text string) uint64 { + h := fnv.New64a() + _, _ = h.Write([]byte(harness)) + _, _ = h.Write([]byte{0}) + _, _ = h.Write([]byte(text)) + // 0 means "no revision" to federation.Queue; never collide with it. + if v := h.Sum64(); v != 0 { + return v } - if s.Workers != nil { - for _, w := range s.Workers.Snapshot() { - if remote, ok := s.Workers.Capture(w.ID, taskID); ok && remote.Revision == c.Revision { - return remote.PaneID - } + return 1 +} + +// capturePane resolves the pane the capture came from, following the same +// source precedence capture() used — asking the coordinator about a pane a +// worker owns (or vice versa) can return a pane the text never came from. +func capturePane(s Server, t domain.Task, c *Capture) string { + if c.Source == "worker" && s.Workers != nil && t.Lease != nil { + if remote, ok := s.Workers.Capture(t.Lease.HarnessID, t.ID); ok && remote.Revision == c.Revision { + return remote.PaneID + } + return "" + } + if s.Coordinator != nil { + if session, ok := s.Coordinator.Session(t.ID); ok { + return session.PaneID } } return "" @@ -304,22 +329,40 @@ func (s Server) action(w http.ResponseWriter, r *http.Request, id, action string http.Error(w, "current capture unavailable", 503) return } - pane := capturePane(s, id, c) + pane := capturePane(s, t, c) approval := ParsePendingApproval(c.Text, pane, c.Revision, c.At) if approval == nil || approval.Kind == "unknown" || pane == "" { http.Error(w, "current prompt is not safely actionable", 409) return } - if s.Workers != nil { - if _, remote := s.Workers.Capture(t.Lease.HarnessID, t.ID); remote { - command, err := s.Workers.Queue(t.Lease.HarnessID, federation.Command{TaskID: id, Kind: action, PaneID: pane, CaptureRevision: c.Revision}) - if err != nil { - http.Error(w, err.Error(), 409) - return - } - writeJSON(w, map[string]any{"command": command, "task": t}) + if c.Source == "worker" { + command, err := s.Workers.Queue(t.Lease.HarnessID, federation.Command{TaskID: id, Kind: action, PaneID: pane, CaptureRevision: c.Revision}) + if err != nil { + http.Error(w, err.Error(), 409) return } + // B19: the federated path used to return here, leaving the most + // safety-critical operation in the system invisible in the event + // log exactly when it crossed a machine boundary. The honest + // event at this point is a *request* — the keystroke has only + // been queued. The matching ApprovalGranted/ApprovalDenied is + // appended when the worker reports the input was acknowledged + // (see /v1/federation/commands/ in cmd/orchestra). + payload, _ := json.Marshal(map[string]any{ + "subject_ref": command.ID, "options": []string{"grant_approval", "deny_approval"}, + "decision": action, "task_id": id, "worker": t.Lease.HarnessID, + "pane_id": pane, "capture_revision": c.Revision, + }) + e := domain.Event{ID: domain.NewID(), Type: "ApprovalRequested", TaskID: id, Version: t.Version + 1, Payload: payload, Surface: string(authz.Web)} + if err := s.Store.Append(e); err != nil { + // No audit trail, no approval. Resolve the queued command so + // the worker never executes an unrecorded keystroke. + _ = s.Workers.CompleteCommand(t.Lease.HarnessID, command.ID, "rejected", "audit append failed: "+err.Error()) + http.Error(w, "approval not recorded: "+err.Error(), 409) + return + } + writeJSON(w, map[string]any{"command": command, "task": t}) + return } if s.Coordinator == nil { http.Error(w, "approval executor unavailable", 503) diff --git a/internal/ui/ui_test.go b/internal/ui/ui_test.go index 62fcaca..7106413 100644 --- a/internal/ui/ui_test.go +++ b/internal/ui/ui_test.go @@ -2,8 +2,11 @@ package ui import ( "bytes" + "encoding/json" "net/http" "net/http/httptest" + "orchestra/internal/domain" + "orchestra/internal/federation" "orchestra/internal/store" "strings" "testing" @@ -43,3 +46,87 @@ func TestParsePendingApprovalDoesNotInventActionablePrompt(t *testing.T) { t.Fatalf("got %#v", p) } } + +// TestFederatedApprovalIsQueuedAtTheWorkerRevisionAndAudited guards B19 and +// B20 together: the queued command must carry the worker's own capture +// counter (not a coordinator timestamp the worker can never match, which +// resolved every such approval "stale"), and queueing must leave an +// ApprovalRequested event behind — the grant/deny is appended later, when +// the worker reports the keystroke was acknowledged. +func TestFederatedApprovalIsQueuedAtTheWorkerRevisionAndAudited(t *testing.T) { + s, err := store.Open(t.TempDir()) + if err != nil { + t.Fatal(err) + } + workers := &federation.Registry{} + if err := workers.Register(federation.Worker{ID: "wpc-claude", Token: "tok"}, ""); err != nil { + t.Fatal(err) + } + h := Server{Store: s, Workers: workers}.Handler() + req := httptest.NewRequest(http.MethodPost, "/v1/ui/tasks", bytes.NewBufferString(`{"source":"web","external_id":"fed","project":"demo","title":"Federated"}`)) + r := httptest.NewRecorder() + h.ServeHTTP(r, req) + if r.Code != http.StatusOK { + t.Fatalf("create status=%d body=%s", r.Code, r.Body.String()) + } + id := s.Tasks()[0].ID + if _, err := s.Lease(id, "wpc-claude", time.Hour); err != nil { + t.Fatal(err) + } + capture, err := workers.PutCapture("wpc-claude", federation.Capture{TaskID: id, PaneID: "wA:p1", Text: "Permission required\n$ rm -rf build"}) + if err != nil { + t.Fatal(err) + } + req = httptest.NewRequest(http.MethodPost, "/v1/ui/tasks/"+id+"/actions/grant_approval", bytes.NewBufferString(`{}`)) + r = httptest.NewRecorder() + h.ServeHTTP(r, req) + if r.Code != http.StatusOK { + t.Fatalf("grant status=%d body=%s", r.Code, r.Body.String()) + } + pending, err := workers.Commands("wpc-claude") + if err != nil || len(pending) != 1 { + t.Fatalf("commands=%#v err=%v", pending, err) + } + if pending[0].CaptureRevision != capture.Revision || pending[0].PaneID != "wA:p1" { + t.Fatalf("queued at revision %d pane %q, worker published %d wA:p1", pending[0].CaptureRevision, pending[0].PaneID, capture.Revision) + } + var requested *domain.Event + for _, e := range s.Events(0) { + if e.Type == "ApprovalRequested" && e.TaskID == id { + x := e + requested = &x + } + if e.Type == "ApprovalGranted" { + t.Fatal("queueing a command must not claim an outcome the worker has not reported") + } + } + if requested == nil { + t.Fatal("federated approval left no audit trail") + } + var p struct { + SubjectRef string `json:"subject_ref"` + Decision string `json:"decision"` + } + if err := json.Unmarshal(requested.Payload, &p); err != nil { + t.Fatal(err) + } + if p.SubjectRef != pending[0].ID || p.Decision != "grant_approval" { + t.Fatalf("audit event does not identify the command: %s", requested.Payload) + } +} + +// TestCoordinatorCaptureRevisionTracksContentNotTime guards B20: the local +// revision used to be UnixNano, so it changed on every read and told a +// caller nothing about whether the pane had changed. +func TestCoordinatorCaptureRevisionTracksContentNotTime(t *testing.T) { + a := captureRevision("claude", "Permission required\n$ ls") + if a != captureRevision("claude", "Permission required\n$ ls") { + t.Fatal("revision changed without the pane text changing") + } + if a == captureRevision("claude", "Permission required\n$ rm -rf /") { + t.Fatal("revision did not change when the pane text changed") + } + if a == 0 { + t.Fatal("revision 0 means \"no revision\" to federation.Queue") + } +}