diff --git a/internal/domain/domain.go b/internal/domain/domain.go index a406d36..9b328d7 100644 --- a/internal/domain/domain.go +++ b/internal/domain/domain.go @@ -241,7 +241,7 @@ type Task struct { // is answered, because after either it is history rather than a live // instruction. PlanMismatch *PlanMismatch `json:"plan_mismatch,omitempty"` - LastError string `json:"last_error,omitempty"` + LastError string `json:"last_error,omitempty"` } // ReviewRef binds a sealed review artifact to one commit. @@ -293,6 +293,23 @@ func NewID() string { _, _ = rand.Read(b[6:]) return ulidEncoding.EncodeToString(b) } + +// EventWithoutTask reports whether an event records something about the +// system rather than about one task's lifecycle. These carry the "system" +// aggregate id, so no task projection has to exist for them. +// +// One list, because there were three: the replay guard, the append guard and +// the transition check each kept their own copy, and adding an event type to +// two of them left it rejected by the third. +func EventWithoutTask(typ string) bool { + switch typ { + case "QuotaReported", "StandupAdvisory", "ApprovalGranted", "ApprovalDenied", + EventObservationIncidentOpened, EventObservationIncidentClosed: + return true + } + return false +} + func ValidateEvent(e Event) error { if e.SchemaVersion > CurrentEventSchema || e.Type == "" || e.TaskID == "" || len(e.Payload) == 0 || len(e.Payload) > 64*1024 { return ErrInvalid @@ -300,7 +317,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} + 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} if !allowed[e.Type] { return fmt.Errorf("%w: unknown type %q", ErrInvalid, e.Type) } @@ -582,6 +599,10 @@ func ValidatePayload(typ string, p map[string]any) error { return ValidatePlanPhaseVerified(p) case EventPlanMismatchRecorded: return ValidatePlanMismatchRecorded(p) + case EventObservationIncidentOpened: + return ValidateObservationIncidentOpened(p) + case EventObservationIncidentClosed: + return ValidateObservationIncidentClosed(p) case EventReviewRecorded: if err := requiredHash(p, "artifact_ref"); err != nil { return err diff --git a/internal/domain/observation.go b/internal/domain/observation.go new file mode 100644 index 0000000..853e1b4 --- /dev/null +++ b/internal/domain/observation.go @@ -0,0 +1,142 @@ +package domain + +import ( + "fmt" + "regexp" + "strings" + "time" +) + +// A worker's observation ring is bounded, lossy, and local: it holds distinct +// failure messages with repeat counts and nothing else, and it disappears when +// the process does. The debt ledger reported that gap about itself, because no +// event carried any of it. +// +// These two events make it durable as incidents rather than as symptoms. Run +// 11 saw the same 409 refusal 301 times; that is one incident with an +// intensity of 301, not 301 pieces of evidence. Recurrence has to mean "this +// happened on four independent leases", or one stuck loop makes everything +// look chronic. +const ( + EventObservationIncidentOpened = "ObservationIncidentOpened" + EventObservationIncidentClosed = "ObservationIncidentClosed" +) + +// ObservationCloseReason is why Orchestra finalized an incident. None of them +// is "the message stopped appearing in the ring": the ring is a bounded +// history, so absence proves eviction as easily as recovery. +type ObservationCloseReason string + +const ( + // ObservationCloseLeaseEnd and ObservationCloseEpochChange are the natural + // boundaries of a lease-scoped incident. The work it was about is over. + ObservationCloseLeaseEnd ObservationCloseReason = "lease_end" + ObservationCloseEpochChange ObservationCloseReason = "epoch_change" + // ObservationCloseWorkerRestart ends every incident of an incarnation. A + // new process cannot continue the old one's symptom. + ObservationCloseWorkerRestart ObservationCloseReason = "worker_restart" + // ObservationCloseQuietTimeout is the only closer for an observation with + // no lease to bound it, and it fires on last_seen going stale rather than + // on the entry vanishing. + ObservationCloseQuietTimeout ObservationCloseReason = "quiet_timeout" +) + +func (r ObservationCloseReason) Valid() bool { + switch r { + case ObservationCloseLeaseEnd, ObservationCloseEpochChange, ObservationCloseWorkerRestart, ObservationCloseQuietTimeout: + return true + } + return false +} + +// WorkerObservation is one entry of a worker's ring as reported on a +// heartbeat. It is the input to the incident projection, never a stored event. +type WorkerObservation struct { + Message string `json:"message"` + Count int `json:"count"` + First time.Time `json:"first"` + Last time.Time `json:"last"` +} + +// ObservationIncident is one durable incident: a signature seen by one worker, +// on one lease when there is one, from its first occurrence to the boundary +// that ended it. +type ObservationIncident struct { + ID string `json:"observation_id"` + WorkerID string `json:"worker_id"` + Incarnation string `json:"incarnation,omitempty"` + TaskID string `json:"task_id,omitempty"` + LeaseEpoch string `json:"lease_epoch,omitempty"` + // Signature is the message with its task ids, commit shas, paths and + // durations replaced, so the same failure on two tasks shares it. Grouping + // on the raw message would make every task its own kind of problem. + Signature string `json:"signature"` + Detail string `json:"detail,omitempty"` + FirstSeen time.Time `json:"first_seen"` + // LastSeen is the last actual occurrence. ClosedAt is when Orchestra + // finalized the incident, which is later and often much later: an incident + // stays open until its lease ends, and open means "not yet final evidence" + // rather than "happening right now". + LastSeen time.Time `json:"last_seen,omitempty"` + ClosedAt time.Time `json:"closed_at,omitempty"` + RepeatCount int `json:"repeat_count,omitempty"` + CloseReason ObservationCloseReason `json:"close_reason,omitempty"` +} + +// Key identifies an incident. Two workers reporting the same failure are two +// incidents, and so are two leases of one task. +func (i ObservationIncident) Key() string { + return strings.Join([]string{i.WorkerID, i.TaskID, i.LeaseEpoch, i.Signature}, "\x00") +} + +var ( + observationID = regexp.MustCompile(`\b[0-9A-HJKMNP-TV-Z]{26}\b`) + observationSHA = regexp.MustCompile(`\b[0-9a-f]{7,64}\b`) + observationDuration = regexp.MustCompile(`\b\d+(\.\d+)?(ns|us|µs|ms|s|m|h)(\d+(\.\d+)?(ns|us|µs|ms|s|m|h))*\b`) + observationNumber = regexp.MustCompile(`\b\d+\b`) + observationPath = regexp.MustCompile(`(/[\w.-]+){2,}`) +) + +// ObservationSignature collapses one message to the kind of failure it is. +// "lease A not renewed" and "lease B not renewed" are the same problem seen +// twice, which is the whole basis of counting recurrence across tasks. +func ObservationSignature(message string) string { + s := strings.TrimSpace(message) + s = observationID.ReplaceAllString(s, "") + s = observationPath.ReplaceAllString(s, "") + s = observationDuration.ReplaceAllString(s, "") + s = observationSHA.ReplaceAllString(s, "") + s = observationNumber.ReplaceAllString(s, "") + s = strings.Join(strings.Fields(s), " ") + if len(s) > 200 { + s = s[:200] + } + return s +} + +func ValidateObservationIncidentOpened(p map[string]any) error { + if id, _ := p["observation_id"].(string); strings.TrimSpace(id) == "" { + return fmt.Errorf("%w: observation_id required", ErrInvalid) + } + if w, _ := p["worker_id"].(string); strings.TrimSpace(w) == "" { + return fmt.Errorf("%w: worker_id required", ErrInvalid) + } + if sig, _ := p["signature"].(string); strings.TrimSpace(sig) == "" { + return fmt.Errorf("%w: signature required", ErrInvalid) + } + return nil +} + +func ValidateObservationIncidentClosed(p map[string]any) error { + if id, _ := p["observation_id"].(string); strings.TrimSpace(id) == "" { + return fmt.Errorf("%w: observation_id required", ErrInvalid) + } + reason, _ := p["close_reason"].(string) + if !ObservationCloseReason(reason).Valid() { + return fmt.Errorf("%w: close_reason %q is not a close reason", ErrInvalid, reason) + } + if c, ok := p["repeat_count"].(float64); ok && c < 0 { + return fmt.Errorf("%w: repeat_count cannot be negative", ErrInvalid) + } + return nil +} diff --git a/internal/operations/observations.go b/internal/operations/observations.go new file mode 100644 index 0000000..ab31475 --- /dev/null +++ b/internal/operations/observations.go @@ -0,0 +1,201 @@ +package operations + +import ( + "encoding/json" + "fmt" + "time" + + "orchestra/internal/authz" + "orchestra/internal/domain" + "orchestra/internal/store" +) + +// QuietTimeout bounds an incident that has no lease to bound it. A worker-level +// observation has no terminal boundary of its own, so staleness of its last +// actual occurrence is the only honest closer. +const QuietTimeout = 10 * time.Minute + +// ObservationTracker turns a worker's bounded, lossy ring into durable +// incidents. It is deliberately not a copy of the ring. +// +// The rules that matter, and why: +// +// - An incident is opened at first sight and appended immediately, so a +// coordinator that dies mid-incident still leaves the fact that it existed. +// - A repeat updates the aggregate and appends nothing. Run 11's 409 loop +// repeated 301 times; appending each would have been 301 pieces of evidence +// for one problem, and would have made every debt item eligible at once. +// - Absence from the ring closes nothing. The ring is a bounded history, so a +// message can vanish because it was evicted rather than because it stopped. +// - The lease is the scope. The same signature going quiet and returning +// inside one epoch is one incident, not two recurrences. +type ObservationTracker struct { + Store *store.Store + // counts is the last count this tracker saw for an open incident, so a + // ring entry that is evicted and recreated accumulates rather than + // restarting. Reported 34, evicted, reported 3 again means 37 occurrences, + // not 3. In-memory: a restart loses the accumulation, never the incident. + counts map[string]int + // incarnations is the last incarnation seen per worker, which is what makes + // a restart detectable at all. + incarnations map[string]string +} + +// WorkerReport is one heartbeat's worth of attributed observations. The +// coordinator attributes them, because the ring carries only messages: the +// worker's active task and that task's current lease epoch are what bind an +// incident to the work it happened during. +type WorkerReport struct { + WorkerID string + Incarnation string + TaskID string + LeaseEpoch string + Observations []domain.WorkerObservation + At time.Time +} + +// Ingest folds one heartbeat into the durable incidents and returns the events +// it appended. Every close it decides is one of the four boundaries; none of +// them is "the message is no longer in the ring". +func (t *ObservationTracker) Ingest(r WorkerReport) ([]domain.Event, error) { + if t == nil || t.Store == nil || r.WorkerID == "" { + return nil, nil + } + if t.counts == nil { + t.counts, t.incarnations = map[string]int{}, map[string]string{} + } + at := r.At + if at.IsZero() { + at = time.Now().UTC() + } + var appended []domain.Event + + // A new process cannot continue the previous one's symptom, so its + // incidents are finalized before anything this heartbeat says is folded in. + if previous, seen := t.incarnations[r.WorkerID]; r.Incarnation != "" && seen && previous != r.Incarnation { + closed, err := t.closeWhere(at, domain.ObservationCloseWorkerRestart, func(inc domain.ObservationIncident) bool { + return inc.WorkerID == r.WorkerID + }) + appended = append(appended, closed...) + if err != nil { + return appended, err + } + } + if r.Incarnation != "" { + t.incarnations[r.WorkerID] = r.Incarnation + } + + open := map[string]domain.ObservationIncident{} + for _, inc := range t.Store.OpenObservations() { + if inc.WorkerID == r.WorkerID { + open[inc.Key()] = inc + } + } + for _, o := range r.Observations { + signature := domain.ObservationSignature(o.Message) + if signature == "" { + continue + } + candidate := domain.ObservationIncident{ + WorkerID: r.WorkerID, Incarnation: r.Incarnation, Signature: signature, + TaskID: r.TaskID, LeaseEpoch: r.LeaseEpoch, + } + existing, isOpen := open[candidate.Key()] + if !isOpen { + candidate.ID = domain.NewID() + candidate.Detail = o.Message + candidate.FirstSeen = firstOr(o.First, at) + candidate.LastSeen = firstOr(o.Last, at) + e, err := t.append(domain.EventObservationIncidentOpened, candidate) + if err != nil { + return appended, err + } + appended = append(appended, e) + t.counts[candidate.ID] = o.Count + continue + } + // Open already: accumulate, append nothing. A count lower than the last + // one means the ring evicted the entry and started it again. + delta := o.Count - t.counts[existing.ID] + if delta < 0 { + delta = o.Count + } + t.counts[existing.ID] += delta + if last := firstOr(o.Last, at); last.After(existing.LastSeen) { + existing.LastSeen = last + t.Store.NoteObservation(existing) // last_seen is durable at close + } + } + + // The boundaries. A lease that ended, an epoch that changed, and a + // worker-level incident whose last occurrence has gone stale. + closed, err := t.closeWhere(at, "", func(inc domain.ObservationIncident) bool { + if inc.WorkerID != r.WorkerID { + return false + } + if inc.TaskID == "" { + return at.Sub(inc.LastSeen) > QuietTimeout + } + return inc.TaskID != r.TaskID || inc.LeaseEpoch != r.LeaseEpoch + }) + appended = append(appended, closed...) + return appended, err +} + +// closeWhere finalizes every open incident the predicate selects. A reason of +// "" is resolved per incident, which is what lets one sweep close a lease that +// ended and a worker-level incident that went quiet. +func (t *ObservationTracker) closeWhere(at time.Time, reason domain.ObservationCloseReason, match func(domain.ObservationIncident) bool) ([]domain.Event, error) { + var out []domain.Event + for _, inc := range t.Store.OpenObservations() { + if !match(inc) { + continue + } + inc.ClosedAt = at + inc.RepeatCount = t.counts[inc.ID] + inc.CloseReason = reason + if inc.CloseReason == "" { + switch { + case inc.TaskID == "": + inc.CloseReason = domain.ObservationCloseQuietTimeout + case inc.LeaseEpoch != "": + inc.CloseReason = domain.ObservationCloseEpochChange + default: + inc.CloseReason = domain.ObservationCloseLeaseEnd + } + } + e, err := t.append(domain.EventObservationIncidentClosed, inc) + if err != nil { + return out, err + } + delete(t.counts, inc.ID) + out = append(out, e) + } + return out, nil +} + +// append writes the incident as a worker-scoped event. The task it happened +// during is carried in the payload rather than in Event.TaskID on purpose: an +// incident is evidence about a worker, and binding it to the task aggregate +// would bump that task's version from a path the lease knows nothing about. +func (t *ObservationTracker) append(typ string, inc domain.ObservationIncident) (domain.Event, error) { + b, err := json.Marshal(inc) + if err != nil { + return domain.Event{}, err + } + // "system" is the same aggregate QuotaReported uses for worker-scoped + // facts: every event needs a task id, and this evidence belongs to a + // worker rather than to any one task. + e := domain.Event{ID: domain.NewID(), Type: typ, TaskID: "system", Payload: b, Surface: string(authz.System)} + if err := t.Store.Append(e); err != nil { + return domain.Event{}, fmt.Errorf("record observation incident: %w", err) + } + return e, nil +} + +func firstOr(t, fallback time.Time) time.Time { + if t.IsZero() { + return fallback + } + return t +} diff --git a/internal/operations/observations_test.go b/internal/operations/observations_test.go new file mode 100644 index 0000000..a140fab --- /dev/null +++ b/internal/operations/observations_test.go @@ -0,0 +1,209 @@ +package operations + +import ( + "encoding/json" + "testing" + "time" + + "orchestra/internal/domain" + "orchestra/internal/store" +) + +func tracker(t *testing.T) (*ObservationTracker, *store.Store) { + t.Helper() + s, err := store.Open(t.TempDir()) + if err != nil { + t.Fatal(err) + } + return &ObservationTracker{Store: s}, s +} + +func ring(message string, count int, last time.Time) []domain.WorkerObservation { + return []domain.WorkerObservation{{Message: message, Count: count, First: last.Add(-time.Minute), Last: last}} +} + +func closedIncident(t *testing.T, s *store.Store) domain.ObservationIncident { + t.Helper() + var out domain.ObservationIncident + found := 0 + for _, e := range s.Events(0) { + if e.Type != domain.EventObservationIncidentClosed { + continue + } + found++ + if err := json.Unmarshal(e.Payload, &out); err != nil { + t.Fatal(err) + } + } + if found != 1 { + t.Fatalf("closed incidents = %d, want 1", found) + } + return out +} + +// 301 repeats of one refusal are one incident with an intensity of 301, not +// 301 pieces of evidence. Appending each would spam the log and make one stuck +// loop look like chronic, recurring debt. +func TestRepeatsAreOneIncident(t *testing.T) { + tr, s := tracker(t) + at := time.Unix(1700000000, 0).UTC() + report := func(count int, when time.Time) { + if _, err := tr.Ingest(WorkerReport{ + WorkerID: "workpc-claude", Incarnation: "boot-1", TaskID: "task-a", LeaseEpoch: "epoch-1", + Observations: ring("release task-a commit: 409 superseded", count, when), At: when, + }); err != nil { + t.Fatal(err) + } + } + report(1, at) + report(40, at.Add(time.Minute)) + report(301, at.Add(2*time.Minute)) + + opened := 0 + for _, e := range s.Events(0) { + if e.Type == domain.EventObservationIncidentOpened { + opened++ + } + if e.Type == domain.EventObservationIncidentClosed { + t.Fatal("an incident was closed while its lease was still running") + } + } + if opened != 1 { + t.Fatalf("opened %d incidents for one repeating failure", opened) + } + if open := s.OpenObservations(); len(open) != 1 || open[0].TaskID != "task-a" { + t.Fatalf("open incidents = %+v", open) + } +} + +// The ring is a bounded history, so an entry that disappears may have been +// evicted rather than resolved. Absence must not close anything, and a +// recreated entry must accumulate rather than restart its count. +func TestEvictionNeitherClosesNorRestartsTheCount(t *testing.T) { + tr, s := tracker(t) + at := time.Unix(1700000000, 0).UTC() + send := func(obs []domain.WorkerObservation, when time.Time) { + if _, err := tr.Ingest(WorkerReport{ + WorkerID: "w", Incarnation: "boot-1", TaskID: "task-a", LeaseEpoch: "epoch-1", + Observations: obs, At: when, + }); err != nil { + t.Fatal(err) + } + } + send(ring("lease task-a not renewed: agent idle", 34, at), at) + // Evicted: the message is simply gone from this heartbeat. + send(nil, at.Add(time.Minute)) + if len(s.OpenObservations()) != 1 { + t.Fatal("an incident was closed because its message left a bounded ring") + } + // Recreated, counting from scratch on the worker side. + send(ring("lease task-a not renewed: agent idle", 3, at.Add(2*time.Minute)), at.Add(2*time.Minute)) + + // The lease ends, which is a real boundary. + if _, err := tr.Ingest(WorkerReport{WorkerID: "w", Incarnation: "boot-1", At: at.Add(3 * time.Minute)}); err != nil { + t.Fatal(err) + } + inc := closedIncident(t, s) + if inc.RepeatCount != 37 { + t.Fatalf("repeat_count = %d, want 37 (34 before eviction plus 3 after)", inc.RepeatCount) + } + if inc.CloseReason != domain.ObservationCloseEpochChange { + t.Fatalf("close_reason = %q", inc.CloseReason) + } + if !inc.LastSeen.Equal(at.Add(2 * time.Minute)) { + t.Fatalf("last_seen = %s, want the last actual occurrence", inc.LastSeen) + } + if !inc.ClosedAt.After(inc.LastSeen) { + t.Fatal("closed_at must be when Orchestra finalized it, not when the failure last happened") + } +} + +// Recurrence is independent incidents. The same signature on two tasks is two, +// which is the evidence that means something; repeats inside one are intensity. +func TestTheSameSignatureOnAnotherTaskIsASecondIncident(t *testing.T) { + tr, s := tracker(t) + at := time.Unix(1700000000, 0).UTC() + if _, err := tr.Ingest(WorkerReport{ + WorkerID: "w", Incarnation: "boot-1", TaskID: "task-a", LeaseEpoch: "e1", + Observations: ring("lease task-a not renewed: agent idle", 5, at), At: at, + }); err != nil { + t.Fatal(err) + } + if _, err := tr.Ingest(WorkerReport{ + WorkerID: "w", Incarnation: "boot-1", TaskID: "task-b", LeaseEpoch: "e2", + Observations: ring("lease task-b not renewed: agent idle", 2, at.Add(time.Minute)), At: at.Add(time.Minute), + }); err != nil { + t.Fatal(err) + } + opened, closed := 0, 0 + for _, e := range s.Events(0) { + switch e.Type { + case domain.EventObservationIncidentOpened: + opened++ + case domain.EventObservationIncidentClosed: + closed++ + } + } + if opened != 2 { + t.Fatalf("opened = %d, want one incident per lease", opened) + } + if closed != 1 { + t.Fatalf("closed = %d, want the first lease finalized when the second began", closed) + } +} + +// A restart cannot continue the previous process's symptom. +func TestAWorkerRestartClosesItsIncidents(t *testing.T) { + tr, s := tracker(t) + at := time.Unix(1700000000, 0).UTC() + if _, err := tr.Ingest(WorkerReport{ + WorkerID: "w", Incarnation: "boot-1", TaskID: "task-a", LeaseEpoch: "e1", + Observations: ring("herdr unreachable", 9, at), At: at, + }); err != nil { + t.Fatal(err) + } + if _, err := tr.Ingest(WorkerReport{WorkerID: "w", Incarnation: "boot-2", At: at.Add(time.Minute)}); err != nil { + t.Fatal(err) + } + inc := closedIncident(t, s) + if inc.CloseReason != domain.ObservationCloseWorkerRestart || inc.RepeatCount != 9 { + t.Fatalf("incident = %+v", inc) + } +} + +// An observation with no lease has no terminal boundary, so staleness of its +// last actual occurrence is what ends it. +func TestAWorkerLevelIncidentClosesOnQuietTimeout(t *testing.T) { + tr, s := tracker(t) + at := time.Unix(1700000000, 0).UTC() + if _, err := tr.Ingest(WorkerReport{ + WorkerID: "w", Incarnation: "boot-1", + Observations: ring("heartbeat: connection refused", 4, at), At: at, + }); err != nil { + t.Fatal(err) + } + if _, err := tr.Ingest(WorkerReport{WorkerID: "w", Incarnation: "boot-1", At: at.Add(time.Minute)}); err != nil { + t.Fatal(err) + } + if len(s.OpenObservations()) != 1 { + t.Fatal("a worker-level incident closed before its quiet timeout") + } + if _, err := tr.Ingest(WorkerReport{WorkerID: "w", Incarnation: "boot-1", At: at.Add(QuietTimeout + time.Minute)}); err != nil { + t.Fatal(err) + } + if inc := closedIncident(t, s); inc.CloseReason != domain.ObservationCloseQuietTimeout { + t.Fatalf("close_reason = %q", inc.CloseReason) + } +} + +// The signature is what makes recurrence countable across tasks. +func TestSignatureCollapsesIdsAndCounts(t *testing.T) { + a := domain.ObservationSignature("lease 06G4WJ9T4F35NZC4Z8QQXM9Z6G not renewed: agent status idle and pane unchanged") + b := domain.ObservationSignature("lease 06G4VF5HZW7Q4JBM3TTY7W1Y64 not renewed: agent status idle and pane unchanged") + if a != b { + t.Fatalf("the same failure on two tasks has two signatures:\n%s\n%s", a, b) + } + if c := domain.ObservationSignature("release 06G4WJ9T4F35NZC4Z8QQXM9Z6G commit: 409 superseded"); c == a { + t.Fatal("two different failures collapsed to one signature") + } +} diff --git a/internal/store/store.go b/internal/store/store.go index 6c86451..0bacd66 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -49,6 +49,9 @@ type Store struct { cursors map[string]string cursorPath string decisionSource map[string]string + // openObservations are the incidents opened and not yet closed, by id. + // Derived from the log, so a restart finds them again. + openObservations map[string]domain.ObservationIncident // PreLease runs immediately before a lease is minted, which is the single // point where ownership of a task begins. Reconciliation of newer human // input belongs here rather than in any individual launch path, because a @@ -96,7 +99,7 @@ func Open(dir string) (*Store, error) { if t, ok := s.tasks[e.TaskID]; ok && e.Version != t.Version+1 { return nil, domain.ErrConflict } - global := e.Type == "QuotaReported" || e.Type == "StandupAdvisory" || e.Type == "ApprovalGranted" || e.Type == "ApprovalDenied" + global := domain.EventWithoutTask(e.Type) if _, ok := s.tasks[e.TaskID]; !ok && e.Type != "TaskCreated" && !global { return nil, domain.ErrNotFound } @@ -490,6 +493,26 @@ func (s *Store) apply(e domain.Event) error { t.Version = e.Version s.replaceTask(e.TaskID, t) return nil + case domain.EventObservationIncidentOpened: + // Open incidents are projected so a coordinator restart resumes them + // instead of orphaning them half-recorded. The event is what makes an + // incident durable at first sight; this is how it is found again. + var inc domain.ObservationIncident + if err := json.Unmarshal(e.Payload, &inc); err != nil { + return err + } + if s.openObservations == nil { + s.openObservations = map[string]domain.ObservationIncident{} + } + s.openObservations[inc.ID] = inc + return nil + case domain.EventObservationIncidentClosed: + var inc domain.ObservationIncident + if err := json.Unmarshal(e.Payload, &inc); err != nil { + return err + } + delete(s.openObservations, inc.ID) + return nil case domain.EventPlanMismatchRecorded: // Projected so the phase this reopens can be told what reopened it. // The event is the record; this is the live instruction derived from @@ -847,7 +870,7 @@ func (s *Store) Append(e domain.Event) error { return fmt.Errorf("%w: corrects references unknown event %q for this task", domain.ErrInvalid, corrects) } } - global := e.Type == "QuotaReported" || e.Type == "StandupAdvisory" || e.Type == "ApprovalGranted" || e.Type == "ApprovalDenied" + global := domain.EventWithoutTask(e.Type) if !taskExists && e.Type != "TaskCreated" && !global { return domain.ErrNotFound } @@ -909,7 +932,7 @@ func (s *Store) Append(e domain.Event) error { // for every lifecycle mutation. func (s *Store) validateTransition(e domain.Event, t domain.Task, exists bool, p map[string]any) error { if !exists { - if e.Type != "TaskCreated" && e.Type != "QuotaReported" && e.Type != "StandupAdvisory" && e.Type != "ApprovalGranted" && e.Type != "ApprovalDenied" { + if e.Type != "TaskCreated" && !domain.EventWithoutTask(e.Type) { return domain.ErrNotFound } return nil @@ -1113,6 +1136,31 @@ func (s *Store) Artifact(ref string) ([]byte, error) { return b, nil } +// NoteObservation updates an open incident's last actual occurrence. It writes +// no event: the aggregate is durable when the incident is finalized, and +// appending one per heartbeat is exactly the spam this design exists to avoid. +func (s *Store) NoteObservation(inc domain.ObservationIncident) { + s.mu.Lock() + defer s.mu.Unlock() + if _, ok := s.openObservations[inc.ID]; ok { + s.openObservations[inc.ID] = inc + } +} + +// OpenObservations returns the incidents that are open, newest first by first +// sight. Open means not yet finalized as durable evidence, never "the failure +// is happening right now". +func (s *Store) OpenObservations() []domain.ObservationIncident { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]domain.ObservationIncident, 0, len(s.openObservations)) + for _, inc := range s.openObservations { + out = append(out, inc) + } + sort.Slice(out, func(i, j int) bool { return out[i].FirstSeen.After(out[j].FirstSeen) }) + return out +} + func (s *Store) Task(id string) (domain.Task, bool) { s.mu.Lock() defer s.mu.Unlock()