From 8bd0f1b94b29a5f4ca60a5066b971020a3e8d744 Mon Sep 17 00:00:00 2001 From: pcvantol Date: Mon, 7 Sep 2026 13:11:55 +0200 Subject: [PATCH] feat: enforce bounded programme authorization --- .../FORGE_V1_BOOTSTRAP_GOVERNANCE_DECISION.md | 25 ++- ...RGE_V1_BOOTSTRAP_ORCHESTRATOR_READINESS.md | 8 +- forge/governance_authority.py | 3 + forge/operator_identity.py | 6 +- forge/programme_authorization.py | 186 ++++++++++++++++++ tests/test_governance_authority.py | 2 +- tests/test_programme_authorization.py | 57 ++++++ 7 files changed, 270 insertions(+), 17 deletions(-) create mode 100644 forge/programme_authorization.py create mode 100644 tests/test_programme_authorization.py diff --git a/docs/architecture/FORGE_V1_BOOTSTRAP_GOVERNANCE_DECISION.md b/docs/architecture/FORGE_V1_BOOTSTRAP_GOVERNANCE_DECISION.md index 3af3bfd..63e6921 100644 --- a/docs/architecture/FORGE_V1_BOOTSTRAP_GOVERNANCE_DECISION.md +++ b/docs/architecture/FORGE_V1_BOOTSTRAP_GOVERNANCE_DECISION.md @@ -32,6 +32,11 @@ authority inputs; derived state/evidence refresh alone is not material. `BOOTSTRAP_AUTHORIZATION_STALENESS = DEFINED` `BOOTSTRAP_NODE_DISPATCH_AUTHORITY = MACHINE_DECIDABLE` +The record is written only through Forge's canonical governance writer with +capability `OWNER_PROGRAMME_AUTHORIZATION`. Its source reference and verified +owner-account binding are immutable evidence; a local status write, a copied +prompt, or a retrospective assertion is not an authorization record. + ## Owner authorization Forge has no inherited EP Owner Authorization workflow. This decision creates @@ -49,7 +54,8 @@ New commits invalidate it. It is distinct from CI and Human UI Review. ## Merge and operating boundary -Bootstrap V1 auto-merge is disabled. A node reaches `MERGE_READY` only after +Within a valid programme authorization, a node may be squash-merged only after +an executable exact-head qualification has passed. A node reaches `MERGE_READY` only after implementation, local and exact-head hosted qualification, resolved reviews, applicable UI/owner/security gates, valid authority, fresh contracts and mergeability. It then enters `WAITING_HUMAN_MERGE`; human merge is followed by @@ -57,18 +63,19 @@ post-merge qualification before `DONE`. A bounded merge packet includes node, PR, exact head, risk, DoR/DoD, CI/reviews/gates, scopes, unlocked dependents and known risks. Parallel PRs re-evaluate after every merge. -`BOOTSTRAP_AUTO_MERGE = FALSE` -`HUMAN_MERGE_REQUIRED = TRUE` +`BOOTSTRAP_AUTO_MERGE = QUALIFIED_SQUASH_ONLY` +`QUALIFIED_SQUASH_MERGE_REQUIRED = TRUE` `MERGE_DECISION_PACKET = DEFINED` `PARALLEL_PR_MERGE_REEVALUATION = TRUE` -Autonomous repair remains disabled: a future runner may classify and stop at -`REPAIR_REQUIRED`, never reinvoke itself without a separate finite policy. -The first operating mode is autonomous preparation/observation only; programme -approval, out-of-envelope decisions, security/UI/owner gates, repair authority -and merge stay human. +Autonomous repair is permitted only when its programme authorization states a +finite per-PR/per-head budget. The current policy supports at most three +attempts for an exact head; a new head requires a fresh exact-head +qualification and its own budget. The runner must stop at scope expansion, +expired or revoked authorization, failed security/review/CI, or exhausted +budget. -`AUTONOMOUS_REPAIR_ENABLED = FALSE` +`AUTONOMOUS_REPAIR_ENABLED = BOUNDED_PER_EXACT_HEAD` `UNBOUNDED_AUTONOMOUS_REPAIR = FALSE` `BOOTSTRAP_RUNNER_CAN_SELF_AUTHORIZE = FALSE` `BOOTSTRAP_GOVERNANCE_FORWARD_COMPATIBLE = TRUE` diff --git a/docs/architecture/FORGE_V1_BOOTSTRAP_ORCHESTRATOR_READINESS.md b/docs/architecture/FORGE_V1_BOOTSTRAP_ORCHESTRATOR_READINESS.md index 5bd8cb5..163968c 100644 --- a/docs/architecture/FORGE_V1_BOOTSTRAP_ORCHESTRATOR_READINESS.md +++ b/docs/architecture/FORGE_V1_BOOTSTRAP_ORCHESTRATOR_READINESS.md @@ -27,7 +27,7 @@ DISPATCHABLE = SEMANTIC_READY AND DEPENDENCY_SAFE AND REPOSITORY_SAFE Temporary local scheduling for this Forge bootstrap repository is `BOOTSTRAP_COORDINATION`, never an execution lease; EP remains the repository-write lease owner. `BOOTSTRAP_RUNNER_IS_LEASE_AUTHORITY = FALSE`; `BOOTSTRAP_PROMPT_IS_AUTHORITY = FALSE`; `CI_GREEN_UNLOCKS_DEPENDENTS = FALSE`; `POST_MERGE_DAG_REEVALUATION = TRUE`; `HUMAN_GATE_GLOBAL_STOP_BY_DEFAULT = FALSE`. -The canonical [V1 Bootstrap Governance Decision](FORGE_V1_BOOTSTRAP_GOVERNANCE_DECISION.md) selects a bounded hybrid programme authorization. `BOOTSTRAP_MISSION_AUTHORITY = BOUNDED_V1_BOOTSTRAP_PROGRAMME_AUTHORIZATION (MODEL_C)`: an explicit human authorization record admits only its immutable node set and authority envelope. It never omits the permanent Roadmap → Candidate → Business approval → Architecture approval → Mission → Action model; a new product, architecture, security, cross-product, repository, scope, or undefined-contract finding returns the node to full governance. `BOOTSTRAP_AUTO_MERGE_POLICY = DISABLED`; a human merge remains required. +The canonical [V1 Bootstrap Governance Decision](FORGE_V1_BOOTSTRAP_GOVERNANCE_DECISION.md) selects a bounded hybrid programme authorization. `BOOTSTRAP_MISSION_AUTHORITY = BOUNDED_V1_BOOTSTRAP_PROGRAMME_AUTHORIZATION (MODEL_C)`: an explicit human authorization record admits only its immutable node set and authority envelope. It never omits the permanent Roadmap → Candidate → Business approval → Architecture approval → Mission → Action model; a new product, architecture, security, cross-product, repository, scope, or undefined-contract finding returns the node to full governance. `BOOTSTRAP_AUTO_MERGE_POLICY = QUALIFIED_SQUASH_ONLY`: an executor may merge only a currently qualified exact head after all actual CI, review, security and Owner Authorization gates pass. ## Scheduler-grade coverage @@ -64,9 +64,9 @@ NOT_READY -> READY -> BOOTSTRAP_CLAIMED -> IMPLEMENTING -> PR_OPEN -> CI_RUNNING `BLOCKED`, `FAILED`, `STALE`, `SUPERSEDED`, and `REPLAN_REQUIRED` require explicit resolution. `BOOTSTRAP_NODE_LIFECYCLE = DEFINED`. `CODE_IMPLEMENTED`, `PR_OPEN`, `CI_GREEN`, `MERGE_READY`, `MERGED`, and `DONE` differ; only `DONE` satisfies a predecessor. -Every qualification names `LOCAL_TEST`, `HOSTED_CI`, `INSTALLED_PRODUCT_CANARY`, `SECURITY_GATE`, `BROWSER_GATE`, `CROSS_PRODUCT_GATE`, `HUMAN_REVIEW`, `OWNER_AUTHORIZATION`, or `POST_MERGE_GATE`, exact SHA, result and new-head invalidation. +Every qualification names `LOCAL_TEST`, `HOSTED_CI`, `INSTALLED_PRODUCT_CANARY`, `SECURITY_GATE`, `BROWSER_GATE`, `CROSS_PRODUCT_GATE`, `HUMAN_REVIEW`, `OWNER_AUTHORIZATION`, or `POST_MERGE_GATE`, exact SHA, result and new-head invalidation. The durable programme record is written by the canonical governance writer and records the verified owner account binding and source reference; it never synthesizes a GitHub check. -Ordinary review is same-node `REPAIR_REQUIRED`; an architecture decision, dependency, new capability or scope expansion is `REPLAN_REQUIRED`. CI failures must first classify as `IMPLEMENTATION_DEFECT`, `TEST_DEFECT`, `FLAKY_INFRASTRUCTURE`, `STALE_BASE`, `DEPENDENCY_CHANGED`, `SECURITY_FAILURE`, `QUALIFICATION_GAP`, or `UNDEFINED_CONTRACT`. No canonical retry limit exists: automatic repair requires a future operator-set finite limit; otherwise it is disallowed. `UNBOUNDED_AUTONOMOUS_REPAIR = FALSE`. +Ordinary review is same-node `REPAIR_REQUIRED`; an architecture decision, dependency, new capability or scope expansion is `REPLAN_REQUIRED`. CI failures must first classify as `IMPLEMENTATION_DEFECT`, `TEST_DEFECT`, `FLAKY_INFRASTRUCTURE`, `STALE_BASE`, `DEPENDENCY_CHANGED`, `SECURITY_FAILURE`, `QUALIFICATION_GAP`, or `UNDEFINED_CONTRACT`. Automatic repair is limited to three attempts per exact PR head under the programme authorization; a changed head must be requalified. `UNBOUNDED_AUTONOMOUS_REPAIR = FALSE`. Human UI review, owner authorization, business approval, architecture approval and security decision pause only their node; each needs named evidence and returns to re-evaluation. Other safe nodes continue. The canonical governance decision supplies Forge's consequence-based risk mapping: `NORMAL_LOW` needs no separate Owner Authorization, `ELEVATED` needs exact-head Owner Authorization, and `HIGH` needs exact-head Owner Authorization plus security review. A new commit invalidates the authorization. This is a Forge contract, not an inference from EP. @@ -92,6 +92,6 @@ After a human merge: refresh main, verify merge SHA, clean context, refresh Repo ## Governance closure -The governance decision resolves the three prior blockers: `BOOTSTRAP_MISSION_AUTHORITY = RESOLVED`, `OWNER_AUTHORIZATION_SCHEDULER_CONTRACT = RESOLVED`, and `BOOTSTRAP_AUTO_MERGE_POLICY = DISABLED`. The future runner is therefore contract-ready for a separately approved implementation Mission; dispatch still requires an actual, non-stale programme authorization record and cannot use repair automation. +The governance decision resolves the three prior blockers: `BOOTSTRAP_MISSION_AUTHORITY = RESOLVED`, `OWNER_AUTHORIZATION_SCHEDULER_CONTRACT = RESOLVED`, and `BOOTSTRAP_AUTO_MERGE_POLICY = QUALIFIED_SQUASH_ONLY`. The future runner is therefore contract-ready for a separately approved implementation Mission; dispatch still requires an actual, non-stale programme authorization record and applies the finite repair budget per exact head. `FORGE_V1_BOOTSTRAP_ORCHESTRATOR_READY_FOR_IMPLEMENTATION = YES`. diff --git a/forge/governance_authority.py b/forge/governance_authority.py index 2958c71..62d3352 100644 --- a/forge/governance_authority.py +++ b/forge/governance_authority.py @@ -18,6 +18,7 @@ class GovernanceCapability(str, Enum): BUSINESS_APPROVAL = "BUSINESS_APPROVAL" ARCHITECTURE_APPROVAL = "ARCHITECTURE_APPROVAL" SECURITY_APPROVAL = "SECURITY_APPROVAL" + OWNER_PROGRAMME_AUTHORIZATION = "OWNER_PROGRAMME_AUTHORIZATION" @dataclass(frozen=True) @@ -30,12 +31,14 @@ class GovernanceDecision: scope: tuple[str, ...] gates: tuple[str, ...] predecessor_digest: str | None = None + evidence: dict[str, object] | None = None def document(self, installation_id: str, operator_id: str, occurred_at: str) -> dict[str, object]: return { **asdict(self), "capability": self.capability.value, "scope": list(sorted(self.scope)), "gates": list(sorted(self.gates)), "installation_id": installation_id, "operator_id": operator_id, "occurred_at": occurred_at, + "evidence": self.evidence or {}, } diff --git a/forge/operator_identity.py b/forge/operator_identity.py index f263636..300eace 100644 --- a/forge/operator_identity.py +++ b/forge/operator_identity.py @@ -41,9 +41,9 @@ def _governance_state(self, context): rows=self.db._connection.execute('SELECT capability FROM governance_authority WHERE installation_id=? AND operator_id=? ORDER BY capability',(context.installation_id,operator)).fetchall() return tuple(row['capability'] for row in rows) def _valid_adoption_provenance(self, context, binding): - expected=('ARCHITECTURE_APPROVAL','BUSINESS_APPROVAL','SECURITY_APPROVAL'); operator=self._governance_operator_id(context) + expected=('ARCHITECTURE_APPROVAL','BUSINESS_APPROVAL','OWNER_PROGRAMME_AUTHORIZATION','SECURITY_APPROVAL'); operator=self._governance_operator_id(context) rows=self.db._connection.execute('SELECT capability,bootstrap_provenance,digest FROM governance_capability_grants WHERE installation_id=? AND operator_id=? ORDER BY capability',(context.installation_id,operator)).fetchall() - if len(rows)!=3 or tuple(row['capability'] for row in rows)!=expected:return False + if len(rows)!=len(expected) or tuple(row['capability'] for row in rows)!=expected:return False for row in rows: try:document=json.loads(row['bootstrap_provenance']) except (TypeError,ValueError):return False @@ -52,7 +52,7 @@ def _valid_adoption_provenance(self, context, binding): return True def _persist_governance_capabilities(self, context, kind, provenance, binding=None): if not self.authorize(context): raise PermissionError('trusted bound operator required') - expected=('ARCHITECTURE_APPROVAL','BUSINESS_APPROVAL','SECURITY_APPROVAL'); state=self._governance_state(context) + expected=('ARCHITECTURE_APPROVAL','BUSINESS_APPROVAL','OWNER_PROGRAMME_AUTHORIZATION','SECURITY_APPROVAL'); state=self._governance_state(context) if state: if state!=expected or kind!='EXISTING_G001_GOVERNANCE_ADOPTION_V1' or binding is None or not self._valid_adoption_provenance(context,binding): raise PermissionError('conflicting governance capability provenance') return diff --git a/forge/programme_authorization.py b/forge/programme_authorization.py new file mode 100644 index 0000000..ce92670 --- /dev/null +++ b/forge/programme_authorization.py @@ -0,0 +1,186 @@ +"""Bounded, programme-scoped owner authorization. + +This module deliberately records authority through ``CanonicalGovernanceRepository``. +It does not write GitHub statuses, replace reviews, or turn a chat instruction into +qualification evidence. Each candidate head is qualified independently. +""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass +from datetime import datetime, timezone +import re +from typing import Iterable + +from forge.governance_authority import ( + CanonicalGovernanceRepository, + GovernanceCapability, + GovernanceDecision, +) +from forge.operator_identity import OperatorContext + +_SHA = re.compile(r"^[0-9a-f]{40}$") +_PROGRAMME_KIND = "OWNER_PROGRAMME_AUTHORIZATION_V1" + + +@dataclass(frozen=True) +class ProgrammeAuthorization: + """Immutable owner grant with deliberately narrow execution boundaries.""" + + authorization_id: str + programme_id: str + owner_login: str + owner_account_binding: str + source_reference: str + repositories: tuple[str, ...] + allowed_scopes: tuple[str, ...] + expires_at: str + repair_attempt_limit: int = 3 + merge_method: str = "squash" + + def __post_init__(self) -> None: + if not all((self.authorization_id, self.programme_id, self.owner_login, + self.owner_account_binding, self.source_reference, + self.repositories, self.allowed_scopes, self.expires_at)): + raise ValueError("programme authorization requires complete provenance and boundaries") + if self.repair_attempt_limit < 0 or self.merge_method != "squash": + raise ValueError("only bounded repair and qualified squash merge are supported") + _parse_time(self.expires_at) + + def document(self) -> dict[str, object]: + value = asdict(self) + value["repositories"] = sorted(self.repositories) + value["allowed_scopes"] = sorted(self.allowed_scopes) + value["kind"] = _PROGRAMME_KIND + return value + + +@dataclass(frozen=True) +class CandidateQualification: + repository: str + pull_request: int + head_sha: str + base_branch: str + changed_scopes: tuple[str, ...] + technical_qualification_passed: bool + ci_passed: bool + reviews_passed: bool + security_passed: bool + owner_workflow_evidence: str + owner_workflow_head_sha: str + merge_method: str = "squash" + + def __post_init__(self) -> None: + if (self.pull_request <= 0 or not _SHA.fullmatch(self.head_sha) + or not self.base_branch or not self.owner_workflow_evidence + or self.owner_workflow_head_sha != self.head_sha): + raise ValueError("candidate requires an exact head, base branch, and workflow evidence") + + +def _parse_time(value: str) -> datetime: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + if parsed.tzinfo is None: + raise ValueError("timestamps must be timezone aware") + return parsed.astimezone(timezone.utc) + + +class ProgrammeAuthorizationGate: + """Fail-closed policy gate for the approved autonomy programme.""" + + def __init__(self, repository: CanonicalGovernanceRepository, context: OperatorContext, + now=lambda: datetime.now(timezone.utc)) -> None: + self.repository, self.context, self.now = repository, context, now + + def record_authorization(self, authorization: ProgrammeAuthorization) -> str: + return self.repository.record(GovernanceDecision( + decision_id=authorization.authorization_id, + subject_id=authorization.programme_id, + subject_revision=authorization.authorization_id, + capability=GovernanceCapability.OWNER_PROGRAMME_AUTHORIZATION, + decision="authorized", + scope=authorization.allowed_scopes, + gates=("exact-head-qualification", "ci", "reviews", "security", "owner-workflow"), + evidence=authorization.document(), + ), self.context) + + def qualify(self, authorization_id: str, candidate: CandidateQualification) -> str: + authorization = self._authorization(authorization_id) + self._validate_candidate(authorization, candidate) + return self.repository.record(GovernanceDecision( + decision_id=f"{authorization_id}:pr-{candidate.pull_request}:{candidate.head_sha}", + subject_id=f"{authorization.programme_id}:pr-{candidate.pull_request}", + subject_revision=candidate.head_sha, + capability=GovernanceCapability.OWNER_PROGRAMME_AUTHORIZATION, + decision="qualified", + scope=candidate.changed_scopes, + gates=("exact-head-qualification", "ci", "reviews", "security", "owner-workflow", "squash"), + predecessor_digest=self._decision_digest(authorization_id), + evidence={"kind": "EXACT_HEAD_QUALIFICATION_V1", **asdict(candidate), + "authorization_id": authorization_id}, + ), self.context) + + def record_repair_attempt(self, authorization_id: str, candidate: CandidateQualification) -> str: + authorization = self._authorization(authorization_id) + self._validate_candidate(authorization, candidate, require_passes=False) + existing = self._repair_attempts(authorization_id, candidate.pull_request, candidate.head_sha) + if existing >= authorization.repair_attempt_limit: + raise PermissionError("bounded repair budget exhausted for this exact PR head") + attempt = existing + 1 + return self.repository.record(GovernanceDecision( + decision_id=f"{authorization_id}:repair:{candidate.pull_request}:{candidate.head_sha}:{attempt}", + subject_id=f"{authorization.programme_id}:repair:pr-{candidate.pull_request}:attempt-{attempt}", + subject_revision=candidate.head_sha, + capability=GovernanceCapability.OWNER_PROGRAMME_AUTHORIZATION, + decision="repair-authorized", + scope=candidate.changed_scopes, + gates=("same-approved-scope", "exact-head"), + predecessor_digest=self._decision_digest(authorization_id), + evidence={"kind": "BOUNDED_REPAIR_ATTEMPT_V1", "authorization_id": authorization_id, + "pull_request": candidate.pull_request, "head_sha": candidate.head_sha, + "attempt": attempt}, + ), self.context) + + def _authorization(self, authorization_id: str) -> ProgrammeAuthorization: + record = self.repository.decision(authorization_id) + if record.get("capability") != GovernanceCapability.OWNER_PROGRAMME_AUTHORIZATION.value or record.get("decision") != "authorized": + raise PermissionError("owner programme authorization is absent") + evidence = record.get("evidence") + if not isinstance(evidence, dict) or evidence.get("kind") != _PROGRAMME_KIND: + raise PermissionError("owner programme authorization provenance is invalid") + return ProgrammeAuthorization(**{key: evidence[key] for key in ProgrammeAuthorization.__dataclass_fields__}) + + def _validate_candidate(self, authorization: ProgrammeAuthorization, candidate: CandidateQualification, + *, require_passes: bool = True) -> None: + if self.now() > _parse_time(authorization.expires_at): + raise PermissionError("owner programme authorization has expired") + if candidate.repository not in authorization.repositories: + raise PermissionError("repository is outside owner-authorized programme scope") + if not set(candidate.changed_scopes).issubset(authorization.allowed_scopes): + raise PermissionError("candidate diff expands the authorized scope") + if candidate.merge_method != authorization.merge_method: + raise PermissionError("only qualified squash merge is allowed") + if require_passes and not all((candidate.technical_qualification_passed, candidate.ci_passed, + candidate.reviews_passed, candidate.security_passed)): + raise PermissionError("candidate has not passed all required qualification gates") + + def _decision_digest(self, decision_id: str) -> str: + row = self.repository.database._connection.execute( + "SELECT digest FROM governance_decisions WHERE decision_id = ?", (decision_id,) + ).fetchone() + if row is None: + raise PermissionError("authorization evidence is absent") + return row["digest"] + + def _repair_attempts(self, authorization_id: str, pull_request: int, head_sha: str) -> int: + rows = self.repository.database._connection.execute( + "SELECT document FROM governance_decisions WHERE capability = ?", + (GovernanceCapability.OWNER_PROGRAMME_AUTHORIZATION.value,), + ).fetchall() + import json + return sum( + 1 for row in rows + if (lambda evidence: evidence.get("kind") == "BOUNDED_REPAIR_ATTEMPT_V1" + and evidence.get("authorization_id") == authorization_id + and evidence.get("pull_request") == pull_request + and evidence.get("head_sha") == head_sha)(json.loads(row["document"]).get("evidence", {})) + ) diff --git a/tests/test_governance_authority.py b/tests/test_governance_authority.py index d976207..044096a 100644 --- a/tests/test_governance_authority.py +++ b/tests/test_governance_authority.py @@ -296,7 +296,7 @@ def test_schema19_pre_governance_fixture_migrates_and_adopts_existing_g001(self) capabilities = tuple(row["capability"] for row in database._connection.execute( "SELECT capability FROM governance_authority ORDER BY capability" )) - self.assertEqual(capabilities, ("ARCHITECTURE_APPROVAL", "BUSINESS_APPROVAL", "SECURITY_APPROVAL")) + self.assertEqual(capabilities, ("ARCHITECTURE_APPROVAL", "BUSINESS_APPROVAL", "OWNER_PROGRAMME_AUTHORIZATION", "SECURITY_APPROVAL")) repository = CanonicalGovernanceRepository._for_test(database, operators) planning = GovernanceAuthorityTests.planning("legacy-19") CanonicalBusinessWorkspace(repository, context).approve( diff --git a/tests/test_programme_authorization.py b/tests/test_programme_authorization.py new file mode 100644 index 0000000..870a3dd --- /dev/null +++ b/tests/test_programme_authorization.py @@ -0,0 +1,57 @@ +import tempfile +import unittest +from datetime import datetime, timezone +from pathlib import Path + +from forge.governance_authority import CanonicalGovernanceRepository +from forge.operator_identity import InstallationOperatorService, NamedOperatorIdentity +from forge.programme_authorization import CandidateQualification, ProgrammeAuthorization, ProgrammeAuthorizationGate +from forge.runtime.database import RuntimeDatabase + + +class ProgrammeAuthorizationGateTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory(); root = Path(self.temp.name) + self.db = RuntimeDatabase(root, path=root / "runtime.db") + self.ops = InstallationOperatorService(self.db, lambda: NamedOperatorIdentity("operator-a", 501)) + self.context = self.ops.first_bind() + self.repository = CanonicalGovernanceRepository._for_test(self.db, self.ops) + self.gate = ProgrammeAuthorizationGate(self.repository, self.context, + now=lambda: datetime(2026, 9, 7, tzinfo=timezone.utc)) + self.authorization = ProgrammeAuthorization( + "owner-programme-1", "AUTONOMY_BOOTSTRAP_CLEAN_EP_AND_FIRST_DELIVERY_LOOP", + "pcvantol", "github:pcvantol:19624891", "owner-task:verified-2026-09-07", + ("pcvantol/engineering-platform",), ("ep-installation", "governance"), + "2026-10-01T00:00:00Z") + + def tearDown(self): self.db.close(); self.temp.cleanup() + + def candidate(self, **changes): + value = dict(repository="pcvantol/engineering-platform", pull_request=77, + head_sha="a" * 40, base_branch="main", changed_scopes=("governance",), + technical_qualification_passed=True, ci_passed=True, reviews_passed=True, + security_passed=True, owner_workflow_evidence="github-run:123", + owner_workflow_head_sha="a" * 40) + value.update(changes) + if "head_sha" in changes and "owner_workflow_head_sha" not in changes: + value["owner_workflow_head_sha"] = changes["head_sha"] + return CandidateQualification(**value) + + def test_persists_owner_grant_and_exact_head_qualification_through_canonical_writer(self): + digest = self.gate.record_authorization(self.authorization) + qualified = self.gate.qualify(self.authorization.authorization_id, self.candidate()) + self.assertTrue(digest.startswith("sha256:")); self.assertTrue(qualified.startswith("sha256:")) + stored = self.repository.decision(self.authorization.authorization_id) + self.assertEqual(stored["evidence"]["source_reference"], "owner-task:verified-2026-09-07") + + def test_fails_closed_for_scope_expansion_or_missing_real_gates(self): + self.gate.record_authorization(self.authorization) + with self.assertRaises(PermissionError): self.gate.qualify(self.authorization.authorization_id, self.candidate(changed_scopes=("outside",))) + with self.assertRaises(PermissionError): self.gate.qualify(self.authorization.authorization_id, self.candidate(security_passed=False)) + + def test_repair_is_limited_per_exact_pr_head_and_new_head_requires_new_qualification(self): + self.gate.record_authorization(self.authorization) + candidate = self.candidate(technical_qualification_passed=False, ci_passed=False) + for _ in range(3): self.gate.record_repair_attempt(self.authorization.authorization_id, candidate) + with self.assertRaises(PermissionError): self.gate.record_repair_attempt(self.authorization.authorization_id, candidate) + self.assertEqual(self.gate.record_repair_attempt(self.authorization.authorization_id, self.candidate(head_sha="b" * 40, technical_qualification_passed=False, ci_passed=False)), self.gate._decision_digest(f"{self.authorization.authorization_id}:repair:77:{'b' * 40}:1"))