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