diff --git a/cmd/orchestra-worker/main.go b/cmd/orchestra-worker/main.go index faf8d39..113ee4a 100644 --- a/cmd/orchestra-worker/main.go +++ b/cmd/orchestra-worker/main.go @@ -103,8 +103,13 @@ func (w *worker) recordError(err error) { w.observations = append(w.observations, federation.Observation{Message: msg, Count: 1, First: now, Last: now}) } +// workerIncarnation identifies this process. A restarted worker cannot +// continue the previous process's failures, and nothing else on the wire says +// a restart happened: build revision and worker id both survive it. +var workerIncarnation = domain.NewID() + func (w *worker) health(ctx context.Context) federation.WorkerHealth { - h := federation.WorkerHealth{HerdrStatus: "unknown"} + h := federation.WorkerHealth{HerdrStatus: "unknown", Incarnation: workerIncarnation} if backend := w.executionBackend(); backend != nil { h.Backend = backend.Kind() } @@ -1884,7 +1889,7 @@ func main() { window: window, // Capacity stays one per identity because a herdr's declared // concurrency is one. Serving N harnesses gives the process N slots. - registration: federation.Worker{ID: spec.ID, Address: spec.Address, Capacity: 1, SupportedProjects: supported, Build: buildinfo.Current()}, + registration: federation.Worker{ID: spec.ID, Address: spec.Address, Capacity: 1, SupportedProjects: supported, Build: buildinfo.Current(), Incarnation: workerIncarnation}, } if w.registration.Address == "" { w.registration.Address = os.Getenv("ORCHESTRA_WORKER_ADDRESS") diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go index 1c98773..9942fc9 100644 --- a/cmd/orchestra/main.go +++ b/cmd/orchestra/main.go @@ -228,6 +228,9 @@ func main() { // Pull-request readers by source name, for reflecting submitted work. pullRequests := map[string]human.PullRequestSource{} localMachine := os.Getenv("ORCHESTRA_MACHINE_ID") + // One tracker for the process: it holds the per-incident accumulation that + // makes an evicted and recreated ring entry add up instead of restarting. + observations := &operations.ObservationTracker{Store: s} workers := &federation.Registry{AdmitToken: os.Getenv("ORCHESTRA_FEDERATION_ADMIT_TOKEN"), StatePath: filepath.Join(dir, "federation-state.json")} if err := workers.Load(); err != nil { log.Fatalf("load federation state: %v", err) @@ -445,6 +448,49 @@ func main() { } json.NewEncoder(w).Encode(s.Events(n)) }) + mux.HandleFunc("/v1/interventions", func(w http.ResponseWriter, r *http.Request) { + // The one command an operator has for saying "I fixed this by hand". + // Everything Orchestra does to itself is already an event; a manual + // repair is the only kind of recovery that leaves no trace unless the + // person who made it says so. + if r.Method == http.MethodGet { + out := make([]domain.Event, 0) + for _, e := range s.Events(0) { + if e.Type == domain.EventOperatorInterventionRecorded { + out = append(out, e) + } + } + json.NewEncoder(w).Encode(out) + return + } + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + if err := authz.AuthorizeEvent(surface(r), domain.EventOperatorInterventionRecorded); err != nil { + http.Error(w, err.Error(), http.StatusForbidden) + return + } + var in domain.OperatorIntervention + if json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<16)).Decode(&in) != nil { + http.Error(w, "invalid json", http.StatusBadRequest) + return + } + e, err := operations.RecordIntervention(s, surface(r), in) + switch { + case errors.Is(err, domain.ErrNotFound): + http.Error(w, err.Error(), http.StatusNotFound) + return + case errors.Is(err, domain.ErrInvalid): + http.Error(w, err.Error(), http.StatusBadRequest) + return + case err != nil: + http.Error(w, err.Error(), http.StatusConflict) + return + } + w.WriteHeader(http.StatusCreated) + json.NewEncoder(w).Encode(e) + }) mux.HandleFunc("/v1/debt", func(w http.ResponseWriter, r *http.Request) { // Read-only, and deliberately so. This projection is evidence about // history, not a new kind of truth: nothing here writes an event, @@ -1391,6 +1437,29 @@ func main() { http.Error(w, err.Error(), 404) return } + // The ring carries messages and nothing else, so the coordinator + // attributes them: the worker's active task, and that task's + // current lease epoch, are what bind an incident to the work it + // happened during. A failed fold is logged rather than failing the + // heartbeat, because losing evidence about a worker is not a reason + // to also stop believing the worker is alive. + report := operations.WorkerReport{ + WorkerID: parts[3], Incarnation: health.Incarnation, + TaskID: health.ActiveTask, At: time.Now().UTC(), + } + if report.TaskID != "" { + if t, ok := s.Task(report.TaskID); ok && t.Lease != nil { + report.LeaseEpoch = t.Lease.Epoch + } + } + for _, o := range health.Observations { + report.Observations = append(report.Observations, domain.WorkerObservation{ + Message: o.Message, Count: o.Count, First: o.First, Last: o.Last, + }) + } + if _, err := observations.Ingest(report); err != nil { + log.Printf("observation incidents for %s: %v", parts[3], err) + } w.WriteHeader(http.StatusNoContent) return } diff --git a/internal/domain/debt.go b/internal/domain/debt.go index 8a6ffca..63f4881 100644 --- a/internal/domain/debt.go +++ b/internal/domain/debt.go @@ -113,6 +113,12 @@ type DebtObservation struct { Detail string `json:"detail,omitempty"` Paths []string `json:"paths,omitempty"` At time.Time `json:"at"` + // Repeats is how many times this one incident recurred. It is intensity, + // never recurrence: one worker stuck in a five-second retry loop produced + // 301 repeats of a single failure, and counting those as 301 pieces of + // evidence would make one broken worker look like chronic, system-wide + // debt. Recurrence is the number of independent observations. + Repeats int `json:"repeats,omitempty"` } func (o DebtObservation) Validate() error { diff --git a/internal/domain/domain.go b/internal/domain/domain.go index 9b328d7..6ef3bf4 100644 --- a/internal/domain/domain.go +++ b/internal/domain/domain.go @@ -305,6 +305,8 @@ func EventWithoutTask(typ string) bool { switch typ { case "QuotaReported", "StandupAdvisory", "ApprovalGranted", "ApprovalDenied", EventObservationIncidentOpened, EventObservationIncidentClosed: + // An intervention is deliberately absent: it names a task when it + // repaired one, and that task must exist. return true } return false @@ -317,7 +319,7 @@ func ValidateEvent(e Event) error { if e.SchemaVersion >= 2 && strings.TrimSpace(e.Surface) == "" { return fmt.Errorf("%w: surface required", ErrInvalid) } - allowed := map[string]bool{"TaskCreated": true, "TaskLeased": true, "TaskLeaseRenewed": true, "TaskReleased": true, "TaskLaunchAcknowledged": true, "TaskPickupValidated": true, "TaskCompleted": true, "TaskFailed": true, "TaskBlocked": true, "TaskNeedsAttention": true, "ApprovalRequested": true, "ApprovalGranted": true, "ApprovalDenied": true, "TaskAmended": true, "TaskCorrected": true, "QuotaReported": true, "StandupAdvisory": true, EventHumanDecisionRecorded: true, EventHumanDecisionSuperseded: true, EventWorkPhaseChanged: true, EventDeferredFindingRecorded: true, EventReviewRecorded: true, EventTaskSubmitted: true, EventTaskChangesRequested: true, EventPlanPhaseVerified: true, EventPlanMismatchRecorded: true, EventObservationIncidentOpened: true, EventObservationIncidentClosed: true} + allowed := map[string]bool{"TaskCreated": true, "TaskLeased": true, "TaskLeaseRenewed": true, "TaskReleased": true, "TaskLaunchAcknowledged": true, "TaskPickupValidated": true, "TaskCompleted": true, "TaskFailed": true, "TaskBlocked": true, "TaskNeedsAttention": true, "ApprovalRequested": true, "ApprovalGranted": true, "ApprovalDenied": true, "TaskAmended": true, "TaskCorrected": true, "QuotaReported": true, "StandupAdvisory": true, EventHumanDecisionRecorded: true, EventHumanDecisionSuperseded: true, EventWorkPhaseChanged: true, EventDeferredFindingRecorded: true, EventReviewRecorded: true, EventTaskSubmitted: true, EventTaskChangesRequested: true, EventPlanPhaseVerified: true, EventPlanMismatchRecorded: true, EventObservationIncidentOpened: true, EventObservationIncidentClosed: true, EventOperatorInterventionRecorded: true} if !allowed[e.Type] { return fmt.Errorf("%w: unknown type %q", ErrInvalid, e.Type) } @@ -603,6 +605,8 @@ func ValidatePayload(typ string, p map[string]any) error { return ValidateObservationIncidentOpened(p) case EventObservationIncidentClosed: return ValidateObservationIncidentClosed(p) + case EventOperatorInterventionRecorded: + return ValidateOperatorInterventionRecorded(p) case EventReviewRecorded: if err := requiredHash(p, "artifact_ref"); err != nil { return err diff --git a/internal/domain/intervention.go b/internal/domain/intervention.go new file mode 100644 index 0000000..ef758b8 --- /dev/null +++ b/internal/domain/intervention.go @@ -0,0 +1,81 @@ +package domain + +import ( + "fmt" + "strings" + "time" +) + +// EventOperatorInterventionRecorded is a human saying what they repaired by +// hand. Nothing infers it: a manual fix happens outside Orchestra by +// definition, so the only honest way to have the evidence is for the operator +// to state it. Without this the debt ledger reported manual recovery as a +// permanent gap, and every repair that kept the system running was invisible +// to the record of how much the system costs to run. +const EventOperatorInterventionRecorded = "OperatorInterventionRecorded" + +// InterventionKind is what the operator did. The list is closed so the ledger +// can group repairs; an unrecognised kind is refused rather than guessed at. +type InterventionKind string + +const ( + InterventionWorkerRestart InterventionKind = "worker_restart" + InterventionTransactionClean InterventionKind = "transaction_cleanup" + InterventionForcedRelease InterventionKind = "forced_release" + InterventionStateRepair InterventionKind = "state_repair" + InterventionManualRequeue InterventionKind = "manual_requeue" + InterventionPhaseRecovery InterventionKind = "manual_phase_recovery" +) + +func (k InterventionKind) Valid() bool { + switch k { + case InterventionWorkerRestart, InterventionTransactionClean, InterventionForcedRelease, + InterventionStateRepair, InterventionManualRequeue, InterventionPhaseRecovery: + return true + } + return false +} + +// OperatorIntervention is one recorded manual repair. +type OperatorIntervention struct { + TaskID string `json:"task_id,omitempty"` + WorkerID string `json:"worker_id,omitempty"` + Kind InterventionKind `json:"kind"` + // Reason is the operator's own account of why it was needed. It is the + // part a later reader cannot reconstruct from anything else. + Reason string `json:"reason"` + // RelatedEventID and RelatedTransactionID point at what was repaired, so a + // reader can find the failure this answered rather than infer it. + RelatedEventID string `json:"related_event_id,omitempty"` + RelatedTransactionID string `json:"related_transaction_id,omitempty"` + Components []string `json:"components,omitempty"` + At time.Time `json:"at,omitempty"` +} + +const maxInterventionReason = 1000 + +func (i OperatorIntervention) Validate() error { + if !i.Kind.Valid() { + return fmt.Errorf("%w: %q is not an intervention kind", ErrInvalid, i.Kind) + } + if strings.TrimSpace(i.Reason) == "" { + return fmt.Errorf("%w: an intervention states why it was needed", ErrInvalid) + } + if len(i.Reason) > maxInterventionReason { + return fmt.Errorf("%w: reason exceeds %d characters", ErrInvalid, maxInterventionReason) + } + if i.TaskID == "" && i.WorkerID == "" { + return fmt.Errorf("%w: an intervention names the task or the worker it repaired", ErrInvalid) + } + return nil +} + +func ValidateOperatorInterventionRecorded(p map[string]any) error { + kind, _ := p["kind"].(string) + reason, _ := p["reason"].(string) + task, _ := p["task_id"].(string) + worker, _ := p["worker_id"].(string) + return OperatorIntervention{ + Kind: InterventionKind(kind), Reason: reason, TaskID: task, WorkerID: worker, + }.Validate() +} diff --git a/internal/federation/federation.go b/internal/federation/federation.go index 2c505e9..77ed9e0 100644 --- a/internal/federation/federation.go +++ b/internal/federation/federation.go @@ -21,10 +21,14 @@ type Worker struct { Capacity int `json:"capacity"` SupportedProjects []string `json:"supported_projects"` Build buildinfo.Info `json:"build"` - LastSeen time.Time `json:"last_seen"` - Online bool `json:"online"` - Health WorkerHealth `json:"health"` - Token string `json:"-"` + // Incarnation identifies one worker process. Nothing else on the wire + // distinguishes a restarted worker from a running one, and an observation + // incident cannot outlive the process that reported it. + Incarnation string `json:"incarnation,omitempty"` + LastSeen time.Time `json:"last_seen"` + Online bool `json:"online"` + Health WorkerHealth `json:"health"` + Token string `json:"-"` } // WorkerHealth is reported by the worker that owns the local execution backend. @@ -42,6 +46,9 @@ type WorkerHealth struct { // Observations is the bounded set of distinct failures behind LastError, // which keeps its wire name and still reports only the newest. Observations []Observation `json:"observations,omitempty"` + // Incarnation repeats the worker's process identity on every heartbeat, so + // the coordinator sees a restart even if it missed the registration. + Incarnation string `json:"incarnation,omitempty"` } // Observation is one distinct worker failure with its repeat count. A single diff --git a/internal/operations/observations.go b/internal/operations/observations.go index ab31475..a3f6d00 100644 --- a/internal/operations/observations.go +++ b/internal/operations/observations.go @@ -199,3 +199,42 @@ func firstOr(t, fallback time.Time) time.Time { } return t } + +// RecordIntervention writes down a repair the operator made by hand. It is +// deliberately an explicit act: Orchestra cannot see a worker someone +// restarted or a transaction someone deleted, and inferring "an operator +// probably intervened" from a gap in the log would put guesses into the +// evidence the ledger is built from. +func RecordIntervention(s *store.Store, surface authz.Surface, in domain.OperatorIntervention) (domain.Event, error) { + if err := in.Validate(); err != nil { + return domain.Event{}, err + } + if in.At.IsZero() { + in.At = time.Now().UTC() + } + taskID := in.TaskID + version := 0 + if taskID != "" { + t, ok := s.Task(taskID) + if !ok { + return domain.Event{}, domain.ErrNotFound + } + version = t.Version + 1 + } else { + // A repair with no task is still about this deployment, so it lands on + // the same aggregate the other worker-scoped facts use. + taskID = "system" + } + b, err := json.Marshal(in) + if err != nil { + return domain.Event{}, err + } + e := domain.Event{ + ID: domain.NewID(), Type: domain.EventOperatorInterventionRecorded, + TaskID: taskID, Version: version, At: in.At, Payload: b, Surface: string(surface), + } + if err := s.Append(e); err != nil { + return domain.Event{}, err + } + return e, nil +} diff --git a/internal/store/debt_projection.go b/internal/store/debt_projection.go index ba243f5..1a5b076 100644 --- a/internal/store/debt_projection.go +++ b/internal/store/debt_projection.go @@ -102,6 +102,36 @@ func projectDebt(events []domain.Event, readArtifact func(string) ([]byte, error o.Detail = str(p["last_error"]) add(class, domain.DebtSignature(class, failure, harness, "lease"), o, "tasks end in "+failure, review.Important) + case domain.EventObservationIncidentClosed: + // One incident, whatever it repeated. The debt class comes from + // the signature's shape rather than a failure class, because a + // worker observation is a symptom the worker described, not a + // lifecycle outcome Orchestra decided. + var inc domain.ObservationIncident + if json.Unmarshal(e.Payload, &inc) != nil || inc.Signature == "" { + continue + } + o := base + o.Kind = domain.ObservationWorkerFailure + o.TaskID = inc.TaskID + o.Detail = inc.Detail + o.Repeats = inc.RepeatCount + add(domain.DebtOperational, + domain.DebtSignature(domain.DebtOperational, inc.Signature, inc.WorkerID, "worker"), + o, "workers report "+inc.Signature, review.Important) + case domain.EventOperatorInterventionRecorded: + var in domain.OperatorIntervention + if json.Unmarshal(e.Payload, &in) != nil || !in.Kind.Valid() { + continue + } + o := base + o.Kind = domain.ObservationManualIntervention + o.TaskID = in.TaskID + o.Detail = in.Reason + o.Paths = in.Components + add(domain.DebtOperational, + domain.DebtSignature(domain.DebtOperational, string(in.Kind), in.WorkerID, "manual"), + o, "an operator repairs this by hand ("+string(in.Kind)+")", review.Important) case domain.EventPlanMismatchRecorded: o := base o.Kind = domain.ObservationPlanMismatch @@ -167,16 +197,15 @@ func projectDebt(events []domain.Event, readArtifact func(string) ([]byte, error // hole in the system. A kind the log could carry and does not is a fact about // this history. func debtGaps(seen map[domain.ObservationKind]bool) []domain.EvidenceGap { - gaps := []domain.EvidenceGap{ - {Kind: domain.ObservationManualIntervention, Durable: false, - Reason: "no event type records an operator repair, so every manual recovery is invisible to this ledger"}, - {Kind: domain.ObservationWorkerFailure, Durable: false, - Reason: "worker observations live in worker memory and reach the coordinator only inside WorkerHealth, which is not persisted"}, - } + // Both of these were once permanent holes in the system. They are ordinary + // evidence now, so their absence is a fact about this history rather than + // about Orchestra. + var gaps []domain.EvidenceGap for _, k := range []domain.ObservationKind{ domain.ObservationBlockReason, domain.ObservationFailureClass, domain.ObservationReviewFinding, domain.ObservationPlanMismatch, - domain.ObservationDeferredFinding, + domain.ObservationDeferredFinding, domain.ObservationWorkerFailure, + domain.ObservationManualIntervention, } { if !seen[k] { gaps = append(gaps, domain.EvidenceGap{Kind: k, Durable: true, diff --git a/internal/store/debt_projection_test.go b/internal/store/debt_projection_test.go index b974e2d..9702e99 100644 --- a/internal/store/debt_projection_test.go +++ b/internal/store/debt_projection_test.go @@ -64,24 +64,92 @@ func TestProjectDebtIgnoresOrdinaryLifecycleStops(t *testing.T) { // not recorded anywhere". func TestProjectDebtReportsWhatItCannotSee(t *testing.T) { ledger := ProjectDebt(nil) - var manual, worker bool - for _, g := range ledger.Gaps { - if g.Durable { - continue - } - switch g.Kind { - case domain.ObservationManualIntervention: - manual = true - case domain.ObservationWorkerFailure: - worker = true - } - } - if !manual || !worker { - t.Fatalf("the two known holes must always be reported: %+v", ledger.Gaps) - } for _, g := range ledger.Gaps { if g.Reason == "" { t.Fatalf("gap %q has no reason", g.Kind) } + // Slice B closed the two holes this ledger used to report about + // itself. Every silence is now a fact about one history, never a kind + // of evidence the system cannot record at all. + if !g.Durable { + t.Fatalf("gap %q is reported as unrecordable: %+v", g.Kind, g) + } + } + var worker, manual bool + for _, g := range ledger.Gaps { + switch g.Kind { + case domain.ObservationWorkerFailure: + worker = true + case domain.ObservationManualIntervention: + manual = true + } + } + if !worker || !manual { + t.Fatalf("an empty history should still name both kinds as absent: %+v", ledger.Gaps) + } +} + +// The whole point of incidents. One worker stuck in a retry loop must not +// manufacture recurrence, while its intensity is still on the record. +func TestRecurrenceCountsIncidentsAndKeepsIntensitySeparate(t *testing.T) { + closed := func(id, worker, task, epoch string, repeats int) domain.Event { + b, _ := json.Marshal(domain.ObservationIncident{ + ID: id, WorkerID: worker, TaskID: task, LeaseEpoch: epoch, + Signature: "lease not renewed: agent status idle and pane unchanged", + Detail: "lease " + task + " not renewed: agent status idle and pane unchanged", + RepeatCount: repeats, CloseReason: domain.ObservationCloseEpochChange, + }) + return domain.Event{ID: id, Type: domain.EventObservationIncidentClosed, TaskID: "system", Payload: b} + } + ledger := ProjectDebt([]domain.Event{ + closed("i1", "workpc-claude", "task-a", "e1", 301), + closed("i2", "workpc-claude", "task-b", "e2", 2), + }) + if len(ledger.Items) != 1 { + t.Fatalf("one kind of failure produced %d items", len(ledger.Items)) + } + item := ledger.Items[0] + if len(item.Observations) != 2 { + t.Fatalf("recurrence = %d, want one per incident", len(item.Observations)) + } + intensity := 0 + for _, o := range item.Observations { + if o.Kind != domain.ObservationWorkerFailure { + t.Fatalf("observation kind = %q", o.Kind) + } + intensity += o.Repeats + } + if intensity != 303 { + t.Fatalf("intensity = %d, want 303 carried alongside a recurrence of 2", intensity) + } + tasks := map[string]bool{} + for _, o := range item.Observations { + tasks[o.TaskID] = true + } + if len(tasks) != 2 { + t.Fatalf("the two incidents are not attributed to their tasks: %+v", item.Observations) + } +} + +// A repair the operator made by hand is evidence like any other, once they say +// it happened. +func TestAnOperatorRepairBecomesDebtEvidence(t *testing.T) { + b, _ := json.Marshal(domain.OperatorIntervention{ + WorkerID: "workpc-opencode", Kind: domain.InterventionTransactionClean, + Reason: "deleted a release transaction stuck at prepared so the pane could be reused", + }) + ledger := ProjectDebt([]domain.Event{{ + ID: "i1", Type: domain.EventOperatorInterventionRecorded, TaskID: "system", Payload: b, + }}) + if len(ledger.Items) != 1 || len(ledger.Items[0].Observations) != 1 { + t.Fatalf("the repair produced no debt evidence: %+v", ledger.Items) + } + if got := ledger.Items[0].Observations[0].Kind; got != domain.ObservationManualIntervention { + t.Fatalf("kind = %q", got) + } + for _, g := range ledger.Gaps { + if g.Kind == domain.ObservationManualIntervention { + t.Fatal("manual intervention is still reported as missing from a history that contains one") + } } }