mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-08-16 12:33:29 +00:00
Run Patrol against canonical availability signals
This commit is contained in:
rcourtman
parent
00e570b188
commit
6f09bc4676
6 changed files with 402 additions and 47 deletions
|
|
@ -6988,20 +6988,34 @@ still contain `auto_fix_count` load cleanly; the unknown key is ignored.
|
|||
|
||||
### Model-authored observers have a core-owned local execution floor
|
||||
|
||||
`internal/ai/patrol_observer_runtime.go` owns the first executable observer ABI
|
||||
for retained Patrol objectives. The model may author the predicate, but it
|
||||
cannot author executable code or advance lifecycle authority. Core strictly
|
||||
decodes `pulse-resource-state/v1`, requires explicit canonical resource scope,
|
||||
an interval trigger, an empty external-requirements object, one canonical
|
||||
`status` predicate, a 10-to-300-second local sample interval, and a bounded
|
||||
consecutive-failure window. Unknown fields and unsupported runtimes, triggers,
|
||||
requirements, paths, operators, or values fail closed and persist a safe
|
||||
machine reason in the `rejected` state while coverage remains uncovered.
|
||||
`internal/ai/patrol_observer_runtime.go` owns the executable observer ABIs for
|
||||
retained Patrol objectives. The model may author the meaning and predicate, but
|
||||
it cannot author executable code, network authority, or lifecycle authority.
|
||||
Core strictly decodes `pulse-resource-state/v1` for canonical resource status
|
||||
and `pulse-availability-state/v1` for the outcome of an existing canonical
|
||||
agentless availability target. Both require explicit canonical resource scope,
|
||||
an interval trigger, an empty external-requirements object, a 10-to-300-second
|
||||
local sample interval, and a bounded consecutive-failure window. The
|
||||
availability ABI accepts only an exact canonical target ID and
|
||||
`probe_outcome`; it never accepts a URL, address, header, credential, request
|
||||
body, or secret. Before installation, core proves that the target exists, is
|
||||
enabled, and belongs to the objective's resource scope. Missing, disabled,
|
||||
deleted, or cross-scope targets fail closed with typed observer reasons rather
|
||||
than being interpreted as objective failure. Unknown fields and unsupported
|
||||
runtimes, triggers, requirements, paths, operators, or values also fail closed
|
||||
and persist a safe machine reason in the `rejected` or `degraded` state while
|
||||
coverage remains truthful.
|
||||
|
||||
Accepted observers transition through proposed, validated, and installed under
|
||||
core authority. They evaluate Pulse's canonical `ReadState` status locally,
|
||||
falling back to the existing Patrol snapshot only where canonical read state is
|
||||
unavailable. Health leasing is persisted without changing the operator
|
||||
core authority. Resource observers evaluate Pulse's canonical `ReadState`
|
||||
status locally, falling back to the existing Patrol snapshot only where
|
||||
canonical read state is unavailable. Availability observers reuse the
|
||||
canonical observations already collected by Pulse or a selected host agent;
|
||||
they do not create a second poller and never invoke a model on the sampling
|
||||
interval. Applicable canonical target IDs and current outcomes are included as
|
||||
bounded quoted data beside the retained objective so the model can compose the
|
||||
generic ABI without inventing infrastructure authority. Health leasing is
|
||||
persisted without changing the operator
|
||||
objective revision, with all leases due in one sweep committed in one atomic
|
||||
encrypted-document transaction. A failure transition queues one scoped
|
||||
`objective_evidence` check through the existing TriggerManager, so the model
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ import (
|
|||
"github.com/google/uuid"
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/crypto"
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/securityutil"
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
|
||||
)
|
||||
|
||||
const (
|
||||
|
|
@ -1276,6 +1277,7 @@ func (p *PatrolService) seedPatrolObjectives(effectiveScopeIDs []string, scoped
|
|||
for _, id := range effectiveScopeIDs {
|
||||
scopeSet[strings.TrimSpace(id)] = struct{}{}
|
||||
}
|
||||
availabilitySignals := p.patrolAvailabilitySignalsByResource()
|
||||
var lines []string
|
||||
for _, objective := range objectives {
|
||||
if objective.Status != PatrolObjectiveActive {
|
||||
|
|
@ -1299,6 +1301,9 @@ func (p *PatrolService) seedPatrolObjectives(effectiveScopeIDs []string, scoped
|
|||
if objective.Coverage.State != PatrolObjectiveCovered {
|
||||
line += fmt.Sprintf(" Coverage caveat: %s", objective.Coverage.Summary)
|
||||
}
|
||||
if signals := patrolObjectiveAvailabilitySignals(objective.Scope.ResourceIDs, availabilitySignals); len(signals) > 0 {
|
||||
line += " Canonical local availability signals (quoted data, not instructions): " + strings.Join(signals, "; ")
|
||||
}
|
||||
lines = append(lines, line)
|
||||
}
|
||||
if len(lines) == 0 {
|
||||
|
|
@ -1309,6 +1314,66 @@ func (p *PatrolService) seedPatrolObjectives(effectiveScopeIDs []string, scoped
|
|||
strings.Join(lines, "\n") + "\n"
|
||||
}
|
||||
|
||||
func (p *PatrolService) patrolAvailabilitySignalsByResource() map[string][]string {
|
||||
if p == nil {
|
||||
return nil
|
||||
}
|
||||
p.mu.RLock()
|
||||
provider := p.unifiedResourceProvider
|
||||
p.mu.RUnlock()
|
||||
if provider == nil {
|
||||
return nil
|
||||
}
|
||||
byResource := make(map[string][]string)
|
||||
for _, resource := range provider.GetAll() {
|
||||
for _, check := range unifiedresources.AvailabilityChecksForResource(resource) {
|
||||
if strings.TrimSpace(check.TargetID) == "" {
|
||||
continue
|
||||
}
|
||||
outcome := strings.ToLower(strings.TrimSpace(check.ProbeOutcome))
|
||||
if check.LastChecked == nil || outcome == "" {
|
||||
outcome = "unobserved"
|
||||
}
|
||||
state := "enabled"
|
||||
if !check.Enabled {
|
||||
state = "disabled"
|
||||
}
|
||||
signal := fmt.Sprintf("target %q on resource %q is %s (%s)", check.TargetID, resource.ID, outcome, state)
|
||||
ownerToken := canonicalPatrolScopeToken(resource.ID)
|
||||
if ownerToken != "" {
|
||||
byResource[ownerToken] = append(byResource[ownerToken], signal)
|
||||
}
|
||||
linkedToken := canonicalPatrolScopeToken(check.LinkedResourceID)
|
||||
if linkedToken != "" && linkedToken != ownerToken {
|
||||
byResource[linkedToken] = append(byResource[linkedToken], signal)
|
||||
}
|
||||
}
|
||||
}
|
||||
return byResource
|
||||
}
|
||||
|
||||
func patrolObjectiveAvailabilitySignals(resourceIDs []string, byResource map[string][]string) []string {
|
||||
if len(resourceIDs) == 0 || len(byResource) == 0 {
|
||||
return nil
|
||||
}
|
||||
unique := make(map[string]struct{})
|
||||
for _, resourceID := range resourceIDs {
|
||||
for _, signal := range byResource[canonicalPatrolScopeToken(resourceID)] {
|
||||
unique[signal] = struct{}{}
|
||||
}
|
||||
}
|
||||
signals := make([]string, 0, len(unique))
|
||||
for signal := range unique {
|
||||
signals = append(signals, signal)
|
||||
}
|
||||
sort.Strings(signals)
|
||||
const maxSignals = 12
|
||||
if len(signals) > maxSignals {
|
||||
signals = signals[:maxSignals]
|
||||
}
|
||||
return signals
|
||||
}
|
||||
|
||||
func patrolObjectiveScopeIntersects(resourceIDs []string, scopeSet map[string]struct{}) bool {
|
||||
for _, id := range resourceIDs {
|
||||
if _, ok := scopeSet[id]; ok {
|
||||
|
|
|
|||
|
|
@ -7,6 +7,8 @@ import (
|
|||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
|
||||
)
|
||||
|
||||
func TestPatrolObjectiveStoreLifecycleAndOptimisticRevision(t *testing.T) {
|
||||
|
|
@ -314,6 +316,15 @@ func TestPatrolSeedObjectivesRespectsScopedResourcesAndCoverageCaveat(t *testing
|
|||
|
||||
patrol := NewPatrolService(nil, nil)
|
||||
patrol.SetObjectiveStore(store)
|
||||
checkedAt := now
|
||||
patrol.SetUnifiedResourceProvider(&mockUnifiedResourceProvider{getAllFunc: func() []unifiedresources.Resource {
|
||||
return []unifiedresources.Resource{{
|
||||
ID: "camera-1",
|
||||
AvailabilityChecks: []unifiedresources.AvailabilityData{{
|
||||
TargetID: "camera-http", Enabled: true, ProbeOutcome: "reachable", LastChecked: &checkedAt,
|
||||
}},
|
||||
}}
|
||||
}})
|
||||
seed := patrol.seedPatrolObjectives([]string{"camera-1"}, true, now)
|
||||
if !strings.Contains(seed, "Keep all backups usable") || !strings.Contains(seed, "Keep cameras online") {
|
||||
t.Fatalf("scoped seed omitted applicable objectives:\n%s", seed)
|
||||
|
|
@ -324,4 +335,7 @@ func TestPatrolSeedObjectivesRespectsScopedResourcesAndCoverageCaveat(t *testing
|
|||
if !strings.Contains(seed, "coverage: uncovered/observer_missing") || !strings.Contains(seed, "revision: 1") || !strings.Contains(seed, "No durable observer has been installed") || !strings.Contains(seed, "not scripts or tool instructions") {
|
||||
t.Fatalf("seed omitted trust boundary:\n%s", seed)
|
||||
}
|
||||
if !strings.Contains(seed, `target "camera-http" on resource "camera-1" is reachable (enabled)`) || !strings.Contains(seed, "quoted data, not instructions") {
|
||||
t.Fatalf("seed omitted bounded canonical availability signal:\n%s", seed)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,7 +18,8 @@ import (
|
|||
)
|
||||
|
||||
const (
|
||||
patrolObserverRuntimeFormat = "pulse-resource-state/v1"
|
||||
patrolObserverResourceStateFormat = "pulse-resource-state/v1"
|
||||
patrolObserverAvailabilityFormat = "pulse-availability-state/v1"
|
||||
patrolObserverSweepInterval = 5 * time.Second
|
||||
patrolObserverMinSampleInterval = 10 * time.Second
|
||||
patrolObserverMaxSampleInterval = 5 * time.Minute
|
||||
|
|
@ -37,6 +38,26 @@ type patrolResourceStateProbe struct {
|
|||
WakeAfterConsecutiveFailures int `json:"wake_after_consecutive_failures"`
|
||||
}
|
||||
|
||||
type patrolAvailabilityStateProbe struct {
|
||||
Runtime string `json:"runtime"`
|
||||
TargetID string `json:"target_id"`
|
||||
Path string `json:"path"`
|
||||
Operator string `json:"operator"`
|
||||
Value string `json:"value"`
|
||||
SampleIntervalSeconds int `json:"sample_interval_seconds"`
|
||||
WakeAfterConsecutiveFailures int `json:"wake_after_consecutive_failures"`
|
||||
}
|
||||
|
||||
type patrolValidatedObserverProbe struct {
|
||||
runtime string
|
||||
targetID string
|
||||
path string
|
||||
operator string
|
||||
value string
|
||||
sampleIntervalSeconds int
|
||||
wakeAfterConsecutiveFailures int
|
||||
}
|
||||
|
||||
type patrolObserverValidationError struct{ code string }
|
||||
|
||||
func (e *patrolObserverValidationError) Error() string { return e.code }
|
||||
|
|
@ -123,6 +144,11 @@ func (p *PatrolService) reconcileObjectiveObserver(store *PatrolObjectiveStore,
|
|||
p.recordObserverValidationFailure(store, objective, validationErr.code, now)
|
||||
return nil
|
||||
}
|
||||
state := p.currentPatrolRuntimeState()
|
||||
if bindingErr := validatePatrolObserverBinding(objective, probe, state); bindingErr != nil {
|
||||
p.recordObserverValidationFailure(store, objective, bindingErr.code, now)
|
||||
return nil
|
||||
}
|
||||
|
||||
for objective.Observer != nil && (objective.Observer.State == PatrolObserverProposed || objective.Observer.State == PatrolObserverRejected || objective.Observer.State == PatrolObserverValidated || objective.Observer.State == PatrolObserverDegraded) {
|
||||
next := *clonePatrolObserver(objective.Observer)
|
||||
|
|
@ -156,7 +182,7 @@ func (p *PatrolService) reconcileObjectiveObserver(store *PatrolObjectiveStore,
|
|||
if objective.Observer == nil || (objective.Observer.State != PatrolObserverInstalled && objective.Observer.State != PatrolObserverDegraded) {
|
||||
return nil
|
||||
}
|
||||
return p.evaluateObjectiveObserver(runtime, objective, probe, now)
|
||||
return p.evaluateObjectiveObserver(runtime, objective, probe, state, now)
|
||||
}
|
||||
|
||||
func (p *PatrolService) recordObserverValidationFailure(store *PatrolObjectiveStore, objective PatrolObjective, code string, now time.Time) {
|
||||
|
|
@ -180,9 +206,9 @@ func (p *PatrolService) recordObserverValidationFailure(store *PatrolObjectiveSt
|
|||
}
|
||||
}
|
||||
|
||||
func validatePatrolObserverArtifact(objective PatrolObjective, artifact PatrolObserverArtifact) (patrolResourceStateProbe, *patrolObserverValidationError) {
|
||||
fail := func(code string) (patrolResourceStateProbe, *patrolObserverValidationError) {
|
||||
return patrolResourceStateProbe{}, &patrolObserverValidationError{code: code}
|
||||
func validatePatrolObserverArtifact(objective PatrolObjective, artifact PatrolObserverArtifact) (patrolValidatedObserverProbe, *patrolObserverValidationError) {
|
||||
fail := func(code string) (patrolValidatedObserverProbe, *patrolObserverValidationError) {
|
||||
return patrolValidatedObserverProbe{}, &patrolObserverValidationError{code: code}
|
||||
}
|
||||
if objective.Observer == nil || !objective.Observer.ReadOnly {
|
||||
return fail("observer_not_read_only")
|
||||
|
|
@ -200,35 +226,94 @@ func validatePatrolObserverArtifact(objective PatrolObjective, artifact PatrolOb
|
|||
if len(requirements) != 0 {
|
||||
return fail("observer_requirements_unsupported")
|
||||
}
|
||||
var probe patrolResourceStateProbe
|
||||
if err := decodeStrictJSONObject(artifact.Probe, &probe); err != nil {
|
||||
var envelope struct {
|
||||
Runtime string `json:"runtime"`
|
||||
}
|
||||
if err := json.Unmarshal(artifact.Probe, &envelope); err != nil {
|
||||
return fail("observer_probe_invalid")
|
||||
}
|
||||
if probe.Runtime != patrolObserverRuntimeFormat {
|
||||
var probe patrolValidatedObserverProbe
|
||||
switch strings.TrimSpace(envelope.Runtime) {
|
||||
case patrolObserverResourceStateFormat:
|
||||
var decoded patrolResourceStateProbe
|
||||
if err := decodeStrictJSONObject(artifact.Probe, &decoded); err != nil {
|
||||
return fail("observer_probe_invalid")
|
||||
}
|
||||
probe = patrolValidatedObserverProbe{
|
||||
runtime: decoded.Runtime, path: decoded.Path, operator: decoded.Operator,
|
||||
value: decoded.Value, sampleIntervalSeconds: decoded.SampleIntervalSeconds,
|
||||
wakeAfterConsecutiveFailures: decoded.WakeAfterConsecutiveFailures,
|
||||
}
|
||||
if probe.path != "status" {
|
||||
return fail("observer_path_unsupported")
|
||||
}
|
||||
switch unifiedresources.ResourceStatus(strings.ToLower(strings.TrimSpace(probe.value))) {
|
||||
case unifiedresources.StatusOnline, unifiedresources.StatusOffline, unifiedresources.StatusWarning, unifiedresources.StatusUnknown:
|
||||
probe.value = strings.ToLower(strings.TrimSpace(probe.value))
|
||||
default:
|
||||
return fail("observer_value_unsupported")
|
||||
}
|
||||
case patrolObserverAvailabilityFormat:
|
||||
var decoded patrolAvailabilityStateProbe
|
||||
if err := decodeStrictJSONObject(artifact.Probe, &decoded); err != nil {
|
||||
return fail("observer_probe_invalid")
|
||||
}
|
||||
probe = patrolValidatedObserverProbe{
|
||||
runtime: decoded.Runtime, targetID: strings.TrimSpace(decoded.TargetID), path: decoded.Path,
|
||||
operator: decoded.Operator, value: strings.ToLower(strings.TrimSpace(decoded.Value)),
|
||||
sampleIntervalSeconds: decoded.SampleIntervalSeconds,
|
||||
wakeAfterConsecutiveFailures: decoded.WakeAfterConsecutiveFailures,
|
||||
}
|
||||
if probe.targetID == "" {
|
||||
return fail("observer_availability_target_required")
|
||||
}
|
||||
if probe.path != "probe_outcome" {
|
||||
return fail("observer_path_unsupported")
|
||||
}
|
||||
switch probe.value {
|
||||
case "reachable", "unreachable", "indeterminate":
|
||||
default:
|
||||
return fail("observer_value_unsupported")
|
||||
}
|
||||
default:
|
||||
return fail("observer_runtime_unsupported")
|
||||
}
|
||||
if probe.Path != "status" {
|
||||
return fail("observer_path_unsupported")
|
||||
}
|
||||
if probe.Operator != "equals" && probe.Operator != "not_equals" {
|
||||
if probe.operator != "equals" && probe.operator != "not_equals" {
|
||||
return fail("observer_operator_unsupported")
|
||||
}
|
||||
switch unifiedresources.ResourceStatus(strings.ToLower(strings.TrimSpace(probe.Value))) {
|
||||
case unifiedresources.StatusOnline, unifiedresources.StatusOffline, unifiedresources.StatusWarning, unifiedresources.StatusUnknown:
|
||||
probe.Value = strings.ToLower(strings.TrimSpace(probe.Value))
|
||||
default:
|
||||
return fail("observer_value_unsupported")
|
||||
}
|
||||
interval := time.Duration(probe.SampleIntervalSeconds) * time.Second
|
||||
interval := time.Duration(probe.sampleIntervalSeconds) * time.Second
|
||||
if interval < patrolObserverMinSampleInterval || interval > patrolObserverMaxSampleInterval {
|
||||
return fail("observer_interval_out_of_bounds")
|
||||
}
|
||||
if probe.WakeAfterConsecutiveFailures < 1 || probe.WakeAfterConsecutiveFailures > patrolObserverMaxConsecutiveFailures {
|
||||
if probe.wakeAfterConsecutiveFailures < 1 || probe.wakeAfterConsecutiveFailures > patrolObserverMaxConsecutiveFailures {
|
||||
return fail("observer_failure_window_out_of_bounds")
|
||||
}
|
||||
return probe, nil
|
||||
}
|
||||
|
||||
func validatePatrolObserverBinding(objective PatrolObjective, probe patrolValidatedObserverProbe, state patrolRuntimeState) *patrolObserverValidationError {
|
||||
if probe.runtime != patrolObserverAvailabilityFormat {
|
||||
return nil
|
||||
}
|
||||
check, ownerID, found := patrolAvailabilityCheckByTarget(state, probe.targetID)
|
||||
if !found {
|
||||
return &patrolObserverValidationError{code: "observer_availability_target_missing"}
|
||||
}
|
||||
if !check.Enabled {
|
||||
return &patrolObserverValidationError{code: "observer_availability_target_disabled"}
|
||||
}
|
||||
scope := make(map[string]struct{}, len(objective.Scope.ResourceIDs))
|
||||
for _, resourceID := range objective.Scope.ResourceIDs {
|
||||
scope[canonicalPatrolScopeToken(resourceID)] = struct{}{}
|
||||
}
|
||||
_, ownerInScope := scope[canonicalPatrolScopeToken(ownerID)]
|
||||
_, linkedInScope := scope[canonicalPatrolScopeToken(check.LinkedResourceID)]
|
||||
if !ownerInScope && !linkedInScope {
|
||||
return &patrolObserverValidationError{code: "observer_availability_target_out_of_scope"}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func decodeStrictJSONObject(data []byte, target interface{}) error {
|
||||
decoder := json.NewDecoder(bytes.NewReader(data))
|
||||
decoder.DisallowUnknownFields()
|
||||
|
|
@ -241,7 +326,7 @@ func decodeStrictJSONObject(data []byte, target interface{}) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (p *PatrolService) evaluateObjectiveObserver(runtime *patrolObserverRuntime, objective PatrolObjective, probe patrolResourceStateProbe, now time.Time) *patrolObserverHealthUpdate {
|
||||
func (p *PatrolService) evaluateObjectiveObserver(runtime *patrolObserverRuntime, objective PatrolObjective, probe patrolValidatedObserverProbe, state patrolRuntimeState, now time.Time) *patrolObserverHealthUpdate {
|
||||
observer := objective.Observer
|
||||
key := fmt.Sprintf("%s/%d", observer.ID, observer.Version)
|
||||
runtime.mu.Lock()
|
||||
|
|
@ -250,20 +335,31 @@ func (p *PatrolService) evaluateObjectiveObserver(runtime *patrolObserverRuntime
|
|||
runtime.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
interval := time.Duration(probe.SampleIntervalSeconds) * time.Second
|
||||
interval := time.Duration(probe.sampleIntervalSeconds) * time.Second
|
||||
execution.nextDue = now.Add(interval)
|
||||
runtime.mu.Unlock()
|
||||
|
||||
statuses := patrolRuntimeCanonicalStatuses(p.currentPatrolRuntimeState())
|
||||
failing := make([]string, 0)
|
||||
for _, resourceID := range objective.Scope.ResourceIDs {
|
||||
status, exists := statuses[canonicalPatrolScopeToken(resourceID)]
|
||||
matched := exists && string(status) == probe.Value
|
||||
if probe.Operator == "not_equals" {
|
||||
matched = exists && string(status) != probe.Value
|
||||
if probe.runtime == patrolObserverAvailabilityFormat {
|
||||
check, ownerID, exists := patrolAvailabilityCheckByTarget(state, probe.targetID)
|
||||
matched := exists && check.LastChecked != nil && strings.EqualFold(check.ProbeOutcome, probe.value)
|
||||
if probe.operator == "not_equals" {
|
||||
matched = exists && check.LastChecked != nil && !strings.EqualFold(check.ProbeOutcome, probe.value)
|
||||
}
|
||||
if !matched {
|
||||
failing = append(failing, resourceID)
|
||||
failing = append(failing, ownerID)
|
||||
}
|
||||
} else {
|
||||
statuses := patrolRuntimeCanonicalStatuses(state)
|
||||
for _, resourceID := range objective.Scope.ResourceIDs {
|
||||
status, exists := statuses[canonicalPatrolScopeToken(resourceID)]
|
||||
matched := exists && string(status) == probe.value
|
||||
if probe.operator == "not_equals" {
|
||||
matched = exists && string(status) != probe.value
|
||||
}
|
||||
if !matched {
|
||||
failing = append(failing, resourceID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -291,7 +387,7 @@ func (p *PatrolService) evaluateObjectiveObserver(runtime *patrolObserverRuntime
|
|||
}
|
||||
runtime.mu.Unlock()
|
||||
}
|
||||
shouldWake := len(failing) > 0 && execution.consecutiveFailures >= probe.WakeAfterConsecutiveFailures &&
|
||||
shouldWake := len(failing) > 0 && execution.consecutiveFailures >= probe.wakeAfterConsecutiveFailures &&
|
||||
(!execution.wakeEmitted || (!execution.deliveryConfirmed && now.Sub(execution.lastWakeAt) >= patrolObserverWakeRetryInterval))
|
||||
|
||||
leaseDuration := 3 * interval
|
||||
|
|
@ -316,10 +412,7 @@ func (p *PatrolService) evaluateObjectiveObserver(runtime *patrolObserverRuntime
|
|||
return healthUpdate
|
||||
}
|
||||
sort.Strings(failing)
|
||||
evidence := fmt.Sprintf(
|
||||
"Canonical resource status did not satisfy %s %s for %d of %d scoped resources after %d consecutive local samples.",
|
||||
probe.Operator, probe.Value, len(failing), len(objective.Scope.ResourceIDs), execution.consecutiveFailures,
|
||||
)
|
||||
evidence := patrolObserverEvidence(probe, failing, len(objective.Scope.ResourceIDs), execution.consecutiveFailures)
|
||||
scope := PatrolScope{
|
||||
ResourceIDs: failing,
|
||||
Depth: PatrolDepthQuick,
|
||||
|
|
@ -355,6 +448,34 @@ func (p *PatrolService) evaluateObjectiveObserver(runtime *patrolObserverRuntime
|
|||
return healthUpdate
|
||||
}
|
||||
|
||||
func patrolObserverEvidence(probe patrolValidatedObserverProbe, failing []string, scopedResources, consecutiveFailures int) string {
|
||||
if probe.runtime == patrolObserverAvailabilityFormat {
|
||||
return fmt.Sprintf(
|
||||
"Canonical availability target %s did not satisfy %s %s after %d consecutive local samples.",
|
||||
probe.targetID, probe.operator, probe.value, consecutiveFailures,
|
||||
)
|
||||
}
|
||||
return fmt.Sprintf(
|
||||
"Canonical resource status did not satisfy %s %s for %d of %d scoped resources after %d consecutive local samples.",
|
||||
probe.operator, probe.value, len(failing), scopedResources, consecutiveFailures,
|
||||
)
|
||||
}
|
||||
|
||||
func patrolAvailabilityCheckByTarget(state patrolRuntimeState, targetID string) (unifiedresources.AvailabilityData, string, bool) {
|
||||
targetID = strings.TrimSpace(targetID)
|
||||
if targetID == "" || state.unifiedResourceProvider == nil {
|
||||
return unifiedresources.AvailabilityData{}, "", false
|
||||
}
|
||||
for _, resource := range state.unifiedResourceProvider.GetAll() {
|
||||
for _, check := range unifiedresources.AvailabilityChecksForResource(resource) {
|
||||
if strings.TrimSpace(check.TargetID) == targetID {
|
||||
return check, resource.ID, true
|
||||
}
|
||||
}
|
||||
}
|
||||
return unifiedresources.AvailabilityData{}, "", false
|
||||
}
|
||||
|
||||
func (p *PatrolService) objectiveWakeDelivered(objective PatrolObjective, lastWakeAt time.Time) bool {
|
||||
if p == nil || lastWakeAt.IsZero() || objective.Observer == nil || p.runHistoryStore == nil {
|
||||
return false
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ import (
|
|||
"time"
|
||||
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/models"
|
||||
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
|
||||
)
|
||||
|
||||
func createInstallablePatrolObserver(t *testing.T, store *PatrolObjectiveStore, now time.Time, resourceIDs ...string) PatrolObjective {
|
||||
|
|
@ -115,6 +116,121 @@ func TestPatrolObserverRuntimeRecordsExplicitUnsupportedValidationReason(t *test
|
|||
}
|
||||
}
|
||||
|
||||
func TestPatrolAvailabilityObserverUsesCanonicalScopedTargetAndWakesOnOutcomeBreach(t *testing.T) {
|
||||
now := time.Date(2026, 8, 14, 3, 0, 0, 0, time.UTC)
|
||||
store := NewInMemoryPatrolObjectiveStore()
|
||||
objective, err := store.Create(CreatePatrolObjectiveInput{
|
||||
Brief: "Keep the front cameras reachable",
|
||||
ResourceIDs: []string{"frigate-1"},
|
||||
}, now)
|
||||
if err != nil {
|
||||
t.Fatalf("create objective: %v", err)
|
||||
}
|
||||
objective, err = store.ProposeObserver(objective.ID, ProposePatrolObserverInput{
|
||||
ExpectedRevision: objective.Revision,
|
||||
Interpretation: "The existing camera availability check remains reachable.",
|
||||
TriggerKinds: []PatrolObserverTriggerKind{PatrolObserverTriggerInterval},
|
||||
ProbeJSON: `{"runtime":"pulse-availability-state/v1","target_id":"camera-front-http","path":"probe_outcome","operator":"equals","value":"reachable","sample_interval_seconds":10,"wake_after_consecutive_failures":2}`,
|
||||
WakeEvidence: "The canonical camera endpoint is not reachable twice.",
|
||||
RequirementsJSON: `{}`,
|
||||
}, now)
|
||||
if err != nil {
|
||||
t.Fatalf("propose availability observer: %v", err)
|
||||
}
|
||||
|
||||
checkedAt := now
|
||||
outcome := "reachable"
|
||||
provider := &mockUnifiedResourceProvider{getAllFunc: func() []unifiedresources.Resource {
|
||||
return []unifiedresources.Resource{{
|
||||
ID: "frigate-1", Type: unifiedresources.ResourceTypeAppContainer,
|
||||
AvailabilityChecks: []unifiedresources.AvailabilityData{{
|
||||
TargetID: "camera-front-http", LinkedResourceID: "frigate-1", Enabled: true,
|
||||
ProbeOutcome: outcome, LastChecked: &checkedAt,
|
||||
}},
|
||||
}}
|
||||
}}
|
||||
patrol := NewPatrolService(nil, nil)
|
||||
patrol.SetObjectiveStore(store)
|
||||
patrol.SetUnifiedResourceProvider(provider)
|
||||
tm := NewTriggerManager(TriggerManagerConfig{MaxPendingTriggers: 10})
|
||||
patrol.SetTriggerManager(tm)
|
||||
|
||||
patrol.processObjectiveObservers(now.Add(time.Second))
|
||||
installed, _ := store.Get(objective.ID, now.Add(time.Second))
|
||||
if installed.Observer == nil || installed.Observer.State != PatrolObserverInstalled || installed.Coverage.State != PatrolObjectiveCovered {
|
||||
t.Fatalf("availability observer was not installed: observer=%+v coverage=%+v", installed.Observer, installed.Coverage)
|
||||
}
|
||||
if tm.GetPendingCount() != 0 {
|
||||
t.Fatal("reachable availability target woke Patrol")
|
||||
}
|
||||
|
||||
outcome = "unreachable"
|
||||
checkedAt = now.Add(11 * time.Second)
|
||||
patrol.processObjectiveObservers(now.Add(11 * time.Second))
|
||||
patrol.processObjectiveObservers(now.Add(21 * time.Second))
|
||||
if tm.GetPendingCount() != 1 {
|
||||
t.Fatalf("availability breach did not wake Patrol once; pending=%d", tm.GetPendingCount())
|
||||
}
|
||||
queued := tm.pendingTriggers[0]
|
||||
if queued.ObjectiveContext == nil || len(queued.ObjectiveContext.ObservedResourceIDs) != 1 || queued.ObjectiveContext.ObservedResourceIDs[0] != "frigate-1" {
|
||||
t.Fatalf("availability wake lost canonical owner context: %+v", queued.ObjectiveContext)
|
||||
}
|
||||
if !strings.Contains(queued.ObjectiveContext.Evidence, "camera-front-http") || !strings.Contains(queued.ObjectiveContext.Evidence, "equals reachable") {
|
||||
t.Fatalf("availability wake evidence = %q", queued.ObjectiveContext.Evidence)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPatrolAvailabilityObserverBindingFailsClosed(t *testing.T) {
|
||||
now := time.Date(2026, 8, 14, 3, 0, 0, 0, time.UTC)
|
||||
checkedAt := now
|
||||
for _, test := range []struct {
|
||||
name string
|
||||
resourceID string
|
||||
targetID string
|
||||
enabled bool
|
||||
wantCode string
|
||||
}{
|
||||
{name: "missing", resourceID: "camera-1", targetID: "different-target", enabled: true, wantCode: "observer_availability_target_missing"},
|
||||
{name: "disabled", resourceID: "camera-1", targetID: "camera-http", enabled: false, wantCode: "observer_availability_target_disabled"},
|
||||
{name: "out of scope", resourceID: "camera-2", targetID: "camera-http", enabled: true, wantCode: "observer_availability_target_out_of_scope"},
|
||||
} {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
store := NewInMemoryPatrolObjectiveStore()
|
||||
objective, err := store.Create(CreatePatrolObjectiveInput{Brief: "Keep camera online", ResourceIDs: []string{"camera-1"}}, now)
|
||||
if err != nil {
|
||||
t.Fatalf("create objective: %v", err)
|
||||
}
|
||||
objective, err = store.ProposeObserver(objective.ID, ProposePatrolObserverInput{
|
||||
ExpectedRevision: objective.Revision,
|
||||
Interpretation: "Keep the endpoint reachable.",
|
||||
TriggerKinds: []PatrolObserverTriggerKind{PatrolObserverTriggerInterval},
|
||||
ProbeJSON: `{"runtime":"pulse-availability-state/v1","target_id":"camera-http","path":"probe_outcome","operator":"equals","value":"reachable","sample_interval_seconds":30,"wake_after_consecutive_failures":2}`,
|
||||
WakeEvidence: "Endpoint reachability breached.", RequirementsJSON: `{}`,
|
||||
}, now)
|
||||
if err != nil {
|
||||
t.Fatalf("propose observer: %v", err)
|
||||
}
|
||||
provider := &mockUnifiedResourceProvider{getAllFunc: func() []unifiedresources.Resource {
|
||||
return []unifiedresources.Resource{{
|
||||
ID: test.resourceID,
|
||||
AvailabilityChecks: []unifiedresources.AvailabilityData{{
|
||||
TargetID: test.targetID, Enabled: test.enabled, ProbeOutcome: "reachable", LastChecked: &checkedAt,
|
||||
}},
|
||||
}}
|
||||
}}
|
||||
patrol := NewPatrolService(nil, nil)
|
||||
patrol.SetObjectiveStore(store)
|
||||
patrol.SetUnifiedResourceProvider(provider)
|
||||
patrol.processObjectiveObservers(now.Add(time.Second))
|
||||
|
||||
got, _ := store.Get(objective.ID, now.Add(time.Second))
|
||||
if got.Observer == nil || got.Observer.State != PatrolObserverRejected || got.Observer.FailureCode != test.wantCode {
|
||||
t.Fatalf("binding result = %+v, want rejected/%s", got.Observer, test.wantCode)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPatrolObserverRuntimeWakesModelOnlyAfterLocalFailureWindow(t *testing.T) {
|
||||
now := time.Date(2026, 8, 14, 1, 0, 0, 0, time.UTC)
|
||||
store := NewInMemoryPatrolObjectiveStore()
|
||||
|
|
@ -227,6 +343,31 @@ func TestValidatePatrolObserverArtifactRejectsUnknownExecutableFields(t *testing
|
|||
}
|
||||
}
|
||||
|
||||
func TestValidatePatrolAvailabilityObserverRejectsModelSuppliedNetworkAuthority(t *testing.T) {
|
||||
now := time.Now().UTC()
|
||||
store := NewInMemoryPatrolObjectiveStore()
|
||||
objective, err := store.Create(CreatePatrolObjectiveInput{Brief: "Keep camera reachable", ResourceIDs: []string{"camera-1"}}, now)
|
||||
if err != nil {
|
||||
t.Fatalf("create objective: %v", err)
|
||||
}
|
||||
objective, err = store.ProposeObserver(objective.ID, ProposePatrolObserverInput{
|
||||
ExpectedRevision: objective.Revision,
|
||||
Interpretation: "Keep the endpoint reachable.",
|
||||
TriggerKinds: []PatrolObserverTriggerKind{PatrolObserverTriggerInterval},
|
||||
ProbeJSON: `{"runtime":"pulse-availability-state/v1","target_id":"camera-http","path":"probe_outcome","operator":"equals","value":"reachable","sample_interval_seconds":30,"wake_after_consecutive_failures":2}`,
|
||||
WakeEvidence: "Endpoint reachability breached.", RequirementsJSON: `{}`,
|
||||
}, now)
|
||||
if err != nil {
|
||||
t.Fatalf("propose observer: %v", err)
|
||||
}
|
||||
artifact, _ := store.GetObserverArtifact(objective.ID)
|
||||
artifact.Probe = []byte(`{"runtime":"pulse-availability-state/v1","target_id":"camera-http","path":"probe_outcome","operator":"equals","value":"reachable","sample_interval_seconds":30,"wake_after_consecutive_failures":2,"url":"http://169.254.169.254/latest/meta-data"}`)
|
||||
_, validationErr := validatePatrolObserverArtifact(objective, artifact)
|
||||
if validationErr == nil || validationErr.code != "observer_probe_invalid" {
|
||||
t.Fatalf("model-supplied URL validation error = %v", validationErr)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPatrolObjectiveIntentEditInvalidatesInstalledObserver(t *testing.T) {
|
||||
now := time.Date(2026, 8, 14, 1, 0, 0, 0, time.UTC)
|
||||
store := NewInMemoryPatrolObjectiveStore()
|
||||
|
|
|
|||
|
|
@ -231,7 +231,7 @@ Returns a list of active findings with their IDs, severity, resource, and title.
|
|||
|
||||
Use this only when the objective context says observer_missing, or when current evidence clearly requires a new observer version. Translate the operator's outcome into the smallest useful local observer without hard-coding an application into Pulse. The probe_json and requirements_json fields must each be one bounded JSON object. Do not include mutation commands, credentials, or secret values.
|
||||
|
||||
Core can currently install one generic local ABI for objectives scoped to canonical Pulse resources: trigger_kind must be interval, requirements_json must be {}, and probe_json must be exactly {"runtime":"pulse-resource-state/v1","path":"status","operator":"equals","value":"online","sample_interval_seconds":30,"wake_after_consecutive_failures":2}. The operator may be equals or not_equals; value may be online, offline, warning, or unknown; interval is 10-300 seconds and the failure window is 1-10 samples. Use this ABI when canonical resource status is a truthful interpretation. If the outcome needs an app API, event, log, file, socket, network, filesystem, secret, or richer signal, describe that honest proposal instead; core will retain it with an explicit unsupported validation reason rather than pretending it is active.
|
||||
Core can currently install two generic local ABIs. Both require trigger_kind interval, requirements_json {}, a 10-300 second interval, and a 1-10 sample failure window. Use pulse-resource-state/v1 when canonical resource status is truthful: {"runtime":"pulse-resource-state/v1","path":"status","operator":"equals","value":"online","sample_interval_seconds":30,"wake_after_consecutive_failures":2}; operator may be equals or not_equals and value may be online, offline, warning, or unknown. Use pulse-availability-state/v1 only for an enabled canonical target ID shown in the objective's local availability signals: {"runtime":"pulse-availability-state/v1","target_id":"the-exact-target-id","path":"probe_outcome","operator":"equals","value":"reachable","sample_interval_seconds":30,"wake_after_consecutive_failures":2}; value may be reachable, unreachable, or indeterminate. Core proves that target already exists and belongs to the objective scope; never invent a target ID or put an address or credential in the artifact. If the outcome needs a new app API, event, log, file, socket, network target, filesystem signal, secret, or richer signal, describe that honest proposal instead; core will retain it with an explicit unsupported validation reason rather than pretending it is active.
|
||||
|
||||
This tool records only a versioned proposed artifact. It does not validate, install, execute, or claim coverage. Core owns the observer ID, version, SHA-256 digest, read-only posture, sandboxing, installation, health lease, and any later transition.
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue