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) + } +}