From 2856bd16b3959ddfbce56ce21a047a5c7d0bd481 Mon Sep 17 00:00:00 2001 From: rcourtman <8825017+rcourtman@users.noreply.github.com> Date: Mon, 21 Sep 2026 17:18:32 +0100 Subject: [PATCH] Recover saved Patrol objectives without observers Saved objectives could remain uncovered indefinitely after their in-memory setup trigger was lost or rejected. Reconcile missing or invalidated observers from retained intent and recheck its revision at run admission. Persist attempt timing before provider work so restart and history pruning cannot erase pacing. Keep coverage dependent on actual observer validation, installation and health evidence. Support case patrol-objective-recovery. --- .../v6/internal/subsystems/ai-runtime.md | 12 + internal/ai/patrol_objective_recovery.go | 77 +++++ internal/ai/patrol_objective_recovery_test.go | 316 ++++++++++++++++++ internal/ai/patrol_objectives.go | 27 +- internal/ai/patrol_observer_runtime.go | 4 +- internal/ai/patrol_run.go | 3 + 6 files changed, 426 insertions(+), 13 deletions(-) create mode 100644 internal/ai/patrol_objective_recovery.go create mode 100644 internal/ai/patrol_objective_recovery_test.go diff --git a/docs/release-control/v6/internal/subsystems/ai-runtime.md b/docs/release-control/v6/internal/subsystems/ai-runtime.md index 08b1b42e27..fbfdc60103 100644 --- a/docs/release-control/v6/internal/subsystems/ai-runtime.md +++ b/docs/release-control/v6/internal/subsystems/ai-runtime.md @@ -831,6 +831,18 @@ trigger kinds, version, state transitions, and machine failure code are validated and persisted by core. Observer authority never includes an infrastructure mutation capability. +Active retained objectives with a missing observer, or an observer disabled by +an intent edit, remain durable setup requests. The observer reconciliation loop +redelivers them through the existing objective-planning trigger after restart, +queue rejection, or a busy worker. It respects disabled or blocked Patrol and +never schedules a paused or archived objective. Admission rechecks the exact +stored revision after acquiring the run slot and atomically persists +`last_planning_attempt_at` before a provider call. Unfinished attempts retry at +the configured Patrol interval, independently of run-history retention. This +timestamp does not change the operator revision, edit time, or coverage, and an +operator edit clears it. A proposal, including a rejected or proxy proposal, +is an explicit model handoff and is not automatically replaced by recovery. + During the first-party Patrol detection profile only, the core may attach the objective store through the narrow `patrol_propose_observer` tool adapter. The model must address an active objective at its current optimistic revision and diff --git a/internal/ai/patrol_objective_recovery.go b/internal/ai/patrol_objective_recovery.go new file mode 100644 index 0000000000..ee68b4fcff --- /dev/null +++ b/internal/ai/patrol_objective_recovery.go @@ -0,0 +1,77 @@ +package ai + +import ( + "time" + + "github.com/rs/zerolog/log" +) + +func patrolObjectiveNeedsPlanning(objective PatrolObjective) bool { + return objective.Status == PatrolObjectiveActive && + (objective.Observer == nil || objective.Observer.State == PatrolObserverDisabled) +} + +// Saved intent is the durable setup request. The trigger queue is only a +// delivery mechanism and may disappear on restart or reject work while busy. +func (p *PatrolService) queueMissingObjectiveCoverage(objectives []PatrolObjective, now time.Time) { + p.mu.RLock() + canPlan := p.config.Enabled && p.config.RuntimeBlockedReason == "" && !p.runInProgress + p.mu.RUnlock() + if !canPlan || IsDemoMode() { + return + } + for _, objective := range objectives { + if patrolObjectiveNeedsPlanning(objective) && p.objectiveCoverageAttemptDue(objective, now) { + p.QueueObjectiveCoverage(objective) + } + } +} + +// A delivered attempt with no retained proposal retries at the operator's +// existing Patrol interval, not at the five-second observer sampling rate. +// The timestamp belongs to retained intent, so pruning run history or +// restarting the service cannot erase the interval between provider calls. +func (p *PatrolService) objectiveCoverageAttemptDue(objective PatrolObjective, now time.Time) bool { + p.mu.RLock() + interval := p.config.GetInterval() + p.mu.RUnlock() + return objective.LastPlanningAttemptAt == nil || !now.Before(objective.LastPlanningAttemptAt.Add(interval)) +} + +// Recheck after acquiring the run slot. A queued snapshot must not start a +// second model call after another run supplied the observer, or act on intent +// that the operator has since edited, paused, or deleted. +func (p *PatrolService) beginObjectivePlanning(scope PatrolScope, now time.Time) bool { + p.mu.RLock() + store := p.objectiveStore + interval := p.config.GetInterval() + p.mu.RUnlock() + if store == nil || scope.ObjectiveContext == nil { + return false + } + accepted, err := store.beginObserverPlanning(scope.ObjectiveContext.ObjectiveID, scope.ObjectiveContext.Revision, now, interval) + if err != nil { + log.Warn().Err(err).Str("objective_id", scope.ObjectiveContext.ObjectiveID).Msg("Patrol objective setup attempt could not be retained") + } + return accepted && err == nil +} + +// Claim the exact revision and persist before calling the provider. This is +// scheduling metadata, not an operator edit or evidence of monitoring coverage. +func (s *PatrolObjectiveStore) beginObserverPlanning(id string, revision uint64, now time.Time, interval time.Duration) (bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + objective, found := s.objectives[id] + if !found || objective.Revision != revision || !patrolObjectiveNeedsPlanning(*objective) || + (objective.LastPlanningAttemptAt != nil && now.Before(objective.LastPlanningAttemptAt.Add(interval))) { + return false, nil + } + next := clonePatrolObjectiveMap(s.objectives) + attemptAt := now.UTC() + next[id].LastPlanningAttemptAt = &attemptAt + if err := s.persistLocked(next); err != nil { + return false, err + } + s.objectives = next + return true, nil +} diff --git a/internal/ai/patrol_objective_recovery_test.go b/internal/ai/patrol_objective_recovery_test.go new file mode 100644 index 0000000000..6868b37bb8 --- /dev/null +++ b/internal/ai/patrol_objective_recovery_test.go @@ -0,0 +1,316 @@ +package ai + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/models" +) + +func TestPatrolObjectiveRecoveryRestoresSavedMissingObserver(t *testing.T) { + now := time.Now().UTC() + dir := t.TempDir() + store, err := NewPatrolObjectiveStore(dir) + if err != nil { + t.Fatal(err) + } + objective, err := store.Create(CreatePatrolObjectiveInput{ + Brief: "Keep critical services available", ResourceIDs: []string{"node-1"}, + }, now.Add(-4*24*time.Hour)) + if err != nil { + t.Fatal(err) + } + // The saved intent survives a restart. The original in-memory trigger does not. + reloaded, err := NewPatrolObjectiveStore(dir) + if err != nil { + t.Fatal(err) + } + patrol := NewPatrolService(nil, mockPatrolStateProvider{state: models.StateSnapshot{ + Nodes: []models.Node{{ID: "node-1", Name: "node-1", Status: "online"}}, + }}) + patrol.SetObjectiveStore(reloaded) + tm := NewTriggerManager(DefaultTriggerManagerConfig()) + patrol.SetTriggerManager(tm) + patrol.processObjectiveObservers(now) + if tm.GetPendingCount() != 1 { + t.Fatalf("saved active objective was stranded without setup: pending=%d", tm.GetPendingCount()) + } + queued := tm.pendingTriggers[0] + if queued.Reason != TriggerReasonObjectiveChanged || queued.ObjectiveContext == nil || queued.ObjectiveContext.ObjectiveID != objective.ID || queued.ObjectiveContext.Revision != objective.Revision { + t.Fatalf("recovery lost retained intent: %+v", queued) + } + patrol.processObjectiveObservers(now.Add(time.Second)) + if tm.GetPendingCount() != 1 { + t.Fatal("repeated reconciliation duplicated setup") + } + got, _ := reloaded.Get(objective.ID, now) + if got.Revision != objective.Revision || got.Observer != nil || got.Coverage.State != PatrolObjectiveUncovered { + t.Fatalf("scheduling invented coverage or changed intent: %+v", got) + } + if !patrol.beginObjectivePlanning(queued, now) { + t.Fatal("recovered setup was not admitted") + } + // Substitute only the model's proposal. Validation, installation, and + // measured coverage still run through the real observer runtime. + _, err = reloaded.ProposeObserver(objective.ID, ProposePatrolObserverInput{ + ExpectedRevision: queued.ObjectiveContext.Revision, EvidenceFit: PatrolObserverEvidenceFitDirect, + Interpretation: "The selected node remains online", TriggerKinds: []PatrolObserverTriggerKind{PatrolObserverTriggerInterval}, + ProbeJSON: `{"runtime":"pulse-resource-state/v1","path":"status","operator":"equals","value":"online","sample_interval_seconds":10,"wake_after_consecutive_failures":2}`, + WakeEvidence: "Node offline", RequirementsJSON: `{}`, + }, now) + if err != nil { + t.Fatal(err) + } + patrol.processObjectiveObservers(now.Add(2 * time.Second)) + covered, _ := reloaded.Get(objective.ID, now.Add(2*time.Second)) + if covered.Coverage.State != PatrolObjectiveCovered || covered.Observer.State != PatrolObserverInstalled { + t.Fatalf("recovered setup did not install and verify the observer: %+v", covered) + } +} + +func TestPatrolObjectiveRecoveryRespectsRuntimeAndOperatorState(t *testing.T) { + for _, state := range []string{"disabled", "blocked", "busy", "paused", "archived"} { + t.Run(state, func(t *testing.T) { + now := time.Now().UTC() + store := NewInMemoryPatrolObjectiveStore() + objective, err := store.Create(CreatePatrolObjectiveInput{Brief: "Keep services available", ResourceIDs: []string{"node-1"}}, now) + if err != nil { + t.Fatal(err) + } + patrol := NewPatrolService(nil, nil) + patrol.SetObjectiveStore(store) + tm := NewTriggerManager(DefaultTriggerManagerConfig()) + patrol.SetTriggerManager(tm) + switch state { + case "disabled": + patrol.config.Enabled = false + case "blocked": + patrol.config.RuntimeBlockedReason = "Provider unavailable" + case "busy": + patrol.runInProgress = true + default: + status := PatrolObjectiveStatus(state) + if _, err := store.Update(objective.ID, UpdatePatrolObjectiveInput{ExpectedRevision: objective.Revision, Status: &status}, now); err != nil { + t.Fatal(err) + } + } + patrol.processObjectiveObservers(now) + if tm.GetPendingCount() != 0 { + t.Fatal("setup bypassed runtime or operator state") + } + if state == "busy" { + patrol.runInProgress = false + patrol.processObjectiveObservers(now.Add(time.Second)) + if tm.GetPendingCount() != 1 { + t.Fatal("setup did not recover when the worker became idle") + } + } + }) + } +} + +func TestPatrolObjectiveRecoveryRetainsIntentWhenQueueRejectsSetup(t *testing.T) { + now := time.Now().UTC() + store := NewInMemoryPatrolObjectiveStore() + if _, err := store.Create(CreatePatrolObjectiveInput{Brief: "Keep services available", ResourceIDs: []string{"node-1"}}, now); err != nil { + t.Fatal(err) + } + patrol := NewPatrolService(nil, nil) + patrol.SetObjectiveStore(store) + tm := NewTriggerManager(TriggerManagerConfig{MaxPendingTriggers: 1}) + patrol.SetTriggerManager(tm) + tm.TriggerPatrol(PatrolScope{Reason: TriggerReasonManual, Priority: triggerPriorityObjective + 1}) + patrol.processObjectiveObservers(now) + if tm.GetPendingCount() != 1 || tm.pendingTriggers[0].Reason != TriggerReasonManual { + t.Fatal("recovery displaced higher-priority work") + } + tm.pendingTriggers = nil + patrol.processObjectiveObservers(now.Add(time.Second)) + if tm.GetPendingCount() != 1 || tm.pendingTriggers[0].Reason != TriggerReasonObjectiveChanged { + t.Fatal("queue rejection permanently lost setup") + } +} + +func TestPatrolObjectiveRecoveryPacesDeliveredAttemptsAcrossRestart(t *testing.T) { + now := time.Now().UTC() + dir := t.TempDir() + store, err := NewPatrolObjectiveStore(dir) + if err != nil { + t.Fatal(err) + } + objective, err := store.Create(CreatePatrolObjectiveInput{Brief: "Keep services available", ResourceIDs: []string{"node-1"}}, now) + if err != nil { + t.Fatal(err) + } + interval := DefaultPatrolConfig().GetInterval() + accepted, err := store.beginObserverPlanning(objective.ID, objective.Revision, now, interval) + if err != nil || !accepted { + t.Fatalf("start setup: accepted=%v err=%v", accepted, err) + } + reloaded, err := NewPatrolObjectiveStore(dir) + if err != nil { + t.Fatal(err) + } + saved, _ := reloaded.Get(objective.ID, now) + if saved.Revision != objective.Revision || !saved.UpdatedAt.Equal(objective.UpdatedAt) || saved.LastPlanningAttemptAt == nil { + t.Fatalf("attempt changed intent or lost pacing: %+v", saved) + } + patrol := NewPatrolService(nil, nil) + patrol.SetObjectiveStore(reloaded) + tm := NewTriggerManager(DefaultTriggerManagerConfig()) + patrol.SetTriggerManager(tm) + // No run history or in-memory scheduling state survives this restart. + patrol.processObjectiveObservers(now.Add(time.Second)) + if tm.GetPendingCount() != 0 { + t.Fatal("restart erased provider-call pacing") + } + patrol.processObjectiveObservers(now.Add(interval)) + if tm.GetPendingCount() != 1 { + t.Fatal("missing proposal did not retry at the configured Patrol interval") + } +} + +func TestPatrolObjectiveRecoveryRejectsStaleQueuedIntent(t *testing.T) { + for _, change := range []string{"edited", "paused", "deleted", "proposal", "attempted"} { + t.Run(change, func(t *testing.T) { + now := time.Now().UTC() + store := NewInMemoryPatrolObjectiveStore() + objective, err := store.Create(CreatePatrolObjectiveInput{Brief: "Keep services available", ResourceIDs: []string{"node-1"}}, now) + if err != nil { + t.Fatal(err) + } + patrol := NewPatrolService(nil, nil) + patrol.SetObjectiveStore(store) + tm := NewTriggerManager(DefaultTriggerManagerConfig()) + patrol.SetTriggerManager(tm) + patrol.QueueObjectiveCoverage(objective) + scope := tm.pendingTriggers[0] + switch change { + case "edited": + brief := "Keep storage available" + _, err = store.Update(objective.ID, UpdatePatrolObjectiveInput{ExpectedRevision: objective.Revision, Brief: &brief}, now) + case "paused": + status := PatrolObjectivePaused + _, err = store.Update(objective.ID, UpdatePatrolObjectiveInput{ExpectedRevision: objective.Revision, Status: &status}, now) + case "deleted": + err = store.Delete(objective.ID, objective.Revision) + case "proposal": + _, err = store.ProposeObserver(objective.ID, ProposePatrolObserverInput{ + ExpectedRevision: objective.Revision, EvidenceFit: PatrolObserverEvidenceFitDirect, + Interpretation: "Observe scoped resource state", TriggerKinds: []PatrolObserverTriggerKind{PatrolObserverTriggerInterval}, + ProbeJSON: `{"runtime":"pulse-resource-state/v1","path":"status","operator":"equals","value":"online","sample_interval_seconds":10,"wake_after_consecutive_failures":2}`, + WakeEvidence: "Resource offline", RequirementsJSON: `{}`, + }, now) + case "attempted": + _, err = store.beginObserverPlanning(objective.ID, objective.Revision, now, patrol.config.GetInterval()) + } + if err != nil { + t.Fatal(err) + } + before := len(patrol.runHistoryStore.GetAll()) + patrol.runScopedPatrol(context.Background(), scope) + if len(patrol.runHistoryStore.GetAll()) != before || patrol.runInProgress { + t.Fatal("obsolete setup started a run or retained the worker slot") + } + }) + } +} + +func TestPatrolObjectiveRecoveryReplansEditedObserverThenInstalls(t *testing.T) { + now := time.Now().UTC() + store := NewInMemoryPatrolObjectiveStore() + objective := createInstallablePatrolObserver(t, store, now, "node-1") + patrol := NewPatrolService(nil, mockPatrolStateProvider{state: models.StateSnapshot{ + Nodes: []models.Node{{ID: "node-1", Name: "node-1", Status: "online"}}, + }}) + patrol.SetObjectiveStore(store) + tm := NewTriggerManager(DefaultTriggerManagerConfig()) + patrol.SetTriggerManager(tm) + patrol.processObjectiveObservers(now) + installed, _ := store.Get(objective.ID, now) + if tm.GetPendingCount() != 0 || installed.Coverage.State != PatrolObjectiveCovered { + t.Fatal("existing observer was replanned instead of installed") + } + brief := "Keep this node available" + updated, err := store.Update(objective.ID, UpdatePatrolObjectiveInput{ExpectedRevision: installed.Revision, Brief: &brief}, now) + if err != nil { + t.Fatal(err) + } + patrol.processObjectiveObservers(now.Add(time.Second)) + if tm.GetPendingCount() != 1 || tm.pendingTriggers[0].ObjectiveContext.Revision != updated.Revision { + t.Fatal("disabled observer left revised intent without setup") + } + _, err = store.ProposeObserver(objective.ID, ProposePatrolObserverInput{ + ExpectedRevision: updated.Revision, EvidenceFit: PatrolObserverEvidenceFitDirect, + Interpretation: "The node remains online", TriggerKinds: []PatrolObserverTriggerKind{PatrolObserverTriggerInterval}, + ProbeJSON: `{"runtime":"pulse-resource-state/v1","path":"status","operator":"equals","value":"online","sample_interval_seconds":10,"wake_after_consecutive_failures":2}`, + WakeEvidence: "Node offline", RequirementsJSON: `{}`, + }, now.Add(2*time.Second)) + if err != nil { + t.Fatal(err) + } + patrol.processObjectiveObservers(now.Add(3 * time.Second)) + covered, _ := store.Get(objective.ID, now.Add(3*time.Second)) + if covered.Coverage.State != PatrolObjectiveCovered || covered.Observer.State != PatrolObserverInstalled { + t.Fatalf("recovered proposal did not acquire real coverage: %+v", covered) + } +} + +func TestPatrolObjectiveRecoveryClaimsAttemptAtomically(t *testing.T) { + now := time.Now().UTC() + store := NewInMemoryPatrolObjectiveStore() + objective, err := store.Create(CreatePatrolObjectiveInput{Brief: "Keep services available", ResourceIDs: []string{"node-1"}}, now) + if err != nil { + t.Fatal(err) + } + var accepted atomic.Int32 + var wg sync.WaitGroup + for range 8 { + wg.Go(func() { + ok, err := store.beginObserverPlanning(objective.ID, objective.Revision, now, time.Hour) + if err != nil { + t.Error(err) + } + if ok { + accepted.Add(1) + } + }) + } + wg.Wait() + if accepted.Load() != 1 { + t.Fatalf("accepted %d concurrent attempts", accepted.Load()) + } + brief := "Keep the selected node available" + updated, err := store.Update(objective.ID, UpdatePatrolObjectiveInput{ExpectedRevision: objective.Revision, Brief: &brief}, now) + if err != nil { + t.Fatal(err) + } + if updated.LastPlanningAttemptAt != nil { + t.Fatal("new intent inherited the previous attempt's delay") + } + if ok, err := store.beginObserverPlanning(updated.ID, updated.Revision, now, time.Hour); !ok || err != nil { + t.Fatalf("new revision was not admitted: accepted=%v err=%v", ok, err) + } +} + +func TestPatrolObjectiveRecoveryDoesNotCallProviderWithoutDurableAttempt(t *testing.T) { + now := time.Now().UTC() + store := NewInMemoryPatrolObjectiveStore() + objective, err := store.Create(CreatePatrolObjectiveInput{Brief: "Keep services available", ResourceIDs: []string{"node-1"}}, now) + if err != nil { + t.Fatal(err) + } + // A directory cannot be replaced by the atomic file rename, on any platform. + store.filePath = t.TempDir() + accepted, err := store.beginObserverPlanning(objective.ID, objective.Revision, now, time.Hour) + if err == nil || accepted { + t.Fatalf("failed persistence admitted provider work: accepted=%v err=%v", accepted, err) + } + got, _ := store.Get(objective.ID, now) + if got.LastPlanningAttemptAt != nil { + t.Fatal("failed persistence suppressed a later recovery attempt") + } +} diff --git a/internal/ai/patrol_objectives.go b/internal/ai/patrol_objectives.go index d5b5b6cc58..2fbe73a3c1 100644 --- a/internal/ai/patrol_objectives.go +++ b/internal/ai/patrol_objectives.go @@ -163,18 +163,19 @@ type PatrolObjectiveCoverage struct { } type PatrolObjective struct { - ID string `json:"id"` - Brief string `json:"brief"` - OptionalContext string `json:"optional_context,omitempty"` - Scope PatrolObjectiveScope `json:"scope"` - Status PatrolObjectiveStatus `json:"status"` - Coverage PatrolObjectiveCoverage `json:"coverage"` - Observer *PatrolObserverRecord `json:"observer,omitempty"` - Revision uint64 `json:"revision"` - CreatedBy string `json:"created_by,omitempty"` - UpdatedBy string `json:"updated_by,omitempty"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` + ID string `json:"id"` + Brief string `json:"brief"` + OptionalContext string `json:"optional_context,omitempty"` + Scope PatrolObjectiveScope `json:"scope"` + Status PatrolObjectiveStatus `json:"status"` + Coverage PatrolObjectiveCoverage `json:"coverage"` + Observer *PatrolObserverRecord `json:"observer,omitempty"` + Revision uint64 `json:"revision"` + CreatedBy string `json:"created_by,omitempty"` + UpdatedBy string `json:"updated_by,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` + LastPlanningAttemptAt *time.Time `json:"last_planning_attempt_at,omitempty"` } type CreatePatrolObjectiveInput struct { @@ -423,6 +424,7 @@ func (s *PatrolObjectiveStore) Update(id string, input UpdatePatrolObjectiveInpu } updated.UpdatedBy = normalizePatrolObjectiveActor(input.Actor) updated.UpdatedAt = now + updated.LastPlanningAttemptAt = nil updated.Revision++ next := clonePatrolObjectiveMap(s.objectives) next[id] = updated @@ -1149,6 +1151,7 @@ func clonePatrolObjective(objective *PatrolObjective) *PatrolObjective { copy := *objective copy.Scope.ResourceIDs = append([]string{}, objective.Scope.ResourceIDs...) copy.Observer = clonePatrolObserver(objective.Observer) + copy.LastPlanningAttemptAt = clonePatrolTime(objective.LastPlanningAttemptAt) copy.Coverage.ValidUntil = clonePatrolTime(objective.Coverage.ValidUntil) copy.Coverage.LastEvidenceAt = clonePatrolTime(objective.Coverage.LastEvidenceAt) return © diff --git a/internal/ai/patrol_observer_runtime.go b/internal/ai/patrol_observer_runtime.go index 73585eb8db..fe97dc7658 100644 --- a/internal/ai/patrol_observer_runtime.go +++ b/internal/ai/patrol_observer_runtime.go @@ -160,7 +160,9 @@ func (p *PatrolService) processObjectiveObservers(now time.Time) { } activeExecutions := make(map[string]struct{}) healthUpdates := make([]patrolObserverHealthUpdate, 0) - for _, objective := range store.List(false, now) { + objectives := store.List(false, now) + p.queueMissingObjectiveCoverage(objectives, now) + for _, objective := range objectives { if objective.Status != PatrolObjectiveActive || objective.Observer == nil { continue } diff --git a/internal/ai/patrol_run.go b/internal/ai/patrol_run.go index 3afb1175ec..fb499b5c4f 100644 --- a/internal/ai/patrol_run.go +++ b/internal/ai/patrol_run.go @@ -962,6 +962,9 @@ func (p *PatrolService) runScopedPatrolWithStart(ctx context.Context, scope Patr start := runStart.startedAt runID := runStart.id executionID := uuid.NewString() + if scope.Reason == TriggerReasonObjectiveChanged && !p.beginObjectivePlanning(scope, start) { + return + } GetPatrolMetrics().RecordRun(string(scope.Reason), "scoped") var runStats struct { resourceCount int