From e0601296e068e92a0ba35b1159ff17fa29362ece Mon Sep 17 00:00:00 2001 From: kami Date: Sun, 30 Aug 2026 06:53:26 +0400 Subject: [PATCH] Record what an operator repaired, and let debt count incidents Slice B, second half. OperatorInterventionRecorded is the one command for saying "I fixed this by hand": a manual repair happens outside Orchestra by definition, so the only honest way to have the evidence is for the person who made it to state it. Inferring "an operator probably intervened" from a gap would put guesses into the record the ledger is built from. The debt projection now consumes both new kinds. A closed incident is one observation carrying its repeat count as intensity, so recurrence stays a count of independent incidents: 301 repeats on one lease and 2 on another is a recurrence of two with an intensity of 303, not a recurrence of 303. Both kinds were previously reported as holes in the system. They are ordinary evidence now, so their absence from a history is a fact about that history, and the gap list says so. The worker also stamps a per-process incarnation on registration and every heartbeat. Nothing else on the wire distinguishes a restarted worker from a running one, and an incident cannot outlive the process that reported it. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01CVbaKucEYBjMqVeUgJUsc1 --- cmd/orchestra-worker/main.go | 9 ++- cmd/orchestra/main.go | 69 ++++++++++++++++++ internal/domain/debt.go | 6 ++ internal/domain/domain.go | 6 +- internal/domain/intervention.go | 81 +++++++++++++++++++++ internal/federation/federation.go | 15 ++-- internal/operations/observations.go | 39 ++++++++++ internal/store/debt_projection.go | 43 +++++++++-- internal/store/debt_projection_test.go | 98 ++++++++++++++++++++++---- 9 files changed, 337 insertions(+), 29 deletions(-) create mode 100644 internal/domain/intervention.go 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") + } } }