Normalize Pulse Intelligence release gates

This commit is contained in:
rcourtman
2026-07-12 14:18:45 +01:00
parent c07927adbd
commit 2d47c12e6d
7 changed files with 866 additions and 5 deletions
@@ -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
+4
View File
@@ -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
@@ -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
@@ -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=<released-final-sha-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", "<release-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", "<full-release-sha>", "--evidence", "<external-evidence.json>"]
}
],
"mutation_gated_commands": []
}
]
}
+28 -5
View File
@@ -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.",
@@ -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())
@@ -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()