fix(bridge): enable governed CREATE_NEW continuation normalization

active-runtime/app/aios_authority_git_executor.py

@@ -11,8 +11,6 @@ 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
@@ -97,8 +95,7 @@ class AuthorityGitExecutor:
         return path
 
     def advance_task_branch_CAS(self, branch: str, expected_oid: str, new_oid: str) -> str:
-        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):
+        if not TASK_BRANCH_RE.fullmatch(branch):
             raise AuthorityGitError("TASK_BRANCH_SCOPE_DENIED")
         _oid(expected_oid, "expected_oid")
         _oid(new_oid, "new_oid")
@@ -115,6 +112,27 @@ class AuthorityGitExecutor:
             raise AuthorityGitError("AUTHORITY_CAS_READBACK_MISMATCH")
         return readback
 
+    def create_task_branch_CAS(self, branch: str, new_oid: str) -> str:
+        """Create an agent branch only if the authoritative ref is absent."""
+        if not TASK_BRANCH_RE.fullmatch(branch):
+            raise AuthorityGitError("TASK_BRANCH_SCOPE_DENIED")
+        _oid(new_oid, "new_oid")
+        current = self.read_ref(branch)
+        if current is not None:
+            if current == new_oid:
+                return new_oid
+            raise AuthorityGitError("AUTHORITY_BRANCH_ALREADY_EXISTS")
+        try:
+            self._git(["update-ref", branch, new_oid, "0" * 40])
+        except AuthorityGitError as exc:
+            if self.read_ref(branch) == new_oid:
+                return new_oid
+            raise AuthorityGitError("AUTHORITY_BRANCH_CREATE_CONFLICT") from exc
+        readback = self.read_ref(branch)
+        if readback != new_oid:
+            raise AuthorityGitError("AUTHORITY_CAS_READBACK_MISMATCH")
+        return readback
+
     def ff_CAS_main_after_verified_protected_merge(self, expected_oid: str, new_oid: str) -> str:
         _oid(expected_oid, "expected_oid")
         _oid(new_oid, "new_oid")

active-runtime/app/aios_chatgpt_bridge_v2_server.py

@@ -9,8 +9,6 @@ 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
 
@@ -126,7 +124,7 @@ class GovernedRuntime:
             workspace = DetachedGitWorkspace(worktree_path)
         if workspace.head_oid() != lease.expected_branch_oid or not workspace.is_detached(): raise RuntimeError("WORKTREE_HEAD_MISMATCH")
         job_root = self._job_root(lease.task_id); controller = GitOpsJobController(JobStore(job_root, payload_root=job_root / "payloads")); credential = ServerSideCredentialProjection(self.config.ssh_key_path, self.config.token_path); remote = GiteaContinuationRemote(workspace=worktree_path, credential=credential, remote_name=self.config.remote_name, repository=lease.repository, api_base=self.config.api_base); executor = ContinuationExecutor(controller, AuthorityGitAdapter(self.authority), workspace, remote, credential)
-        binding = ContinuationRuntimeBinding(request=lease, workspace_path=worktree_path, lease_not_before=datetime.now(timezone.utc) - timedelta(seconds=1), lease_expires_at=datetime.now(timezone.utc) + timedelta(seconds=lease.lease_seconds), original_receipt_valid=True, original_grant_consumed=True, permissions=ContinuationPermissions())
+        binding = ContinuationRuntimeBinding(request=lease, workspace_path=worktree_path, lease_not_before=datetime.now(timezone.utc) - timedelta(seconds=1), lease_expires_at=datetime.now(timezone.utc) + timedelta(seconds=lease.lease_seconds), original_receipt_valid=True, original_grant_consumed=True, permissions=ContinuationPermissions(create_target_branch=lease.branch_mode == "CREATE_NEW"))
         result = (binding, controller, executor); self.runtimes[key] = result; self.controllers[lease.task_id] = controller; return result
 
     def create_task(self, request: Mapping[str, Any]) -> dict[str, Any]:
@@ -150,7 +148,16 @@ class GovernedRuntime:
         if request["target_branch"] != binding.request.target_branch or request["expected_branch_oid"] != binding.request.expected_branch_oid: raise RuntimeError("LEASE_BINDING_CONFLICT")
         commit = ContinuationCommit(idempotency_key=str(request["idempotency_key"]), files=self._files(request), message=str(request["message"]))
         executor.workspace.materialize_files(commit.files)
-        return executor.execute(binding, commit)
+        result = executor.execute(binding, commit)
+        receipt = result.get("receipt") if isinstance(result, Mapping) else None
+        receipt = receipt if isinstance(receipt, Mapping) else {}
+        commit_oid = receipt.get("authority_oid") or receipt.get("desired_remote_oid")
+        if not isinstance(commit_oid, str) or len(commit_oid) != 40 or any(ch not in "0123456789abcdef" for ch in commit_oid):
+            raise RuntimeError("COMMIT_RECEIPT_OID_MISSING")
+        response = dict(result)
+        response["commit_oid"] = commit_oid
+        response["commit_visible"] = bool(receipt.get("commit_visible") or receipt.get("remote_verified"))
+        return response
 
     def _controller_for_job(self, job_id: str) -> GitOpsJobController:
         for controller in self.controllers.values():
@@ -212,7 +219,8 @@ class GovernedRuntime:
         if (
             not task_id
             or repository != self.config.repository
-            or not validate_target_branch(branch)
+            or not branch.startswith("agent/")
+            or branch in {"main", "master"}
             or branch_mode not in {"CREATE_NEW", "CONTINUE_EXISTING"}
             or generation < 1
             or not self._valid_oid(base)
@@ -288,7 +296,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 validate_target_branch(branch):
+            if not task_id or not branch.startswith("agent/"):
                 continue
             authority = self.authority.read_ref("refs/heads/" + branch)
             try:

active-runtime/app/aios_continuation_executor.py

@@ -52,11 +52,12 @@ class AuthorityPort(Protocol):
     def branch_head(self, branch: str) -> str: ...
     def is_ancestor(self, older: str, newer: str) -> bool: ...
     def compare_and_swap_branch(self, branch: str, expected: str, new: str) -> str: ...
+    def create_task_branch_CAS(self, branch: str, new: str) -> str: ...
 
 
 class RemotePort(Protocol):
     def repository_identity(self) -> str: ...
-    def push_exact(self, branch: str, expected: str, new: str) -> None: ...
+    def push_exact(self, branch: str, expected: str | None, new: str) -> None: ...
     def read_git_ref(self, branch: str) -> str | None: ...
     def read_api_branch(self, branch: str) -> str | None: ...
     def commit_visible(self, oid: str) -> bool: ...
@@ -106,6 +107,9 @@ class AuthorityGitAdapter:
     def compare_and_swap_branch(self, branch: str, expected: str, new: str) -> str:
         return self.executor.advance_task_branch_CAS(self._ref(branch), expected, new)
 
+    def create_task_branch_CAS(self, branch: str, new: str) -> str:
+        return self.executor.create_task_branch_CAS(self._ref(branch), new)
+
 
 @dataclass(slots=True)
 class DetachedGitWorkspace:
@@ -258,10 +262,16 @@ class GiteaContinuationRemote:
             raise ContinuationExecutionError("GITEA_GIT_READBACK_INVALID")
         return oid
 
-    def push_exact(self, branch: str, expected: str, new: str) -> None:
+    def push_exact(self, branch: str, expected: str | None, new: str) -> None:
         self._require_projected()
         self.repository_identity()
-        if self.read_git_ref(branch) != expected:
+        observed = self.read_git_ref(branch)
+        if observed == new:
+            return
+        if expected is None:
+            if observed is not None:
+                raise ContinuationExecutionError("GITEA_BRANCH_EXPECTATION_MISMATCH")
+        elif observed != expected:
             raise ContinuationExecutionError("GITEA_BRANCH_EXPECTATION_MISMATCH")
         _run(["git", "push", self.remote_name, f"{new}:refs/heads/{branch}"], cwd=self.workspace)
 
@@ -321,8 +331,18 @@ class ContinuationPermissions:
     production_write: bool = False
     main_write: bool = False
 
-    def validate(self) -> None:
-        if any(asdict(self).values()):
+    def validate(self, *, branch_mode: str) -> None:
+        if branch_mode not in {"CREATE_NEW", "CONTINUE_EXISTING"}:
+            raise ContinuationExecutionError("BRANCH_MODE_INVALID")
+        expected_create = branch_mode == "CREATE_NEW"
+        forbidden = (
+            self.force_push,
+            self.tag_write,
+            self.baseline_write,
+            self.production_write,
+            self.main_write,
+        )
+        if self.create_target_branch != expected_create or any(forbidden):
             raise ContinuationExecutionError("MUTATION_PERMISSION_DENIED")
 
 
@@ -415,12 +435,14 @@ class ContinuationExecutor:
 
     def _validate_preflight(self, binding: ContinuationRuntimeBinding) -> None:
         request = binding.request
-        normalized = validate_writer_lease_request_v2(self._request_mapping(request), self.authority)
-        if normalized != {**self._request_mapping(request), "create_target_branch": False, "worktree_mode": "detached_at_expected_head", "cas_ref_update_required": True}:
-            raise ContinuationExecutionError("CONTINUATION_NORMALIZATION_INVALID")
+        # WriterLeaseRequestV2 validation is the single normalization authority.
+        # CREATE_NEW legitimately normalizes to create_target_branch=True and
+        # detached_at_base_oid; pinning the CONTINUE_EXISTING shape here rejects
+        # valid first commits before the governed CAS ref update can run.
+        validate_writer_lease_request_v2(self._request_mapping(request), self.authority)
         if not binding.original_receipt_valid or not binding.original_grant_consumed:
             raise ContinuationExecutionError("ORIGINAL_GRANT_LINEAGE_INVALID")
-        binding.permissions.validate()
+        binding.permissions.validate(branch_mode=request.branch_mode)
         now = self.now()
         if now < binding.lease_not_before:
             raise ContinuationExecutionError("LEASE_NOT_ACTIVE")
@@ -442,7 +464,7 @@ class ContinuationExecutor:
             "campaign_id": request.task_id,
             "task_id": request.task_id,
             "repository": request.repository,
-            "operation": "CONTINUATION_COMMIT_FILES",
+            "operation": "CONTINUATION_COMMIT_FILES_CREATE_NEW" if request.branch_mode == "CREATE_NEW" else "CONTINUATION_COMMIT_FILES",
             "branch": request.target_branch,
             "expected_branch_oid": request.expected_branch_oid,
             "base_oid": request.base_oid,
@@ -488,7 +510,10 @@ class ContinuationExecutor:
             new_oid = self.workspace.commit_files(commit.files, commit.message)
             if not OID_RE.fullmatch(new_oid) or not self.authority.is_ancestor(request.expected_branch_oid, new_oid):
                 raise ContinuationExecutionError("NON_FAST_FORWARD_COMMIT")
-            authority_oid = self.authority.compare_and_swap_branch(request.target_branch, request.expected_branch_oid, new_oid)
+            if request.branch_mode == "CREATE_NEW":
+                authority_oid = self.authority.create_task_branch_CAS(request.target_branch, new_oid)
+            else:
+                authority_oid = self.authority.compare_and_swap_branch(request.target_branch, request.expected_branch_oid, new_oid)
             if authority_oid != new_oid:
                 raise ContinuationExecutionError("AUTHORITY_CAS_READBACK_MISMATCH")
             self.controller.advance(
@@ -499,7 +524,7 @@ class ContinuationExecutor:
             )
             try:
                 self.controller.advance(submitted.job_id, JobState.REMOTE_SYNCING, side_effect_state="IN_PROGRESS")
-                self.remote.push_exact(request.target_branch, request.expected_branch_oid, new_oid)
+                self.remote.push_exact(request.target_branch, None if request.branch_mode == "CREATE_NEW" else request.expected_branch_oid, new_oid)
             except Exception:
                 return self._remote_pending(submitted.job_id, authority_oid, new_oid)
             return self._finish_remote_verification(submitted.job_id, authority_oid, new_oid)
@@ -575,7 +600,7 @@ class ContinuationExecutor:
             if authority_oid == git_oid == api_oid == new_oid and visible:
                 self.controller.advance(job_id, JobState.RECONCILING, side_effect_state="OBSERVATION_FIRST")
                 return self._finish_remote_verification(job_id, new_oid, new_oid)
-            expected_oid = request["expected_branch_oid"]
+            expected_oid = None if request.get("branch_mode") == "CREATE_NEW" else request["expected_branch_oid"]
             if (
                 self.authority.is_ancestor(new_oid, authority_oid)
                 and git_oid is not None

active-runtime/app/aios_continuation_lease.py

@@ -10,8 +10,6 @@ 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
 
 
@@ -62,7 +60,7 @@ class WriterLeaseRequestV2:
             raise ContinuationError("branch_mode_invalid")
         _oid(self.base_oid, "base_oid")
         _oid(self.expected_branch_oid, "expected_branch_oid")
-        if not validate_target_branch(self.target_branch):
+        if self.target_branch in {"main", "master"} or self.target_branch.startswith(("refs/", "tags/")):
             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,8 +8,6 @@ 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:
@@ -46,7 +44,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 validate_target_branch(target_branch) or generation < 1 or not 1 <= lease_seconds <= 86400:
+        if repository != "aios/aios" or not target_branch.startswith("agent/") 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_gitea_publisher.py

@@ -11,7 +11,6 @@ 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)
@@ -36,7 +35,7 @@ class GiteaPublisher:
 
     @staticmethod
     def validate_branch(branch: str) -> None:
-        if not validate_target_branch(branch):
+        if not isinstance(branch, str) or not branch.startswith("agent/") or branch in {"main", "master"}:
             raise ValueError("BRANCH_SCOPE_DENIED")
 
     def repository_identity(self) -> str:

active-runtime/app/aios_gitops_job_controller.py

@@ -18,8 +18,6 @@ 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
@@ -276,7 +274,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 not validate_target_branch(request["branch"]):
+        if request["branch"] in {"main", "master"} or request["branch"].startswith(("refs/", "tags/")):
             raise JobError("PROTECTED_BRANCH_REJECTED")
         binding = {field: str(request[field]) for field in fields}
         # Operation-specific, non-secret binding inputs must survive the
@@ -385,7 +383,7 @@ class GitOpsJobController:
                     raise JobError("JOB_NOT_FOUND")
                 if int(row["revision"]) != expected_revision:
                     raise JobError("REVISION_CAS_CONFLICT")
-                if not validate_target_branch(row["branch"]):
+                if row["branch"] in {"main", "master"}:
                     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,8 +4,6 @@ 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}$")
@@ -54,7 +52,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 validate_target_branch(observation.branch):
+    if not observation.branch.startswith("agent/"):
         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,8 +3,6 @@ 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}$")
@@ -41,7 +39,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 validate_target_branch(value.target_branch):
+    if not value.target_branch.startswith("agent/") or value.target_branch in {"main", "master"}:
         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}$")
-from aios_target_branch_policy import BRANCH_RE, validate_target_branch
+BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
 
 
 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 validate_target_branch(values["target_branch"]): raise BootstrapError("TARGET_BRANCH_INVALID")
+        if not BRANCH_RE.fullmatch(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

@@ -1,56 +0,0 @@
-"""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}$")
-from aios_target_branch_policy import BRANCH_RE, validate_target_branch
+BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
 OWNER_REF_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/@+-]{3,255}$")
 
 REQUIRED_REQUEST_FIELDS = frozenset(
@@ -353,7 +353,11 @@ 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 validate_target_branch(raw["target_branch"]):
+    if (
+        not isinstance(raw["target_branch"], str)
+        or not BRANCH_RE.fullmatch(raw["target_branch"])
+        or raw["target_branch"] in {"main", "master"}
+    ):
         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,7 +23,6 @@ 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,
@@ -336,7 +335,11 @@ def _validate_request(
         or not OID_RE.fullmatch(raw["base_oid"])
     ):
         raise OwnerAuthorizationError("BASE_OID_REJECTED")
-    if not validate_target_branch(raw["target_branch"]):
+    if (
+        not isinstance(raw["target_branch"], str)
+        or not BRANCH_RE.fullmatch(raw["target_branch"])
+        or raw["target_branch"] in {"main", "master"}
+    ):
         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}$")
-from aios_target_branch_policy import validate_target_branch
+BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9._/-]{1,220}$")
 
 MAX_FRAME = 64 * 1024
 CHALLENGE_TTL_SECONDS = 120
@@ -207,7 +207,10 @@ 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 validate_target_branch(request["target_branch"]):
+    if (
+        not isinstance(request["target_branch"], str)
+        or not BRANCH_RE.fullmatch(request["target_branch"])
+    ):
         raise BootstrapError("TARGET_BRANCH_REJECTED")
     if (
         not isinstance(request["task_id"], str)

active-runtime/app/task_identity_overlay.py

@@ -17,8 +17,6 @@ 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
@@ -194,7 +192,8 @@ def _validate_config(config: IdentityOverlayConfig) -> None:
     if not OID_RE.fullmatch(binding.base_oid):
         raise IdentityOverlayError("OVERLAY_BASE_OID_INVALID", 503)
     if (
-        not validate_target_branch(binding.target_branch)
+        not BRANCH_RE.fullmatch(binding.target_branch)
+        or binding.target_branch in {"main", "master"}
     ):
         raise IdentityOverlayError("OVERLAY_TARGET_BRANCH_INVALID", 503)
     if not OWNER_IDENTITY_RE.fullmatch(config.owner_identity):

active-runtime/app/test_continuation_create_new_normalization_20260815.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+import inspect
+
+from aios_authority_git_executor import AuthorityGitExecutor
+from aios_continuation_executor import ContinuationExecutor, ContinuationPermissions
+from aios_continuation_lease import WriterLeaseRequestV2
+from authorized_task_writer_provisioning import validate_writer_lease_request_v2
+
+
+BASE = "a" * 40
+BRANCH = "agent/bridge-create-new-normalization-test"
+
+
+class LocalAuthority:
+    def __init__(self, branch_head: str | None) -> None:
+        self._branch_head = branch_head
+
+    def branch_exists(self, branch: str) -> bool:
+        return self._branch_head is not None
+
+    def branch_head(self, branch: str) -> str:
+        assert self._branch_head is not None
+        return self._branch_head
+
+    def commit_exists(self, oid: str) -> bool:
+        return oid == BASE
+
+    def is_ancestor(self, older: str, newer: str) -> bool:
+        return older == BASE and newer == BASE
+
+
+def request(branch_mode: str) -> dict[str, object]:
+    return {
+        "schema": WriterLeaseRequestV2.schema,
+        "task_id": "bridge-create-new-normalization-test",
+        "repository": "aios/aios",
+        "branch_mode": branch_mode,
+        "base_oid": BASE,
+        "expected_branch_oid": BASE,
+        "target_branch": BRANCH,
+        "generation": 1,
+        "lease_seconds": 120,
+    }
+
+
+def main() -> None:
+    create_new = validate_writer_lease_request_v2(request("CREATE_NEW"), LocalAuthority(None))
+    assert create_new["create_target_branch"] is True
+    assert create_new["worktree_mode"] == "detached_at_base_oid"
+    assert create_new["cas_ref_update_required"] is True
+
+    continue_existing = validate_writer_lease_request_v2(request("CONTINUE_EXISTING"), LocalAuthority(BASE))
+    assert continue_existing["create_target_branch"] is False
+    assert continue_existing["worktree_mode"] == "detached_at_expected_head"
+    assert continue_existing["cas_ref_update_required"] is True
+
+    preflight = inspect.getsource(ContinuationExecutor._validate_preflight)
+    assert "validate_writer_lease_request_v2" in preflight
+    assert '"create_target_branch": False' not in preflight
+    assert '"worktree_mode": "detached_at_expected_head"' not in preflight
+
+    ContinuationPermissions(create_target_branch=True).validate(branch_mode="CREATE_NEW")
+    ContinuationPermissions().validate(branch_mode="CONTINUE_EXISTING")
+
+    cas = inspect.getsource(AuthorityGitExecutor.create_task_branch_CAS)
+    assert "update-ref" in cas
+    assert '"0" * 40' in cas
+    print("CREATE_NEW_NORMALIZATION_REGRESSION_PASS")
+
+
+if __name__ == "__main__":
+    main()

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"), "release/not-allowed", "TARGET_BRANCH_REJECTED"),
+        (("request", "target_branch"), "patch/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

@@ -1,55 +0,0 @@
-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)