fix(gitops): centralize implementation branch policy

active-runtime/app/aios_authority_git_executor.py

@@ -11,6 +11,8 @@ from dataclasses import dataclass
 import os
 from pathlib import Path
 import re
+
+from aios_target_branch_policy import validate_target_branch
 import subprocess
 import tempfile
 from typing import Mapping, Sequence
@@ -95,7 +97,8 @@ class AuthorityGitExecutor:
         return path
 
     def advance_task_branch_CAS(self, branch: str, expected_oid: str, new_oid: str) -> str:
-        if not TASK_BRANCH_RE.fullmatch(branch):
+        branch_path = branch[len("refs/heads/"):] if branch.startswith("refs/heads/") else ""
+        if not branch.startswith("refs/heads/") or not validate_target_branch(branch_path):
             raise AuthorityGitError("TASK_BRANCH_SCOPE_DENIED")
         _oid(expected_oid, "expected_oid")
         _oid(new_oid, "new_oid")

active-runtime/app/aios_chatgpt_bridge_v2_server.py

@@ -9,6 +9,8 @@ import os
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 import re
+
+from aios_target_branch_policy import validate_target_branch
 import sqlite3
 from typing import Any, Mapping
 
@@ -210,8 +212,7 @@ class GovernedRuntime:
         if (
             not task_id
             or repository != self.config.repository
-            or not branch.startswith("agent/")
-            or branch in {"main", "master"}
+            or not validate_target_branch(branch)
             or branch_mode not in {"CREATE_NEW", "CONTINUE_EXISTING"}
             or generation < 1
             or not self._valid_oid(base)
@@ -287,7 +288,7 @@ class GovernedRuntime:
         for value in self.bindings.values():
             task_id = str(value.get("task_id") or "").strip()
             branch = str(value.get("target_branch") or "").strip()
-            if not task_id or not branch.startswith("agent/"):
+            if not task_id or not validate_target_branch(branch):
                 continue
             authority = self.authority.read_ref("refs/heads/" + branch)
             try:

active-runtime/app/aios_continuation_lease.py

@@ -10,6 +10,8 @@ from __future__ import annotations
 from dataclasses import dataclass, replace
 from datetime import datetime, timedelta, timezone
 import re
+
+from aios_target_branch_policy import validate_target_branch
 from typing import Protocol
 
 
@@ -60,7 +62,7 @@ class WriterLeaseRequestV2:
             raise ContinuationError("branch_mode_invalid")
         _oid(self.base_oid, "base_oid")
         _oid(self.expected_branch_oid, "expected_branch_oid")
-        if self.target_branch in {"main", "master"} or self.target_branch.startswith(("refs/", "tags/")):
+        if not validate_target_branch(self.target_branch):
             raise ContinuationError("protected_branch_rejected")
         if not isinstance(self.generation, int) or isinstance(self.generation, bool) or self.generation < 1:
             raise ContinuationError("generation_invalid")

active-runtime/app/aios_credential_broker.py

@@ -8,6 +8,8 @@ from pathlib import Path
 import secrets
 from typing import Any, Callable, Mapping
 
+from aios_target_branch_policy import validate_target_branch
+
 
 @dataclass(frozen=True, slots=True)
 class CredentialLease:
@@ -44,7 +46,7 @@ class CredentialBroker:
         self._leases: dict[str, CredentialLease] = {}
 
     def issue(self, *, task_id: str, repository: str, target_branch: str, generation: int, scopes: tuple[str, ...] = ("git:push", "gitea:read"), lease_seconds: int = 900) -> dict[str, Any]:
-        if repository != "aios/aios" or not target_branch.startswith("agent/") or generation < 1 or not 1 <= lease_seconds <= 86400:
+        if repository != "aios/aios" or not validate_target_branch(target_branch) or generation < 1 or not 1 <= lease_seconds <= 86400:
             raise ValueError("CREDENTIAL_LEASE_BINDING_INVALID")
         now = self.now().astimezone(timezone.utc)
         lease = CredentialLease("lease-" + secrets.token_hex(16), task_id, repository, target_branch, generation, now.isoformat(), (now + timedelta(seconds=lease_seconds)).isoformat(), tuple(scopes), type(self.provider).__name__)

active-runtime/app/aios_existing_branch_approval.py

@@ -1,352 +0,0 @@
-"""Cryptographic Owner-approval boundary for governed existing-branch publication.
-
-Only a complete signed envelope from an authoritative trust-root adapter is
-accepted.  References, chat assertions, and locally supplied keys are not an
-approval.  Missing public-key or revocation evidence fails closed.
-"""
-from __future__ import annotations
-
-import base64
-from dataclasses import dataclass
-from datetime import datetime, timezone
-import hashlib
-import json
-import os
-from pathlib import Path
-import re
-from typing import Any, Mapping, Protocol
-
-from cryptography.exceptions import InvalidSignature
-from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
-
-
-APPROVAL_SCHEMA = "aios.existing-branch-owner-approval.v1"
-PUBLISH_OPERATION = "publishExistingBranchGoverned"
-OID_RE = re.compile(r"^[0-9a-f]{40}$")
-DIGEST_RE = re.compile(r"^[0-9a-f]{64}$")
-IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._@:+/-]{0,254}$")
-BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
-
-
-class OwnerApprovalRejected(RuntimeError):
-    """Stable, audit-safe error for a rejected Owner approval."""
-
-    def __init__(self, code: str) -> None:
-        super().__init__(code)
-        self.code = code
-
-
-def canonical_json(value: Any) -> bytes:
-    """Encode a signed payload without implementation-dependent spacing."""
-    try:
-        encoded = json.dumps(
-            value,
-            sort_keys=True,
-            separators=(",", ":"),
-            ensure_ascii=False,
-            allow_nan=False,
-        )
-    except (TypeError, ValueError) as exc:
-        raise OwnerApprovalRejected("OWNER_APPROVAL_CANONICALIZATION_REJECTED") from exc
-    return (encoded + "\n").encode("utf-8")
-
-
-def digest_json(value: Any) -> str:
-    return hashlib.sha256(canonical_json(value)).hexdigest()
-
-
-def _require_identifier(field: str, value: object) -> str:
-    if not isinstance(value, str) or not IDENTIFIER_RE.fullmatch(value):
-        raise OwnerApprovalRejected(f"OWNER_APPROVAL_{field.upper()}_REJECTED")
-    return value
-
-
-def _require_digest(field: str, value: object) -> str:
-    if not isinstance(value, str) or not DIGEST_RE.fullmatch(value):
-        raise OwnerApprovalRejected(f"OWNER_APPROVAL_{field.upper()}_REJECTED")
-    return value
-
-
-def _require_oid(field: str, value: object) -> str:
-    if not isinstance(value, str) or not OID_RE.fullmatch(value):
-        raise OwnerApprovalRejected(f"OWNER_APPROVAL_{field.upper()}_REJECTED")
-    return value
-
-
-def _parse_utc(field: str, value: object) -> datetime:
-    if not isinstance(value, str):
-        raise OwnerApprovalRejected(f"OWNER_APPROVAL_{field.upper()}_REJECTED")
-    try:
-        instant = datetime.fromisoformat(value.replace("Z", "+00:00"))
-    except ValueError as exc:
-        raise OwnerApprovalRejected(f"OWNER_APPROVAL_{field.upper()}_REJECTED") from exc
-    if instant.tzinfo is None:
-        raise OwnerApprovalRejected(f"OWNER_APPROVAL_{field.upper()}_REJECTED")
-    return instant.astimezone(timezone.utc)
-
-
-@dataclass(frozen=True, slots=True)
-class PublicationApprovalBinding:
-    repository: str
-    branch: str
-    expected_branch_oid: str
-    operation: str
-    request_digest: str
-    idempotency_digest: str
-
-    def validate(self) -> "PublicationApprovalBinding":
-        _require_identifier("repository", self.repository)
-        if not BRANCH_RE.fullmatch(self.branch):
-            raise OwnerApprovalRejected("OWNER_APPROVAL_BRANCH_REJECTED")
-        _require_oid("expected_branch_oid", self.expected_branch_oid)
-        if self.operation != PUBLISH_OPERATION:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_OPERATION_REJECTED")
-        _require_digest("request_digest", self.request_digest)
-        _require_digest("idempotency_digest", self.idempotency_digest)
-        return self
-
-    @property
-    def as_dict(self) -> dict[str, str]:
-        return {
-            "repository": self.repository,
-            "branch": self.branch,
-            "expected_branch_oid": self.expected_branch_oid,
-            "operation": self.operation,
-            "request_digest": self.request_digest,
-            "idempotency_digest": self.idempotency_digest,
-        }
-
-
-@dataclass(frozen=True, slots=True)
-class OwnerApprovalEnvelope:
-    approval_id: str
-    repository: str
-    branch: str
-    expected_branch_oid: str
-    operation: str
-    issued_at: str
-    not_before: str
-    expires_at: str
-    nonce: str
-    owner_identity: str
-    trust_domain: str
-    key_id: str
-    request_digest: str
-    idempotency_digest: str
-    signature: str
-
-    @property
-    def canonical_payload(self) -> dict[str, str]:
-        return {
-            "schema": APPROVAL_SCHEMA,
-            "approval_id": self.approval_id,
-            "repository": self.repository,
-            "branch": self.branch,
-            "expected_branch_oid": self.expected_branch_oid,
-            "operation": self.operation,
-            "issued_at": self.issued_at,
-            "not_before": self.not_before,
-            "expires_at": self.expires_at,
-            "nonce": self.nonce,
-            "owner_identity": self.owner_identity,
-            "trust_domain": self.trust_domain,
-            "key_id": self.key_id,
-            "request_digest": self.request_digest,
-            "idempotency_digest": self.idempotency_digest,
-        }
-
-    def canonical_bytes(self) -> bytes:
-        return canonical_json(self.canonical_payload)
-
-    @property
-    def fingerprint(self) -> str:
-        return hashlib.sha256(self.canonical_bytes()).hexdigest()
-
-    @property
-    def binding(self) -> PublicationApprovalBinding:
-        return PublicationApprovalBinding(
-            repository=self.repository,
-            branch=self.branch,
-            expected_branch_oid=self.expected_branch_oid,
-            operation=self.operation,
-            request_digest=self.request_digest,
-            idempotency_digest=self.idempotency_digest,
-        )
-
-    @classmethod
-    def from_mapping(cls, value: object) -> "OwnerApprovalEnvelope":
-        if not isinstance(value, Mapping):
-            raise OwnerApprovalRejected("OWNER_APPROVAL_ENVELOPE_REQUIRED")
-        expected_fields = {
-            "schema", "approval_id", "repository", "branch", "expected_branch_oid",
-            "operation", "issued_at", "not_before", "expires_at", "nonce",
-            "owner_identity", "trust_domain", "key_id", "request_digest",
-            "idempotency_digest", "signature",
-        }
-        if set(value) != expected_fields or value.get("schema") != APPROVAL_SCHEMA:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_ENVELOPE_REJECTED")
-        envelope = cls(
-            approval_id=_require_identifier("approval_id", value["approval_id"]),
-            repository=_require_identifier("repository", value["repository"]),
-            branch=str(value["branch"]),
-            expected_branch_oid=_require_oid("expected_branch_oid", value["expected_branch_oid"]),
-            operation=str(value["operation"]),
-            issued_at=str(value["issued_at"]),
-            not_before=str(value["not_before"]),
-            expires_at=str(value["expires_at"]),
-            nonce=_require_identifier("nonce", value["nonce"]),
-            owner_identity=_require_identifier("owner_identity", value["owner_identity"]),
-            trust_domain=_require_identifier("trust_domain", value["trust_domain"]),
-            key_id=_require_identifier("key_id", value["key_id"]),
-            request_digest=_require_digest("request_digest", value["request_digest"]),
-            idempotency_digest=_require_digest("idempotency_digest", value["idempotency_digest"]),
-            signature=str(value["signature"]),
-        )
-        envelope.binding.validate()
-        _parse_utc("issued_at", envelope.issued_at)
-        _parse_utc("not_before", envelope.not_before)
-        _parse_utc("expires_at", envelope.expires_at)
-        if not envelope.signature:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_SIGNATURE_REJECTED")
-        return envelope
-
-
-@dataclass(frozen=True, slots=True)
-class TrustRootEvidence:
-    """Public-key and revocation evidence from the authoritative trust root."""
-
-    key_id: str
-    owner_identity: str
-    trust_domain: str
-    public_key: bytes | None
-    authority_confirmed: bool
-    revocation_state: str
-
-
-class AuthoritativeTrustRootAdapter(Protocol):
-    def resolve(
-        self, *, key_id: str, owner_identity: str, trust_domain: str
-    ) -> TrustRootEvidence | None: ...
-
-
-class ReplayLedger(Protocol):
-    def consume(
-        self, approval: OwnerApprovalEnvelope, binding: PublicationApprovalBinding
-    ) -> None: ...
-
-
-class DurableNonceReplayLedger:
-    """Single-use nonce storage that remains conservative across restarts."""
-
-    def __init__(self, root: Path) -> None:
-        self.root = Path(root)
-
-    def consume(
-        self, approval: OwnerApprovalEnvelope, binding: PublicationApprovalBinding
-    ) -> None:
-        binding.validate()
-        if self.root.is_symlink():
-            raise OwnerApprovalRejected("OWNER_APPROVAL_REPLAY_LEDGER_REJECTED")
-        try:
-            self.root.mkdir(parents=True, exist_ok=True)
-            if not self.root.is_dir() or self.root.is_symlink():
-                raise OwnerApprovalRejected("OWNER_APPROVAL_REPLAY_LEDGER_REJECTED")
-            name = hashlib.sha256(approval.nonce.encode("utf-8")).hexdigest()
-            record_path = self.root / name
-            flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
-            if hasattr(os, "O_NOFOLLOW"):
-                flags |= os.O_NOFOLLOW
-            descriptor = os.open(record_path, flags, 0o600)
-        except FileExistsError as exc:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_REPLAY_REJECTED") from exc
-        except OwnerApprovalRejected:
-            raise
-        except OSError as exc:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_REPLAY_LEDGER_REJECTED") from exc
-        record = {
-            "schema": "aios.existing-branch-owner-approval-replay.v1",
-            "nonce_sha256": name,
-            "approval_fingerprint": approval.fingerprint,
-            "binding_digest": digest_json(binding.as_dict),
-        }
-        try:
-            with os.fdopen(descriptor, "wb", closefd=True) as stream:
-                stream.write(canonical_json(record))
-                stream.flush()
-                os.fsync(stream.fileno())
-            directory = os.open(self.root, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
-            try:
-                os.fsync(directory)
-            finally:
-                os.close(directory)
-        except OSError as exc:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_REPLAY_LEDGER_REJECTED") from exc
-
-
-class OwnerApprovalVerifier:
-    def __init__(
-        self,
-        trust_root: AuthoritativeTrustRootAdapter,
-        *,
-        allowed_trust_domains: set[str] | frozenset[str],
-        maximum_lifetime_seconds: int = 900,
-    ) -> None:
-        if not allowed_trust_domains or maximum_lifetime_seconds < 1:
-            raise ValueError("OWNER_APPROVAL_VERIFIER_CONFIGURATION_REJECTED")
-        self.trust_root = trust_root
-        self.allowed_trust_domains = frozenset(allowed_trust_domains)
-        self.maximum_lifetime_seconds = maximum_lifetime_seconds
-
-    def verify(
-        self,
-        value: object,
-        binding: PublicationApprovalBinding,
-        *,
-        now: datetime | None = None,
-    ) -> OwnerApprovalEnvelope:
-        binding.validate()
-        approval = OwnerApprovalEnvelope.from_mapping(value)
-        if approval.binding != binding:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_BINDING_REJECTED")
-        instant = datetime.now(timezone.utc) if now is None else now
-        if instant.tzinfo is None:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_TIME_REJECTED")
-        instant = instant.astimezone(timezone.utc)
-        issued_at = _parse_utc("issued_at", approval.issued_at)
-        not_before = _parse_utc("not_before", approval.not_before)
-        expires_at = _parse_utc("expires_at", approval.expires_at)
-        if not issued_at <= not_before <= expires_at:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_TIME_REJECTED")
-        if not not_before <= instant <= expires_at:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_TIME_REJECTED")
-        if (expires_at - issued_at).total_seconds() > self.maximum_lifetime_seconds:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_TIME_REJECTED")
-        if approval.trust_domain not in self.allowed_trust_domains:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_TRUST_DOMAIN_REJECTED")
-        evidence = self.trust_root.resolve(
-            key_id=approval.key_id,
-            owner_identity=approval.owner_identity,
-            trust_domain=approval.trust_domain,
-        )
-        if evidence is None or not evidence.authority_confirmed:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_TRUST_ROOT_EVIDENCE_MISSING")
-        if (
-            evidence.key_id != approval.key_id
-            or evidence.owner_identity != approval.owner_identity
-            or evidence.trust_domain != approval.trust_domain
-        ):
-            raise OwnerApprovalRejected("OWNER_APPROVAL_TRUST_ROOT_BINDING_REJECTED")
-        if evidence.revocation_state == "REVOKED":
-            raise OwnerApprovalRejected("OWNER_APPROVAL_REVOKED")
-        if evidence.revocation_state != "ACTIVE":
-            raise OwnerApprovalRejected("OWNER_APPROVAL_REVOCATION_EVIDENCE_MISSING")
-        if evidence.public_key is None or len(evidence.public_key) != 32:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_TRUST_ROOT_EVIDENCE_MISSING")
-        try:
-            signature = base64.b64decode(approval.signature, validate=True)
-            Ed25519PublicKey.from_public_bytes(evidence.public_key).verify(
-                signature, approval.canonical_bytes()
-            )
-        except (ValueError, TypeError, InvalidSignature) as exc:
-            raise OwnerApprovalRejected("OWNER_APPROVAL_SIGNATURE_REJECTED") from exc
-        return approval

active-runtime/app/aios_existing_branch_publisher.py

@@ -1,331 +0,0 @@
-"""Narrow governed publisher for an already-existing task branch.
-
-Its only writing port is a compare-and-swap publication after a complete
-signed Owner approval, a stable remote observation, and a single-use nonce
-claim.  It neither owns trust material nor accepts direct reference strings.
-"""
-from __future__ import annotations
-
-from dataclasses import dataclass
-import re
-from typing import Any, Mapping, Protocol
-
-from aios_existing_branch_approval import (
-    OwnerApprovalEnvelope,
-    OwnerApprovalRejected,
-    OwnerApprovalVerifier,
-    PublicationApprovalBinding,
-    ReplayLedger,
-    canonical_json,
-    digest_json,
-)
-
-
-REQUEST_SCHEMA = "aios.existing-branch-publication.v1"
-PUBLISH_OPERATION = "publishExistingBranchGoverned"
-OID_RE = re.compile(r"^[0-9a-f]{40}$")
-BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
-
-
-class ExistingBranchPublicationRejected(RuntimeError):
-    def __init__(self, code: str) -> None:
-        super().__init__(code)
-        self.code = code
-
-
-class RemoteOidConflict(RuntimeError):
-    pass
-
-
-class RemoteObservationAmbiguous(RuntimeError):
-    pass
-
-
-@dataclass(frozen=True, slots=True)
-class ExistingBranchPublicationRequest:
-    repository: str
-    branch: str
-    expected_branch_oid: str
-    idempotency_key: str
-    payload: Mapping[str, Any]
-
-    @classmethod
-    def from_mapping(cls, value: object) -> "ExistingBranchPublicationRequest":
-        if not isinstance(value, Mapping):
-            raise ExistingBranchPublicationRejected("PUBLICATION_REQUEST_REJECTED")
-        expected_fields = {
-            "schema",
-            "repository",
-            "branch",
-            "expected_branch_oid",
-            "operation",
-            "idempotency_key",
-            "payload",
-        }
-        if set(value) != expected_fields or value.get("schema") != REQUEST_SCHEMA:
-            raise ExistingBranchPublicationRejected("PUBLICATION_REQUEST_REJECTED")
-        repository = value["repository"]
-        branch = value["branch"]
-        expected_oid = value["expected_branch_oid"]
-        key = value["idempotency_key"]
-        payload = value["payload"]
-        if (
-            not isinstance(repository, str)
-            or repository != "aios/aios"
-            or not BRANCH_RE.fullmatch(str(branch))
-            or not isinstance(expected_oid, str)
-            or not OID_RE.fullmatch(expected_oid)
-            or value.get("operation") != PUBLISH_OPERATION
-            or not isinstance(key, str)
-            or not key.strip()
-            or not isinstance(payload, Mapping)
-        ):
-            raise ExistingBranchPublicationRejected("PUBLICATION_REQUEST_REJECTED")
-        try:
-            canonical_json(payload)
-        except OwnerApprovalRejected as exc:
-            raise ExistingBranchPublicationRejected("PUBLICATION_PAYLOAD_REJECTED") from exc
-        return cls(repository, str(branch), expected_oid, key, dict(payload))
-
-    @property
-    def request_digest(self) -> str:
-        return digest_json(
-            {
-                "schema": REQUEST_SCHEMA,
-                "repository": self.repository,
-                "branch": self.branch,
-                "expected_branch_oid": self.expected_branch_oid,
-                "operation": PUBLISH_OPERATION,
-                "payload": self.payload,
-            }
-        )
-
-    @property
-    def idempotency_digest(self) -> str:
-        return digest_json(
-            {
-                "idempotency_key": self.idempotency_key,
-                "request_digest": self.request_digest,
-            }
-        )
-
-    @property
-    def approval_binding(self) -> PublicationApprovalBinding:
-        return PublicationApprovalBinding(
-            repository=self.repository,
-            branch=self.branch,
-            expected_branch_oid=self.expected_branch_oid,
-            operation=PUBLISH_OPERATION,
-            request_digest=self.request_digest,
-            idempotency_digest=self.idempotency_digest,
-        )
-
-
-@dataclass(frozen=True, slots=True)
-class RemoteBranchObservation:
-    state: str
-    git_oid: str | None
-    api_oid: str | None
-    expected_commit_visible: bool | None
-
-
-@dataclass(frozen=True, slots=True)
-class RemotePublicationEvidence:
-    authority_oid: str | None
-    gitea_git_oid: str | None
-    gitea_api_oid: str | None
-    commit_visible: bool | None
-
-
-class ExistingBranchRemotePort(Protocol):
-    def observe(self, branch: str) -> RemoteBranchObservation: ...
-
-    def publish_compare_and_swap(
-        self, branch: str, expected_branch_oid: str, commit_oid: str
-    ) -> RemotePublicationEvidence: ...
-
-
-class AuthorityBranchPort(Protocol):
-    def read_branch_oid(self, branch: str) -> str | None: ...
-
-
-class CredentialProjection(Protocol):
-    def project(self) -> None: ...
-
-    def revoke(self) -> None: ...
-
-
-class PublicationWorkspace(Protocol):
-    def head_oid(self) -> str: ...
-
-    def commit_payload(self, payload: Mapping[str, Any]) -> str: ...
-
-
-class DetachedWorkspaceFactory(Protocol):
-    def open_detached(self, expected_branch_oid: str) -> PublicationWorkspace: ...
-
-
-class AuditSink(Protocol):
-    def record(self, event: Mapping[str, Any]) -> None: ...
-
-
-class ExistingBranchPublisher:
-    """Validate once, observe once, then make one bounded publication attempt."""
-
-    def __init__(
-        self,
-        *,
-        verifier: OwnerApprovalVerifier,
-        replay_ledger: ReplayLedger,
-        authority: AuthorityBranchPort,
-        remote: ExistingBranchRemotePort,
-        credentials: CredentialProjection,
-        workspaces: DetachedWorkspaceFactory,
-        audit: AuditSink,
-    ) -> None:
-        self.verifier = verifier
-        self.replay_ledger = replay_ledger
-        self.authority = authority
-        self.remote = remote
-        self.credentials = credentials
-        self.workspaces = workspaces
-        self.audit = audit
-
-    def _audit(
-        self,
-        *,
-        status: str,
-        request: ExistingBranchPublicationRequest,
-        approval: OwnerApprovalEnvelope | None = None,
-        commit_oid: str | None = None,
-    ) -> None:
-        event: dict[str, Any] = {
-            "event": "governed_existing_branch_publication",
-            "status": status,
-            "repository": request.repository,
-            "branch": request.branch,
-            "expected_branch_oid": request.expected_branch_oid,
-            "operation": PUBLISH_OPERATION,
-            "request_digest": request.request_digest,
-            "idempotency_digest": request.idempotency_digest,
-        }
-        if approval is not None:
-            event.update(
-                {
-                    "approval_id": approval.approval_id,
-                    "owner_identity": approval.owner_identity,
-                    "trust_domain": approval.trust_domain,
-                    "approval_fingerprint": approval.fingerprint,
-                }
-            )
-        if commit_oid is not None:
-            event["commit_oid"] = commit_oid
-        self.audit.record(event)
-
-    def _preflight_remote(
-        self, request: ExistingBranchPublicationRequest
-    ) -> RemoteBranchObservation | dict[str, str]:
-        try:
-            authority_oid = self.authority.read_branch_oid(request.branch)
-        except Exception:
-            return {"status": "RECONCILIATION_REQUIRED"}
-        if authority_oid != request.expected_branch_oid:
-            return {"status": "MANUAL_OWNER_REVIEW_REQUIRED"}
-        try:
-            observation = self.remote.observe(request.branch)
-        except RemoteObservationAmbiguous:
-            return {"status": "RECONCILIATION_REQUIRED"}
-        if observation.state == "AMBIGUOUS":
-            return {"status": "RECONCILIATION_REQUIRED"}
-        if observation.state != "PRESENT":
-            return {"status": "MANUAL_OWNER_REVIEW_REQUIRED"}
-        if (
-            observation.git_oid is not None
-            and observation.api_oid is not None
-            and observation.git_oid != observation.api_oid
-        ):
-            return {"status": "RECONCILIATION_REQUIRED"}
-        if (
-            observation.git_oid != request.expected_branch_oid
-            or observation.api_oid != request.expected_branch_oid
-        ):
-            return {"status": "MANUAL_OWNER_REVIEW_REQUIRED"}
-        if observation.expected_commit_visible is not True:
-            return {"status": "RECONCILIATION_REQUIRED"}
-        return observation
-
-    def publish(
-        self, request_value: object, approval_value: object
-    ) -> dict[str, str | bool]:
-        request = ExistingBranchPublicationRequest.from_mapping(request_value)
-        try:
-            approval = self.verifier.verify(approval_value, request.approval_binding)
-        except OwnerApprovalRejected:
-            self._audit(status="REJECTED", request=request)
-            raise
-        preflight = self._preflight_remote(request)
-        if isinstance(preflight, dict):
-            self._audit(status=preflight["status"], request=request, approval=approval)
-            return preflight
-        self.replay_ledger.consume(approval, request.approval_binding)
-        workspace = self.workspaces.open_detached(request.expected_branch_oid)
-        if workspace.head_oid() != request.expected_branch_oid:
-            self._audit(status="REJECTED", request=request, approval=approval)
-            raise ExistingBranchPublicationRejected("DETACHED_WORKSPACE_HEAD_REJECTED")
-        projected = False
-        try:
-            self.credentials.project()
-            projected = True
-            commit_oid = workspace.commit_payload(request.payload)
-            if not isinstance(commit_oid, str) or not OID_RE.fullmatch(commit_oid):
-                raise ExistingBranchPublicationRejected("PUBLICATION_COMMIT_REJECTED")
-            try:
-                evidence = self.remote.publish_compare_and_swap(
-                    request.branch, request.expected_branch_oid, commit_oid
-                )
-            except RemoteOidConflict:
-                self._audit(
-                    status="MANUAL_OWNER_REVIEW_REQUIRED",
-                    request=request,
-                    approval=approval,
-                    commit_oid=commit_oid,
-                )
-                return {"status": "MANUAL_OWNER_REVIEW_REQUIRED"}
-            except RemoteObservationAmbiguous:
-                self._audit(
-                    status="RECONCILIATION_REQUIRED",
-                    request=request,
-                    approval=approval,
-                    commit_oid=commit_oid,
-                )
-                return {"status": "RECONCILIATION_REQUIRED"}
-            if (
-                evidence.authority_oid != commit_oid
-                or evidence.gitea_git_oid != commit_oid
-                or evidence.gitea_api_oid != commit_oid
-                or evidence.commit_visible is not True
-            ):
-                self._audit(
-                    status="RECONCILIATION_REQUIRED",
-                    request=request,
-                    approval=approval,
-                    commit_oid=commit_oid,
-                )
-                return {"status": "RECONCILIATION_REQUIRED"}
-            self._audit(
-                status="PUBLISHED",
-                request=request,
-                approval=approval,
-                commit_oid=commit_oid,
-            )
-            return {
-                "status": "PUBLISHED",
-                "commit_oid": commit_oid,
-                "authority_oid": evidence.authority_oid,
-                "gitea_git_oid": evidence.gitea_git_oid,
-                "gitea_api_oid": evidence.gitea_api_oid,
-                "commit_visible": evidence.commit_visible,
-            }
-        finally:
-            if projected:
-                self.credentials.revoke()

active-runtime/app/aios_gitea_publisher.py

@@ -11,6 +11,7 @@ from urllib.parse import quote
 from urllib.request import Request, urlopen
 
 from aios_gitops_reconciliation_core import RemoteObservation
+from aios_target_branch_policy import validate_target_branch
 
 
 @dataclass(frozen=True, slots=True)
@@ -35,7 +36,7 @@ class GiteaPublisher:
 
     @staticmethod
     def validate_branch(branch: str) -> None:
-        if not isinstance(branch, str) or not branch.startswith("agent/") or branch in {"main", "master"}:
+        if not validate_target_branch(branch):
             raise ValueError("BRANCH_SCOPE_DENIED")
 
     def repository_identity(self) -> str:

active-runtime/app/aios_gitops_job_controller.py

@@ -18,6 +18,8 @@ import tempfile
 from datetime import datetime, timezone
 from typing import Any, Callable, Mapping
 
+from aios_target_branch_policy import validate_target_branch
+
 
 class JobError(RuntimeError):
     pass
@@ -274,7 +276,7 @@ class GitOpsJobController:
             raise JobError("EXPECTED_BRANCH_OID_INVALID")
         if len(digest) != 64 or any(ch not in "0123456789abcdef" for ch in digest):
             raise JobError("PAYLOAD_DIGEST_INVALID")
-        if request["branch"] in {"main", "master"} or request["branch"].startswith(("refs/", "tags/")):
+        if not validate_target_branch(request["branch"]):
             raise JobError("PROTECTED_BRANCH_REJECTED")
         binding = {field: str(request[field]) for field in fields}
         # Operation-specific, non-secret binding inputs must survive the
@@ -383,7 +385,7 @@ class GitOpsJobController:
                     raise JobError("JOB_NOT_FOUND")
                 if int(row["revision"]) != expected_revision:
                     raise JobError("REVISION_CAS_CONFLICT")
-                if row["branch"] in {"main", "master"}:
+                if not validate_target_branch(row["branch"]):
                     raise JobError("PROTECTED_BRANCH_REJECTED")
                 now = utc_now()
                 connection.execute("INSERT INTO branch_leases(lease_id,job_id,branch,lineage_anchor_oid,current_head_oid,generation,lease_expires_at,state,revision) VALUES(?,?,?,?,?,?,?,?,?)", (lease_id, job_id, row["branch"], lineage_anchor_oid, current_head_oid, generation, lease_expires_at, "ACTIVE", 0))

active-runtime/app/aios_gitops_reconciliation_core.py

@@ -4,6 +4,8 @@ from __future__ import annotations
 from dataclasses import dataclass
 from enum import Enum
 import re
+
+from aios_target_branch_policy import validate_target_branch
 from typing import Optional
 
 OID_RE = re.compile(r"^[0-9a-f]{40}$")
@@ -52,7 +54,7 @@ def decide_reconciliation(observation: RemoteObservation) -> ReconciliationDecis
     """Decide from a fresh observation; never complete on a missing remote."""
     if observation.repository_identity != "aios/aios":
         return ReconciliationDecision(ReconciliationAction.FAIL_CONFLICT, "CONFLICT", "REPOSITORY_IDENTITY_MISMATCH")
-    if not observation.branch.startswith("agent/"):
+    if not validate_target_branch(observation.branch):
         return ReconciliationDecision(ReconciliationAction.FAIL_CONFLICT, "CONFLICT", "BRANCH_SCOPE_DENIED")
     if not _valid_oid(observation.desired_oid) or not _valid_oid(observation.expected_oid):
         return ReconciliationDecision(ReconciliationAction.FAIL_CONFLICT, "CONFLICT", "OID_INVALID")

active-runtime/app/aios_governed_policy.py

@@ -3,6 +3,8 @@ from __future__ import annotations
 
 from dataclasses import dataclass
 import re
+
+from aios_target_branch_policy import validate_target_branch
 from typing import Any, Mapping
 
 OID_RE = re.compile(r"^[0-9a-f]{40}$")
@@ -39,7 +41,7 @@ def evaluate_policy(value: GovernedPolicyInput | Mapping[str, Any]) -> PolicyDec
         return PolicyDecision("DENY", "REPOSITORY_IDENTITY_MISMATCH")
     if not OID_RE.fullmatch(value.base_oid) or not OID_RE.fullmatch(value.expected_branch_oid):
         return PolicyDecision("DENY", "OID_INVALID")
-    if not value.target_branch.startswith("agent/") or value.target_branch in {"main", "master"}:
+    if not validate_target_branch(value.target_branch):
         return PolicyDecision("DENY", "BRANCH_SCOPE_DENIED")
     if "force_push" in value.requested_capabilities or "force" in value.operation.lower():
         return PolicyDecision("DENY", "FORCE_PUSH_DENIED")

active-runtime/app/aios_governed_task_bootstrap.py

@@ -11,7 +11,7 @@ from typing import Any, Mapping
 from aios_gitops_job_controller import GitOpsJobController, JobStore
 
 OID_RE = re.compile(r"^[0-9a-f]{40}$")
-BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
+from aios_target_branch_policy import BRANCH_RE, validate_target_branch
 
 
 class BootstrapError(RuntimeError):
@@ -61,7 +61,7 @@ class GovernedTaskBootstrap:
     def create(self, request: Mapping[str, Any]) -> dict[str, Any]:
         values = {key: str(request[key]) for key in ("task_id", "repository", "base_oid", "target_branch", "risk", "description", "idempotency_key")}
         if values["repository"] != "aios/aios" or not OID_RE.fullmatch(values["base_oid"]): raise BootstrapError("TASK_BINDING_INVALID")
-        if not BRANCH_RE.fullmatch(values["target_branch"]): raise BootstrapError("TARGET_BRANCH_INVALID")
+        if not validate_target_branch(values["target_branch"]): raise BootstrapError("TARGET_BRANCH_INVALID")
         existing = self._read_existing(values["task_id"])
         if existing is not None:
             if any(getattr(existing, key) != values[key] for key in ("repository", "base_oid", "target_branch", "risk", "description")): raise BootstrapError("TASK_BINDING_CONFLICT")

active-runtime/app/aios_target_branch_policy.py

@@ -0,0 +1,56 @@
+"""Single-source validation for governed implementation branch names.
+
+The policy accepts ordinary implementation branches while rejecting protected
+refs, malformed Git ref names, traversal-like components, and characters that
+Git treats specially. Callers translate a false result into their own stable
+error contract.
+"""
+from __future__ import annotations
+
+import re
+from typing import Any
+
+
+BRANCH_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
+PROTECTED_BRANCHES = frozenset({"HEAD", "main", "master"})
+PROTECTED_PREFIXES = (
+    "main/",
+    "master/",
+    "refs/",
+    "tags/",
+    "tag/",
+    "release/",
+    "releases/",
+    "baseline/",
+    "production/",
+)
+
+
+def validate_target_branch(value: Any) -> bool:
+    """Return whether *value* is a safe, non-protected implementation branch."""
+    if not isinstance(value, str) or not 3 <= len(value) <= 180:
+        return False
+    if not BRANCH_RE.fullmatch(value):
+        return False
+    if value in PROTECTED_BRANCHES or value == "@":
+        return False
+    if value.startswith(PROTECTED_PREFIXES):
+        return False
+    if value.startswith("/") or value.endswith("/"):
+        return False
+    if value.endswith(".") or value.endswith(".lock"):
+        return False
+    if ".." in value or "//" in value or "@{" in value:
+        return False
+    if any(ord(char) < 0x20 or ord(char) == 0x7F or char in " ~^:?*[\\" for char in value):
+        return False
+    components = value.split("/")
+    if any(
+        not component
+        or component in {".", "..", "@"}
+        or component.startswith(".")
+        or component.endswith(".lock")
+        for component in components
+    ):
+        return False
+    return True

active-runtime/app/authorized_task_writer_provisioning.py

@@ -33,7 +33,7 @@ from signer_contract_compatibility import (
 OID_RE = re.compile(r"^[0-9a-f]{40}$")
 SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
 TASK_RE = re.compile(r"^[A-Z0-9][A-Z0-9._:-]{7,127}$")
-BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
+from aios_target_branch_policy import BRANCH_RE, validate_target_branch
 OWNER_REF_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/@+-]{3,255}$")
 
 REQUIRED_REQUEST_FIELDS = frozenset(
@@ -353,11 +353,7 @@ def _validate_request(
         raise ProvisioningError("BASE_REF_REJECTED", 409)
     if not isinstance(raw["base_oid"], str) or not OID_RE.fullmatch(raw["base_oid"]):
         raise ProvisioningError("BASE_OID_REJECTED")
-    if (
-        not isinstance(raw["target_branch"], str)
-        or not BRANCH_RE.fullmatch(raw["target_branch"])
-        or raw["target_branch"] in {"main", "master"}
-    ):
+    if not validate_target_branch(raw["target_branch"]):
         raise ProvisioningError("TARGET_BRANCH_REJECTED")
     if not isinstance(raw["task_id"], str) or not TASK_RE.fullmatch(raw["task_id"]):
         raise ProvisioningError("TASK_ID_REJECTED")

active-runtime/app/owner_authorization_broker.py

@@ -23,6 +23,7 @@ from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import Any, Mapping, Protocol
 
+from aios_target_branch_policy import validate_target_branch
 from authorized_task_writer_provisioning import (
     AuthorizedTaskWriterProvisioner,
     BRANCH_RE,
@@ -335,11 +336,7 @@ def _validate_request(
         or not OID_RE.fullmatch(raw["base_oid"])
     ):
         raise OwnerAuthorizationError("BASE_OID_REJECTED")
-    if (
-        not isinstance(raw["target_branch"], str)
-        or not BRANCH_RE.fullmatch(raw["target_branch"])
-        or raw["target_branch"] in {"main", "master"}
-    ):
+    if not validate_target_branch(raw["target_branch"]):
         raise OwnerAuthorizationError("TARGET_BRANCH_REJECTED")
     if (
         not isinstance(raw["task_id"], str)

active-runtime/app/owner_local_auth_ingress.py

@@ -49,7 +49,7 @@ OID_RE = re.compile(r"^[0-9a-f]{40}$")
 IDENTITY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._@:+/-]{1,254}$")
 SCOPE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{1,127}$")
 TASK_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{1,255}$")
-BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9._/-]{1,220}$")
+from aios_target_branch_policy import validate_target_branch
 
 MAX_FRAME = 64 * 1024
 CHALLENGE_TTL_SECONDS = 120
@@ -207,10 +207,7 @@ def validate_binding(value: Mapping[str, Any]) -> dict[str, Any]:
         raise BootstrapError("BASE_REF_REJECTED")
     if not isinstance(request["base_oid"], str) or not OID_RE.fullmatch(request["base_oid"]):
         raise BootstrapError("BASE_OID_REJECTED")
-    if (
-        not isinstance(request["target_branch"], str)
-        or not BRANCH_RE.fullmatch(request["target_branch"])
-    ):
+    if not validate_target_branch(request["target_branch"]):
         raise BootstrapError("TARGET_BRANCH_REJECTED")
     if (
         not isinstance(request["task_id"], str)

active-runtime/app/task_identity_overlay.py

@@ -17,6 +17,8 @@ import hashlib
 import json
 import os
 import re
+
+from aios_target_branch_policy import validate_target_branch
 import secrets
 import tempfile
 from dataclasses import dataclass
@@ -192,8 +194,7 @@ def _validate_config(config: IdentityOverlayConfig) -> None:
     if not OID_RE.fullmatch(binding.base_oid):
         raise IdentityOverlayError("OVERLAY_BASE_OID_INVALID", 503)
     if (
-        not BRANCH_RE.fullmatch(binding.target_branch)
-        or binding.target_branch in {"main", "master"}
+        not validate_target_branch(binding.target_branch)
     ):
         raise IdentityOverlayError("OVERLAY_TARGET_BRANCH_INVALID", 503)
     if not OWNER_IDENTITY_RE.fullmatch(config.owner_identity):

tests/test_existing_branch_owner_approval.py

@@ -1,285 +0,0 @@
-from __future__ import annotations
-
-import base64
-from datetime import datetime, timedelta, timezone
-import json
-
-from cryptography.hazmat.primitives import serialization
-from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
-import pytest
-
-from aios_existing_branch_approval import (
-    DurableNonceReplayLedger,
-    OwnerApprovalEnvelope,
-    OwnerApprovalRejected,
-    OwnerApprovalVerifier,
-    TrustRootEvidence,
-)
-from aios_existing_branch_publisher import (
-    ExistingBranchPublicationRejected,
-    ExistingBranchPublisher,
-    RemoteBranchObservation,
-    RemotePublicationEvidence,
-)
-
-
-BASE_OID = "a" * 40
-COMMIT_OID = "b" * 40
-BRANCH = "agent/owner-approval-contract-20260810"
-
-
-class TrustRoot:
-    def __init__(self, public_key: bytes, *, revocation: str = "ACTIVE") -> None:
-        self.public_key = public_key
-        self.revocation = revocation
-        self.calls: list[str] = []
-
-    def resolve(self, *, key_id: str, owner_identity: str, trust_domain: str):
-        self.calls.append("trust-root")
-        return TrustRootEvidence(
-            key_id=key_id,
-            owner_identity=owner_identity,
-            trust_domain=trust_domain,
-            public_key=self.public_key,
-            authority_confirmed=True,
-            revocation_state=self.revocation,
-        )
-
-
-class Credentials:
-    def __init__(self, events: list[str]) -> None:
-        self.events = events
-
-    def project(self) -> None:
-        self.events.append("credential-project")
-
-    def revoke(self) -> None:
-        self.events.append("credential-revoke")
-
-
-class Workspace:
-    def __init__(self, events: list[str]) -> None:
-        self.events = events
-
-    def head_oid(self) -> str:
-        self.events.append("workspace-head")
-        return BASE_OID
-
-    def commit_payload(self, payload):
-        self.events.append("workspace-commit")
-        return COMMIT_OID
-
-
-class Workspaces:
-    def __init__(self, events: list[str]) -> None:
-        self.events = events
-
-    def open_detached(self, expected_branch_oid: str) -> Workspace:
-        assert expected_branch_oid == BASE_OID
-        self.events.append("workspace-open")
-        return Workspace(self.events)
-
-
-class Remote:
-    def __init__(self, events: list[str]) -> None:
-        self.events = events
-        self.observation = RemoteBranchObservation("PRESENT", BASE_OID, BASE_OID, True)
-        self.authority_oid = BASE_OID
-        self.pushes: list[tuple[str, str, str]] = []
-
-    def read_branch_oid(self, branch: str) -> str:
-        self.events.append("authority-observe")
-        assert branch == BRANCH
-        return self.authority_oid
-
-    def observe(self, branch: str) -> RemoteBranchObservation:
-        self.events.append("remote-observe")
-        assert branch == BRANCH
-        return self.observation
-
-    def publish_compare_and_swap(self, branch: str, expected: str, commit: str):
-        self.events.append("remote-publish")
-        self.pushes.append((branch, expected, commit))
-        return RemotePublicationEvidence(COMMIT_OID, COMMIT_OID, COMMIT_OID, True)
-
-
-class Audit:
-    def __init__(self) -> None:
-        self.rows: list[dict] = []
-
-    def record(self, event) -> None:
-        self.rows.append(dict(event))
-
-
-def request(*, payload=None, branch=BRANCH, expected_oid=BASE_OID):
-    return {
-        "schema": "aios.existing-branch-publication.v1",
-        "repository": "aios/aios",
-        "branch": branch,
-        "expected_branch_oid": expected_oid,
-        "operation": "publishExistingBranchGoverned",
-        "idempotency_key": "owner-approval-binding-20260810",
-        "payload": {"files": ["active-runtime/app/example.py"]} if payload is None else payload,
-    }
-
-
-def signed_approval(private: Ed25519PrivateKey, request_value, **replacements):
-    from aios_existing_branch_publisher import ExistingBranchPublicationRequest
-
-    parsed = ExistingBranchPublicationRequest.from_mapping(request_value)
-    now = datetime.now(timezone.utc)
-    raw = {
-        "schema": "aios.existing-branch-owner-approval.v1",
-        "approval_id": "approval-owner-binding-20260810",
-        "repository": parsed.repository,
-        "branch": parsed.branch,
-        "expected_branch_oid": parsed.expected_branch_oid,
-        "operation": "publishExistingBranchGoverned",
-        "issued_at": (now - timedelta(seconds=30)).isoformat(),
-        "not_before": (now - timedelta(seconds=1)).isoformat(),
-        "expires_at": (now + timedelta(minutes=5)).isoformat(),
-        "nonce": "nonce-owner-binding-20260810",
-        "owner_identity": "owner.aios",
-        "trust_domain": "aios-owner-root",
-        "key_id": "owner-key-20260810",
-        "request_digest": parsed.request_digest,
-        "idempotency_digest": parsed.idempotency_digest,
-        "signature": "AA==",
-    }
-    raw.update(replacements)
-    envelope = OwnerApprovalEnvelope.from_mapping(raw)
-    raw["signature"] = base64.b64encode(private.sign(envelope.canonical_bytes())).decode("ascii")
-    return raw
-
-
-def service(tmp_path, *, revocation="ACTIVE"):
-    private = Ed25519PrivateKey.generate()
-    public_key = private.public_key().public_bytes(
-        serialization.Encoding.Raw, serialization.PublicFormat.Raw
-    )
-    root = TrustRoot(public_key, revocation=revocation)
-    events: list[str] = []
-    remote = Remote(events)
-    audit = Audit()
-    publisher = ExistingBranchPublisher(
-        verifier=OwnerApprovalVerifier(root, allowed_trust_domains={"aios-owner-root"}),
-        replay_ledger=DurableNonceReplayLedger(tmp_path / "replay"),
-        authority=remote,
-        remote=remote,
-        credentials=Credentials(events),
-        workspaces=Workspaces(events),
-        audit=audit,
-    )
-    return private, root, events, remote, audit, publisher
-
-
-def test_valid_signed_owner_approval_is_verified_before_capability_projection(tmp_path):
-    private, root, events, remote, audit, publisher = service(tmp_path)
-    value = request()
-    approval = signed_approval(private, value)
-
-    result = publisher.publish(value, approval)
-
-    assert result["status"] == "PUBLISHED"
-    assert result["commit_oid"] == COMMIT_OID
-    assert remote.pushes == [(BRANCH, BASE_OID, COMMIT_OID)]
-    assert events == [
-        "authority-observe",
-        "remote-observe",
-        "workspace-open",
-        "workspace-head",
-        "credential-project",
-        "workspace-commit",
-        "remote-publish",
-        "credential-revoke",
-    ]
-    assert root.calls == ["trust-root"]
-    assert approval["signature"] not in json.dumps(audit.rows)
-    assert "signature" not in audit.rows[-1]
-
-
-@pytest.mark.parametrize(
-    "case",
-    [
-        "invalid_signature",
-        "expired",
-        "revoked",
-        "missing_key_evidence",
-        "missing_revocation_evidence",
-        "wrong_repository",
-        "wrong_branch",
-        "wrong_oid",
-        "wrong_operation",
-        "arbitrary_reference",
-    ],
-)
-def test_invalid_owner_approval_inputs_are_rejected_without_publication(tmp_path, case):
-    private, root, events, remote, _, publisher = service(
-        tmp_path, revocation="REVOKED" if case == "revoked" else "ACTIVE"
-    )
-    value = request()
-    approval = signed_approval(private, value)
-    if case == "invalid_signature":
-        approval["signature"] = "AAAA"
-    elif case == "expired":
-        approval = signed_approval(
-            private,
-            value,
-            issued_at="2000-01-01T00:00:00+00:00",
-            not_before="2000-01-01T00:00:01+00:00",
-            expires_at="2000-01-01T00:01:00+00:00",
-        )
-    elif case == "wrong_repository":
-        approval = signed_approval(private, value, repository="other/repository")
-    elif case == "wrong_branch":
-        approval = signed_approval(private, value, branch="agent/another-approved-branch")
-    elif case == "wrong_oid":
-        approval = signed_approval(private, value, expected_branch_oid="c" * 40)
-    elif case == "wrong_operation":
-        approval["operation"] = "differentOperation"
-    elif case == "arbitrary_reference":
-        approval = "owner-approval-reference-123"
-    elif case == "missing_key_evidence":
-        root.public_key = None
-    elif case == "missing_revocation_evidence":
-        root.revocation = "UNKNOWN"
-
-    with pytest.raises((OwnerApprovalRejected, ExistingBranchPublicationRejected)):
-        publisher.publish(value, approval)
-
-    assert remote.pushes == []
-    assert "credential-project" not in events
-    assert "workspace-open" not in events
-
-
-@pytest.mark.parametrize(
-    "changed",
-    [
-        lambda value: {**value, "branch": "agent/approval-replay-other-branch"},
-        lambda value: {**value, "expected_branch_oid": "c" * 40},
-        lambda value: {**value, "payload": {"files": ["other.py"]}},
-    ],
-)
-def test_signed_approval_cannot_be_replayed_for_another_branch_oid_or_payload(tmp_path, changed):
-    private, _, events, remote, _, publisher = service(tmp_path)
-    original = request()
-    approval = signed_approval(private, original)
-
-    with pytest.raises(OwnerApprovalRejected, match="OWNER_APPROVAL_BINDING_REJECTED"):
-        publisher.publish(changed(original), approval)
-
-    assert remote.pushes == []
-    assert events == []
-
-
-def test_signed_approval_nonce_is_single_use(tmp_path):
-    private, _, events, remote, _, publisher = service(tmp_path)
-    value = request()
-    approval = signed_approval(private, value)
-
-    assert publisher.publish(value, approval)["status"] == "PUBLISHED"
-    with pytest.raises(OwnerApprovalRejected, match="OWNER_APPROVAL_REPLAY_REJECTED"):
-        publisher.publish(value, approval)
-
-    assert len(remote.pushes) == 1
-    assert events.count("credential-project") == 1

tests/test_existing_branch_publication_security.py

@@ -1,72 +0,0 @@
-from __future__ import annotations
-
-import inspect
-
-from aios_existing_branch_publisher import (
-    ExistingBranchPublisher,
-    RemoteBranchObservation,
-)
-from test_existing_branch_owner_approval import (
-    BASE_OID,
-    COMMIT_OID,
-    request,
-    service,
-    signed_approval,
-)
-
-
-def test_remote_oid_conflict_requires_manual_owner_review_before_capability_projection(tmp_path):
-    private, _, events, remote, _, publisher = service(tmp_path)
-    remote.observation = RemoteBranchObservation("PRESENT", "c" * 40, "c" * 40, True)
-    value = request()
-
-    result = publisher.publish(value, signed_approval(private, value))
-
-    assert result == {"status": "MANUAL_OWNER_REVIEW_REQUIRED"}
-    assert events == ["authority-observe", "remote-observe"]
-    assert remote.pushes == []
-
-
-def test_ambiguous_remote_requires_reconciliation_after_one_observation(tmp_path):
-    private, _, events, remote, _, publisher = service(tmp_path)
-    remote.observation = RemoteBranchObservation("AMBIGUOUS", None, None, None)
-    value = request()
-
-    result = publisher.publish(value, signed_approval(private, value))
-
-    assert result == {"status": "RECONCILIATION_REQUIRED"}
-    assert events == ["authority-observe", "remote-observe"]
-    assert remote.pushes == []
-
-
-def test_authority_oid_conflict_requires_manual_owner_review(tmp_path):
-    private, _, events, remote, _, publisher = service(tmp_path)
-    remote.authority_oid = "c" * 40
-    value = request()
-
-    result = publisher.publish(value, signed_approval(private, value))
-
-    assert result == {"status": "MANUAL_OWNER_REVIEW_REQUIRED"}
-    assert events == ["authority-observe"]
-    assert remote.pushes == []
-
-
-def test_success_reports_all_remote_agreement_fields(tmp_path):
-    private, _, _, _, _, publisher = service(tmp_path)
-    value = request()
-
-    result = publisher.publish(value, signed_approval(private, value))
-
-    assert result == {
-        "status": "PUBLISHED",
-        "commit_oid": COMMIT_OID,
-        "authority_oid": COMMIT_OID,
-        "gitea_git_oid": COMMIT_OID,
-        "gitea_api_oid": COMMIT_OID,
-        "commit_visible": True,
-    }
-
-
-def test_publisher_exposes_only_compare_and_swap_publication_path():
-    source = inspect.getsource(ExistingBranchPublisher).lower()
-    assert "force" not in source

tests/test_owner_local_auth_ingress.py

@@ -53,7 +53,7 @@ def test_exact_binding_passes():
     [
         (("owner_identity",), "owner", None),
         (("authenticated_scope",), "*", "OWNER_SCOPE_INVALID"),
-        (("request", "target_branch"), "patch/not-allowed", "TARGET_BRANCH_REJECTED"),
+        (("request", "target_branch"), "release/not-allowed", "TARGET_BRANCH_REJECTED"),
         (("request", "production_change_authorized"), True, "CONSEQUENTIAL_CAPABILITY_REJECTED"),
         (("request", "tag_change_authorized"), True, "CONSEQUENTIAL_CAPABILITY_REJECTED"),
         (("request", "baseline_change_authorized"), True, "CONSEQUENTIAL_CAPABILITY_REJECTED"),

tests/test_target_branch_policy.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+from aios_target_branch_policy import validate_target_branch
+
+
+def test_accepts_ordinary_implementation_branches() -> None:
+    for branch in (
+        "agent/worker-change-20260815",
+        "patch/not-allowed-anymore",
+        "fix/target-branch-policy",
+        "feature/ordinary-implementation",
+        "chore/code_stats-v2",
+    ):
+        assert validate_target_branch(branch), branch
+
+
+def test_rejects_protected_namespaces() -> None:
+    for branch in (
+        "main",
+        "master",
+        "main/hotfix",
+        "master/hotfix",
+        "refs/heads/patch/change",
+        "tags/release-v1",
+        "tag/release-v1",
+        "release/v1",
+        "releases/v1",
+        "baseline/current",
+        "production/hotfix",
+    ):
+        assert not validate_target_branch(branch), branch
+
+
+def test_rejects_malformed_git_refs_and_traversal() -> None:
+    for branch in (
+        "../escape",
+        "patch/../escape",
+        "patch//escape",
+        "patch/.",
+        "patch/.hidden",
+        "patch/name.lock",
+        "patch/name.",
+        "patch/name~bad",
+        "patch/name^bad",
+        "patch/name:bad",
+        "patch/name?bad",
+        "patch/name*bad",
+        "patch/name[bad",
+        "patch/name\\\\bad",
+        "patch/name@{bad}",
+        "patch/name with-space",
+        "patch/line\\nbreak",
+        "@",
+    ):
+        assert not validate_target_branch(branch), repr(branch)