From d6838d3a257529f0fcfe549d2f8618489e04c4b1 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Sun, 12 Jul 2026 05:23:02 +0100 Subject: [PATCH] Complete durable APT workflow continuity --- docs/release-control/v6/internal/status.json | 6 +- .../v6/internal/subsystems/agent-lifecycle.md | 26 +- .../v6/internal/subsystems/ai-runtime.md | 21 +- .../v6/internal/subsystems/api-contracts.md | 32 +- .../internal/subsystems/storage-recovery.md | 9 +- internal/agentexec/apt_codec.go | 72 ++++- internal/agentexec/apt_codec_test.go | 36 ++- internal/agentexec/operation_query_test.go | 79 ++++- internal/agentexec/server_websocket_test.go | 2 +- internal/agentexec/types.go | 23 +- internal/ai/findings_apt_workflows.go | 58 +++- internal/ai/findings_apt_workflows_test.go | 7 +- internal/api/apt_workflow_integration_test.go | 296 ++++++++++++++++++ internal/api/contract_test.go | 35 ++- internal/api/host_apt_action_result.go | 31 +- internal/api/host_apt_action_result_test.go | 26 +- internal/api/host_apt_partial_truth_test.go | 184 +++++++++++ .../host_storage_cleanup_action_executor.go | 23 +- ...st_storage_cleanup_action_executor_test.go | 71 ++++- internal/api/host_update_action_executor.go | 30 +- .../api/host_update_action_executor_test.go | 111 ++++++- .../hostagent/commands_host_update_test.go | 3 + ...tion_receipt_websocket_integration_test.go | 114 +++++++ internal/hostagent/package_updates.go | 45 ++- internal/hostagent/package_updates_test.go | 143 ++++++++- internal/hostagent/storage_cleanup_test.go | 42 +++ 26 files changed, 1407 insertions(+), 118 deletions(-) create mode 100644 internal/api/apt_workflow_integration_test.go create mode 100644 internal/api/host_apt_partial_truth_test.go diff --git a/docs/release-control/v6/internal/status.json b/docs/release-control/v6/internal/status.json index 37b74c913..b99563cd2 100644 --- a/docs/release-control/v6/internal/status.json +++ b/docs/release-control/v6/internal/status.json @@ -4000,7 +4000,7 @@ "./internal/hostagent", "./internal/unifiedresources", "-run", - "TestPatrolFullModeRunsStorageCleanupThroughCanonicalLifecycle|TestExecuteHostStorageCleanupRoundTripUsesPathAndCommandFreeEnvelope|TestStorageCleanupManagerApplyUsesClosedAPTCatalogAndVerifiesBytes|TestStorageCleanupManagerRefusesFingerprintDriftBeforeMutation|TestHostStorageCleanupTargetDiskUsesLongestContainingMount", + "TestAPTCacheCleanupDetectorProposalApprovalDispatchAuditAndFindingReconciliation|TestAPTCacheCleanupCallbackLossServerRestartReconcilesTerminalAuditAndFindingWithoutResend|TestHostStorageCleanupReconcileDelayedTerminalReceiptPreservesAgentAttestedEvidenceWithoutResend|TestHostStorageCleanupPartialTruthIsNonRollbackableAndRequiresRescan|TestPatrolFullModeRunsStorageCleanupThroughCanonicalLifecycle|TestExecuteHostStorageCleanupRoundTripUsesPathAndCommandFreeEnvelope|TestStorageCleanupManagerApplyUsesClosedAPTCatalogAndVerifiesBytes|TestStorageCleanupManagerFailurePhasesPreserveMeasuredEffect|TestStorageCleanupManagerRefusesFingerprintDriftBeforeMutation|TestHostStorageCleanupTargetDiskUsesLongestContainingMount", "-count=1" ] }, @@ -4013,7 +4013,7 @@ "./internal/agentexec", "./internal/hostagent", "-run", - "TestPatrolFullModeRunsHostUpdateThroughCanonicalLifecycle|TestExecuteHostUpdateRoundTripUsesTypedCommandFreeEnvelope|TestPackageUpdateManagerApplyUsesClosedAPTCommandCatalogAndVerifies|TestPackageUpdateManagerRefusesRefreshTimeInventoryDriftBeforeInstall", + "TestAPTUpdateDetectorProposalApprovalDispatchAuditAndFindingReconciliation|TestAPTUpdateCallbackLossServerRestartReconcilesTerminalAuditAndFindingWithoutResend|TestHostUpdateReconcileDelayedTerminalReceiptPreservesAgentAttestedEvidenceWithoutResend|TestHostUpdateReconcileLegacyTerminalReceiptDowngradesUnknownHealthWithoutResend|TestHostUpdatePartialTruthProjectsPhaseHealthRemainingAndRecovery|TestLegacyHostUpdateResultWithoutHealthFieldsRemainsStrictlyDecodableAsUnknown|TestPackageUpdateManagerApplyUsesClosedAPTCommandCatalogAndVerifies|TestPackageUpdateManagerFailsClosedWhenRefreshFails|TestPackageUpdateManagerPreInstallHealthUnknownRefusesWithoutMutationOrRecoveryClaim|TestPackageUpdateManagerPartialInstallAndVerifyFailureCarryExplicitHealthAndRecovery|TestPackageUpdateManagerRefusesRefreshTimeInventoryDriftBeforeInstall|TestPackageUpdateManagerZeroPendingAfterRefreshIsDriftWithoutInstallOrVerifiedClaim|TestRealServerAndUnifiedAgentWebSocketExecutesAPTThroughFakeTypedManagersAndReplaysWithoutMutation", "-count=1" ] }, @@ -7981,7 +7981,7 @@ }, { "id": "action-governance-auditability-post-rc-hardening", - "summary": "Task 07 Phase B1 has landed the core durable delivery/reconnect foundation: transactional action admission plus deterministic attempt/outbox, one-shot pre-send CAS, receipt correlation, explicit expiry, query-only restart reconciliation, and tenant-scoped detail/pending/settled APIs. Phase B2 generic agent-operation receipt continuity is implemented locally pending commit gate: keyed SQLite admission/start/terminal transitions, immutable attempt/action/operation/digest/agent binding, strict versioned sanitized terminal envelopes, payload compaction to permanent replay-denial tombstones, exact authenticated query correlation, legacy capability fail-closed gating, and APT update/cleanup adapters over the generic owner. Tombstone metadata is monotonic and not claimed bounded. Task 10 remains sole owner of execution, verification, evidence, compensation, and rollback truth. Claims 16 and 17 remain open until Task 09 resumes, browser/lab evidence passes, and Task 12 certifies. Mobile/relay consumption remains separate with no local workaround for absent core APIs. Raw model command, file write, arbitrary pod exec, legacy run_command, /api/ai/run-command, and enterprise command remediation remain retired with no replacement or executable historical authority.", + "summary": "Task 07 durable delivery and generic agent-operation receipts are accepted: transactional dispatch admission, one-shot send CAS, immutable attempt/action/operation/digest/agent binding, query-only restart reconciliation, strict sanitized terminal envelopes, and permanent replay-denial tombstones. Task 09 now consumes that owner for both APT workflows and completes the fake-only detector-to-finding, exact empty-parameter proposal, shared policy/approval, typed dispatch, delayed/callback-loss/reopened-server receipt reconciliation, ActionResultV2 truth, terminal audit, and finding reconciliation floor without blind resend. Host update health is explicit tri-state evidence from a fixed read-only catalog; legacy v1 receipts lacking additive health facts terminalize as health-unknown/inconclusive. Task 10 remains sole owner of execution, verification, evidence, compensation, and rollback truth. Claims 16 and 17 and both workflow scorecards remain open below operational completion until browser proof, disposable Debian/Ubuntu lab proof, and Task 12 certification pass. Mobile/relay consumption remains separate with no local workaround for absent core APIs. Raw model command, file write, arbitrary pod exec, legacy run_command, /api/ai/run-command, and enterprise command remediation remain retired with no replacement or executable historical authority.", "owner": "project-owner", "status": "planned", "recorded_at": "2026-04-25", diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 02343b717..d6ba95ff3 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -4392,11 +4392,33 @@ generic owner, persist only bounded sanitized typed results, and still share the package-manager lease. Raw commands, paths, package selectors, stderr, secrets, and unbounded output are outside the receipt schema. +Task 09 now consumes that continuity end to end. A delayed terminal receipt is +validated against its immutable identity/digest and the agent-attested durable +terminal boundary, not the later query time. The server retains the later +query receipt separately. Impossible observation/terminal chronology, a +future terminal timestamp, stale-at-completion evidence, before-state drift, +or a hostile envelope still fails closed. Callback loss and a reopened server +store reconcile both APT actions by query only; the original typed dispatch is +never resent. Legacy APT v1 terminal payloads that predate additive package- +manager health facts remain structurally valid receipts, but their verified +claim is projected as health-unknown/inconclusive rather than confirmed. + +For host updates, `MutationStarted` begins immediately before the fixed install +command, not during metadata refresh or simulation. A fixed read-only +`dpkg --audit` health check under the same shared lease supplies explicit +`health_checked`, `package_manager_healthy`, and `recovery_required` facts. +Refresh failure, refreshed inventory drift, zero-pending state, and pre-install +health refusal therefore cannot claim partial installation or action-caused +recovery. Install/verify failures preserve phase, remaining count, tri-state +health, and recovery posture through the existing ActionResultV2 summaries. + Durable receipt protocol support is explicit agent-reported and server-observed capability metadata. Missing, legacy, or future versions stay connected for monitoring but advertise no APT mutation capability, emit no actionable APT finding, and fail dispatch readiness. Product version strings never imply this authority. The raw protocol integer stays internal to ingest, registry, and live dispatch checks; customer/frontend resource JSON consumes only derived -capabilities and readiness. Claims 16 and 17 remain open until Task 09 resumes and the required -browser/lab evidence plus Task 12 certification are complete. +capabilities and readiness. The fake-only code/test floor for claims 16 and 17 +is implemented; both claims and workflow scorecards remain below operational +completion until browser proof, disposable Debian/Ubuntu lab proof, and Task 12 +certification are complete. diff --git a/docs/release-control/v6/internal/subsystems/ai-runtime.md b/docs/release-control/v6/internal/subsystems/ai-runtime.md index af5718cc6..e77bd197c 100644 --- a/docs/release-control/v6/internal/subsystems/ai-runtime.md +++ b/docs/release-control/v6/internal/subsystems/ai-runtime.md @@ -5818,18 +5818,23 @@ verification, and execution success remains separate from outcome truth. Task 11 still owns the explicit acknowledgement UX and browser/device proof; Task 12 still owns final certification. -### Task 09 APT finding producer (partial) +### Task 09 deterministic APT workflow routing (non-lab floor) `internal/ai/findings_apt_workflows.go` deterministically produces agent-managed APT update and pressure-gated package-cache findings. Admission requires fresh dual-timestamp telemetry and the exact current canonical capability/handler. Stale, skewed, replayed, errored, or capability-less state is unknown and must not resolve an active finding; only fresh authoritative clear evidence or -explicit resource removal reconciles it. Durable disconnect/restart -reconciliation remains open on the Task 07 generic agent-operation receipt -continuation. +explicit resource removal reconciles it. The same pure detector entry point is +exercised through an exact empty-parameter proposal, shared planning/policy and +human approval, durable typed dispatch, canonical terminal audit, and finding +resolution for both workflows. Finding evidence stays bounded and contains no +command, path, package selector, raw APT output, stderr, or reboot authority. -This partial producer/transport slice does not close L20 or RA35, claims 16 or -17, either APT workflow scorecard, durable disconnect/restart/late-receipt/no- -blind-replay evidence, browser workflow evidence, or Debian/Ubuntu real-lab -evidence. Those remain explicit continuation or certification inputs. +Fake-only callback-loss tests reopen the server-side action store and consume +Task 07 terminal receipts without resending either typed mutation, then +reconcile the audit and originating finding. This closes the non-lab detector- +to-reconciliation code/test floor only. Claims 16/17 and both APT scorecards +remain below operational completion because browser workflow evidence, +disposable Debian/Ubuntu real-lab evidence, and Task 12 final-SHA certification +remain explicit residuals; no tier-5 or tier-6 evidence is claimed here. diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index df6411590..021a9af27 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -7459,13 +7459,27 @@ missing, stale, wrong-organization, wrong-actor, expired, revoked, malformed, conflicting, and unavailable-store evidence. Task 09 APT executors consume Task 10 `ActionResultV2` without local truth -enums. APT metadata refresh is an external package-manager effect, so refresh -failure or cancellation is `inconclusive/possible_partial_effect`, never -`not_run`. Readback from the mutating agent is `agent_attested`, preserves -agent-observed and server-received timestamps, and declares compensation -unavailable. Durable post-disconnect reconciliation remains a Task 07 -continuation dependency. +enums. Host-update `MutationStarted` means the fixed install command began; +metadata refresh, refreshed-inventory drift, zero-pending state, and pre-install +health refusal do not claim install effects or action-caused recovery. Install +and verify outcomes project the validated phase, bounded remaining count, +explicit health-check/healthy facts, and recovery requirement through existing +execution/verification summaries. Package-manager health comes only from the +fixed read-only agent health catalog under the shared lease, never raw stderr or +absence of an error string. Cache-cleanup failures preserve clean/verify phase, +measured reclaim effect, non-rollbackable posture, and rescan requirement. -This is a partial non-lab producer migration only. It does not change L20 or -RA35 completion, claims 16/17, workflow scorecards, browser certification, or -real Debian/Ubuntu execution status. +Readback from the mutating agent remains `agent_attested`, preserves agent- +observed and server-received timestamps, and declares compensation unavailable. +Durable delayed receipts use their agent-attested terminal boundary for +chronology/freshness while retaining server query receipt time separately. +Additive health facts do not invalidate older Task 07 APT v1 terminal receipts: +missing facts are unknown and downgrade a verified claim to +`inconclusive/package_manager_health_unknown` with no evidence and no resend. +Explicit contradictory health/recovery combinations fail closed. + +The non-lab detector/proposal/policy/receipt/result/reconciliation code and test +floor is implemented for both workflows. Claims 16/17 and both workflow +scorecards remain below operational completion; browser certification, +disposable Debian/Ubuntu execution, and Task 12 final-SHA certification remain +open and are not inferred from fake-only proof. diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index be4dae04b..d2540bb6a 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -2062,8 +2062,13 @@ Task 09 package-cache cleanup remains a bounded typed action over the fixed APT cache provider and requires fresh canonical reclaimable-byte and containing- filesystem pressure evidence. It accepts no path, package selector, installed-package removal, reboot, or rollback authority. Compensation is -unavailable; durable restart/replay continuity remains an explicit Task 07 -dependency. +unavailable. Clean and verify failure results retain their validated phase, +bounded before/after/reclaimed bytes, explicit non-rollbackable posture, and a +rescan requirement without implying restoration. Task 09 now consumes the +accepted Task 07 generic receipt owner for callback-loss and reopened-server +query-only reconciliation; interrupted, not-found, or tombstoned receipts never +authorize resend. Browser and disposable Debian/Ubuntu proof remain explicit +operational residuals. ### Patrol Autopilot config recovery diff --git a/internal/agentexec/apt_codec.go b/internal/agentexec/apt_codec.go index 7d8c7eb40..918d28d44 100644 --- a/internal/agentexec/apt_codec.go +++ b/internal/agentexec/apt_codec.go @@ -89,16 +89,40 @@ const ( HostAPTReceiptVersion = 1 ) +// ValidateOperationQueryResultForIdentity validates a durable receipt against +// its immutable dispatch identity. Receipt replay can happen long after the +// agent observed the terminal result, so evidence freshness is anchored to the +// durable agent-authored terminal boundary rather than the later query time. +// receivedAt remains the server's transport receipt time and is used only to +// reject an implausibly future agent terminal timestamp. func ValidateOperationQueryResultForIdentity(result operationreceipt.QueryResult, identity operationreceipt.Identity, receivedAt time.Time) error { + if result.Version != operationreceipt.ProtocolVersion { + return fmt.Errorf("unsupported operation query result version %d", result.Version) + } if result.Status == operationreceipt.QueryNotFound { + if result.Record != nil { + return fmt.Errorf("not-found operation query result contains a record") + } return nil } if result.Record == nil || result.Record.Identity != identity { return fmt.Errorf("operation query result identity mismatch") } + if err := operationreceipt.ValidateRecord(*result.Record); err != nil { + return err + } if result.Status != operationreceipt.QueryFoundTerminal { + if result.Status != operationreceipt.QueryFoundInterrupted || result.Record.State == operationreceipt.StateTerminal { + return fmt.Errorf("operation query result status and record state mismatch") + } return nil } + if result.Record.State != operationreceipt.StateTerminal { + return fmt.Errorf("terminal operation query result requires terminal record") + } + if receivedAt.IsZero() || result.Record.TerminalAt.After(receivedAt.UTC().Add(5*time.Minute)) { + return fmt.Errorf("operation query terminal time is invalid or implausibly future") + } switch identity.OperationKind { case HostUpdateOperationInstall: if result.Record.ResultKind != HostUpdateReceiptKind || result.Record.ResultVersion != HostAPTReceiptVersion { @@ -112,7 +136,10 @@ func ValidateOperationQueryResultForIdentity(result operationreceipt.QueryResult if err := ValidateHostUpdatePayload(&req); err != nil { return err } - return ValidateHostUpdateResultForRequestAt(req, payload, receivedAt) + if err := ValidateHostUpdateResultForRequest(req, payload); err != nil { + return err + } + return validateDurableAPTObservation(payload.Before.CheckedAt, payload.After.CheckedAt, result.Record.TerminalAt) case HostStorageCleanupOperationPackageCache: if result.Record.ResultKind != HostStorageCleanupReceiptKind || result.Record.ResultVersion != HostAPTReceiptVersion { return fmt.Errorf("host cleanup query result envelope mismatch") @@ -125,12 +152,22 @@ func ValidateOperationQueryResultForIdentity(result operationreceipt.QueryResult if err := ValidateHostStorageCleanupPayload(&req); err != nil { return err } - return ValidateHostStorageCleanupResultForRequestAt(req, payload, receivedAt) + if err := ValidateHostStorageCleanupResultForRequest(req, payload); err != nil { + return err + } + return validateDurableAPTObservation(payload.Before.CheckedAt, payload.After.CheckedAt, result.Record.TerminalAt) default: return fmt.Errorf("unsupported operation query kind %q", identity.OperationKind) } } +func validateDurableAPTObservation(before, after, terminalAt time.Time) error { + if !validAPTResultObservationChronology(before, after) || terminalAt.IsZero() || terminalAt.Before(after) || after.Before(terminalAt.Add(-hostAPTResultFreshness)) { + return fmt.Errorf("durable APT result observation is stale or has invalid terminal chronology") + } + return nil +} + func BindHostUpdatePayload(payload *HostUpdatePayload) error { if payload == nil { return fmt.Errorf("host update payload is required") @@ -197,15 +234,33 @@ func ValidateHostUpdateResultPayload(result *HostUpdateResultPayload) error { default: return fmt.Errorf("unsupported host update execution phase %q", result.ExecutionPhase) } - if result.Success && result.ExecutionPhase != HostUpdatePhaseComplete { - return fmt.Errorf("successful host update must be complete") + if result.Success && result.ExecutionPhase != HostUpdatePhaseVerify && result.ExecutionPhase != HostUpdatePhaseComplete { + return fmt.Errorf("successful host update mutation must be in verify or complete phase") } if (result.Verification == HostUpdateVerificationVerified || result.Verification == HostUpdateVerificationFailed) && !validAPTResultObservationChronology(result.Before.CheckedAt, result.After.CheckedAt) { return fmt.Errorf("evidence-bearing host update observation timestamps are invalid") } - if result.MutationStarted && result.ExecutionPhase != HostUpdatePhaseRefresh && result.ExecutionPhase != HostUpdatePhaseInstall && result.ExecutionPhase != HostUpdatePhaseVerify && result.ExecutionPhase != HostUpdatePhaseComplete { + if result.MutationStarted && result.ExecutionPhase != HostUpdatePhaseInstall && result.ExecutionPhase != HostUpdatePhaseVerify && result.ExecutionPhase != HostUpdatePhaseComplete { return fmt.Errorf("host update mutation state conflicts with execution phase") } + if result.RecoveryRequired && !result.MutationStarted { + return fmt.Errorf("host update recovery requirement conflicts with mutation state") + } + if result.PackageManagerHealthy && !result.HealthChecked { + return fmt.Errorf("healthy package manager claim requires a completed health check") + } + if result.HealthChecked && !result.PackageManagerHealthy && result.MutationStarted && !result.RecoveryRequired { + return fmt.Errorf("unhealthy package manager after mutation requires recovery") + } + if result.ExecutionPhase == HostUpdatePhaseInstall && result.MutationStarted && !result.Success && !result.RecoveryRequired { + return fmt.Errorf("partial host update install requires recovery") + } + if result.Success && result.ExecutionPhase == HostUpdatePhaseComplete && result.RecoveryRequired { + return fmt.Errorf("successful host update completion cannot require recovery") + } + if result.Verification == HostUpdateVerificationVerified && result.ExecutionPhase != HostUpdatePhaseComplete { + return fmt.Errorf("verified host update must be complete") + } return nil } @@ -223,8 +278,11 @@ func ValidateHostStorageCleanupResultPayload(result *HostStorageCleanupResultPay default: return fmt.Errorf("unsupported host storage cleanup execution phase %q", result.ExecutionPhase) } - if result.Success && result.ExecutionPhase != HostStorageCleanupPhaseComplete { - return fmt.Errorf("successful host storage cleanup must be complete") + if result.Success && result.ExecutionPhase != HostStorageCleanupPhaseVerify && result.ExecutionPhase != HostStorageCleanupPhaseComplete { + return fmt.Errorf("successful host storage cleanup mutation must be in verify or complete phase") + } + if result.Verification == HostStorageCleanupVerificationVerified && result.ExecutionPhase != HostStorageCleanupPhaseComplete { + return fmt.Errorf("verified host storage cleanup must be complete") } if (result.Verification == HostStorageCleanupVerificationVerified || result.Verification == HostStorageCleanupVerificationFailed) && !validAPTResultObservationChronology(result.Before.CheckedAt, result.After.CheckedAt) { return fmt.Errorf("evidence-bearing host storage cleanup observation timestamps are invalid") diff --git a/internal/agentexec/apt_codec_test.go b/internal/agentexec/apt_codec_test.go index 22c94c7db..c5e1b5b19 100644 --- a/internal/agentexec/apt_codec_test.go +++ b/internal/agentexec/apt_codec_test.go @@ -72,7 +72,7 @@ func TestMalformedOrCrossTypeAPTResultCannotPoisonPendingUpdate(t *testing.T) { } now := time.Now().UTC() - valid := HostUpdateResultPayload{RequestID: "attempt-1", ActionID: "action-1", ExecutionPhase: HostUpdatePhaseComplete, Success: true, Verification: HostUpdateVerificationVerified, + valid := HostUpdateResultPayload{RequestID: "attempt-1", ActionID: "action-1", ExecutionPhase: HostUpdatePhaseComplete, Success: true, HealthChecked: true, PackageManagerHealthy: true, Verification: HostUpdateVerificationVerified, Before: HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: hash, PendingCount: 1, CheckedAt: now}, After: HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: empty, PendingCount: 0, CheckedAt: now}} if err := conn.WriteJSON(mustNewMessage(t, MsgTypeHostUpdateResult, "attempt-1", valid)); err != nil { @@ -157,6 +157,40 @@ func TestAPTResultReceiptTimeValidationUsesControlledClock(t *testing.T) { } } +func TestLegacyHostUpdateResultWithoutHealthFieldsRemainsStrictlyDecodableAsUnknown(t *testing.T) { + now := time.Now().UTC() + payload := HostUpdateResultPayload{ + RequestID: "legacy.dispatch.1", ActionID: "legacy", Success: true, MutationStarted: true, ExecutionPhase: HostUpdatePhaseComplete, + Before: HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: "sha256:" + strings.Repeat("a", 64), PendingCount: 1, CheckedAt: now.Add(-time.Second)}, + After: HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: "sha256:" + strings.Repeat("b", 64), CheckedAt: now}, + Verification: HostUpdateVerificationVerified, + } + raw, err := json.Marshal(payload) + if err != nil { + t.Fatal(err) + } + var legacy map[string]any + if err := json.Unmarshal(raw, &legacy); err != nil { + t.Fatal(err) + } + delete(legacy, "health_checked") + delete(legacy, "package_manager_healthy") + delete(legacy, "recovery_required") + raw, _ = json.Marshal(legacy) + decoded, err := DecodeHostUpdateResultPayload(raw) + if err != nil { + t.Fatalf("legacy terminal result rejected: %v", err) + } + if decoded.HealthChecked || decoded.PackageManagerHealthy || decoded.RecoveryRequired { + t.Fatalf("legacy health must remain unknown: %#v", decoded) + } + legacy["unexpected_authority"] = true + hostile, _ := json.Marshal(legacy) + if _, err := DecodeHostUpdateResultPayload(hostile); err == nil { + t.Fatal("unknown field bypassed strict legacy decode") + } +} + func wsMessageForTest(messageType MessageType, id string, payload json.RawMessage) Message { return Message{Type: messageType, ID: id, Timestamp: time.Now().UTC(), Payload: payload} } diff --git a/internal/agentexec/operation_query_test.go b/internal/agentexec/operation_query_test.go index f94479c1e..26207bd86 100644 --- a/internal/agentexec/operation_query_test.go +++ b/internal/agentexec/operation_query_test.go @@ -157,7 +157,7 @@ func TestOperationQueryBeforeStateDigestMismatchFailsClosed(t *testing.T) { req := HostUpdatePayload{RequestID: "u.dispatch.1", ActionID: "u", Operation: HostUpdateOperationInstall, ExpectedInventoryHash: "sha256:" + strings.Repeat("a", 64)} _ = BindHostUpdatePayload(&req) id := HostUpdateOperationIdentity("agent", req) - result := HostUpdateResultPayload{RequestID: req.RequestID, ActionID: req.ActionID, Success: true, ExecutionPhase: HostUpdatePhaseComplete, MutationStarted: true, Before: HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: "sha256:" + strings.Repeat("b", 64), PendingCount: 1, CheckedAt: now.Add(-time.Second)}, After: HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: "sha256:" + strings.Repeat("c", 64), PendingCount: 0, CheckedAt: now}, Verification: HostUpdateVerificationVerified} + result := HostUpdateResultPayload{RequestID: req.RequestID, ActionID: req.ActionID, Success: true, ExecutionPhase: HostUpdatePhaseComplete, MutationStarted: true, HealthChecked: true, PackageManagerHealthy: true, Before: HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: "sha256:" + strings.Repeat("b", 64), PendingCount: 1, CheckedAt: now.Add(-time.Second)}, After: HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: "sha256:" + strings.Repeat("c", 64), PendingCount: 0, CheckedAt: now}, Verification: HostUpdateVerificationVerified} raw, _ := json.Marshal(result) return struct { name string @@ -191,6 +191,83 @@ func TestOperationQueryBeforeStateDigestMismatchFailsClosed(t *testing.T) { } } +func TestOperationQueryDurableTerminalTimingUsesAgentCommitBoundary(t *testing.T) { + for _, operation := range []string{HostUpdateOperationInstall, HostStorageCleanupOperationPackageCache} { + t.Run(operation, func(t *testing.T) { + terminalAt := time.Now().UTC().Add(-2 * time.Hour) + query, identity := durableAPTTerminalQuery(t, operation, terminalAt, terminalAt.Add(-2*time.Second), terminalAt.Add(-time.Second)) + if err := ValidateOperationQueryResultForIdentity(query, identity, terminalAt.Add(2*time.Hour)); err != nil { + t.Fatalf("delayed valid terminal receipt rejected: %v", err) + } + + stale, staleIdentity := durableAPTTerminalQuery(t, operation, terminalAt, terminalAt.Add(-17*time.Minute), terminalAt.Add(-16*time.Minute)) + if err := ValidateOperationQueryResultForIdentity(stale, staleIdentity, terminalAt.Add(2*time.Hour)); err == nil || !strings.Contains(err.Error(), "stale") { + t.Fatalf("stale-at-completion receipt err=%v", err) + } + + impossible, impossibleIdentity := durableAPTTerminalQuery(t, operation, terminalAt, terminalAt.Add(-time.Second), terminalAt.Add(-2*time.Second)) + if err := ValidateOperationQueryResultForIdentity(impossible, impossibleIdentity, terminalAt.Add(2*time.Hour)); err == nil || !strings.Contains(err.Error(), "timestamps") { + t.Fatalf("impossible observation chronology err=%v", err) + } + + afterTerminal, afterTerminalIdentity := durableAPTTerminalQuery(t, operation, terminalAt, terminalAt.Add(-time.Second), terminalAt.Add(time.Second)) + if err := ValidateOperationQueryResultForIdentity(afterTerminal, afterTerminalIdentity, terminalAt.Add(2*time.Hour)); err == nil || !strings.Contains(err.Error(), "terminal chronology") { + t.Fatalf("terminal-before-observation err=%v", err) + } + + futureTerminal := time.Now().UTC().Add(6 * time.Minute) + future, futureIdentity := durableAPTTerminalQuery(t, operation, futureTerminal, futureTerminal.Add(-2*time.Second), futureTerminal.Add(-time.Second)) + if err := ValidateOperationQueryResultForIdentity(future, futureIdentity, time.Now().UTC()); err == nil || !strings.Contains(err.Error(), "future") { + t.Fatalf("future terminal err=%v", err) + } + }) + } +} + +func durableAPTTerminalQuery(t *testing.T, operation string, terminalAt, beforeAt, afterAt time.Time) (operationreceipt.QueryResult, operationreceipt.Identity) { + t.Helper() + var identity operationreceipt.Identity + var kind string + var payload []byte + var err error + switch operation { + case HostUpdateOperationInstall: + req := HostUpdatePayload{RequestID: "update.dispatch.1", ActionID: "update", Operation: operation, ExpectedInventoryHash: "sha256:" + strings.Repeat("a", 64)} + if err = BindHostUpdatePayload(&req); err == nil { + identity = HostUpdateOperationIdentity("agent", req) + payload, err = json.Marshal(HostUpdateResultPayload{ + RequestID: req.RequestID, ActionID: req.ActionID, Success: true, MutationStarted: true, ExecutionPhase: HostUpdatePhaseComplete, + Before: HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: req.ExpectedInventoryHash, PendingCount: 2, CheckedAt: beforeAt}, + After: HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: "sha256:" + strings.Repeat("b", 64), CheckedAt: afterAt}, + HealthChecked: true, PackageManagerHealthy: true, Verification: HostUpdateVerificationVerified, + }) + } + kind = HostUpdateReceiptKind + case HostStorageCleanupOperationPackageCache: + req := HostStorageCleanupPayload{RequestID: "cleanup.dispatch.1", ActionID: "cleanup", Operation: operation, ExpectedFingerprint: "sha256:" + strings.Repeat("c", 64)} + if err = BindHostStorageCleanupPayload(&req); err == nil { + identity = HostStorageCleanupOperationIdentity("agent", req) + payload, err = json.Marshal(HostStorageCleanupResultPayload{ + RequestID: req.RequestID, ActionID: req.ActionID, Success: true, MutationStarted: true, ExecutionPhase: HostStorageCleanupPhaseComplete, + Before: HostStorageCleanupSnapshot{Supported: true, Provider: "apt-package-cache", Fingerprint: req.ExpectedFingerprint, ReclaimableBytes: 10, CheckedAt: beforeAt}, + After: HostStorageCleanupSnapshot{Supported: true, Provider: "apt-package-cache", Fingerprint: "sha256:" + strings.Repeat("d", 64), ReclaimableBytes: 1, CheckedAt: afterAt}, + ReclaimedBytes: 9, Verification: HostStorageCleanupVerificationVerified, + }) + } + kind = HostStorageCleanupReceiptKind + default: + t.Fatalf("unsupported test operation %q", operation) + } + if err != nil { + t.Fatal(err) + } + record := operationreceipt.Record{ + Identity: identity, State: operationreceipt.StateTerminal, AcceptedAt: terminalAt.Add(-20 * time.Minute), StartedAt: terminalAt.Add(-19 * time.Minute), TerminalAt: terminalAt, + ResultKind: kind, ResultVersion: HostAPTReceiptVersion, Result: payload, + } + return operationreceipt.QueryResult{Version: operationreceipt.ProtocolVersion, Status: operationreceipt.QueryFoundTerminal, Record: &record}, identity +} + func TestLegacyAgentWithoutReceiptProtocolRemainsConnectedButTypedMutationFailsClosed(t *testing.T) { s := NewServer(func(token, agent, host string) bool { return token == "ok" }) conn, cleanup := registeredTestAgentLegacy(t, s, "legacy-agent") diff --git a/internal/agentexec/server_websocket_test.go b/internal/agentexec/server_websocket_test.go index 2db45ca83..9200f4f38 100644 --- a/internal/agentexec/server_websocket_test.go +++ b/internal/agentexec/server_websocket_test.go @@ -568,7 +568,7 @@ func TestExecuteHostUpdateRoundTripUsesTypedCommandFreeEnvelope(t *testing.T) { ExecutionPhase: HostUpdatePhaseComplete, Before: HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: inventoryHash, PendingCount: 2, CheckedAt: time.Now().UTC()}, After: HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: emptyInventoryHash, PendingCount: 0, RebootRequired: true, CheckedAt: time.Now().UTC()}, - Verification: HostUpdateVerificationVerified, + HealthChecked: true, PackageManagerHealthy: true, Verification: HostUpdateVerificationVerified, } if err := conn.WriteJSON(mustNewMessage(t, MsgTypeHostUpdateResult, payload.RequestID, response)); err != nil { agentErr <- err diff --git a/internal/agentexec/types.go b/internal/agentexec/types.go index cfbc3e213..e3bda967d 100644 --- a/internal/agentexec/types.go +++ b/internal/agentexec/types.go @@ -197,16 +197,19 @@ const ( // independently. Success means the package-manager mutation completed; // Verification states whether the postcondition was actually observed. type HostUpdateResultPayload struct { - RequestID string `json:"request_id"` - ActionID string `json:"action_id"` - Success bool `json:"success"` - ExecutionPhase string `json:"execution_phase"` - MutationStarted bool `json:"mutation_started"` - Before HostPackageUpdateSnapshot `json:"before"` - After HostPackageUpdateSnapshot `json:"after"` - Verification string `json:"verification"` - Error string `json:"error,omitempty"` - Duration int64 `json:"duration_ms"` + RequestID string `json:"request_id"` + ActionID string `json:"action_id"` + Success bool `json:"success"` + ExecutionPhase string `json:"execution_phase"` + MutationStarted bool `json:"mutation_started"` + Before HostPackageUpdateSnapshot `json:"before"` + After HostPackageUpdateSnapshot `json:"after"` + HealthChecked bool `json:"health_checked"` + PackageManagerHealthy bool `json:"package_manager_healthy"` + RecoveryRequired bool `json:"recovery_required"` + Verification string `json:"verification"` + Error string `json:"error,omitempty"` + Duration int64 `json:"duration_ms"` } const ( diff --git a/internal/ai/findings_apt_workflows.go b/internal/ai/findings_apt_workflows.go index 6b4030123..60aa4bcd2 100644 --- a/internal/ai/findings_apt_workflows.go +++ b/internal/ai/findings_apt_workflows.go @@ -23,6 +23,31 @@ type aptWorkflowWatcher struct{} func newAPTWorkflowWatcher() *aptWorkflowWatcher { return &aptWorkflowWatcher{} } +// DetectAPTWorkflowFindings is the deterministic, model-free detector entry +// point used by Patrol and end-to-end action lifecycle proofs. It emits only +// findings supported by fresh canonical host views; active-finding resolution +// remains owned by aptWorkflowWatcher so missing or stale authority cannot be +// mistaken for a cleared condition. +func DetectAPTWorkflowFindings(readState unifiedresources.ReadState, now time.Time) []*Finding { + if readState == nil { + return nil + } + now = now.UTC() + var findings []*Finding + for _, host := range readState.Hosts() { + if host == nil || strings.TrimSpace(host.ID()) == "" { + continue + } + if finding, _, _ := aptHostUpdateFinding(host, now); finding != nil { + findings = append(findings, finding) + } + if finding, _, _ := aptPackageCacheFinding(host, now); finding != nil { + findings = append(findings, finding) + } + } + return findings +} + func (w *aptWorkflowWatcher) Observe(state patrolRuntimeState, active []*Finding, now time.Time) (emit []*Finding, resolve []resolveSentinel) { if w == nil { return nil, nil @@ -36,25 +61,34 @@ func (w *aptWorkflowWatcher) Observe(state patrolRuntimeState, active []*Finding now = now.UTC() present := make(map[string]struct{}) resolved := make(map[string]struct{}) + detected := make(map[string]struct{}) + for _, finding := range DetectAPTWorkflowFindings(state.readState, now) { + emit = append(emit, finding) + detected[finding.ID] = struct{}{} + } for _, host := range state.readState.Hosts() { if host == nil || strings.TrimSpace(host.ID()) == "" { continue } resourceID := strings.TrimSpace(host.ID()) present[resourceID] = struct{}{} - if finding, known, cleared := aptHostUpdateFinding(host, now); finding != nil { - emit = append(emit, finding) - } else if known && cleared { - id := aptHostUpdateFindingIDPrefix + resourceID - resolve = append(resolve, resolveSentinel{DedupKey: id, Reason: aptWorkflowResolveCleared}) - resolved[id] = struct{}{} + updateID := aptHostUpdateFindingIDPrefix + resourceID + if _, found := detected[updateID]; !found { + _, known, cleared := aptHostUpdateFinding(host, now) + if known && cleared { + id := aptHostUpdateFindingIDPrefix + resourceID + resolve = append(resolve, resolveSentinel{DedupKey: id, Reason: aptWorkflowResolveCleared}) + resolved[id] = struct{}{} + } } - if finding, known, cleared := aptPackageCacheFinding(host, now); finding != nil { - emit = append(emit, finding) - } else if known && cleared { - id := aptPackageCacheFindingIDPrefix + resourceID - resolve = append(resolve, resolveSentinel{DedupKey: id, Reason: aptWorkflowResolveCleared}) - resolved[id] = struct{}{} + cleanupID := aptPackageCacheFindingIDPrefix + resourceID + if _, found := detected[cleanupID]; !found { + _, known, cleared := aptPackageCacheFinding(host, now) + if known && cleared { + id := aptPackageCacheFindingIDPrefix + resourceID + resolve = append(resolve, resolveSentinel{DedupKey: id, Reason: aptWorkflowResolveCleared}) + resolved[id] = struct{}{} + } } } for _, finding := range active { diff --git a/internal/ai/findings_apt_workflows_test.go b/internal/ai/findings_apt_workflows_test.go index 229675274..a4bfbf537 100644 --- a/internal/ai/findings_apt_workflows_test.go +++ b/internal/ai/findings_apt_workflows_test.go @@ -31,7 +31,12 @@ func aptWorkflowTestHost(now time.Time) *ur.HostView { func TestAPTWorkflowWatcherEmitsDeterministicCapabilityBoundFindings(t *testing.T) { now := time.Date(2026, 7, 12, 8, 0, 0, 0, time.UTC) host := aptWorkflowTestHost(now) - emit, resolve := newAPTWorkflowWatcher().Observe(patrolRuntimeState{readState: &mockReadState{hosts: []*ur.HostView{host}}}, nil, now) + readState := &mockReadState{hosts: []*ur.HostView{host}} + detected := DetectAPTWorkflowFindings(readState, now) + if len(detected) != 2 || detected[0].ID != "apt:host-update:agent:host-1" || detected[1].ID != "apt:cache-cleanup:agent:host-1" { + t.Fatalf("canonical detector findings=%#v", detected) + } + emit, resolve := newAPTWorkflowWatcher().Observe(patrolRuntimeState{readState: readState}, nil, now) if len(emit) != 2 || len(resolve) != 0 { t.Fatalf("emit=%d resolve=%d", len(emit), len(resolve)) } diff --git a/internal/api/apt_workflow_integration_test.go b/internal/api/apt_workflow_integration_test.go new file mode 100644 index 000000000..6124aef3c --- /dev/null +++ b/internal/api/apt_workflow_integration_test.go @@ -0,0 +1,296 @@ +package api + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/agentexec" + "github.com/rcourtman/pulse-go-rewrite/internal/ai" + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/operationreceipt" + "github.com/rcourtman/pulse-go-rewrite/internal/relay" + unified "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" + "github.com/rcourtman/pulse-go-rewrite/pkg/aicontracts" + "github.com/rcourtman/pulse-go-rewrite/pkg/auth" +) + +func TestAPTUpdateDetectorProposalApprovalDispatchAuditAndFindingReconciliation(t *testing.T) { + now := time.Now().UTC() + resource := hostUpdateActionResource(now) + finding := detectSingleAPTWorkflowFinding(t, resource, now) + if finding.ID != "apt:host-update:agent:host-1" || finding.Key != "apt-host-updates" { + t.Fatalf("finding=%#v", finding) + } + assertBoundedAPTWorkflowEvidence(t, finding) + + resources := newActionTestResourceHandlers(t, &config.Config{DataPath: t.TempDir()}) + resources.SetStateProvider(resourceUnifiedSeedProvider{snapshot: models.StateSnapshot{LastUpdate: now}, resources: []unified.Resource{resource}}) + agents := &fakeHostUpdateAgent{connected: true, result: &agentexec.HostUpdateResultPayload{ + Success: true, MutationStarted: true, ExecutionPhase: agentexec.HostUpdatePhaseComplete, + Before: agentexec.HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: testHostPackageInventoryHash, PendingCount: 3, Packages: []agentexec.HostPackageUpdate{{Name: "private-package"}}, CheckedAt: now.Add(-time.Second)}, + After: agentexec.HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: testHostPackageEmptyInventoryHash, Packages: []agentexec.HostPackageUpdate{{Name: "private-package"}}, CheckedAt: now}, + HealthChecked: true, PackageManagerHealthy: true, Verification: agentexec.HostUpdateVerificationVerified, + Error: "raw stderr token secret /private/cache/path", + }} + resources.SetActionExecutor(newRoutedActionExecutor(resources, newHostUpdateActionExecutor(resources, agents))) + runAPTWorkflowFindingJourney(t, resources, finding, hostPackageUpdateCapability) + if len(agents.requests) != 1 || agents.requests[0].Operation != agentexec.HostUpdateOperationInstall { + t.Fatalf("typed update requests=%#v", agents.requests) + } +} + +func TestAPTUpdateCallbackLossServerRestartReconcilesTerminalAuditAndFindingWithoutResend(t *testing.T) { + now := time.Now().UTC() + resource := hostUpdateActionResource(now) + finding := detectSingleAPTWorkflowFinding(t, resource, now) + agents := &fakeHostUpdateAgent{connected: true, err: errors.New("controlled callback loss")} + runAPTWorkflowReceiptRecoveryJourney(t, resource, finding, hostPackageUpdateCapability, + func(resources *ResourceHandlers) ActionExecutor { + return newHostUpdateActionExecutor(resources, agents) + }, + func(attempt unified.ActionDispatchAttempt) { + payload := agentexec.HostUpdateResultPayload{ + RequestID: attempt.ID, ActionID: attempt.ActionID, Success: true, MutationStarted: true, ExecutionPhase: agentexec.HostUpdatePhaseComplete, + Before: agentexec.HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: testHostPackageInventoryHash, PendingCount: 3, CheckedAt: now.Add(-2 * time.Second)}, + After: agentexec.HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: testHostPackageEmptyInventoryHash, CheckedAt: now.Add(-time.Second)}, + HealthChecked: true, PackageManagerHealthy: true, Verification: agentexec.HostUpdateVerificationVerified, + } + raw, _ := json.Marshal(payload) + identity := operationreceipt.Identity{AttemptID: attempt.ID, ActionID: attempt.ActionID, OperationKind: attempt.OperationKind, OperationVersion: attempt.OperationVersion, RequestDigest: attempt.RequestDigest, AgentID: attempt.AgentID} + agents.queryResult = terminalAPTQuery(identity, agentexec.HostUpdateReceiptKind, raw, now) + }, + func() (int, int) { return len(agents.requests), len(agents.queries) }) +} + +func TestAPTCacheCleanupCallbackLossServerRestartReconcilesTerminalAuditAndFindingWithoutResend(t *testing.T) { + now := time.Now().UTC() + resource := hostStorageCleanupActionResource(now) + finding := detectSingleAPTWorkflowFinding(t, resource, now) + agents := &fakeHostStorageCleanupAgent{connected: true, err: errors.New("controlled callback loss")} + runAPTWorkflowReceiptRecoveryJourney(t, resource, finding, hostStorageCleanupCapability, + func(resources *ResourceHandlers) ActionExecutor { + return newHostStorageCleanupActionExecutor(resources, agents) + }, + func(attempt unified.ActionDispatchAttempt) { + payload := agentexec.HostStorageCleanupResultPayload{ + RequestID: attempt.ID, ActionID: attempt.ActionID, Success: true, MutationStarted: true, ExecutionPhase: agentexec.HostStorageCleanupPhaseComplete, + Before: agentexec.HostStorageCleanupSnapshot{Supported: true, Provider: "apt-package-cache", Fingerprint: testHostStorageCleanupFingerprint, ReclaimableBytes: 512 * 1024 * 1024, CheckedAt: now.Add(-2 * time.Second)}, + After: agentexec.HostStorageCleanupSnapshot{Supported: true, Provider: "apt-package-cache", Fingerprint: testHostStorageCleanupAfterFingerprint, ReclaimableBytes: 8 * 1024 * 1024, CheckedAt: now.Add(-time.Second)}, + ReclaimedBytes: 504 * 1024 * 1024, Verification: agentexec.HostStorageCleanupVerificationVerified, + } + raw, _ := json.Marshal(payload) + identity := operationreceipt.Identity{AttemptID: attempt.ID, ActionID: attempt.ActionID, OperationKind: attempt.OperationKind, OperationVersion: attempt.OperationVersion, RequestDigest: attempt.RequestDigest, AgentID: attempt.AgentID} + agents.queryResult = terminalAPTQuery(identity, agentexec.HostStorageCleanupReceiptKind, raw, now) + }, + func() (int, int) { return len(agents.requests), len(agents.queries) }) +} + +func terminalAPTQuery(identity operationreceipt.Identity, kind string, payload []byte, terminalAt time.Time) operationreceipt.QueryResult { + return operationreceipt.QueryResult{Version: operationreceipt.ProtocolVersion, Status: operationreceipt.QueryFoundTerminal, Record: &operationreceipt.Record{ + Identity: identity, State: operationreceipt.StateTerminal, AcceptedAt: terminalAt.Add(-4 * time.Second), StartedAt: terminalAt.Add(-3 * time.Second), TerminalAt: terminalAt, + ResultKind: kind, ResultVersion: agentexec.HostAPTReceiptVersion, Result: payload, + }} +} + +func TestAPTCacheCleanupDetectorProposalApprovalDispatchAuditAndFindingReconciliation(t *testing.T) { + now := time.Now().UTC() + resource := hostStorageCleanupActionResource(now) + finding := detectSingleAPTWorkflowFinding(t, resource, now) + if finding.ID != "apt:cache-cleanup:agent:host-cleanup" || finding.Key != "apt-package-cache-pressure" { + t.Fatalf("finding=%#v", finding) + } + assertBoundedAPTWorkflowEvidence(t, finding) + + resources := newActionTestResourceHandlers(t, &config.Config{DataPath: t.TempDir()}) + resources.SetStateProvider(resourceUnifiedSeedProvider{snapshot: models.StateSnapshot{LastUpdate: now}, resources: []unified.Resource{resource}}) + agents := &fakeHostStorageCleanupAgent{connected: true, result: &agentexec.HostStorageCleanupResultPayload{ + Success: true, MutationStarted: true, ExecutionPhase: agentexec.HostStorageCleanupPhaseComplete, + Before: agentexec.HostStorageCleanupSnapshot{Supported: true, Provider: "apt-package-cache", Fingerprint: testHostStorageCleanupFingerprint, ReclaimableBytes: 512 * 1024 * 1024, CheckedAt: now.Add(-time.Second)}, + After: agentexec.HostStorageCleanupSnapshot{Supported: true, Provider: "apt-package-cache", Fingerprint: testHostStorageCleanupAfterFingerprint, ReclaimableBytes: 8 * 1024 * 1024, CheckedAt: now}, + ReclaimedBytes: 504 * 1024 * 1024, Verification: agentexec.HostStorageCleanupVerificationVerified, + }} + resources.SetActionExecutor(newRoutedActionExecutor(resources, newHostStorageCleanupActionExecutor(resources, agents))) + runAPTWorkflowFindingJourney(t, resources, finding, hostStorageCleanupCapability) + if len(agents.requests) != 1 || agents.requests[0].Operation != agentexec.HostStorageCleanupOperationPackageCache { + t.Fatalf("typed cleanup requests=%#v", agents.requests) + } +} + +func detectSingleAPTWorkflowFinding(t *testing.T, resource unified.Resource, now time.Time) *ai.Finding { + t.Helper() + registry := unified.NewRegistry(nil) + registry.IngestResources([]unified.Resource{resource}) + findings := ai.DetectAPTWorkflowFindings(registry, now) + if len(findings) != 1 { + t.Fatalf("detected findings=%#v", findings) + } + return findings[0] +} + +func assertBoundedAPTWorkflowEvidence(t *testing.T, finding *ai.Finding) { + t.Helper() + if finding == nil || finding.Evidence == "" || len(finding.Evidence) > 512 { + t.Fatalf("unbounded finding evidence=%#v", finding) + } + for _, forbidden := range []string{"apt-get", "/var/cache", "package=", "stderr", "--no-remove", "reboot "} { + if strings.Contains(strings.ToLower(finding.Evidence), forbidden) { + t.Fatalf("finding evidence exposes command/path/package/reboot authority: %q", finding.Evidence) + } + } +} + +func runAPTWorkflowFindingJourney(t *testing.T, resources *ResourceHandlers, finding *ai.Finding, capability string) { + t.Helper() + aiHandler, patrol, _, _ := setupAIHandlerWithPatrol(t) + notifications := make(chan relay.PushNotificationPayload, 1) + patrol.SetPushNotifyCallback(func(payload relay.PushNotificationPayload) { notifications <- payload }) + if !patrol.GetFindings().Add(finding) { + t.Fatal("deterministic finding was not admitted") + } + investigations := newTestInvestigationStore() + investigation := investigations.Create(finding.ID, "bounded-investigation-session") + aiHandler.investigationStores = map[string]aicontracts.InvestigationStore{"default": investigations} + aiHandler.SetResourceStoreProvider(resources.getStore) + resources.SetActionTransitionPublisher(aiHandler.ReconcilePatrolActionTransition) + + proposal := aicontracts.ActionProposal{ + ProposalID: "proposal-" + finding.ID, FindingID: finding.ID, InvestigationID: investigation.ID, + ResourceID: finding.ResourceID, CapabilityName: capability, Params: map[string]any{}, + Reason: "Apply the exact typed action supported by the bounded APT finding evidence.", + EvidenceIDs: []string{"finding-evidence:" + finding.ID}, + } + if len(proposal.Params) != 0 { + t.Fatalf("APT proposal params=%#v, want exact empty object", proposal.Params) + } + disposition, err := NewPatrolActionBroker("default", resources).Submit(context.Background(), proposal) + if err != nil { + t.Fatalf("Submit: %v", err) + } + if disposition.State != string(unified.ActionStatePending) || !disposition.Plan.RequiresApproval { + t.Fatalf("planned disposition=%#v", disposition) + } + + decision := httptest.NewRecorder() + decisionReq := httptest.NewRequest(http.MethodPost, "/api/actions/"+disposition.ActionID+"/decision", bytes.NewBufferString(`{"outcome":"approved","reason":"controlled fake-only proof"}`)) + decisionReq.SetPathValue("id", disposition.ActionID) + decisionReq = decisionReq.WithContext(auth.WithUser(decisionReq.Context(), "operator@example.com")) + resources.HandleDecideAction(decision, actionHandlerTestRequest(decisionReq, "")) + if decision.Code != http.StatusOK { + t.Fatalf("decision status=%d body=%s", decision.Code, decision.Body.String()) + } + + execution := httptest.NewRecorder() + executionReq := httptest.NewRequest(http.MethodPost, "/api/actions/"+disposition.ActionID+"/execute", bytes.NewBufferString(`{"reason":"execute approved typed APT action"}`)) + executionReq.SetPathValue("id", disposition.ActionID) + executionReq = executionReq.WithContext(auth.WithUser(executionReq.Context(), "operator@example.com")) + resources.HandleExecuteAction(execution, actionHandlerTestRequest(executionReq, "")) + if execution.Code != http.StatusOK { + t.Fatalf("execution status=%d body=%s", execution.Code, execution.Body.String()) + } + + store, err := resources.getStore("default") + if err != nil { + t.Fatal(err) + } + audit, found, err := store.GetActionAudit(disposition.ActionID) + if err != nil || !found || audit.State != unified.ActionStateCompleted || audit.Origin == nil || audit.Origin.FindingID != finding.ID || len(audit.Request.Params) != 0 { + t.Fatalf("terminal audit found=%v err=%v audit=%#v", found, err, audit) + } + if audit.Result == nil || audit.Result.ActionResultV2 == nil || audit.Result.ActionResultV2.Verification.EvidenceClass != unified.ActionEvidenceAgentAttested { + t.Fatalf("terminal action truth=%#v", audit.Result) + } + reconciled := patrol.GetFindings().Get(finding.ID) + if reconciled == nil || reconciled.ResolvedAt == nil || reconciled.InvestigationOutcome != string(aicontracts.OutcomeFixVerified) { + t.Fatalf("reconciled finding=%#v", reconciled) + } + var notification relay.PushNotificationPayload + select { + case notification = <-notifications: + default: + t.Fatal("terminal finding notification was not published") + } + encoded, err := json.Marshal(struct { + Audit unified.ActionAuditRecord `json:"audit"` + Finding *ai.Finding `json:"finding"` + Notification relay.PushNotificationPayload `json:"notification"` + }{Audit: audit, Finding: reconciled, Notification: notification}) + if err != nil { + t.Fatal(err) + } + for _, forbidden := range []string{"private-package", "raw stderr", "token secret", "/private/cache/path"} { + if strings.Contains(string(encoded), forbidden) { + t.Fatalf("raw APT detail %q escaped terminal projections: %s", forbidden, encoded) + } + } +} + +func runAPTWorkflowReceiptRecoveryJourney(t *testing.T, resource unified.Resource, finding *ai.Finding, capability string, executorFactory func(*ResourceHandlers) ActionExecutor, setTerminal func(unified.ActionDispatchAttempt), counts func() (requests, queries int)) { + t.Helper() + dataPath := t.TempDir() + cfg := &config.Config{DataPath: dataPath} + resources := newActionTestResourceHandlers(t, cfg) + resources.SetStateProvider(resourceUnifiedSeedProvider{snapshot: models.StateSnapshot{LastUpdate: time.Now().UTC()}, resources: []unified.Resource{resource}}) + resources.SetActionExecutor(newRoutedActionExecutor(resources, executorFactory(resources))) + aiHandler, patrol, _, _ := setupAIHandlerWithPatrol(t) + if !patrol.GetFindings().Add(finding) { + t.Fatal("deterministic finding was not admitted") + } + investigations := newTestInvestigationStore() + investigation := investigations.Create(finding.ID, "receipt-recovery-session") + aiHandler.investigationStores = map[string]aicontracts.InvestigationStore{"default": investigations} + aiHandler.SetResourceStoreProvider(resources.getStore) + resources.SetActionTransitionPublisher(aiHandler.ReconcilePatrolActionTransition) + disposition, err := NewPatrolActionBroker("default", resources).Submit(context.Background(), aicontracts.ActionProposal{ + ProposalID: "recovery-" + finding.ID, FindingID: finding.ID, InvestigationID: investigation.ID, ResourceID: finding.ResourceID, CapabilityName: capability, Params: map[string]any{}, Reason: "Exercise durable typed receipt recovery.", + }) + if err != nil { + t.Fatal(err) + } + decisionReq := httptest.NewRequest(http.MethodPost, "/api/actions/"+disposition.ActionID+"/decision", bytes.NewBufferString(`{"outcome":"approved","reason":"controlled recovery proof"}`)) + decisionReq.SetPathValue("id", disposition.ActionID) + decisionReq = actionHandlerTestRequest(decisionReq, "operator@example.com") + decision := httptest.NewRecorder() + resources.HandleDecideAction(decision, decisionReq) + if decision.Code != http.StatusOK { + t.Fatalf("decision status=%d body=%s", decision.Code, decision.Body.String()) + } + executionReq := httptest.NewRequest(http.MethodPost, "/api/actions/"+disposition.ActionID+"/execute", bytes.NewBufferString(`{"reason":"controlled callback loss"}`)) + executionReq.SetPathValue("id", disposition.ActionID) + executionReq = actionHandlerTestRequest(executionReq, "operator@example.com") + resources.HandleExecuteAction(httptest.NewRecorder(), executionReq) + store, err := resources.getStore("default") + if err != nil { + t.Fatal(err) + } + attempt, found, err := store.GetActionDispatchAttempt(disposition.ActionID) + if err != nil || !found || attempt.State != unified.ActionDispatchReceiptPending { + t.Fatalf("pending attempt found=%v err=%v attempt=%#v", found, err, attempt) + } + setTerminal(attempt) + + restarted := NewResourceHandlers(cfg) + restarted.SetStateProvider(resourceUnifiedSeedProvider{snapshot: models.StateSnapshot{LastUpdate: time.Now().UTC()}, resources: []unified.Resource{resource}}) + restarted.SetActionExecutor(newRoutedActionExecutor(restarted, executorFactory(restarted))) + restarted.SetActionTransitionPublisher(aiHandler.ReconcilePatrolActionTransition) + aiHandler.SetResourceStoreProvider(restarted.getStore) + recovered, err := restarted.ActionLifecycle().RecoverExecutingActions(context.Background(), "default", "system:apt-receipt-recovery", 10) + if err != nil || len(recovered) != 1 || recovered[0].State != unified.ActionStateCompleted { + t.Fatalf("recovered=%#v err=%v", recovered, err) + } + requests, queries := counts() + if requests != 1 || queries != 1 { + t.Fatalf("requests=%d queries=%d; recovery must query once without resend", requests, queries) + } + reconciled := patrol.GetFindings().Get(finding.ID) + if reconciled == nil || reconciled.ResolvedAt == nil || reconciled.InvestigationOutcome != string(aicontracts.OutcomeFixVerified) { + t.Fatalf("reconciled finding=%#v", reconciled) + } +} diff --git a/internal/api/contract_test.go b/internal/api/contract_test.go index 44b44503b..8b6ebfea0 100644 --- a/internal/api/contract_test.go +++ b/internal/api/contract_test.go @@ -20044,7 +20044,10 @@ func TestContract_HostUpdatesUseTypedFingerprintBoundAgentOperation(t *testing.T "ExpectedInventoryHash: resource.Agent.PackageUpdates.InventoryHash", "agentexec.HostUpdateOperationInstall", "beforeBound := result.Before.InventoryHash == resource.Agent.PackageUpdates.InventoryHash", - "hostAPTExecutionResult(record.Request.ResourceID, agentID, agentexec.HostUpdateOperationInstall, output, result.Success, result.MutationStarted, result.Verification, beforeBound, result.Before.CheckedAt, result.After.CheckedAt, e.currentTime())", + "agentexec.ValidateHostUpdateResultForRequestAt(req, *result, receivedAt)", + "receivedAt := e.currentTime()", + "hostAPTExecutionResult(record.Request.ResourceID, agentID, agentexec.HostUpdateOperationInstall, output, result.Success, result.MutationStarted, result.Verification, beforeBound, true, result.HealthChecked, result.PackageManagerHealthy, result.RecoveryRequired, result.Before.CheckedAt, result.After.CheckedAt, receivedAt, receivedAt)", + "query.Record.TerminalAt, receivedAt)", } { if !strings.Contains(executorSrc, snippet) { t.Fatalf("host update executor missing typed operation invariant %q", snippet) @@ -20055,6 +20058,9 @@ func TestContract_HostUpdatesUseTypedFingerprintBoundAgentOperation(t *testing.T } for _, snippet := range []string{ "ExpectedInventoryHash string `json:\"expected_inventory_hash\"`", + "HealthChecked bool `json:\"health_checked\"`", + "PackageManagerHealthy bool `json:\"package_manager_healthy\"`", + "RecoveryRequired bool `json:\"recovery_required\"`", "const HostUpdateOperationInstall = \"install_os_updates\"", } { if !strings.Contains(typesSrc, snippet) { @@ -20065,11 +20071,27 @@ func TestContract_HostUpdatesUseTypedFingerprintBoundAgentOperation(t *testing.T "if before.InventoryHash != strings.TrimSpace(req.ExpectedInventoryHash)", "\"--no-remove\"", "\"Dpkg::Options::=--force-confold\"", + "m.run(ctx, nil, \"dpkg\", \"--audit\")", } { if !strings.Contains(runtimeSrc, snippet) { t.Fatalf("host update runtime missing fail-closed command-catalog invariant %q", snippet) } } + refreshIndex := strings.Index(runtimeSrc, `m.run(ctx, nil, "apt-get", "update")`) + mutationIndex := strings.Index(runtimeSrc, "result.MutationStarted = true") + installIndex := strings.Index(runtimeSrc, `m.run(ctx, env, "apt-get", "-y", "--no-remove"`) + if refreshIndex < 0 || mutationIndex <= refreshIndex || installIndex <= mutationIndex { + t.Fatal("host update MutationStarted must be set only immediately before the fixed install command, after refresh and preflight") + } + truthSource, err := os.ReadFile("host_apt_action_result.go") + if err != nil { + t.Fatal(err) + } + for _, snippet := range []string{"package_manager_health_unknown", "package_manager_unhealthy", "ActionEvidenceAgentAttested"} { + if !strings.Contains(string(truthSource), snippet) { + t.Fatalf("host update truth projection missing legacy/health invariant %q", snippet) + } + } } func TestContract_HostStorageCleanupIsTypedFingerprintBoundAndPathFree(t *testing.T) { @@ -20099,7 +20121,11 @@ func TestContract_HostStorageCleanupIsTypedFingerprintBoundAndPathFree(t *testin "agentexec.HostStorageCleanupOperationPackageCache", "unified.HostStorageCleanupPressureDisk(resource.Agent.Disks)", "beforeBound := result.Before.Fingerprint == resource.Agent.StorageCleanup.Fingerprint", - "hostAPTExecutionResult(record.Request.ResourceID, resource.Agent.AgentID, agentexec.HostStorageCleanupOperationPackageCache, output, result.Success, result.MutationStarted, result.Verification, beforeBound, result.Before.CheckedAt, result.After.CheckedAt, e.currentTime())", + "agentexec.ValidateHostStorageCleanupResultForRequestAt(req, *result, receivedAt)", + "hostStorageCleanupResultSummary(*result)", + "receivedAt := e.currentTime()", + "hostAPTExecutionResult(record.Request.ResourceID, resource.Agent.AgentID, agentexec.HostStorageCleanupOperationPackageCache, output, result.Success, result.MutationStarted, result.Verification, beforeBound, false, false, false, false, result.Before.CheckedAt, result.After.CheckedAt, receivedAt, receivedAt)", + "query.Record.TerminalAt, receivedAt)", } { if !strings.Contains(executorSrc, snippet) { t.Fatalf("host storage cleanup executor missing invariant %q", snippet) @@ -20130,6 +20156,11 @@ func TestContract_HostStorageCleanupIsTypedFingerprintBoundAndPathFree(t *testin t.Fatalf("host storage cleanup runtime missing fail-closed invariant %q", snippet) } } + for _, snippet := range []string{"rollback available: false", "rescan required: %t"} { + if !strings.Contains(executorSrc, snippet) { + t.Fatalf("host storage cleanup truth projection missing %q", snippet) + } + } } func TestContract_DurableOperationReceiptCapabilityGatesAPTActions(t *testing.T) { diff --git a/internal/api/host_apt_action_result.go b/internal/api/host_apt_action_result.go index daaf408d8..ab8cfe7de 100644 --- a/internal/api/host_apt_action_result.go +++ b/internal/api/host_apt_action_result.go @@ -9,22 +9,33 @@ import ( unified "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" ) -func hostAPTExecutionResult(resourceID, agentID, operation, output string, success, mutationStarted bool, verification string, beforeStateBound bool, beforeObservedAt, afterObservedAt, receivedAt time.Time) (*unified.ExecutionResult, error) { +func hostAPTExecutionResult(resourceID, agentID, operation, output string, success, mutationStarted bool, verification string, beforeStateBound, packageManagerHealthRequired, healthChecked, packageManagerHealthy, recoveryRequired bool, beforeObservedAt, afterObservedAt, observationBoundaryAt, receivedAt time.Time) (*unified.ExecutionResult, error) { + output = strings.TrimSpace(output) execution := unified.ActionExecutionTruth{Status: unified.ActionExecutionSucceeded, Summary: output} if !success { if mutationStarted { - execution = unified.ActionExecutionTruth{Status: unified.ActionExecutionInconclusive, ReasonCode: "possible_partial_effect", Summary: "The agent could not establish the final mutation outcome."} + execution = unified.ActionExecutionTruth{Status: unified.ActionExecutionInconclusive, ReasonCode: "possible_partial_effect", Summary: output} } else { - execution = unified.ActionExecutionTruth{Status: unified.ActionExecutionNotRun, ReasonCode: "preflight_refused", Summary: "The typed host operation did not begin mutation."} + execution = unified.ActionExecutionTruth{Status: unified.ActionExecutionNotRun, ReasonCode: "preflight_refused", Summary: output} } } - verificationTruth := unified.ActionVerificationTruth{Status: unified.ActionVerificationInconclusive, EvidenceClass: unified.ActionEvidenceNone, ReasonCode: "agent_readback_inconclusive", Summary: "The agent readback was inconclusive and did not establish the requested postcondition."} + verificationSummary := "The agent readback was inconclusive and did not establish the requested postcondition." + if output != "" { + verificationSummary = output + } + verificationTruth := unified.ActionVerificationTruth{Status: unified.ActionVerificationInconclusive, EvidenceClass: unified.ActionEvidenceNone, ReasonCode: "agent_readback_inconclusive", Summary: verificationSummary} if verification == agentexec.HostUpdateVerificationVerified || verification == agentexec.HostStorageCleanupVerificationVerified || verification == agentexec.HostUpdateVerificationFailed || verification == agentexec.HostStorageCleanupVerificationFailed { if !beforeStateBound { verificationTruth.ReasonCode = "before_state_mismatch" verificationTruth.Summary = "The agent readback did not match the request-bound before state." - } else if !freshHostAPTResultObservation(beforeObservedAt, afterObservedAt, receivedAt) { + } else if packageManagerHealthRequired && verification == agentexec.HostUpdateVerificationVerified && !healthChecked { + verificationTruth.ReasonCode = "package_manager_health_unknown" + verificationTruth.Summary = "The terminal receipt predates or lacks a completed bounded package-manager health check." + } else if packageManagerHealthRequired && verification == agentexec.HostUpdateVerificationVerified && !packageManagerHealthy { + verificationTruth.ReasonCode = "package_manager_unhealthy" + verificationTruth.Summary = "The package-manager health check contradicted the claimed healthy postcondition." + } else if !freshHostAPTResultObservation(beforeObservedAt, afterObservedAt, observationBoundaryAt) { verificationTruth.ReasonCode = "stale_agent_readback" verificationTruth.Summary = "The agent readback was stale, skewed, or had invalid mutation chronology." } else { @@ -32,7 +43,7 @@ func hostAPTExecutionResult(resourceID, agentID, operation, output string, succe Version: unified.ActionEvidenceVersion, ID: operation + "-agent-readback", ObserverID: agentID, ObserverKind: "unified_agent", ObserverTrustDomain: "agent:" + agentID, ExecutorTrustDomain: "agent:" + agentID, Method: "typed_read_after_write", SubjectID: resourceID, ObservedAt: afterObservedAt.UTC(), ReceivedAt: receivedAt.UTC(), - Summary: "Bounded typed readback received from the mutating agent.", + Summary: output, }) if err == nil { status := unified.ActionVerificationConfirmed @@ -44,11 +55,17 @@ func hostAPTExecutionResult(resourceID, agentID, operation, output string, succe } } + compensationSummary := "Rollback is unavailable and no recovery action is required." + if operation == agentexec.HostStorageCleanupOperationPackageCache { + compensationSummary = "Package-cache cleanup is irreversible and non-rollbackable; a rescan reports the actual effect and never restoration." + } else if recoveryRequired { + compensationSummary = "No automatic rollback is available; separately governed recovery is required before another update attempt." + } canonical := unified.ActionResultV2{ Version: unified.ActionResultV2Version, Execution: execution, Verification: verificationTruth, - Compensation: unified.ActionCompensationTruth{Support: unified.ActionCompensationUnavailable, Status: unified.ActionCompensationNotAvailable}, + Compensation: unified.ActionCompensationTruth{Support: unified.ActionCompensationUnavailable, Status: unified.ActionCompensationNotAvailable, Summary: compensationSummary}, } legacy := &unified.ExecutionResult{Output: strings.TrimSpace(output)} projected, _, err := unified.ApplyActionResultV2(legacy, canonical) diff --git a/internal/api/host_apt_action_result_test.go b/internal/api/host_apt_action_result_test.go index 245f9aa70..20fc09b14 100644 --- a/internal/api/host_apt_action_result_test.go +++ b/internal/api/host_apt_action_result_test.go @@ -8,13 +8,13 @@ import ( unified "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" ) -func TestHostAPTActionResultRefreshFailureIsPossiblePartialEffect(t *testing.T) { +func TestHostAPTActionResultRefreshFailureIsNotRunBeforeInstallMutation(t *testing.T) { now := time.Date(2026, 7, 12, 9, 0, 0, 0, time.UTC) - result, err := hostAPTExecutionResult("agent:host-1", "host-1", agentexec.HostUpdateOperationInstall, "refresh failed", false, true, agentexec.HostUpdateVerificationInconclusive, true, time.Time{}, time.Time{}, now) + result, err := hostAPTExecutionResult("agent:host-1", "host-1", agentexec.HostUpdateOperationInstall, "refresh failed", false, false, agentexec.HostUpdateVerificationInconclusive, true, true, false, false, false, time.Time{}, time.Time{}, now, now) if err != nil { t.Fatal(err) } - if result.ActionResultV2.Execution.Status != unified.ActionExecutionInconclusive || result.ActionResultV2.Execution.ReasonCode != "possible_partial_effect" || result.ActionResultV2.Execution.Status == unified.ActionExecutionNotRun { + if result.ActionResultV2.Execution.Status != unified.ActionExecutionNotRun || result.ActionResultV2.Execution.ReasonCode != "preflight_refused" { t.Fatalf("canonical execution truth = %#v", result.ActionResultV2.Execution) } if result.ActionResultV2.Compensation.Support != unified.ActionCompensationUnavailable || result.ActionResultV2.Compensation.Status != unified.ActionCompensationNotAvailable { @@ -25,7 +25,7 @@ func TestHostAPTActionResultRefreshFailureIsPossiblePartialEffect(t *testing.T) func TestHostAPTActionResultPreservesAgentObservedAndServerReceivedTimes(t *testing.T) { checkedAt := time.Date(2026, 7, 12, 8, 59, 0, 0, time.UTC) observedAt := time.Date(2026, 7, 12, 9, 0, 0, 0, time.UTC) - result, err := hostAPTExecutionResult("agent:host-1", "host-1", agentexec.HostUpdateOperationInstall, "updates complete", true, true, agentexec.HostUpdateVerificationVerified, true, checkedAt.Add(-time.Second), checkedAt, observedAt) + result, err := hostAPTExecutionResult("agent:host-1", "host-1", agentexec.HostUpdateOperationInstall, "updates complete", true, true, agentexec.HostUpdateVerificationVerified, true, true, true, true, false, checkedAt.Add(-time.Second), checkedAt, observedAt, observedAt) if err != nil { t.Fatal(err) } @@ -43,7 +43,7 @@ func TestHostAPTActionResultPreservesAgentObservedAndServerReceivedTimes(t *test func TestHostAPTActionResultFutureVerifiedClaimFailsClosed(t *testing.T) { now := time.Date(2026, 7, 12, 9, 0, 0, 0, time.UTC) - result, err := hostAPTExecutionResult("agent:host-1", "host-1", agentexec.HostUpdateOperationInstall, "updates complete", true, true, agentexec.HostUpdateVerificationVerified, true, now, now.Add(unified.HostAPTTelemetryMaxClockSkew+time.Second), now) + result, err := hostAPTExecutionResult("agent:host-1", "host-1", agentexec.HostUpdateOperationInstall, "updates complete", true, true, agentexec.HostUpdateVerificationVerified, true, true, true, true, false, now, now.Add(unified.HostAPTTelemetryMaxClockSkew+time.Second), now, now) if err != nil { t.Fatal(err) } @@ -54,7 +54,7 @@ func TestHostAPTActionResultFutureVerifiedClaimFailsClosed(t *testing.T) { func TestHostAPTActionResultStaleReadbackFailsClosed(t *testing.T) { now := time.Date(2026, 7, 12, 9, 0, 0, 0, time.UTC) - result, err := hostAPTExecutionResult("agent:host-1", "host-1", agentexec.HostUpdateOperationInstall, "updates complete", true, true, agentexec.HostUpdateVerificationVerified, true, now.Add(-time.Hour-time.Minute), now.Add(-time.Hour), now) + result, err := hostAPTExecutionResult("agent:host-1", "host-1", agentexec.HostUpdateOperationInstall, "updates complete", true, true, agentexec.HostUpdateVerificationVerified, true, true, true, true, false, now.Add(-time.Hour-time.Minute), now.Add(-time.Hour), now, now) if err != nil { t.Fatal(err) } @@ -66,7 +66,7 @@ func TestHostAPTActionResultStaleReadbackFailsClosed(t *testing.T) { func TestHostAPTActionResultBeforeStateMismatchCannotBecomeContradictionEvidence(t *testing.T) { now := time.Date(2026, 7, 12, 9, 0, 0, 0, time.UTC) - result, err := hostAPTExecutionResult("agent:host-1", "host-1", agentexec.HostStorageCleanupOperationPackageCache, "cleanup contradicted", true, true, agentexec.HostStorageCleanupVerificationFailed, false, now.Add(-time.Minute), now, now) + result, err := hostAPTExecutionResult("agent:host-1", "host-1", agentexec.HostStorageCleanupOperationPackageCache, "cleanup contradicted", true, true, agentexec.HostStorageCleanupVerificationFailed, false, false, false, false, false, now.Add(-time.Minute), now, now, now) if err != nil { t.Fatal(err) } @@ -75,3 +75,15 @@ func TestHostAPTActionResultBeforeStateMismatchCannotBecomeContradictionEvidence t.Fatalf("mismatched before-state truth=%#v", truth) } } + +func TestHostAPTActionResultExplicitUnhealthyManagerCannotConfirmVerifiedClaim(t *testing.T) { + now := time.Date(2026, 7, 12, 9, 0, 0, 0, time.UTC) + result, err := hostAPTExecutionResult("agent:host-1", "host-1", agentexec.HostUpdateOperationInstall, "phase=complete; package manager health: unhealthy", true, true, agentexec.HostUpdateVerificationVerified, true, true, true, false, false, now.Add(-time.Second), now, now, now) + if err != nil { + t.Fatal(err) + } + truth := result.ActionResultV2.Verification + if truth.Status != unified.ActionVerificationInconclusive || truth.EvidenceClass != unified.ActionEvidenceNone || truth.ReasonCode != "package_manager_unhealthy" { + t.Fatalf("unhealthy manager verification truth=%#v", truth) + } +} diff --git a/internal/api/host_apt_partial_truth_test.go b/internal/api/host_apt_partial_truth_test.go new file mode 100644 index 000000000..b7b9b9bef --- /dev/null +++ b/internal/api/host_apt_partial_truth_test.go @@ -0,0 +1,184 @@ +package api + +import ( + "encoding/json" + "strings" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/agentexec" + unified "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" +) + +func TestHostUpdatePartialTruthProjectsPhaseHealthRemainingAndRecovery(t *testing.T) { + now := time.Date(2026, 7, 12, 10, 0, 0, 0, time.UTC) + snapshot := func(hash string, pending int, checkedAt time.Time) agentexec.HostPackageUpdateSnapshot { + return agentexec.HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: hash, PendingCount: pending, CheckedAt: checkedAt} + } + cases := []struct { + name string + payload agentexec.HostUpdateResultPayload + wantExecution unified.ActionExecutionStatus + wantVerification unified.ActionVerificationStatus + wantSummary []string + }{ + { + name: "preflight refusal", + payload: agentexec.HostUpdateResultPayload{RequestID: "preflight.dispatch.1", ActionID: "preflight", ExecutionPhase: agentexec.HostUpdatePhasePreflight, + Before: snapshot(testHostPackageInventoryHash, 3, now), After: snapshot(testHostPackageInventoryHash, 3, now), Verification: agentexec.HostUpdateVerificationInconclusive}, + wantExecution: unified.ActionExecutionNotRun, wantVerification: unified.ActionVerificationInconclusive, + wantSummary: []string{"phase=preflight", "3 pending after", "health: unknown", "recovery required: false"}, + }, + { + name: "refresh failure", + payload: agentexec.HostUpdateResultPayload{RequestID: "refresh.dispatch.1", ActionID: "refresh", ExecutionPhase: agentexec.HostUpdatePhaseRefresh, + Before: snapshot(testHostPackageInventoryHash, 3, now), After: snapshot(testHostPackageInventoryHash, 3, now), Verification: agentexec.HostUpdateVerificationInconclusive}, + wantExecution: unified.ActionExecutionNotRun, wantVerification: unified.ActionVerificationInconclusive, + wantSummary: []string{"phase=refresh", "3 pending after", "health: unknown", "recovery required: false"}, + }, + { + name: "partial install unhealthy", + payload: agentexec.HostUpdateResultPayload{RequestID: "install.dispatch.1", ActionID: "install", ExecutionPhase: agentexec.HostUpdatePhaseInstall, MutationStarted: true, HealthChecked: true, PackageManagerHealthy: false, RecoveryRequired: true, + Before: snapshot(testHostPackageInventoryHash, 3, now.Add(-time.Second)), After: snapshot(testHostPackageInventoryHash, 2, now), Verification: agentexec.HostUpdateVerificationFailed}, + wantExecution: unified.ActionExecutionInconclusive, wantVerification: unified.ActionVerificationContradicted, + wantSummary: []string{"phase=install", "2 pending after", "health: unhealthy", "recovery required: true"}, + }, + { + name: "verify failure with remaining updates", + payload: agentexec.HostUpdateResultPayload{RequestID: "verify.dispatch.1", ActionID: "verify", Success: true, ExecutionPhase: agentexec.HostUpdatePhaseVerify, MutationStarted: true, HealthChecked: true, PackageManagerHealthy: true, RecoveryRequired: true, + Before: snapshot(testHostPackageInventoryHash, 3, now.Add(-time.Second)), After: snapshot(testHostPackageEmptyInventoryHash, 1, now), Verification: agentexec.HostUpdateVerificationFailed}, + wantExecution: unified.ActionExecutionSucceeded, wantVerification: unified.ActionVerificationContradicted, + wantSummary: []string{"phase=verify", "1 pending after", "health: healthy", "recovery required: true"}, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if err := agentexec.ValidateHostUpdateResultPayload(&tc.payload); err != nil { + t.Fatalf("valid typed result rejected: %v", err) + } + summary := hostUpdateResultSummary(tc.payload) + result, err := hostAPTExecutionResult("agent:host-1", "agent-1", agentexec.HostUpdateOperationInstall, summary, tc.payload.Success, tc.payload.MutationStarted, tc.payload.Verification, true, true, tc.payload.HealthChecked, tc.payload.PackageManagerHealthy, tc.payload.RecoveryRequired, tc.payload.Before.CheckedAt, tc.payload.After.CheckedAt, now, now) + if err != nil { + t.Fatal(err) + } + if result.ActionResultV2.Execution.Status != tc.wantExecution || result.ActionResultV2.Verification.Status != tc.wantVerification { + t.Fatalf("truth=%#v", result.ActionResultV2) + } + for _, want := range tc.wantSummary { + if !strings.Contains(result.ActionResultV2.Execution.Summary, want) { + t.Fatalf("execution summary %q missing %q", result.ActionResultV2.Execution.Summary, want) + } + } + if result.ActionResultV2.Compensation.Support != unified.ActionCompensationUnavailable || result.ActionResultV2.Compensation.Status != unified.ActionCompensationNotAvailable { + t.Fatalf("compensation=%#v", result.ActionResultV2.Compensation) + } + if tc.payload.RecoveryRequired { + if !strings.Contains(result.ActionResultV2.Compensation.Summary, "separately governed recovery is required") { + t.Fatalf("partial update compensation=%#v", result.ActionResultV2.Compensation) + } + } else if strings.Contains(result.ActionResultV2.Compensation.Summary, "recovery is required") || !strings.Contains(result.ActionResultV2.Compensation.Summary, "no recovery action is required") { + t.Fatalf("non-recovery update compensation=%#v", result.ActionResultV2.Compensation) + } + }) + } +} + +func TestHostStorageCleanupPartialTruthIsNonRollbackableAndRequiresRescan(t *testing.T) { + now := time.Date(2026, 7, 12, 10, 0, 0, 0, time.UTC) + for _, tc := range []struct { + name string + phase string + reclaimed int64 + }{ + {name: "clean failure", phase: agentexec.HostStorageCleanupPhaseClean}, + {name: "verify failure with measured effect", phase: agentexec.HostStorageCleanupPhaseVerify, reclaimed: 128 * 1024 * 1024}, + } { + t.Run(tc.name, func(t *testing.T) { + payload := agentexec.HostStorageCleanupResultPayload{ + RequestID: "cleanup.dispatch.1", ActionID: "cleanup", ExecutionPhase: tc.phase, MutationStarted: true, + Before: agentexec.HostStorageCleanupSnapshot{Supported: true, Provider: "apt-package-cache", Fingerprint: testHostStorageCleanupFingerprint, ReclaimableBytes: 512 * 1024 * 1024, CheckedAt: now.Add(-time.Second)}, + After: agentexec.HostStorageCleanupSnapshot{Supported: true, Provider: "apt-package-cache", Fingerprint: testHostStorageCleanupAfterFingerprint, ReclaimableBytes: 512*1024*1024 - tc.reclaimed, CheckedAt: now}, + ReclaimedBytes: tc.reclaimed, Verification: agentexec.HostStorageCleanupVerificationFailed, + } + if tc.phase == agentexec.HostStorageCleanupPhaseVerify { + payload.Success = true + } + if err := agentexec.ValidateHostStorageCleanupResultPayload(&payload); err != nil { + t.Fatal(err) + } + summary := hostStorageCleanupResultSummary(payload) + result, err := hostAPTExecutionResult("agent:host-cleanup", "agent-1", agentexec.HostStorageCleanupOperationPackageCache, summary, payload.Success, true, payload.Verification, true, false, false, false, false, payload.Before.CheckedAt, payload.After.CheckedAt, now, now) + if err != nil { + t.Fatal(err) + } + wantExecution := unified.ActionExecutionInconclusive + if tc.phase == agentexec.HostStorageCleanupPhaseVerify { + wantExecution = unified.ActionExecutionSucceeded + } + if result.ActionResultV2.Execution.Status != wantExecution || result.ActionResultV2.Verification.Status != unified.ActionVerificationContradicted || !strings.Contains(summary, "rollback available: false") || !strings.Contains(summary, "rescan required: true") || !strings.Contains(summary, "phase="+tc.phase) { + t.Fatalf("truth=%#v summary=%q", result.ActionResultV2, summary) + } + if !strings.Contains(result.ActionResultV2.Compensation.Summary, "irreversible and non-rollbackable") || !strings.Contains(result.ActionResultV2.Compensation.Summary, "never restoration") { + t.Fatalf("cleanup compensation=%#v", result.ActionResultV2.Compensation) + } + }) + } +} + +func TestHostUpdateResultRejectsImpossibleHealthAndRecoveryCombinations(t *testing.T) { + base := agentexec.HostUpdateResultPayload{RequestID: "bad.dispatch.1", ActionID: "bad", ExecutionPhase: agentexec.HostUpdatePhaseInstall, MutationStarted: true, RecoveryRequired: true, Verification: agentexec.HostUpdateVerificationInconclusive} + cases := map[string]func(*agentexec.HostUpdateResultPayload){ + "refresh cannot claim install mutation": func(p *agentexec.HostUpdateResultPayload) { p.ExecutionPhase = agentexec.HostUpdatePhaseRefresh }, + "healthy without check": func(p *agentexec.HostUpdateResultPayload) { p.PackageManagerHealthy = true; p.HealthChecked = false }, + "partial install without recovery": func(p *agentexec.HostUpdateResultPayload) { + p.ExecutionPhase = agentexec.HostUpdatePhaseInstall + p.RecoveryRequired = false + }, + "success complete with recovery": func(p *agentexec.HostUpdateResultPayload) { + p.Success = true + p.ExecutionPhase = agentexec.HostUpdatePhaseComplete + p.HealthChecked = true + p.PackageManagerHealthy = true + }, + "preflight recovery without mutation": func(p *agentexec.HostUpdateResultPayload) { + p.ExecutionPhase = agentexec.HostUpdatePhasePreflight + p.MutationStarted = false + }, + } + for name, mutate := range cases { + t.Run(name, func(t *testing.T) { + payload := base + mutate(&payload) + if err := agentexec.ValidateHostUpdateResultPayload(&payload); err == nil { + t.Fatalf("impossible result accepted: %#v", payload) + } + }) + } +} + +func TestHostAPTActionTruthRedactsRawAgentAndPackageDetail(t *testing.T) { + now := time.Now().UTC() + payload := agentexec.HostUpdateResultPayload{ + RequestID: "redact.dispatch.1", ActionID: "redact", ExecutionPhase: agentexec.HostUpdatePhaseInstall, MutationStarted: true, RecoveryRequired: true, + Before: agentexec.HostPackageUpdateSnapshot{PendingCount: 2, Packages: []agentexec.HostPackageUpdate{{Name: "private-package"}}, Error: "repo token secret", CheckedAt: now.Add(-time.Second)}, + After: agentexec.HostPackageUpdateSnapshot{PendingCount: 1, Packages: []agentexec.HostPackageUpdate{{Name: "private-package"}}, Error: "stderr /private/cache/path", CheckedAt: now}, + Verification: agentexec.HostUpdateVerificationInconclusive, Error: "raw stderr token secret", + } + result, err := hostAPTExecutionResult("agent:host", "agent", agentexec.HostUpdateOperationInstall, hostUpdateResultSummary(payload), false, true, payload.Verification, true, true, false, false, true, payload.Before.CheckedAt, payload.After.CheckedAt, now, now) + if err != nil { + t.Fatal(err) + } + encoded, err := json.Marshal(struct { + Result *unified.ExecutionResult `json:"result"` + Finding string `json:"finding"` + Notice string `json:"notification"` + }{Result: result, Finding: "APT update requires recovery", Notice: "Host update outcome needs review"}) + if err != nil { + t.Fatal(err) + } + for _, forbidden := range []string{"private-package", "repo token", "raw stderr", "/private/cache/path", "token secret"} { + if strings.Contains(string(encoded), forbidden) { + t.Fatalf("raw detail %q escaped canonical projections: %s", forbidden, encoded) + } + } +} diff --git a/internal/api/host_storage_cleanup_action_executor.go b/internal/api/host_storage_cleanup_action_executor.go index a88c5d754..f0b7eb9a2 100644 --- a/internal/api/host_storage_cleanup_action_executor.go +++ b/internal/api/host_storage_cleanup_action_executor.go @@ -110,10 +110,14 @@ func (e hostStorageCleanupActionExecutor) ExecuteAction(ctx context.Context, rec if result == nil { return nil, fmt.Errorf("host storage cleanup agent returned no result") } + receivedAt := e.currentTime() + if err := agentexec.ValidateHostStorageCleanupResultForRequestAt(req, *result, receivedAt); err != nil { + return nil, fmt.Errorf("invalid host storage cleanup agent result: %w", err) + } - output := fmt.Sprintf("APT package cache: %d bytes before, %d bytes after, %d bytes reclaimed", result.Before.ReclaimableBytes, result.After.ReclaimableBytes, result.ReclaimedBytes) + output := hostStorageCleanupResultSummary(*result) beforeBound := result.Before.Fingerprint == resource.Agent.StorageCleanup.Fingerprint - return hostAPTExecutionResult(record.Request.ResourceID, resource.Agent.AgentID, agentexec.HostStorageCleanupOperationPackageCache, output, result.Success, result.MutationStarted, result.Verification, beforeBound, result.Before.CheckedAt, result.After.CheckedAt, e.currentTime()) + return hostAPTExecutionResult(record.Request.ResourceID, resource.Agent.AgentID, agentexec.HostStorageCleanupOperationPackageCache, output, result.Success, result.MutationStarted, result.Verification, beforeBound, false, false, false, false, result.Before.CheckedAt, result.After.CheckedAt, receivedAt, receivedAt) } func (e hostStorageCleanupActionExecutor) ReconcileActionDispatch(ctx context.Context, record unified.ActionAuditRecord, attempt unified.ActionDispatchAttempt) (*unified.ExecutionResult, unified.ActionDispatchReceipt, bool, error) { @@ -129,7 +133,8 @@ func (e hostStorageCleanupActionExecutor) ReconcileActionDispatch(ctx context.Co if query.Status != operationreceipt.QueryFoundTerminal { return nil, unified.ActionDispatchReceipt{}, false, nil } - if err := agentexec.ValidateOperationQueryResultForIdentity(query, identity, e.currentTime()); err != nil { + receivedAt := e.currentTime() + if err := agentexec.ValidateOperationQueryResultForIdentity(query, identity, receivedAt); err != nil { return nil, unified.ActionDispatchReceipt{}, false, err } result, err := agentexec.DecodeHostStorageCleanupResultPayload(query.Record.Result) @@ -137,18 +142,22 @@ func (e hostStorageCleanupActionExecutor) ReconcileActionDispatch(ctx context.Co return nil, unified.ActionDispatchReceipt{}, false, err } req := agentexec.HostStorageCleanupPayload{RequestID: attempt.ID, ActionID: record.ID, Operation: attempt.OperationKind, OperationVersion: attempt.OperationVersion, RequestDigest: attempt.RequestDigest, ExpectedFingerprint: result.Before.Fingerprint} - if err := agentexec.ValidateHostStorageCleanupResultForRequestAt(req, result, e.currentTime()); err != nil { + if err := agentexec.ValidateHostStorageCleanupResultForRequest(req, result); err != nil { return nil, unified.ActionDispatchReceipt{}, false, err } - output := fmt.Sprintf("APT package cache: %d bytes before, %d bytes after, %d bytes reclaimed", result.Before.ReclaimableBytes, result.After.ReclaimableBytes, result.ReclaimedBytes) - execution, buildErr := hostAPTExecutionResult(record.Request.ResourceID, attempt.AgentID, attempt.OperationKind, output, result.Success, result.MutationStarted, result.Verification, true, result.Before.CheckedAt, result.After.CheckedAt, e.currentTime()) + output := hostStorageCleanupResultSummary(result) + execution, buildErr := hostAPTExecutionResult(record.Request.ResourceID, attempt.AgentID, attempt.OperationKind, output, result.Success, result.MutationStarted, result.Verification, true, false, false, false, false, result.Before.CheckedAt, result.After.CheckedAt, query.Record.TerminalAt, receivedAt) if buildErr != nil { return nil, unified.ActionDispatchReceipt{}, false, buildErr } - receipt := unified.ActionDispatchReceipt{AttemptID: attempt.ID, ActionID: record.ID, TransportRequestID: attempt.ID, ReceivedAt: e.currentTime()} + receipt := unified.ActionDispatchReceipt{AttemptID: attempt.ID, ActionID: record.ID, TransportRequestID: attempt.ID, ReceivedAt: receivedAt} return execution, receipt, true, nil } +func hostStorageCleanupResultSummary(result agentexec.HostStorageCleanupResultPayload) string { + return fmt.Sprintf("APT package cache: phase=%s; %d bytes before, %d bytes after, %d bytes reclaimed; rollback available: false; rescan required: %t", result.ExecutionPhase, result.Before.ReclaimableBytes, result.After.ReclaimableBytes, result.ReclaimedBytes, result.MutationStarted && result.Verification != agentexec.HostStorageCleanupVerificationVerified) +} + func (e hostStorageCleanupActionExecutor) currentResource(ctx context.Context, resourceID string) (unified.Resource, error) { if e.resources == nil { return unified.Resource{}, fmt.Errorf("resource handler unavailable") diff --git a/internal/api/host_storage_cleanup_action_executor_test.go b/internal/api/host_storage_cleanup_action_executor_test.go index 88d29caec..6dd03bd32 100644 --- a/internal/api/host_storage_cleanup_action_executor_test.go +++ b/internal/api/host_storage_cleanup_action_executor_test.go @@ -2,6 +2,7 @@ package api import ( "context" + "encoding/json" "strings" "testing" "time" @@ -23,6 +24,14 @@ type fakeHostStorageCleanupAgent struct { requests []agentexec.HostStorageCleanupPayload receiptVersion int receiptVersionSet bool + queryResult operationreceipt.QueryResult + queryErr error + queries []operationreceipt.Identity +} + +func (f *fakeHostStorageCleanupAgent) QueryAgentOperation(_ context.Context, _ string, identity operationreceipt.Identity) (operationreceipt.QueryResult, error) { + f.queries = append(f.queries, identity) + return f.queryResult, f.queryErr } var testHostStorageCleanupFingerprint = "sha256:" + strings.Repeat("c", 64) @@ -31,7 +40,13 @@ var testHostStorageCleanupAfterFingerprint = "sha256:" + strings.Repeat("d", 64) func (f *fakeHostStorageCleanupAgent) ExecuteHostStorageCleanup(_ context.Context, agentID string, req agentexec.HostStorageCleanupPayload) (*agentexec.HostStorageCleanupResultPayload, error) { f.agentID = agentID f.requests = append(f.requests, req) - return f.result, f.err + if f.result == nil { + return nil, f.err + } + result := *f.result + result.RequestID = req.RequestID + result.ActionID = req.ActionID + return &result, f.err } func (f *fakeHostStorageCleanupAgent) IsAgentConnected(string) bool { return f.connected } @@ -44,7 +59,13 @@ func (f *fakeHostStorageCleanupAgent) AgentOperationReceiptVersion(string) int { func hostCleanupDispatchTestContext(t *testing.T, actionID string) context.Context { t.Helper() - attempt, err := unified.NewActionDispatchAttempt(actionID, time.Now()) + attempt := hostCleanupDispatchAttempt(t, actionID, time.Now()) + return actionlifecycle.ContextWithCommittedDispatchAttempt(context.Background(), attempt) +} + +func hostCleanupDispatchAttempt(t *testing.T, actionID string, now time.Time) unified.ActionDispatchAttempt { + t.Helper() + attempt, err := unified.NewActionDispatchAttempt(actionID, now) if err != nil { t.Fatal(err) } @@ -56,7 +77,7 @@ func hostCleanupDispatchTestContext(t *testing.T, actionID string) context.Conte if err != nil { t.Fatal(err) } - return actionlifecycle.ContextWithCommittedDispatchAttempt(context.Background(), attempt) + return attempt } func TestHostStorageCleanupActionExecutorDispatchesFingerprintBoundOperation(t *testing.T) { @@ -89,7 +110,7 @@ func TestHostStorageCleanupActionExecutorDoesNotExposeAgentErrorText(t *testing. h := NewResourceHandlers(&config.Config{DataPath: t.TempDir()}) h.SetStateProvider(resourceUnifiedSeedProvider{snapshot: models.StateSnapshot{LastUpdate: now}, resources: []unified.Resource{hostStorageCleanupActionResource(now)}}) agents := &fakeHostStorageCleanupAgent{connected: true, result: &agentexec.HostStorageCleanupResultPayload{ - RequestID: "action-cleanup", Verification: agentexec.HostStorageCleanupVerificationFailed, + RequestID: "action-cleanup", ExecutionPhase: agentexec.HostStorageCleanupPhasePreflight, Verification: agentexec.HostStorageCleanupVerificationInconclusive, Before: agentexec.HostStorageCleanupSnapshot{ReclaimableBytes: 512 * 1024 * 1024}, After: agentexec.HostStorageCleanupSnapshot{ReclaimableBytes: 512 * 1024 * 1024}, Error: "private repository package path and token", @@ -105,6 +126,46 @@ func TestHostStorageCleanupActionExecutorDoesNotExposeAgentErrorText(t *testing. } } +func TestHostStorageCleanupReconcileDelayedTerminalReceiptPreservesAgentAttestedEvidenceWithoutResend(t *testing.T) { + terminalAt := time.Now().UTC().Add(-2 * time.Hour) + receivedAt := terminalAt.Add(2 * time.Hour) + attempt := hostCleanupDispatchAttempt(t, "action-cleanup", terminalAt) + payload := agentexec.HostStorageCleanupResultPayload{ + RequestID: attempt.ID, ActionID: attempt.ActionID, Success: true, MutationStarted: true, ExecutionPhase: agentexec.HostStorageCleanupPhaseComplete, + Before: agentexec.HostStorageCleanupSnapshot{Supported: true, Provider: "apt-package-cache", Fingerprint: testHostStorageCleanupFingerprint, ReclaimableBytes: 512 * 1024 * 1024, CheckedAt: terminalAt.Add(-2 * time.Second)}, + After: agentexec.HostStorageCleanupSnapshot{Supported: true, Provider: "apt-package-cache", Fingerprint: testHostStorageCleanupAfterFingerprint, ReclaimableBytes: 8 * 1024 * 1024, CheckedAt: terminalAt.Add(-time.Second)}, + ReclaimedBytes: 504 * 1024 * 1024, Verification: agentexec.HostStorageCleanupVerificationVerified, + } + raw, err := json.Marshal(payload) + if err != nil { + t.Fatal(err) + } + identity := operationreceipt.Identity{AttemptID: attempt.ID, ActionID: attempt.ActionID, OperationKind: attempt.OperationKind, OperationVersion: attempt.OperationVersion, RequestDigest: attempt.RequestDigest, AgentID: attempt.AgentID} + agents := &fakeHostStorageCleanupAgent{queryResult: operationreceipt.QueryResult{Version: operationreceipt.ProtocolVersion, Status: operationreceipt.QueryFoundTerminal, Record: &operationreceipt.Record{ + Identity: identity, State: operationreceipt.StateTerminal, AcceptedAt: terminalAt.Add(-4 * time.Second), StartedAt: terminalAt.Add(-3 * time.Second), TerminalAt: terminalAt, + ResultKind: agentexec.HostStorageCleanupReceiptKind, ResultVersion: agentexec.HostAPTReceiptVersion, Result: raw, + }}} + executor := hostStorageCleanupActionExecutor{agents: agents, now: func() time.Time { return receivedAt }} + + result, receipt, found, err := executor.ReconcileActionDispatch(context.Background(), hostStorageCleanupActionRecord(attempt.ActionID), attempt) + if err != nil || !found { + t.Fatalf("ReconcileActionDispatch: found=%v err=%v", found, err) + } + if len(agents.requests) != 0 || len(agents.queries) != 1 { + t.Fatalf("mutation requests=%d queries=%d", len(agents.requests), len(agents.queries)) + } + if result == nil || result.ActionResultV2 == nil || result.ActionResultV2.Verification.Status != unified.ActionVerificationConfirmed || result.ActionResultV2.Verification.EvidenceClass != unified.ActionEvidenceAgentAttested { + t.Fatalf("result=%#v", result) + } + evidence := result.ActionResultV2.Verification.Evidence + if len(evidence) != 1 || !evidence[0].ObservedAt.Equal(payload.After.CheckedAt) || !evidence[0].ReceivedAt.Equal(receivedAt) { + t.Fatalf("evidence=%#v", evidence) + } + if !receipt.ReceivedAt.Equal(receivedAt) || receipt.TransportRequestID != attempt.ID { + t.Fatalf("receipt=%#v", receipt) + } +} + func TestPatrolFullModeRunsStorageCleanupThroughCanonicalLifecycle(t *testing.T) { now := time.Now().UTC() h := NewResourceHandlers(&config.Config{DataPath: t.TempDir()}) @@ -200,7 +261,7 @@ func TestHostCleanupSelfReportedReceiptSupportCannotBypassLiveServerVersion(t *t func verifiedHostStorageCleanupAgent() *fakeHostStorageCleanupAgent { now := time.Now().UTC() return &fakeHostStorageCleanupAgent{connected: true, result: &agentexec.HostStorageCleanupResultPayload{ - RequestID: "filled-by-executor", Success: true, + RequestID: "filled-by-executor", Success: true, MutationStarted: true, ExecutionPhase: agentexec.HostStorageCleanupPhaseComplete, Before: agentexec.HostStorageCleanupSnapshot{Supported: true, Provider: "apt-package-cache", Fingerprint: testHostStorageCleanupFingerprint, ReclaimableBytes: 512 * 1024 * 1024, CheckedAt: now.Add(-time.Second)}, After: agentexec.HostStorageCleanupSnapshot{Supported: true, Provider: "apt-package-cache", Fingerprint: testHostStorageCleanupAfterFingerprint, ReclaimableBytes: 8 * 1024 * 1024, CheckedAt: now}, ReclaimedBytes: 504 * 1024 * 1024, diff --git a/internal/api/host_update_action_executor.go b/internal/api/host_update_action_executor.go index 81ec711cc..3ca8da063 100644 --- a/internal/api/host_update_action_executor.go +++ b/internal/api/host_update_action_executor.go @@ -115,10 +115,14 @@ func (e hostUpdateActionExecutor) ExecuteAction(ctx context.Context, record unif if result == nil { return nil, fmt.Errorf("host update agent returned no result") } + receivedAt := e.currentTime() + if err := agentexec.ValidateHostUpdateResultForRequestAt(req, *result, receivedAt); err != nil { + return nil, fmt.Errorf("invalid host update agent result: %w", err) + } - output := fmt.Sprintf("APT package updates: %d pending before, %d pending after; reboot required: %t", result.Before.PendingCount, result.After.PendingCount, result.After.RebootRequired) + output := hostUpdateResultSummary(*result) beforeBound := result.Before.InventoryHash == resource.Agent.PackageUpdates.InventoryHash - return hostAPTExecutionResult(record.Request.ResourceID, agentID, agentexec.HostUpdateOperationInstall, output, result.Success, result.MutationStarted, result.Verification, beforeBound, result.Before.CheckedAt, result.After.CheckedAt, e.currentTime()) + return hostAPTExecutionResult(record.Request.ResourceID, agentID, agentexec.HostUpdateOperationInstall, output, result.Success, result.MutationStarted, result.Verification, beforeBound, true, result.HealthChecked, result.PackageManagerHealthy, result.RecoveryRequired, result.Before.CheckedAt, result.After.CheckedAt, receivedAt, receivedAt) } func (e hostUpdateActionExecutor) ReconcileActionDispatch(ctx context.Context, record unified.ActionAuditRecord, attempt unified.ActionDispatchAttempt) (*unified.ExecutionResult, unified.ActionDispatchReceipt, bool, error) { @@ -134,7 +138,8 @@ func (e hostUpdateActionExecutor) ReconcileActionDispatch(ctx context.Context, r if query.Status != operationreceipt.QueryFoundTerminal { return nil, unified.ActionDispatchReceipt{}, false, nil } - if err := agentexec.ValidateOperationQueryResultForIdentity(query, identity, e.currentTime()); err != nil { + receivedAt := e.currentTime() + if err := agentexec.ValidateOperationQueryResultForIdentity(query, identity, receivedAt); err != nil { return nil, unified.ActionDispatchReceipt{}, false, err } result, err := agentexec.DecodeHostUpdateResultPayload(query.Record.Result) @@ -142,18 +147,29 @@ func (e hostUpdateActionExecutor) ReconcileActionDispatch(ctx context.Context, r return nil, unified.ActionDispatchReceipt{}, false, err } req := agentexec.HostUpdatePayload{RequestID: attempt.ID, ActionID: record.ID, Operation: attempt.OperationKind, OperationVersion: attempt.OperationVersion, RequestDigest: attempt.RequestDigest, ExpectedInventoryHash: result.Before.InventoryHash} - if err := agentexec.ValidateHostUpdateResultForRequestAt(req, result, e.currentTime()); err != nil { + if err := agentexec.ValidateHostUpdateResultForRequest(req, result); err != nil { return nil, unified.ActionDispatchReceipt{}, false, err } - output := fmt.Sprintf("APT package updates: %d pending before, %d pending after; reboot required: %t", result.Before.PendingCount, result.After.PendingCount, result.After.RebootRequired) - execution, buildErr := hostAPTExecutionResult(record.Request.ResourceID, attempt.AgentID, attempt.OperationKind, output, result.Success, result.MutationStarted, result.Verification, true, result.Before.CheckedAt, result.After.CheckedAt, e.currentTime()) + output := hostUpdateResultSummary(result) + execution, buildErr := hostAPTExecutionResult(record.Request.ResourceID, attempt.AgentID, attempt.OperationKind, output, result.Success, result.MutationStarted, result.Verification, true, true, result.HealthChecked, result.PackageManagerHealthy, result.RecoveryRequired, result.Before.CheckedAt, result.After.CheckedAt, query.Record.TerminalAt, receivedAt) if buildErr != nil { return nil, unified.ActionDispatchReceipt{}, false, buildErr } - receipt := unified.ActionDispatchReceipt{AttemptID: attempt.ID, ActionID: record.ID, TransportRequestID: attempt.ID, ReceivedAt: e.currentTime()} + receipt := unified.ActionDispatchReceipt{AttemptID: attempt.ID, ActionID: record.ID, TransportRequestID: attempt.ID, ReceivedAt: receivedAt} return execution, receipt, true, nil } +func hostUpdateResultSummary(result agentexec.HostUpdateResultPayload) string { + health := "unknown" + if result.HealthChecked { + health = "unhealthy" + if result.PackageManagerHealthy { + health = "healthy" + } + } + return fmt.Sprintf("APT package updates: phase=%s; %d pending before, %d pending after; package manager health: %s; recovery required: %t; reboot required: %t", result.ExecutionPhase, result.Before.PendingCount, result.After.PendingCount, health, result.RecoveryRequired, result.After.RebootRequired) +} + func (e hostUpdateActionExecutor) currentResource(ctx context.Context, resourceID string) (unified.Resource, error) { if e.resources == nil { return unified.Resource{}, fmt.Errorf("resource handler unavailable") diff --git a/internal/api/host_update_action_executor_test.go b/internal/api/host_update_action_executor_test.go index ee4064b97..15bbf7f3f 100644 --- a/internal/api/host_update_action_executor_test.go +++ b/internal/api/host_update_action_executor_test.go @@ -2,6 +2,7 @@ package api import ( "context" + "encoding/json" "strings" "testing" "time" @@ -23,6 +24,14 @@ type fakeHostUpdateAgent struct { requests []agentexec.HostUpdatePayload receiptVersion int receiptVersionSet bool + queryResult operationreceipt.QueryResult + queryErr error + queries []operationreceipt.Identity +} + +func (f *fakeHostUpdateAgent) QueryAgentOperation(_ context.Context, _ string, identity operationreceipt.Identity) (operationreceipt.QueryResult, error) { + f.queries = append(f.queries, identity) + return f.queryResult, f.queryErr } var testHostPackageInventoryHash = "sha256:" + strings.Repeat("a", 64) @@ -31,7 +40,13 @@ var testHostPackageEmptyInventoryHash = "sha256:" + strings.Repeat("b", 64) func (f *fakeHostUpdateAgent) ExecuteHostUpdate(_ context.Context, agentID string, req agentexec.HostUpdatePayload) (*agentexec.HostUpdateResultPayload, error) { f.agentID = agentID f.requests = append(f.requests, req) - return f.result, f.err + if f.result == nil { + return nil, f.err + } + result := *f.result + result.RequestID = req.RequestID + result.ActionID = req.ActionID + return &result, f.err } func (f *fakeHostUpdateAgent) IsAgentConnected(string) bool { return f.connected } @@ -44,7 +59,13 @@ func (f *fakeHostUpdateAgent) AgentOperationReceiptVersion(string) int { func hostUpdateDispatchTestContext(t *testing.T, actionID string) context.Context { t.Helper() - attempt, err := unified.NewActionDispatchAttempt(actionID, time.Now()) + attempt := hostUpdateDispatchAttempt(t, actionID, time.Now()) + return actionlifecycle.ContextWithCommittedDispatchAttempt(context.Background(), attempt) +} + +func hostUpdateDispatchAttempt(t *testing.T, actionID string, now time.Time) unified.ActionDispatchAttempt { + t.Helper() + attempt, err := unified.NewActionDispatchAttempt(actionID, now) if err != nil { t.Fatal(err) } @@ -56,7 +77,7 @@ func hostUpdateDispatchTestContext(t *testing.T, actionID string) context.Contex if err != nil { t.Fatal(err) } - return actionlifecycle.ContextWithCommittedDispatchAttempt(context.Background(), attempt) + return attempt } func TestHostUpdateActionExecutorDispatchesTypedOperationAndProjectsVerification(t *testing.T) { @@ -67,11 +88,10 @@ func TestHostUpdateActionExecutorDispatchesTypedOperationAndProjectsVerification resources: []unified.Resource{hostUpdateActionResource(now)}, }) agents := &fakeHostUpdateAgent{connected: true, result: &agentexec.HostUpdateResultPayload{ - RequestID: "action-host-update", - Success: true, - Before: agentexec.HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: testHostPackageInventoryHash, PendingCount: 3, CheckedAt: now.Add(-time.Second)}, - After: agentexec.HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: testHostPackageEmptyInventoryHash, PendingCount: 0, RebootRequired: true, CheckedAt: now}, - Verification: agentexec.HostUpdateVerificationVerified, + RequestID: "action-host-update", Success: true, MutationStarted: true, ExecutionPhase: agentexec.HostUpdatePhaseComplete, + Before: agentexec.HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: testHostPackageInventoryHash, PendingCount: 3, CheckedAt: now.Add(-time.Second)}, + After: agentexec.HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: testHostPackageEmptyInventoryHash, PendingCount: 0, RebootRequired: true, CheckedAt: now}, + HealthChecked: true, PackageManagerHealthy: true, Verification: agentexec.HostUpdateVerificationVerified, }} executor := newHostUpdateActionExecutor(h, agents) @@ -99,7 +119,7 @@ func TestHostUpdateActionExecutorReportsInconclusiveVerificationHonestly(t *test h := NewResourceHandlers(&config.Config{DataPath: t.TempDir()}) h.SetStateProvider(resourceUnifiedSeedProvider{snapshot: models.StateSnapshot{LastUpdate: now}, resources: []unified.Resource{hostUpdateActionResource(now)}}) agents := &fakeHostUpdateAgent{connected: true, result: &agentexec.HostUpdateResultPayload{ - RequestID: "action-host-update", Success: true, + RequestID: "action-host-update", Success: true, ExecutionPhase: agentexec.HostUpdatePhaseComplete, Before: agentexec.HostPackageUpdateSnapshot{PendingCount: 2}, After: agentexec.HostPackageUpdateSnapshot{PendingCount: 0}, Verification: agentexec.HostUpdateVerificationInconclusive, @@ -115,15 +135,82 @@ func TestHostUpdateActionExecutorReportsInconclusiveVerificationHonestly(t *test } } +func TestHostUpdateReconcileDelayedTerminalReceiptPreservesAgentAttestedEvidenceWithoutResend(t *testing.T) { + terminalAt := time.Now().UTC().Add(-2 * time.Hour) + receivedAt := terminalAt.Add(2 * time.Hour) + attempt := hostUpdateDispatchAttempt(t, "action-host-update", terminalAt) + payload := agentexec.HostUpdateResultPayload{ + RequestID: attempt.ID, ActionID: attempt.ActionID, Success: true, MutationStarted: true, ExecutionPhase: agentexec.HostUpdatePhaseComplete, + Before: agentexec.HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: testHostPackageInventoryHash, PendingCount: 3, CheckedAt: terminalAt.Add(-2 * time.Second)}, + After: agentexec.HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: testHostPackageEmptyInventoryHash, PendingCount: 0, RebootRequired: true, CheckedAt: terminalAt.Add(-time.Second)}, + HealthChecked: true, PackageManagerHealthy: true, Verification: agentexec.HostUpdateVerificationVerified, + } + raw, err := json.Marshal(payload) + if err != nil { + t.Fatal(err) + } + identity := operationreceipt.Identity{AttemptID: attempt.ID, ActionID: attempt.ActionID, OperationKind: attempt.OperationKind, OperationVersion: attempt.OperationVersion, RequestDigest: attempt.RequestDigest, AgentID: attempt.AgentID} + agents := &fakeHostUpdateAgent{queryResult: operationreceipt.QueryResult{Version: operationreceipt.ProtocolVersion, Status: operationreceipt.QueryFoundTerminal, Record: &operationreceipt.Record{ + Identity: identity, State: operationreceipt.StateTerminal, AcceptedAt: terminalAt.Add(-4 * time.Second), StartedAt: terminalAt.Add(-3 * time.Second), TerminalAt: terminalAt, + ResultKind: agentexec.HostUpdateReceiptKind, ResultVersion: agentexec.HostAPTReceiptVersion, Result: raw, + }}} + executor := hostUpdateActionExecutor{agents: agents, now: func() time.Time { return receivedAt }} + + result, receipt, found, err := executor.ReconcileActionDispatch(context.Background(), hostUpdateActionRecord(attempt.ActionID), attempt) + if err != nil || !found { + t.Fatalf("ReconcileActionDispatch: found=%v err=%v", found, err) + } + if len(agents.requests) != 0 || len(agents.queries) != 1 { + t.Fatalf("mutation requests=%d queries=%d", len(agents.requests), len(agents.queries)) + } + if result == nil || result.ActionResultV2 == nil || result.ActionResultV2.Verification.Status != unified.ActionVerificationConfirmed || result.ActionResultV2.Verification.EvidenceClass != unified.ActionEvidenceAgentAttested { + t.Fatalf("result=%#v", result) + } + evidence := result.ActionResultV2.Verification.Evidence + if len(evidence) != 1 || !evidence[0].ObservedAt.Equal(payload.After.CheckedAt) || !evidence[0].ReceivedAt.Equal(receivedAt) { + t.Fatalf("evidence=%#v", evidence) + } + if !receipt.ReceivedAt.Equal(receivedAt) || receipt.TransportRequestID != attempt.ID { + t.Fatalf("receipt=%#v", receipt) + } +} + +func TestHostUpdateReconcileLegacyTerminalReceiptDowngradesUnknownHealthWithoutResend(t *testing.T) { + terminalAt := time.Now().UTC().Add(-2 * time.Hour) + receivedAt := terminalAt.Add(2 * time.Hour) + attempt := hostUpdateDispatchAttempt(t, "action-host-update", terminalAt) + payload := agentexec.HostUpdateResultPayload{ + RequestID: attempt.ID, ActionID: attempt.ActionID, Success: true, MutationStarted: true, ExecutionPhase: agentexec.HostUpdatePhaseComplete, + Before: agentexec.HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: testHostPackageInventoryHash, PendingCount: 3, CheckedAt: terminalAt.Add(-2 * time.Second)}, + After: agentexec.HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: testHostPackageEmptyInventoryHash, CheckedAt: terminalAt.Add(-time.Second)}, + Verification: agentexec.HostUpdateVerificationVerified, + } + raw, _ := json.Marshal(payload) + identity := operationreceipt.Identity{AttemptID: attempt.ID, ActionID: attempt.ActionID, OperationKind: attempt.OperationKind, OperationVersion: attempt.OperationVersion, RequestDigest: attempt.RequestDigest, AgentID: attempt.AgentID} + agents := &fakeHostUpdateAgent{queryResult: terminalAPTQuery(identity, agentexec.HostUpdateReceiptKind, raw, terminalAt)} + executor := hostUpdateActionExecutor{agents: agents, now: func() time.Time { return receivedAt }} + result, _, found, err := executor.ReconcileActionDispatch(context.Background(), hostUpdateActionRecord(attempt.ActionID), attempt) + if err != nil || !found { + t.Fatalf("ReconcileActionDispatch: found=%v err=%v", found, err) + } + truth := result.ActionResultV2.Verification + if truth.Status != unified.ActionVerificationInconclusive || truth.EvidenceClass != unified.ActionEvidenceNone || truth.ReasonCode != "package_manager_health_unknown" || len(truth.Evidence) != 0 { + t.Fatalf("legacy verification truth=%#v", truth) + } + if len(agents.requests) != 0 || len(agents.queries) != 1 { + t.Fatalf("legacy receipt requests=%d queries=%d", len(agents.requests), len(agents.queries)) + } +} + func TestPatrolFullModeRunsHostUpdateThroughCanonicalLifecycle(t *testing.T) { now := time.Now().UTC() h := NewResourceHandlers(&config.Config{DataPath: t.TempDir()}) h.SetStateProvider(resourceUnifiedSeedProvider{snapshot: models.StateSnapshot{LastUpdate: now}, resources: []unified.Resource{hostUpdateActionResource(now)}}) agents := &fakeHostUpdateAgent{connected: true, result: &agentexec.HostUpdateResultPayload{ RequestID: "filled-by-executor", Success: true, - Before: agentexec.HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: testHostPackageInventoryHash, PendingCount: 3, CheckedAt: now.Add(-time.Second)}, - After: agentexec.HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: testHostPackageEmptyInventoryHash, PendingCount: 0, CheckedAt: now}, - Verification: agentexec.HostUpdateVerificationVerified, + Before: agentexec.HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: testHostPackageInventoryHash, PendingCount: 3, CheckedAt: now.Add(-time.Second)}, + After: agentexec.HostPackageUpdateSnapshot{Supported: true, Manager: "apt", InventoryHash: testHostPackageEmptyInventoryHash, PendingCount: 0, CheckedAt: now}, + HealthChecked: true, PackageManagerHealthy: true, Verification: agentexec.HostUpdateVerificationVerified, }} h.SetActionExecutor(newRoutedActionExecutor(h, newHostUpdateActionExecutor(h, agents))) store, err := h.getStore("default") diff --git a/internal/hostagent/commands_host_update_test.go b/internal/hostagent/commands_host_update_test.go index df87988ac..f300a0524 100644 --- a/internal/hostagent/commands_host_update_test.go +++ b/internal/hostagent/commands_host_update_test.go @@ -24,6 +24,9 @@ func TestCommandClientHandlesTypedHostUpdateWithoutExecuteCommand(t *testing.T) simulations := 0 mutations := 0 manager.run = func(_ context.Context, _ []string, name string, args ...string) packageUpdateCommandResult { + if name == "dpkg" && strings.Join(args, " ") == "--audit" { + return packageUpdateCommandResult{} + } if name != "apt-get" { t.Fatalf("executable = %q, want apt-get", name) } diff --git a/internal/hostagent/operation_receipt_websocket_integration_test.go b/internal/hostagent/operation_receipt_websocket_integration_test.go index 6b55c5804..4ddd49cd2 100644 --- a/internal/hostagent/operation_receipt_websocket_integration_test.go +++ b/internal/hostagent/operation_receipt_websocket_integration_test.go @@ -2,9 +2,12 @@ package hostagent import ( "context" + "errors" "net/http" "net/http/httptest" + "os" "strings" + "sync" "testing" "time" @@ -60,3 +63,114 @@ func TestRealServerAndUnifiedAgentWebSocketQueriesFakeTypedOperationWithoutPacka t.Fatal("client did not stop") } } + +func TestRealServerAndUnifiedAgentWebSocketExecutesAPTThroughFakeTypedManagersAndReplaysWithoutMutation(t *testing.T) { + server := agentexec.NewServer(func(token, agent, host string) bool { return token == "token" }) + httpServer := httptest.NewServer(http.HandlerFunc(server.HandleWebSocket)) + defer httpServer.Close() + logger := zerolog.Nop() + client := NewCommandClient(Config{PulseURL: httpServer.URL, APIToken: "token", StateDir: t.TempDir(), Logger: &logger}, "agent-apt", "host", "linux", "6") + lease := newPackageManagerLease() + updates := newPackageUpdateManager("linux", lease) + cleanup := newStorageCleanupManager("linux", lease) + + const pendingSimulation = "Inst pulse-safe [1.0] (1.1 stable [amd64])\n" + var mu sync.Mutex + upgraded := false + cleaned := false + refreshCalls := 0 + upgradeCalls := 0 + cleanCalls := 0 + fakeRun := func(_ context.Context, _ []string, command string, args ...string) packageUpdateCommandResult { + if command == "dpkg" && strings.Join(args, " ") == "--audit" { + return packageUpdateCommandResult{} + } + if command != "apt-get" { + return packageUpdateCommandResult{err: errors.New("unexpected fake command")} + } + mu.Lock() + defer mu.Unlock() + joined := strings.Join(args, " ") + switch { + case joined == "update": + refreshCalls++ + case strings.Contains(joined, "-y --no-remove"): + upgradeCalls++ + upgraded = true + case joined == "clean": + cleanCalls++ + cleaned = true + case strings.Contains(joined, "-s -o Debug::NoLocking=1 upgrade"): + if !upgraded { + return packageUpdateCommandResult{stdout: pendingSimulation} + } + default: + return packageUpdateCommandResult{err: errors.New("unexpected fake apt catalog call")} + } + return packageUpdateCommandResult{} + } + updates.run = fakeRun + updates.lookPath = func(string) (string, error) { return "/fake/apt-get", nil } + updates.stat = func(string) (os.FileInfo, error) { return nil, os.ErrNotExist } + updates.cacheTTL = 0 + cleanup.run = fakeRun + cleanup.lookPath = func(string) (string, error) { return "/fake/apt-get", nil } + cleanup.cacheTTL = 0 + cleanup.scan = func() (agentexec.HostStorageCleanupSnapshot, error) { + mu.Lock() + defer mu.Unlock() + if cleaned { + return agentexec.HostStorageCleanupSnapshot{Fingerprint: "sha256:" + strings.Repeat("d", 64), ReclaimableBytes: 8 * 1024 * 1024}, nil + } + return agentexec.HostStorageCleanupSnapshot{Fingerprint: "sha256:" + strings.Repeat("c", 64), ReclaimableBytes: 512 * 1024 * 1024}, nil + } + client.packageUpdates = updates + client.storageCleanup = cleanup + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- client.Run(ctx) }() + defer func() { + cancel() + _ = client.Close() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Error("client did not stop") + } + }() + deadline := time.Now().Add(3 * time.Second) + for !server.IsAgentConnected("agent-apt") && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + if !server.IsAgentConnected("agent-apt") { + t.Fatal("real command client did not connect") + } + + updateReq := agentexec.HostUpdatePayload{RequestID: "update.dispatch.1", ActionID: "update", Operation: agentexec.HostUpdateOperationInstall, ExpectedInventoryHash: aptUpgradeInventoryHash(pendingSimulation), Timeout: 5} + if err := agentexec.BindHostUpdatePayload(&updateReq); err != nil { + t.Fatal(err) + } + for i := 0; i < 2; i++ { + result, err := server.ExecuteHostUpdate(context.Background(), "agent-apt", updateReq) + if err != nil || result == nil || !result.Success || result.Verification != agentexec.HostUpdateVerificationVerified { + t.Fatalf("update replay %d result=%+v err=%v", i, result, err) + } + } + + cleanupReq := agentexec.HostStorageCleanupPayload{RequestID: "cleanup.dispatch.1", ActionID: "cleanup", Operation: agentexec.HostStorageCleanupOperationPackageCache, ExpectedFingerprint: "sha256:" + strings.Repeat("c", 64), Timeout: 5} + if err := agentexec.BindHostStorageCleanupPayload(&cleanupReq); err != nil { + t.Fatal(err) + } + for i := 0; i < 2; i++ { + result, err := server.ExecuteHostStorageCleanup(context.Background(), "agent-apt", cleanupReq) + if err != nil || result == nil || !result.Success || result.Verification != agentexec.HostStorageCleanupVerificationVerified { + t.Fatalf("cleanup replay %d result=%+v err=%v", i, result, err) + } + } + mu.Lock() + defer mu.Unlock() + if refreshCalls != 1 || upgradeCalls != 1 || cleanCalls != 1 { + t.Fatalf("fake mutation catalog calls: refresh=%d upgrade=%d clean=%d", refreshCalls, upgradeCalls, cleanCalls) + } +} diff --git a/internal/hostagent/package_updates.go b/internal/hostagent/package_updates.go index c158612c8..2af66bfa5 100644 --- a/internal/hostagent/package_updates.go +++ b/internal/hostagent/package_updates.go @@ -137,11 +137,11 @@ func (m *packageUpdateManager) Apply(ctx context.Context, req agentexec.HostUpda } result.ExecutionPhase = agentexec.HostUpdatePhaseRefresh - result.MutationStarted = true refresh := m.run(ctx, nil, "apt-get", "update") if refresh.err != nil || refresh.exitCode != 0 { result.Before = probe result.After = probe + result.HealthChecked, result.PackageManagerHealthy = m.checkPackageManagerHealth(ctx) result.Verification = agentexec.HostUpdateVerificationInconclusive result.Error = "package index refresh failed" return result @@ -151,23 +151,39 @@ func (m *packageUpdateManager) Apply(ctx context.Context, req agentexec.HostUpda result.Before = before if before.Error != "" { result.After = before + result.HealthChecked, result.PackageManagerHealthy = m.checkPackageManagerHealth(ctx) result.Verification = agentexec.HostUpdateVerificationInconclusive result.Error = "package update preflight failed" return result } if before.InventoryHash != strings.TrimSpace(req.ExpectedInventoryHash) { result.After = before + result.HealthChecked, result.PackageManagerHealthy = m.checkPackageManagerHealth(ctx) result.Verification = agentexec.HostUpdateVerificationInconclusive result.Error = "package update inventory changed; replan required" return result } if before.PendingCount == 0 { + result.After = before + result.HealthChecked, result.PackageManagerHealthy = m.checkPackageManagerHealth(ctx) + if !result.HealthChecked || !result.PackageManagerHealthy { + result.Verification = agentexec.HostUpdateVerificationInconclusive + result.Error = "package manager health could not be established" + return result + } result.Success = true result.ExecutionPhase = agentexec.HostUpdatePhaseComplete - result.After = before result.Verification = agentexec.HostUpdateVerificationVerified return result } + result.HealthChecked, result.PackageManagerHealthy = m.checkPackageManagerHealth(ctx) + if !result.HealthChecked || !result.PackageManagerHealthy { + result.After = before + result.ExecutionPhase = agentexec.HostUpdatePhasePreflight + result.Verification = agentexec.HostUpdateVerificationInconclusive + result.Error = "package manager health check refused installation" + return result + } env := []string{ "DEBIAN_FRONTEND=noninteractive", @@ -175,10 +191,13 @@ func (m *packageUpdateManager) Apply(ctx context.Context, req agentexec.HostUpda "NEEDRESTART_MODE=a", } result.ExecutionPhase = agentexec.HostUpdatePhaseInstall + result.MutationStarted = true install := m.run(ctx, env, "apt-get", "-y", "--no-remove", "-o", "Dpkg::Options::=--force-confold", "upgrade") if install.err != nil || install.exitCode != 0 { after := m.snapshotLocked(ctx, true) result.After = after + result.HealthChecked, result.PackageManagerHealthy = m.checkPackageManagerHealth(ctx) + result.RecoveryRequired = true result.Verification = agentexec.HostUpdateVerificationFailed result.Error = "package installation failed" return result @@ -189,20 +208,42 @@ func (m *packageUpdateManager) Apply(ctx context.Context, req agentexec.HostUpda after := m.snapshotLocked(ctx, true) result.After = after if after.Error != "" { + result.HealthChecked, result.PackageManagerHealthy = m.checkPackageManagerHealth(ctx) + result.RecoveryRequired = true result.Verification = agentexec.HostUpdateVerificationInconclusive result.Error = "package installation completed but verification was inconclusive" return result } if after.PendingCount != 0 { + result.HealthChecked, result.PackageManagerHealthy = m.checkPackageManagerHealth(ctx) + result.RecoveryRequired = true result.Verification = agentexec.HostUpdateVerificationFailed result.Error = "package installation completed but updates remain pending" return result } + result.HealthChecked, result.PackageManagerHealthy = m.checkPackageManagerHealth(ctx) + if !result.HealthChecked || !result.PackageManagerHealthy { + result.Verification = agentexec.HostUpdateVerificationInconclusive + result.RecoveryRequired = true + result.Error = "package installation completed but package manager health was not established" + return result + } result.Verification = agentexec.HostUpdateVerificationVerified result.ExecutionPhase = agentexec.HostUpdatePhaseComplete return result } +func (m *packageUpdateManager) checkPackageManagerHealth(ctx context.Context) (checked, healthy bool) { + if _, err := m.lookPath("dpkg"); err != nil { + return false, false + } + result := m.run(ctx, nil, "dpkg", "--audit") + if result.err != nil && result.exitCode < 0 { + return false, false + } + return true, result.err == nil && result.exitCode == 0 && strings.TrimSpace(result.stdout) == "" +} + type packageUpdateCommandResult struct { stdout string stderr string diff --git a/internal/hostagent/package_updates_test.go b/internal/hostagent/package_updates_test.go index 4ed142c82..690935de6 100644 --- a/internal/hostagent/package_updates_test.go +++ b/internal/hostagent/package_updates_test.go @@ -35,10 +35,10 @@ func TestPackageUpdateManagerApplyUsesClosedAPTCommandCatalogAndVerifies(t *test beforeOutput := "Inst openssl [1.0] (1.1 repo [amd64])\nInst curl [8.0] (8.1 repo [amd64])\n" m := newPackageUpdateManager("linux", newPackageManagerLease()) m.lookPath = func(name string) (string, error) { - if name != "apt-get" { - t.Fatalf("lookPath(%q), want apt-get", name) + if name != "apt-get" && name != "dpkg" { + t.Fatalf("lookPath(%q), want fixed apt-get/dpkg catalog", name) } - return "/usr/bin/apt-get", nil + return "/usr/bin/" + name, nil } fileInfo, err := os.Stat(t.TempDir()) if err != nil { @@ -60,6 +60,9 @@ func TestPackageUpdateManagerApplyUsesClosedAPTCommandCatalogAndVerifies(t *test simulations := 0 m.run = func(_ context.Context, env []string, name string, args ...string) packageUpdateCommandResult { calls = append(calls, invocation{env: append([]string(nil), env...), name: name, args: append([]string(nil), args...)}) + if name == "dpkg" && strings.Join(args, " ") == "--audit" { + return packageUpdateCommandResult{} + } if name != "apt-get" { return packageUpdateCommandResult{exitCode: -1, err: errors.New("unexpected executable")} } @@ -85,13 +88,13 @@ func TestPackageUpdateManagerApplyUsesClosedAPTCommandCatalogAndVerifies(t *test if result.Before.PendingCount != 2 || result.After.PendingCount != 0 || !result.After.RebootRequired { t.Fatalf("before/after = %#v / %#v", result.Before, result.After) } - if len(calls) != 5 { - t.Fatalf("calls = %d, want probe, refresh, preflight, install, verify", len(calls)) + if len(calls) != 7 { + t.Fatalf("calls = %d, want probe, refresh, preflight, pre-install health, install, verify, post-install health", len(calls)) } if got := strings.Join(calls[1].args, " "); got != "update" { t.Fatalf("refresh args = %q", got) } - install := calls[3] + install := calls[4] if install.name != "apt-get" || !strings.Contains(strings.Join(install.args, " "), "--no-remove") || !strings.Contains(strings.Join(install.args, " "), "--force-confold") { t.Fatalf("install invocation = %#v", install) } @@ -110,8 +113,11 @@ func TestPackageUpdateManagerFailsClosedWhenRefreshFails(t *testing.T) { m.lookPath = func(string) (string, error) { return "/usr/bin/apt-get", nil } m.stat = func(string) (os.FileInfo, error) { return nil, os.ErrNotExist } calls := 0 - m.run = func(_ context.Context, _ []string, _ string, args ...string) packageUpdateCommandResult { + m.run = func(_ context.Context, _ []string, name string, args ...string) packageUpdateCommandResult { calls++ + if name == "dpkg" { + return packageUpdateCommandResult{} + } if strings.Join(args, " ") == "update" { return packageUpdateCommandResult{exitCode: 100, err: errors.New("exit status 100")} } @@ -122,11 +128,11 @@ func TestPackageUpdateManagerFailsClosedWhenRefreshFails(t *testing.T) { if result.Success || result.Verification != agentexec.HostUpdateVerificationInconclusive || result.Error != "package index refresh failed" { t.Fatalf("result = %#v", result) } - if !result.MutationStarted || result.ExecutionPhase != agentexec.HostUpdatePhaseRefresh { - t.Fatalf("refresh failure must report possible external effect: %#v", result) + if result.MutationStarted || result.RecoveryRequired || result.ExecutionPhase != agentexec.HostUpdatePhaseRefresh { + t.Fatalf("refresh failure must not claim install mutation or recovery: %#v", result) } - if calls != 2 { - t.Fatalf("calls = %d, want probe and refresh only", calls) + if calls != 3 { + t.Fatalf("calls = %d, want probe, refresh, and bounded health check", calls) } } @@ -141,7 +147,7 @@ func TestPackageUpdateManagerCanceledRefreshReportsPossibleExternalEffect(t *tes return packageUpdateCommandResult{stdout: "Inst openssl [1.0] (1.1 repo [amd64])\n"} } result := m.Apply(context.Background(), agentexec.HostUpdatePayload{RequestID: "r1", ActionID: "a1", Operation: agentexec.HostUpdateOperationInstall, ExpectedInventoryHash: aptUpgradeInventoryHash("Inst openssl [1.0] (1.1 repo [amd64])\n")}) - if !result.MutationStarted || result.ExecutionPhase != agentexec.HostUpdatePhaseRefresh || result.Success { + if result.MutationStarted || result.RecoveryRequired || result.ExecutionPhase != agentexec.HostUpdatePhaseRefresh || result.Success { t.Fatalf("canceled refresh truth = %#v", result) } } @@ -178,6 +184,119 @@ func TestPackageUpdateManagerRefusesRefreshTimeInventoryDriftBeforeInstall(t *te if installCalls != 0 { t.Fatalf("install calls = %d, want zero on preflight drift", installCalls) } + if result.MutationStarted || result.RecoveryRequired { + t.Fatalf("refresh-time drift claimed install mutation/recovery: %#v", result) + } +} + +func TestPackageUpdateManagerPartialInstallAndVerifyFailureCarryExplicitHealthAndRecovery(t *testing.T) { + pending := "Inst openssl [1.0] (1.1 repo [amd64])\nInst curl [8.0] (8.1 repo [amd64])\n" + for _, tc := range []struct { + name string + installFails bool + auditOutput string + wantPhase string + wantHealthy bool + wantRemaining int + wantSuccess bool + }{ + {name: "partial install unhealthy", installFails: true, auditOutput: "Packages are only half configured", wantPhase: agentexec.HostUpdatePhaseInstall, wantRemaining: 2}, + {name: "verify failure healthy manager", wantPhase: agentexec.HostUpdatePhaseVerify, wantHealthy: true, wantRemaining: 1, wantSuccess: true}, + } { + t.Run(tc.name, func(t *testing.T) { + m := newPackageUpdateManager("linux", newPackageManagerLease()) + m.lookPath = func(name string) (string, error) { return "/fake/" + name, nil } + m.stat = func(string) (os.FileInfo, error) { return nil, os.ErrNotExist } + simulations := 0 + audits := 0 + m.run = func(_ context.Context, _ []string, name string, args ...string) packageUpdateCommandResult { + joined := strings.Join(args, " ") + if name == "dpkg" && joined == "--audit" { + audits++ + if audits == 1 { + return packageUpdateCommandResult{} + } + return packageUpdateCommandResult{stdout: tc.auditOutput} + } + if strings.Contains(joined, "-s") { + simulations++ + if simulations <= 2 || tc.installFails { + return packageUpdateCommandResult{stdout: pending} + } + return packageUpdateCommandResult{stdout: "Inst curl [8.0] (8.1 repo [amd64])\n"} + } + if tc.installFails && strings.Contains(joined, "-y --no-remove") { + return packageUpdateCommandResult{exitCode: 100, err: errors.New("controlled install failure")} + } + return packageUpdateCommandResult{} + } + result := m.Apply(context.Background(), agentexec.HostUpdatePayload{RequestID: "partial.dispatch.1", ActionID: "partial", Operation: agentexec.HostUpdateOperationInstall, ExpectedInventoryHash: aptUpgradeInventoryHash(pending)}) + if result.ExecutionPhase != tc.wantPhase || result.Success != tc.wantSuccess || !result.MutationStarted || !result.HealthChecked || result.PackageManagerHealthy != tc.wantHealthy || !result.RecoveryRequired || result.After.PendingCount != tc.wantRemaining { + t.Fatalf("partial truth=%#v", result) + } + if err := agentexec.ValidateHostUpdateResultPayload(&result); err != nil { + t.Fatalf("agent produced invalid partial truth: %v", err) + } + }) + } +} + +func TestPackageUpdateManagerPreInstallHealthUnknownRefusesWithoutMutationOrRecoveryClaim(t *testing.T) { + pending := "Inst openssl [1.0] (1.1 repo [amd64])\n" + m := newPackageUpdateManager("linux", newPackageManagerLease()) + m.lookPath = func(name string) (string, error) { return "/fake/" + name, nil } + m.stat = func(string) (os.FileInfo, error) { return nil, os.ErrNotExist } + installCalls := 0 + m.run = func(_ context.Context, _ []string, name string, args ...string) packageUpdateCommandResult { + joined := strings.Join(args, " ") + if name == "dpkg" { + return packageUpdateCommandResult{exitCode: -1, err: errors.New("controlled audit unavailable")} + } + if strings.Contains(joined, "-s") { + return packageUpdateCommandResult{stdout: pending} + } + if strings.Contains(joined, "-y --no-remove") { + installCalls++ + } + return packageUpdateCommandResult{} + } + result := m.Apply(context.Background(), agentexec.HostUpdatePayload{RequestID: "health.dispatch.1", ActionID: "health", Operation: agentexec.HostUpdateOperationInstall, ExpectedInventoryHash: aptUpgradeInventoryHash(pending)}) + if result.Success || result.MutationStarted || result.RecoveryRequired || result.HealthChecked || result.ExecutionPhase != agentexec.HostUpdatePhasePreflight || installCalls != 0 { + t.Fatalf("pre-install health refusal=%#v installCalls=%d", result, installCalls) + } + if err := agentexec.ValidateHostUpdateResultPayload(&result); err != nil { + t.Fatalf("agent produced invalid refusal truth: %v", err) + } +} + +func TestPackageUpdateManagerZeroPendingAfterRefreshIsDriftWithoutInstallOrVerifiedClaim(t *testing.T) { + pending := "Inst openssl [1.0] (1.1 repo [amd64])\n" + m := newPackageUpdateManager("linux", newPackageManagerLease()) + m.lookPath = func(name string) (string, error) { return "/fake/" + name, nil } + m.stat = func(string) (os.FileInfo, error) { return nil, os.ErrNotExist } + simulations := 0 + installCalls := 0 + m.run = func(_ context.Context, _ []string, name string, args ...string) packageUpdateCommandResult { + joined := strings.Join(args, " ") + if name == "dpkg" { + return packageUpdateCommandResult{} + } + if strings.Contains(joined, "-s") { + simulations++ + if simulations == 1 { + return packageUpdateCommandResult{stdout: pending} + } + return packageUpdateCommandResult{} + } + if strings.Contains(joined, "-y --no-remove") { + installCalls++ + } + return packageUpdateCommandResult{} + } + result := m.Apply(context.Background(), agentexec.HostUpdatePayload{RequestID: "noop.dispatch.1", ActionID: "noop", Operation: agentexec.HostUpdateOperationInstall, ExpectedInventoryHash: aptUpgradeInventoryHash(pending)}) + if result.Success || result.MutationStarted || result.RecoveryRequired || result.ExecutionPhase != agentexec.HostUpdatePhaseRefresh || result.Verification != agentexec.HostUpdateVerificationInconclusive || !strings.Contains(result.Error, "replan") || installCalls != 0 { + t.Fatalf("zero-pending drift=%#v installCalls=%d", result, installCalls) + } } func TestPackageUpdateSnapshotCachesAndUnsupportedPlatformsFailClosed(t *testing.T) { diff --git a/internal/hostagent/storage_cleanup_test.go b/internal/hostagent/storage_cleanup_test.go index d20121c50..c360e49fa 100644 --- a/internal/hostagent/storage_cleanup_test.go +++ b/internal/hostagent/storage_cleanup_test.go @@ -2,6 +2,7 @@ package hostagent import ( "context" + "errors" "os" "path/filepath" "reflect" @@ -100,3 +101,44 @@ func TestStorageCleanupManagerRefusesFingerprintDriftBeforeMutation(t *testing.T t.Fatalf("unexpected drift result: %#v", result) } } + +func TestStorageCleanupManagerFailurePhasesPreserveMeasuredEffect(t *testing.T) { + before := agentexec.HostStorageCleanupSnapshot{Supported: true, Provider: "apt-package-cache", Fingerprint: "sha256:" + strings.Repeat("a", 64), ReclaimableBytes: 400, CheckedAt: time.Now().UTC().Add(-time.Second)} + for _, tc := range []struct { + name string + cleanFails bool + afterBytes int64 + wantPhase string + wantSuccess bool + wantReclaimed int64 + }{ + {name: "clean failure", cleanFails: true, afterBytes: 400, wantPhase: agentexec.HostStorageCleanupPhaseClean}, + {name: "verify failure", afterBytes: 400, wantPhase: agentexec.HostStorageCleanupPhaseVerify, wantSuccess: true}, + } { + t.Run(tc.name, func(t *testing.T) { + manager := newStorageCleanupManager("linux", newPackageManagerLease()) + manager.lookPath = func(string) (string, error) { return "/fake/apt-get", nil } + scans := 0 + manager.scan = func() (agentexec.HostStorageCleanupSnapshot, error) { + scans++ + if scans == 1 { + return before, nil + } + return agentexec.HostStorageCleanupSnapshot{Supported: true, Provider: "apt-package-cache", Fingerprint: "sha256:" + strings.Repeat("b", 64), ReclaimableBytes: tc.afterBytes, CheckedAt: time.Now().UTC()}, nil + } + manager.run = func(context.Context, []string, string, ...string) packageUpdateCommandResult { + if tc.cleanFails { + return packageUpdateCommandResult{exitCode: 100, err: errors.New("controlled clean failure")} + } + return packageUpdateCommandResult{} + } + result := manager.Apply(context.Background(), agentexec.HostStorageCleanupPayload{RequestID: "cleanup.dispatch.1", ActionID: "cleanup", Operation: agentexec.HostStorageCleanupOperationPackageCache, ExpectedFingerprint: before.Fingerprint}) + if result.ExecutionPhase != tc.wantPhase || result.Success != tc.wantSuccess || !result.MutationStarted || result.ReclaimedBytes != tc.wantReclaimed || result.Verification != agentexec.HostStorageCleanupVerificationFailed { + t.Fatalf("cleanup truth=%#v", result) + } + if err := agentexec.ValidateHostStorageCleanupResultPayload(&result); err != nil { + t.Fatalf("agent produced invalid cleanup truth: %v", err) + } + }) + } +}