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")

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
 		}

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 {

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

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

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

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

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,

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