diff --git a/docs/release-control/v6/internal/SOURCE_OF_TRUTH.md b/docs/release-control/v6/internal/SOURCE_OF_TRUTH.md index b8605761a..96f63168f 100644 --- a/docs/release-control/v6/internal/SOURCE_OF_TRUTH.md +++ b/docs/release-control/v6/internal/SOURCE_OF_TRUTH.md @@ -213,6 +213,14 @@ The retained foundation is therefore the hidden backend layer: canonical resources and relationships, standardized agent-emitted resource/signal/change envelopes, policy metadata and routing hooks, governed action and approval boundaries with auditability, and first-class fleet governance. +Governed action identity is create-once and lifecycle state is monotonic. A +deterministic replay returns the authoritative persisted record; it cannot +replace approvals, execution results, verification, origin, or terminal state. +Only the store-level compare-and-swap winner that persists the `executing` +transition and its event may admit the executor. This is an exactly-one +admission guarantee, not permission to claim exactly-once external effects +after a process crash; durable effect recovery remains a separately governed +continuity obligation. ## Evergreen Readiness Assertions diff --git a/docs/release-control/v6/internal/status.json b/docs/release-control/v6/internal/status.json index 6f9358405..36e0ec05d 100644 --- a/docs/release-control/v6/internal/status.json +++ b/docs/release-control/v6/internal/status.json @@ -4135,6 +4135,85 @@ "kind": "file" } ] + }, + { + "id": "RA36", + "summary": "Deterministic governed-action replay is create-once and monotonic: concurrent identical proposals return the authoritative current audit, terminal states cannot rewind, record and lifecycle events commit atomically, and exactly one successful transition to executing may admit the executor.", + "kind": "invariant", + "blocking_level": "repo-ready", + "proof_type": "automated", + "lane_ids": [ + "L6", + "L13", + "L20" + ], + "subsystem_ids": [ + "api-contracts", + "unified-resources" + ], + "release_gate_ids": [], + "proof_commands": [ + { + "id": "ra36-action-lifecycle-contract", + "run": [ + "go", + "test", + "./internal/unifiedresources", + "./internal/api", + "-run", + "TestProductionActionLifecycleDoesNotUseRecordActionAuditAsUpsert|TestContract_ActionLifecycleReplayIsCreateOnceAndMonotonic|TestContract_ActionExecutorAdmissionRequiresExecutingCASWinner", + "-count=1" + ] + }, + { + "id": "ra36-atomic-action-lifecycle", + "run": [ + "go", + "test", + "./internal/unifiedresources", + "./internal/actionlifecycle", + "./internal/api", + "-run", + "TestMemoryStoreCreateActionAuditConcurrentReturnsCurrent|TestSQLiteStoreCreateActionAuditConcurrentAcrossTwoInstances|TestSQLiteStoreActionTransitionCASAcrossTwoInstances|TestSQLiteStoreConcurrentExecutionStartAcrossTwoInstancesHasOneWinner|TestSQLiteStoreCreateActionAuditRollsBackWhenInitialEventInsertFails|TestSQLiteStoreActionTransitionRollsBackWhenEventInsertFails|TestSQLiteStoreLifecycleRestartPreservesMonotonicState|TestSQLiteStoreRestartDoesNotReadmitExecutingAction|TestConcurrentPlanReplayCannotRewindTerminalActionMemoryStore|TestConcurrentPlanReplayCannotRewindTerminalActionSQLiteStore|TestConcurrentExecuteAdmitsExecutorExactlyOnceMemoryStore|TestConcurrentExecuteAdmitsExecutorExactlyOnceSQLiteStore|TestPlanReplayRejectsConflictingOriginForDeterministicActionID|TestPatrolActionBrokerBarrierReplayAdmitsExecutorExactlyOnce|TestPatrolActionBrokerTerminalReplayPreservesAuditAndEvents|TestPatrolActionReconciliationHydratesTerminalAuditAfterRestart|TestPatrolActionReconciliationCannotRegressFromOutOfOrderCallback", + "-count=1" + ] + } + ], + "evidence": [ + { + "repo": "pulse", + "path": "internal/actionlifecycle/service.go", + "kind": "file" + }, + { + "repo": "pulse", + "path": "internal/actionlifecycle/service_test.go", + "kind": "file", + "evidence_tier": "test-proof" + }, + { + "repo": "pulse", + "path": "internal/api/patrol_action_broker_test.go", + "kind": "file", + "evidence_tier": "test-proof" + }, + { + "repo": "pulse", + "path": "internal/unifiedresources/actions.go", + "kind": "file" + }, + { + "repo": "pulse", + "path": "internal/unifiedresources/store.go", + "kind": "file" + }, + { + "repo": "pulse", + "path": "internal/unifiedresources/store_test.go", + "kind": "file", + "evidence_tier": "test-proof" + } + ] } ], "evidence_reference_policy": { diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 8b12c92a9..b63353a75 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -1738,6 +1738,13 @@ Agent` secondary handoff against the live setup wizard instead of relying ## Current State +Agent dispatch begins only after the canonical action store wins and commits +the transition to `executing`. Concurrent replay, another SQLite connection, +or a process restart cannot obtain a second executor admission for that action. +This contract intentionally does not claim exactly-once infrastructure effects +after a crash; durable attempt recovery and downstream effect reconciliation +remain the action-continuity layer's responsibility. + Deploy fan-out concurrency is one shared protocol contract in `internal/agentexec`: server request normalization and host-agent semaphore allocation both cap `max_parallel` at the same bound, including payloads that diff --git a/docs/release-control/v6/internal/subsystems/ai-runtime.md b/docs/release-control/v6/internal/subsystems/ai-runtime.md index 5909e377e..678096d01 100644 --- a/docs/release-control/v6/internal/subsystems/ai-runtime.md +++ b/docs/release-control/v6/internal/subsystems/ai-runtime.md @@ -3518,6 +3518,13 @@ query...`, and `Reading storage...` before streamed tool arguments are ## Current State +Assistant compatibility audit writers no longer use whole-record action-audit +replacement. Fresh approval and direct-action records use atomic creation with +their initial events, while decisions, refusals, execution starts, and results +use the same typed CAS transitions as the canonical action lifecycle. This +keeps legacy boundary producers from reopening the state-rewind path while +their mutation-plane consolidation remains separately governed. + AI provider model-cache identity must never expose reusable credential material or deterministic unkeyed credential hashes. `internal/ai/service.go` derives cache-only credential identities with a process-local random HMAC key, so an diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index e96223ecb..e047e2325 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -7297,3 +7297,12 @@ action path. `ActionPlanInfo` carries canonical preflight detail across the broker boundary. Transition publication is org-scoped, persistence precedes publication, and API reconciliation treats the callback payload only as an id to re-read from the action lifecycle store. +Planning uses atomic `CreateActionAudit` with its initial lifecycle events; it +does not perform a separate existence read followed by an upsert. Identical +replay returns the authoritative current plan and disposition, while a +deterministic action-ID collision with different stable request, plan hash, or +broker-owned origin fails without a write. Decision, refusal, execution-start, +and execution-result writes are conditional state transitions whose event and +record update commit together. Only the successful execution-start CAS winner +may call the executor; a concurrent or post-restart duplicate returns the +current executing or terminal record without another admission. diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index 7188d380d..e3d2a8d3f 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -1682,6 +1682,12 @@ must not treat starter ## Current State +Storage and recovery consumers inherit create-once action identity, monotonic +terminal state, and exactly-one executor admission from the shared lifecycle. +They must not retry an `executing` action as a new dispatch after restart; +recovery must resolve the existing attempt through the separately governed +continuity contract. + Unified Agent lifecycle fields added to the shared host and connections API are adjacent monitoring/API state only. Applied config fingerprints, updater status, and Host, Docker/Podman, or Kubernetes module readiness do not become diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index 0ebf16de9..435a57470 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -2883,9 +2883,15 @@ That same shared store now also persists append-only action lifecycle, action audit, and export audit records, giving the control-plane verbs a durable home next to the resource timeline instead of leaving those records isolated in memory-only models. -The in-memory store mirrors the durable audit contract by upserting action -audits on action ID, so tests and runtime callers observe the same current -record state that SQLite persists for the control-plane execution trail. +The in-memory store mirrors the durable audit contract through atomic +create-or-return-current action identity and monotonic typed transitions; it +must never replace an existing action record merely because an ID collides. +SQLite uses insert-on-conflict-do-nothing for creation and conditional state +updates for transitions, with the lifecycle event committed in the same +transaction. Both stores treat rejected, completed, and failed states as +absorbing and admit exactly one successful transition to executing. Lifecycle +events are unique per action and state, with migration deduplication for rows +written before this invariant became database-enforced. It also mirrors the durable decision contract: approval/rejection writes must target an existing pending action and must fail rather than creating a decision-only record or overwriting an already decided action. diff --git a/internal/actionlifecycle/service.go b/internal/actionlifecycle/service.go index 50f03ba9b..179313151 100644 --- a/internal/actionlifecycle/service.go +++ b/internal/actionlifecycle/service.go @@ -35,11 +35,12 @@ type AvailabilityChecker interface { // structural subset of unified.ResourceStore so the canonical store // satisfies it without adaptation. type Store interface { - RecordActionAudit(record unified.ActionAuditRecord) error + CreateActionAudit(record unified.ActionAuditRecord, initialEvents []unified.ActionLifecycleEvent) (unified.ActionAuditRecord, bool, error) GetActionAudit(actionID string) (unified.ActionAuditRecord, bool, error) RecordActionDecision(record unified.ActionAuditRecord, event unified.ActionLifecycleEvent) error RecordActionExecutionStart(record unified.ActionAuditRecord, event unified.ActionLifecycleEvent) error RecordActionExecutionResult(record unified.ActionAuditRecord, event unified.ActionLifecycleEvent) error + RecordActionExecutionRefusal(record unified.ActionAuditRecord, event unified.ActionLifecycleEvent) error RecordActionLifecycleEvent(event unified.ActionLifecycleEvent) error GetActionLifecycleEvents(actionID string, since time.Time, limit int) ([]unified.ActionLifecycleEvent, error) GetResourceOperatorState(canonicalID string) (unified.ResourceOperatorState, bool, error) @@ -264,18 +265,13 @@ func (s *Service) PlanWithOptions(ctx context.Context, orgID string, req unified if err != nil { return unified.ActionPlan{}, err } - if existing, found, queryErr := store.GetActionAudit(plan.ActionID); queryErr != nil { - return unified.ActionPlan{}, &QueryError{Op: "idempotent action audit", Err: queryErr} - } else if found { - // Replayed proposals must never rewind an approved, executing, or - // terminal action back to its initial plan state. The deterministic - // action id already binds request, resource, and capability policy. - return existing.Plan, nil - } - record, err := persistPlanAudit(store, req, plan, opts.Origin) + record, created, err := persistPlanAudit(store, req, plan, opts.Origin) if err != nil { return unified.ActionPlan{}, &PersistError{Op: "action plan audit", Err: err} } + if !created { + return record.Plan, nil + } s.publishTransition(orgID, record) return plan, nil } @@ -334,11 +330,11 @@ func (s *Service) Capabilities(ctx context.Context, orgID, resourceID string) ([ // initial lifecycle events, deduplicating states that were already // recorded for the same action ID (idempotent replans). func PersistPlanAudit(store Store, req unified.ActionRequest, plan unified.ActionPlan) error { - _, err := persistPlanAudit(store, req, plan, nil) + _, _, err := persistPlanAudit(store, req, plan, nil) return err } -func persistPlanAudit(store Store, req unified.ActionRequest, plan unified.ActionPlan, origin *unified.ActionOrigin) (unified.ActionAuditRecord, error) { +func persistPlanAudit(store Store, req unified.ActionRequest, plan unified.ActionPlan, origin *unified.ActionOrigin) (unified.ActionAuditRecord, bool, error) { state := PlannedActionState(plan) record := unified.ActionAuditRecord{ ID: plan.ActionID, @@ -349,42 +345,25 @@ func persistPlanAudit(store Store, req unified.ActionRequest, plan unified.Actio Plan: plan, Origin: unified.NormalizeActionOrigin(origin), } - if err := store.RecordActionAudit(record); err != nil { - return unified.ActionAuditRecord{}, err - } - - existingEvents, err := store.GetActionLifecycleEvents(plan.ActionID, time.Time{}, 100) - if err != nil { - return unified.ActionAuditRecord{}, err - } - seenStates := map[unified.ActionState]bool{} - for _, event := range existingEvents { - seenStates[event.State] = true - } - - if !seenStates[unified.ActionStatePlanned] { - if err := store.RecordActionLifecycleEvent(unified.ActionLifecycleEvent{ + events := []unified.ActionLifecycleEvent{ + { ActionID: plan.ActionID, Timestamp: plan.PlannedAt, State: unified.ActionStatePlanned, Actor: req.RequestedBy, Message: "Action plan created.", - }); err != nil { - return unified.ActionAuditRecord{}, err - } + }, } - if state != unified.ActionStatePlanned && !seenStates[state] { - if err := store.RecordActionLifecycleEvent(unified.ActionLifecycleEvent{ + if state != unified.ActionStatePlanned { + events = append(events, unified.ActionLifecycleEvent{ ActionID: plan.ActionID, Timestamp: plan.PlannedAt, State: state, Actor: req.RequestedBy, Message: "Action is waiting for approval before execution.", - }); err != nil { - return unified.ActionAuditRecord{}, err - } + }) } - return record, nil + return store.CreateActionAudit(record, events) } // PlannedActionState is the initial audit state for a fresh plan: pending @@ -417,6 +396,14 @@ func (s *Service) Decide(ctx context.Context, orgID, actionID string, approval u if !ok { return unified.ActionAuditRecord{}, &ActionNotFoundError{ActionID: actionID} } + if record.State != unified.ActionStatePending { + for _, existing := range record.Approvals { + if existing.Outcome == approval.Outcome { + return record, nil + } + } + return unified.ActionAuditRecord{}, unified.ErrActionNotPending + } now := s.now() if approval.Timestamp.IsZero() { @@ -428,6 +415,14 @@ func (s *Service) Decide(ctx context.Context, orgID, actionID string, approval u } if err := store.RecordActionDecision(updated, event); err != nil { if errors.Is(err, unified.ErrActionNotPending) { + current, found, queryErr := store.GetActionAudit(actionID) + if queryErr == nil && found { + for _, existing := range current.Approvals { + if existing.Outcome == approval.Outcome { + return current, nil + } + } + } return unified.ActionAuditRecord{}, err } return unified.ActionAuditRecord{}, &PersistError{Op: "action decision", Err: err} @@ -458,6 +453,9 @@ func (s *Service) Execute(ctx context.Context, orgID, actionID, actor, reason st if !ok { return unified.ActionAuditRecord{}, &ActionNotFoundError{ActionID: actionID} } + if record.State == unified.ActionStateExecuting || record.State == unified.ActionStateCompleted || record.State == unified.ActionStateFailed { + return record, nil + } now := s.now() if err := unified.ValidateActionExecutionStart(record, now); err != nil { @@ -508,8 +506,15 @@ func (s *Service) Execute(ctx context.Context, orgID, actionID, actor, reason st startEvent.Message = "Action execution started: " + reason } if err := store.RecordActionExecutionStart(started, startEvent); err != nil { + if errors.Is(err, unified.ErrActionAlreadyExecuting) || errors.Is(err, unified.ErrActionExecutionFinal) { + current, found, queryErr := store.GetActionAudit(actionID) + if queryErr == nil && found { + return current, nil + } + } return unified.ActionAuditRecord{}, &PersistError{Op: "action execution start", Err: err} } + s.publishTransition(orgID, started) result, execErr := s.Executor.ExecuteAction(ctx, started) if execErr != nil { @@ -597,10 +602,7 @@ func RecordRefusedExecution(store Store, record unified.ActionAuditRecord, actor if store == nil { return unified.ActionAuditRecord{}, errors.New("action audit store unavailable") } - if err := store.RecordActionAudit(failed); err != nil { - return unified.ActionAuditRecord{}, err - } - if err := store.RecordActionLifecycleEvent(event); err != nil { + if err := store.RecordActionExecutionRefusal(failed, event); err != nil { return unified.ActionAuditRecord{}, err } return failed, nil diff --git a/internal/actionlifecycle/service_test.go b/internal/actionlifecycle/service_test.go index b011d6baf..d77f3e8da 100644 --- a/internal/actionlifecycle/service_test.go +++ b/internal/actionlifecycle/service_test.go @@ -3,6 +3,7 @@ package actionlifecycle import ( "context" "errors" + "sync/atomic" "testing" "time" @@ -10,6 +11,248 @@ import ( unified "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" ) +type delayedCreateStore struct { + unified.ResourceStore + calls atomic.Int32 + secondArrived chan struct{} + releaseSecond chan struct{} +} + +func (s *delayedCreateStore) CreateActionAudit(record unified.ActionAuditRecord, events []unified.ActionLifecycleEvent) (unified.ActionAuditRecord, bool, error) { + if s.calls.Add(1) == 2 { + close(s.secondArrived) + <-s.releaseSecond + } + return s.ResourceStore.CreateActionAudit(record, events) +} + +type blockingExecutor struct { + calls atomic.Int32 + entered chan struct{} + release chan struct{} +} + +func (e *blockingExecutor) ExecuteAction(_ context.Context, record unified.ActionAuditRecord) (*unified.ExecutionResult, error) { + if e.calls.Add(1) == 1 { + close(e.entered) + } + <-e.release + return &unified.ExecutionResult{Success: true, Output: record.ID}, nil +} + +func serviceForStore(t *testing.T, store unified.ResourceStore, resource unified.Resource, executor Executor) *Service { + t.Helper() + registry := unified.NewRegistry(store) + registry.IngestResources([]unified.Resource{resource}) + return &Service{ + Registry: func(string) (*unified.ResourceRegistry, error) { return registry, nil }, + Store: func(string) (Store, error) { return store, nil }, + Executor: executor, + } +} + +func runConcurrentPlanReplayCannotRewindTerminalAction(t *testing.T, store unified.ResourceStore) { + t.Helper() + delayed := &delayedCreateStore{ResourceStore: store, secondArrived: make(chan struct{}), releaseSecond: make(chan struct{})} + executor := &stubExecutor{result: &unified.ExecutionResult{Success: true}} + service := serviceForStore(t, delayed, testResource(time.Now().UTC(), unified.ApprovalAdmin), executor) + firstPlan, err := service.Plan(context.Background(), "default", restartRequest()) + if err != nil { + t.Fatal(err) + } + secondResult := make(chan error, 1) + go func() { _, err := service.Plan(context.Background(), "default", restartRequest()); secondResult <- err }() + <-delayed.secondArrived + if _, err := service.Decide(context.Background(), "default", firstPlan.ActionID, unified.ActionApprovalRecord{Actor: "operator", Method: unified.MethodAPI, Outcome: unified.OutcomeApproved}); err != nil { + t.Fatal(err) + } + if _, err := service.Execute(context.Background(), "default", firstPlan.ActionID, "operator", "proof"); err != nil { + t.Fatal(err) + } + close(delayed.releaseSecond) + if err := <-secondResult; err != nil { + t.Fatal(err) + } + current, found, err := store.GetActionAudit(firstPlan.ActionID) + if err != nil || !found || current.State != unified.ActionStateCompleted { + t.Fatalf("current=%#v found=%v err=%v", current, found, err) + } + if executor.calls != 1 { + t.Fatalf("executor calls=%d, want 1", executor.calls) + } +} + +func TestConcurrentPlanReplayCannotRewindTerminalActionMemoryStore(t *testing.T) { + runConcurrentPlanReplayCannotRewindTerminalAction(t, unified.NewMemoryStore()) +} + +func TestConcurrentPlanReplayCannotRewindTerminalActionSQLiteStore(t *testing.T) { + store, err := unified.NewSQLiteResourceStore(t.TempDir(), "default") + if err != nil { + t.Fatal(err) + } + defer store.Close() + runConcurrentPlanReplayCannotRewindTerminalAction(t, store) +} + +func runConcurrentExecuteAdmitsExecutorExactlyOnce(t *testing.T, store unified.ResourceStore) { + t.Helper() + executor := &blockingExecutor{entered: make(chan struct{}), release: make(chan struct{})} + service := serviceForStore(t, store, testResource(time.Now().UTC(), unified.ApprovalNone), executor) + plan, err := service.Plan(context.Background(), "default", restartRequest()) + if err != nil { + t.Fatal(err) + } + first := make(chan error, 1) + go func() { + _, err := service.Execute(context.Background(), "default", plan.ActionID, "operator", "first") + first <- err + }() + <-executor.entered + secondRecord, secondErr := service.Execute(context.Background(), "default", plan.ActionID, "operator", "second") + if secondErr != nil { + t.Fatalf("duplicate Execute: %v", secondErr) + } + if secondRecord.State != unified.ActionStateExecuting { + t.Fatalf("duplicate state=%q", secondRecord.State) + } + close(executor.release) + if err := <-first; err != nil { + t.Fatal(err) + } + if executor.calls.Load() != 1 { + t.Fatalf("executor calls=%d, want 1", executor.calls.Load()) + } +} + +func TestConcurrentExecuteAdmitsExecutorExactlyOnceMemoryStore(t *testing.T) { + runConcurrentExecuteAdmitsExecutorExactlyOnce(t, unified.NewMemoryStore()) +} + +func TestConcurrentExecuteAdmitsExecutorExactlyOnceSQLiteStore(t *testing.T) { + store, err := unified.NewSQLiteResourceStore(t.TempDir(), "default") + if err != nil { + t.Fatal(err) + } + defer store.Close() + runConcurrentExecuteAdmitsExecutorExactlyOnce(t, store) +} + +func TestPlanReplayReturnsAuthoritativeApprovedExecutingAndTerminalRecords(t *testing.T) { + states := []unified.ActionState{unified.ActionStateApproved, unified.ActionStateExecuting, unified.ActionStateRejected, unified.ActionStateCompleted, unified.ActionStateFailed} + for _, state := range states { + t.Run(string(state), func(t *testing.T) { + store := unified.NewMemoryStore() + service := serviceForStore(t, store, testResource(time.Now().UTC(), unified.ApprovalAdmin), &stubExecutor{}) + plan, err := service.Plan(context.Background(), "default", restartRequest()) + if err != nil { + t.Fatal(err) + } + record, _, _ := store.GetActionAudit(plan.ActionID) + decisionOutcome := unified.OutcomeApproved + if state == unified.ActionStateRejected { + decisionOutcome = unified.OutcomeRejected + } + decided, decisionEvent, err := unified.ApplyActionDecision(record, unified.ActionApprovalRecord{Actor: "operator", Outcome: decisionOutcome}, time.Now().UTC()) + if err != nil || store.RecordActionDecision(decided, decisionEvent) != nil { + t.Fatalf("decision: %v", err) + } + if state != unified.ActionStateApproved && state != unified.ActionStateRejected { + started, startEvent, err := unified.BeginActionExecution(decided, "operator", time.Now().UTC()) + if err != nil || store.RecordActionExecutionStart(started, startEvent) != nil { + t.Fatalf("start: %v", err) + } + if state == unified.ActionStateCompleted || state == unified.ActionStateFailed { + result := &unified.ExecutionResult{Success: state == unified.ActionStateCompleted} + completed, doneEvent, err := unified.CompleteActionExecution(started, result, "operator", time.Now().UTC()) + if err != nil || store.RecordActionExecutionResult(completed, doneEvent) != nil { + t.Fatalf("complete: %v", err) + } + } + } + returned, err := service.Plan(context.Background(), "default", restartRequest()) + if err != nil { + t.Fatal(err) + } + if returned.ActionID != plan.ActionID { + t.Fatalf("action id=%q", returned.ActionID) + } + current, _, _ := store.GetActionAudit(plan.ActionID) + if current.State != state { + t.Fatalf("state=%q want=%q", current.State, state) + } + }) + } +} + +func TestPlanReplayRejectsConflictingOriginForDeterministicActionID(t *testing.T) { + store := unified.NewMemoryStore() + service := serviceForStore(t, store, testResource(time.Now().UTC(), unified.ApprovalAdmin), &stubExecutor{}) + first := PlanOptions{Origin: &unified.ActionOrigin{Surface: "patrol", FindingID: "finding-1", InvestigationID: "inv-1", ProposalID: "proposal-1"}} + if _, err := service.PlanWithOptions(context.Background(), "default", restartRequest(), first); err != nil { + t.Fatal(err) + } + conflict := first + conflict.Origin = &unified.ActionOrigin{Surface: "patrol", FindingID: "finding-2", InvestigationID: "inv-2", ProposalID: "proposal-1"} + if _, err := service.PlanWithOptions(context.Background(), "default", restartRequest(), conflict); !errors.Is(err, unified.ErrActionIdentityConflict) { + t.Fatalf("error=%v", err) + } +} + +func TestExecutePublishesPersistedExecutingTransition(t *testing.T) { + store := unified.NewMemoryStore() + executor := &stubExecutor{result: &unified.ExecutionResult{Success: true}} + service := serviceForStore(t, store, testResource(time.Now().UTC(), unified.ApprovalNone), executor) + var states []unified.ActionState + service.OnActionTransition = func(_ string, record unified.ActionAuditRecord) { states = append(states, record.State) } + plan, err := service.Plan(context.Background(), "default", restartRequest()) + if err != nil { + t.Fatal(err) + } + if _, err := service.Execute(context.Background(), "default", plan.ActionID, "operator", ""); err != nil { + t.Fatal(err) + } + want := []unified.ActionState{unified.ActionStatePlanned, unified.ActionStateExecuting, unified.ActionStateCompleted} + if len(states) != len(want) { + t.Fatalf("states=%v", states) + } + for i := range want { + if states[i] != want[i] { + t.Fatalf("states=%v want=%v", states, want) + } + } +} + +func TestExecuteAfterSQLiteRestartDoesNotReadmitExecutingAction(t *testing.T) { + dir := t.TempDir() + first, err := unified.NewSQLiteResourceStore(dir, "default") + if err != nil { + t.Fatal(err) + } + executor := &stubExecutor{result: &unified.ExecutionResult{Success: true}} + service := serviceForStore(t, first, testResource(time.Now().UTC(), unified.ApprovalNone), executor) + plan, err := service.Plan(context.Background(), "default", restartRequest()) + if err != nil { + t.Fatal(err) + } + record, _, _ := first.GetActionAudit(plan.ActionID) + started, event, _ := unified.BeginActionExecution(record, "operator", time.Now().UTC()) + if err := first.RecordActionExecutionStart(started, event); err != nil { + t.Fatal(err) + } + first.Close() + second, err := unified.NewSQLiteResourceStore(dir, "default") + if err != nil { + t.Fatal(err) + } + defer second.Close() + service = serviceForStore(t, second, testResource(time.Now().UTC(), unified.ApprovalNone), executor) + current, err := service.Execute(context.Background(), "default", plan.ActionID, "operator", "retry") + if err != nil || current.State != unified.ActionStateExecuting || executor.calls != 0 { + t.Fatalf("current=%#v err=%v calls=%d", current, err, executor.calls) + } +} + type stubExecutor struct { result *unified.ExecutionResult err error @@ -474,7 +717,7 @@ func TestOnActionTransitionFiresAfterEachPersistedState(t *testing.T) { t.Fatalf("Execute: %v", err) } - want := []unified.ActionState{unified.ActionStatePending, unified.ActionStateApproved, unified.ActionStateCompleted} + want := []unified.ActionState{unified.ActionStatePending, unified.ActionStateApproved, unified.ActionStateExecuting, unified.ActionStateCompleted} if len(transitions) != len(want) { t.Fatalf("transitions = %v, want %v", transitions, want) } diff --git a/internal/ai/tools/action_audit.go b/internal/ai/tools/action_audit.go index 4bba74b98..72a48f68e 100644 --- a/internal/ai/tools/action_audit.go +++ b/internal/ai/tools/action_audit.go @@ -141,8 +141,9 @@ func (e *PulseToolExecutor) executeCommandWithAudit( Success: false, ErrorMessage: fmt.Sprintf("plan_drift: %s", driftErr.Error()), } - e.recordActionAudit(record) - e.recordActionLifecycle(record.ID, unifiedresources.ActionStateFailed, requestedBy, "plan drift refused") + if err := persistFailedActionAudit(e.actionAuditStore, record, requestedBy, "plan drift refused"); err != nil { + log.Warn().Err(err).Str("action_id", record.ID).Msg("failed to persist plan-drift refusal") + } e.publishActionCompleted(record) return nil, driftErr } @@ -380,7 +381,12 @@ func (e *PulseToolExecutor) recordActionExecutionStart(record unifiedresources.A return record, err } } else if e != nil && e.actionAuditStore != nil { - if err := e.actionAuditStore.RecordActionAudit(record); err != nil { + plannedEvent := unifiedresources.ActionLifecycleEvent{ + ActionID: record.ID, Timestamp: now, State: unifiedresources.ActionStatePlanned, + Actor: actor, Message: strings.TrimSpace(record.Request.Reason), + } + current, _, err := e.actionAuditStore.CreateActionAudit(record, []unifiedresources.ActionLifecycleEvent{plannedEvent}) + if err != nil { log.Warn(). Err(err). Str("action_id", record.ID). @@ -388,21 +394,7 @@ func (e *PulseToolExecutor) recordActionExecutionStart(record unifiedresources.A Msg("failed to persist planned action audit") return record, err } - event := unifiedresources.ActionLifecycleEvent{ - ActionID: record.ID, - Timestamp: now, - State: unifiedresources.ActionStatePlanned, - Actor: actor, - Message: strings.TrimSpace(record.Request.Reason), - } - if err := e.actionAuditStore.RecordActionLifecycleEvent(event); err != nil { - log.Warn(). - Err(err). - Str("action_id", record.ID). - Str("state", string(unifiedresources.ActionStatePlanned)). - Msg("failed to persist planned action lifecycle event") - return record, err - } + record = current } started, event, err := unifiedresources.BeginActionExecution(record, actor, now) @@ -443,21 +435,23 @@ func (e *PulseToolExecutor) recordActionExecutionRefusal(record unifiedresources e.publishActionCompleted(failed) return failed, reason } - if err := e.actionAuditStore.RecordActionAudit(failed); err != nil { + _, found, queryErr := e.actionAuditStore.GetActionAudit(failed.ID) + if queryErr != nil { + return record, queryErr + } + var persistErr error + if found { + persistErr = e.actionAuditStore.RecordActionExecutionRefusal(failed, event) + } else { + _, _, persistErr = e.actionAuditStore.CreateActionAudit(failed, []unifiedresources.ActionLifecycleEvent{event}) + } + if persistErr != nil { log.Warn(). - Err(err). + Err(persistErr). Str("action_id", failed.ID). Str("state", string(failed.State)). Msg("failed to persist action execution refusal") - return record, err - } - if err := e.actionAuditStore.RecordActionLifecycleEvent(event); err != nil { - log.Warn(). - Err(err). - Str("action_id", failed.ID). - Str("state", string(failed.State)). - Msg("failed to persist action execution refusal lifecycle event") - return record, err + return record, persistErr } e.publishActionCompleted(failed) return failed, reason @@ -484,11 +478,13 @@ func (e *PulseToolExecutor) ensureApprovalDecisionBeforeExecution(record unified return record, err } if !ok { - if err := e.actionAuditStore.RecordActionAudit(record); err != nil { + decisionEvent := unifiedresources.ActionLifecycleEvent{ActionID: record.ID, Timestamp: now, State: record.State, Actor: actor, Message: "Action approval was recorded before execution."} + current, _, err := e.actionAuditStore.CreateActionAudit(record, []unifiedresources.ActionLifecycleEvent{decisionEvent}) + if err != nil { log.Warn().Err(err).Str("action_id", record.ID).Msg("failed to persist approved action audit before execution") return record, err } - return record, nil + return current, nil } if current.State != unifiedresources.ActionStatePending { return current, nil @@ -552,9 +548,7 @@ func (e *PulseToolExecutor) recordActionExecutionResult(record unifiedresources. } else { record.State = unifiedresources.ActionStateFailed } - e.recordActionAudit(record) - e.recordActionLifecycle(record.ID, record.State, actor, message) - e.publishActionCompleted(record) + log.Warn().Str("action_id", record.ID).Msg("refusing to persist a synthetic terminal action result after transition normalization failed") return record } if strings.TrimSpace(message) != "" { @@ -575,19 +569,6 @@ func (e *PulseToolExecutor) recordActionExecutionResult(record unifiedresources. return completed } -func (e *PulseToolExecutor) recordActionAudit(record unifiedresources.ActionAuditRecord) { - if e == nil || e.actionAuditStore == nil { - return - } - if err := e.actionAuditStore.RecordActionAudit(record); err != nil { - log.Warn(). - Err(err). - Str("action_id", record.ID). - Str("resource_id", record.Request.ResourceID). - Msg("failed to persist action audit") - } -} - // publishActionCompleted dispatches the executor's post-completion // callback, if installed. Fire-and-forget on its own goroutine: the // callback runs after the audit record has already been persisted, @@ -612,26 +593,6 @@ func (e *PulseToolExecutor) publishActionCompleted(record unifiedresources.Actio go cb(record) } -func (e *PulseToolExecutor) recordActionLifecycle(actionID string, state unifiedresources.ActionState, actor, message string) { - if e == nil || e.actionAuditStore == nil || strings.TrimSpace(actionID) == "" { - return - } - event := unifiedresources.ActionLifecycleEvent{ - ActionID: actionID, - Timestamp: time.Now().UTC(), - State: state, - Actor: actor, - Message: message, - } - if err := e.actionAuditStore.RecordActionLifecycleEvent(event); err != nil { - log.Warn(). - Err(err). - Str("action_id", actionID). - Str("state", string(state)). - Msg("failed to persist action lifecycle event") - } -} - // RecordApprovalDecision updates the unified action audit for an approval that // reached a terminal or pre-execution decision state. func (e *PulseToolExecutor) RecordApprovalDecision(approvalID string, state unifiedresources.ActionState, actor, message string) { @@ -660,8 +621,10 @@ func RecordApprovalDecision(store unifiedresources.ResourceStore, approvalID str if recordApprovalDecisionAtomically(store, req.ID, record, actor) { return } - recordActionAudit(store, record) - recordActionLifecycle(store, req.Plan.ActionID, state, actor, message) + event := unifiedresources.ActionLifecycleEvent{ActionID: req.Plan.ActionID, Timestamp: time.Now().UTC(), State: state, Actor: actor, Message: message} + if _, _, err := store.CreateActionAudit(record, []unifiedresources.ActionLifecycleEvent{event}); err != nil { + log.Warn().Err(err).Str("action_id", record.ID).Msg("failed to create action audit for approval decision") + } } func (e *PulseToolExecutor) recordApprovalDecisionAtomically(approvalID string, record unifiedresources.ActionAuditRecord, actor string) bool { @@ -716,39 +679,6 @@ func recordApprovalDecisionAtomically(store unifiedresources.ResourceStore, appr return true } -func recordActionAudit(store unifiedresources.ResourceStore, record unifiedresources.ActionAuditRecord) { - if store == nil { - return - } - if err := store.RecordActionAudit(record); err != nil { - log.Warn(). - Err(err). - Str("action_id", record.ID). - Str("resource_id", record.Request.ResourceID). - Msg("failed to persist action audit") - } -} - -func recordActionLifecycle(store unifiedresources.ResourceStore, actionID string, state unifiedresources.ActionState, actor, message string) { - if store == nil || strings.TrimSpace(actionID) == "" { - return - } - event := unifiedresources.ActionLifecycleEvent{ - ActionID: actionID, - Timestamp: time.Now().UTC(), - State: state, - Actor: actor, - Message: message, - } - if err := store.RecordActionLifecycleEvent(event); err != nil { - log.Warn(). - Err(err). - Str("action_id", actionID). - Str("state", string(state)). - Msg("failed to persist action lifecycle event") - } -} - func (e *PulseToolExecutor) recordPendingApprovalAction(req *approval.ApprovalRequest) { if e == nil { return @@ -821,12 +751,13 @@ func RecordPendingApprovalAction(store unifiedresources.ResourceStore, req *appr } actor := approval.RequesterForRequest(req) record := actionAuditRecordFromApproval(req, unifiedresources.ActionStatePending, actor) - if err := store.RecordActionAudit(record); err != nil { - log.Warn().Err(err).Str("action_id", record.ID).Msg("failed to persist pending approval action") - return + events := []unifiedresources.ActionLifecycleEvent{ + {ActionID: req.Plan.ActionID, State: unifiedresources.ActionStatePlanned, Timestamp: record.CreatedAt, Actor: actor, Message: strings.TrimSpace(req.Context)}, + {ActionID: req.Plan.ActionID, State: unifiedresources.ActionStatePending, Timestamp: record.CreatedAt, Actor: actor, Message: "waiting for approval"}, + } + if _, _, err := store.CreateActionAudit(record, events); err != nil { + log.Warn().Err(err).Str("action_id", record.ID).Msg("failed to persist pending approval action") } - recordApprovalLifecycle(store, req.Plan.ActionID, unifiedresources.ActionStatePlanned, actor, strings.TrimSpace(req.Context)) - recordApprovalLifecycle(store, req.Plan.ActionID, unifiedresources.ActionStatePending, actor, "waiting for approval") } func recordApprovalLifecycle(store unifiedresources.ResourceStore, actionID string, state unifiedresources.ActionState, actor, message string) { @@ -1178,12 +1109,34 @@ func (e *PulseToolExecutor) refuseDispatchForRemediationLock(record unifiedresou ErrorMessage: remediationLockRefusalMessage(refusal), } record.Result = result - e.recordActionAudit(record) - e.recordActionLifecycle(record.ID, unifiedresources.ActionStateFailed, requestedBy, remediationLockLifecycleMessage(refusal)) + if err := persistFailedActionAudit(e.actionAuditStore, record, requestedBy, remediationLockLifecycleMessage(refusal)); err != nil { + log.Warn().Err(err).Str("action_id", record.ID).Msg("failed to persist remediation-lock refusal") + } e.publishActionCompleted(record) return result } +func persistFailedActionAudit(store unifiedresources.ResourceStore, record unifiedresources.ActionAuditRecord, actor, message string) error { + if store == nil { + return nil + } + event := unifiedresources.ActionLifecycleEvent{ActionID: record.ID, Timestamp: record.UpdatedAt, State: unifiedresources.ActionStateFailed, Actor: actor, Message: message} + current, found, err := store.GetActionAudit(record.ID) + if err != nil { + return err + } + if found { + current.State = unifiedresources.ActionStateFailed + current.UpdatedAt = record.UpdatedAt + current.Result = record.Result + current.Verification = record.Verification + current.VerificationOutcome = record.VerificationOutcome + return store.RecordActionExecutionRefusal(current, event) + } + _, _, err = store.CreateActionAudit(record, []unifiedresources.ActionLifecycleEvent{event}) + return err +} + func remediationLockRefusalMessage(refusal error) string { if errors.Is(refusal, ErrRemediationLockStateUnknown) { return fmt.Sprintf("remediation_lock_state_unknown: %s", refusal.Error()) diff --git a/internal/ai/tools/action_audit_execution_test.go b/internal/ai/tools/action_audit_execution_test.go index 2c9c41a7e..454ac964a 100644 --- a/internal/ai/tools/action_audit_execution_test.go +++ b/internal/ai/tools/action_audit_execution_test.go @@ -568,7 +568,7 @@ func TestExecuteCommandWithAuditRefusesPayloadDriftAgainstApprovedPlan(t *testin // WARN logs. Operators reviewing the action audit trail need to see // "Pulse caught this drift attempt" recorded as a Failed action with // a plan_drift error message. - audits, err := actionStore.GetActionAudits("agent-1", time.Time{}, 10) + audits, err := actionStore.GetActionAudits("agent:agent-1", time.Time{}, 10) if err != nil { t.Fatalf("GetActionAudits: %v", err) } @@ -1251,7 +1251,7 @@ func TestExecuteCommandWithAuditRefusesWhenResourceIsRemediationLocked(t *testin // Refusal must be observable in the audit history with the canonical // `resource_remediation_locked:` ErrorMessage prefix so audit-UI // filters and alert rules can branch on the stable token. - audits, err := actionStore.GetActionAudits("agent-locked", time.Time{}, 10) + audits, err := actionStore.GetActionAudits("agent:agent-locked", time.Time{}, 10) if err != nil { t.Fatalf("GetActionAudits: %v", err) } diff --git a/internal/api/actions_test.go b/internal/api/actions_test.go index e1ff4ba7d..234d87fb5 100644 --- a/internal/api/actions_test.go +++ b/internal/api/actions_test.go @@ -886,7 +886,7 @@ func TestHandleExecuteActionRejectsExpiredPlanAsFailedAudit(t *testing.T) { } } -func TestPersistActionPlanAuditFillsMissingLifecycleState(t *testing.T) { +func TestPersistActionPlanAuditRejectsOrphanLifecycleState(t *testing.T) { now := time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC) store := unified.NewMemoryStore() req := unified.ActionRequest{ @@ -918,30 +918,11 @@ func TestPersistActionPlanAuditFillsMissingLifecycleState(t *testing.T) { t.Fatalf("seed lifecycle event: %v", err) } - if err := actionlifecycle.PersistPlanAudit(store, req, plan); err != nil { - t.Fatalf("persistActionPlanAudit: %v", err) + if err := actionlifecycle.PersistPlanAudit(store, req, plan); err == nil { + t.Fatal("orphan lifecycle state must make atomic plan creation fail") } - events, err := store.GetActionLifecycleEvents(plan.ActionID, time.Time{}, 10) - if err != nil { - t.Fatalf("GetActionLifecycleEvents: %v", err) - } - seenStates := map[unified.ActionState]bool{} - for _, event := range events { - seenStates[event.State] = true - } - if len(events) != 2 || !seenStates[unified.ActionStatePlanned] || !seenStates[unified.ActionStatePending] { - t.Fatalf("events = %#v, want one planned and one pending event", events) - } - - if err := actionlifecycle.PersistPlanAudit(store, req, plan); err != nil { - t.Fatalf("persistActionPlanAudit retry: %v", err) - } - events, err = store.GetActionLifecycleEvents(plan.ActionID, time.Time{}, 10) - if err != nil { - t.Fatalf("GetActionLifecycleEvents retry: %v", err) - } - if len(events) != 2 { - t.Fatalf("retry duplicated lifecycle events: %#v", events) + if _, found, err := store.GetActionAudit(plan.ActionID); err != nil || found { + t.Fatalf("atomic creation left audit behind: found=%v err=%v", found, err) } } diff --git a/internal/api/ai_handlers_investigation_additional_test.go b/internal/api/ai_handlers_investigation_additional_test.go index 1b22ccba5..77d7ad99a 100644 --- a/internal/api/ai_handlers_investigation_additional_test.go +++ b/internal/api/ai_handlers_investigation_additional_test.go @@ -223,7 +223,7 @@ func (s *testInvestigationStore) CountFixed() int { return 0 } func (s *testInvestigationStore) Cleanup(_ time.Duration) int { return 0 } func (s *testInvestigationStore) EnforceSizeLimit(_ int) int { return 0 } -func TestReconcilePatrolActionTransitionUsesAuthoritativeAudit(t *testing.T) { +func TestPatrolActionReconciliationCannotRegressFromOutOfOrderCallback(t *testing.T) { investigations := newTestInvestigationStore() investigation := investigations.Create("finding-1", "session-1") audits := unifiedresources.NewMemoryStore() @@ -234,15 +234,12 @@ func TestReconcilePatrolActionTransitionUsesAuthoritativeAudit(t *testing.T) { Plan: unifiedresources.ActionPlan{ActionID: "act-1", RequestID: "proposal-1", Allowed: true, RequiresApproval: true}, Origin: &unifiedresources.ActionOrigin{Surface: patrolActionOriginSurface, FindingID: "finding-1", InvestigationID: investigation.ID, ProposalID: "proposal-1"}, } - if err := audits.RecordActionAudit(pending); err != nil { - t.Fatalf("RecordActionAudit(pending): %v", err) - } completed := pending completed.State = unifiedresources.ActionStateCompleted completed.UpdatedAt = now.Add(time.Second) completed.VerificationOutcome = unifiedresources.VerificationOutcome{Status: unifiedresources.VerificationVerified} - if err := audits.RecordActionAudit(completed); err != nil { - t.Fatalf("RecordActionAudit(completed): %v", err) + if _, _, err := audits.CreateActionAudit(completed, nil); err != nil { + t.Fatalf("CreateActionAudit(completed): %v", err) } handler := &AISettingsHandler{ @@ -260,19 +257,31 @@ func TestReconcilePatrolActionTransitionUsesAuthoritativeAudit(t *testing.T) { } } -func TestHydratePatrolInvestigationActionRepairsMissedCallbackByOrigin(t *testing.T) { +func TestPatrolActionReconciliationHydratesTerminalAuditAfterRestart(t *testing.T) { investigations := newTestInvestigationStore() investigation := investigations.Create("finding-1", "session-1") - audits := unifiedresources.NewMemoryStore() + dir := t.TempDir() + audits, err := unifiedresources.NewSQLiteResourceStore(dir, "default") + if err != nil { + t.Fatal(err) + } record := unifiedresources.ActionAuditRecord{ ID: "act-missed", CreatedAt: time.Now().UTC(), UpdatedAt: time.Now().UTC(), State: unifiedresources.ActionStateRejected, Request: unifiedresources.ActionRequest{RequestID: "proposal-1", ResourceID: "vm:42", CapabilityName: "restart", RequestedBy: "pulse_patrol"}, Plan: unifiedresources.ActionPlan{ActionID: "act-missed", RequestID: "proposal-1", Allowed: true, RequiresApproval: true}, Origin: &unifiedresources.ActionOrigin{Surface: patrolActionOriginSurface, FindingID: "finding-1", InvestigationID: investigation.ID, ProposalID: "proposal-1"}, } - if err := audits.RecordActionAudit(record); err != nil { - t.Fatalf("RecordActionAudit: %v", err) + if _, _, err := audits.CreateActionAudit(record, nil); err != nil { + t.Fatalf("CreateActionAudit: %v", err) } + if err := audits.Close(); err != nil { + t.Fatal(err) + } + audits, err = unifiedresources.NewSQLiteResourceStore(dir, "default") + if err != nil { + t.Fatal(err) + } + defer audits.Close() handler := &AISettingsHandler{ investigationStores: map[string]aicontracts.InvestigationStore{"default": investigations}, resourceStoreProvider: func(string) (unifiedresources.ResourceStore, error) { return audits, nil }, diff --git a/internal/api/contract_test.go b/internal/api/contract_test.go index 26b9e2ef3..05632f1fc 100644 --- a/internal/api/contract_test.go +++ b/internal/api/contract_test.go @@ -61,6 +61,38 @@ import ( tmock "github.com/stretchr/testify/mock" ) +func TestContract_ActionLifecycleReplayIsCreateOnceAndMonotonic(t *testing.T) { + service, err := os.ReadFile("../actionlifecycle/service.go") + if err != nil { + t.Fatal(err) + } + store, err := os.ReadFile("../unifiedresources/store.go") + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(service), "CreateActionAudit(record, events)") { + t.Fatal("planning must atomically create the audit and initial events") + } + for _, required := range []string{"ON CONFLICT(id) DO NOTHING", "updateActionAuditSQL", "idx_action_lifecycle_events_action_state_unique"} { + if !strings.Contains(string(store), required) { + t.Errorf("store missing %q", required) + } + } +} + +func TestContract_ActionExecutorAdmissionRequiresExecutingCASWinner(t *testing.T) { + service, err := os.ReadFile("../actionlifecycle/service.go") + if err != nil { + t.Fatal(err) + } + src := string(service) + start := strings.Index(src, "store.RecordActionExecutionStart(started, startEvent)") + dispatch := strings.Index(src, "s.Executor.ExecuteAction(ctx, started)") + if start < 0 || dispatch < 0 || start > dispatch { + t.Fatal("executor dispatch must occur only after the executing-state CAS") + } +} + type resourceContractSnapshot struct { ID string Name string @@ -15802,14 +15834,11 @@ func TestContract_ExecutorPostCompletionCallback(t *testing.T) { if !strings.Contains(auditSrc, "go cb(record)") { t.Error("post-completion callback must run on its own goroutine to keep the dispatch hot path off any consumer's slowness") } - // Pin that all four terminal sites call publishActionCompleted. + // Pin the terminal persistence sites and the shared completion publisher. terminalSites := []string{ - `e.recordActionLifecycle(record.ID, unifiedresources.ActionStateFailed, requestedBy, "plan drift refused") - e.publishActionCompleted(record)`, - `e.recordActionLifecycle(record.ID, unifiedresources.ActionStateFailed, requestedBy, remediationLockLifecycleMessage(refusal))` + - "\n\t" + `e.publishActionCompleted(record)`, - `e.recordActionLifecycle(record.ID, record.State, actor, message) - e.publishActionCompleted(record)`, + `persistFailedActionAudit(e.actionAuditStore, record, requestedBy, "plan drift refused")`, + `e.actionAuditStore.RecordActionExecutionResult(completed, event)`, + `persistFailedActionAudit(e.actionAuditStore, record, requestedBy, remediationLockLifecycleMessage(refusal))`, } for _, site := range terminalSites { if !strings.Contains(auditSrc, site) { diff --git a/internal/api/patrol_action_broker_test.go b/internal/api/patrol_action_broker_test.go index e39cb0460..26fd4b8c8 100644 --- a/internal/api/patrol_action_broker_test.go +++ b/internal/api/patrol_action_broker_test.go @@ -6,6 +6,7 @@ import ( "errors" "net/http" "net/http/httptest" + "sync/atomic" "testing" "time" @@ -17,6 +18,107 @@ import ( "github.com/rcourtman/pulse-go-rewrite/pkg/auth" ) +type barrierPatrolExecutor struct { + calls atomic.Int32 + entered chan struct{} + release chan struct{} +} + +func (e *barrierPatrolExecutor) ExecuteAction(_ context.Context, _ unified.ActionAuditRecord) (*unified.ExecutionResult, error) { + if e.calls.Add(1) == 1 { + close(e.entered) + } + <-e.release + return &unified.ExecutionResult{Success: true}, nil +} + +func configurePatrolAutoAuthorization(t *testing.T, h *ResourceHandlers) { + t.Helper() + store, err := h.getStore("default") + if err != nil { + t.Fatal(err) + } + if err := store.SetResourceOperatorState(unified.ResourceOperatorState{CanonicalID: "vm:42", AutoRemediationPolicy: unified.AutoRemediationPolicy{Enabled: true, CapabilityNames: []string{"restart"}}, SetAt: time.Now().UTC()}); err != nil { + t.Fatal(err) + } +} + +func TestPatrolActionBrokerBarrierReplayAdmitsExecutorExactlyOnce(t *testing.T) { + h, _ := newPatrolBrokerTestHandlers(t, unified.ApprovalAdmin) + executor := &barrierPatrolExecutor{entered: make(chan struct{}), release: make(chan struct{})} + h.SetActionExecutor(executor) + configurePatrolAutoAuthorization(t, h) + broker := NewPatrolActionBroker("default", h, func(context.Context, string) (PatrolActionPolicySnapshot, error) { + return PatrolActionPolicySnapshot{EffectiveAutonomyLevel: "assisted"}, nil + }) + first := make(chan error, 1) + go func() { _, err := broker.Submit(context.Background(), patrolTestProposal()); first <- err }() + <-executor.entered + secondDisposition, secondErr := broker.Submit(context.Background(), patrolTestProposal()) + if secondErr != nil { + t.Fatalf("second Submit: %v", secondErr) + } + if secondDisposition.State != string(unified.ActionStateExecuting) { + t.Fatalf("second state=%q", secondDisposition.State) + } + close(executor.release) + if err := <-first; err != nil { + t.Fatal(err) + } + if executor.calls.Load() != 1 { + t.Fatalf("executor calls=%d, want 1", executor.calls.Load()) + } +} + +func TestPatrolActionBrokerReplayDuringExecutionReturnsCurrentDisposition(t *testing.T) { + TestPatrolActionBrokerBarrierReplayAdmitsExecutorExactlyOnce(t) +} + +func TestPatrolActionBrokerTerminalReplayPreservesAuditAndEvents(t *testing.T) { + h, executor := newPatrolBrokerTestHandlers(t, unified.ApprovalAdmin) + configurePatrolAutoAuthorization(t, h) + broker := NewPatrolActionBroker("default", h, func(context.Context, string) (PatrolActionPolicySnapshot, error) { + return PatrolActionPolicySnapshot{EffectiveAutonomyLevel: "assisted"}, nil + }) + first, err := broker.Submit(context.Background(), patrolTestProposal()) + if err != nil { + t.Fatal(err) + } + store, _ := h.getStore("default") + before, err := store.GetActionLifecycleEvents(first.ActionID, time.Time{}, 100) + if err != nil { + t.Fatal(err) + } + second, err := broker.Submit(context.Background(), patrolTestProposal()) + if err != nil { + t.Fatal(err) + } + after, err := store.GetActionLifecycleEvents(first.ActionID, time.Time{}, 100) + if err != nil { + t.Fatal(err) + } + if second.State != string(unified.ActionStateCompleted) || len(after) != len(before) || executor.calls != 1 { + t.Fatalf("second=%#v events=%d/%d calls=%d", second, len(before), len(after), executor.calls) + } +} + +func TestPatrolActionBrokerConflictingOriginReplayFailsWithoutDispatch(t *testing.T) { + h, executor := newPatrolBrokerTestHandlers(t, unified.ApprovalAdmin) + broker := NewPatrolActionBroker("default", h) + if _, err := broker.Submit(context.Background(), patrolTestProposal()); err != nil { + t.Fatal(err) + } + conflict := patrolTestProposal() + conflict.FindingID = "finding-2" + conflict.InvestigationID = "inv-2" + if _, err := broker.Submit(context.Background(), conflict); !errors.Is(err, unified.ErrActionIdentityConflict) { + t.Fatalf("error=%v", err) + } + if executor.calls != 0 { + t.Fatalf("executor calls=%d", executor.calls) + } +} + func newPatrolBrokerTestHandlers(t *testing.T, minimumApproval unified.ActionApprovalLevel) (*ResourceHandlers, *stubActionExecutor) { return newPatrolBrokerTestHandlersWithEligibility(t, minimumApproval, unified.AutoAuthorizeLowRisk) } diff --git a/internal/unifiedresources/actions.go b/internal/unifiedresources/actions.go index 204b39a8e..d31385f9b 100644 --- a/internal/unifiedresources/actions.go +++ b/internal/unifiedresources/actions.go @@ -1,6 +1,8 @@ package unifiedresources import ( + "bytes" + "encoding/json" "errors" "fmt" "strings" @@ -321,8 +323,33 @@ var ( ErrActionDryRunOnly = errors.New("action plan is dry-run only") ErrActionExecutionRefusal = errors.New("action execution refusal is not a permanent terminal refusal") ErrInvalidApprovalOutcome = errors.New("invalid approval outcome") + ErrActionAuditAlreadyExists = errors.New("action audit already exists") + ErrActionIdentityConflict = errors.New("action audit identity conflicts with the persisted record") ) +// ActionAuditIdentityMatches reports whether a replay addresses the same +// immutable governed action. Lifecycle state, timestamps, approvals, results, +// and verification are deliberately excluded because the persisted record is +// authoritative for those mutable fields. +func ActionAuditIdentityMatches(existing, replay ActionAuditRecord) bool { + existing, existingErr := NormalizeActionAuditRecord(existing) + replay, replayErr := NormalizeActionAuditRecord(replay) + if existingErr != nil || replayErr != nil { + return false + } + return existing.ID == replay.ID && + existing.Plan.ActionID == replay.Plan.ActionID && + existing.Plan.PlanHash == replay.Plan.PlanHash && + canonicalActionIdentityJSONEqual(existing.Request, replay.Request) && + canonicalActionIdentityJSONEqual(existing.Origin, replay.Origin) +} + +func canonicalActionIdentityJSONEqual(left, right any) bool { + leftJSON, leftErr := json.Marshal(left) + rightJSON, rightErr := json.Marshal(right) + return leftErr == nil && rightErr == nil && bytes.Equal(leftJSON, rightJSON) +} + // ApplyActionDecision records an explicit approval or rejection against a // pending governed action without starting execution. Execution remains a // separate contract so approvals cannot become an implicit control bypass. @@ -577,6 +604,13 @@ func ValidateActionExecutionStart(record ActionAuditRecord, now time.Time) error // deterministic defaults, but rejects records that cannot identify the action, // state, resource, capability, or requester. func NormalizeActionAuditRecord(record ActionAuditRecord) (ActionAuditRecord, error) { + if record.Plan.Preflight != nil { + preflight := *record.Plan.Preflight + preflight.SafetyChecks = append([]string(nil), record.Plan.Preflight.SafetyChecks...) + preflight.VerificationSteps = append([]string(nil), record.Plan.Preflight.VerificationSteps...) + record.Plan.Preflight = &preflight + } + record.Approvals = append([]ActionApprovalRecord(nil), record.Approvals...) record.ID = strings.TrimSpace(record.ID) record.Plan.ActionID = strings.TrimSpace(record.Plan.ActionID) if record.ID == "" { diff --git a/internal/unifiedresources/code_standards_test.go b/internal/unifiedresources/code_standards_test.go index 15eafe8b7..031223eab 100644 --- a/internal/unifiedresources/code_standards_test.go +++ b/internal/unifiedresources/code_standards_test.go @@ -69,6 +69,19 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/models" ) +func TestProductionActionLifecycleDoesNotUseRecordActionAuditAsUpsert(t *testing.T) { + paths := []string{"../actionlifecycle/service.go", "../ai/tools/action_audit.go", "../api/patrol_action_broker.go"} + for _, path := range paths { + src, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read %s: %v", path, err) + } + if strings.Contains(string(src), ".RecordActionAudit(") { + t.Errorf("%s must use CreateActionAudit or a typed CAS transition", path) + } + } +} + // readConsumerGoFiles returns the contents of all non-test .go files in the // specified directory (relative to the repo internal/ root). func readConsumerGoFiles(t *testing.T, relDir string) map[string]string { diff --git a/internal/unifiedresources/store.go b/internal/unifiedresources/store.go index 89ed5cdbd..38bcdf764 100644 --- a/internal/unifiedresources/store.go +++ b/internal/unifiedresources/store.go @@ -41,12 +41,14 @@ type ResourceStore interface { CountRecentChangesBySourceTypeFiltered(canonicalID string, since time.Time, filters ResourceChangeFilters) (map[ChangeSourceType]int, error) CountRecentChangesBySourceAdapter(canonicalID string, since time.Time) (map[ChangeSourceAdapter]int, error) CountRecentChangesBySourceAdapterFiltered(canonicalID string, since time.Time, filters ResourceChangeFilters) (map[ChangeSourceAdapter]int, error) + CreateActionAudit(record ActionAuditRecord, initialEvents []ActionLifecycleEvent) (ActionAuditRecord, bool, error) RecordActionAudit(record ActionAuditRecord) error GetActionAudit(actionID string) (ActionAuditRecord, bool, error) GetActionAudits(canonicalID string, since time.Time, limit int) ([]ActionAuditRecord, error) RecordActionDecision(record ActionAuditRecord, event ActionLifecycleEvent) error RecordActionExecutionStart(record ActionAuditRecord, event ActionLifecycleEvent) error RecordActionExecutionResult(record ActionAuditRecord, event ActionLifecycleEvent) error + RecordActionExecutionRefusal(record ActionAuditRecord, event ActionLifecycleEvent) error RecordActionLifecycleEvent(event ActionLifecycleEvent) error GetActionLifecycleEvents(actionID string, since time.Time, limit int) ([]ActionLifecycleEvent, error) RecordExportAudit(record ExportAuditRecord) error @@ -534,6 +536,9 @@ func (s *SQLiteResourceStore) initSchema() error { if err := s.migrateActionAuditsSchema(); err != nil { return err } + if err := s.migrateActionLifecycleEventsSchema(); err != nil { + return err + } if err := s.migrateResourceOperatorStateSchema(); err != nil { return err } @@ -619,6 +624,24 @@ func (s *SQLiteResourceStore) migrateActionAuditsSchema() error { return nil } +func (s *SQLiteResourceStore) migrateActionLifecycleEventsSchema() error { + if _, err := s.db.Exec(` + DELETE FROM action_lifecycle_events + WHERE id NOT IN ( + SELECT MIN(id) FROM action_lifecycle_events GROUP BY action_id, state + ) + `); err != nil { + return fmt.Errorf("deduplicate action lifecycle state events: %w", err) + } + if _, err := s.db.Exec(` + CREATE UNIQUE INDEX IF NOT EXISTS idx_action_lifecycle_events_action_state_unique + ON action_lifecycle_events(action_id, state) + `); err != nil { + return fmt.Errorf("create unique action lifecycle state index: %w", err) + } + return nil +} + func (s *SQLiteResourceStore) migrateActionAuditRedaction() error { rows, err := s.db.Query(` SELECT id, result_json @@ -1705,74 +1728,189 @@ type sqlExecutor interface { Exec(query string, args ...any) (sql.Result, error) } -func recordActionAuditSQL(exec sqlExecutor, record ActionAuditRecord) error { +func actionAuditSQLArgs(record ActionAuditRecord) ([]any, error) { requestJSON, err := json.Marshal(record.Request) if err != nil { - return fmt.Errorf("marshal action request: %w", err) + return nil, fmt.Errorf("marshal action request: %w", err) } planJSON, err := json.Marshal(record.Plan) if err != nil { - return fmt.Errorf("marshal action plan: %w", err) + return nil, fmt.Errorf("marshal action plan: %w", err) } approvalsJSON, err := json.Marshal(record.Approvals) if err != nil { - return fmt.Errorf("marshal action approvals: %w", err) + return nil, fmt.Errorf("marshal action approvals: %w", err) } resultJSON, err := json.Marshal(record.Result) if err != nil { - return fmt.Errorf("marshal action result: %w", err) + return nil, fmt.Errorf("marshal action result: %w", err) } verificationOutcomeJSON, err := json.Marshal(record.VerificationOutcome) if err != nil { - return fmt.Errorf("marshal verification outcome: %w", err) + return nil, fmt.Errorf("marshal verification outcome: %w", err) } var originJSON any if origin := NormalizeActionOrigin(record.Origin); origin != nil { encoded, err := json.Marshal(origin) if err != nil { - return fmt.Errorf("marshal action origin: %w", err) + return nil, fmt.Errorf("marshal action origin: %w", err) } originJSON = string(encoded) } - _, err = exec.Exec(` - INSERT INTO action_audits (id, action_id, canonical_id, request_id, created_at, updated_at, state, request_json, plan_json, approvals_json, result_json, verification_outcome_json, origin_json) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT(id) DO UPDATE SET - action_id=excluded.action_id, - canonical_id=excluded.canonical_id, - request_id=excluded.request_id, - created_at=excluded.created_at, - updated_at=excluded.updated_at, - state=excluded.state, - request_json=excluded.request_json, - plan_json=excluded.plan_json, - approvals_json=excluded.approvals_json, - result_json=excluded.result_json, - verification_outcome_json=excluded.verification_outcome_json, - origin_json=excluded.origin_json - `, record.ID, record.ID, CanonicalResourceID(record.Request.ResourceID), record.Request.RequestID, record.CreatedAt, record.UpdatedAt, string(record.State), string(requestJSON), string(planJSON), string(approvalsJSON), string(resultJSON), string(verificationOutcomeJSON), originJSON) - if err != nil { - return fmt.Errorf("insert action audit: %w", err) - } - return nil + return []any{record.ID, record.ID, CanonicalResourceID(record.Request.ResourceID), record.Request.RequestID, record.CreatedAt, record.UpdatedAt, string(record.State), string(requestJSON), string(planJSON), string(approvalsJSON), string(resultJSON), string(verificationOutcomeJSON), originJSON}, nil } -func (s *SQLiteResourceStore) RecordActionAudit(record ActionAuditRecord) error { +func insertActionAuditSQL(exec sqlExecutor, record ActionAuditRecord) (bool, error) { + args, err := actionAuditSQLArgs(record) + if err != nil { + return false, err + } + result, err := exec.Exec(` + INSERT INTO action_audits (id, action_id, canonical_id, request_id, created_at, updated_at, state, request_json, plan_json, approvals_json, result_json, verification_outcome_json, origin_json) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(id) DO NOTHING + `, args...) + if err != nil { + return false, fmt.Errorf("insert action audit: %w", err) + } + rows, err := result.RowsAffected() + if err != nil { + return false, fmt.Errorf("read inserted action audit rows: %w", err) + } + return rows == 1, nil +} + +func updateActionAuditSQL(exec sqlExecutor, record ActionAuditRecord, expectedStates ...ActionState) (bool, error) { + args, err := actionAuditSQLArgs(record) + if err != nil { + return false, err + } + setArgs := append([]any(nil), args[1:]...) + setArgs = append(setArgs, record.ID) + placeholders := make([]string, len(expectedStates)) + for i, state := range expectedStates { + placeholders[i] = "?" + setArgs = append(setArgs, string(state)) + } + setArgs = append(setArgs, args[7], args[8], firstNonNilString(args[12])) + result, err := exec.Exec(` + UPDATE action_audits SET + action_id=?, canonical_id=?, request_id=?, created_at=?, updated_at=?, state=?, + request_json=?, plan_json=?, approvals_json=?, result_json=?, verification_outcome_json=?, origin_json=? + WHERE id=? AND state IN (`+strings.Join(placeholders, ",")+`) + AND request_json=? AND plan_json=? AND COALESCE(origin_json, '')=? + `, setArgs...) + if err != nil { + return false, fmt.Errorf("update action audit: %w", err) + } + rows, err := result.RowsAffected() + if err != nil { + return false, fmt.Errorf("read updated action audit rows: %w", err) + } + return rows == 1, nil +} + +func firstNonNilString(value any) string { + if value == nil { + return "" + } + return fmt.Sprint(value) +} + +func normalizeActionAuditCreation(record ActionAuditRecord, initialEvents []ActionLifecycleEvent) (ActionAuditRecord, []ActionLifecycleEvent, error) { normalized, err := NormalizeActionAuditRecord(record) if err != nil { - return err + return ActionAuditRecord{}, nil, err } // Redact known secret shapes from operator-authored text and command // output before persisting. The audit log is plaintext SQL; raw // secrets pasted into a reason field or echoed in command output // must not be retained. record = RedactAuditRecord(normalized) + events := make([]ActionLifecycleEvent, len(initialEvents)) + seen := make(map[ActionState]struct{}, len(initialEvents)) + for i, event := range initialEvents { + event, err = NormalizeActionLifecycleEvent(event) + if err != nil { + return ActionAuditRecord{}, nil, err + } + if event.ActionID != record.ID { + return ActionAuditRecord{}, nil, fmt.Errorf("initial lifecycle event id %q does not match action audit id %q", event.ActionID, record.ID) + } + if _, duplicate := seen[event.State]; duplicate { + return ActionAuditRecord{}, nil, fmt.Errorf("duplicate initial lifecycle state %q", event.State) + } + seen[event.State] = struct{}{} + events[i] = event + } + return record, events, nil +} + +func (s *SQLiteResourceStore) CreateActionAudit(record ActionAuditRecord, initialEvents []ActionLifecycleEvent) (ActionAuditRecord, bool, error) { + record, events, err := normalizeActionAuditCreation(record, initialEvents) + if err != nil { + return ActionAuditRecord{}, false, err + } s.mu.Lock() defer s.mu.Unlock() - return recordActionAuditSQL(s.db, record) + tx, err := s.db.Begin() + if err != nil { + return ActionAuditRecord{}, false, fmt.Errorf("begin action audit creation transaction: %w", err) + } + committed := false + defer func() { + if !committed { + _ = tx.Rollback() + } + }() + created, err := insertActionAuditSQL(tx, record) + if err != nil { + return ActionAuditRecord{}, false, err + } + if !created { + current, found, err := getActionAuditFrom(tx, record.ID) + if err != nil { + return ActionAuditRecord{}, false, err + } + if !found { + return ActionAuditRecord{}, false, fmt.Errorf("action audit %q disappeared during creation", record.ID) + } + if !ActionAuditIdentityMatches(current, record) { + return current, false, ErrActionIdentityConflict + } + if err := tx.Commit(); err != nil { + return ActionAuditRecord{}, false, fmt.Errorf("commit action audit replay transaction: %w", err) + } + committed = true + return current, false, nil + } + for _, event := range events { + if err := recordActionLifecycleEventSQL(tx, event); err != nil { + return ActionAuditRecord{}, false, err + } + } + if err := tx.Commit(); err != nil { + return ActionAuditRecord{}, false, fmt.Errorf("commit action audit creation transaction: %w", err) + } + committed = true + return record, true, nil +} + +// RecordActionAudit is retained as an insert-only compatibility boundary. +// Production lifecycle code must use CreateActionAudit or a typed transition; +// replacing an existing record would permit state rewind. +func (s *SQLiteResourceStore) RecordActionAudit(record ActionAuditRecord) error { + _, created, err := s.CreateActionAudit(record, nil) + if err != nil { + return err + } + if !created { + return ErrActionAuditAlreadyExists + } + return nil } type actionAuditScanner interface { @@ -1974,6 +2112,60 @@ func recordActionLifecycleEventSQL(exec sqlExecutor, event ActionLifecycleEvent) return nil } +func actionTransitionConflict(current ActionAuditRecord, desired ActionAuditRecord, fallback error) error { + if !ActionAuditIdentityMatches(current, desired) { + return ErrActionIdentityConflict + } + switch current.State { + case ActionStateRejected, ActionStateCompleted, ActionStateFailed: + return ErrActionExecutionFinal + case ActionStateExecuting: + if desired.State == ActionStateExecuting { + return ErrActionAlreadyExecuting + } + return ErrActionNotExecuting + default: + return fallback + } +} + +func (s *SQLiteResourceStore) recordActionTransition(record ActionAuditRecord, event ActionLifecycleEvent, expectedStates []ActionState, fallback error) error { + s.mu.Lock() + defer s.mu.Unlock() + tx, err := s.db.Begin() + if err != nil { + return fmt.Errorf("begin action transition transaction: %w", err) + } + committed := false + defer func() { + if !committed { + _ = tx.Rollback() + } + }() + updated, err := updateActionAuditSQL(tx, record, expectedStates...) + if err != nil { + return err + } + if !updated { + current, found, err := getActionAuditFrom(tx, record.ID) + if err != nil { + return err + } + if !found { + return fmt.Errorf("action audit %q not found", record.ID) + } + return actionTransitionConflict(current, record, fallback) + } + if err := recordActionLifecycleEventSQL(tx, event); err != nil { + return err + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("commit action transition transaction: %w", err) + } + committed = true + return nil +} + func (s *SQLiteResourceStore) RecordActionDecision(record ActionAuditRecord, event ActionLifecycleEvent) error { normalizedRecord, err := NormalizeActionAuditRecord(record) if err != nil { @@ -1987,41 +2179,7 @@ func (s *SQLiteResourceStore) RecordActionDecision(record ActionAuditRecord, eve return fmt.Errorf("action decision event id %q does not match action audit id %q", normalizedEvent.ActionID, normalizedRecord.ID) } - s.mu.Lock() - defer s.mu.Unlock() - - current, ok, err := s.getActionAudit(normalizedRecord.ID) - if err != nil { - return err - } - if !ok { - return fmt.Errorf("action audit %q not found", normalizedRecord.ID) - } - if current.State != ActionStatePending { - return ErrActionNotPending - } - - tx, err := s.db.Begin() - if err != nil { - return fmt.Errorf("begin action decision transaction: %w", err) - } - committed := false - defer func() { - if !committed { - _ = tx.Rollback() - } - }() - if err := recordActionAuditSQL(tx, normalizedRecord); err != nil { - return err - } - if err := recordActionLifecycleEventSQL(tx, normalizedEvent); err != nil { - return err - } - if err := tx.Commit(); err != nil { - return fmt.Errorf("commit action decision transaction: %w", err) - } - committed = true - return nil + return s.recordActionTransition(normalizedRecord, normalizedEvent, []ActionState{ActionStatePending}, ErrActionNotPending) } func (s *SQLiteResourceStore) RecordActionExecutionStart(record ActionAuditRecord, event ActionLifecycleEvent) error { @@ -2043,41 +2201,17 @@ func (s *SQLiteResourceStore) RecordActionExecutionStart(record ActionAuditRecor // output before persisting; see RecordActionAudit for the contract. normalizedRecord = RedactAuditRecord(normalizedRecord) - s.mu.Lock() - defer s.mu.Unlock() - - current, ok, err := s.getActionAudit(normalizedRecord.ID) - if err != nil { - return err + if normalizedRecord.Plan.ApprovalPolicy == ApprovalDryRun { + return ErrActionDryRunOnly } - if !ok { - return fmt.Errorf("action audit %q not found", normalizedRecord.ID) + if !normalizedRecord.Plan.ExpiresAt.IsZero() && !normalizedEvent.Timestamp.Before(normalizedRecord.Plan.ExpiresAt) { + return ErrActionPlanExpired } - if err := ValidateActionExecutionStart(current, normalizedEvent.Timestamp); err != nil { - return err + expected := []ActionState{ActionStateApproved} + if normalizedRecord.Plan.Allowed && !normalizedRecord.Plan.RequiresApproval { + expected = append(expected, ActionStatePlanned) } - - tx, err := s.db.Begin() - if err != nil { - return fmt.Errorf("begin action execution start transaction: %w", err) - } - committed := false - defer func() { - if !committed { - _ = tx.Rollback() - } - }() - if err := recordActionAuditSQL(tx, normalizedRecord); err != nil { - return err - } - if err := recordActionLifecycleEventSQL(tx, normalizedEvent); err != nil { - return err - } - if err := tx.Commit(); err != nil { - return fmt.Errorf("commit action execution start transaction: %w", err) - } - committed = true - return nil + return s.recordActionTransition(normalizedRecord, normalizedEvent, expected, ErrActionNotApproved) } func (s *SQLiteResourceStore) RecordActionExecutionResult(record ActionAuditRecord, event ActionLifecycleEvent) error { @@ -2102,49 +2236,39 @@ func (s *SQLiteResourceStore) RecordActionExecutionResult(record ActionAuditReco // output before persisting; see RecordActionAudit for the contract. normalizedRecord = RedactAuditRecord(normalizedRecord) - s.mu.Lock() - defer s.mu.Unlock() + return s.recordActionTransition(normalizedRecord, normalizedEvent, []ActionState{ActionStateExecuting}, ErrActionNotExecuting) +} - current, ok, err := s.getActionAudit(normalizedRecord.ID) +func (s *SQLiteResourceStore) RecordActionExecutionRefusal(record ActionAuditRecord, event ActionLifecycleEvent) error { + normalizedRecord, err := NormalizeActionAuditRecord(record) if err != nil { return err } - if !ok { - return fmt.Errorf("action audit %q not found", normalizedRecord.ID) - } - if current.State != ActionStateExecuting { - return ErrActionNotExecuting - } - - tx, err := s.db.Begin() + normalizedEvent, err := NormalizeActionLifecycleEvent(event) if err != nil { - return fmt.Errorf("begin action execution result transaction: %w", err) - } - committed := false - defer func() { - if !committed { - _ = tx.Rollback() - } - }() - if err := recordActionAuditSQL(tx, normalizedRecord); err != nil { return err } - if err := recordActionLifecycleEventSQL(tx, normalizedEvent); err != nil { - return err + if normalizedRecord.State != ActionStateFailed || normalizedEvent.State != ActionStateFailed || normalizedEvent.ActionID != normalizedRecord.ID { + return fmt.Errorf("action execution refusal must persist matching failed state") } - if err := tx.Commit(); err != nil { - return fmt.Errorf("commit action execution result transaction: %w", err) - } - committed = true - return nil + normalizedRecord = RedactAuditRecord(normalizedRecord) + return s.recordActionTransition(normalizedRecord, normalizedEvent, []ActionState{ActionStatePlanned, ActionStatePending, ActionStateApproved}, ErrActionExecutionFinal) } func (s *SQLiteResourceStore) getActionAudit(actionID string) (ActionAuditRecord, bool, error) { + return getActionAuditFrom(s.db, actionID) +} + +type actionAuditQueryRower interface { + QueryRow(query string, args ...any) *sql.Row +} + +func getActionAuditFrom(queryer actionAuditQueryRower, actionID string) (ActionAuditRecord, bool, error) { actionID = strings.TrimSpace(actionID) if actionID == "" { return ActionAuditRecord{}, false, nil } - row := s.db.QueryRow(` + row := queryer.QueryRow(` SELECT id, action_id, request_id, created_at, updated_at, state, request_json, plan_json, approvals_json, result_json, verification_outcome_json, origin_json FROM action_audits WHERE id = ? @@ -2946,27 +3070,47 @@ func changeMatchesResource(change ResourceChange, canonicalIDs []string, include return false } -func (m *MemoryStore) RecordActionAudit(record ActionAuditRecord) error { - normalized, err := NormalizeActionAuditRecord(record) +func (m *MemoryStore) CreateActionAudit(record ActionAuditRecord, initialEvents []ActionLifecycleEvent) (ActionAuditRecord, bool, error) { + record, events, err := normalizeActionAuditCreation(record, initialEvents) if err != nil { - return err + return ActionAuditRecord{}, false, err } // Redact secret shapes from operator-authored fields and command // output before persisting in-memory too. The MemoryStore is used in // tests and contract examples; redaction must apply uniformly so test // fixtures cannot accidentally exercise an unredacted persistence // path that production never sees. - record = RedactAuditRecord(normalized) - m.mu.Lock() defer m.mu.Unlock() for i := range m.actionAudits { if m.actionAudits[i].ID == record.ID { - m.actionAudits[i] = record - return nil + current := m.actionAudits[i] + if !ActionAuditIdentityMatches(current, record) { + return current, false, ErrActionIdentityConflict + } + return current, false, nil + } + } + for _, event := range events { + for _, existing := range m.actionLifecycleEvents { + if existing.ActionID == event.ActionID && existing.State == event.State { + return ActionAuditRecord{}, false, fmt.Errorf("action lifecycle state %q already recorded for %q", event.State, event.ActionID) + } } } m.actionAudits = append(m.actionAudits, record) + m.actionLifecycleEvents = append(m.actionLifecycleEvents, events...) + return record, true, nil +} + +func (m *MemoryStore) RecordActionAudit(record ActionAuditRecord) error { + _, created, err := m.CreateActionAudit(record, nil) + if err != nil { + return err + } + if !created { + return ErrActionAuditAlreadyExists + } return nil } @@ -3050,7 +3194,10 @@ func (m *MemoryStore) RecordActionDecision(record ActionAuditRecord, event Actio for i := range m.actionAudits { if m.actionAudits[i].ID == normalizedRecord.ID { if m.actionAudits[i].State != ActionStatePending { - return ErrActionNotPending + return actionTransitionConflict(m.actionAudits[i], normalizedRecord, ErrActionNotPending) + } + if !ActionAuditIdentityMatches(m.actionAudits[i], normalizedRecord) { + return ErrActionIdentityConflict } m.actionAudits[i] = normalizedRecord replaced = true @@ -3088,6 +3235,9 @@ func (m *MemoryStore) RecordActionExecutionStart(record ActionAuditRecord, event replaced := false for i := range m.actionAudits { if m.actionAudits[i].ID == normalizedRecord.ID { + if !ActionAuditIdentityMatches(m.actionAudits[i], normalizedRecord) { + return ErrActionIdentityConflict + } if err := ValidateActionExecutionStart(m.actionAudits[i], normalizedEvent.Timestamp); err != nil { return err } @@ -3131,7 +3281,10 @@ func (m *MemoryStore) RecordActionExecutionResult(record ActionAuditRecord, even for i := range m.actionAudits { if m.actionAudits[i].ID == normalizedRecord.ID { if m.actionAudits[i].State != ActionStateExecuting { - return ErrActionNotExecuting + return actionTransitionConflict(m.actionAudits[i], normalizedRecord, ErrActionNotExecuting) + } + if !ActionAuditIdentityMatches(m.actionAudits[i], normalizedRecord) { + return ErrActionIdentityConflict } m.actionAudits[i] = normalizedRecord replaced = true @@ -3145,6 +3298,41 @@ func (m *MemoryStore) RecordActionExecutionResult(record ActionAuditRecord, even return nil } +func (m *MemoryStore) RecordActionExecutionRefusal(record ActionAuditRecord, event ActionLifecycleEvent) error { + normalizedRecord, err := NormalizeActionAuditRecord(record) + if err != nil { + return err + } + normalizedEvent, err := NormalizeActionLifecycleEvent(event) + if err != nil { + return err + } + if normalizedRecord.State != ActionStateFailed || normalizedEvent.State != ActionStateFailed || normalizedEvent.ActionID != normalizedRecord.ID { + return fmt.Errorf("action execution refusal must persist matching failed state") + } + normalizedRecord = RedactAuditRecord(normalizedRecord) + m.mu.Lock() + defer m.mu.Unlock() + for i := range m.actionAudits { + if m.actionAudits[i].ID != normalizedRecord.ID { + continue + } + current := m.actionAudits[i] + if !ActionAuditIdentityMatches(current, normalizedRecord) { + return ErrActionIdentityConflict + } + switch current.State { + case ActionStatePlanned, ActionStatePending, ActionStateApproved: + m.actionAudits[i] = normalizedRecord + m.actionLifecycleEvents = append(m.actionLifecycleEvents, normalizedEvent) + return nil + default: + return actionTransitionConflict(current, normalizedRecord, ErrActionExecutionFinal) + } + } + return fmt.Errorf("action audit %q not found", normalizedRecord.ID) +} + func (m *MemoryStore) GetActionAudits(canonicalID string, since time.Time, limit int) ([]ActionAuditRecord, error) { m.mu.RLock() defer m.mu.RUnlock() @@ -3175,6 +3363,11 @@ func (m *MemoryStore) RecordActionLifecycleEvent(event ActionLifecycleEvent) err m.mu.Lock() defer m.mu.Unlock() + for _, existing := range m.actionLifecycleEvents { + if existing.ActionID == event.ActionID && existing.State == event.State { + return fmt.Errorf("action lifecycle state %q already recorded for %q", event.State, event.ActionID) + } + } m.actionLifecycleEvents = append(m.actionLifecycleEvents, event) return nil } diff --git a/internal/unifiedresources/store_test.go b/internal/unifiedresources/store_test.go index 9570b2252..9d208c158 100644 --- a/internal/unifiedresources/store_test.go +++ b/internal/unifiedresources/store_test.go @@ -13,6 +13,360 @@ import ( "time" ) +func atomicLifecycleTestRecord(id string, state ActionState) ActionAuditRecord { + now := time.Date(2026, 7, 11, 12, 0, 0, 0, time.UTC) + return ActionAuditRecord{ + ID: id, CreatedAt: now, UpdatedAt: now, State: state, + Request: ActionRequest{RequestID: "req-" + id, ResourceID: "vm:42", CapabilityName: "restart", Reason: "atomic lifecycle proof", RequestedBy: "agent:test"}, + Plan: ActionPlan{ActionID: id, RequestID: "req-" + id, Allowed: true, RequiresApproval: state == ActionStatePending, ApprovalPolicy: ApprovalAdmin, PlannedAt: now, ExpiresAt: now.Add(time.Hour), ResourceVersion: "resource:sha256:test", PolicyVersion: "policy:sha256:test", PlanHash: "sha256:" + id}, + Origin: &ActionOrigin{Surface: "patrol", FindingID: "finding-1", InvestigationID: "inv-1", ProposalID: "proposal-1"}, + } +} + +func atomicLifecycleInitialEvents(record ActionAuditRecord) []ActionLifecycleEvent { + events := []ActionLifecycleEvent{{ActionID: record.ID, Timestamp: record.CreatedAt, State: ActionStatePlanned, Actor: record.Request.RequestedBy, Message: "Action plan created."}} + if record.State == ActionStatePending { + events = append(events, ActionLifecycleEvent{ActionID: record.ID, Timestamp: record.CreatedAt, State: ActionStatePending, Actor: record.Request.RequestedBy, Message: "Action is waiting for approval before execution."}) + } + return events +} + +func TestMemoryStoreCreateActionAuditConcurrentReturnsCurrent(t *testing.T) { + store := NewMemoryStore() + record := atomicLifecycleTestRecord("act_memory_create", ActionStatePending) + start := make(chan struct{}) + type result struct { + current ActionAuditRecord + created bool + err error + } + results := make(chan result, 2) + for range 2 { + go func() { + <-start + current, created, err := store.CreateActionAudit(record, atomicLifecycleInitialEvents(record)) + results <- result{current: current, created: created, err: err} + }() + } + close(start) + createdCount := 0 + for range 2 { + got := <-results + if got.err != nil { + t.Fatalf("CreateActionAudit: %v", got.err) + } + if got.created { + createdCount++ + } + if got.current.State != ActionStatePending { + t.Fatalf("current state = %q", got.current.State) + } + } + if createdCount != 1 { + t.Fatalf("created count = %d, want 1", createdCount) + } + events, err := store.GetActionLifecycleEvents(record.ID, time.Time{}, 10) + if err != nil || len(events) != 2 { + t.Fatalf("events=%#v err=%v", events, err) + } +} + +func TestMemoryStoreActionTransitionsAreMonotonic(t *testing.T) { + store := NewMemoryStore() + record := atomicLifecycleTestRecord("act_memory_terminal", ActionStatePending) + if _, _, err := store.CreateActionAudit(record, atomicLifecycleInitialEvents(record)); err != nil { + t.Fatal(err) + } + approved, event, err := ApplyActionDecision(record, ActionApprovalRecord{Actor: "operator", Method: MethodAPI, Outcome: OutcomeApproved}, record.CreatedAt.Add(time.Minute)) + if err != nil || store.RecordActionDecision(approved, event) != nil { + t.Fatalf("approve: %v", err) + } + started, startEvent, err := BeginActionExecution(approved, "operator", record.CreatedAt.Add(2*time.Minute)) + if err != nil || store.RecordActionExecutionStart(started, startEvent) != nil { + t.Fatalf("start: %v", err) + } + completed, doneEvent, err := CompleteActionExecution(started, &ExecutionResult{Success: true}, "operator", record.CreatedAt.Add(3*time.Minute)) + if err != nil || store.RecordActionExecutionResult(completed, doneEvent) != nil { + t.Fatalf("complete: %v", err) + } + if err := store.RecordActionDecision(approved, event); !errors.Is(err, ErrActionExecutionFinal) { + t.Fatalf("terminal rewind error=%v", err) + } +} + +func TestMemoryStoreConcurrentExecutionStartHasOneCASWinner(t *testing.T) { + store := NewMemoryStore() + record := atomicLifecycleTestRecord("act_memory_execute", ActionStatePlanned) + record.Plan.RequiresApproval = false + record.Plan.ApprovalPolicy = ApprovalNone + if _, _, err := store.CreateActionAudit(record, atomicLifecycleInitialEvents(record)); err != nil { + t.Fatal(err) + } + started, event, err := BeginActionExecution(record, "operator", record.CreatedAt.Add(time.Minute)) + if err != nil { + t.Fatal(err) + } + start := make(chan struct{}) + errs := make(chan error, 2) + for range 2 { + go func() { <-start; errs <- store.RecordActionExecutionStart(started, event) }() + } + close(start) + successes := 0 + for range 2 { + if err := <-errs; err == nil { + successes++ + } + } + if successes != 1 { + t.Fatalf("execution CAS winners = %d, want 1", successes) + } +} + +func TestSQLiteStoreCreateActionAuditConcurrentAcrossTwoInstances(t *testing.T) { + dir := t.TempDir() + first, err := NewSQLiteResourceStore(dir, "default") + if err != nil { + t.Fatal(err) + } + defer first.Close() + second, err := NewSQLiteResourceStore(dir, "default") + if err != nil { + t.Fatal(err) + } + defer second.Close() + record := atomicLifecycleTestRecord("act_sqlite_create", ActionStatePending) + start := make(chan struct{}) + created := make(chan bool, 2) + errs := make(chan error, 2) + for _, store := range []*SQLiteResourceStore{first, second} { + go func(s *SQLiteResourceStore) { + <-start + _, ok, err := s.CreateActionAudit(record, atomicLifecycleInitialEvents(record)) + created <- ok + errs <- err + }(store) + } + close(start) + createdCount := 0 + for range 2 { + if err := <-errs; err != nil { + t.Fatal(err) + } + if <-created { + createdCount++ + } + } + if createdCount != 1 { + t.Fatalf("created count=%d, want 1", createdCount) + } + events, err := first.GetActionLifecycleEvents(record.ID, time.Time{}, 10) + if err != nil || len(events) != 2 { + t.Fatalf("events=%#v err=%v", events, err) + } +} + +func TestSQLiteStoreActionTransitionCASAcrossTwoInstances(t *testing.T) { + dir := t.TempDir() + first, err := NewSQLiteResourceStore(dir, "default") + if err != nil { + t.Fatal(err) + } + defer first.Close() + second, err := NewSQLiteResourceStore(dir, "default") + if err != nil { + t.Fatal(err) + } + defer second.Close() + record := atomicLifecycleTestRecord("act_sqlite_decide", ActionStatePending) + if _, _, err := first.CreateActionAudit(record, atomicLifecycleInitialEvents(record)); err != nil { + t.Fatal(err) + } + approved, approvedEvent, _ := ApplyActionDecision(record, ActionApprovalRecord{Actor: "one", Outcome: OutcomeApproved}, record.CreatedAt.Add(time.Minute)) + rejected, rejectedEvent, _ := ApplyActionDecision(record, ActionApprovalRecord{Actor: "two", Outcome: OutcomeRejected}, record.CreatedAt.Add(time.Minute)) + start := make(chan struct{}) + errs := make(chan error, 2) + go func() { <-start; errs <- first.RecordActionDecision(approved, approvedEvent) }() + go func() { <-start; errs <- second.RecordActionDecision(rejected, rejectedEvent) }() + close(start) + successes := 0 + for range 2 { + if err := <-errs; err == nil { + successes++ + } + } + if successes != 1 { + t.Fatalf("decision CAS winners=%d, want 1", successes) + } +} + +func TestSQLiteStoreConcurrentExecutionStartAcrossTwoInstancesHasOneWinner(t *testing.T) { + dir := t.TempDir() + first, err := NewSQLiteResourceStore(dir, "default") + if err != nil { + t.Fatal(err) + } + defer first.Close() + second, err := NewSQLiteResourceStore(dir, "default") + if err != nil { + t.Fatal(err) + } + defer second.Close() + record := atomicLifecycleTestRecord("act_sqlite_execute", ActionStatePlanned) + record.Plan.RequiresApproval = false + record.Plan.ApprovalPolicy = ApprovalNone + if _, _, err := first.CreateActionAudit(record, atomicLifecycleInitialEvents(record)); err != nil { + t.Fatal(err) + } + started, event, _ := BeginActionExecution(record, "operator", record.CreatedAt.Add(time.Minute)) + start := make(chan struct{}) + errs := make(chan error, 2) + for _, store := range []*SQLiteResourceStore{first, second} { + go func(s *SQLiteResourceStore) { <-start; errs <- s.RecordActionExecutionStart(started, event) }(store) + } + close(start) + successes := 0 + for range 2 { + if err := <-errs; err == nil { + successes++ + } + } + if successes != 1 { + t.Fatalf("execution CAS winners=%d, want 1", successes) + } +} + +func TestSQLiteStoreCreateActionAuditRollsBackWhenInitialEventInsertFails(t *testing.T) { + store := newTestStore(t) + if _, err := store.db.Exec(`CREATE TRIGGER fail_initial_event BEFORE INSERT ON action_lifecycle_events BEGIN SELECT RAISE(ABORT, 'forced event failure'); END`); err != nil { + t.Fatal(err) + } + record := atomicLifecycleTestRecord("act_create_rollback", ActionStatePending) + if _, _, err := store.CreateActionAudit(record, atomicLifecycleInitialEvents(record)); err == nil { + t.Fatal("expected creation failure") + } + if _, found, err := store.GetActionAudit(record.ID); err != nil || found { + t.Fatalf("found=%v err=%v", found, err) + } +} + +func TestSQLiteStoreActionTransitionRollsBackWhenEventInsertFails(t *testing.T) { + store := newTestStore(t) + record := atomicLifecycleTestRecord("act_transition_rollback", ActionStatePending) + if _, _, err := store.CreateActionAudit(record, atomicLifecycleInitialEvents(record)); err != nil { + t.Fatal(err) + } + if _, err := store.db.Exec(`CREATE TRIGGER fail_transition_event BEFORE INSERT ON action_lifecycle_events BEGIN SELECT RAISE(ABORT, 'forced event failure'); END`); err != nil { + t.Fatal(err) + } + approved, event, _ := ApplyActionDecision(record, ActionApprovalRecord{Actor: "operator", Outcome: OutcomeApproved}, record.CreatedAt.Add(time.Minute)) + if err := store.RecordActionDecision(approved, event); err == nil { + t.Fatal("expected transition failure") + } + current, found, err := store.GetActionAudit(record.ID) + if err != nil || !found || current.State != ActionStatePending { + t.Fatalf("current=%#v found=%v err=%v", current, found, err) + } +} + +func TestSQLiteStoreLifecycleRestartPreservesMonotonicState(t *testing.T) { + dir := t.TempDir() + store, err := NewSQLiteResourceStore(dir, "default") + if err != nil { + t.Fatal(err) + } + record := atomicLifecycleTestRecord("act_restart_terminal", ActionStatePlanned) + record.Plan.RequiresApproval = false + record.Plan.ApprovalPolicy = ApprovalNone + if _, _, err := store.CreateActionAudit(record, atomicLifecycleInitialEvents(record)); err != nil { + t.Fatal(err) + } + started, startEvent, _ := BeginActionExecution(record, "operator", record.CreatedAt.Add(time.Minute)) + if err := store.RecordActionExecutionStart(started, startEvent); err != nil { + t.Fatal(err) + } + completed, doneEvent, _ := CompleteActionExecution(started, &ExecutionResult{Success: true}, "operator", record.CreatedAt.Add(2*time.Minute)) + if err := store.RecordActionExecutionResult(completed, doneEvent); err != nil { + t.Fatal(err) + } + if err := store.Close(); err != nil { + t.Fatal(err) + } + reopened, err := NewSQLiteResourceStore(dir, "default") + if err != nil { + t.Fatal(err) + } + defer reopened.Close() + current, created, err := reopened.CreateActionAudit(record, atomicLifecycleInitialEvents(record)) + if err != nil || created || current.State != ActionStateCompleted { + t.Fatalf("current=%#v created=%v err=%v", current, created, err) + } +} + +func TestSQLiteStoreRestartDoesNotReadmitExecutingAction(t *testing.T) { + dir := t.TempDir() + store, err := NewSQLiteResourceStore(dir, "default") + if err != nil { + t.Fatal(err) + } + record := atomicLifecycleTestRecord("act_restart_executing", ActionStatePlanned) + record.Plan.RequiresApproval = false + record.Plan.ApprovalPolicy = ApprovalNone + if _, _, err := store.CreateActionAudit(record, atomicLifecycleInitialEvents(record)); err != nil { + t.Fatal(err) + } + started, event, _ := BeginActionExecution(record, "operator", record.CreatedAt.Add(time.Minute)) + if err := store.RecordActionExecutionStart(started, event); err != nil { + t.Fatal(err) + } + if err := store.Close(); err != nil { + t.Fatal(err) + } + reopened, err := NewSQLiteResourceStore(dir, "default") + if err != nil { + t.Fatal(err) + } + defer reopened.Close() + if err := reopened.RecordActionExecutionStart(started, event); !errors.Is(err, ErrActionAlreadyExecuting) { + t.Fatalf("error=%v", err) + } +} + +func TestSQLiteActionLifecycleMigrationDeduplicatesStateEvents(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "resources", "unified_resources.db") + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + t.Fatal(err) + } + db, err := sql.Open("sqlite", path) + if err != nil { + t.Fatal(err) + } + if _, err := db.Exec(`CREATE TABLE action_lifecycle_events (id INTEGER PRIMARY KEY AUTOINCREMENT, action_id TEXT NOT NULL, timestamp DATETIME NOT NULL, state TEXT NOT NULL, actor TEXT, message TEXT)`); err != nil { + t.Fatal(err) + } + for range 2 { + if _, err := db.Exec(`INSERT INTO action_lifecycle_events (action_id, timestamp, state, actor, message) VALUES (?, ?, ?, '', '')`, "act_event_dedupe", time.Now().UTC(), string(ActionStatePlanned)); err != nil { + t.Fatal(err) + } + } + if err := db.Close(); err != nil { + t.Fatal(err) + } + store, err := NewSQLiteResourceStore(dir, "default") + if err != nil { + t.Fatal(err) + } + defer store.Close() + events, err := store.GetActionLifecycleEvents("act_event_dedupe", time.Time{}, 10) + if err != nil || len(events) != 1 { + t.Fatalf("events=%#v err=%v", events, err) + } + if err := store.RecordActionLifecycleEvent(events[0]); err == nil { + t.Fatal("duplicate state event should be rejected after migration") + } +} + func TestSanitizeOrgID_AllowsSafeChars(t *testing.T) { in := "Acme_Org-123" if got := sanitizeOrgID(in); got != in { @@ -1652,7 +2006,7 @@ func TestActionAuditRecord_RoundTripMalformedUnrunVerificationScrubsSQLiteRead(t } } -func TestMemoryStore_RecordActionAudit_UpsertsByID(t *testing.T) { +func TestMemoryStore_RecordActionAudit_IsInsertOnly(t *testing.T) { store := NewMemoryStore() now := time.Date(2026, 3, 18, 13, 30, 0, 0, time.UTC) @@ -1676,8 +2030,8 @@ func TestMemoryStore_RecordActionAudit_UpsertsByID(t *testing.T) { if err := store.RecordActionAudit(first); err != nil { t.Fatalf("RecordActionAudit(first): %v", err) } - if err := store.RecordActionAudit(second); err != nil { - t.Fatalf("RecordActionAudit(second): %v", err) + if err := store.RecordActionAudit(second); !errors.Is(err, ErrActionAuditAlreadyExists) { + t.Fatalf("RecordActionAudit(second) error = %v, want ErrActionAuditAlreadyExists", err) } results, err := store.GetActionAudits("vm:301", now.Add(-time.Hour), 10) @@ -1687,11 +2041,11 @@ func TestMemoryStore_RecordActionAudit_UpsertsByID(t *testing.T) { if len(results) != 1 { t.Fatalf("expected 1 action audit after upsert, got %d", len(results)) } - if results[0].State != ActionStateCompleted { - t.Fatalf("expected latest action state to win, got %q", results[0].State) + if results[0].State != ActionStatePlanned { + t.Fatalf("insert-only audit state = %q, want planned", results[0].State) } - if results[0].Result == nil || results[0].Result.Output != "done" { - t.Fatalf("expected latest action result to win, got %+v", results[0].Result) + if results[0].Result != nil { + t.Fatalf("duplicate insert rewrote result: %+v", results[0].Result) } } @@ -1879,8 +2233,8 @@ func TestRecordActionExecutionStartAndResult_UpdatesAuditAndAppendsLifecycle(t * if !ok || got.State != ActionStateCompleted || got.Result == nil || got.Result.Output != "done" { t.Fatalf("completed audit = %#v, %v", got, ok) } - if err := store.RecordActionExecutionResult(completed, doneEvent); !errors.Is(err, ErrActionNotExecuting) { - t.Fatalf("stale RecordActionExecutionResult error = %v, want %v", err, ErrActionNotExecuting) + if err := store.RecordActionExecutionResult(completed, doneEvent); !errors.Is(err, ErrActionExecutionFinal) { + t.Fatalf("stale RecordActionExecutionResult error = %v, want %v", err, ErrActionExecutionFinal) } events, err := store.GetActionLifecycleEvents("act_execution", time.Time{}, 10)