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.
This commit is contained in:
rcourtman
2026-09-21 17:18:32 +01:00
parent eb47a4e653
commit 2856bd16b3
6 changed files with 426 additions and 13 deletions
@@ -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
+77
View File
@@ -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
}
@@ -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")
}
}
+15 -12
View File
@@ -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 &copy
+3 -1
View File
@@ -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
}
+3
View File
@@ -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