From 0ead6d2d022e2b45e6f25291dcf34852766f5cbb Mon Sep 17 00:00:00 2001 From: kami Date: Wed, 26 Aug 2026 23:36:10 +0400 Subject: [PATCH] Close F7, F5 and F8 before resuming burn-in F7, security. An unset surface token makes the middleware skip its check, so a full-control surface with no credential is an open control plane rather than a closed one. With ORCHESTRA_TUI_TOKEN unset, any LAN caller could lease, release, complete or block any task by declaring one header, which is how this session's manual leases were issued. authz.RequireCredentials now refuses startup instead of logging. Web is exempt: Sessions makes its login mandatory. F5, lifecycle. router.go's silent `continue` was the first bug, not the predicate behind it. Every eligibility gate now records a router.Rejection with task, herdr and reason, exposed at GET /v1/router/health, reset per pass. No gate was weakened: a direct Store.Lease succeeding proves the lease path, not that eligibility should have selected that worker. F8, correctness. Reconcile iterated every configured source for every task, so a task's external id was looked up in whatever repository each source pointed at. Once two repositories share an issue number, an unrelated human comment becomes an authoritative decision for the wrong task. Reconciliation is now bound to task.Source, the provider:project identity the ingest stamped, and a source that cannot prove it owns the task is skipped. A task with no matching source reconciles to nothing and still launches, because nothing to import is not a failure to read. The integration fixture ingested from "jsonl" while reconciling from "gitea", which is exactly the shape F8 makes impossible; it now ingests from the source it reconciles. Co-Authored-By: Claude Opus 5 --- cmd/orchestra/main.go | 16 ++++++ internal/authz/authz.go | 16 ++++++ internal/authz/authz_test.go | 19 +++++++ internal/human/reconcile.go | 32 +++++++---- internal/human/reconcile_test.go | 48 ++++++++++++++++ internal/integration/end_to_end_test.go | 5 +- internal/router/router.go | 73 ++++++++++++++++++++++-- internal/router/router_test.go | 74 +++++++++++++++++++++++++ 8 files changed, 265 insertions(+), 18 deletions(-) diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go index 155577c..5828297 100644 --- a/cmd/orchestra/main.go +++ b/cmd/orchestra/main.go @@ -1089,6 +1089,16 @@ func main() { } json.NewEncoder(w).Encode(out) }) + mux.HandleFunc("/v1/router/health", func(w http.ResponseWriter, r *http.Request) { + // Why the last scheduling pass placed nothing. A silent eligibility + // gate is indistinguishable from an empty queue, which cost a burn-in + // run to diagnose by hand. + if rt == nil { + http.Error(w, "router not configured", http.StatusServiceUnavailable) + return + } + json.NewEncoder(w).Encode(map[string]any{"rejections": rt.Rejections()}) + }) mux.HandleFunc("/v1/orchestrator/health", func(w http.ResponseWriter, r *http.Request) { if coordinator == nil { http.Error(w, "orchestrator unavailable", http.StatusServiceUnavailable) @@ -1682,6 +1692,12 @@ func main() { // 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"), } + // A full-control surface with no credential is an open control plane, not a + // disabled one, because an unset token makes the middleware skip its check. + // Refuse to start rather than log it and serve. + if err := authz.RequireCredentials(tokens); err != nil { + log.Fatalf("refusing to serve: %v", err) + } log.Fatal(http.ListenAndServe(":"+port, authz.HTTPWithSessions(tokens, sessions, mux))) } diff --git a/internal/authz/authz.go b/internal/authz/authz.go index b16b2dd..a7b6f3d 100644 --- a/internal/authz/authz.go +++ b/internal/authz/authz.go @@ -72,6 +72,22 @@ func (s Surface) RequiresApproval(typ string) bool { return CapabilityFor(s) == GatedWrite && typ != "ApprovalRequested" } +// RequireCredentials refuses to serve a full-control surface that has no token. +// An unset token means the middleware performs no check for that surface, so an +// unconfigured FullControl surface is an unauthenticated control plane, not a +// closed one. Found live during burn-in: with ORCHESTRA_TUI_TOKEN unset, any +// LAN caller could lease, release, complete or block any task by declaring one +// header. Web is exempt because Sessions makes its login mandatory and it +// carries its own credentials. +func RequireCredentials(tokens map[Surface]string) error { + for _, s := range []Surface{TUI} { + if CapabilityFor(s) == FullControl && strings.TrimSpace(tokens[s]) == "" { + return fmt.Errorf("surface %q is full control and has no token: set its credential or leave the surface unused", s) + } + } + return nil +} + func AuthorizeEvent(s Surface, typ string) error { if !s.CanEmit(typ) { return fmt.Errorf("surface %q cannot emit %s", s, typ) diff --git a/internal/authz/authz_test.go b/internal/authz/authz_test.go index ccebd34..e384cfd 100644 --- a/internal/authz/authz_test.go +++ b/internal/authz/authz_test.go @@ -278,3 +278,22 @@ func TestHarnessTurnBypassesSurfaceGate(t *testing.T) { t.Fatalf("harness turn = %d, want 204", w.Code) } } + +// An unset surface token means no check, so a full-control surface without a +// credential is an open control plane. Startup must refuse, not warn. +func TestRequireCredentialsRefusesUncredentialedFullControl(t *testing.T) { + if err := RequireCredentials(map[Surface]string{}); err == nil { + t.Fatal("an uncredentialed TUI surface was accepted") + } + if err := RequireCredentials(map[Surface]string{TUI: " "}); err == nil { + t.Fatal("a blank TUI token was accepted") + } + if err := RequireCredentials(map[Surface]string{TUI: "tui-secret"}); err != nil { + t.Fatalf("a credentialed deployment was refused: %v", err) + } + // Gated and notify-only surfaces are bounded by capability, so an unset + // token there is a deployment choice rather than an open control plane. + if err := RequireCredentials(map[Surface]string{TUI: "tui-secret", MCP: "", Agent: ""}); err != nil { + t.Fatalf("gated surfaces must not block startup: %v", err) + } +} diff --git a/internal/human/reconcile.go b/internal/human/reconcile.go index 594bb46..cd051d7 100644 --- a/internal/human/reconcile.go +++ b/internal/human/reconcile.go @@ -9,7 +9,6 @@ import ( "encoding/json" "errors" "fmt" - "sort" "strings" "time" @@ -72,18 +71,27 @@ func (r *Reconciler) Reconcile(ctx context.Context, taskID string) error { ctx, cancel = context.WithTimeout(ctx, r.Timeout) defer cancel() } - // Deterministic provider order, so two runs over the same pending inputs - // produce the same log. - providers := make([]string, 0, len(r.Sources)) - for name := range r.Sources { - providers = append(providers, name) + // Only the source this task came from may reconcile it. A source is + // identified by the same string the ingest stamped on the task + // (provider:project, e.g. "gitea:test-e2e"), which binds provider, + // instance and repository together. + // + // Iterating every configured source was wrong and not merely noisy: a + // task's external id was looked up in whatever repository each source + // happened to point at, so once two repositories used the same issue + // number, an unrelated human comment became an authoritative decision for + // the wrong task. Found during burn-in with three correx tasks being + // reconciled against kami/test-e2e. + // + // A source that cannot prove it owns the task is skipped, not guessed at. + // Nothing to import is not the same as a failure to read, so a task with no + // matching source reconciles to nothing and the launch proceeds. + src, ok := r.Sources[task.Source] + if !ok { + return nil } - sort.Strings(providers) - - for _, name := range providers { - if err := r.reconcileSource(ctx, task, name, r.Sources[name]); err != nil { - return fmt.Errorf("%s: %w", name, err) - } + if err := r.reconcileSource(ctx, task, task.Source, src); err != nil { + return fmt.Errorf("%s: %w", task.Source, err) } return nil } diff --git a/internal/human/reconcile_test.go b/internal/human/reconcile_test.go index 0615d29..17bf8af 100644 --- a/internal/human/reconcile_test.go +++ b/internal/human/reconcile_test.go @@ -2,6 +2,7 @@ package human import ( "context" + "encoding/json" "errors" "os" "path/filepath" @@ -240,3 +241,50 @@ func TestInputWithoutExternalIDRejected(t *testing.T) { t.Fatalf("want ErrInvalid, got %v", err) } } + +// Only the source a task came from may reconcile it. Iterating every configured +// source looked up a task's external id in whatever repository each source +// pointed at, so a colliding issue number would turn an unrelated human comment +// into an authoritative decision for the wrong task. +func TestReconcileUsesOnlyTheTaskOwnSource(t *testing.T) { + s, err := store.Open(t.TempDir()) + if err != nil { + t.Fatal(err) + } + b, _ := json.Marshal(map[string]any{"source": "gitea:correx", "external_id": "17", "project": "correx"}) + if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskCreated", TaskID: "t1", Version: 1, Payload: b, Surface: string(authz.System)}); err != nil { + t.Fatal(err) + } + owner := &fakeSource{inputs: []Input{{Provider: "gitea:correx", ExternalID: "c1", Author: "kami", Body: "use b"}}, next: "c1"} + other := &fakeSource{inputs: []Input{{Provider: "gitea:test-e2e", ExternalID: "x9", Author: "kami", Body: "delete everything"}}, next: "x9"} + r := &Reconciler{Store: s, Sources: map[string]Source{"gitea:correx": owner, "gitea:test-e2e": other}} + if err := r.Reconcile(context.Background(), "t1"); err != nil { + t.Fatal(err) + } + if other.calls != 0 { + t.Fatalf("a foreign source was asked about this task %d times", other.calls) + } + if owner.calls != 1 { + t.Fatalf("the owning source was called %d times", owner.calls) + } + intent, err := s.EffectiveIntent("t1") + if err != nil { + t.Fatal(err) + } + if len(intent.Decisions) != 1 || intent.Decisions[0].Value != "use b" { + t.Fatalf("standing set = %+v", intent.Decisions) + } + + // A task whose source is not configured reconciles to nothing. Nothing to + // import is not a failure to read, so the launch must not be refused. + b2, _ := json.Marshal(map[string]any{"source": "jsonl", "external_id": "5", "project": "correx"}) + if err := s.Append(domain.Event{ID: domain.NewID(), Type: "TaskCreated", TaskID: "t2", Version: 1, Payload: b2, Surface: string(authz.System)}); err != nil { + t.Fatal(err) + } + if err := r.Reconcile(context.Background(), "t2"); err != nil { + t.Fatalf("unconfigured source refused the launch: %v", err) + } + if other.calls != 0 || owner.calls != 1 { + t.Fatalf("an unowned task reached a source: owner %d, other %d", owner.calls, other.calls) + } +} diff --git a/internal/integration/end_to_end_test.go b/internal/integration/end_to_end_test.go index 223d9c1..e31a182 100644 --- a/internal/integration/end_to_end_test.go +++ b/internal/integration/end_to_end_test.go @@ -86,7 +86,10 @@ func setup(t *testing.T) (*store.Store, registry.Registry, string) { func ingest(t *testing.T, s *store.Store, external string) domain.Task { t.Helper() - _, err := (provider.JSONL{}).Ingest(strings.NewReader(`{"source":"jsonl","external_id":"`+external+`","project":"p","capability":["go"],"title":"demo"}`+"\n"), s) + // The source name is the one a human source is keyed by, because only the + // source a task came from may reconcile it. A fixture that ingests from one + // source and reconciles from another is testing a shape that cannot occur. + _, err := (provider.JSONL{}).Ingest(strings.NewReader(`{"source":"gitea","external_id":"`+external+`","project":"p","capability":["go"],"title":"demo"}`+"\n"), s) if err != nil { t.Fatal(err) } diff --git a/internal/router/router.go b/internal/router/router.go index c401488..def8990 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -102,9 +102,48 @@ type Router struct { // A scheduling pass always has a coherent health snapshot; this cache also // prevents bursts of TaskCreated events from repeatedly dialing the same // herdr between passes. - HealthTTL time.Duration - healthMu sync.Mutex - health map[string]healthProbe + HealthTTL time.Duration + healthMu sync.Mutex + health map[string]healthProbe + rejectMu sync.Mutex + rejections []Rejection +} + +// Rejection is why one scheduling pass did not give one task to one herdr. +// Every eligibility gate records one, because a silent `continue` is +// indistinguishable from "nothing was queued": during burn-in a task sat +// queued while every gate checked out by hand, and the router said nothing. +type Rejection struct { + TaskID string `json:"task_id"` + HerdrID string `json:"herdr_id,omitempty"` + Reason string `json:"reason"` +} + +// maxRejections bounds the recorded set so a large queue cannot grow it without +// limit. The newest pass always fits, because the list is reset per pass. +const maxRejections = 200 + +func (r *Router) reject(taskID, herdrID, reason string) { + r.rejectMu.Lock() + defer r.rejectMu.Unlock() + if len(r.rejections) >= maxRejections { + return + } + r.rejections = append(r.rejections, Rejection{TaskID: taskID, HerdrID: herdrID, Reason: reason}) +} + +// Rejections returns why the most recent scheduling pass assigned nothing to +// the tasks it could not place. +func (r *Router) Rejections() []Rejection { + r.rejectMu.Lock() + defer r.rejectMu.Unlock() + return append([]Rejection(nil), r.rejections...) +} + +func (r *Router) resetRejections() { + r.rejectMu.Lock() + r.rejections = nil + r.rejectMu.Unlock() } type healthProbe struct { @@ -139,6 +178,7 @@ func (r *Router) AssignPending() ([]domain.Event, error) { return nil, errors.New("router: store required") } now := r.Now() + r.resetRejections() snapshot := r.Store.SchedulingSnapshot() var queued []domain.Task for _, t := range snapshot.Tasks { @@ -169,10 +209,16 @@ func (r *Router) AssignPending() ([]domain.Event, error) { var err error cs, err = r.Registry.CandidatesWithHealth(t.Project, health) if err != nil { + r.reject(t.ID, "", "no candidate herdr for project "+t.Project+": "+err.Error()) continue } candidates[t.Project] = cs } + if len(cs) == 0 { + r.reject(t.ID, "", "no candidate herdr for project "+t.Project+" (affinity, capability or reachability)") + continue + } + placed := false for _, h := range cs { projectKey := h.ID + "\x00" + t.Project projectOK, checked := projectSupport[projectKey], projectSupportKnown[projectKey] @@ -183,7 +229,12 @@ func (r *Router) AssignPending() ([]domain.Event, error) { } projectSupport[projectKey], projectSupportKnown[projectKey] = projectOK, true } - if !projectOK || !matches(t.Capability, h.Capabilities) { + if !projectOK { + r.reject(t.ID, h.ID, "worker has not declared project "+t.Project) + continue + } + if !matches(t.Capability, h.Capabilities) { + r.reject(t.ID, h.ID, "capability mismatch") continue } available, checked := availability[h.ID], availabilityKnown[h.ID] @@ -191,13 +242,22 @@ func (r *Router) AssignPending() ([]domain.Event, error) { available = r.Availability.Available(h) availability[h.ID], availabilityKnown[h.ID] = available, true } - if !available || occupiedCount(activeLeases[h.ID], h.Concurrency) { + if !available { + r.reject(t.ID, h.ID, "worker unavailable: stale heartbeat or unhealthy local backend") + continue + } + if occupiedCount(activeLeases[h.ID], h.Concurrency) { + r.reject(t.ID, h.ID, "no free concurrency") continue } e, err := r.Store.Lease(t.ID, h.ID, 30*time.Minute) if err != nil { + // Includes a pre-lease reconciliation refusal, which is the + // one gate that fails closed on purpose. + r.reject(t.ID, h.ID, "lease refused: "+err.Error()) continue } + placed = true out = append(out, e) activeLeases[h.ID]++ if r.OnLease != nil { @@ -207,6 +267,9 @@ func (r *Router) AssignPending() ([]domain.Event, error) { } break } + if !placed { + r.reject(t.ID, "", "no candidate accepted the task") + } } return out, nil } diff --git a/internal/router/router_test.go b/internal/router/router_test.go index ec4c328..8802ccd 100644 --- a/internal/router/router_test.go +++ b/internal/router/router_test.go @@ -2,6 +2,9 @@ package router import ( "encoding/json" + "errors" + "strings" + "orchestra/internal/authz" "orchestra/internal/domain" "orchestra/internal/registry" @@ -299,3 +302,74 @@ func TestQuotaWindowsAreIndependent(t *testing.T) { t.Fatal("bounded harness without a native usage receipt must fail closed") } } + +// Every eligibility gate must say why. A silent `continue` is +// indistinguishable from an empty queue: during burn-in a task sat queued +// while every gate checked out by hand, and the router reported nothing. +func TestAssignPendingRecordsWhyItPlacedNothing(t *testing.T) { + s, err := store.Open(t.TempDir()) + if err != nil { + t.Fatal(err) + } + r, err := registry.New(registry.Config{ + Projects: []registry.Project{{ID: "p", MachineAffinity: []string{"m"}}}, + Machines: []registry.Machine{{ID: "m", Address: "unused"}}, + Herdrs: []registry.Herdr{{ID: "h", MachineID: "m", Concurrency: 1}}, + }) + if err != nil { + t.Fatal(err) + } + b, _ := json.Marshal(map[string]any{"source": "test", "external_id": "why", "project": "p"}) + if err := s.Append(domain.Event{ID: "create", TaskID: "t", Type: "TaskCreated", Version: 1, Payload: b, Surface: string(authz.System)}); err != nil { + t.Fatal(err) + } + + // The live shape that went undiagnosed: a worker that has not declared the + // project. Everything else about it looks healthy. + rt := Router{Store: s, Registry: r, Reachability: reachable{}, Availability: projectAvailability{projects: map[string]bool{}}} + if got, err := rt.AssignPending(); err != nil || len(got) != 0 { + t.Fatalf("lease = %#v, %v", got, err) + } + reasons := rt.Rejections() + if len(reasons) == 0 { + t.Fatal("the router placed nothing and said nothing") + } + var named bool + for _, rej := range reasons { + if rej.TaskID != "t" { + t.Fatalf("rejection for the wrong task: %+v", rej) + } + if rej.HerdrID == "h" && strings.Contains(rej.Reason, "has not declared project") { + named = true + } + } + if !named { + t.Fatalf("no rejection names the failing gate: %+v", reasons) + } + + // A pre-lease refusal is the gate that fails closed on purpose, and it must + // also be visible rather than looking like "no candidates". + rt.Availability = projectAvailability{projects: map[string]bool{"h/p": true}} + s.PreLease = func(string) error { return errors.New("gitea unreachable") } + if got, err := rt.AssignPending(); err != nil || len(got) != 0 { + t.Fatalf("lease despite pre-lease refusal = %#v, %v", got, err) + } + found := false + for _, rej := range rt.Rejections() { + if strings.Contains(rej.Reason, "lease refused") && strings.Contains(rej.Reason, "gitea unreachable") { + found = true + } + } + if !found { + t.Fatalf("pre-lease refusal not reported: %+v", rt.Rejections()) + } + + // A successful pass leaves nothing behind to misread. + s.PreLease = nil + if got, err := rt.AssignPending(); err != nil || len(got) != 1 { + t.Fatalf("lease = %#v, %v", got, err) + } + if reasons := rt.Rejections(); len(reasons) != 0 { + t.Fatalf("stale rejections after a successful pass: %+v", reasons) + } +}