diff --git a/.github/workflows/canonical-governance.yml b/.github/workflows/canonical-governance.yml index a559b82a2..ae03f9425 100644 --- a/.github/workflows/canonical-governance.yml +++ b/.github/workflows/canonical-governance.yml @@ -104,6 +104,12 @@ jobs: PULSE_REPO_ROOT_PULSE_MOBILE: ${{ github.workspace }}/repos/pulse-mobile run: python3 scripts/release_control/status_audit.py --check + - name: Validate Pulse Intelligence release-gate schema + run: python3 scripts/release_control/pulse_intelligence_gate.py --validate-only --matrix docs/release-control/v6/internal/pulse-intelligence-release-gate.json + + - name: Run Pulse Intelligence release-gate unit tests + run: python3 scripts/release_control/pulse_intelligence_gate_test.py + - name: Run control plane audit env: PULSE_REPO_ROOT_PULSE: ${{ github.workspace }}/repos/pulse diff --git a/.husky/pre-commit b/.husky/pre-commit index 100b80498..dda495911 100755 --- a/.husky/pre-commit +++ b/.husky/pre-commit @@ -100,6 +100,9 @@ python3 scripts/release_control/canonical_completion_guard.py echo "Running status audit..." python3 scripts/release_control/status_audit.py --check --staged +echo "Validating Pulse Intelligence release-gate schema..." +python3 scripts/release_control/pulse_intelligence_gate.py --validate-only --matrix docs/release-control/v6/internal/pulse-intelligence-release-gate.json + echo "Running registry audit..." python3 scripts/release_control/registry_audit.py --check --staged @@ -119,6 +122,7 @@ python3 scripts/release_control/control_plane_audit_test.py python3 scripts/release_control/contract_audit_test.py python3 scripts/release_control/format_staged_go_test.py python3 scripts/release_control/governance_stage_guard_test.py +python3 scripts/release_control/pulse_intelligence_gate_test.py python3 scripts/release_control/release_promotion_policy_support_test.py python3 scripts/release_control/registry_audit_test.py python3 scripts/release_control/readiness_assertion_guard_test.py diff --git a/docs/release-control/v6/internal/SOURCE_OF_TRUTH.md b/docs/release-control/v6/internal/SOURCE_OF_TRUTH.md index b780ae36f..0bf4f6236 100644 --- a/docs/release-control/v6/internal/SOURCE_OF_TRUTH.md +++ b/docs/release-control/v6/internal/SOURCE_OF_TRUTH.md @@ -463,6 +463,14 @@ Assertion design rules: current stable demo network/SSH/browser path without mutation, and the publish workflow must await demo deployment plus definitive verification; manual SSH deployment is not an acceptable release completion path. +23. Do not authorize limited unattended Pulse Intelligence mutation until the + closed `RG-01` through `RG-12` matrix reports `GO` at the exact release + Git SHA. `scripts/release_control/pulse_intelligence_gate.py` is a + read-only verdict checker: the tracked matrix contains requirements only, + while concrete results come from a separate external or untracked evidence + document bound to the audited SHA selected at runtime. The checker must + never execute real-lab, device, relay, or other mutation-gated commands + itself. ## Locked Decisions diff --git a/docs/release-control/v6/internal/pulse-intelligence-release-gate.json b/docs/release-control/v6/internal/pulse-intelligence-release-gate.json new file mode 100644 index 000000000..b44d5d48d --- /dev/null +++ b/docs/release-control/v6/internal/pulse-intelligence-release-gate.json @@ -0,0 +1,206 @@ +{ + "schema_version": 2, + "matrix_id": "pulse-intelligence-rg-01-rg-12", + "program": "pulse-intelligence", + "result_semantics": { + "PASS": "External evidence satisfies the gate at the audited Git SHA.", + "FAIL": "External evidence bound to the audited Git SHA records a failed proof or an invalid binding.", + "NOT_RUN": "No qualifying external evidence exists for the gate at the audited Git SHA." + }, + "evidence_tiers": [ + "unit", + "integration", + "browser", + "real-lab", + "physical-device", + "live-relay" + ], + "gates": [ + { + "id": "RG-01", + "title": "Read-only investigation and origin isolation", + "owner": "task-01", + "required_evidence_tier": "integration", + "requires_triple_zero": true, + "allows_ancestor_provenance": false, + "non_mutating_proof_commands": [ + { + "id": "rg-01-profile-boundary", + "argv": ["go", "test", "./internal/ai/tools", "-run", "TestPatrolDetectionProfileEnforcesAllowlistedPulseState|TestPatrolInvestigationProfileIsStructurallyReadOnly|TestProposeActionIsInvestigationProfileOnly", "-count=1"] + } + ], + "mutation_gated_commands": [] + }, + { + "id": "RG-02", + "title": "Canonical lifecycle admission", + "owner": "task-02", + "required_evidence_tier": "integration", + "requires_triple_zero": true, + "allows_ancestor_provenance": false, + "non_mutating_proof_commands": [ + { + "id": "rg-02-lifecycle-admission", + "argv": ["go", "test", "./internal/actionlifecycle", "./internal/api", "-run", "TestPolicyAdmissionCommitsApprovalAndExecutingAtomicallyMemoryStore|TestPolicyAdmissionCommitsApprovalAndExecutingAtomicallySQLite|TestPatrolActionBrokerDispatchTimePolicyRevocation", "-count=1"] + } + ], + "mutation_gated_commands": [] + }, + { + "id": "RG-03", + "title": "Typed capability and provider contracts", + "owner": "task-03", + "required_evidence_tier": "integration", + "requires_triple_zero": true, + "allows_ancestor_provenance": false, + "non_mutating_proof_commands": [ + { + "id": "rg-03-capability-contracts", + "argv": ["go", "test", "./internal/agentcapabilities", "./internal/agentexec", "-run", "Test(CanonicalManifestOwnsAgentSurface|CanonicalManifestPinsPulseIntelligenceSurfaceContract|DockerContainerLifecycleWireCarriesFactsNotActionTruth|DockerAgentWireAndRuntimeDeclareNoLocalActionTruthModel|PendingDockerOperationBindsFullIdentityAndRejectsCrossTypeReuse)", "-count=1"] + } + ], + "mutation_gated_commands": [] + }, + { + "id": "RG-04", + "title": "Policy, approval floor, and plan identity", + "owner": "task-04", + "required_evidence_tier": "integration", + "requires_triple_zero": true, + "allows_ancestor_provenance": false, + "non_mutating_proof_commands": [ + { + "id": "rg-04-policy-plan-hash", + "argv": ["go", "test", "./internal/actionlifecycle", "./internal/api", "-run", "TestExecuteUnderPolicyBarrierRevocationsMemoryStore|TestExecuteUnderPolicyBarrierRevocationsSQLite|TestHumanApprovalSurvivesAutomaticPolicyRevocationMemoryStore|TestHumanApprovalSurvivesAutomaticPolicyRevocationSQLite|TestEmergencyStopBlocksHumanAndPolicyAdmissionMemoryStore|TestEmergencyStopBlocksHumanAndPolicyAdmissionSQLite", "-count=1"] + } + ], + "mutation_gated_commands": [] + }, + { + "id": "RG-05", + "title": "Closed canonical mutation registry", + "owner": "task-05", + "required_evidence_tier": "integration", + "requires_triple_zero": true, + "allows_ancestor_provenance": false, + "non_mutating_proof_commands": [ + { + "id": "rg-05-mutation-registry", + "argv": ["go", "test", "./internal/mutationregistry", "./internal/ai/tools", "-run", "Test(EveryRegisteredMutationHasDisposition|InfrastructureAPIRoutesResolveToRegistry|TransportCommandCatalogsResolveToRegistry|PatrolJobRegistrationResolvesToRegistry|RuntimeCandidateAuditNegativeFixtures|ActionRouteMethodAuthorityIsExactAndLookalikesFailClosed|NonAdmittingTransportMessagesCannotCarryDispatchAuthority|UnknownTransportLookalikeFailsClosed|RegisteredModelMutationSchemasResolveToClosedRegistry|RetiredMutationAliasesCannotShadowExtensions)", "-count=1"] + } + ], + "mutation_gated_commands": [] + }, + { + "id": "RG-06", + "title": "Limited unattended autonomy release authorization", + "owner": "task-08", + "required_evidence_tier": "real-lab", + "requires_triple_zero": true, + "allows_ancestor_provenance": false, + "non_mutating_proof_commands": [ + { + "id": "rg-06-autonomy-authority", + "argv": ["go", "test", "./internal/api", "./internal/actionlifecycle", "-run", "Test(PatrolFullModeRunsStorageCleanupThroughCanonicalLifecycle|PatrolFullModeRunsHostUpdateThroughCanonicalLifecycle|PatrolAutopilotAcknowledgementAPIRejectsPublicAuthorityAndAPIToken|PatrolAutopilotActivationRequiresCurrentBoundAcknowledgement|PatrolAutopilotVersionRotationAndRevocationRaceFailClosed|PatrolAutopilotStoreUnavailableDoesNotChangeModeOrFabricateEvidence|EmergencyStopBlocksHumanAndPolicyAdmissionMemoryStore|EmergencyStopBlocksHumanAndPolicyAdmissionSQLite)", "-count=1"] + } + ], + "mutation_gated_commands": [] + }, + { + "id": "RG-07", + "title": "Durable dispatch and reconnect continuity", + "owner": "task-07", + "required_evidence_tier": "integration", + "requires_triple_zero": true, + "allows_ancestor_provenance": false, + "non_mutating_proof_commands": [ + { + "id": "rg-07-durable-delivery", + "argv": ["go", "test", "./internal/actionlifecycle", "./internal/api", "./internal/agentexec", "-run", "Test(SQLiteRestartRecoveryReconcilesReceiptPendingWithoutResend|SQLiteRestartRecoveryNotFoundRemainsReceiptPendingWithoutResend|AuthenticatedServerNotFoundRecoveryStaysQueryOnlyAndReceiptPending|DockerContainerActionExecutorCallbackLossReconcilesReceiptWithoutRedispatch|HostStorageCleanupReconcileDelayedTerminalReceiptPreservesAgentAttestedEvidenceWithoutResend|HostUpdateReconcileDelayedTerminalReceiptPreservesAgentAttestedEvidenceWithoutResend|LegacyAgentWithoutReceiptProtocolRemainsConnectedButTypedMutationFailsClosed)", "-count=1"] + } + ], + "mutation_gated_commands": [] + }, + { + "id": "RG-08", + "title": "Final-SHA Docker lifecycle lab and browser certification", + "owner": "task-06", + "required_evidence_tier": "real-lab", + "requires_triple_zero": false, + "allows_ancestor_provenance": true, + "non_mutating_proof_commands": [ + { + "id": "rg-08-docker-artifact-browser", + "argv": ["env", "PULSE_INTELLIGENCE_LAB_ARTIFACT=", "tests/integration/node_modules/.bin/playwright", "test", "--config=tests/integration/playwright.config.ts", "tests/84-docker-restart-real-lab-artifact.spec.ts", "--project=chromium"] + } + ], + "mutation_gated_commands": [ + { + "id": "rg-08-colima-lab-mutation", + "argv": ["python3", "scripts/intelligence_lab/docker_restart_colima.py", "--run-id", ""], + "authorization": "EXPLICIT_RELEASE_REQUIRED" + } + ] + }, + { + "id": "RG-09", + "title": "Debian and Ubuntu APT update and cleanup certification", + "owner": "task-09", + "required_evidence_tier": "real-lab", + "requires_triple_zero": false, + "allows_ancestor_provenance": false, + "non_mutating_proof_commands": [ + { + "id": "rg-09-apt-fake-contracts", + "argv": ["go", "test", "./internal/api", "./internal/agentexec", "./internal/hostagent", "-run", "Test(APTUpdateDetectorProposalApprovalDispatchAuditAndFindingReconciliation|APTUpdateCallbackLossServerRestartReconcilesTerminalAuditAndFindingWithoutResend|APTCacheCleanupDetectorProposalApprovalDispatchAuditAndFindingReconciliation|APTCacheCleanupCallbackLossServerRestartReconcilesTerminalAuditAndFindingWithoutResend|PackageUpdateManagerApplyUsesClosedAPTCommandCatalogAndVerifies|PackageUpdateManagerFailsClosedWhenRefreshFails|PackageUpdateManagerRefusesRefreshTimeInventoryDriftBeforeInstall|StorageCleanupManagerApplyUsesClosedAPTCatalogAndVerifiesBytes|StorageCleanupManagerRefusesFingerprintDriftBeforeMutation)", "-count=1"] + } + ], + "mutation_gated_commands": [] + }, + { + "id": "RG-10", + "title": "Canonical execution, verification, and compensation truth", + "owner": "task-10", + "required_evidence_tier": "integration", + "requires_triple_zero": false, + "allows_ancestor_provenance": false, + "non_mutating_proof_commands": [ + { + "id": "rg-10-action-result-v2", + "argv": ["go", "test", "./internal/unifiedresources", "./internal/actionlifecycle", "./internal/api", "-run", "Test(ActionResultV2MemoryStoreRoundTrip|ActionResultV2SQLiteRoundTrip|ActionResultV2SurvivesSQLiteReopen|MalformedActionResultV2PersistsFailClosed|MalformedStoredActionResultV2FailsClosed|ExecutorTuplePostDispatchErrorPreservesReceiptPending)", "-count=1"] + } + ], + "mutation_gated_commands": [] + }, + { + "id": "RG-11", + "title": "Clean Product Trust browser certification", + "owner": "task-11", + "required_evidence_tier": "browser", + "requires_triple_zero": true, + "allows_ancestor_provenance": false, + "non_mutating_proof_commands": [ + { + "id": "rg-11-product-trust-playwright", + "argv": ["env", "PLAYWRIGHT_BASE_URL=http://127.0.0.1:5173", "tests/integration/node_modules/.bin/playwright", "test", "--config=tests/integration/playwright.config.ts", "tests/78-monitor-first-patrol-workbench.spec.ts", "tests/81-actions-inbox.spec.ts", "--project=chromium", "--grep", "APT"] + } + ], + "mutation_gated_commands": [] + }, + { + "id": "RG-12", + "title": "Final-SHA physical-device, live-relay, and aggregate certification", + "owner": "task-12", + "required_evidence_tier": "live-relay", + "requires_triple_zero": true, + "allows_ancestor_provenance": false, + "non_mutating_proof_commands": [ + { + "id": "rg-12-aggregate-verdict", + "argv": ["python3", "scripts/release_control/pulse_intelligence_gate.py", "--check", "--matrix", "docs/release-control/v6/internal/pulse-intelligence-release-gate.json", "--sha", "", "--evidence", ""] + } + ], + "mutation_gated_commands": [] + } + ] +} diff --git a/docs/release-control/v6/internal/status.json b/docs/release-control/v6/internal/status.json index 19a5be75e..d17ac0190 100644 --- a/docs/release-control/v6/internal/status.json +++ b/docs/release-control/v6/internal/status.json @@ -3949,10 +3949,10 @@ }, { "id": "RA35", - "summary": "Every Patrol-initiated infrastructure mutation originates as a typed action proposal and reaches execution only through the canonical action lifecycle; unsupported proposals fail closed with no raw-command fallback.", + "summary": "Passed unit and integration proof shows that covered Patrol mutation candidates use typed proposals and the canonical action lifecycle, while covered unsupported and raw-command paths fail closed. This does not certify final-SHA real-lab, browser, physical-device, or live-relay behavior; the aggregate Pulse Intelligence release gate remains blocked.", "kind": "invariant", - "blocking_level": "repo-ready", - "proof_type": "automated", + "blocking_level": "release-ready", + "proof_type": "hybrid", "lane_ids": [ "L6", "L20" @@ -3961,7 +3961,9 @@ "ai-runtime", "api-contracts" ], - "release_gate_ids": [], + "release_gate_ids": [ + "pulse-intelligence-rg-01-rg-12" + ], "proof_commands": [ { "id": "ra35-apt-product-trust-browser", @@ -6207,7 +6209,7 @@ "status": "partial", "completion": { "state": "bounded-residual", - "summary": "Action governance has a first-class governed floor AND the typed action lifecycle is now the only autonomous route for Patrol: investigations run under a core-owned, non-interactive, read-only execution profile and can only emit a side-effect-free typed action proposal (patrol_propose_action), which the shared action lifecycle plans, approves, and executes with plan hashing, remediation locks, plan-drift revalidation, and read-after-write verification. Three policy-authorized verticals are live. Docker/Podman restart is capability-owned low-risk eligible. Agent-managed APT host updates are capability-owned elevated eligible and cross a closed typed host_update protocol with no command, package-selection, removal, or reboot authority; the agent owns metadata refresh, preflight simulation, guarded installation, and post-install simulation, while the audit records verified, failed, or honestly inconclusive outcomes plus reboot-required state. Agent-managed package-cache cleanup is capability-owned low-risk eligible and crosses a separate fingerprint-bound host_storage_cleanup protocol with no command, path, package selector, installed-package removal, or reboot authority; it is available only for meaningful reclaimable bytes on the actual pressured cache filesystem, and the audit verifies the bytes reclaimed. All three require explicit per-resource capability allowlists and optional recurring timezone windows, remain bounded by tenant Patrol mode and full-mode unlock, recheck policy/readiness before decision and dispatch, and record server-owned policy approval plus terminal verification in the same audit trail. Never-auto-remediate, missing/unknown policy, unsupported capabilities, stale/error/empty package or cleanup inventories, cleared storage pressure, dry-run/MFA floors, closed windows, and mode downgrades all fail closed. Action continuity is audit-authoritative and now has a Phase B1 durable transport foundation: execution admission atomically creates the deterministic dispatch attempt/outbox, a one-shot pre-send CAS prevents concurrent duplicate sends, expired pre-send claims safely requeue, post-send uncertainty remains receipt-pending for query-only reconciliation, explicit expiry is durable, and canonical detail plus pending/settled reads are tenant-scoped. Generic transport errors cannot fabricate receipts or terminal truth, legacy executing rows without attempts remain inert, callbacks trigger org-scoped re-reads, investigation reads recover missed transitions by trusted origin, desktop and mobile decisions use the canonical /api/actions routes, terminal verification maps honestly onto Patrol findings and push outcomes, and legacy command-shaped history remains inert. Task 08 Phase B1 also makes requester and decision authority server-owned, binds actor and approval requirement into plan identity, rechecks current tenant/RBAC/token scope, and records multi-actor approvals through revision-CAS append-only decision facts plus atomic lifecycle transitions. MFA-required decisions remain explicitly unavailable until a core verifier validates and consumes action-bound cryptographic evidence; labels and local biometrics are not MFA. Task 08 Phase B2 now makes Patrol Autopilot acknowledgement and activation server-owned: immutable human actor, credential, organization, current version, accepted limits, timestamps, and digest are persisted as append-only config evidence; effective full mode requires the exact current activation and falls back safely for legacy booleans, stale, revoked, expired, malformed, API-token, or cross-tenant evidence. This backend prerequisite does not close M7; Task 11 still owns explicit acknowledgement UX and browser/device proof, and Task 12 owns final certification. Task 10 Phase B1 now owns one versioned ActionResultV2 with separate execution and verification axes, bounded trust-domain evidence, fail-closed redaction and stable digesting, nested compensation truth, durable Memory/SQLite compatibility persistence, executor tuple enforcement, and non-collapsing event, finding, context, telemetry, and relay projections. M4, M6, and claim 19 remain open pending Task 05 producer migration, Task 07 recovery, Tasks 06/09 workflow lab observations, Task 11 browser presentation, and Task 12 certification against the final SHA. Remaining post-RC hardening also includes mobile/relay inbox and push-dedup consumption, broader tenant defaults/inheritance and automation rate budgets, richer dry-run providers, multi-actor UI and MSP delegation/inheritance, device-key/WebAuthn enrollment and verifier integration, additional package-manager and bounded cleanup providers beyond APT, separately governed reboot orchestration, and a real-lab detect-plan-execute-verify-rollback journey proof.", + "summary": "Action governance has a first-class typed lifecycle floor backed by unit and integration evidence: covered Patrol investigations are read-only, typed proposals cross shared planning, policy, approval, plan-hash, durable-delivery, audit, verification, and ActionResultV2 contracts, and covered raw or unsupported paths fail closed. Docker restart is the only implemented typed infrastructure vertical with an ancestor-bound Colima artifact; it is not certified against the final SHA. APT update and package-cache cleanup have typed fake-backed contracts but remain operationally open until disposable Debian and Ubuntu tier-6 evidence exists. Limited unattended autonomy remains NO-GO. Current continuity and fail-closed claims apply only to the passed unit and integration profiles, not to clean Product Trust browser, physical-device, live-relay, or final-SHA real-lab behavior. Task 10 remains the sole owner of execution, verification, evidence, compensation, and rollback truth. Task 12 owns the final RG-01 through RG-12 verdict, which remains blocked by RG-06, RG-08, RG-09, RG-11, and RG-12.", "tracking": [ { "kind": "lane-followup", @@ -7663,6 +7665,27 @@ } ] }, + { + "id": "pulse-intelligence-rg-01-rg-12", + "summary": "Require the closed RG-01 through RG-12 Pulse Intelligence matrix to reach GO at the final Git SHA. Implemented unit and integration contracts do not substitute for missing Docker final-SHA lab/browser, Debian and Ubuntu APT tier-6, clean Product Trust Playwright, physical-device, or live-relay evidence.", + "owner": "task-12", + "blocking_level": "release-ready", + "minimum_evidence_tier": "real-external-e2e", + "status": "blocked", + "verification_doc": "docs/release-control/v6/internal/HIGH_RISK_RELEASE_VERIFICATION_MATRIX.md", + "lane_ids": [ + "L6", + "L20" + ], + "evidence": [ + { + "repo": "pulse", + "path": "docs/release-control/v6/internal/pulse-intelligence-release-gate.json", + "kind": "file", + "evidence_tier": "test-proof" + } + ] + }, { "id": "rc-to-ga-promotion-readiness", "summary": "Confirm Pulse v6 GA promotion is coming from the current pulse/v6-release branch after accumulated post-RC7 fixes, with the 2026-07-02 owner risk acceptance for no RC8 or further current-branch validation, rollback instructions, and a written v5 maintenance-only policy.", diff --git a/scripts/release_control/pulse_intelligence_gate.py b/scripts/release_control/pulse_intelligence_gate.py new file mode 100644 index 000000000..3285a37b4 --- /dev/null +++ b/scripts/release_control/pulse_intelligence_gate.py @@ -0,0 +1,423 @@ +#!/usr/bin/env python3 +"""Validate and evaluate the closed Pulse Intelligence release-gate contract. + +The tracked matrix contains requirements only. Concrete results are read from +an external evidence document whose records are bound to the audited Git SHA. +This program never executes proof or mutation-gated commands. +""" + +from __future__ import annotations + +import argparse +import json +import re +import subprocess +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Callable + + +REPO_ROOT = Path(__file__).resolve().parents[2] +DEFAULT_MATRIX = REPO_ROOT / "docs/release-control/v6/internal/pulse-intelligence-release-gate.json" +GATE_IDS = tuple(f"RG-{number:02d}" for number in range(1, 13)) +APPROVED_OWNERS = { + "RG-01": "task-01", + "RG-02": "task-02", + "RG-03": "task-03", + "RG-04": "task-04", + "RG-05": "task-05", + "RG-06": "task-08", + "RG-07": "task-07", + "RG-08": "task-06", + "RG-09": "task-09", + "RG-10": "task-10", + "RG-11": "task-11", + "RG-12": "task-12", +} +EVIDENCE_TIERS = ("unit", "integration", "browser", "real-lab", "physical-device", "live-relay") +RESULTS = {"PASS", "FAIL", "NOT_RUN"} +KIND_MAX_TIER = { + "source": "unit", + "mock": "integration", + "task-summary": "unit", + "automated-test": "integration", + "browser-artifact": "browser", + "real-lab-artifact": "real-lab", + "physical-device-artifact": "physical-device", + "live-relay-artifact": "live-relay", +} +SHA_RE = re.compile(r"^[0-9a-f]{40}$") + + +class MatrixError(ValueError): + """Raised for malformed governance or evidence input.""" + + +@dataclass(frozen=True) +class Evaluation: + audited_sha: str + gate_results: dict[str, str] + details: tuple[str, ...] + + @property + def verdict(self) -> str: + return "GO" if all(value == "PASS" for value in self.gate_results.values()) else "NO-GO" + + +def _object(value: Any, context: str) -> dict[str, Any]: + if not isinstance(value, dict): + raise MatrixError(f"{context} must be an object") + return value + + +def _array(value: Any, context: str) -> list[Any]: + if not isinstance(value, list): + raise MatrixError(f"{context} must be an array") + return value + + +def _string(value: Any, context: str) -> str: + if not isinstance(value, str) or not value: + raise MatrixError(f"{context} must be a non-empty string") + return value + + +def _full_sha(value: Any, context: str) -> str: + result = _string(value, context) + if not SHA_RE.fullmatch(result): + raise MatrixError(f"{context} must be a full lowercase Git SHA") + return result + + +def _exact_keys(value: dict[str, Any], required: set[str], optional: set[str], context: str) -> None: + missing = sorted(required - value.keys()) + unknown = sorted(value.keys() - required - optional) + if missing: + raise MatrixError(f"{context} missing fields: {', '.join(missing)}") + if unknown: + raise MatrixError(f"{context} has unknown fields: {', '.join(unknown)}") + + +def _validate_command(command: Any, context: str, *, mutation_gated: bool) -> str: + value = _object(command, context) + required = {"id", "argv"} | ({"authorization"} if mutation_gated else set()) + _exact_keys(value, required, set(), context) + command_id = _string(value["id"], f"{context}.id") + argv = _array(value["argv"], f"{context}.argv") + if not argv or any(not isinstance(part, str) or not part for part in argv): + raise MatrixError(f"{context}.argv must contain non-empty string arguments") + if mutation_gated and value["authorization"] != "EXPLICIT_RELEASE_REQUIRED": + raise MatrixError(f"{context}.authorization must be EXPLICIT_RELEASE_REQUIRED") + return command_id + + +def validate_matrix(payload: Any) -> dict[str, Any]: + root = _object(payload, "matrix") + _exact_keys( + root, + {"schema_version", "matrix_id", "program", "result_semantics", "evidence_tiers", "gates"}, + set(), + "matrix", + ) + if root["schema_version"] != 2: + raise MatrixError("matrix.schema_version must be 2") + if root["matrix_id"] != "pulse-intelligence-rg-01-rg-12": + raise MatrixError("matrix.matrix_id is not the canonical Pulse Intelligence matrix") + if root["program"] != "pulse-intelligence": + raise MatrixError("matrix.program must be pulse-intelligence") + if root["result_semantics"] != { + "PASS": "External evidence satisfies the gate at the audited Git SHA.", + "FAIL": "External evidence bound to the audited Git SHA records a failed proof or an invalid binding.", + "NOT_RUN": "No qualifying external evidence exists for the gate at the audited Git SHA.", + }: + raise MatrixError("matrix.result_semantics must define the canonical PASS/FAIL/NOT_RUN meanings") + if tuple(root["evidence_tiers"]) != EVIDENCE_TIERS: + raise MatrixError("matrix.evidence_tiers must contain the canonical ordered tiers") + + gates = _array(root["gates"], "matrix.gates") + gate_ids: list[str] = [] + all_command_ids: set[str] = set() + for index, raw_gate in enumerate(gates): + context = f"matrix.gates[{index}]" + gate = _object(raw_gate, context) + _exact_keys( + gate, + { + "id", + "title", + "owner", + "required_evidence_tier", + "requires_triple_zero", + "allows_ancestor_provenance", + "non_mutating_proof_commands", + "mutation_gated_commands", + }, + set(), + context, + ) + gate_id = _string(gate["id"], f"{context}.id") + gate_ids.append(gate_id) + _string(gate["title"], f"{context}.title") + if gate.get("owner") != APPROVED_OWNERS.get(gate_id): + raise MatrixError(f"{gate_id} owner must be {APPROVED_OWNERS.get(gate_id)!r}") + if gate["required_evidence_tier"] not in EVIDENCE_TIERS: + raise MatrixError(f"{gate_id} has an unknown required evidence tier") + for field in ("requires_triple_zero", "allows_ancestor_provenance"): + if not isinstance(gate[field], bool): + raise MatrixError(f"{gate_id}.{field} must be boolean") + proof_commands = _array(gate["non_mutating_proof_commands"], f"{context}.non_mutating_proof_commands") + if not proof_commands: + raise MatrixError(f"{gate_id} must declare at least one non-mutating proof command") + for command_index, command in enumerate(proof_commands): + command_id = _validate_command( + command, + f"{context}.non_mutating_proof_commands[{command_index}]", + mutation_gated=False, + ) + if command_id in all_command_ids: + raise MatrixError(f"duplicate proof command id {command_id!r}") + all_command_ids.add(command_id) + for command_index, command in enumerate( + _array(gate["mutation_gated_commands"], f"{context}.mutation_gated_commands") + ): + command_id = _validate_command( + command, + f"{context}.mutation_gated_commands[{command_index}]", + mutation_gated=True, + ) + if command_id in all_command_ids: + raise MatrixError(f"duplicate proof command id {command_id!r}") + all_command_ids.add(command_id) + if len(gate_ids) != len(set(gate_ids)): + raise MatrixError("matrix.gates contains duplicate gate ids") + if tuple(gate_ids) != GATE_IDS: + raise MatrixError(f"matrix.gates must contain exactly {', '.join(GATE_IDS)} in order") + return root + + +def _gate_command_ids(gate: dict[str, Any]) -> set[str]: + return { + command["id"] + for field in ("non_mutating_proof_commands", "mutation_gated_commands") + for command in gate[field] + } + + +def validate_evidence(payload: Any, matrix: dict[str, Any]) -> dict[str, Any]: + root = _object(payload, "evidence") + _exact_keys(root, {"schema_version", "matrix_id", "audited_git_sha", "records"}, set(), "evidence") + if root["schema_version"] != 1: + raise MatrixError("evidence.schema_version must be 1") + if root["matrix_id"] != matrix["matrix_id"]: + raise MatrixError("evidence.matrix_id does not match the selected matrix") + _full_sha(root["audited_git_sha"], "evidence.audited_git_sha") + gates = {gate["id"]: gate for gate in matrix["gates"]} + record_ids: set[str] = set() + for index, raw_record in enumerate(_array(root["records"], "evidence.records")): + context = f"evidence.records[{index}]" + record = _object(raw_record, context) + _exact_keys( + record, + {"id", "gate_id", "result", "tier", "kind", "git_sha", "command_id"}, + {"triple_zero", "note"}, + context, + ) + record_id = _string(record["id"], f"{context}.id") + if record_id in record_ids: + raise MatrixError(f"evidence has duplicate record id {record_id!r}") + record_ids.add(record_id) + gate_id = _string(record["gate_id"], f"{context}.gate_id") + if gate_id not in gates: + raise MatrixError(f"{context}.gate_id is not in the closed matrix") + if record["result"] not in RESULTS: + raise MatrixError(f"{context}.result is invalid") + tier = record["tier"] + kind = record["kind"] + if tier not in EVIDENCE_TIERS: + raise MatrixError(f"{context}.tier is invalid") + if kind not in KIND_MAX_TIER: + raise MatrixError(f"{context}.kind is invalid") + if EVIDENCE_TIERS.index(tier) > EVIDENCE_TIERS.index(KIND_MAX_TIER[kind]): + raise MatrixError(f"{context} overstates {kind!r} as {tier!r} evidence") + _full_sha(record["git_sha"], f"{context}.git_sha") + if record["command_id"] not in _gate_command_ids(gates[gate_id]): + raise MatrixError(f"{context}.command_id does not name a command declared by {gate_id}") + if "triple_zero" in record: + triple_zero = _object(record["triple_zero"], f"{context}.triple_zero") + _exact_keys( + triple_zero, + {"unauthorized_mutations", "transport_dispatches", "authority_writes"}, + set(), + f"{context}.triple_zero", + ) + if any(not isinstance(value, int) or isinstance(value, bool) or value < 0 for value in triple_zero.values()): + raise MatrixError(f"{context}.triple_zero values must be non-negative integers") + return root + + +def evaluate_matrix( + matrix: dict[str, Any], + audited_sha: str, + evidence: dict[str, Any] | None, + *, + is_ancestor: Callable[[str, str], bool] | None = None, +) -> Evaluation: + _full_sha(audited_sha, "audited SHA") + details: list[str] = [] + results: dict[str, str] = {} + records = evidence["records"] if evidence is not None else [] + evidence_binding_valid = evidence is None or evidence["audited_git_sha"] == audited_sha + if evidence is not None and not evidence_binding_valid: + details.append( + f"external evidence is bound to {evidence['audited_git_sha']}, not audited SHA {audited_sha}" + ) + + for gate in matrix["gates"]: + gate_id = gate["id"] + gate_records = [record for record in records if record["gate_id"] == gate_id] + qualifying = False + failed = not evidence_binding_valid + for record in gate_records: + if record["git_sha"] != audited_sha: + ancestor = ( + gate["allows_ancestor_provenance"] + and is_ancestor is not None + and is_ancestor(record["git_sha"], audited_sha) + ) + if ancestor: + details.append( + f"{gate_id} provenance {record['id']} is ancestor-bound to {record['git_sha']} and does not certify {audited_sha}" + ) + else: + details.append( + f"{gate_id} evidence {record['id']} is bound to wrong SHA {record['git_sha']}; expected {audited_sha}" + ) + failed = True + continue + if record["result"] == "FAIL": + failed = True + details.append(f"{gate_id} evidence {record['id']} records FAIL") + continue + if record["result"] != "PASS": + continue + if EVIDENCE_TIERS.index(record["tier"]) < EVIDENCE_TIERS.index(gate["required_evidence_tier"]): + details.append( + f"{gate_id} evidence {record['id']} is only {record['tier']}; requires {gate['required_evidence_tier']}" + ) + continue + if gate["requires_triple_zero"]: + triple_zero = record.get("triple_zero") + if triple_zero is None: + failed = True + details.append(f"{gate_id} evidence {record['id']} is missing required triple-zero fields") + continue + if any( + triple_zero[field] != 0 + for field in ("unauthorized_mutations", "transport_dispatches", "authority_writes") + ): + failed = True + details.append(f"{gate_id} evidence {record['id']} does not prove triple zero") + continue + qualifying = True + if failed: + results[gate_id] = "FAIL" + elif qualifying: + results[gate_id] = "PASS" + else: + results[gate_id] = "NOT_RUN" + return Evaluation(audited_sha=audited_sha, gate_results=results, details=tuple(details)) + + +def _git_head() -> str: + return subprocess.run( + ["git", "rev-parse", "HEAD"], + cwd=REPO_ROOT, + check=True, + capture_output=True, + text=True, + ).stdout.strip() + + +def _require_commit_sha(value: str) -> str: + sha = _full_sha(value, "audited SHA") + result = subprocess.run( + ["git", "cat-file", "-e", f"{sha}^{{commit}}"], + cwd=REPO_ROOT, + check=False, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + if result.returncode != 0: + raise MatrixError(f"audited SHA {sha} is not a commit in this repository") + return sha + + +def _require_external_evidence_path(path: Path) -> Path: + resolved = path.resolve() + try: + relative = resolved.relative_to(REPO_ROOT) + except ValueError: + return resolved + tracked = subprocess.run( + ["git", "ls-files", "--error-unmatch", "--", str(relative)], + cwd=REPO_ROOT, + check=False, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + if tracked.returncode == 0: + raise MatrixError("--evidence must reference an external or untracked JSON file") + return resolved + + +def _is_ancestor(older: str, newer: str) -> bool: + return subprocess.run( + ["git", "merge-base", "--is-ancestor", older, newer], + cwd=REPO_ROOT, + check=False, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ).returncode == 0 + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + mode = parser.add_mutually_exclusive_group(required=True) + mode.add_argument("--check", action="store_true", help="derive a verdict from external evidence") + mode.add_argument("--validate-only", action="store_true", help="validate the tracked requirements schema") + parser.add_argument("--matrix", type=Path, default=DEFAULT_MATRIX) + parser.add_argument("--sha", help="full Git SHA to audit; defaults to current HEAD") + parser.add_argument("--evidence", type=Path, help="external, untracked SHA-bound evidence JSON") + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv) + try: + matrix = validate_matrix(json.loads(args.matrix.read_text(encoding="utf-8"))) + if args.validate_only: + if args.sha or args.evidence: + raise MatrixError("--validate-only accepts the tracked matrix only") + print(f"Pulse Intelligence gate matrix schema valid: {len(matrix['gates'])} requirement-only gates") + return 0 + audited_sha = _require_commit_sha(args.sha or _git_head()) + evidence = None + if args.evidence is not None: + evidence_path = _require_external_evidence_path(args.evidence) + evidence = validate_evidence(json.loads(evidence_path.read_text(encoding="utf-8")), matrix) + evaluation = evaluate_matrix(matrix, audited_sha, evidence, is_ancestor=_is_ancestor) + except (OSError, json.JSONDecodeError, MatrixError, subprocess.CalledProcessError) as exc: + print(f"BLOCKED: {exc}") + return 1 + + print(f"Pulse Intelligence audited SHA: {evaluation.audited_sha}") + for gate_id in GATE_IDS: + print(f"{gate_id}: {evaluation.gate_results[gate_id]}") + for detail in evaluation.details: + print(f"DETAIL: {detail}") + print(f"Pulse Intelligence release verdict: {evaluation.verdict}") + return 0 if evaluation.verdict == "GO" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/release_control/pulse_intelligence_gate_test.py b/scripts/release_control/pulse_intelligence_gate_test.py new file mode 100644 index 000000000..fec236bac --- /dev/null +++ b/scripts/release_control/pulse_intelligence_gate_test.py @@ -0,0 +1,191 @@ +#!/usr/bin/env python3 +"""Deterministic tests for the Pulse Intelligence aggregate release gate.""" + +from __future__ import annotations + +import json +import re +import unittest +from pathlib import Path + +import pulse_intelligence_gate as gate + + +SHA = "1" * 40 +NEXT_SHA = "2" * 40 +ANCESTOR_SHA = "3" * 40 + + +def valid_matrix() -> dict[str, object]: + gates: list[dict[str, object]] = [] + for gate_id in gate.GATE_IDS: + gates.append( + { + "id": gate_id, + "title": f"Test gate {gate_id}", + "owner": gate.APPROVED_OWNERS[gate_id], + "required_evidence_tier": "integration", + "requires_triple_zero": True, + "allows_ancestor_provenance": gate_id == "RG-08", + "non_mutating_proof_commands": [ + {"id": f"{gate_id.lower()}-proof", "argv": ["go", "test", "./internal/example"]} + ], + "mutation_gated_commands": [], + } + ) + return { + "schema_version": 2, + "matrix_id": "pulse-intelligence-rg-01-rg-12", + "program": "pulse-intelligence", + "result_semantics": { + "PASS": "External evidence satisfies the gate at the audited Git SHA.", + "FAIL": "External evidence bound to the audited Git SHA records a failed proof or an invalid binding.", + "NOT_RUN": "No qualifying external evidence exists for the gate at the audited Git SHA.", + }, + "evidence_tiers": list(gate.EVIDENCE_TIERS), + "gates": gates, + } + + +def evidence_for(matrix: dict[str, object], audited_sha: str) -> dict[str, object]: + records: list[dict[str, object]] = [] + kind_for_tier = { + "unit": "source", + "integration": "automated-test", + "browser": "browser-artifact", + "real-lab": "real-lab-artifact", + "physical-device": "physical-device-artifact", + "live-relay": "live-relay-artifact", + } + for item in matrix["gates"]: + record: dict[str, object] = { + "id": f"{item['id'].lower()}-evidence", + "gate_id": item["id"], + "result": "PASS", + "tier": item["required_evidence_tier"], + "kind": kind_for_tier[item["required_evidence_tier"]], + "git_sha": audited_sha, + "command_id": item["non_mutating_proof_commands"][0]["id"], + } + if item["requires_triple_zero"]: + record["triple_zero"] = { + "unauthorized_mutations": 0, + "transport_dispatches": 0, + "authority_writes": 0, + } + records.append(record) + return { + "schema_version": 1, + "matrix_id": matrix["matrix_id"], + "audited_git_sha": audited_sha, + "records": records, + } + + +class PulseIntelligenceGateTest(unittest.TestCase): + def test_malformed_matrix_is_rejected(self) -> None: + payload = valid_matrix() + del payload["result_semantics"] + with self.assertRaisesRegex(gate.MatrixError, "missing fields: result_semantics"): + gate.validate_matrix(payload) + + def test_duplicate_gate_is_rejected(self) -> None: + payload = valid_matrix() + payload["gates"][1]["id"] = "RG-01" + payload["gates"][1]["owner"] = "task-01" + with self.assertRaisesRegex(gate.MatrixError, "duplicate gate ids"): + gate.validate_matrix(payload) + + def test_wrong_owner_is_rejected(self) -> None: + payload = valid_matrix() + payload["gates"][4]["owner"] = "task-12" + with self.assertRaisesRegex(gate.MatrixError, "RG-05 owner must be 'task-05'"): + gate.validate_matrix(payload) + + def test_over_tier_evidence_is_rejected(self) -> None: + matrix = gate.validate_matrix(valid_matrix()) + payload = evidence_for(matrix, SHA) + payload["records"][0]["kind"] = "mock" + payload["records"][0]["tier"] = "real-lab" + with self.assertRaisesRegex(gate.MatrixError, "overstates 'mock' as 'real-lab'"): + gate.validate_evidence(payload, matrix) + + def test_external_exact_sha_evidence_can_certify_go(self) -> None: + matrix = gate.validate_matrix(valid_matrix()) + evidence = gate.validate_evidence(evidence_for(matrix, SHA), matrix) + evaluation = gate.evaluate_matrix(matrix, SHA, evidence) + self.assertEqual(evaluation.verdict, "GO") + self.assertEqual(set(evaluation.gate_results.values()), {"PASS"}) + + def test_committing_matrix_does_not_inherently_stale_it(self) -> None: + matrix = gate.validate_matrix(valid_matrix()) + self.assertNotIn("evaluation_git_sha", matrix) + for audited_sha in (SHA, NEXT_SHA): + with self.subTest(audited_sha=audited_sha): + evidence = gate.validate_evidence(evidence_for(matrix, audited_sha), matrix) + self.assertEqual(gate.evaluate_matrix(matrix, audited_sha, evidence).verdict, "GO") + + def test_ancestor_evidence_is_visible_but_non_qualifying(self) -> None: + matrix = gate.validate_matrix(valid_matrix()) + payload = evidence_for(matrix, SHA) + payload["records"][7]["git_sha"] = ANCESTOR_SHA + evidence = gate.validate_evidence(payload, matrix) + evaluation = gate.evaluate_matrix( + matrix, + SHA, + evidence, + is_ancestor=lambda older, newer: (older, newer) == (ANCESTOR_SHA, SHA), + ) + self.assertEqual(evaluation.gate_results["RG-08"], "NOT_RUN") + self.assertEqual(evaluation.verdict, "NO-GO") + self.assertTrue(any("RG-08 provenance" in detail and "does not certify" in detail for detail in evaluation.details)) + + def test_wrong_sha_external_evidence_fails(self) -> None: + matrix = gate.validate_matrix(valid_matrix()) + payload = evidence_for(matrix, SHA) + payload["records"][0]["git_sha"] = NEXT_SHA + evidence = gate.validate_evidence(payload, matrix) + evaluation = gate.evaluate_matrix(matrix, SHA, evidence, is_ancestor=lambda _older, _newer: False) + self.assertEqual(evaluation.gate_results["RG-01"], "FAIL") + self.assertTrue(any("bound to wrong SHA" in detail for detail in evaluation.details)) + + def test_wrong_external_document_binding_fails(self) -> None: + matrix = gate.validate_matrix(valid_matrix()) + evidence = gate.validate_evidence(evidence_for(matrix, NEXT_SHA), matrix) + evaluation = gate.evaluate_matrix(matrix, SHA, evidence) + self.assertEqual(set(evaluation.gate_results.values()), {"FAIL"}) + self.assertIn("not audited SHA", evaluation.details[0]) + + def test_missing_external_evidence_is_honest_not_run(self) -> None: + matrix = gate.validate_matrix(valid_matrix()) + evaluation = gate.evaluate_matrix(matrix, SHA, None) + self.assertEqual(evaluation.verdict, "NO-GO") + self.assertEqual(set(evaluation.gate_results.values()), {"NOT_RUN"}) + self.assertEqual(evaluation.details, ()) + + def test_missing_triple_zero_fails_applicable_gate(self) -> None: + matrix = gate.validate_matrix(valid_matrix()) + payload = evidence_for(matrix, SHA) + del payload["records"][0]["triple_zero"] + evidence = gate.validate_evidence(payload, matrix) + evaluation = gate.evaluate_matrix(matrix, SHA, evidence) + self.assertEqual(evaluation.gate_results["RG-01"], "FAIL") + self.assertTrue(any("missing required triple-zero" in detail for detail in evaluation.details)) + + def test_repository_matrix_schema_is_deterministic_and_requirement_only(self) -> None: + matrix_path = Path(__file__).resolve().parents[2] / "docs/release-control/v6/internal/pulse-intelligence-release-gate.json" + matrix_text = matrix_path.read_text(encoding="utf-8") + raw = json.loads(matrix_text) + first = gate.validate_matrix(raw) + second = gate.validate_matrix(json.loads(json.dumps(raw, sort_keys=True))) + self.assertEqual(first, second) + self.assertNotIn("evaluation_git_sha", first) + self.assertNotIn("overall_result", first) + self.assertIsNone(re.search(r"\b[0-9a-f]{40}\b", matrix_text)) + for item in first["gates"]: + self.assertNotIn("result", item) + self.assertNotIn("evidence", item) + + +if __name__ == "__main__": + unittest.main()