Serialize Patrol action projection and finalize hung runs

Three gaps remained in how Pulse projects a Patrol investigation's
action and closes its runs.

Projections of one investigation could interleave: the transition
callback, an investigation read and a run's close each read the action
audit and applied it, so one that read an older state could land after
one that read a newer state, and an approval could land between a
run's reconciliation and its finalization. All of them now read and
apply under the investigation's lock, and a close finalizes before it
releases it.

The projection replaced the whole session through Update, so it could
undo run progress the orchestrator saved in between. It now writes
only the changed action and outcome when the store offers SetAction
(aicontracts.InvestigationActionStore), and the reference it writes
carries the broker's canonical result and verification status and the
model's causal attribution, which the audit does not hold.

A run that hung past its limit without closing was recovered by the
stuck-run sweep but never finalized. The sweep now finalizes it with a
copy of its session marked failed, leaving the orchestrator's session
as it is. A later real close is finalized instead, and a refresh no
longer reopens a record that finalization closed.
This commit is contained in:
rcourtman
2026-10-23 22:26:09 +01:00
parent 7b39194ae2
commit a4bfca4d32
10 changed files with 492 additions and 40 deletions
@@ -10210,3 +10210,8 @@ runs when they close, failed runs included, and publishes the action plans the
inbox expiry sweep expires, enrols, commands, updates or removes no agent and
touches no agent identity or credential. Investigations still reach hosts only
through the read-only evidence tools and the governed action lifecycle.
The `internal/api` change of 2026-10-23 that serializes the projection of a
Patrol investigation's action and finalizes Patrol runs that ran past their
limit enrols, commands, updates or removes no agent and touches no agent
identity or credential.
@@ -7635,3 +7635,30 @@ refresh plans a replacement under the same investigation and leaves the
replaced plan to expire on its own, and that expiry must not take the
investigation back from its replacement. Proof:
`internal/api/patrol_action_expiry_sweep_test.go`.
### Patrol action projection is serialized and hung runs are finalized (23 October 2026)
Every projection of a Patrol investigation's canonical action reads the action
audit and applies it under that investigation's lock: the lifecycle's
transition callback (`ReconcilePatrolActionTransition`), a read that hydrates
the investigation, and a run's close (`closePatrolInvestigationRun`), which
also finalizes the run before it releases the lock. A projection read from an
older state therefore cannot land after one read from a newer state, and no
approval or rejection lands between a run's reconciliation and its
finalization. The projection writes only the changed action and outcome when
the investigation store offers `aicontracts.InvestigationActionStore`, so it
cannot undo run progress the orchestrator saved after the projection read the
session; the orchestrator in turn saves its progress without replacing the
action Pulse recorded. The action reference carries the canonical result and
verification status the broker reports (`dispositionFromRecord`) and keeps the
model's causal attribution, which the action audit does not hold.
The stuck-run sweep now also finalizes a run that ran past its limit without
closing, with a copy of its session marked failed at the sweep ("Patrol stopped
waiting for this investigation after it ran past its time limit."). The
orchestrator's own session is left as it is. If the run closes later, its own
close is finalized, and a close finalizes the stored session rather than the
copy once the orchestrator has closed it. A refresh of the record from a
session that has not closed no longer reopens a record that finalization
closed. Proofs: `internal/api/ai_handlers_investigation_additional_test.go` and
`internal/ai/patrol_investigation_finalize_test.go`.
@@ -12545,3 +12545,18 @@ changes. Proofs: `internal/actionlifecycle/service_test.go`
`TestRefreshedActionIDNamesTheReplacedAction`),
`internal/api/patrol_action_expiry_sweep_test.go` and
`internal/api/ai_handlers_investigation_finalize_test.go`.
### Field-level action writes and serialized Patrol action projection (23 October 2026)
`aicontracts.InvestigationActionStore` is a new optional interface on the
Pulse–enterprise boundary: an investigation store that offers
`SetAction(id, action)` lets Pulse record an action lifecycle transition
without replacing the session. Pulse's Patrol action projection uses it with
`SetOutcome` when the store offers it and falls back to `Update` for a store
without it, so an older store keeps working. Projections of one investigation,
and a run's close, are serialized under that investigation's lock. The
investigation's action reference now includes `action_result_v2` and
`verification_status`, the projection the broker's disposition already
carries, and keeps `causal_resource_id` for the same action, so investigation
payloads may carry these optional fields where they were empty before. No route
changes. Proof: `internal/api/ai_handlers_investigation_additional_test.go`.
@@ -7198,3 +7198,7 @@ The `internal/api` change of 2026-10-23 that finalizes Patrol investigation
runs when they close and publishes the action plans the inbox expiry sweep
expires opens no storage, backup or recovery path. Recovery evidence, restore
rules and storage capabilities are unchanged.
The `internal/api` change of 2026-10-23 that serializes the projection of a
Patrol investigation's action and finalizes Patrol runs that ran past their
limit opens no storage, backup or recovery path.
+30 -5
View File
@@ -1594,9 +1594,11 @@ func (p *PatrolService) recoverStuckInvestigations() {
Str("resource", f.ResourceName).
Str("outcome", string(result.outcome)).
Msg("AI Patrol: Recovered stuck investigation")
// A run whose session closed without its close reaching the finding,
// such as one a restart interrupted after it planned a fix, is
// finalized here. A run still going finalizes when its close lands.
// The recovered run is finalized here: a run whose session closed
// without its close reaching the finding, such as one a restart
// interrupted after it planned a fix, and a run that ran past its limit
// without closing. If that run does close later, its close lands and
// finalizes it again with its own result.
if result.session != nil {
p.closeInvestigationRun(f.ID, result.session)
}
@@ -1611,7 +1613,8 @@ func (p *PatrolService) recoverStuckInvestigations() {
// runStartedAt is the start of the run's session, zero when the orchestrator
// holds none, so the store can check the run belongs to the finding's current
// occurrence before it applies the session's result. session is the run's
// session once it has closed, which finalizes the run after recovery.
// session as it is finalized after recovery: closed, or failed as of the sweep
// for a run that ran past its limit.
type stuckInvestigation struct {
status InvestigationStatus
outcome InvestigationOutcome
@@ -1653,6 +1656,7 @@ func stuckInvestigationResult(f *Finding, orchestrator InvestigationOrchestrator
if now.Before(limit) {
return stuckInvestigation{}, false
}
closed.session = abandonedInvestigationSession(session, now)
return closed, true
}
closed.session = session
@@ -1679,6 +1683,22 @@ func stuckInvestigationResult(f *Finding, orchestrator InvestigationOrchestrator
return closed, true
}
// abandonedRunError is the error a run the stuck-run sweep gave up on shows.
const abandonedRunError = "Patrol stopped waiting for this investigation after it ran past its time limit."
// abandonedInvestigationSession is a run's session as the stuck-run sweep gives
// the run up at now: failed, because it ran past its limit without closing.
// The orchestrator's own session is left as it is.
func abandonedInvestigationSession(session *InvestigationSession, now time.Time) *InvestigationSession {
abandoned := *session
abandoned.Status = aicontracts.InvestigationStatusFailed
abandoned.CompletedAt = &now
if strings.TrimSpace(abandoned.Error) == "" {
abandoned.Error = abandonedRunError
}
return &abandoned
}
// investigationStoreProvider is implemented by orchestrators that expose their
// session store, which holds every recorded run of a finding.
type investigationStoreProvider interface {
@@ -2025,7 +2045,12 @@ func (p *PatrolService) storeFindingInvestigationRecord(findingID string, sessio
// may no longer retain all of that original context after restart.
if previous := finding.InvestigationRecord; preserveEvidence && previous != nil && previous.ID == record.ID {
retained := previous.NormalizeCollections()
retained.Status = record.Status
// A record its run's close finalized stays closed: a later projection
// of the action, read from a session the orchestrator has not closed,
// such as one the stuck-run sweep gave up on, must not reopen it.
if !investigationRecordStatusIsTerminal(retained.Status) || investigationRecordStatusIsTerminal(record.Status) {
retained.Status = record.Status
}
retained.Outcome = record.Outcome
retained.Action = record.Action
retained.Verification = record.Verification
@@ -126,6 +126,32 @@ func TestCompleteInvestigationKeepsAnOutcomeTheActionReached(t *testing.T) {
assertInvestigationState(t, findings, "unplanned", InvestigationStatusFailed, InvestigationOutcomeTimedOut)
}
// A projection of the action that reads a session the orchestrator has not
// closed, such as one the stuck-run sweep gave up on, records the action's new
// state without reopening the record that run's finalization closed.
func TestRefreshDoesNotReopenAFinalizedRecord(t *testing.T) {
run := newFinalizedRun(t, InvestigationOutcomeFixQueued)
closed := time.Now()
running := &InvestigationSession{ID: "investigation-1", FindingID: "finding-1", SessionID: "chat-1",
Status: aicontracts.InvestigationStatusRunning, StartedAt: closed.Add(-time.Hour),
Action: &aicontracts.ActionReference{ActionID: "action-1", State: "pending_approval"}}
abandoned := *running
abandoned.Status = aicontracts.InvestigationStatusFailed
abandoned.CompletedAt = &closed
abandoned.Error = abandonedRunError
run.patrol.FinalizeInvestigationRun("finding-1", &abandoned)
approved := *running
approved.Action = &aicontracts.ActionReference{ActionID: "action-1", State: "approved"}
run.patrol.RefreshFindingInvestigationRecord("finding-1", &approved)
record := run.store.Get("finding-1").InvestigationRecord
if record == nil || record.Status != aicontracts.InvestigationStatusFailed || record.Error != abandonedRunError ||
record.CompletedAt == nil || record.Action == nil || record.Action.State != "approved" {
t.Fatalf("record = %#v, want it still closed, with the approval recorded", record)
}
}
// An action's outcome is pushed when its transition is reconciled
// (PublishFindingLifecycleUpdate), so a run that closes on a verified fix is
// published without a second push.
@@ -143,11 +169,12 @@ func TestFinalizeInvestigationRunLeavesActionOutcomePushesToReconciliation(t *te
}
}
// The stuck-run sweep finalizes a run it recovers once the run's session has
// closed without its close reaching the finding, such as a run a restart
// interrupted after it planned a fix. A run still going past its limit is not
// finalized: it finalizes when its own close lands.
func TestRecoverStuckInvestigationsFinalizesAClosedRun(t *testing.T) {
// The stuck-run sweep finalizes each run it recovers: one whose session closed
// without its close reaching the finding, such as a run a restart interrupted
// after it planned a fix, and one still going past its limit, as failed. The
// orchestrator's own session for a run past its limit is left as it is, so the
// run's real close can still land and finalize it again.
func TestRecoverStuckInvestigationsFinalizesTheRunsItRecovers(t *testing.T) {
now := time.Now()
closedAt := now.Add(-time.Second)
findings := NewFindingsStore()
@@ -164,16 +191,26 @@ func TestRecoverStuckInvestigationsFinalizesAClosedRun(t *testing.T) {
"hung": {ID: "inv-hung", FindingID: "hung", SessionID: "chat-hung", Status: aicontracts.InvestigationStatusRunning, StartedAt: now.Add(-time.Hour)},
}}
patrol := &PatrolService{findings: findings, investigationOrchestrator: orchestrator}
var closed []string
closed := map[string]*InvestigationSession{}
patrol.SetInvestigationRunCloser(func(findingID string, session *InvestigationSession) {
closed = append(closed, findingID+"/"+session.SessionID)
closed[findingID] = session
})
patrol.recoverStuckInvestigations()
assertInvestigationState(t, findings, "restarted", InvestigationStatusCompleted, InvestigationOutcomeFixQueued)
assertInvestigationState(t, findings, "hung", InvestigationStatusFailed, InvestigationOutcomeTimedOut)
if len(closed) != 1 || closed[0] != "restarted/chat-restarted" {
t.Fatalf("closed runs = %v, want only the restarted run's closed session", closed)
if len(closed) != 2 {
t.Fatalf("closed runs = %v, want both recovered runs", closed)
}
if restarted := closed["restarted"]; restarted == nil || restarted.SessionID != "chat-restarted" || restarted.Outcome != aicontracts.OutcomeInterrupted {
t.Fatalf("restarted run closed with %+v, want its closed session", restarted)
}
if hung := closed["hung"]; hung == nil || hung.SessionID != "chat-hung" || hung.Status != aicontracts.InvestigationStatusFailed ||
hung.CompletedAt == nil || hung.Error != abandonedRunError {
t.Fatalf("hung run closed with %+v, want a failed session past its limit", hung)
}
if own := orchestrator.sessions["hung"]; own.Status != aicontracts.InvestigationStatusRunning || own.CompletedAt != nil {
t.Fatalf("the sweep changed the orchestrator's own session: %+v", own)
}
}
+5 -4
View File
@@ -1932,11 +1932,12 @@ func (h *AISettingsHandler) setupInvestigationOrchestrator(orgID string, svc *ai
log.Warn().Str("orgID", orgID).Msg("Findings store not available for orchestrator")
return
}
// A closed run's action is reconciled from the action audit before the run
// is finalized: the orchestrator writes the action back as it was planned,
// which an approval or rejection during the run may have overtaken.
// A closed run's action is reconciled from the action audit, and the run
// finalized, under the investigation's lock (closePatrolInvestigationRun).
closeRun := func(findingID string, session *aicontracts.InvestigationSession) {
patrol.FinalizeInvestigationRun(findingID, h.hydratePatrolInvestigationAction(orgID, session))
h.closePatrolInvestigationRun(orgID, session, func(closed *aicontracts.InvestigationSession) {
patrol.FinalizeInvestigationRun(findingID, closed)
})
}
patrol.SetInvestigationRunCloser(closeRun)
findingsAdapter := &orchestratorFindingsAdapter{
@@ -1143,3 +1143,210 @@ func TestOrchestratorRunResultSkipsANewerRun(t *testing.T) {
}
assertOrchestratorRunState(t, findings, "retried", "chat-retried-b", aicontracts.InvestigationStatusFailed, aicontracts.OutcomeTimedOut, 2)
}
// actionRecordingStore records an investigation's action on its own, as the
// enterprise store does, and counts how often a whole session is replaced.
type actionRecordingStore struct {
*testInvestigationStore
replaced int
}
func (s *actionRecordingStore) Update(session *aicontracts.InvestigationSession) bool {
s.replaced++
return s.testInvestigationStore.Update(session)
}
func (s *actionRecordingStore) SetAction(id string, action *aicontracts.ActionReference) bool {
s.mu.Lock()
defer s.mu.Unlock()
session, ok := s.sessions[id]
if !ok {
return false
}
session.Action = aicontracts.CloneActionReference(action)
return true
}
// patrolProjectionFixture is a running Patrol investigation, the action audit
// store its plans live in, and the handler that projects them onto it.
type patrolProjectionFixture struct {
handler *AISettingsHandler
audits *unifiedresources.MemoryStore
sessions *actionRecordingStore
investigation *aicontracts.InvestigationSession
}
func newPatrolProjectionFixture(t *testing.T) patrolProjectionFixture {
t.Helper()
svc := ai.NewService(nil, nil)
svc.SetStateProvider(&MockStateProvider{})
svc.GetPatrolService().GetFindings().Add(&ai.Finding{ID: "finding-1", ResourceID: "vm:42", Title: "Unhealthy service",
Severity: ai.FindingSeverityWarning, InvestigationSessionID: "session-1",
InvestigationStatus: string(aicontracts.InvestigationStatusRunning)})
sessions := &actionRecordingStore{testInvestigationStore: newTestInvestigationStore()}
investigation := sessions.Create("finding-1", "session-1")
audits := unifiedresources.NewMemoryStore()
handler := &AISettingsHandler{defaultAIService: svc,
investigationStores: map[string]aicontracts.InvestigationStore{"default": sessions},
resourceStoreProvider: func(string) (unifiedresources.ResourceStore, error) { return audits, nil },
}
return patrolProjectionFixture{handler: handler, audits: audits, sessions: sessions, investigation: investigation}
}
func (f patrolProjectionFixture) recordPlan(t *testing.T, audit unifiedresources.ActionAuditRecord) unifiedresources.ActionAuditRecord {
t.Helper()
if err := f.audits.RecordActionAudit(audit); err != nil {
t.Fatalf("RecordActionAudit(%s): %v", audit.ID, err)
}
recorded, found, err := f.audits.GetActionAudit(audit.ID)
if err != nil || !found {
t.Fatalf("GetActionAudit(%s): found=%v err=%v", audit.ID, found, err)
}
return recorded
}
// A projection records the action and the outcome on their own when the store
// can, so it never replaces the session and cannot undo progress the
// orchestrator saved after the projection read it.
func TestPatrolActionProjectionRecordsOnlyTheActionAndOutcome(t *testing.T) {
f := newPatrolProjectionFixture(t)
now := time.Now().UTC().Truncate(time.Second)
plan := f.recordPlan(t, patrolPlanAudit("act-1", f.investigation.ID, now, now.Add(time.Hour)))
f.handler.ReconcilePatrolActionTransition("default", plan)
got := f.sessions.Get(f.investigation.ID)
if got.Action == nil || got.Action.ActionID != "act-1" || got.Outcome != aicontracts.OutcomeFixQueued {
t.Fatalf("session = %+v, want the planned action and its queued fix", got)
}
if f.sessions.replaced != 0 {
t.Fatalf("the projection replaced the whole session %d times", f.sessions.replaced)
}
}
// A projection waits for any other projection of the same investigation, so one
// that read an older state of the action cannot land after one that read a
// newer state. Other investigations do not wait for it.
func TestPatrolActionProjectionWaitsForTheInvestigationsLock(t *testing.T) {
f := newPatrolProjectionFixture(t)
now := time.Now().UTC().Truncate(time.Second)
plan := f.recordPlan(t, patrolPlanAudit("act-1", f.investigation.ID, now, now.Add(time.Hour)))
release := lockPatrolInvestigation("default", f.investigation.ID)
done := make(chan struct{})
go func() {
f.handler.ReconcilePatrolActionTransition("default", plan)
close(done)
}()
select {
case <-done:
release()
t.Fatal("a projection ran while another held the investigation's lock")
case <-time.After(100 * time.Millisecond):
}
lockPatrolInvestigation("default", "another-investigation")()
release()
select {
case <-done:
case <-time.After(5 * time.Second):
t.Fatal("the projection did not run once the lock was free")
}
if got := f.sessions.Get(f.investigation.ID); got.Action == nil || got.Action.ActionID != "act-1" {
t.Fatalf("session = %+v, want the action projected once the lock was free", got)
}
}
// Hydration returns the action and outcome the investigation's store holds after
// the repair, not a projection of the caller's copy. After a refresh replaced
// the plan, a copy taken before the refresh must not bring back the replaced
// plan, here expired.
func TestPatrolActionHydrationUsesTheInvestigationsCurrentAction(t *testing.T) {
f := newPatrolProjectionFixture(t)
now := time.Now().UTC().Truncate(time.Second)
f.recordPlan(t, patrolPlanAudit("act-1", f.investigation.ID, now.Add(-10*time.Minute), now.Add(-5*time.Minute)))
replacement := patrolPlanAudit("act-2", f.investigation.ID, now.Add(-time.Minute), now.Add(time.Hour))
replacement.Request.RequestID = "refresh:act-1"
replacement.Plan.RequestID = "refresh:act-1"
f.recordPlan(t, replacement)
if expired, err := f.audits.ExpireActionAudits(now, 10); err != nil || len(expired) != 1 || expired[0].ID != "act-1" {
t.Fatalf("expired = %+v err=%v, want the replaced plan", expired, err)
}
f.sessions.SetAction(f.investigation.ID, &aicontracts.ActionReference{ActionID: "act-2", State: string(unifiedresources.ActionStatePending)})
f.sessions.SetOutcome(f.investigation.ID, aicontracts.OutcomeFixQueued)
stale := *f.sessions.Get(f.investigation.ID)
stale.Action = &aicontracts.ActionReference{ActionID: "act-1", State: string(unifiedresources.ActionStatePending)}
got := f.handler.hydratePatrolInvestigationAction("default", &stale)
if got.Action == nil || got.Action.ActionID != "act-2" || got.Outcome != aicontracts.OutcomeFixQueued {
t.Fatalf("hydrated = action %+v outcome %q, want the replacement still queued", got.Action, got.Outcome)
}
}
// A run's close finalizes, under the investigation's lock, the session the
// orchestrator closed: a copy the stuck-run sweep marked failed gives way to the
// run's own close once that has landed.
func TestPatrolRunCloseFinalizesTheClosedSessionUnderTheLock(t *testing.T) {
f := newPatrolProjectionFixture(t)
f.sessions.Complete(f.investigation.ID, aicontracts.OutcomeResolved, "The service recovered on its own.", nil)
abandoned := *f.sessions.Get(f.investigation.ID)
abandoned.Status = aicontracts.InvestigationStatusFailed
abandoned.Error = "Patrol stopped waiting for this investigation after it ran past its time limit."
var finalized *aicontracts.InvestigationSession
f.handler.closePatrolInvestigationRun("default", &abandoned, func(closed *aicontracts.InvestigationSession) {
finalized = closed
acquired := make(chan struct{})
go func() {
lockPatrolInvestigation("default", f.investigation.ID)()
close(acquired)
}()
select {
case <-acquired:
t.Error("a projection of the investigation ran while the run was being finalized")
case <-time.After(50 * time.Millisecond):
}
})
if finalized == nil || finalized.Status != aicontracts.InvestigationStatusCompleted || finalized.Outcome != aicontracts.OutcomeResolved || finalized.Error != "" {
t.Fatalf("finalized %+v, want the run's own completed close", finalized)
}
}
// A run the orchestrator still shows as running, which the stuck-run sweep gave
// up on, is finalized as the sweep marked it.
func TestPatrolRunCloseFinalizesAnAbandonedRunAsGivenUp(t *testing.T) {
f := newPatrolProjectionFixture(t)
f.sessions.UpdateStatus(f.investigation.ID, aicontracts.InvestigationStatusRunning)
abandoned := *f.sessions.Get(f.investigation.ID)
abandoned.Status = aicontracts.InvestigationStatusFailed
abandoned.Error = "Patrol stopped waiting for this investigation after it ran past its time limit."
var finalized *aicontracts.InvestigationSession
f.handler.closePatrolInvestigationRun("default", &abandoned, func(closed *aicontracts.InvestigationSession) {
finalized = closed
})
if finalized == nil || finalized.Status != aicontracts.InvestigationStatusFailed || finalized.Error != abandoned.Error {
t.Fatalf("finalized %+v, want the run as the sweep gave it up", finalized)
}
}
// The investigation keeps the action's canonical result and verification, as
// the broker reports them, and the model's causal attribution, which the
// action audit does not hold.
func TestPatrolActionProjectionKeepsTheFullActionReference(t *testing.T) {
f := newPatrolProjectionFixture(t)
now := time.Now().UTC().Truncate(time.Second)
plan := f.recordPlan(t, patrolPlanAudit("act-1", f.investigation.ID, now, now.Add(time.Hour)))
f.sessions.SetAction(f.investigation.ID, &aicontracts.ActionReference{ActionID: "act-1", State: string(unifiedresources.ActionStatePending), CausalResourceID: "storage:nfs-1"})
f.handler.ReconcilePatrolActionTransition("default", plan)
got := f.sessions.Get(f.investigation.ID).Action
want := dispositionFromRecord(plan)
if got == nil || got.CausalResourceID != "storage:nfs-1" || len(got.ActionResultV2) == 0 ||
string(got.ActionResultV2) != string(want.ActionResultV2) || got.VerificationStatus != want.VerificationStatus {
t.Fatalf("action = %+v, want the broker's result and verification with the causal attribution kept", got)
}
}
+145 -22
View File
@@ -5,6 +5,7 @@ import (
"fmt"
"reflect"
"strings"
"sync"
"github.com/rcourtman/pulse-go-rewrite/internal/actionlifecycle"
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
@@ -12,6 +13,53 @@ import (
"github.com/rs/zerolog/log"
)
// patrolInvestigationLocks serializes the projection of each Patrol
// investigation's action. A transition callback, a read that hydrates the
// investigation and a run's close each read the action audit and apply it under
// the investigation's lock, so a projection read from an older audit state
// cannot land after one read from a newer state.
var patrolInvestigationLocks keyedLocks
// keyedLocks is a mutex per key, held only while some caller uses it.
type keyedLocks struct {
mu sync.Mutex
locks map[string]*keyedLock
}
type keyedLock struct {
mu sync.Mutex
users int
}
// lock acquires key's mutex and returns its release.
func (k *keyedLocks) lock(key string) func() {
k.mu.Lock()
if k.locks == nil {
k.locks = make(map[string]*keyedLock)
}
entry := k.locks[key]
if entry == nil {
entry = &keyedLock{}
k.locks[key] = entry
}
entry.users++
k.mu.Unlock()
entry.mu.Lock()
return func() {
entry.mu.Unlock()
k.mu.Lock()
entry.users--
if entry.users == 0 {
delete(k.locks, key)
}
k.mu.Unlock()
}
}
func lockPatrolInvestigation(orgID, investigationID string) func() {
return patrolInvestigationLocks.lock(strings.TrimSpace(orgID) + "\x00" + strings.TrimSpace(investigationID))
}
// ReconcilePatrolActionTransition projects the authoritative canonical action
// audit onto the Patrol investigation and finding that originated it. The
// callback payload is only a wake-up signal: re-reading by action ID prevents
@@ -25,6 +73,7 @@ func (h *AISettingsHandler) ReconcilePatrolActionTransition(orgID string, transi
log.Warn().Err(err).Str("orgID", orgID).Str("actionID", transition.ID).Msg("Unable to reconcile Patrol action transition")
return
}
defer lockPatrolInvestigation(orgID, transition.Origin.InvestigationID)()
authoritative, found, err := store.GetActionAudit(transition.ID)
if err != nil {
log.Warn().Err(err).Str("orgID", orgID).Str("actionID", transition.ID).Msg("Unable to read authoritative Patrol action transition")
@@ -36,14 +85,41 @@ func (h *AISettingsHandler) ReconcilePatrolActionTransition(orgID string, transi
h.applyPatrolActionAudit(orgID, authoritative)
}
// closePatrolInvestigationRun reconciles a closed run's action and finalizes the
// run under the investigation's lock, so no projection of the action lands
// between the two: the orchestrator writes the action back as it was planned,
// which an approval, rejection or expiry during the run may have overtaken.
// Once the orchestrator's own close has landed, its stored session is the one
// finalized, rather than a copy the stuck-run sweep marked failed.
func (h *AISettingsHandler) closePatrolInvestigationRun(orgID string, session *aicontracts.InvestigationSession, finalize func(*aicontracts.InvestigationSession)) {
if h == nil || session == nil || finalize == nil {
return
}
defer lockPatrolInvestigation(orgID, session.ID)()
if stored := h.storedInvestigation(orgID, session.ID); stored != nil &&
stored.Status != aicontracts.InvestigationStatusPending && stored.Status != aicontracts.InvestigationStatusRunning {
session = stored
}
finalize(h.hydratePatrolInvestigationActionLocked(orgID, session))
}
// hydratePatrolInvestigationAction repairs a missed transition callback before
// an investigation is returned. Action audit remains authoritative: existing
// references hydrate by action ID, while older/missed records recover through
// the broker-owned origin index.
// the broker-owned origin index. The caller's session comes back with the
// action and outcome its store holds after the repair, so a copy taken before
// a refresh replaced its plan does not bring the replaced plan back.
func (h *AISettingsHandler) hydratePatrolInvestigationAction(orgID string, investigation *aicontracts.InvestigationSession) *aicontracts.InvestigationSession {
if h == nil || investigation == nil {
return investigation
}
defer lockPatrolInvestigation(orgID, investigation.ID)()
return h.hydratePatrolInvestigationActionLocked(orgID, investigation)
}
// hydratePatrolInvestigationActionLocked is hydratePatrolInvestigationAction for
// a caller that holds the investigation's lock.
func (h *AISettingsHandler) hydratePatrolInvestigationActionLocked(orgID string, investigation *aicontracts.InvestigationSession) *aicontracts.InvestigationSession {
store, err := h.actionAuditStore(orgID)
if err != nil {
log.Warn().Err(err).Str("orgID", orgID).Str("investigationID", investigation.ID).Msg("Unable to hydrate Patrol action")
@@ -66,9 +142,14 @@ func (h *AISettingsHandler) hydratePatrolInvestigationAction(orgID string, inves
}
h.applyPatrolActionAudit(orgID, audit)
// Return the hydrated projection even if the backing investigation store
// disappeared between lookup and repair.
// Fall back to the audit's own projection if the backing investigation
// store disappeared between lookup and repair.
copy := *investigation
if stored := h.storedInvestigation(orgID, investigation.ID); stored != nil {
copy.Action = stored.Action
copy.Outcome = stored.Outcome
return &copy
}
copy.Action = actionReferenceFromAudit(audit)
if outcome := patrolOutcomeForActionAudit(audit); outcome != "" {
copy.Outcome = outcome
@@ -76,6 +157,17 @@ func (h *AISettingsHandler) hydratePatrolInvestigationAction(orgID string, inves
return &copy
}
// storedInvestigation returns the investigation as its org's store holds it.
func (h *AISettingsHandler) storedInvestigation(orgID, investigationID string) *aicontracts.InvestigationSession {
h.investigationMu.RLock()
store := h.investigationStores[orgID]
h.investigationMu.RUnlock()
if store == nil {
return nil
}
return store.Get(investigationID)
}
func (h *AISettingsHandler) applyPatrolActionAudit(orgID string, audit unifiedresources.ActionAuditRecord) {
origin := audit.Origin
if !isPatrolActionOrigin(origin) {
@@ -95,25 +187,51 @@ func (h *AISettingsHandler) applyPatrolActionAudit(orgID string, audit unifiedre
return
}
reference := actionReferenceFromAudit(audit)
changed := false
if !reflect.DeepEqual(investigation.Action, reference) {
investigation.Action = reference
changed = true
// The model's causal attribution is not part of the action audit, so the
// one the investigation recorded for this action stays.
if investigation.Action != nil && investigation.Action.ActionID == reference.ActionID {
reference.CausalResourceID = investigation.Action.CausalResourceID
}
if outcome := patrolOutcomeForActionAudit(audit); outcome != "" {
if investigation.Outcome != outcome {
investigation.Outcome = outcome
changed = true
}
if changed {
store.Update(investigation)
actionChanged := !reflect.DeepEqual(investigation.Action, reference)
if actionChanged {
investigation.Action = reference
}
outcome := patrolOutcomeForActionAudit(audit)
outcomeChanged := outcome != "" && investigation.Outcome != outcome
if outcomeChanged {
investigation.Outcome = outcome
}
saveInvestigationProjection(store, investigation, actionChanged, outcomeChanged)
if outcome != "" {
// The finding's record follows the session as stored, which a store
// that merges the action, such as keeping the causal attribution,
// may hold differently from this projection.
if stored := store.Get(investigation.ID); stored != nil {
investigation = stored
}
ctx := context.WithValue(context.Background(), OrgIDContextKey, orgID)
h.updateFindingInvestigationOutcome(ctx, orgID, origin.FindingID, string(outcome), investigation)
}
}
// saveInvestigationProjection records a reconciled action and outcome on the
// investigation's session. A store that records the action on its own does not
// replace the session, so progress the orchestrator saved after the session
// was read stays.
func saveInvestigationProjection(store aicontracts.InvestigationStore, investigation *aicontracts.InvestigationSession, actionChanged, outcomeChanged bool) {
if !actionChanged && !outcomeChanged {
return
}
if changed {
actions, ok := store.(aicontracts.InvestigationActionStore)
if !ok {
store.Update(investigation)
return
}
if actionChanged {
actions.SetAction(investigation.ID, investigation.Action)
}
if outcomeChanged {
store.SetOutcome(investigation.ID, investigation.Outcome)
}
}
@@ -168,15 +286,20 @@ func isPatrolActionOrigin(origin *unifiedresources.ActionOrigin) bool {
strings.TrimSpace(origin.InvestigationID) != ""
}
// actionReferenceFromAudit projects an action audit onto the reference an
// investigation keeps, with the canonical result and verification status the
// broker reports for it (dispositionFromRecord).
func actionReferenceFromAudit(audit unifiedresources.ActionAuditRecord) *aicontracts.ActionReference {
plan := approvalPlanRequestToInfo(&audit.Plan)
disposition := dispositionFromRecord(audit)
return &aicontracts.ActionReference{
ActionID: audit.ID,
ProposalID: audit.Origin.ProposalID,
ResourceID: audit.Request.ResourceID,
CapabilityName: audit.Request.CapabilityName,
State: string(audit.State),
Plan: *plan,
ActionResultV2: disposition.ActionResultV2,
VerificationStatus: disposition.VerificationStatus,
ActionID: audit.ID,
ProposalID: audit.Origin.ProposalID,
ResourceID: audit.Request.ResourceID,
CapabilityName: audit.Request.CapabilityName,
State: string(audit.State),
Plan: disposition.Plan,
}
}
+8
View File
@@ -137,3 +137,11 @@ type InvestigationStore interface {
Cleanup(maxAge time.Duration) int
EnforceSizeLimit(maxSessions int) int
}
// InvestigationActionStore is implemented by investigation stores that record
// an investigation's action without replacing the rest of the session. Pulse
// uses it, when the store offers it, to record the action lifecycle's
// transitions without undoing progress the orchestrator saved meanwhile.
type InvestigationActionStore interface {
SetAction(id string, action *ActionReference) bool
}