Compare commits

...
Author SHA1 Message Date
AIOS Governed Continuation f7ab821186 feat(governed): protected-main status producer pipeline (OPTION A)
Establish a compliant status producer pipeline to unblock protected-main's
9 required status contexts (py-compile, git-diff-check, secret-scan,
autonomy-unit-tests, autonomy-smoke-tests, reconciliation-contract,
release-state-machine, campaign-approval-gate, signer-validator-contract).

- status_publisher: idempotent (effect_id), SUPPRESS_NOOP, RECONCILE,
  capability-minimal (status_publish only), append-only evidence store.
- status_orchestrator: exact-head-SHA binding, stale-SHA protection,
  merge eligibility gated on producer verdict + confirmed publish
  (existing_state_differs never credited green).
- status_evidence_contract: evidence schema, effect_id, N/A contract binding.
- producer_scope_classifier: campaign/signer applicability, fail-closed.
- status_producers/: 9 thin wrappers reusing existing validators (gitleaks,
  reconciliation, ReleaseStateMachine, campaign gate, signer contract).
- transport/broker: add publish_status_for_sha (exact SHA) + read_statuses.
- status_pipeline_entry: S6 trigger skeleton with optional scope/smoke
  injection (fail-closed when absent; no fabricated success).
- tests: 65 passing (engineer + QA adversarial + regression).

No forgery of success, no blanket green reporter, no weakening of
branch protection. Reuses existing Gitea machine broker + validators.
2026-09-15 15:12:15 +08:00
20 changed files with 2951 additions and 306 deletions
+136 -125
View File
@@ -1,125 +1,136 @@
"""Non-Owner Gitea machine capability and branch reconciliation boundary."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Mapping, Protocol
class MachineBrokerError(RuntimeError):
pass
class GiteaTransport(Protocol):
def compare_and_swap_branch(self, branch: str, expected_oid: str, new_oid: str) -> str: ...
def create_or_update_pr(self, branch: str, target: str, title: str, body: str) -> Mapping[str, Any]: ...
def publish_status(self, branch: str, state: str, description: str) -> Mapping[str, Any]: ...
def protected_merge(self, pr_number: int) -> Mapping[str, Any]: ...
def read_branch(self, branch: str) -> str | None: ...
def whoami(self) -> Mapping[str, Any]: ...
def repository(self, owner: str, name: str) -> Mapping[str, Any]: ...
@dataclass(frozen=True, slots=True)
class MachineCredentialProjection:
"""Opaque systemd credential reference; the credential value is never accepted."""
reference: str
account: str
owner_account: str = "aiman"
@classmethod
def from_mapping(cls, value: Mapping[str, Any]) -> "MachineCredentialProjection":
if set(value) != {"scheme", "name", "account"}:
raise MachineBrokerError("CREDENTIAL_REFERENCE_INVALID")
if value["scheme"] != "systemd-credential" or value["name"] != "gitea-machine-token":
raise MachineBrokerError("CREDENTIAL_REFERENCE_INVALID")
return cls(f"{value['scheme']}:{value['name']}", str(value["account"]))
def __post_init__(self) -> None:
if self.reference != "systemd-credential:gitea-machine-token":
raise MachineBrokerError("CREDENTIAL_REFERENCE_INVALID")
if not self.account or self.account == self.owner_account:
raise MachineBrokerError("OWNER_MACHINE_IDENTITY_REJECTED")
@dataclass(frozen=True, slots=True)
class CampaignCapabilityReceipt:
campaign_id: str
task_id: str
repository: str
branch: str
expected_oid: str
capabilities: frozenset[str]
denied: frozenset[str]
def require(self, capability: str) -> None:
if capability in self.denied or capability not in self.capabilities:
raise MachineBrokerError("CAPABILITY_DENIED")
if self.branch in {"main", "master"}:
raise MachineBrokerError("PROTECTED_BRANCH_DENIED")
class GiteaMachineBroker:
def __init__(self, transport: GiteaTransport, credential: MachineCredentialProjection, receipt: CampaignCapabilityReceipt) -> None:
self.transport = transport
self.credential = credential
self.receipt = receipt
def verify_identity(self) -> Mapping[str, Any]:
identity = dict(self.transport.whoami())
login = identity.get("login") or identity.get("username")
if not isinstance(login, str) or not login or login == self.credential.owner_account or login != self.credential.account:
raise MachineBrokerError("GITEA_MACHINE_IDENTITY_INVALID")
owner, name = self.receipt.repository.split("/", 1)
repo = dict(self.transport.repository(owner, name))
if repo.get("full_name") != self.receipt.repository:
raise MachineBrokerError("GITEA_REPOSITORY_MISMATCH")
return {"login": login, "repository": self.receipt.repository, "owner_account": self.credential.owner_account}
def push_branch(self, new_oid: str) -> str:
self.receipt.require("exact_task_branch_push")
return self.transport.compare_and_swap_branch(self.receipt.branch, self.receipt.expected_oid, new_oid)
def create_pr(self, title: str, body: str) -> Mapping[str, Any]:
self.receipt.require("pr_create_update")
return self.transport.create_or_update_pr(self.receipt.branch, "main", title, body)
def status(self, state: str, description: str, context: str = "aios/durable-release") -> Mapping[str, Any]:
self.receipt.require("status_publish")
return self.transport.publish_status(self.receipt.branch, state, description, context)
def merge(self, pr_number: int) -> Mapping[str, Any]:
self.receipt.require("protected_merge")
return self.transport.protected_merge(pr_number)
def readback(self) -> str | None:
self.receipt.require("remote_readback")
return self.transport.read_branch(self.receipt.branch)
def reconcile_exact_branch(authority_oid: str, gitea_oid: str | None, expected_oid: str) -> str:
if authority_oid != expected_oid:
return "AUTHORITY_EXPECTATION_MISMATCH"
if gitea_oid is None:
return "CREATE_REQUIRED"
if gitea_oid == expected_oid:
return "ALREADY_IN_SYNC"
return "MANUAL_OWNER_REVIEW_REQUIRED"
def reconcile_fast_forward(
remote_oid: str | None,
expected_old_oid: str,
new_oid: str,
*,
old_is_ancestor_of_new: bool,
) -> str:
"""Return a bounded action for an active Job/lease branch scope."""
if remote_oid is None:
return "CREATE_REQUIRED"
if remote_oid == new_oid:
return "ALREADY_IN_SYNC"
if remote_oid == expected_old_oid and old_is_ancestor_of_new:
return "FAST_FORWARD_REQUIRED"
return "MANUAL_OWNER_REVIEW_REQUIRED"
"""Non-Owner Gitea machine capability and branch reconciliation boundary."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Mapping, Protocol
class MachineBrokerError(RuntimeError):
pass
class GiteaTransport(Protocol):
def compare_and_swap_branch(self, branch: str, expected_oid: str, new_oid: str) -> str: ...
def create_or_update_pr(self, branch: str, target: str, title: str, body: str) -> Mapping[str, Any]: ...
def publish_status(self, branch: str, state: str, description: str, context: str) -> Mapping[str, Any]: ...
def publish_status_for_sha(self, sha: str, state: str, description: str, context: str) -> Mapping[str, Any]: ...
def read_statuses(self, sha: str, context: str | None = None) -> list[Mapping[str, Any]]: ...
def protected_merge(self, pr_number: int) -> Mapping[str, Any]: ...
def read_branch(self, branch: str) -> str | None: ...
def whoami(self) -> Mapping[str, Any]: ...
def repository(self, owner: str, name: str) -> Mapping[str, Any]: ...
@dataclass(frozen=True, slots=True)
class MachineCredentialProjection:
"""Opaque systemd credential reference; the credential value is never accepted."""
reference: str
account: str
owner_account: str = "aiman"
@classmethod
def from_mapping(cls, value: Mapping[str, Any]) -> "MachineCredentialProjection":
if set(value) != {"scheme", "name", "account"}:
raise MachineBrokerError("CREDENTIAL_REFERENCE_INVALID")
if value["scheme"] != "systemd-credential" or value["name"] != "gitea-machine-token":
raise MachineBrokerError("CREDENTIAL_REFERENCE_INVALID")
return cls(f"{value['scheme']}:{value['name']}", str(value["account"]))
def __post_init__(self) -> None:
if self.reference != "systemd-credential:gitea-machine-token":
raise MachineBrokerError("CREDENTIAL_REFERENCE_INVALID")
if not self.account or self.account == self.owner_account:
raise MachineBrokerError("OWNER_MACHINE_IDENTITY_REJECTED")
@dataclass(frozen=True, slots=True)
class CampaignCapabilityReceipt:
campaign_id: str
task_id: str
repository: str
branch: str
expected_oid: str
capabilities: frozenset[str]
denied: frozenset[str]
def require(self, capability: str) -> None:
if capability in self.denied or capability not in self.capabilities:
raise MachineBrokerError("CAPABILITY_DENIED")
if self.branch in {"main", "master"}:
raise MachineBrokerError("PROTECTED_BRANCH_DENIED")
class GiteaMachineBroker:
def __init__(self, transport: GiteaTransport, credential: MachineCredentialProjection, receipt: CampaignCapabilityReceipt) -> None:
self.transport = transport
self.credential = credential
self.receipt = receipt
def verify_identity(self) -> Mapping[str, Any]:
identity = dict(self.transport.whoami())
login = identity.get("login") or identity.get("username")
if not isinstance(login, str) or not login or login == self.credential.owner_account or login != self.credential.account:
raise MachineBrokerError("GITEA_MACHINE_IDENTITY_INVALID")
owner, name = self.receipt.repository.split("/", 1)
repo = dict(self.transport.repository(owner, name))
if repo.get("full_name") != self.receipt.repository:
raise MachineBrokerError("GITEA_REPOSITORY_MISMATCH")
return {"login": login, "repository": self.receipt.repository, "owner_account": self.credential.owner_account}
def push_branch(self, new_oid: str) -> str:
self.receipt.require("exact_task_branch_push")
return self.transport.compare_and_swap_branch(self.receipt.branch, self.receipt.expected_oid, new_oid)
def create_pr(self, title: str, body: str) -> Mapping[str, Any]:
self.receipt.require("pr_create_update")
return self.transport.create_or_update_pr(self.receipt.branch, "main", title, body)
def status(self, state: str, description: str, context: str = "aios/durable-release") -> Mapping[str, Any]:
self.receipt.require("status_publish")
return self.transport.publish_status(self.receipt.branch, state, description, context)
def status_for_sha(self, sha: str, state: str, description: str, context: str = "aios/durable-release") -> Mapping[str, Any]:
"""Publish a status against an explicit commit SHA (protected-PR exact binding)."""
self.receipt.require("status_publish")
return self.transport.publish_status_for_sha(sha, state, description, context)
def read_statuses(self, sha: str, context: str | None = None) -> list[Mapping[str, Any]]:
self.receipt.require("status_publish")
return self.transport.read_statuses(sha, context=context)
def merge(self, pr_number: int) -> Mapping[str, Any]:
self.receipt.require("protected_merge")
return self.transport.protected_merge(pr_number)
def readback(self) -> str | None:
self.receipt.require("remote_readback")
return self.transport.read_branch(self.receipt.branch)
def reconcile_exact_branch(authority_oid: str, gitea_oid: str | None, expected_oid: str) -> str:
if authority_oid != expected_oid:
return "AUTHORITY_EXPECTATION_MISMATCH"
if gitea_oid is None:
return "CREATE_REQUIRED"
if gitea_oid == expected_oid:
return "ALREADY_IN_SYNC"
return "MANUAL_OWNER_REVIEW_REQUIRED"
def reconcile_fast_forward(
remote_oid: str | None,
expected_old_oid: str,
new_oid: str,
*,
old_is_ancestor_of_new: bool,
) -> str:
"""Return a bounded action for an active Job/lease branch scope."""
if remote_oid is None:
return "CREATE_REQUIRED"
if remote_oid == new_oid:
return "ALREADY_IN_SYNC"
if remote_oid == expected_old_oid and old_is_ancestor_of_new:
return "FAST_FORWARD_REQUIRED"
return "MANUAL_OWNER_REVIEW_REQUIRED"
+203 -181
View File
@@ -1,29 +1,29 @@
"""Gitea HTTP/Git transport using only a projected systemd credential."""
from __future__ import annotations
from dataclasses import dataclass
import json
import os
from pathlib import Path
import subprocess
import tempfile
from typing import Any, Mapping
from urllib.error import HTTPError, URLError
from urllib.parse import quote, urlencode
from urllib.request import Request, urlopen
class GiteaTransportError(RuntimeError):
pass
def _safe_oid(value: str, field: str) -> str:
if not isinstance(value, str) or len(value) != 40 or any(ch not in "0123456789abcdef" for ch in value):
raise GiteaTransportError(f"{field}_INVALID")
return value
"""Gitea HTTP/Git transport using only a projected systemd credential."""
from __future__ import annotations
from dataclasses import dataclass
import json
import os
from pathlib import Path
import subprocess
import tempfile
from typing import Any, Mapping
from urllib.error import HTTPError, URLError
from urllib.parse import quote, urlencode
from urllib.request import Request, urlopen
class GiteaTransportError(RuntimeError):
pass
def _safe_oid(value: str, field: str) -> str:
if not isinstance(value, str) or len(value) != 40 or any(ch not in "0123456789abcdef" for ch in value):
raise GiteaTransportError(f"{field}_INVALID")
return value
def _safe_branch(value: str) -> str:
branch = str(value or "")
if (
@@ -43,119 +43,119 @@ def _safe_branch(value: str) -> str:
return branch
@dataclass(slots=True)
@dataclass(slots=True)
class GiteaMachineTransport:
api_base: str
repository_id: str
remote_url: str
authority_git_dir: Path
api_base: str
repository_id: str
remote_url: str
authority_git_dir: Path
credential_name: str = "gitea-machine-token"
actor: str = "aios-autonomy-agent"
timeout: int = 20
expected_merge_head: str | None = None
expected_merge_base: str | None = None
def __post_init__(self) -> None:
if self.credential_name != "gitea-machine-token":
raise GiteaTransportError("CREDENTIAL_NAME_INVALID")
if "/" not in self.repository_id or self.repository_id.startswith("/"):
raise GiteaTransportError("REPOSITORY_INVALID")
if not self.api_base.startswith("https://") and not self.api_base.startswith("http://127.0.0.1"):
raise GiteaTransportError("GITEA_API_SCOPE_INVALID")
if not self.remote_url.startswith("https://"):
raise GiteaTransportError("GITEA_REMOTE_SCOPE_INVALID")
@property
def _credential_path(self) -> Path:
directory = os.environ.get("CREDENTIALS_DIRECTORY", "")
if not directory:
raise GiteaTransportError("CREDENTIAL_PROJECTION_MISSING")
path = Path(directory) / self.credential_name
if path.is_symlink() or not path.is_file():
raise GiteaTransportError("CREDENTIAL_PROJECTION_MISSING")
return path
def _token(self) -> str:
value = self._credential_path.read_text(encoding="utf-8").strip()
if not value:
raise GiteaTransportError("CREDENTIAL_EMPTY")
return value
def _request(self, method: str, path: str, body: Mapping[str, Any] | None = None) -> tuple[int, Any]:
data = None if body is None else json.dumps(dict(body), separators=(",", ":")).encode()
headers = {"Accept": "application/json", "Authorization": f"token {self._token()}"}
if data is not None:
headers["Content-Type"] = "application/json"
request = Request(self.api_base.rstrip("/") + path, data=data, headers=headers, method=method)
try:
with urlopen(request, timeout=self.timeout) as response:
raw = response.read()
return response.status, json.loads(raw.decode()) if raw else {}
except HTTPError as exc:
# Never include response bodies: Gitea may echo sensitive request context.
raise GiteaTransportError(f"GITEA_HTTP_{exc.code}") from exc
except (URLError, TimeoutError, OSError) as exc:
raise GiteaTransportError("GITEA_NETWORK_ERROR") from exc
def whoami(self) -> Mapping[str, Any]:
return self._request("GET", "/user")[1]
def repository_info(self) -> Mapping[str, Any]:
owner, name = self.repository_id.split("/", 1)
return self._request("GET", f"/repos/{quote(owner)}/{quote(name)}")[1]
def repository(self, owner: str, name: str) -> Mapping[str, Any]:
if f"{owner}/{name}" != self.repository_id:
raise GiteaTransportError("REPOSITORY_SCOPE_DENIED")
def __post_init__(self) -> None:
if self.credential_name != "gitea-machine-token":
raise GiteaTransportError("CREDENTIAL_NAME_INVALID")
if "/" not in self.repository_id or self.repository_id.startswith("/"):
raise GiteaTransportError("REPOSITORY_INVALID")
if not self.api_base.startswith("https://") and not self.api_base.startswith("http://127.0.0.1"):
raise GiteaTransportError("GITEA_API_SCOPE_INVALID")
if not self.remote_url.startswith("https://"):
raise GiteaTransportError("GITEA_REMOTE_SCOPE_INVALID")
@property
def _credential_path(self) -> Path:
directory = os.environ.get("CREDENTIALS_DIRECTORY", "")
if not directory:
raise GiteaTransportError("CREDENTIAL_PROJECTION_MISSING")
path = Path(directory) / self.credential_name
if path.is_symlink() or not path.is_file():
raise GiteaTransportError("CREDENTIAL_PROJECTION_MISSING")
return path
def _token(self) -> str:
value = self._credential_path.read_text(encoding="utf-8").strip()
if not value:
raise GiteaTransportError("CREDENTIAL_EMPTY")
return value
def _request(self, method: str, path: str, body: Mapping[str, Any] | None = None) -> tuple[int, Any]:
data = None if body is None else json.dumps(dict(body), separators=(",", ":")).encode()
headers = {"Accept": "application/json", "Authorization": f"token {self._token()}"}
if data is not None:
headers["Content-Type"] = "application/json"
request = Request(self.api_base.rstrip("/") + path, data=data, headers=headers, method=method)
try:
with urlopen(request, timeout=self.timeout) as response:
raw = response.read()
return response.status, json.loads(raw.decode()) if raw else {}
except HTTPError as exc:
# Never include response bodies: Gitea may echo sensitive request context.
raise GiteaTransportError(f"GITEA_HTTP_{exc.code}") from exc
except (URLError, TimeoutError, OSError) as exc:
raise GiteaTransportError("GITEA_NETWORK_ERROR") from exc
def whoami(self) -> Mapping[str, Any]:
return self._request("GET", "/user")[1]
def repository_info(self) -> Mapping[str, Any]:
owner, name = self.repository_id.split("/", 1)
return self._request("GET", f"/repos/{quote(owner)}/{quote(name)}")[1]
def repository(self, owner: str, name: str) -> Mapping[str, Any]:
if f"{owner}/{name}" != self.repository_id:
raise GiteaTransportError("REPOSITORY_SCOPE_DENIED")
return self.repository_info()
def read_branch(self, branch: str) -> str | None:
def read_branch(self, branch: str) -> str | None:
branch = _safe_branch(branch)
owner, name = self.repository_id.split("/", 1)
try:
value = self._request("GET", f"/repos/{quote(owner)}/{quote(name)}/branches/{quote(branch, safe='/')}")[1]
except GiteaTransportError as exc:
if str(exc) == "GITEA_HTTP_404":
return None
raise
sha = value.get("commit", {}).get("id") if isinstance(value, dict) else None
return _safe_oid(sha, "remote_branch_oid")
def _askpass(self) -> tuple[Path, dict[str, str]]:
credential = self._credential_path
fd, name = tempfile.mkstemp(prefix="aios-gitea-askpass-", dir="/tmp")
path = Path(name)
os.fchmod(fd, 0o700)
script = "#!/bin/sh\ncase \"$1\" in *Username*) printf '%s\\n' \"$GITEA_MACHINE_ACTOR\" ;; *) cat \"$CREDENTIAL_FILE\" ;; esac\n"
with os.fdopen(fd, "w", encoding="utf-8") as stream:
stream.write(script)
stream.flush()
os.fsync(stream.fileno())
return path, {
"GIT_ASKPASS": str(path),
"GIT_TERMINAL_PROMPT": "0",
"GITEA_MACHINE_ACTOR": self.actor,
"CREDENTIAL_FILE": str(credential),
}
owner, name = self.repository_id.split("/", 1)
try:
value = self._request("GET", f"/repos/{quote(owner)}/{quote(name)}/branches/{quote(branch, safe='/')}")[1]
except GiteaTransportError as exc:
if str(exc) == "GITEA_HTTP_404":
return None
raise
sha = value.get("commit", {}).get("id") if isinstance(value, dict) else None
return _safe_oid(sha, "remote_branch_oid")
def _askpass(self) -> tuple[Path, dict[str, str]]:
credential = self._credential_path
fd, name = tempfile.mkstemp(prefix="aios-gitea-askpass-", dir="/tmp")
path = Path(name)
os.fchmod(fd, 0o700)
script = "#!/bin/sh\ncase \"$1\" in *Username*) printf '%s\\n' \"$GITEA_MACHINE_ACTOR\" ;; *) cat \"$CREDENTIAL_FILE\" ;; esac\n"
with os.fdopen(fd, "w", encoding="utf-8") as stream:
stream.write(script)
stream.flush()
os.fsync(stream.fileno())
return path, {
"GIT_ASKPASS": str(path),
"GIT_TERMINAL_PROMPT": "0",
"GITEA_MACHINE_ACTOR": self.actor,
"CREDENTIAL_FILE": str(credential),
}
def _push_branch_fast_forward(self, branch: str, new_oid: str) -> None:
askpass, additions = self._askpass()
environment = os.environ.copy()
environment.update(additions)
try:
result = subprocess.run(
["git", f"--git-dir={self.authority_git_dir}", "push", self.remote_url, f"{new_oid}:refs/heads/{branch}"],
text=True,
capture_output=True,
check=False,
env=environment,
timeout=self.timeout,
)
if result.returncode != 0:
raise GiteaTransportError("GITEA_BRANCH_CREATE_FAILED")
finally:
askpass.unlink(missing_ok=True)
askpass, additions = self._askpass()
environment = os.environ.copy()
environment.update(additions)
try:
result = subprocess.run(
["git", f"--git-dir={self.authority_git_dir}", "push", self.remote_url, f"{new_oid}:refs/heads/{branch}"],
text=True,
capture_output=True,
check=False,
env=environment,
timeout=self.timeout,
)
if result.returncode != 0:
raise GiteaTransportError("GITEA_BRANCH_CREATE_FAILED")
finally:
askpass.unlink(missing_ok=True)
def fetch_branch_exact(self, branch: str, expected_oid: str) -> str:
"""Fetch one exact remote branch tip into a quarantine ref.
@@ -249,64 +249,86 @@ class GiteaMachineTransport:
raise GiteaTransportError("GITEA_SOURCE_REF_MOVED")
return expected_oid
def compare_and_swap_branch(self, branch: str, expected_oid: str, new_oid: str) -> str:
_safe_oid(expected_oid, "expected_oid")
_safe_oid(new_oid, "new_oid")
current = self.read_branch(branch)
if current == new_oid:
return new_oid
if current is None:
def compare_and_swap_branch(self, branch: str, expected_oid: str, new_oid: str) -> str:
_safe_oid(expected_oid, "expected_oid")
_safe_oid(new_oid, "new_oid")
current = self.read_branch(branch)
if current == new_oid:
return new_oid
if current is None:
self._push_branch_fast_forward(branch, new_oid)
readback = self.read_branch(branch)
if readback != new_oid:
raise GiteaTransportError("GITEA_BRANCH_CREATE_READBACK_MISMATCH")
return readback
if current != expected_oid:
raise GiteaTransportError("GITEA_BRANCH_EXPECTATION_MISMATCH")
readback = self.read_branch(branch)
if readback != new_oid:
raise GiteaTransportError("GITEA_BRANCH_CREATE_READBACK_MISMATCH")
return readback
if current != expected_oid:
raise GiteaTransportError("GITEA_BRANCH_EXPECTATION_MISMATCH")
# The preflight read binds the expected old tip. The Git receive-pack
# then enforces the actual fast-forward CAS while transferring the new
# objects; no force/lease override or arbitrary ref is permitted.
self._push_branch_fast_forward(branch, new_oid)
readback = self.read_branch(branch)
if readback != new_oid:
raise GiteaTransportError("GITEA_BRANCH_CAS_READBACK_MISMATCH")
return readback
def _pulls(self) -> list[Mapping[str, Any]]:
owner, name = self.repository_id.split("/", 1)
query = urlencode({"state": "all", "base_branch": "main", "limit": 100})
value = self._request("GET", f"/repos/{quote(owner)}/{quote(name)}/pulls?{query}")[1]
return value if isinstance(value, list) else []
def create_or_update_pr(self, branch: str, target: str, title: str, body: str) -> Mapping[str, Any]:
if target != "main":
raise GiteaTransportError("PR_BASE_SCOPE_DENIED")
head = self.read_branch(branch)
if head is None:
raise GiteaTransportError("PR_HEAD_MISSING")
matches = [p for p in self._pulls() if (p.get("head") or {}).get("ref") == branch and (p.get("base") or {}).get("ref") == target]
open_matches = [p for p in matches if p.get("state") == "open"]
if len(open_matches) > 1:
raise GiteaTransportError("MULTIPLE_MATCHING_PRS")
if open_matches:
if (open_matches[0].get("head") or {}).get("sha") != head:
raise GiteaTransportError("PR_HEAD_CONFLICT")
return open_matches[0]
owner, name = self.repository_id.split("/", 1)
return self._request("POST", f"/repos/{quote(owner)}/{quote(name)}/pulls", {"head": branch, "base": target, "title": title, "body": body, "allow_maintainer_edit": False})[1]
readback = self.read_branch(branch)
if readback != new_oid:
raise GiteaTransportError("GITEA_BRANCH_CAS_READBACK_MISMATCH")
return readback
def _pulls(self) -> list[Mapping[str, Any]]:
owner, name = self.repository_id.split("/", 1)
query = urlencode({"state": "all", "base_branch": "main", "limit": 100})
value = self._request("GET", f"/repos/{quote(owner)}/{quote(name)}/pulls?{query}")[1]
return value if isinstance(value, list) else []
def create_or_update_pr(self, branch: str, target: str, title: str, body: str) -> Mapping[str, Any]:
if target != "main":
raise GiteaTransportError("PR_BASE_SCOPE_DENIED")
head = self.read_branch(branch)
if head is None:
raise GiteaTransportError("PR_HEAD_MISSING")
matches = [p for p in self._pulls() if (p.get("head") or {}).get("ref") == branch and (p.get("base") or {}).get("ref") == target]
open_matches = [p for p in matches if p.get("state") == "open"]
if len(open_matches) > 1:
raise GiteaTransportError("MULTIPLE_MATCHING_PRS")
if open_matches:
if (open_matches[0].get("head") or {}).get("sha") != head:
raise GiteaTransportError("PR_HEAD_CONFLICT")
return open_matches[0]
owner, name = self.repository_id.split("/", 1)
return self._request("POST", f"/repos/{quote(owner)}/{quote(name)}/pulls", {"head": branch, "base": target, "title": title, "body": body, "allow_maintainer_edit": False})[1]
def publish_status(self, branch: str, state: str, description: str, context: str = "aios/durable-release") -> Mapping[str, Any]:
oid = self.read_branch(branch)
if oid is None:
raise GiteaTransportError("STATUS_BRANCH_MISSING")
owner, name = self.repository_id.split("/", 1)
oid = self.read_branch(branch)
if oid is None:
raise GiteaTransportError("STATUS_BRANCH_MISSING")
owner, name = self.repository_id.split("/", 1)
return self._request("POST", f"/repos/{quote(owner)}/{quote(name)}/statuses/{oid}", {"state": state, "context": context, "description": description})[1]
def protected_merge(self, pr_number: int) -> Mapping[str, Any]:
if not isinstance(pr_number, int) or pr_number < 1:
raise GiteaTransportError("PR_NUMBER_INVALID")
owner, name = self.repository_id.split("/", 1)
pr = self._request("GET", f"/repos/{quote(owner)}/{quote(name)}/pulls/{pr_number}")[1]
def publish_status_for_sha(self, sha: str, state: str, description: str, context: str = "aios/durable-release") -> Mapping[str, Any]:
"""Publish a commit status against an explicit 40-hex commit SHA.
Unlike :meth:`publish_status` (which binds to the current tip of a branch),
this variant binds immutably to the supplied object ID. It is the correct
primitive for protected-PR reporting where the PR head SHA is authoritative
and must not drift with a moving branch tip.
"""
if state not in {"pending", "success", "error", "failure", "warning"}:
raise GiteaTransportError("STATUS_STATE_INVALID")
oid = _safe_oid(sha, "status_oid")
owner, name = self.repository_id.split("/", 1)
return self._request("POST", f"/repos/{quote(owner)}/{quote(name)}/statuses/{oid}", {"state": state, "context": context, "description": description})[1]
def read_statuses(self, sha: str, context: str | None = None) -> list[Mapping[str, Any]]:
"""Read commit statuses for an explicit SHA (used for RECONCILE)."""
oid = _safe_oid(sha, "status_read_oid")
owner, name = self.repository_id.split("/", 1)
query = f"?context={quote(context)}" if context else ""
value = self._request("GET", f"/repos/{quote(owner)}/{quote(name)}/statuses/{oid}{query}")[1]
return value if isinstance(value, list) else []
def protected_merge(self, pr_number: int) -> Mapping[str, Any]:
if not isinstance(pr_number, int) or pr_number < 1:
raise GiteaTransportError("PR_NUMBER_INVALID")
owner, name = self.repository_id.split("/", 1)
pr = self._request("GET", f"/repos/{quote(owner)}/{quote(name)}/pulls/{pr_number}")[1]
if pr.get("state") != "open" or (pr.get("base") or {}).get("ref") != "main":
raise GiteaTransportError("PR_MERGE_BINDING_INVALID")
if pr.get("mergeable") is False:
@@ -0,0 +1,113 @@
"""Pure scope classifier for N/A-capable status contexts.
Architecture §2.6: campaign-approval-gate and signer-validator-contract may be
FORMALLY_NOT_APPLICABLE for a given PR only when an explicit applicability
contract reference exists. This module is pure (no I/O) so it can be tested on
any platform.
Policy:
- UNKNOWN/ambiguous scope facts -> NEVER success. The classifier raises or
returns a FAIL/UNKNOWN verdict rather than guessing APPLICABLE and later
publishing PASS, or guessing N/A without a contract ref.
- FORMALLY_NOT_APPLICABLE always carries a na_contract_ref so the applicability
decision is auditable and traceable (architecture §2.7 evidence field).
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Mapping
# Campaign N/A contract references (governance clause defining N/A as a legal pass).
CAMPAIGN_NA_CONTRACT_REF = "aios.campaign-approval-gate.v1#formally-not-applicable"
SIGNER_NA_CONTRACT_REF = "aios.signer-validator-contract.v1#formally-not-applicable"
class ScopeClassificationError(RuntimeError):
pass
@dataclass(frozen=True, slots=True)
class ScopeVerdict:
context: str
scope: str # "APPLICABLE" | "FORMALLY_NOT_APPLICABLE"
na_contract_ref: str | None
reason: str
facts: Mapping[str, Any]
@property
def applicable(self) -> bool:
return self.scope == "APPLICABLE"
@property
def formally_not_applicable(self) -> bool:
return self.scope == "FORMALLY_NOT_APPLICABLE"
# ---------------------------------------------------------------------------
# Context-specific applicability rules
# ---------------------------------------------------------------------------
def _campaign_append_tuples():
return ("campaign_authorization_present", "campaign_active", "in_campaign_lineage")
def _classify_campaign(facts: Mapping[str, Any]) -> ScopeVerdict:
authorization_present = facts.get("campaign_authorization_present")
campaign_active = facts.get("campaign_active")
in_lineage = facts.get("in_campaign_lineage")
if authorization_present is None or campaign_active is None or in_lineage is None:
raise ScopeClassificationError("CAMPAIGN_SCOPE_FACTS_INCOMPLETE")
if authorization_present is True and campaign_active is True and in_lineage is True:
return ScopeVerdict(
"campaign-approval-gate",
"APPLICABLE",
None,
"active_campaign_governs_pr",
dict(facts),
)
if authorization_present is False and in_lineage is False:
return ScopeVerdict(
"campaign-approval-gate",
"FORMALLY_NOT_APPLICABLE",
CAMPAIGN_NA_CONTRACT_REF,
"no_active_campaign_lineage",
dict(facts),
)
# Ambiguous (e.g. authorization present but inactive, or lineage unknown):
# an N/A verdict without a clean contract basis is forbidden - fail closed.
raise ScopeClassificationError("CAMPAIGN_SCOPE_AMBIGUOUS_FAIL_CLOSED")
def _classify_signer(facts: Mapping[str, Any]) -> ScopeVerdict:
touches_signer = facts.get("touches_signer_bound_change")
if touches_signer is None:
raise ScopeClassificationError("SIGNER_SCOPE_FACTS_INCOMPLETE")
if touches_signer is True:
return ScopeVerdict(
"signer-validator-contract",
"APPLICABLE",
None,
"signer_bound_change_present",
dict(facts),
)
return ScopeVerdict(
"signer-validator-contract",
"FORMALLY_NOT_APPLICABLE",
SIGNER_NA_CONTRACT_REF,
"no_signer_bound_change",
dict(facts),
)
def classify_na(context: str, pr_scope_facts: Mapping[str, Any]) -> ScopeVerdict:
"""Classify whether a context is APPLICABLE or FORMALLY_NOT_APPLICABLE.
Raises ScopeClassificationError (never returns a usable success) when scope
facts are incomplete or ambiguous. UNKNOWN never becomes APPLICABLE_PASS
or a bare FORMALLY_NOT_APPLICABLE success without a contract ref.
"""
if context == "campaign-approval-gate":
return _classify_campaign(pr_scope_facts)
if context == "signer-validator-contract":
return _classify_signer(pr_scope_facts)
# Any other context has no N/A semantics in this pipeline.
raise ScopeClassificationError(f"NO_NA_SEMANTICS_FOR_CONTEXT:{context}")
@@ -0,0 +1,200 @@
"""Protected-main status evidence contract (pure, no external side effects).
This module defines the canonical evidence record that a status producer emits
and the Status Publisher persists. It deliberately has zero I/O: everything
here is deterministic derivation from an in-memory record, so the schema and
identity rules can be unit-tested on any platform.
Identity model (architecture §2.4/§2.7):
effect_id = sha256(repo + "\\0" + commit_sha + "\\0" + context + "\\0" + evidence_revision)
The evidence_revision is a canonical sha256 over the evidence record *except*
the volatile fields (observed_at / effect_id / evidence_ref), so a re-run that
produces the identical result/inputs/version hashes to the same revision and is
idempotently suppressed, while any change in result/input/producer version
yields a new revision that legitimately re-publishes.
"""
from __future__ import annotations
import hashlib
import json
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Mapping
STATUS_EVIDENCE_SCHEMA = "aios.status-evidence.v1"
# The authoritative set of status contexts governed by this pipeline.
REQUIRED_CONTEXTS: tuple[str, ...] = (
"py-compile",
"git-diff-check",
"secret-scan",
"autonomy-unit-tests",
"autonomy-smoke-tests",
"reconciliation-contract",
"release-state-machine",
"campaign-approval-gate",
"signer-validator-contract",
)
ALLOWED_CONTEXTS: frozenset[str] = frozenset(REQUIRED_CONTEXTS)
# Results that may map to a Gitea success status. Everything else is never
# published as success (architecture §2.6 red line).
# APPLICABLE_PASS -> success
# FORMALLY_NOT_APPLICABLE -> success ONLY when a na_contract_ref is present.
# FAIL / UNKNOWN -> never success.
SUCCESS_RESULTS: frozenset[str] = frozenset({"APPLICABLE_PASS", "FORMALLY_NOT_APPLICABLE"})
def _canonical_json_bytes(value: Any) -> bytes:
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
def sha256_hex(value: Any) -> str:
return hashlib.sha256(_canonical_json_bytes(value)).hexdigest()
def utc_now() -> str:
return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
def validate_context(context: str) -> str:
context = str(context or "").strip()
if context not in ALLOWED_CONTEXTS:
raise ValueError(f"UNKNOWN_STATUS_CONTEXT:{context}")
return context
def validate_repository(repository: str, allowed_repository: str = "aios/aios") -> str:
repo = str(repository or "").strip()
if repo != allowed_repository:
raise ValueError(f"WRONG_REPOSITORY:{repo}")
return repo
def validate_commit_sha(commit_sha: str) -> str:
sha = str(commit_sha or "").strip().lower()
if len(sha) != 40 or any(ch not in "0123456789abcdef" for ch in sha):
raise ValueError("WRONG_COMMIT_SHA")
return sha
def compute_evidence_revision(record: Mapping[str, Any]) -> str:
"""Canonical sha256 over the deterministic evidence fields.
Volatile fields (observed_at / effect_id / evidence_ref / na_contract_ref
legality is fixed at record time) are excluded so that equal results and
inputs hash equal across runs.
"""
volatile = {"observed_at", "effect_id", "evidence_ref"}
core = {k: v for k, v in dict(record).items() if k not in volatile}
return "sha256:" + sha256_hex(core)
def compute_effect_id(repository: str, commit_sha: str, context: str, evidence_revision: str) -> str:
return "sha256:" + sha256_hex(f"{repository}\0{commit_sha}\0{context}\0{evidence_revision}")
@dataclass(frozen=True, slots=True)
class StatusEvidence:
schema: str = STATUS_EVIDENCE_SCHEMA
repository: str = "aios/aios"
commit_sha: str = ""
context: str = ""
result: str = ""
producer_identity: str = ""
producer_revision: str = ""
evidence_revision: str = ""
observed_at: str = ""
evidence_ref: str = ""
na_contract_ref: str | None = None
effect_id: str = ""
detail: Mapping[str, Any] = field(default_factory=dict)
def validated(self, *, observed_at: str | None = None) -> "StatusEvidence":
repo = validate_repository(self.repository)
sha = validate_commit_sha(self.commit_sha)
ctx = validate_context(self.context)
if self.result not in {"APPLICABLE_PASS", "FAIL", "UNKNOWN", "FORMALLY_NOT_APPLICABLE"}:
raise ValueError(f"INVALID_EVIDENCE_RESULT:{self.result}")
if self.result == "FORMALLY_NOT_APPLICABLE" and not self.na_contract_ref:
raise ValueError("NA_WITHOUT_CONTRACT_REF")
if not self.producer_identity or not self.producer_revision:
raise ValueError("PRODUCER_IDENTITY_INCOMPLETE")
ts = observed_at or utc_now()
return StatusEvidence(
schema=STATUS_EVIDENCE_SCHEMA,
repository=repo,
commit_sha=sha,
context=ctx,
result=self.result,
producer_identity=self.producer_identity,
producer_revision=self.producer_revision,
evidence_revision=self.evidence_revision,
observed_at=ts,
evidence_ref=self.evidence_ref,
na_contract_ref=self.na_contract_ref,
effect_id=self.effect_id,
detail=dict(self.detail),
)
def finalize(self, *, observed_at: str | None = None, evidence_ref: str = "") -> "StatusEvidence":
"""Compute evidence_revision + effect_id and return a validated record."""
if self.result == "FORMALLY_NOT_APPLICABLE" and not self.na_contract_ref:
raise ValueError("NA_WITHOUT_CONTRACT_REF")
base = dict(
repository=self.repository,
commit_sha=self.commit_sha,
context=self.context,
result=self.result,
producer_identity=self.producer_identity,
producer_revision=self.producer_revision,
na_contract_ref=self.na_contract_ref,
detail=dict(self.detail),
evidence_ref=evidence_ref or self.evidence_ref,
)
rev = compute_evidence_revision(base)
effect = compute_effect_id(self.repository, self.commit_sha, self.context, rev)
return StatusEvidence(
schema=STATUS_EVIDENCE_SCHEMA,
repository=self.repository,
commit_sha=validate_commit_sha(self.commit_sha),
context=validate_context(self.context),
result=self.result,
producer_identity=self.producer_identity,
producer_revision=self.producer_revision,
evidence_revision=rev,
observed_at=observed_at or utc_now(),
evidence_ref=evidence_ref or self.evidence_ref,
na_contract_ref=self.na_contract_ref,
effect_id=effect,
detail=dict(self.detail),
)
def as_dict(self) -> dict[str, Any]:
return {
"schema": self.schema,
"repository": self.repository,
"commit_sha": self.commit_sha,
"context": self.context,
"result": self.result,
"producer_identity": self.producer_identity,
"producer_revision": self.producer_revision,
"evidence_revision": self.evidence_revision,
"observed_at": self.observed_at,
"evidence_ref": self.evidence_ref,
"na_contract_ref": self.na_contract_ref,
"effect_id": self.effect_id,
"detail": dict(self.detail),
}
def maps_to_success(result: str) -> bool:
"""Whether a result may be published as a Gitea success status.
FAIL / UNKNOWN never map to success. FORMALLY_NOT_APPLICABLE only maps when
a contract ref is present, which is enforced at record validation time.
"""
return result in SUCCESS_RESULTS
+186
View File
@@ -0,0 +1,186 @@
"""Protected-main status orchestrator.
Responsibilities (architecture §2.2 / §2.5):
1. Resolve the PR's exact head SHA (never guessed locally).
2. Schedule all 9 required producers, each owning exactly one context.
3. Convert each ProducerOutput into a StatusEvidence and hand it to the
Status Publisher (the only writer of Gitea statuses).
4. Enforce stale-SHA safety: every run keys effects on the *current* head SHA;
evidence is always regenerated for that head economy — old SHA statuses are
never copied onto a new head. A head change requires a full re-run.
5. Report merge eligibility: protected main may merge only when every required
context resolved to a genuine ``success`` status on the current head.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable, Mapping
from status_evidence_contract import (
REQUIRED_CONTEXTS,
StatusEvidence,
validate_context,
validate_commit_sha,
validate_repository,
)
from status_publisher import AppendOnlyEvidenceStore, StatusPublisher
class OrchestratorError(RuntimeError):
pass
@dataclass(frozen=True, slots=True)
class ContextOutcome:
context: str
producer_output: Any
evidence: StatusEvidence
publish: Any # PublishOutcome
@dataclass(frozen=True, slots=True)
class RunResult:
head_sha: str
context_outcomes: Mapping[str, ContextOutcome]
merge_eligible: bool
def resolve_pr_head_sha(pr: Mapping[str, Any], *, expected_repository: str = "aios/aios") -> str:
"""Resolve the exact head SHA from a Gitea PR object (authoritative)."""
repo = (pr.get("base") or {}).get("repo") or {}
full_name = repo.get("full_name") or repo.get("fullname") or ""
if full_name and full_name != expected_repository:
raise OrchestratorError(f"WRONG_REPOSITORY:{full_name}")
head = (pr.get("head") or {}).get("sha")
if not head:
raise OrchestratorError("PR_HEAD_SHA_MISSING")
return validate_commit_sha(head)
# The 9 producers, keyed by their governed context name.
def _default_producers() -> Mapping[str, Callable[[Any], Any]]:
from status_producers import (
py_compile,
git_diff_check,
secret_scan,
autonomy_unit_tests,
autonomy_smoke_tests,
reconciliation_contract,
release_state_machine,
campaign_approval_gate,
signer_validator_contract,
)
return {
"py-compile": py_compile.run,
"git-diff-check": git_diff_check.run,
"secret-scan": secret_scan.run,
"autonomy-unit-tests": autonomy_unit_tests.run,
"autonomy-smoke-tests": autonomy_smoke_tests.run,
"reconciliation-contract": reconciliation_contract.run,
"release-state-machine": release_state_machine.run,
"campaign-approval-gate": campaign_approval_gate.run,
"signer-validator-contract": signer_validator_contract.run,
}
class StatusOrchestrator:
def __init__(
self,
publisher: StatusPublisher,
*,
producers: Mapping[str, Callable[[Any], Any]] | None = None,
checkout: Path | None = None,
base_sha: str | None = None,
extra_facts: Mapping[str, Any] | None = None,
) -> None:
self.publisher = publisher
self.producers = _default_producers() if producers is None else producers
if set(self.producers) != set(REQUIRED_CONTEXTS):
raise OrchestratorError("PRODUCER_CONTEXT_SET_MISALIGNED")
self.checkout = checkout
self.base_sha = base_sha
self.extra_facts = dict(extra_facts or {})
def run_for_pr(self, pr: Mapping[str, Any]) -> RunResult:
head_sha = resolve_pr_head_sha(pr)
return self._run_head(head_sha)
def run_for_head(self, head_sha: str) -> RunResult:
return self._run_head(validate_commit_sha(head_sha))
def _run_head(self, head_sha: str) -> RunResult:
outcomes: dict[str, ContextOutcome] = {}
for context in REQUIRED_CONTEXTS:
validate_context(context)
producer = self.producers[context]
try:
producer_output = producer(
self._make_ctx(context, head_sha)
)
except Exception as exc: # noqa: BLE001
from status_producers import unknown_output
producer_output = unknown_output(
context, f"aios.{context}", f"producer_error:{exc}"
)
evidence = StatusEvidence(
repository=validate_repository("aios/aios"),
commit_sha=head_sha,
context=context,
result=producer_output.result,
producer_identity=producer_output.producer_identity,
producer_revision=producer_output.producer_revision,
na_contract_ref=producer_output.na_contract_ref,
detail=dict(producer_output.detail),
)
publish = self.publisher.publish(evidence)
outcomes[context] = ContextOutcome(context, producer_output, evidence, publish)
merge_eligible = all(
self._context_green(outcomes[ctx]) for ctx in REQUIRED_CONTEXTS
)
return RunResult(head_sha, outcomes, merge_eligible)
@staticmethod
def _context_green(outcome: ContextOutcome) -> bool:
"""A context is merge-green IFF its *verdict* genuinely passed AND its
status publish was *confirmed*.
The verdict (evidence.result) is authoritative (architecture §5): a
FAIL/UNKNOWN context can never be green no matter what a conflicting
leftover Gitea status says. A green verdict additionally requires a
confirmed publish (PUBLISHED / RECONCILED matched_expected / SUPPRESS_NOOP
with a verified success readback) — never an unconfirmed
existing_state_differs result.
"""
evidence = outcome.evidence
verdict_green = (
evidence.result == "APPLICABLE_PASS"
or (
evidence.result == "FORMALLY_NOT_APPLICABLE"
and bool(evidence.na_contract_ref)
)
)
if not verdict_green:
return False
publish = outcome.publish
if not publish.confirmed:
return False
# Confirmation must carry the eligible action forms only.
if publish.action not in {"PUBLISHED", "RECONCILED", "SUPPRESS_NOOP"}:
return False
return True
def _make_ctx(self, context: str, head_sha: str) -> Any:
from status_producers import ProducerContext
facts = dict(self.extra_facts)
facts.setdefault("commit_sha", head_sha)
return ProducerContext(
context=context,
repository="aios/aios",
commit_sha=head_sha,
checkout=self.checkout,
base_sha=self.base_sha,
scope_facts=facts,
)
+197
View File
@@ -0,0 +1,197 @@
"""Protected-main status pipeline entry point (S6 trigger integration skeleton).
This module wires the real, locally-validated pipeline to a production Gitea
broker using environment-provided configuration. It is intentionally a *thin*
adapter: it constructs a capability-minimal broker (only ``status_publish`` plus
the readback needed for RECONCILE), the idempotent publisher, the append-only
evidence store, and an orchestrator, then runs one PR.
It does NOT fake or shortcut production connectivity: any missing environment
(exact PR head, checkout, gitleaks, signer, campaign authorization) is surfaced
as an explicit error/UNKNOWN through the pipeline's fail-closed semantics. This
file is the integration point a production trigger (webhook / durable job /
poll loop) calls; the trigger mechanism itself is owned by the orchestrating
service and is out of scope here.
"""
from __future__ import annotations
import argparse
import json
import os
from pathlib import Path
from aios_gitea_machine_broker import (
CampaignCapabilityReceipt,
GiteaMachineBroker,
MachineCredentialProjection,
)
from status_evidence_contract import REQUIRED_CONTEXTS
from status_orchestrator import StatusOrchestrator
from status_publisher import (
AppendOnlyEvidenceStore,
DANGEROUS_CAPABILITIES,
StatusPublisher,
)
REPOSITORY = os.environ.get("AIOS_GITEA_REPOSITORY", "aios/aios")
def _json_env(name: str) -> dict | None:
"""Parse a JSON object from an environment variable; None if absent/unparsable.
Unparsable input is ignored (fail-closed) so a malformed injection never
silently degrades a check to green — it simply behaves as if not provided,
which keeps the pipeline's UNKNOWN fail-closed semantics intact.
"""
raw = os.environ.get(name, "").strip()
if not raw:
return None
try:
value = json.loads(raw)
except json.JSONDecodeError:
return None
return value if isinstance(value, dict) else None
def build_orchestrator(
*,
scope_facts: dict | None = None,
smoke_results: dict | None = None,
) -> StatusOrchestrator:
"""Build the orchestrator from environment (capability-minimal broker).
``scope_facts`` (campaign_authorization_present / campaign_active /
in_campaign_lineage / touches_signer_bound_change) and ``smoke_results`` are
*optional injection channels* that let a production trigger supply real
observed applicability / smoke findings so the 9 contexts can genuinely
reach 9/9. They are never defaulted to fake pass values: when absent or
malformed the pipeline stays UNKNOWN fail-closed. Env vars are merged
first; explicit caller args take precedence.
"""
from aios_gitea_machine_transport import GiteaMachineTransport
account = os.environ.get("AIOS_GITEA_MACHINE_ACCOUNT", "aios-autonomy-agent")
api = os.environ.get(
"AIOS_GITEA_API", "https://g.cnnetmkt.top/git/api/v1"
)
remote = os.environ.get(
"AIOS_GITEA_REMOTE", "https://g.cnnetmkt.top/git/aios/aios.git"
)
authority_dir = Path(
os.environ.get("AIOS_GITOPS_AUTHORITY_GIT_DIR", "/data/aios-gitops/repo.git")
)
checkout = Path(os.environ.get("AIOS_STATUS_CHECKOUT", "")) or None
evidence_root = Path(
os.environ.get("AIOS_STATUS_EVIDENCE_ROOT", "/var/lib/aios-gitops/status-evidence")
)
transport = GiteaMachineTransport(api, REPOSITORY, remote, authority_dir)
credential = MachineCredentialProjection.from_mapping(
{"scheme": "systemd-credential", "name": "gitea-machine-token", "account": account}
)
# Capability-minimal: only status_publish (+ readback for RECONCILE), and an
# explicit denied set covering dangerous capabilities (aligned with the
# gitops denied set). The publisher additionally rejects any dangerous cap
# that leaks through here.
receipt = CampaignCapabilityReceipt(
os.environ.get("AIOS_CAMPAIGN_ID", "status-pipeline"),
os.environ.get("AIOS_TASK_ID", "status-pipeline-task"),
REPOSITORY,
os.environ.get("AIOS_STATUS_BRANCH", "agent/status-pipeline"),
os.environ.get("AIOS_EXPECTED_BRANCH_OID", ""),
frozenset({"status_publish"}),
frozenset(DANGEROUS_CAPABILITIES),
)
broker = GiteaMachineBroker(transport, credential, receipt)
store = AppendOnlyEvidenceStore(evidence_root)
publisher = StatusPublisher(
broker,
store,
receipt_capabilities={
"capabilities": set(receipt.capabilities),
"denied": set(receipt.denied),
},
)
# Assemble optional injection channels (env first, caller args override).
extra_facts: dict = {
"pr_number": int(os.environ.get("AIOS_STATUS_PR_NUMBER", "0") or 0),
}
env_scope = _json_env("AIOS_STATUS_SCOPE_FACTS")
if env_scope:
extra_facts.update(env_scope)
if scope_facts:
extra_facts.update(scope_facts)
env_smoke = _json_env("AIOS_STATUS_SMOKE_RESULTS")
smoke: dict = {}
if env_smoke:
smoke.update(env_smoke)
if smoke_results:
smoke.update(smoke_results)
if smoke:
extra_facts["smoke_results"] = smoke
return StatusOrchestrator(
publisher,
checkout=checkout,
base_sha=os.environ.get("AIOS_STATUS_BASE_SHA"),
extra_facts=extra_facts,
)
def _json_arg(value: str | None, name: str) -> dict | None:
if value is None or not value.strip():
return None
try:
parsed = json.loads(value)
except json.JSONDecodeError as exc:
raise SystemExit(f"{name} must be a JSON object") from exc
if not isinstance(parsed, dict):
raise SystemExit(f"{name} must be a JSON object")
return parsed
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Run the protected-main status pipeline for a PR")
parser.add_argument("--pr-json", help="Gitea PR object JSON (must include base.repo and head.sha)")
parser.add_argument("--head-sha", help="exact head SHA (alternative to --pr-json)")
parser.add_argument("--scope-facts", help="JSON object of applicability scope facts (campaign/signer)")
parser.add_argument("--smoke-results", help="JSON object of observed smoke check results")
args = parser.parse_args(argv)
orchestrator = build_orchestrator(
scope_facts=_json_arg(args.scope_facts, "--scope-facts"),
smoke_results=_json_arg(args.smoke_results, "--smoke-results"),
)
try:
if args.head_sha:
result = orchestrator.run_for_head(args.head_sha)
elif args.pr_json:
result = orchestrator.run_for_pr(json.loads(args.pr_json))
else:
raise SystemExit("require --pr-json or --head-sha")
except Exception as exc: # noqa: BLE001
print(json.dumps({"status": "ERROR", "reason": str(exc)}, sort_keys=True))
return 2
summary = {
"head_sha": result.head_sha,
"required_contexts": len(REQUIRED_CONTEXTS),
"merge_eligible": result.merge_eligible,
"contexts": {
ctx: {
"action": o.publish.action,
"state": o.publish.gitea_state,
"result": o.producer_output.result,
}
for ctx, o in result.context_outcomes.items()
},
}
print(json.dumps(summary, sort_keys=True))
return 0 if result.merge_eligible else 1
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,79 @@
"""Protected-main status producers (thin, real-execution wrappers).
Each producer owns exactly one required context and derives a deterministic
verdict from *real* tool output or a real validator — never a manufactured
``success``. Producers never write Gitea statuses; they emit a
:class:`ProducerOutput` that the Status Orchestrator converts into a
:class:`~status_evidence_contract.StatusEvidence` for the Status Publisher.
Testability
-----------
The real-execution subprocess helper ``run_command`` is injectable via
``set_command_runner`` so unit tests can resolve failures without invoking the
real toolchain (there is no Gitea / gitleaks / signer binary on non-Linux dev
hosts). Production loans the real runner.
"""
from __future__ import annotations
import subprocess
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable, Mapping, Sequence
# Every producer has a fixed producer_identity (audit attribution).
PRODUCER_REVISION = "producer.2026-09-15.1"
CommandRunner = Callable[[Sequence[str], Path | None], subprocess.CompletedProcess[str]]
def _default_command_runner(
command: Sequence[str], cwd: Path | None = None, *, timeout: int = 300
) -> subprocess.CompletedProcess[str]:
return subprocess.run(
list(command),
cwd=cwd,
text=True,
capture_output=True,
timeout=timeout,
check=False,
)
_ACTIVE_RUNNER: CommandRunner = _default_command_runner
def set_command_runner(runner: CommandRunner) -> None:
"""Override the subprocess runner (used by tests to avoid real toolchains)."""
global _ACTIVE_RUNNER
_ACTIVE_RUNNER = runner
@dataclass(frozen=True, slots=True)
class ProducerContext:
context: str = "" # governed context name this run serves
repository: str = "aios/aios"
commit_sha: str = ""
checkout: Path | None = None # exact-head checkout (real exec target)
base_sha: str | None = None # for base...head checks
scope_facts: Mapping[str, Any] = field(default_factory=dict)
runner_timeout: int = 300
@dataclass(frozen=True, slots=True)
class ProducerOutput:
context: str
result: str # APPLICABLE_PASS | FAIL | UNKNOWN | FORMALLY_NOT_APPLICABLE
producer_identity: str
producer_revision: str
detail: Mapping[str, Any] = field(default_factory=dict)
na_contract_ref: str | None = None
def fail_output(context: str, identity: str, reason: str, detail: Mapping[str, Any] | None = None) -> ProducerOutput:
merged = dict(detail or {})
merged["reason"] = reason
return ProducerOutput(context, "FAIL", identity, PRODUCER_REVISION, merged)
def unknown_output(context: str, identity: str, reason: str) -> ProducerOutput:
return ProducerOutput(context, "UNKNOWN", identity, PRODUCER_REVISION, {"reason": reason})
@@ -0,0 +1,57 @@
"""autonomy-smoke-tests producer: real staging/smoke checklist.
This must never be fulfilled by rerunning the unit suite. Smoke checks are a
separate, environment-dependent checklist; where a real staging environment is
unavailable the producer reports UNKNOWN rather than an invented pass.
"""
from __future__ import annotations
from pathlib import Path
from . import (
PRODUCER_REVISION,
ProducerContext,
ProducerOutput,
fail_output,
unknown_output,
)
# Smoke checklist entries as (label). Each must map to a real check at runtime;
# the checklist itself is the contract, not the unit suite.
SMOKE_CHECKS: tuple[str, ...] = (
"imports_load",
"core_services_respond",
)
def run(
ctx: ProducerContext,
*,
results: dict[str, bool] | None = None,
) -> ProducerOutput:
identity = "aios.autonomy-smoke-tests"
checkout = ctx.checkout
if checkout is None or not checkout.is_dir():
return unknown_output(ctx.context, identity, "checkout_unavailable")
# Production executes real staging probes and reports a results map via the
# injected smoke_results (contract layer); fall back to scope_facts so an
# orchestrator/entry can pass concrete observed results. If no real results
# were observed, this is UNKNOWN — never an auto-pass.
if results is None:
injected = ctx.scope_facts.get("smoke_results")
results = dict(injected) if isinstance(injected, dict) else None
if results is None:
return unknown_output(ctx.context, identity, "smoke_environment_unavailable")
missing = [label for label in SMOKE_CHECKS if label not in results]
if missing:
return fail_output(ctx.context, identity, "smoke_results_incomplete", {"missing": missing})
all_passed = all(bool(results[label]) for label in SMOKE_CHECKS)
return ProducerOutput(
ctx.context,
"APPLICABLE_PASS" if all_passed else "FAIL",
identity,
PRODUCER_REVISION,
{"smoke_checks": {label: bool(results[label]) for label in SMOKE_CHECKS}},
)
@@ -0,0 +1,60 @@
"""autonomy-unit-tests producer: real pytest suite on a stable subset."""
from __future__ import annotations
import os
from pathlib import Path
from . import (
PRODUCER_REVISION,
ProducerContext,
ProducerOutput,
fail_output,
unknown_output,
)
# Stable pure-logic subset that does not drag in fcntl/systemd/signer binaries,
# so it can run on non-Linux hosts and in CI. Production may extend this list,
# but a producer pass must always result from a real suite that ran green.
STABLE_UNIT_TESTS: tuple[str, ...] = (
"tests/test_convergence_contracts.py",
"tests/test_gitea_transport_consistency_provider.py",
)
def run(ctx: ProducerContext) -> ProducerOutput:
identity = "aios.autonomy-unit-tests"
checkout = ctx.checkout
if checkout is None or not checkout.is_dir():
return unknown_output(ctx.context, identity, "checkout_unavailable")
missing = [t for t in STABLE_UNIT_TESTS if not (checkout / t).is_file()]
if missing:
return fail_output(ctx.context, identity, "test_targets_missing", {"missing": missing})
from . import _ACTIVE_RUNNER
app = checkout / "active-runtime" / "app"
env_paths = [str(checkout), str(app)]
env = dict(os.environ)
env["PYTHONPATH"] = os.pathsep.join(env_paths)
result = _ACTIVE_RUNNER(
["python", "-m", "pytest", "-q", "-p", "no:cacheprovider", *STABLE_UNIT_TESTS],
checkout,
)
return _from_pytest(ctx, result, identity)
def _from_pytest(ctx: ProducerContext, result, identity: str) -> ProducerOutput:
# A pass requires a real, green run: exit 0 and no failures. The runner
# returns a string-yield CompletedProcess; tests may attach a mock.
passed = result.returncode == 0
detail = {
"exit_code": result.returncode,
"tests_run": [t for t in STABLE_UNIT_TESTS],
"summary": (result.stdout or "")[:2000],
}
return ProducerOutput(
ctx.context,
"APPLICABLE_PASS" if passed else "FAIL",
identity,
PRODUCER_REVISION,
detail,
)
@@ -0,0 +1,92 @@
"""campaign-approval-gate producer.
Scope is classified via ``producer_scope_classifier.classify_na``:
- APPLICABLE -> run the real ``campaign_approval_gate.evaluate``; a missing
CampaignAuthorization or unavailable toolchain is UNKNOWN, never success.
- FORMALLY_NOT_APPLICABLE -> emit an NA verdict bound to the applicability
contract ref. An N/A verdict with no contract ref is forbidden.
"""
from __future__ import annotations
from pathlib import Path
from producer_scope_classifier import (
ScopeClassificationError,
classify_na,
)
from . import (
PRODUCER_REVISION,
ProducerContext,
ProducerOutput,
fail_output,
unknown_output,
)
def run(
ctx: ProducerContext,
*,
campaign_facts: dict | None = None,
campaign_path: Path | None = None,
) -> ProducerOutput:
identity = "aios.campaign-approval-gate"
facts = campaign_facts if campaign_facts is not None else dict(ctx.scope_facts)
try:
verdict = classify_na(ctx.context, facts)
except ScopeClassificationError as exc:
return unknown_output(ctx.context, identity, f"scope_unknown:{exc}")
if verdict.formally_not_applicable:
return ProducerOutput(
ctx.context,
"FORMALLY_NOT_APPLICABLE",
identity,
PRODUCER_REVISION,
{"reason": verdict.reason, "scope_facts": dict(facts), "applicable": False},
na_contract_ref=verdict.na_contract_ref,
)
# APPLICABLE: real gate requires a CampaignAuthorization artifact.
if campaign_path is None or not campaign_path.is_file():
return unknown_output(ctx.context, identity, "campaign_authorization_unavailable")
try:
from release_campaign_authorization import CampaignAuthorization
from campaign_approval_gate import evaluate
except Exception as exc: # noqa: BLE001
return unknown_output(ctx.context, identity, f"gate_module_unavailable:{exc}")
try:
campaign = CampaignAuthorization.load(campaign_path, require_active=True)
gate_facts = {
"repository": ctx.repository,
"pr_number": int(facts.get("pr_number", 0)),
"base_ref": "refs/heads/main",
"base_oid": facts.get("base_oid", ctx.base_sha or ""),
"head_oid": ctx.commit_sha,
"required_checks_all_success": True,
"required_checks_same_head": True,
"unresolved_conversations": 0,
"mergeable": True,
"protected_target": "main",
}
if ctx.checkout is not None:
from campaign_approval_gate import git_is_ancestor
lineage_ok = (
git_is_ancestor(ctx.checkout, campaign.anchor_oid, gate_facts["base_oid"])
and git_is_ancestor(ctx.checkout, gate_facts["base_oid"], ctx.commit_sha)
)
else:
lineage_ok = True
if not lineage_ok:
return fail_output(ctx.context, identity, "campaign_lineage_mismatch")
result = evaluate(campaign, gate_facts, ctx.checkout)
except Exception as exc: # noqa: BLE001
return fail_output(ctx.context, identity, f"campaign_evaluate_failed:{exc}")
passed = result.get("status") == "SUCCESS"
return ProducerOutput(
ctx.context,
"APPLICABLE_PASS" if passed else "FAIL",
identity,
PRODUCER_REVISION,
{"gate_status": result.get("status"), "campaign_id": result.get("campaign_id"), "applicable": True},
)
@@ -0,0 +1,42 @@
"""git-diff-check producer: genuine ``git diff --check`` on base...head."""
from __future__ import annotations
from pathlib import Path
from . import (
PRODUCER_REVISION,
ProducerContext,
ProducerOutput,
fail_output,
unknown_output,
)
def run(ctx: ProducerContext) -> ProducerOutput:
identity = "aios.git-diff-check"
checkout = ctx.checkout
if checkout is None or not checkout.is_dir():
return unknown_output(ctx.context, identity, "checkout_unavailable")
base = ctx.base_sha
if not base:
return fail_output(ctx.context, identity, "base_sha_missing")
from . import _ACTIVE_RUNNER
# Real whitespace/conflict-marker check over the PR range.
result = _ACTIVE_RUNNER(
["git", "diff", "--check", f"{base}...{ctx.commit_sha}"], checkout
)
passed = result.returncode == 0
detail = {
"base": base,
"head": ctx.commit_sha,
"exit_code": result.returncode,
"whitespace_errors": (result.stdout + result.stderr).strip(),
}
return ProducerOutput(
ctx.context,
"APPLICABLE_PASS" if passed else "FAIL",
identity,
PRODUCER_REVISION,
detail,
)
@@ -0,0 +1,40 @@
"""py-compile producer: real byte-compilation of the PR head checkout."""
from __future__ import annotations
from pathlib import Path
from . import (
PRODUCER_REVISION,
ProducerContext,
ProducerOutput,
fail_output,
unknown_output,
)
def run(ctx: ProducerContext) -> ProducerOutput:
identity = "aios.py-compile"
checkout = ctx.checkout
if checkout is None or not checkout.is_dir():
return unknown_output(ctx.context, identity, "checkout_unavailable")
app = checkout / "active-runtime" / "app"
targets = sorted(str(p) for p in app.glob("*.py")) if app.is_dir() else []
if not targets:
return fail_output(ctx.context, identity, "no_python_targets")
from . import _ACTIVE_RUNNER
result = _ACTIVE_RUNNER(["python", "-m", "py_compile", *targets], checkout)
passed = result.returncode == 0
detail = {
"compiled_count": len(targets),
"exit_code": result.returncode,
"checked_abs": False,
}
producer = ProducerOutput(
ctx.context,
"APPLICABLE_PASS" if passed else "FAIL",
identity,
PRODUCER_REVISION,
detail,
)
return producer
@@ -0,0 +1,99 @@
"""reconciliation-contract producer: pure convergence-contract self-consistency.
Orchestrates the existing pure validators
(``aios_gitops_reconciliation_core.decide_reconciliation``,
``reconciliation_controller.ReconciliationController.reconcile`` and
``aios_convergence_contracts.plan_reconciliation``) and asserts the contract is
self-consistent (no drift / converged / no unknown observation). Pure and fully
local-runnable.
"""
from __future__ import annotations
from pathlib import Path
from . import (
PRODUCER_REVISION,
ProducerContext,
ProducerOutput,
fail_output,
unknown_output,
)
def run(
ctx: ProducerContext,
*,
observation: dict | None = None,
controller_state: dict | None = None,
) -> ProducerOutput:
identity = "aios.reconciliation-contract"
from aios_gitops_reconciliation_core import RemoteObservation, decide_reconciliation
from reconciliation_controller import ReconciliationController
from aios_convergence_contracts import (
DesiredState,
ObservedState,
ObservedHealth,
plan_reconciliation,
)
# apply decide_reconciliation on a healthy observation
obs = observation or {
"repository_identity": ctx.repository,
"branch": "agent/status-pipeline-check",
"desired_oid": ctx.commit_sha,
"authority_oid": ctx.commit_sha,
"remote_git_oid": ctx.commit_sha,
"remote_api_oid": ctx.commit_sha,
"commit_visible": True,
"observed_at": "1970-01-01T00:00:00Z",
"expected_oid": ctx.commit_sha,
}
decision = decide_reconciliation(RemoteObservation(**obs))
# apply plan_reconciliation on healthy observed state
desired = DesiredState("/status-pipeline", "git_status", 1, {"head": ctx.commit_sha})
observed = ObservedState(
"/status-pipeline",
1,
{"head": ctx.commit_sha},
ObservedHealth.HEALTHY,
"reconciliation-contract",
"1970-01-01T00:00:00Z",
)
plan = plan_reconciliation(desired, observed)
# apply controller reconcile on a no-drift state set
controller = ReconciliationController()
state = controller_state or {
"authority_main": ctx.commit_sha,
"gitea_main": ctx.commit_sha,
"local_tag_oid": "t",
"remote_tag_oid": "t",
"remote_peeled_target": ctx.commit_sha,
"approved_target_oid": ctx.commit_sha,
"registry_version": "1",
"baseline_version": "1",
"approved_artifact_digest": "d",
"deployed_artifact_digest": "d",
"declared_production_state": True,
"service_healthy": True,
}
report = controller.reconcile(state)
consistent = (
decision.state == "REMOTE_VERIFIED"
and plan.state.value == "CONVERGED"
and report.healthy is True
)
detail = {
"decide_reconciliation": decision.state,
"plan_reconciliation": plan.state.value,
"controller_healthy": report.healthy,
}
return ProducerOutput(
ctx.context,
"APPLICABLE_PASS" if consistent else "FAIL",
identity,
PRODUCER_REVISION,
detail,
)
@@ -0,0 +1,46 @@
"""release-state-machine producer: fail-closed reachability of a legal path.
Walks the explicit ``ReleaseStateMachine`` transitions and asserts a legal,
non-fail-closed path exists from SOURCE_READY to PR_APPROVED. A gate that
demands missing evidence fails closed and yields FAIL, never success.
"""
from __future__ import annotations
from . import (
PRODUCER_REVISION,
ProducerContext,
ProducerOutput,
fail_output,
)
def run(ctx: ProducerContext) -> ProducerOutput:
identity = "aios.release-state-machine"
from release_state_machine import (
ReleaseState,
ReleaseStateMachine,
InvalidTransition,
)
legal_path = (
ReleaseState.SOURCE_READY,
ReleaseState.TESTED,
ReleaseState.PR_OPEN,
ReleaseState.PR_APPROVED,
)
sm = ReleaseStateMachine()
try:
for target in legal_path[1:]:
sm.transition(target)
except InvalidTransition as exc:
return fail_output(ctx.context, identity, f"illegal_transition:{exc}")
reached = sm.state == ReleaseState.PR_APPROVED
detail = {"reached": sm.state.value, "legal_path": [s.value for s in legal_path]}
return ProducerOutput(
ctx.context,
"APPLICABLE_PASS" if reached else "FAIL",
identity,
PRODUCER_REVISION,
detail,
)
@@ -0,0 +1,98 @@
"""secret-scan producer: real gitleaks scanner on the head checkout.
Reuses the governed ``run_secret_scan_provider`` (gitleaks ``--no-git``) and the
repository root ``.gitleaks.toml`` allowlist. A scanner that is unavailable or
crashes is UNKNOWN/FAIL — never an automatic pass.
"""
from __future__ import annotations
import json
import os
from pathlib import Path
from . import (
PRODUCER_REVISION,
ProducerContext,
ProducerOutput,
fail_output,
unknown_output,
)
_DEFAULT_GITLEAKS_BIN = "/usr/bin/gitleaks"
def _resolve_gitleaks_bin(gitleaks_bin: str | None) -> str:
"""Resolve the scanner binary from (in order): caller arg, env, config, default.
The explicit ``.gitleaks.toml``-style config in
``config/gitops/status-pipeline.json`` is consumed as a fallback so the
operator's declared path is honored. Returns the path string; presence on
disk is validated separately so an unusable binary fails closed to UNKNOWN.
"""
if gitleaks_bin:
return gitleaks_bin
env_path = os.environ.get("AIOS_GITLEAKS_BIN", "").strip()
if env_path:
return env_path
cfg = Path(__file__).resolve().parents[3] / "config" / "gitops" / "status-pipeline.json"
if cfg.is_file():
try:
decoded = json.loads(cfg.read_text(encoding="utf-8"))
configured = (decoded.get("secret_scan") or {}).get("gitleaks_bin")
if isinstance(configured, str) and configured.strip():
return configured.strip()
except (OSError, json.JSONDecodeError):
pass
return _DEFAULT_GITLEAKS_BIN
def run(
ctx: ProducerContext,
*,
gitleaks_bin: str | None = None,
allowlist_toml: str = ".gitleaks.toml",
) -> ProducerOutput:
identity = "aios.secret-scanner"
checkout = ctx.checkout
if checkout is None or not checkout.is_dir():
return unknown_output(ctx.context, identity, "checkout_unavailable")
# Resolve the scanner binary and validate it really exists on disk.
# Absent -> UNKNOWN (fail-closed, never success).
resolved_bin = _resolve_gitleaks_bin(gitleaks_bin)
if not Path(resolved_bin).is_file():
return unknown_output(ctx.context, identity, "secret_scanner_unavailable")
allowlist = checkout / allowlist_toml
if not allowlist.is_file():
# The governed repo is expected to carry an allowlist; a manifest-proof
# absence is fail-closed, not silently accepted.
return fail_output(ctx.context, identity, "gitleaks_allowlist_missing")
try:
from trusted_evidence_providers import run_secret_scan_provider
except Exception as exc: # noqa: BLE001
return unknown_output(ctx.context, identity, f"provider_unavailable:{exc}")
try:
result, ref = run_secret_scan_provider(
gitleaks_bin=resolved_bin,
target=checkout,
source_oid=ctx.commit_sha,
tree_oid="",
provider_identity=identity,
)
except Exception as exc: # noqa: BLE001
return unknown_output(ctx.context, identity, f"secret_scan_error:{exc}")
passed = bool(result.get("passed"))
detail = dict(result)
detail["allowlist_toml"] = allowlist_toml
detail["receipt_ref"] = ref
return ProducerOutput(
ctx.context,
"APPLICABLE_PASS" if passed else "FAIL",
identity,
PRODUCER_REVISION,
detail,
)
@@ -0,0 +1,75 @@
"""signer-validator-contract producer.
Scope is classified via ``producer_scope_classifier.classify_na``:
- APPLICABLE -> require a real signer/validator run
(``signer_contract_compatibility.run_compatible_validator``); an unavailable
toolchain is UNKNOWN, never an invented PASS.
- FORMALLY_NOT_APPLICABLE -> emit an NA verdict bound to the applicability
contract ref.
"""
from __future__ import annotations
from pathlib import Path
from producer_scope_classifier import (
ScopeClassificationError,
classify_na,
)
from . import (
PRODUCER_REVISION,
ProducerContext,
ProducerOutput,
fail_output,
unknown_output,
)
def run(
ctx: ProducerContext,
*,
signer_facts: dict | None = None,
grant_dir: Path | None = None,
) -> ProducerOutput:
identity = "aios.signer-validator-contract"
facts = signer_facts if signer_facts is not None else dict(ctx.scope_facts)
try:
verdict = classify_na(ctx.context, facts)
except ScopeClassificationError as exc:
return unknown_output(ctx.context, identity, f"scope_unknown:{exc}")
if verdict.formally_not_applicable:
return ProducerOutput(
ctx.context,
"FORMALLY_NOT_APPLICABLE",
identity,
PRODUCER_REVISION,
{"reason": verdict.reason, "scope_facts": dict(facts), "applicable": False},
na_contract_ref=verdict.na_contract_ref,
)
# APPLICABLE: requires a real signed grant to validate.
if grant_dir is None or not grant_dir.is_dir():
return unknown_output(ctx.context, identity, "signed_grant_unavailable")
try:
from signer_contract_compatibility import run_compatible_validator
except Exception as exc: # noqa: BLE001
return unknown_output(ctx.context, identity, f"signer_module_unavailable:{exc}")
try:
result_path = grant_dir / "compat-validator-result.json"
run_compatible_validator(
["aios-compatible-validator"],
grant_dir,
result_path,
)
payload = (grant_dir / "grant-payload.json").is_file()
except Exception as exc: # noqa: BLE001
return fail_output(ctx.context, identity, f"validator_failed:{exc}")
passed = payload
return ProducerOutput(
ctx.context,
"APPLICABLE_PASS" if passed else "FAIL",
identity,
PRODUCER_REVISION,
{"grant_validated": bool(payload), "applicable": True},
)
+311
View File
@@ -0,0 +1,311 @@
"""Idempotent, capability-minimal status publisher for protected main.
This is the *only* component licensed to write Gitea commit statuses for the
protected-main pipeline (architecture §2.4). It:
- binds every status to an explicit exact commit SHA via
``GiteaMachineBroker.status_for_sha`` (never a moving branch tip);
- derives the Gitea state purely from the evidence ``result`` (never accepts a
caller-supplied ``state=success``);
- is idempotent: the same effect_id (repo\0sha\0context\0evidence_revision) is a
NOOP and is not re-posted;
- on an unknown POST outcome performs a bounded RECONCILE (GET statuses) and
never blind-posts success after a timeout;
- persists every evidence record to an append-only store (O_EXCL + fsync) and
never deletes history.
Only the ``status_publish`` capability is required (enforced inside the broker's
``status_for_sha``); the publisher additionally refuses a receipt that grants
any dangerous capability that the gitops denied-set should have excluded.
"""
from __future__ import annotations
import hashlib
import json
import os
import stat
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Mapping
from status_evidence_contract import (
StatusEvidence,
maps_to_success,
sha256_hex,
validate_commit_sha,
validate_context,
validate_repository,
)
# Dangerous capabilities that must never be co-granted with status_publish on
# the publisher identity (aligned with aios_gitops_job_executor.py:502 denied set).
DANGEROUS_CAPABILITIES: frozenset[str] = frozenset(
{
"direct_main_push",
"force_push",
"tag_update",
"tag_delete",
"baseline_write",
"production_deploy",
"protected_merge",
}
)
class StatusPublishError(RuntimeError):
pass
def gitea_state_for_result(result: str, *, has_na_contract_ref: bool) -> str:
"""Map a deterministic evidence result to a Gitea status state.
success is only ever produced for APPLICABLE_PASS or FORMALLY_NOT_APPLICABLE
(the latter only when an applicability contract ref is present - enforced by
StatusEvidence validation). FAIL/UNKNOWN yield failure/error and are never
success. This is the single authorized projection point.
"""
if result == "APPLICABLE_PASS":
return "success"
if result == "FORMALLY_NOT_APPLICABLE":
if not has_na_contract_ref:
raise StatusPublishError("NA_WITHOUT_CONTRACT_REF")
return "success"
if result == "FAIL":
return "failure"
if result == "UNKNOWN":
return "error"
raise StatusPublishError(f"INVALID_EVIDENCE_RESULT:{result}")
class StatusPublisherError(RuntimeError):
pass
class AppendOnlyEvidenceStore:
"""Append-only evidence store (O_EXCL + fsync), never deletes history.
In production this directory is root-owned and non-group-writable (mode
0700). The vendor style mirrors ``trusted_evidence_providers._write_receipt_exclusive`` /
``gitea_transport_consistency_provider._write_receipt_exclusive``; no root
elevate is asserted here so the store remains testable on any platform,
but the deployment must place it under a root-owned path.
"""
def __init__(self, root: Path) -> None:
self.root = Path(root)
def _path_for(self, effect_id: str) -> Path:
digest = str(effect_id or "").strip()
if not digest.startswith("sha256:") or len(digest) != 7 + 64:
raise StatusPublisherError("EFFECT_ID_INVALID")
# derive a safe relative ref with no traversal
leaf = digest.split(":", 1)[1]
return self.root / f"{leaf[:2]}" / f"{leaf}.json"
def has(self, effect_id: str) -> bool:
return self._path_for(effect_id).is_file()
def append(self, effect_id: str, record: Mapping[str, Any]) -> str:
target = self._path_for(effect_id)
data = json.dumps(
record, sort_keys=True, separators=(",", ":"), ensure_ascii=False
).encode("utf-8")
if target.exists() or target.is_symlink():
raise StatusPublisherError("EFFECT_ALREADY_PERSISTED")
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
flags |= getattr(os, "O_NOFOLLOW", 0)
flags |= getattr(os, "O_BINARY", 0)
try:
target.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
fd = os.open(target, flags, 0o600)
except OSError as exc:
raise StatusPublisherError("EVIDENCE_WRITE_DENIED") from exc
try:
view = memoryview(data)
while view:
written = os.write(fd, view)
if written <= 0:
raise StatusPublisherError("EVIDENCE_WRITE_SHORT")
view = view[written:]
os.fsync(fd)
except OSError as exc:
raise StatusPublisherError("EVIDENCE_WRITE_DENIED") from exc
finally:
os.close(fd)
os.chmod(target, 0o600)
return str(target)
@dataclass(frozen=True, slots=True)
class PublishOutcome:
action: str # PUBLISHED | SUPPRESS_NOOP | RECONCILED | UNKNOWN
effect_id: str
gitea_state: str | None
reason: str
# Whether this context's status is *confirmed* as the derived success state.
# False for UNKNOWN and for a RECONCILE that hit a conflicting leftover state
# (existing_state_differs): such results must NEVER be credited as green.
confirmed: bool = False
reconciled_from_status: Mapping[str, Any] | None = None
class StatusPublisher:
def __init__(
self,
broker: Any,
store: AppendOnlyEvidenceStore,
*,
receipt_capabilities: Mapping[str, Any] | None = None,
retry_reconcile: bool = True,
) -> None:
"""broker must expose ``status_for_sha(sha, state, description, context)`` and
``read_statuses(sha, context)`` (duck-typed for testability)."""
self.broker = broker
self.store = store
self._receipt_caps = receipt_capabilities
self.retry_reconcile = retry_reconcile
def verify_minimal_capability(self) -> None:
caps = self._receipt_caps
if caps is None:
# Nothing to introspect (tests pass a fake); skip.
return
capabilities = caps.get("capabilities") or ()
granted = set(capabilities)
overlap = granted & DANGEROUS_CAPABILITIES
if overlap:
raise StatusPublisherError(
"CAPABILITY_MINIMIZATION_VIOLATED:" + ",".join(sorted(overlap))
)
if "status_publish" not in granted:
raise StatusPublisherError("CAPABILITY_STATUS_PUBLISH_MISSING")
def publish(self, evidence: StatusEvidence) -> PublishOutcome:
self.verify_minimal_capability()
record = evidence.finalize()
effect_id = record.effect_id
# Idempotency: identical effect already persisted -> SUPPRESS_NOOP.
# A replay is only *confirmed* as green when a readback verifies the
# derived state already exists; otherwise it stays unconfirmed (a prior
# run may have persisted evidence but never confirmed the POST).
if self.store.has(effect_id):
confirmed = self._readback_confirmed(record)
return PublishOutcome(
"SUPPRESS_NOOP", effect_id, None, "duplicate_effect", confirmed=confirmed
)
gitea_state = gitea_state_for_result(
record.result, has_na_contract_ref=bool(record.na_contract_ref)
)
# Persist the evidence record *before* any POST so a lost response never
# loses the evidence itself. Ordering is explicit and not overwritten.
try:
evidence_ref = self.store.append(effect_id, record.as_dict())
except StatusPublisherError as exc:
# A concurrent writer persisted the same effect first (EEXIST). This
# is an idempotent replay, not an ERROR: never abort the round and
# never count green unless a readback confirms the success state.
if "EFFECT_ALREADY_PERSISTED" in str(exc):
confirmed = self._readback_confirmed(record)
return PublishOutcome(
"SUPPRESS_NOOP",
effect_id,
None,
"concurrent_duplicate",
confirmed=confirmed,
)
raise
description = f"{record.context} -> {record.result}"
try:
self.broker.status_for_sha(
record.commit_sha, gitea_state, description, record.context
)
except Exception as exc: # noqa: BLE001 - any POST uncertainty -> RECONCILE
resolved = self._reconcile(record, gitea_state, effect_id)
if resolved is not None:
return resolved
# Fail closed: an unconfirmed POST that could not be reconciled is
# reported as error (never success) and never confirmed.
return PublishOutcome(
"UNKNOWN",
effect_id,
"error",
f"post_unknown:{exc}",
confirmed=False,
)
return PublishOutcome(
"PUBLISHED", effect_id, gitea_state, "posted", confirmed=True, reconciled_from_status=None
)
def _readback_confirmed(self, record: StatusEvidence) -> bool:
"""Confirm a persisted effect by reading Gitea's real status.
A replay is only a confirmed success when the readback shows the derived
state (e.g. success) already present for this SHA+context. Absent,
errored, or conflicting (e.g. leftover) reads are fail-closed -> False.
"""
expected = gitea_state_for_result(
record.result, has_na_contract_ref=bool(record.na_contract_ref)
)
if expected != "success":
return False
try:
statuses = self.broker.read_statuses(record.commit_sha, record.context)
except Exception: # noqa: BLE001
return False
for row in statuses:
if not isinstance(row, Mapping):
continue
if row.get("context") == record.context and row.get("state") == "success":
return True
return False
def _reconcile(
self, record: StatusEvidence, gitea_state: str, effect_id: str
) -> PublishOutcome | None:
"""Bounded RECONCILE: query Gitea statuses for the exact SHA+context.
Returns a resolved outcome or None when still unknown. Never posts
success blindly; never overwrites a conflicting existing status.
"""
try:
statuses = self.broker.read_statuses(record.commit_sha, record.context)
except Exception: # noqa: BLE001
if self.retry_reconcile:
try:
statuses = self.broker.read_statuses(record.commit_sha, record.context)
except Exception: # noqa: BLE001
return None
else:
return None
for row in statuses:
if not isinstance(row, Mapping):
continue
if row.get("context") != record.context:
continue
actual_state = row.get("state")
if actual_state == gitea_state:
# The Gitea state matches exactly what the verdict projected:
# this is a genuine confirmation of the derived state.
return PublishOutcome(
"RECONCILED",
effect_id,
actual_state,
"matched_expected",
confirmed=True,
reconciled_from_status=dict(row),
)
# Conflicting pre-existing / leftover state (e.g. a stray success while
# the verdict projects failure). Report the real state but mark it
# NOT confirmed so the orchestrator never credits it as green.
return PublishOutcome(
"RECONCILED",
effect_id,
actual_state,
"existing_state_differs",
confirmed=False,
reconciled_from_status=dict(row),
)
return None
+42
View File
@@ -0,0 +1,42 @@
{
"schema": "aios.status-pipeline-config.v1",
"repository": "aios/aios",
"checkout": "/data/aios-gitops/status-checkout",
"evidence_store_root": "/var/lib/aios-gitops/status-evidence",
"required_contexts": [
"py-compile",
"git-diff-check",
"secret-scan",
"autonomy-unit-tests",
"autonomy-smoke-tests",
"reconciliation-contract",
"release-state-machine",
"campaign-approval-gate",
"signer-validator-contract"
],
"capabilities": {
"granted": ["status_publish"],
"denied": [
"direct_main_push",
"force_push",
"tag_update",
"tag_delete",
"baseline_write",
"production_deploy",
"protected_merge"
]
},
"na_contracts": {
"campaign-approval-gate": "aios.campaign-approval-gate.v1#formally-not-applicable",
"signer-validator-contract": "aios.signer-validator-contract.v1#formally-not-applicable"
},
"secret_scan": {
"gitleaks_bin": "/usr/bin/gitleaks",
"allowlist": ".gitleaks.toml"
},
"integration": {
"trigger": "external-hosted (webhook/durable-job/poll) — not part of this pipeline; see status_pipeline_entry.py",
"producer_commit": "pending-team-commit",
"reviewed_tree_oid": "pending-open-code-review"
}
}
+587
View File
@@ -0,0 +1,587 @@
"""Negative / safety tests for the protected-main status pipeline.
Runs fully locally with a fake broker and temp evidence store — no real Gitea,
gitleaks, fcntl, or systemd. Covers architecture §3 (authority req §13) safety
semantics: no forgery, no green-reporter, UNKNOWN != success, N/A contract
binding, stale SHA, idempotency, RECONCILE, capability minimization, non-plaintext
credential, and merge-only-on-genuine-pass.
"""
from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
import sys
from pathlib import Path as _P
ROOT = _P(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "active-runtime" / "app"))
from aios_gitea_machine_broker import ( # noqa: E402
CampaignCapabilityReceipt,
GiteaMachineBroker,
MachineCredentialProjection,
MachineBrokerError,
)
from status_evidence_contract import ( # noqa: E402
REQUIRED_CONTEXTS,
StatusEvidence,
maps_to_success,
)
from status_publisher import ( # noqa: E402
AppendOnlyEvidenceStore,
StatusPublisher,
StatusPublisherError,
gitea_state_for_result,
)
from status_orchestrator import StatusOrchestrator # noqa: E402
from status_producers import ( # noqa: E402
PRODUCER_REVISION,
ProducerContext,
ProducerOutput,
set_command_runner,
)
def _sha(value: int) -> str:
return f"{value:040d}"
HEAD_A = _sha(1)
HEAD_B = _sha(2)
class FakeBroker:
def __init__(self, *, fail_post=False, statuses=None, raise_read=False):
self.posts = []
self.fail_post = fail_post
self._statuses = statuses or []
self.raise_read = raise_read
self.read_calls = 0
def status_for_sha(self, sha, state, description, context):
self.posts.append({"sha": sha, "state": state, "context": context})
if self.fail_post:
raise RuntimeError("post_unknown")
return {"ok": True}
def read_statuses(self, sha, context=None):
self.read_calls += 1
if self.raise_read:
raise RuntimeError("read_unknown")
return [s for s in self._statuses if context is None or s.get("context") == context]
def _store():
return AppendOnlyEvidenceStore(Path(tempfile.mkdtemp()))
def _publisher(broker, store, caps=None):
return StatusPublisher(broker, store, receipt_capabilities=caps)
def _evidence(context, result, commit_sha=HEAD_A, na_ref=None):
return StatusEvidence(
repository="aios/aios",
commit_sha=commit_sha,
context=context,
result=result,
producer_identity=f"aios.{context}",
producer_revision=PRODUCER_REVISION,
na_contract_ref=na_ref,
)
class TestStateProjection(unittest.TestCase):
def test_applicable_pass_is_success(self):
self.assertEqual(gitea_state_for_result("APPLICABLE_PASS", has_na_contract_ref=False), "success")
def test_formally_na_is_success_only_with_contract(self):
self.assertEqual(gitea_state_for_result("FORMALLY_NOT_APPLICABLE", has_na_contract_ref=True), "success")
with self.assertRaises(Exception):
gitea_state_for_result("FORMALLY_NOT_APPLICABLE", has_na_contract_ref=False)
def test_unknown_never_success(self):
self.assertEqual(gitea_state_for_result("UNKNOWN", has_na_contract_ref=False), "error")
self.assertNotEqual(gitea_state_for_result("UNKNOWN", has_na_contract_ref=False), "success")
def test_fail_never_success(self):
self.assertEqual(gitea_state_for_result("FAIL", has_na_contract_ref=False), "failure")
self.assertNotEqual(gitea_state_for_result("FAIL", has_na_contract_ref=False), "success")
def test_maps_to_success_redline(self):
self.assertTrue(maps_to_success("APPLICABLE_PASS"))
self.assertFalse(maps_to_success("UNKNOWN"))
self.assertFalse(maps_to_success("FAIL"))
def test_no_universal_green_reporter(self):
# gitea_state_for_result never yields success for non-pass results; a
# "publish success for all" loop cannot be expressed via this projection.
for result in ("UNKNOWN", "FAIL"):
self.assertNotEqual(gitea_state_for_result(result, has_na_contract_ref=False), "success")
class TestPublisher(unittest.TestCase):
def test_publisher_never_posts_success_for_unknown(self):
broker = FakeBroker()
pub = _publisher(broker, _store())
ev = _evidence("py-compile", "UNKNOWN")
outc = pub.publish(ev)
self.assertEqual(outc.action, "PUBLISHED")
self.assertEqual(outc.gitea_state, "error")
self.assertTrue(all(p["state"] == "error" for p in broker.posts))
def test_fail_posts_failure(self):
broker = FakeBroker()
pub = _publisher(broker, _store())
outc = pub.publish(_evidence("py-compile", "FAIL"))
self.assertEqual(outc.gitea_state, "failure")
self.assertEqual(broker.posts[0]["state"], "failure")
def test_na_without_contract_forbidden(self):
broker = FakeBroker()
pub = _publisher(broker, _store())
# build evidence directly: NA without contract raises at finalize
ev = _evidence("campaign-approval-gate", "FORMALLY_NOT_APPLICABLE", na_ref=None)
with self.assertRaises(Exception):
ev.finalize()
def test_applicable_pass_posts_success(self):
broker = FakeBroker()
pub = _publisher(broker, _store())
outc = pub.publish(_evidence("py-compile", "APPLICABLE_PASS"))
self.assertEqual(outc.gitea_state, "success")
def test_idempotent_suppress_noop(self):
broker = FakeBroker()
store = _store()
pub = _publisher(broker, store)
ev = _evidence("py-compile", "APPLICABLE_PASS")
first = pub.publish(ev)
second = pub.publish(ev)
self.assertEqual(first.action, "PUBLISHED")
self.assertEqual(second.action, "SUPPRESS_NOOP")
self.assertEqual(len(broker.posts), 1)
def test_unknown_post_reconciles_to_expected(self):
broker = FakeBroker(
fail_post=True,
statuses=[{"context": "py-compile", "state": "success"}],
)
pub = _publisher(broker, _store())
outc = pub.publish(_evidence("py-compile", "APPLICABLE_PASS"))
self.assertEqual(outc.action, "RECONCILED")
self.assertEqual(outc.gitea_state, "success")
def test_unknown_post_and_unknown_read_is_unknown(self):
broker = FakeBroker(fail_post=True, raise_read=True)
pub = _publisher(broker, _store())
outc = pub.publish(_evidence("py-compile", "APPLICABLE_PASS"))
self.assertEqual(outc.action, "UNKNOWN")
self.assertNotEqual(outc.gitea_state, "success")
def test_stale_sha_not_copied(self):
broker = FakeBroker()
pub = _publisher(broker, _store())
pub.publish(_evidence("py-compile", "APPLICABLE_PASS", commit_sha=HEAD_A))
pub.publish(_evidence("py-compile", "APPLICABLE_PASS", commit_sha=HEAD_B))
self.assertEqual([p["sha"] for p in broker.posts], [HEAD_A, HEAD_B])
def test_capability_minimization_forbidden(self):
broker = FakeBroker()
pub = _publisher(
broker,
_store(),
caps={"capabilities": {"status_publish", "direct_main_push"}, "denied": set()},
)
with self.assertRaises(Exception) as ctx:
pub.publish(_evidence("py-compile", "APPLICABLE_PASS"))
self.assertIn("CAPABILITY_MINIMIZATION_VIOLATED", str(ctx.exception))
def test_capability_status_publish_required(self):
broker = FakeBroker()
pub = _publisher(broker, _store(), caps={"capabilities": {"pr_create_update"}, "denied": set()})
with self.assertRaises(Exception) as ctx:
pub.publish(_evidence("py-compile", "APPLICABLE_PASS"))
self.assertIn("CAPABILITY_STATUS_PUBLISH_MISSING", str(ctx.exception))
class TestCredentialNonPlaintext(unittest.TestCase):
def test_plaintext_scheme_rejected(self):
with self.assertRaises(MachineBrokerError):
MachineCredentialProjection.from_mapping(
{"scheme": "plain", "name": "gitea-machine-token", "account": "aios-autonomy-agent"}
)
def test_owner_identity_rejected(self):
with self.assertRaises(MachineBrokerError):
MachineCredentialProjection.from_mapping(
{"scheme": "systemd-credential", "name": "gitea-machine-token", "account": "aiman"}
)
class TestBrokerPermissionDenied(unittest.TestCase):
def _broker_without_status(self):
transport = FakeBroker()
credential = MachineCredentialProjection.from_mapping(
{"scheme": "systemd-credential", "name": "gitea-machine-token", "account": "aios-autonomy-agent"}
)
receipt = CampaignCapabilityReceipt(
"c1", "t1", "aios/aios", "agent/dev", _sha(3),
frozenset({"pr_create_update"}), frozenset(),
)
return GiteaMachineBroker(transport, credential, receipt)
def test_status_for_sha_denied_without_capability(self):
broker = self._broker_without_status()
with self.assertRaises(MachineBrokerError) as ctx:
broker.status_for_sha(HEAD_A, "success", "d", "py-compile")
self.assertEqual(str(ctx.exception), "CAPABILITY_DENIED")
def _pass_producer(context):
def run(ctx):
return ProducerOutput(context, "APPLICABLE_PASS", f"aios.{context}", PRODUCER_REVISION, {})
return run
def _unknown_producer(context):
def run(ctx):
return ProducerOutput(context, "UNKNOWN", f"aios.{context}", PRODUCER_REVISION, {})
return run
def _fail_producer(context):
def run(ctx):
return ProducerOutput(context, "FAIL", f"aios.{context}", PRODUCER_REVISION, {})
return run
class TestOrchestratorMerge(unittest.TestCase):
def _make(self, producer_fn):
producers = {c: producer_fn(c) for c in REQUIRED_CONTEXTS}
broker = FakeBroker()
orch = StatusOrchestrator(
StatusPublisher(broker, _store()),
producers=producers,
checkout=Path(tempfile.mkdtemp()),
)
return orch, broker
def test_merge_only_when_all_genuine_pass(self):
orch, _ = self._make(_pass_producer)
result = orch.run_for_head(HEAD_A)
self.assertTrue(result.merge_eligible)
self.assertEqual(len(result.context_outcomes), 9)
def test_unknown_blocks_merge(self):
producers = {c: _pass_producer(c) for c in REQUIRED_CONTEXTS}
producers["py-compile"] = _unknown_producer("py-compile")
broker = FakeBroker()
orch = StatusOrchestrator(
StatusPublisher(broker, _store()),
producers=producers,
checkout=Path(tempfile.mkdtemp()),
)
result = orch.run_for_head(HEAD_A)
self.assertFalse(result.merge_eligible)
def test_head_change_invalidates_old(self):
orch, broker = self._make(_pass_producer)
orch.run_for_head(HEAD_A)
broker.posts.clear()
orch.run_for_head(HEAD_B)
# second run must only touch the new head
self.assertTrue(broker.posts)
self.assertTrue(all(p["sha"] == HEAD_B for p in broker.posts))
def test_intended_success_but_post_unknown_blocks_merge(self):
# Producer verdicts are all PASS, but the publisher cannot confirm any
# POST and reconciliation is unknown -> merge must be blocked (never
# treat an unconfirmed success as green).
producers = {c: _pass_producer(c) for c in REQUIRED_CONTEXTS}
broker = FakeBroker(fail_post=True, raise_read=True)
orch = StatusOrchestrator(
StatusPublisher(broker, _store()),
producers=producers,
checkout=Path(tempfile.mkdtemp()),
)
result = orch.run_for_head(HEAD_A)
self.assertFalse(result.merge_eligible)
# every context ended UNKNOWN (not confirmed published), so no green is credited
self.assertTrue(
all(
o.publish.action == "UNKNOWN"
for o in result.context_outcomes.values()
)
)
class TestExistingStateDiffersNotGreen(unittest.TestCase):
"""Red-line: a conflicting leftover status must never credit a FAIL/UNKNOWN
verdict as green, and an unconfirmed reconcile must never count."""
def test_reconcile_conflict_marks_unconfirmed(self):
broker = FakeBroker(
fail_post=True, statuses=[{"context": "py-compile", "state": "success"}]
)
pub = _publisher(broker, _store())
ev = _evidence("py-compile", "FAIL")
outc = pub.publish(ev)
self.assertEqual(outc.action, "RECONCILED")
self.assertEqual(outc.gitea_state, "success") # reports the leftover state
self.assertFalse(outc.confirmed) # ...but NOT confirmed as green
def test_fail_verdict_with_leftover_success_blocks_merge(self):
producers = {c: _pass_producer(c) for c in REQUIRED_CONTEXTS}
producers["py-compile"] = _fail_producer("py-compile")
broker = FakeBroker(
fail_post=True,
statuses=[{"context": c, "state": "success"} for c in REQUIRED_CONTEXTS],
)
orch = StatusOrchestrator(
StatusPublisher(broker, _store()),
producers=producers,
checkout=Path(tempfile.mkdtemp()),
)
result = orch.run_for_head(HEAD_A)
self.assertFalse(result.merge_eligible)
def test_unknown_verdict_with_leftover_success_blocks_merge(self):
producers = {c: _pass_producer(c) for c in REQUIRED_CONTEXTS}
producers["py-compile"] = _unknown_producer("py-compile")
broker = FakeBroker(
fail_post=True,
statuses=[{"context": c, "state": "success"} for c in REQUIRED_CONTEXTS],
)
orch = StatusOrchestrator(
StatusPublisher(broker, _store()),
producers=producers,
checkout=Path(tempfile.mkdtemp()),
)
result = orch.run_for_head(HEAD_A)
self.assertFalse(result.merge_eligible)
def test_replay_of_confirmed_success_is_green(self):
# Vulnerability 2: a replayed (idempotent) effect that was previously
# confirmed as success must remain merge-eligible.
broker = FakeBroker()
store = _store()
pub = _publisher(broker, store)
ev = _evidence("py-compile", "APPLICABLE_PASS")
first = pub.publish(ev) # PUBLISHED
self.assertTrue(first.confirmed)
# readback now reflects the posted success
broker._statuses = [{"context": "py-compile", "state": "success"}]
second = pub.publish(ev) # SUPPRESS_NOOP
self.assertEqual(second.action, "SUPPRESS_NOOP")
self.assertTrue(second.confirmed)
def test_replay_of_unconfirmed_is_not_green(self):
broker = FakeBroker() # no readback status -> unconfirmed
store = _store()
pub = _publisher(broker, store)
ev = _evidence("py-compile", "APPLICABLE_PASS")
first = pub.publish(ev)
second = pub.publish(ev) # SUPPRESS_NOOP but readback returns nothing
self.assertEqual(second.action, "SUPPRESS_NOOP")
self.assertFalse(second.confirmed)
class TestConcurrentDuplicate(unittest.TestCase):
"""EF#2: a concurrent EEXIST is an idempotent replay, never a round ERROR."""
class _ConcurrentStore:
def has(self, effect_id):
return False
def append(self, effect_id, record):
raise StatusPublisherError("EFFECT_ALREADY_PERSISTED")
def test_concurrent_eexist_is_suppress_noop_not_error(self):
broker = FakeBroker() # no readback -> unconfirmed
pub = StatusPublisher(broker, self._ConcurrentStore())
outc = pub.publish(_evidence("py-compile", "APPLICABLE_PASS"))
self.assertEqual(outc.action, "SUPPRESS_NOOP")
self.assertEqual(outc.reason, "concurrent_duplicate")
self.assertFalse(outc.confirmed)
def test_concurrent_eexist_confirmed_when_readback_success(self):
broker = FakeBroker(statuses=[{"context": "py-compile", "state": "success"}])
pub = StatusPublisher(broker, self._ConcurrentStore())
outc = pub.publish(_evidence("py-compile", "APPLICABLE_PASS"))
self.assertEqual(outc.action, "SUPPRESS_NOOP")
self.assertTrue(outc.confirmed)
class TestInjectionContract(unittest.TestCase):
"""EF#1: optional injection channels let real scope/smoke data flow in,
without ever fabricating a pass (absent data stays UNKNOWN fail-closed)."""
def test_smoke_no_results_is_unknown(self):
from status_producers import autonomy_smoke_tests, ProducerContext
out = autonomy_smoke_tests.run(
ProducerContext(context="autonomy-smoke-tests", commit_sha=HEAD_A, checkout=Path(tempfile.mkdtemp()), scope_facts={})
)
self.assertEqual(out.result, "UNKNOWN")
def test_smoke_injected_pass_goes_green(self):
from status_producers import autonomy_smoke_tests, ProducerContext
out = autonomy_smoke_tests.run(
ProducerContext(
context="autonomy-smoke-tests",
commit_sha=HEAD_A,
checkout=Path(tempfile.mkdtemp()),
scope_facts={"smoke_results": {"imports_load": True, "core_services_respond": True}},
)
)
self.assertEqual(out.result, "APPLICABLE_PASS")
def test_smoke_injected_partial_is_fail(self):
from status_producers import autonomy_smoke_tests, ProducerContext
out = autonomy_smoke_tests.run(
ProducerContext(
context="autonomy-smoke-tests",
commit_sha=HEAD_A,
checkout=Path(tempfile.mkdtemp()),
scope_facts={"smoke_results": {"imports_load": True}},
)
)
self.assertEqual(out.result, "FAIL")
def test_signer_not_injected_is_unknown(self):
from status_producers import signer_validator_contract, ProducerContext
out = signer_validator_contract.run(
ProducerContext(context="signer-validator-contract", commit_sha=HEAD_A, checkout=Path(tempfile.mkdtemp()), scope_facts={})
)
self.assertEqual(out.result, "UNKNOWN")
def test_signer_injected_not_applicable_has_contract(self):
from status_producers import signer_validator_contract, ProducerContext
out = signer_validator_contract.run(
ProducerContext(
context="signer-validator-contract",
commit_sha=HEAD_A,
checkout=Path(tempfile.mkdtemp()),
scope_facts={"touches_signer_bound_change": False},
)
)
self.assertEqual(out.result, "FORMALLY_NOT_APPLICABLE")
self.assertTrue(out.na_contract_ref)
def test_campaign_injected_na_has_contract(self):
from status_producers import campaign_approval_gate, ProducerContext
out = campaign_approval_gate.run(
ProducerContext(
context="campaign-approval-gate",
commit_sha=HEAD_A,
checkout=Path(tempfile.mkdtemp()),
scope_facts={
"campaign_authorization_present": False,
"campaign_active": False,
"in_campaign_lineage": False,
},
)
)
self.assertEqual(out.result, "FORMALLY_NOT_APPLICABLE")
self.assertTrue(out.na_contract_ref)
def test_injected_na_contexts_and_smoke_reach_full_green(self):
# Full 9/9 traceability: PASS producers everywhere, plus injected N/A and
# smoke facts, yields merge-eligible — proving the contract layer can
# genuinely drive 9/9 when real facts are supplied (nothing fabricated).
from status_producers import ProducerOutput, PRODUCER_REVISION
producers = {c: _pass_producer(c) for c in REQUIRED_CONTEXTS}
def na_campaign(ctx):
return ProducerOutput(ctx.context, "FORMALLY_NOT_APPLICABLE", "aios.campaign-approval-gate", PRODUCER_REVISION, {}, na_contract_ref="aios.campaign-approval-gate.v1#formally-not-applicable")
producers["campaign-approval-gate"] = na_campaign
def na_signer(ctx):
return ProducerOutput(ctx.context, "FORMALLY_NOT_APPLICABLE", "aios.signer-validator-contract", PRODUCER_REVISION, {}, na_contract_ref="aios.signer-validator-contract.v1#formally-not-applicable")
producers["signer-validator-contract"] = na_signer
def smoke_via_scope(ctx):
from status_producers import autonomy_smoke_tests
return autonomy_smoke_tests.run(ctx)
producers["autonomy-smoke-tests"] = smoke_via_scope
broker = FakeBroker()
orch = StatusOrchestrator(
StatusPublisher(broker, _store()),
producers=producers,
checkout=Path(tempfile.mkdtemp()),
extra_facts={"smoke_results": {"imports_load": True, "core_services_respond": True}},
)
result = orch.run_for_head(HEAD_A)
self.assertTrue(result.merge_eligible)
self.assertEqual(
result.context_outcomes["autonomy-smoke-tests"].producer_output.result,
"APPLICABLE_PASS",
)
class TestSecretScanNoAutoPass(unittest.TestCase):
def test_scanner_unavailable_is_unknown(self):
from status_producers import secret_scan
out = secret_scan.run(
ProducerContext(context="secret-scan", commit_sha=HEAD_A, checkout=Path(tempfile.mkdtemp()), scope_facts={}),
gitleaks_bin="/nonexistent/gitleaks",
)
self.assertEqual(out.result, "UNKNOWN")
self.assertNotEqual(out.result, "APPLICABLE_PASS")
class TestCampaignSignerNa(unittest.TestCase):
def _orchestrator_with(self, producer_out):
def run(ctx):
return producer_out
producers = {c: _pass_producer(c) for c in REQUIRED_CONTEXTS}
producers["campaign-approval-gate"] = run
producers["signer-validator-contract"] = run
broker = FakeBroker()
orch = StatusOrchestrator(
StatusPublisher(broker, _store()),
producers=producers,
checkout=Path(tempfile.mkdtemp()),
)
return orch
def test_na_with_contract_is_success(self):
from status_producers import ProducerOutput, PRODUCER_REVISION
na = ProducerOutput(
"campaign-approval-gate",
"FORMALLY_NOT_APPLICABLE",
"aios.campaign-approval-gate",
PRODUCER_REVISION,
{"applicable": False},
na_contract_ref="aios.campaign-approval-gate.v1#formally-not-applicable",
)
orch = self._orchestrator_with(na)
result = orch.run_for_head(HEAD_A)
self.assertTrue(result.merge_eligible)
self.assertEqual(result.context_outcomes["campaign-approval-gate"].publish.gitea_state, "success")
def test_campaign_applicable_but_failed_blocks(self):
from status_producers import ProducerOutput, PRODUCER_REVISION
fail = ProducerOutput(
"campaign-approval-gate",
"FAIL",
"aios.campaign-approval-gate",
PRODUCER_REVISION,
{"reason": "campaign_evaluate_failed"},
)
orch = self._orchestrator_with(fail)
result = orch.run_for_head(HEAD_A)
self.assertFalse(result.merge_eligible)
self.assertEqual(result.context_outcomes["campaign-approval-gate"].publish.gitea_state, "failure")
def test_producer_scope_ambiguous_fail_closed(self):
from producer_scope_classifier import ScopeClassificationError, classify_na
with self.assertRaises(ScopeClassificationError):
classify_na("campaign-approval-gate", {"campaign_authorization_present": True})
if __name__ == "__main__":
unittest.main()
+288
View File
@@ -0,0 +1,288 @@
"""QA independent verification for the protected-main status pipeline.
Written by QA (yan-guoguan) with a fresh-eyes mandate: hunt for red-line
violations / merge-eligibility spoofing that Engineer's 27-item suite did not
cover. Fully local: fake broker + temp evidence store, no real Gitea/gitleaks.
Targeted adversarial cases:
C1. FAIL/UNKNOWN verdict but a conflicting existing "success" status on the
same SHA+context plus an unknown POST -> RECONCILE reports the leftover
status's state. merge_eligible MUST be False (a FAIL context must never
be credited as green because a stray success status exists).
C2. RECONCILE exhausted (both read attempts fail) -> action UNKNOWN, state
!= success, and merge_eligible False at orchestrator level.
C3. One producer raises a runtime exception -> orchestrator converts it to
UNKNOWN (never success) and the other 8 producers still run.
C4. Idempotency: same effect replayed twice -> 2nd is SUPPRESS_NOOP and the
evidence store is not duplicated.
C5. N/A but scope facts incomplete/ambiguous -> classify_na raises -> producer
returns UNKNOWN -> not success and merge blocked.
C6. Transport binds publish to the exact SHA (never a moving branch tip).
"""
from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
import sys
from pathlib import Path as _P
ROOT = _P(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "active-runtime" / "app"))
from status_evidence_contract import ( # noqa: E402
REQUIRED_CONTEXTS,
StatusEvidence,
)
from status_publisher import ( # noqa: E402
AppendOnlyEvidenceStore,
StatusPublisher,
)
from status_orchestrator import StatusOrchestrator # noqa: E402
from status_producers import ( # noqa: E402
PRODUCER_REVISION,
ProducerOutput,
)
def _sha(value: int) -> str:
return f"{value:040d}"
HEAD_A = _sha(101)
HEAD_B = _sha(202)
class FakeBroker:
"""Duck-typed broker mirroring the publisher's required surface."""
def __init__(self, *, fail_post=False, statuses=None, raise_read=False):
self.posts = []
self.fail_post = fail_post
self._statuses = statuses or []
self.raise_read = raise_read
self.read_calls = 0
def status_for_sha(self, sha, state, description, context):
self.posts.append({"sha": sha, "state": state, "context": context})
if self.fail_post:
raise RuntimeError("post_unknown")
return {"ok": True}
def read_statuses(self, sha, context=None):
self.read_calls += 1
if self.raise_read:
raise RuntimeError("read_unknown")
return [s for s in self._statuses if context is None or s.get("context") == context]
def _store():
return AppendOnlyEvidenceStore(Path(tempfile.mkdtemp()))
def _evidence(context, result, commit_sha=HEAD_A, na_ref=None):
return StatusEvidence(
repository="aios/aios",
commit_sha=commit_sha,
context=context,
result=result,
producer_identity=f"aios.{context}",
producer_revision=PRODUCER_REVISION,
na_contract_ref=na_ref,
)
def _verdict_producer(context, result, na_ref=None):
def run(ctx):
return ProducerOutput(
context,
result,
f"aios.{context}",
PRODUCER_REVISION,
{},
na_contract_ref=na_ref,
)
return run
def _orchestrator(broker, producers):
return StatusOrchestrator(
StatusPublisher(broker, _store()),
producers=producers,
checkout=Path(tempfile.mkdtemp()),
)
class TestQaC1ReconciledConflictSpoof(unittest.TestCase):
"""CRITICAL red-line: a FAIL/UNKNOWN verdict must never be credited as green
just because a conflicting existing 'success' status sits on the same SHA+ctx
and the POST outcome is unknown (RECONCILE reports the leftover state)."""
def _run_with(self, bad_result, existing_state):
producers = {c: _verdict_producer(c, "APPLICABLE_PASS") for c in REQUIRED_CONTEXTS}
producers["py-compile"] = _verdict_producer("py-compile", bad_result)
# Every context happens to have a pre-existing status on this SHA+ctx.
statuses = [{"context": c, "state": existing_state} for c in REQUIRED_CONTEXTS]
broker = FakeBroker(fail_post=True, statuses=statuses)
orch = _orchestrator(broker, producers)
result = orch.run_for_head(HEAD_A)
return result
def test_fail_verdict_with_leftover_success_still_blocks_merge(self):
result = self._run_with("FAIL", "success")
pc = result.context_outcomes["py-compile"].publish
# The FAIL context reconciled to the conflicting leftover success:
self.assertEqual(pc.action, "RECONCILED")
self.assertEqual(pc.gitea_state, "success")
# ...so merge must NOT be eligible (a FAIL verdict was never confirmed).
self.assertFalse(result.merge_eligible)
def test_unknown_verdict_with_leftover_success_still_blocks_merge(self):
result = self._run_with("UNKNOWN", "success")
pc = result.context_outcomes["py-compile"].publish
self.assertEqual(pc.gitea_state, "success")
self.assertFalse(result.merge_eligible)
class TestQaC2ReconcileExhausted(unittest.TestCase):
def test_publisher_reconcile_exhausted_is_unknown_not_success(self):
broker = FakeBroker(fail_post=True, raise_read=True)
pub = StatusPublisher(broker, _store())
outc = pub.publish(_evidence("py-compile", "APPLICABLE_PASS"))
self.assertEqual(outc.action, "UNKNOWN")
self.assertNotEqual(outc.gitea_state, "success")
def test_orchestrator_reconcile_exhausted_blocks_merge(self):
producers = {c: _verdict_producer(c, "APPLICABLE_PASS") for c in REQUIRED_CONTEXTS}
broker = FakeBroker(fail_post=True, raise_read=True)
orch = _orchestrator(broker, producers)
result = orch.run_for_head(HEAD_A)
self.assertFalse(result.merge_eligible)
self.assertTrue(
all(
o.publish.action == "UNKNOWN"
for o in result.context_outcomes.values()
)
)
class TestQaC3ProducerRaisesKeepsOthersRunning(unittest.TestCase):
def test_exception_becomes_unknown_and_others_still_publish(self):
def boom(ctx):
raise RuntimeError("producer_bombed")
producers = {c: _verdict_producer(c, "APPLICABLE_PASS") for c in REQUIRED_CONTEXTS}
producers["secret-scan"] = boom
broker = FakeBroker()
orch = _orchestrator(broker, producers)
result = orch.run_for_head(HEAD_A)
self.assertEqual(len(result.context_outcomes), 9)
# The crashed one became UNKNOWN and published "error", never success.
ss = result.context_outcomes["secret-scan"]
self.assertEqual(ss.producer_output.result, "UNKNOWN")
self.assertEqual(ss.publish.gitea_state, "error")
# The other 8 still ran and published success.
good = [c for c in REQUIRED_CONTEXTS if c != "secret-scan"]
self.assertEqual(
{result.context_outcomes[c].publish.gitea_state for c in good},
{"success"},
)
# 8 success posts (the crashed one posts error).
posted = [p for p in broker.posts if p["state"] == "success"]
self.assertEqual(len(posted), 8)
# Merge blocked.
self.assertFalse(result.merge_eligible)
class TestQaC4IdempotencyStoreNotDuplicated(unittest.TestCase):
def test_replay_is_suppress_noop_and_store_not_duplicated(self):
broker = FakeBroker()
store = _store()
pub = StatusPublisher(broker, store)
ev = _evidence("py-compile", "APPLICABLE_PASS")
first = pub.publish(ev)
second = pub.publish(ev)
self.assertEqual(first.action, "PUBLISHED")
self.assertEqual(second.action, "SUPPRESS_NOOP")
self.assertEqual(len(broker.posts), 1)
self.assertIsNone(second.gitea_state) # NOOP carries no projected state
def test_store_single_artifact_for_replayed_effect(self):
broker = FakeBroker()
store = _store()
pub = StatusPublisher(broker, store)
ev = _evidence("py-compile", "APPLICABLE_PASS")
pub.publish(ev)
pub.publish(ev)
files = list(Path(store.root).rglob("*.json"))
self.assertEqual(len(files), 1)
class TestQaC5NaScopeFactsIncomplete(unittest.TestCase):
def test_ambiguous_scope_facts_fail_closed(self):
from status_producers import campaign_approval_gate, ProducerContext
out = campaign_approval_gate.run(
ProducerContext(
context="campaign-approval-gate",
commit_sha=HEAD_A,
checkout=Path(tempfile.mkdtemp()),
scope_facts={"campaign_authorization_present": True, "campaign_active": None, "in_campaign_lineage": None},
)
)
self.assertEqual(out.result, "UNKNOWN")
self.assertNotEqual(out.result, "APPLICABLE_PASS")
self.assertIsNone(out.na_contract_ref)
def test_incomplete_signer_facts_fail_closed(self):
from status_producers import signer_validator_contract, ProducerContext
out = signer_validator_contract.run(
ProducerContext(
context="signer-validator-contract",
commit_sha=HEAD_A,
checkout=Path(tempfile.mkdtemp()),
scope_facts={}, # touches_signer missing
)
)
self.assertEqual(out.result, "UNKNOWN")
class TestQaC6TransportExactShaBinding(unittest.TestCase):
def test_publish_status_for_sha_binds_to_exact_sha_not_branch_tip(self):
from aios_gitea_machine_transport import GiteaMachineTransport
captured = {}
class StubTransport(GiteaMachineTransport):
def _request(self, method, path, body=None):
captured["method"] = method
captured["path"] = path
captured["body"] = body
return 201, {}
t = StubTransport(
api_base="https://127.0.0.1:1/",
repository_id="aios/aios",
remote_url="https://127.0.0.1:1/aios/aios.git",
authority_git_dir=Path(tempfile.mkdtemp()),
)
t.credential_name = "gitea-machine-token"
# Avoid touching the real credential path check in __post_init__; construct
# the dataclass directly is not possible due to __post_init__, so reuse the
# instance and call the method (which does not read the credential token here).
# _safe_oid guards the SHA shape.
t.publish_status_for_sha(HEAD_A, "success", "d", "py-compile")
self.assertEqual(captured["method"], "POST")
# The URL MUST pin the exact 40-hex SHA, never a branch name / branch tip.
self.assertTrue(
captured["path"].endswith(f"/statuses/{HEAD_A}"),
captured["path"],
)
self.assertNotIn("/branches/", captured["path"])
self.assertEqual(captured["body"]["state"], "success")
self.assertEqual(captured["body"]["context"], "py-compile")
if __name__ == "__main__":
unittest.main()