fix(agent-task): recover runtime projection after source capture

active-runtime/app/aios_all_modified_files_packager.py

@@ -0,0 +1,470 @@
+#!/usr/bin/env python3
+"""Canonical AIOS all-modified-files evidence packager.
+
+The bundle contract is intentionally compatible with the historical
+aios_package_all_modified_files_v1.sh output while keeping secret scanning in
+version-controlled, tested code.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+import argparse
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+import tempfile
+import zipfile
+from typing import Iterable
+
+OID_RE = re.compile(r"^[0-9a-f]{40}$")
+TASK_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
+
+# A PEM private-key finding requires the complete PEM header. Source code,
+# tests, docs, or error text containing only "BEGIN PRIVATE KEY" are not
+# credentials and must not be rejected.
+PEM_PRIVATE_KEY_RE = re.compile(
+    rb"-----BEGIN (?:RSA |EC |DSA |OPENSSH )?PRIVATE KEY-----",
+    re.IGNORECASE,
+)
+BEARER_RE = re.compile(
+    rb"authorization\s*:\s*bearer\s+[A-Za-z0-9._~+/=-]{20,}",
+    re.IGNORECASE,
+)
+KEY_VALUE_SECRET_RE = re.compile(
+    rb"(?:password|passwd|secret|api[_-]?key)\s*[:=]\s*[^\s,;]{12,}",
+    re.IGNORECASE,
+)
+
+DOCUMENT_SUFFIXES = {".md", ".rst", ".adoc", ".txt"}
+DOCUMENT_PREFIXES = (
+    "README",
+    "CHANGELOG",
+    "RUNBOOK",
+    "ADR",
+    "ARCHITECTURE",
+    "DEPLOYMENT",
+    "TROUBLESHOOTING",
+)
+
+
+class PackagerError(RuntimeError):
+    pass
+
+
+def sha256(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def secret_finding(data: bytes) -> str | None:
+    if PEM_PRIVATE_KEY_RE.search(data):
+        return "PEM_PRIVATE_KEY"
+    if BEARER_RE.search(data):
+        return "AUTHORIZATION_BEARER"
+    if KEY_VALUE_SECRET_RE.search(data):
+        return "KEY_VALUE_SECRET"
+    return None
+
+
+def scan_secret(data: bytes, label: str) -> None:
+    finding = secret_finding(data)
+    if finding is not None:
+        raise PackagerError(
+            f"MODIFIED_FILES_SECRET_SCAN_MATCH:{finding}:{label}"
+        )
+
+
+def git_bytes(git_dir: Path, oid: str, path: str) -> bytes:
+    try:
+        return subprocess.check_output(
+            [
+                "/usr/bin/git",
+                f"--git-dir={git_dir}",
+                "show",
+                f"{oid}:{path}",
+            ],
+            timeout=30,
+        )
+    except subprocess.CalledProcessError as exc:
+        raise PackagerError(
+            "MODIFIED_FILES_GIT_SHOW_FAILED:" + path
+        ) from exc
+
+
+def git_text(git_dir: Path, *args: str) -> str:
+    try:
+        return subprocess.check_output(
+            ["/usr/bin/git", f"--git-dir={git_dir}", *args],
+            text=True,
+            timeout=30,
+        )
+    except subprocess.CalledProcessError as exc:
+        raise PackagerError(
+            "MODIFIED_FILES_GIT_COMMAND_FAILED:" + " ".join(args)
+        ) from exc
+
+
+def validate_lineage(git_dir: Path, base: str, candidate: str) -> None:
+    if OID_RE.fullmatch(base) is None or OID_RE.fullmatch(candidate) is None:
+        raise PackagerError("MODIFIED_FILES_OID_INVALID")
+    for oid in (base, candidate):
+        try:
+            subprocess.run(
+                [
+                    "/usr/bin/git",
+                    f"--git-dir={git_dir}",
+                    "cat-file",
+                    "-e",
+                    f"{oid}^{{commit}}",
+                ],
+                check=True,
+                timeout=30,
+            )
+        except subprocess.CalledProcessError as exc:
+            raise PackagerError("MODIFIED_FILES_COMMIT_MISSING:" + oid) from exc
+    if subprocess.run(
+        [
+            "/usr/bin/git",
+            f"--git-dir={git_dir}",
+            "merge-base",
+            "--is-ancestor",
+            base,
+            candidate,
+        ],
+        check=False,
+        timeout=30,
+    ).returncode != 0:
+        raise PackagerError("MODIFIED_FILES_LINEAGE_INVALID")
+
+
+def safe_destination(root: Path, relative: str) -> Path:
+    path = Path(relative)
+    if path.is_absolute() or ".." in path.parts:
+        raise PackagerError("MODIFIED_FILE_PATH_INVALID:" + relative)
+    destination = root / path
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    return destination
+
+
+def is_document_path(path: str) -> bool:
+    value = Path(path)
+    if path.startswith("docs/"):
+        return True
+    if value.suffix.lower() in DOCUMENT_SUFFIXES:
+        return True
+    upper = value.name.upper()
+    return any(upper.startswith(prefix) for prefix in DOCUMENT_PREFIXES)
+
+
+def diff_entries(
+    git_dir: Path,
+    base: str,
+    candidate: str,
+    before_root: Path,
+    after_root: Path,
+) -> list[dict[str, object]]:
+    raw = git_text(
+        git_dir,
+        "diff",
+        "--name-status",
+        "-M",
+        "-C",
+        base,
+        candidate,
+    )
+    entries: list[dict[str, object]] = []
+    for line in raw.splitlines():
+        if not line.strip():
+            continue
+        parts = line.split("\t")
+        status = parts[0]
+        code = status[0]
+        old_path: str | None = None
+        new_path: str | None = None
+        if code in {"R", "C"}:
+            if len(parts) != 3:
+                raise PackagerError(
+                    "MODIFIED_FILES_DIFF_PARSE_INVALID:" + line
+                )
+            old_path, new_path = parts[1], parts[2]
+        else:
+            if len(parts) != 2:
+                raise PackagerError(
+                    "MODIFIED_FILES_DIFF_PARSE_INVALID:" + line
+                )
+            new_path = parts[1] if code != "D" else None
+            old_path = parts[1] if code != "A" else None
+
+        entry: dict[str, object] = {"status": status}
+        if old_path is not None:
+            entry["previous_path"] = old_path
+            before = git_bytes(git_dir, base, old_path)
+            scan_secret(before, "before:" + old_path)
+            safe_destination(before_root, old_path).write_bytes(before)
+            entry["before_sha256"] = sha256(before)
+            entry["before_bytes"] = len(before)
+        if new_path is not None:
+            entry["path"] = new_path
+            after = git_bytes(git_dir, candidate, new_path)
+            scan_secret(after, "after:" + new_path)
+            safe_destination(after_root, new_path).write_bytes(after)
+            entry["after_sha256"] = sha256(after)
+            entry["after_bytes"] = len(after)
+        entry["deleted"] = code == "D"
+        entry["added"] = code == "A"
+        entry["renamed"] = code == "R"
+        entry["copied"] = code == "C"
+        entries.append(entry)
+    return entries
+
+
+def execution_entries(
+    extra_root: Path | None,
+    execution_root: Path,
+) -> list[dict[str, object]]:
+    if extra_root is None or not extra_root.exists():
+        return []
+    if extra_root.is_symlink() or not extra_root.is_dir():
+        raise PackagerError("EXTRA_FILES_DIR_INVALID")
+
+    entries: list[dict[str, object]] = []
+    for path in sorted(extra_root.rglob("*")):
+        if path.is_symlink():
+            raise PackagerError(
+                "EXTRA_FILE_SYMLINK_REJECTED:" + str(path)
+            )
+        if not path.is_file():
+            continue
+        relative = path.relative_to(extra_root).as_posix()
+        data = path.read_bytes()
+        scan_secret(data, "execution:" + relative)
+        safe_destination(execution_root, relative).write_bytes(data)
+        entries.append(
+            {
+                "path": relative,
+                "sha256": sha256(data),
+                "bytes": len(data),
+            }
+        )
+    return entries
+
+
+def write_checksums(bundle: Path) -> None:
+    checks: list[tuple[str, str]] = []
+    for path in sorted(bundle.rglob("*")):
+        if path.is_file() and path.name != "checksums.sha256":
+            checks.append(
+                (
+                    sha256(path.read_bytes()),
+                    path.relative_to(bundle).as_posix(),
+                )
+            )
+    (bundle / "checksums.sha256").write_text(
+        "".join(f"{digest}  {name}\n" for digest, name in checks),
+        encoding="utf-8",
+    )
+
+
+def verify_checksums(bundle: Path) -> None:
+    checksum_path = bundle / "checksums.sha256"
+    for line in checksum_path.read_text(encoding="utf-8").splitlines():
+        digest, relative = line.split("  ", 1)
+        path = bundle / relative
+        if not path.is_file() or sha256(path.read_bytes()) != digest:
+            raise PackagerError(
+                "MODIFIED_FILES_CHECKSUM_CLOSURE_INVALID:" + relative
+            )
+
+
+def build_bundle(
+    *,
+    git_dir: Path,
+    base: str,
+    candidate: str,
+    task_id: str,
+    extra_dir: Path | None = None,
+    output_root: Path | None = None,
+    now: datetime | None = None,
+) -> dict[str, object]:
+    if TASK_RE.fullmatch(task_id) is None:
+        raise PackagerError("TASK_ID_INVALID")
+    validate_lineage(git_dir, base, candidate)
+
+    timestamp = (now or datetime.now(timezone.utc)).strftime(
+        "%Y%m%dT%H%M%SZ"
+    )
+    safe_task = re.sub(r"[^A-Za-z0-9._-]+", "-", task_id).strip("-")
+    if output_root is None:
+        output_root = Path(
+            tempfile.mkdtemp(
+                prefix=f"aios-modified-files-bundle-{safe_task}-{timestamp}-",
+                dir="/var/tmp",
+            )
+        )
+    else:
+        output_root.mkdir(parents=True, exist_ok=True)
+
+    bundle = output_root / "bundle"
+    before_root = bundle / "repo" / "before"
+    after_root = bundle / "repo" / "after"
+    execution_root = bundle / "execution"
+    before_root.mkdir(parents=True)
+    after_root.mkdir(parents=True)
+    execution_root.mkdir(parents=True)
+
+    changes = diff_entries(
+        git_dir,
+        base,
+        candidate,
+        before_root,
+        after_root,
+    )
+    execution_files = execution_entries(extra_dir, execution_root)
+
+    document_changes = [
+        entry
+        for entry in changes
+        if is_document_path(
+            str(entry.get("path") or entry.get("previous_path") or "")
+        )
+    ]
+    execution_documents = [
+        entry
+        for entry in execution_files
+        if is_document_path(str(entry["path"]))
+    ]
+
+    documents_index = {
+        "schema": "aios.modified-documents.index.v1",
+        "task_id": task_id,
+        "base_oid": base,
+        "candidate_oid": candidate,
+        "repository_documents": document_changes,
+        "execution_documents": execution_documents,
+        "repository_document_count": len(document_changes),
+        "execution_document_count": len(execution_documents),
+    }
+    (bundle / "documents-index.json").write_text(
+        json.dumps(documents_index, indent=2, sort_keys=True) + "\n",
+        encoding="utf-8",
+    )
+
+    manifest = {
+        "schema": "aios.all-modified-files.bundle.v1",
+        "task_id": task_id,
+        "base_oid": base,
+        "candidate_oid": candidate,
+        "generated_at": (
+            now or datetime.now(timezone.utc)
+        ).isoformat(),
+        "changed_file_count": len(changes),
+        "execution_file_count": len(execution_files),
+        "repository_document_count": len(document_changes),
+        "execution_document_count": len(execution_documents),
+        "changes": changes,
+        "execution_files": execution_files,
+    }
+    (bundle / "all-modified-files-manifest.json").write_text(
+        json.dumps(manifest, indent=2, sort_keys=True) + "\n",
+        encoding="utf-8",
+    )
+    write_checksums(bundle)
+    verify_checksums(bundle)
+
+    zip_name = (
+        f"AIOS_{task_id}_ALL_MODIFIED_FILES_{candidate[:12]}.zip"
+        .replace("/", "-")
+    )
+    zip_path = output_root / zip_name
+    with zipfile.ZipFile(
+        zip_path,
+        "w",
+        compression=zipfile.ZIP_DEFLATED,
+    ) as archive:
+        for path in sorted(bundle.rglob("*")):
+            if path.is_symlink():
+                raise PackagerError(
+                    "BUNDLE_SYMLINK_REJECTED:" + str(path)
+                )
+            if path.is_file():
+                archive.write(
+                    path,
+                    arcname=path.relative_to(bundle).as_posix(),
+                )
+
+    zip_sha = sha256(zip_path.read_bytes())
+    summary = {
+        "schema": "aios.all-modified-files.bundle-summary.v1",
+        "task_id": task_id,
+        "base_oid": base,
+        "candidate_oid": candidate,
+        "bundle_path": str(zip_path),
+        "bundle_sha256": zip_sha,
+        "changed_file_count": len(changes),
+        "execution_file_count": len(execution_files),
+        "repository_document_count": len(document_changes),
+        "execution_document_count": len(execution_documents),
+    }
+    (output_root / "ALL_MODIFIED_FILES_BUNDLE.json").write_text(
+        json.dumps(summary, indent=2, sort_keys=True) + "\n",
+        encoding="utf-8",
+    )
+    return summary
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("base")
+    parser.add_argument("candidate")
+    parser.add_argument("task_id")
+    parser.add_argument("extra_dir", nargs="?")
+    parser.add_argument(
+        "--authority-git-dir",
+        default=os.environ.get(
+            "AIOS_AUTHORITY_GIT_DIR",
+            "/data/aios-gitops/repo.git",
+        ),
+    )
+    parser.add_argument("--output-root")
+    args = parser.parse_args()
+
+    try:
+        summary = build_bundle(
+            git_dir=Path(args.authority_git_dir),
+            base=args.base,
+            candidate=args.candidate,
+            task_id=args.task_id,
+            extra_dir=Path(args.extra_dir) if args.extra_dir else None,
+            output_root=(
+                Path(args.output_root) if args.output_root else None
+            ),
+        )
+    except PackagerError as exc:
+        print("FATAL:" + str(exc))
+        return 20
+
+    print("ALL_MODIFIED_FILES_BUNDLE=PASS")
+    print("ALL_MODIFIED_FILES_BUNDLE_PATH=" + str(summary["bundle_path"]))
+    print("ALL_MODIFIED_FILES_BUNDLE_SHA256=" + str(summary["bundle_sha256"]))
+    print("CHANGED_FILE_COUNT=" + str(summary["changed_file_count"]))
+    print("EXECUTION_FILE_COUNT=" + str(summary["execution_file_count"]))
+    print(
+        "REPOSITORY_DOCUMENT_COUNT="
+        + str(summary["repository_document_count"])
+    )
+    print(
+        "EXECUTION_DOCUMENT_COUNT="
+        + str(summary["execution_document_count"])
+    )
+    print(
+        "ALL_MODIFIED_FILES_EVIDENCE="
+        + str(Path(str(summary["bundle_path"])).parent)
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

active-runtime/app/aios_autonomous_control_plane.py

@@ -0,0 +1,288 @@
+"""Unified autonomous control-plane primitives.
+
+This is deliberately a library, not another daemon. Task API, workers and host
+recovery code consume the same Admission -> Invariant -> Repair semantics.
+"""
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+from enum import Enum
+import re
+from typing import Callable, Iterable, Mapping, Sequence
+
+from aios_capability_contract import (
+    capability_contract,
+    has_capabilities,
+    normalize_required_capabilities,
+    normalize_worker_capabilities,
+    registered_capability,
+    validate_capability_contract_registry,
+)
+
+
+class FailureClass(str, Enum):
+    CAPABILITY_CONTRACT_ERROR = "CAPABILITY_CONTRACT_ERROR"
+    CREDENTIAL_PROJECTION_ERROR = "CREDENTIAL_PROJECTION_ERROR"
+    DEPENDENCY_UNAVAILABLE = "DEPENDENCY_UNAVAILABLE"
+    POLICY_DENIED = "POLICY_DENIED"
+    RESOURCE_UNAVAILABLE = "RESOURCE_UNAVAILABLE"
+    WORKER_CRASH = "WORKER_CRASH"
+    LEASE_LOST = "LEASE_LOST"
+    BUSINESS_PERMANENT = "BUSINESS_PERMANENT"
+    UNKNOWN = "UNKNOWN"
+
+
+class RepairLevel(int, Enum):
+    AUTO_LOCAL = 0
+    AUTO_REPAIR_TASK = 1
+    OWNER_APPROVAL = 2
+    MANUAL = 3
+
+
+@dataclass(frozen=True)
+class InvariantResult:
+    name: str
+    passed: bool
+    detail: str = ""
+
+    def as_dict(self) -> dict[str, object]:
+        return asdict(self)
+
+
+@dataclass(frozen=True)
+class AdmissionDecision:
+    admitted: bool
+    state: str
+    invariants: tuple[InvariantResult, ...]
+    failure_class: FailureClass | None = None
+    repair_level: RepairLevel | None = None
+    repair_action: str | None = None
+
+    def as_dict(self) -> dict[str, object]:
+        return {
+            "schema": "aios.runtime-admission.decision.v1",
+            "admitted": self.admitted,
+            "state": self.state,
+            "invariants": [item.as_dict() for item in self.invariants],
+            "failure_class": None if self.failure_class is None else self.failure_class.value,
+            "repair_level": None if self.repair_level is None else int(self.repair_level),
+            "repair_action": self.repair_action,
+        }
+
+
+@dataclass(frozen=True)
+class RepairDecision:
+    failure_class: FailureClass
+    level: RepairLevel
+    action: str
+    consumes_business_attempt: bool
+
+    def as_dict(self) -> dict[str, object]:
+        return {
+            "failure_class": self.failure_class.value,
+            "level": int(self.level),
+            "action": self.action,
+            "consumes_business_attempt": self.consumes_business_attempt,
+        }
+
+
+def classify_failure(error: object, *, phase: str = "") -> FailureClass:
+    text = (str(error) + " " + str(phase)).lower()
+    if any(token in text for token in (
+        "credential", "permission denied", "token registry", "secret", "unauthorized"
+    )):
+        return FailureClass.CREDENTIAL_PROJECTION_ERROR
+    if any(token in text for token in (
+        "transport", "unavailable", "timeout", "connection", "bad gateway", "http_5"
+    )):
+        return FailureClass.DEPENDENCY_UNAVAILABLE
+    if any(token in text for token in (
+        "capability", "eligible", "matcher", "contract"
+    )):
+        return FailureClass.CAPABILITY_CONTRACT_ERROR
+    if any(token in text for token in (
+        "policy", "forbidden", "approval required", "not authorized"
+    )):
+        return FailureClass.POLICY_DENIED
+    if any(token in text for token in ("lease lost", "lease_expired", "heartbeat_failed")):
+        return FailureClass.LEASE_LOST
+    if any(token in text for token in ("out of memory", "disk full", "resource")):
+        return FailureClass.RESOURCE_UNAVAILABLE
+    if any(token in text for token in ("worker crash", "process exited", "signal")):
+        return FailureClass.WORKER_CRASH
+    if any(token in text for token in ("invalid release", "business permanent")):
+        return FailureClass.BUSINESS_PERMANENT
+    return FailureClass.UNKNOWN
+
+
+def repair_for_failure(failure: FailureClass) -> RepairDecision:
+    table = {
+        FailureClass.CAPABILITY_CONTRACT_ERROR: RepairDecision(
+            failure, RepairLevel.AUTO_REPAIR_TASK,
+            "create_same_lineage_capability_contract_repair", False,
+        ),
+        FailureClass.CREDENTIAL_PROJECTION_ERROR: RepairDecision(
+            failure, RepairLevel.AUTO_REPAIR_TASK,
+            "restore_service_private_credential_projection", False,
+        ),
+        FailureClass.DEPENDENCY_UNAVAILABLE: RepairDecision(
+            failure, RepairLevel.AUTO_LOCAL,
+            "retry_dependency_health_and_resume", False,
+        ),
+        FailureClass.RESOURCE_UNAVAILABLE: RepairDecision(
+            failure, RepairLevel.AUTO_LOCAL,
+            "heal_runtime_resource_and_resume", False,
+        ),
+        FailureClass.WORKER_CRASH: RepairDecision(
+            failure, RepairLevel.AUTO_LOCAL,
+            "restart_worker_and_resume_same_lineage", False,
+        ),
+        FailureClass.LEASE_LOST: RepairDecision(
+            failure, RepairLevel.AUTO_LOCAL,
+            "recover_lease_and_resume_same_lineage", False,
+        ),
+        FailureClass.POLICY_DENIED: RepairDecision(
+            failure, RepairLevel.OWNER_APPROVAL,
+            "request_bound_owner_authorization", False,
+        ),
+        FailureClass.BUSINESS_PERMANENT: RepairDecision(
+            failure, RepairLevel.MANUAL,
+            "terminal_business_failure", True,
+        ),
+        FailureClass.UNKNOWN: RepairDecision(
+            failure, RepairLevel.AUTO_REPAIR_TASK,
+            "create_diagnostic_repair_task", False,
+        ),
+    }
+    return table[failure]
+
+
+def _probe(
+    name: str,
+    probes: Mapping[str, Callable[[], bool]],
+) -> InvariantResult:
+    fn = probes.get(name)
+    if fn is None:
+        return InvariantResult(name, False, "PROBE_MISSING")
+    try:
+        passed = bool(fn())
+    except Exception as exc:
+        return InvariantResult(name, False, "PROBE_ERROR:" + type(exc).__name__)
+    return InvariantResult(name, passed, "PASS" if passed else "NOT_READY")
+
+
+def runtime_admission(
+    *,
+    worker_capabilities: Iterable[object],
+    required_capabilities: Iterable[object],
+    requested_operations: Sequence[str] = (),
+    probes: Mapping[str, Callable[[], bool]] | None = None,
+) -> AdmissionDecision:
+    """Admission is evaluated before claim; failures do not consume attempts."""
+
+    worker_values = tuple(worker_capabilities)
+    required_values = tuple(required_capabilities)
+    probe_map = probes or {}
+    invariants: list[InvariantResult] = []
+
+    registry = validate_capability_contract_registry()
+    invariants.append(InvariantResult(
+        "capability_contract_registry",
+        bool(registry.get("valid")),
+        "PASS" if registry.get("valid") else "REGISTRY_INVALID",
+    ))
+
+    task_known = all(registered_capability(item) for item in required_values)
+    worker_known = all(registered_capability(item, worker=True) for item in worker_values)
+    invariants.extend((
+        InvariantResult("task_capabilities_registered", task_known, "PASS" if task_known else "UNKNOWN_TASK_CAPABILITY"),
+        InvariantResult("worker_capabilities_registered", worker_known, "PASS" if worker_known else "UNKNOWN_WORKER_CAPABILITY"),
+    ))
+
+    capability_match = (
+        task_known
+        and worker_known
+        and has_capabilities(worker_values, required_values)
+    )
+    invariants.append(InvariantResult(
+        "capability_match",
+        capability_match,
+        "PASS" if capability_match else "WORKER_NOT_ELIGIBLE",
+    ))
+
+    normalized = normalize_required_capabilities(required_values)
+    contracts = [
+        capability_contract(item)
+        for item in normalized
+        if not str(item).startswith("exact:")
+    ]
+    contracts = [item for item in contracts if item is not None]
+
+    forbidden: set[str] = set()
+    dependencies: set[str] = set()
+    health: set[str] = set()
+    for contract in contracts:
+        forbidden.update(contract.forbidden)
+        dependencies.update(contract.dependencies)
+        health.update(contract.health_requirements)
+
+    requested = {str(item).strip().lower() for item in requested_operations if str(item).strip()}
+    forbidden_hit = sorted(requested & {item.lower() for item in forbidden})
+    invariants.append(InvariantResult(
+        "forbidden_operations_absent",
+        not forbidden_hit,
+        "PASS" if not forbidden_hit else "FORBIDDEN:" + ",".join(forbidden_hit),
+    ))
+
+    for name in sorted(dependencies | health):
+        invariants.append(_probe(name, probe_map))
+
+    failed = [item for item in invariants if not item.passed]
+    if not failed:
+        return AdmissionDecision(
+            True,
+            "EXECUTION_READY",
+            tuple(invariants),
+        )
+
+    details = " ".join(item.name + ":" + item.detail for item in failed)
+    failure = classify_failure(details, phase="admission")
+    if any(item.name == "forbidden_operations_absent" and not item.passed for item in failed):
+        failure = FailureClass.POLICY_DENIED
+    elif any("credential" in item.name for item in failed):
+        failure = FailureClass.CREDENTIAL_PROJECTION_ERROR
+    elif any(item.name in {"capability_contract_registry", "task_capabilities_registered", "worker_capabilities_registered", "capability_match"} for item in failed):
+        failure = FailureClass.CAPABILITY_CONTRACT_ERROR
+    elif any(item.name in dependencies or item.name in health for item in failed):
+        failure = FailureClass.DEPENDENCY_UNAVAILABLE
+
+    repair = repair_for_failure(failure)
+    return AdmissionDecision(
+        False,
+        "EXECUTION_READY_FAILED",
+        tuple(invariants),
+        failure,
+        repair.level,
+        repair.action,
+    )
+
+
+def system_invariant_snapshot(
+    *,
+    worker_capabilities: Iterable[object],
+    required_capabilities: Iterable[object],
+    requested_operations: Sequence[str] = (),
+    probes: Mapping[str, Callable[[], bool]] | None = None,
+) -> dict[str, object]:
+    decision = runtime_admission(
+        worker_capabilities=worker_capabilities,
+        required_capabilities=required_capabilities,
+        requested_operations=requested_operations,
+        probes=probes,
+    )
+    return {
+        "schema": "aios.system-invariants.v1",
+        "status": "HEALTHY" if decision.admitted else "DEGRADED",
+        "execution_ready": decision.admitted,
+        "admission": decision.as_dict(),
+    }

active-runtime/app/aios_capability_contract.py

@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 from collections.abc import Iterable
+from dataclasses import asdict, dataclass
+from typing import Any
 
 
 _TASK_ALIASES: dict[str, frozenset[str]] = {
@@ -37,6 +39,7 @@ _TASK_ALIASES: dict[str, frozenset[str]] = {
     "gpu.train": frozenset({"legacy.gpu.train"}),
     "agent.runtime.hermes": frozenset({"agent.runtime.hermes"}),
     "agent.runtime.openclaw": frozenset({"agent.runtime.openclaw"}),
+    "release:candidate": frozenset({"release:candidate"}),
 }
 
 _WORKER_ALIASES: dict[str, frozenset[str]] = {
@@ -71,8 +74,172 @@ _WORKER_ALIASES: dict[str, frozenset[str]] = {
     "gpu.train": frozenset({"legacy.gpu.train"}),
     "agent.runtime.hermes": frozenset({"agent.runtime.hermes"}),
     "agent.runtime.openclaw": frozenset({"agent.runtime.openclaw"}),
+    "release:candidate": frozenset({"release:candidate"}),
 }
 
+
+
+@dataclass(frozen=True)
+class CapabilityContract:
+    """Versioned runtime contract attached to one canonical semantic capability."""
+
+    name: str
+    min_version: int = 1
+    max_version: int = 1
+    permissions: tuple[str, ...] = ()
+    forbidden: tuple[str, ...] = ()
+    dependencies: tuple[str, ...] = ()
+    health_requirements: tuple[str, ...] = ()
+
+    def as_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+# This registry is the semantic source of truth. Alias tables below are retained
+# as compatibility adapters; startup/tests validate every alias target against
+# this registry so a name cannot be accepted in one layer and be unknown in
+# another.
+_CAPABILITY_CONTRACTS: dict[str, CapabilityContract] = {
+    "repo.read": CapabilityContract("repo.read", permissions=("repository_read",)),
+    "test.execute": CapabilityContract("test.execute", permissions=("test_execution",)),
+    "governed.git.write.branch": CapabilityContract(
+        "governed.git.write.branch",
+        permissions=("governed_branch_write",),
+        dependencies=("writer_lease", "governed_gitops"),
+    ),
+    "governed.git.write.file": CapabilityContract(
+        "governed.git.write.file",
+        permissions=("governed_file_write",),
+        dependencies=("writer_lease", "governed_gitops"),
+    ),
+    "governed.git.write": CapabilityContract(
+        "governed.git.write",
+        permissions=("governed_git_write",),
+        dependencies=("writer_lease", "governed_gitops"),
+    ),
+    "runtime.shell": CapabilityContract("runtime.shell", permissions=("runtime_shell",)),
+    "runtime.git": CapabilityContract("runtime.git", permissions=("runtime_git",)),
+    "runtime.python": CapabilityContract("runtime.python", permissions=("runtime_python",)),
+    "legacy.code.read": CapabilityContract("legacy.code.read"),
+    "legacy.code.review": CapabilityContract("legacy.code.review"),
+    "legacy.code.write": CapabilityContract("legacy.code.write"),
+    "legacy.code.write.patch": CapabilityContract("legacy.code.write.patch"),
+    "legacy.test.python": CapabilityContract("legacy.test.python"),
+    "legacy.test.shell.allowlisted": CapabilityContract("legacy.test.shell.allowlisted"),
+    "legacy.git.commit": CapabilityContract("legacy.git.commit"),
+    "legacy.git.diff.read": CapabilityContract("legacy.git.diff.read"),
+    "legacy.git.force_push": CapabilityContract(
+        "legacy.git.force_push",
+        forbidden=("protected_ref_bypass",),
+    ),
+    "legacy.deploy.production": CapabilityContract(
+        "legacy.deploy.production",
+        permissions=("production_deploy",),
+        dependencies=("owner_approval", "deployment_controller"),
+        health_requirements=("approval_envelope_verified", "runtime_acceptance_available"),
+    ),
+    "legacy.secret.transform": CapabilityContract("legacy.secret.transform"),
+    "legacy.secure_file_transformer": CapabilityContract("legacy.secure_file_transformer"),
+    "legacy.runtime.shadow": CapabilityContract("legacy.runtime.shadow"),
+    "legacy.review.security": CapabilityContract("legacy.review.security"),
+    "legacy.review.api-contract": CapabilityContract("legacy.review.api-contract"),
+    "legacy.gpu.train": CapabilityContract("legacy.gpu.train"),
+    "agent.runtime.hermes": CapabilityContract(
+        "agent.runtime.hermes",
+        dependencies=("hermes_runtime",),
+        health_requirements=("worker_liveness",),
+    ),
+    "agent.runtime.openclaw": CapabilityContract(
+        "agent.runtime.openclaw",
+        dependencies=("openclaw_runtime",),
+        health_requirements=("worker_liveness",),
+    ),
+    "release:candidate": CapabilityContract(
+        "release:candidate",
+        min_version=1,
+        max_version=1,
+        permissions=("prepare_release",),
+        forbidden=(
+            "production_deploy",
+            "submit_owner_approval",
+            "tag_publish",
+            "baseline_registry_finalize",
+            "runtime_switch",
+        ),
+        dependencies=(
+            "credential_projection",
+            "trusted_builder",
+        ),
+        health_requirements=(
+            "authenticated_probe",
+            "release_api_health",
+            "worker_liveness",
+        ),
+    ),
+}
+
+
+def capability_contract(name: object) -> CapabilityContract | None:
+    """Return the contract for one canonical or registered capability name."""
+
+    text = _clean(name)
+    direct = _CAPABILITY_CONTRACTS.get(text)
+    if direct is not None:
+        return direct
+    mapped = _TASK_ALIASES.get(text) or _WORKER_ALIASES.get(text)
+    if mapped is None or len(mapped) != 1:
+        return None
+    return _CAPABILITY_CONTRACTS.get(next(iter(mapped)))
+
+
+def registered_capability(name: object, *, worker: bool = False) -> bool:
+    """Fail-closed validation used by registration/task admission."""
+
+    text = _clean(name)
+    if "*" in text:
+        return False
+    aliases = _WORKER_ALIASES if worker else _TASK_ALIASES
+    mapped = aliases.get(text)
+    if mapped is None:
+        return text in _CAPABILITY_CONTRACTS
+    return all(item in _CAPABILITY_CONTRACTS for item in mapped)
+
+
+def validate_capability_contract_registry() -> dict[str, object]:
+    """Prove alias/contract closure and contract safety invariants."""
+
+    alias_targets: set[str] = set()
+    for mapping in (_TASK_ALIASES, _WORKER_ALIASES):
+        for values in mapping.values():
+            alias_targets.update(values)
+    missing = sorted(alias_targets - set(_CAPABILITY_CONTRACTS))
+    invalid_versions = sorted(
+        name
+        for name, contract in _CAPABILITY_CONTRACTS.items()
+        if contract.min_version < 1 or contract.max_version < contract.min_version
+    )
+    permission_forbidden_overlap = sorted(
+        name
+        for name, contract in _CAPABILITY_CONTRACTS.items()
+        if set(contract.permissions) & set(contract.forbidden)
+    )
+    return {
+        "schema": "aios.capability-contract-registry.v1",
+        "contract_count": len(_CAPABILITY_CONTRACTS),
+        "missing_alias_targets": missing,
+        "invalid_versions": invalid_versions,
+        "permission_forbidden_overlap": permission_forbidden_overlap,
+        "valid": not missing and not invalid_versions and not permission_forbidden_overlap,
+    }
+
+
+def capability_contract_snapshot() -> dict[str, dict[str, Any]]:
+    return {
+        name: contract.as_dict()
+        for name, contract in sorted(_CAPABILITY_CONTRACTS.items())
+    }
+
+
 _UNKNOWN_PREFIX = "exact:"
 
 
@@ -98,7 +265,11 @@ def normalize_required_capabilities(values: Iterable[object]) -> frozenset[str]:
         if "*" in text:
             result.add(_UNKNOWN_PREFIX + text)
         else:
-            result.update(_TASK_ALIASES.get(text, frozenset({_UNKNOWN_PREFIX + text})))
+            mapped = _TASK_ALIASES.get(text)
+            if mapped is None or any(item not in _CAPABILITY_CONTRACTS for item in mapped):
+                result.add(_UNKNOWN_PREFIX + text)
+            else:
+                result.update(mapped)
     return frozenset(result)
 
 
@@ -111,7 +282,7 @@ def normalize_worker_capabilities(values: Iterable[object]) -> frozenset[str]:
         if "*" in text:
             continue
         mapped = _WORKER_ALIASES.get(text)
-        if mapped is not None:
+        if mapped is not None and all(item in _CAPABILITY_CONTRACTS for item in mapped):
             result.update(mapped)
     if {
         "governed.git.write.branch",

active-runtime/app/aios_gpt_https_edge.py

@@ -0,0 +1,456 @@
+#!/usr/bin/env python3
+"""Disabled-by-default TLS edge for the read-only GPT Online Task ingress.
+
+The internal AIOS ingress remains loopback-only. This service is a separate edge
+that may bind non-loopback only when the owner explicitly approves it. It
+forwards only the tiny read-only ingress API and never proxies the canonical
+Bridge or Task API.
+"""
+
+from __future__ import annotations
+
+from collections import defaultdict, deque
+from dataclasses import dataclass
+import hashlib
+import hmac
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+import os
+from pathlib import Path
+import re
+import secrets
+import ssl
+import sys
+import threading
+import time
+import urllib.error
+import urllib.request
+from typing import Any, Mapping
+from urllib.parse import urlsplit
+
+UPSTREAM = "http://127.0.0.1:18945"
+MAX_BODY = 64 * 1024
+MAX_RESPONSE = 256 * 1024
+MAX_CONCURRENT_REQUESTS = 32
+REQUEST_TIMEOUT_SECONDS = 20
+RATE_WINDOW_SECONDS = 60
+RATE_LIMIT_REQUESTS = 60
+RATE_BURST_REQUESTS = 15
+RATE_MAX_CLIENTS = 4096
+_AUDIT_SALT = secrets.token_bytes(32)
+PUBLIC_APPROVAL_ENV = "AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED"
+PUBLIC_DOMAIN_ENV = "AIOS_GPT_EDGE_PUBLIC_DOMAIN"
+TASK_PATH_RE = re.compile(r"^/v1/read-only-tasks(?:/[A-Za-z0-9][A-Za-z0-9._:-]{7,127})?$")
+
+
+def _client_pseudonym(value: str) -> str:
+    return hashlib.sha256(_AUDIT_SALT + value.encode("utf-8", "replace")).hexdigest()[:16]
+
+
+def audit(event: str, **fields: object) -> None:
+    safe = {
+        "event": event,
+        "ts_unix_ms": int(time.time() * 1000),
+    }
+    for key, value in fields.items():
+        if key in {"authorization", "body", "token", "cookie", "client_ip"}:
+            continue
+        safe[key] = value
+    print(
+        json.dumps(safe, sort_keys=True, separators=(",", ":")),
+        file=sys.stderr,
+        flush=True,
+    )
+
+
+class EdgeError(RuntimeError):
+    def __init__(self, code: str, status: int = 400) -> None:
+        super().__init__(code)
+        self.code = code
+        self.status = status
+
+
+def canonical(value: object) -> bytes:
+    return json.dumps(
+        value,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    ).encode("utf-8")
+
+
+def public_bind_allowed(host: str, approval: str) -> str:
+    value = host.strip()
+    if value in {"127.0.0.1", "::1", "localhost"}:
+        return value
+    if approval != "1":
+        raise RuntimeError("GPT_EDGE_PUBLIC_BIND_OWNER_APPROVAL_REQUIRED")
+    if value not in {"0.0.0.0", "::"}:
+        raise RuntimeError("GPT_EDGE_BIND_ADDRESS_INVALID")
+    return value
+
+
+def validated_public_domain(value: str) -> str:
+    domain = value.strip().lower()
+    if not domain:
+        return ""
+    if len(domain) > 253 or re.fullmatch(
+        r"(?=.{1,253}\Z)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?",
+        domain,
+    ) is None:
+        raise RuntimeError("GPT_EDGE_PUBLIC_DOMAIN_INVALID")
+    return domain
+
+
+def validated_edge_identity(host: str, approval: str, domain: str) -> tuple[str, str]:
+    bind_host = public_bind_allowed(host, approval)
+    public_domain = validated_public_domain(domain)
+    if bind_host not in {"127.0.0.1", "::1", "localhost"} and not public_domain:
+        raise RuntimeError("GPT_EDGE_PUBLIC_DOMAIN_REQUIRED")
+    return bind_host, public_domain
+
+
+def validate_tls_file(path: Path, *, key: bool) -> Path:
+    if path.is_symlink() or not path.is_file():
+        raise RuntimeError("GPT_EDGE_TLS_FILE_INVALID")
+    size = path.stat().st_size
+    if size < 16 or size > 256 * 1024:
+        raise RuntimeError("GPT_EDGE_TLS_FILE_INVALID")
+    if key and (path.stat().st_mode & 0o077):
+        raise RuntimeError("GPT_EDGE_TLS_KEY_MODE_INVALID")
+    return path
+
+
+def build_tls_context(cert_path: Path, key_path: Path) -> ssl.SSLContext:
+    cert = validate_tls_file(cert_path, key=False)
+    key = validate_tls_file(key_path, key=True)
+    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+    context.minimum_version = ssl.TLSVersion.TLSv1_2
+    context.options |= ssl.OP_NO_COMPRESSION
+    context.load_cert_chain(certfile=str(cert), keyfile=str(key))
+    return context
+
+
+def allowed_route(method: str, raw_path: str) -> str:
+    parsed = urlsplit(raw_path)
+    if parsed.query or parsed.fragment:
+        raise EdgeError("QUERY_OR_FRAGMENT_DENY", 400)
+    path = parsed.path
+    if method == "GET" and path in {"/health", "/openapi.json"}:
+        return path
+    if method == "POST" and path == "/v1/read-only-tasks":
+        return path
+    if method == "GET" and TASK_PATH_RE.fullmatch(path):
+        return path
+    raise EdgeError("ROUTE_DENY", 404)
+
+
+def route_label(method: str, path: str) -> str:
+    if method == "GET" and path == "/health":
+        return "health"
+    if method == "GET" and path == "/openapi.json":
+        return "openapi"
+    if method == "POST" and path == "/v1/read-only-tasks":
+        return "task_create"
+    if method == "GET" and path.startswith("/v1/read-only-tasks/"):
+        return "task_status"
+    return "unknown"
+
+
+def forwarded_headers(headers: Mapping[str, str], *, body: bool) -> dict[str, str]:
+    result = {"Accept": "application/json"}
+    authorization = headers.get("Authorization")
+    if not isinstance(authorization, str) or not authorization.startswith("Bearer "):
+        raise EdgeError("AUTHORIZATION_REQUIRED", 401)
+    token = authorization[7:]
+    if len(token) < 32 or len(token) > 4096 or any(ch.isspace() for ch in token):
+        raise EdgeError("AUTHORIZATION_INVALID", 401)
+    result["Authorization"] = authorization
+    if body:
+        content_type = headers.get("Content-Type", "")
+        if content_type.split(";", 1)[0].strip().lower() != "application/json":
+            raise EdgeError("CONTENT_TYPE_INVALID", 415)
+        result["Content-Type"] = "application/json"
+    return result
+
+
+class RateLimiter:
+    def __init__(
+        self,
+        *,
+        window_seconds: int = RATE_WINDOW_SECONDS,
+        limit: int = RATE_LIMIT_REQUESTS,
+        burst: int = RATE_BURST_REQUESTS,
+        max_clients: int = RATE_MAX_CLIENTS,
+        now_fn=time.monotonic,
+    ) -> None:
+        self.window_seconds = window_seconds
+        self.limit = limit
+        self.burst = burst
+        self.max_clients = max_clients
+        self.now_fn = now_fn
+        self._lock = threading.Lock()
+        self._events: dict[str, deque[float]] = defaultdict(deque)
+
+    def _prune_stale(self, cutoff: float) -> None:
+        stale = [
+            key
+            for key, events in self._events.items()
+            if not events or events[-1] <= cutoff
+        ]
+        for key in stale:
+            self._events.pop(key, None)
+
+    def allowed(self, key: str) -> bool:
+        now = self.now_fn()
+        cutoff = now - self.window_seconds
+        with self._lock:
+            if key not in self._events and len(self._events) >= self.max_clients:
+                self._prune_stale(cutoff)
+                if len(self._events) >= self.max_clients:
+                    return False
+            events = self._events[key]
+            while events and events[0] <= cutoff:
+                events.popleft()
+            if len(events) >= self.limit + self.burst:
+                return False
+            events.append(now)
+            return True
+
+
+@dataclass(frozen=True)
+class Proxy:
+    upstream: str = UPSTREAM
+    timeout: int = REQUEST_TIMEOUT_SECONDS
+
+    def request(
+        self,
+        method: str,
+        path: str,
+        headers: Mapping[str, str],
+        body: bytes | None,
+    ) -> tuple[int, bytes, str]:
+        if self.upstream != UPSTREAM:
+            raise EdgeError("UPSTREAM_DRIFT_DENY", 503)
+        req = urllib.request.Request(
+            self.upstream + path,
+            method=method,
+            data=body,
+            headers=dict(headers),
+        )
+        try:
+            with urllib.request.urlopen(req, timeout=self.timeout) as response:
+                raw = response.read(MAX_RESPONSE + 1)
+                status = response.status
+                content_type = response.headers.get("Content-Type", "application/json")
+        except urllib.error.HTTPError as exc:
+            raw = exc.read(MAX_RESPONSE + 1)
+            status = exc.code
+            content_type = exc.headers.get("Content-Type", "application/json")
+        except Exception as exc:
+            raise EdgeError("UPSTREAM_UNAVAILABLE", 502) from exc
+        if len(raw) > MAX_RESPONSE:
+            raise EdgeError("UPSTREAM_RESPONSE_TOO_LARGE", 502)
+        if content_type.split(";", 1)[0].strip().lower() != "application/json":
+            raise EdgeError("UPSTREAM_CONTENT_TYPE_INVALID", 502)
+        return status, raw, "application/json"
+
+
+class EdgeServer(ThreadingHTTPServer):
+    daemon_threads = True
+    request_queue_size = 64
+    allow_reuse_address = True
+
+
+class Handler(BaseHTTPRequestHandler):
+    proxy = Proxy()
+    limiter = RateLimiter()
+    concurrency = threading.BoundedSemaphore(MAX_CONCURRENT_REQUESTS)
+    expected_domain = ""
+
+    server_version = "AIOSGPTIngressEdge/1.0"
+    sys_version = ""
+
+    def log_message(self, *_args: Any) -> None:
+        return
+
+    def respond_json(self, value: Mapping[str, Any], status: int) -> None:
+        raw = canonical(value)
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Cache-Control", "no-store")
+        self.send_header("X-Content-Type-Options", "nosniff")
+        self.send_header("Referrer-Policy", "no-referrer")
+        self.send_header("Content-Security-Policy", "default-src 'none'")
+        self.send_header("Strict-Transport-Security", "max-age=31536000")
+        self.send_header("Content-Length", str(len(raw)))
+        self.end_headers()
+        self.wfile.write(raw)
+
+    def send_proxy_response(self, status: int, raw: bytes, content_type: str) -> None:
+        self.send_response(status)
+        self.send_header("Content-Type", content_type)
+        self.send_header("Cache-Control", "no-store")
+        self.send_header("X-Content-Type-Options", "nosniff")
+        self.send_header("Referrer-Policy", "no-referrer")
+        self.send_header("Content-Security-Policy", "default-src 'none'")
+        self.send_header("Strict-Transport-Security", "max-age=31536000")
+        self.send_header("Content-Length", str(len(raw)))
+        self.end_headers()
+        self.wfile.write(raw)
+
+    def client_key(self) -> str:
+        return str(self.client_address[0])
+
+    def validate_host(self) -> None:
+        if not self.expected_domain:
+            return
+        host = self.headers.get("Host", "").split(":", 1)[0].lower()
+        if not hmac.compare_digest(host, self.expected_domain):
+            raise EdgeError("HOST_HEADER_DENY", 421)
+
+    def body(self) -> bytes:
+        try:
+            length = int(self.headers.get("Content-Length", "-1"))
+        except ValueError as exc:
+            raise EdgeError("CONTENT_LENGTH_INVALID") from exc
+        if length < 1 or length > MAX_BODY:
+            raise EdgeError("REQUEST_SIZE_INVALID", 413)
+        raw = self.rfile.read(length)
+        if len(raw) != length:
+            raise EdgeError("REQUEST_BODY_TRUNCATED", 400)
+        # Validate JSON at the edge so malformed bodies do not consume upstream work.
+        try:
+            value = json.loads(raw.decode("utf-8"))
+        except Exception as exc:
+            raise EdgeError("REQUEST_JSON_INVALID", 400) from exc
+        if not isinstance(value, Mapping):
+            raise EdgeError("REQUEST_JSON_OBJECT_REQUIRED", 400)
+        return canonical(value)
+
+    def handle_proxy(self, method: str) -> None:
+        client_id = _client_pseudonym(self.client_key())
+        started = time.monotonic()
+        acquired = self.concurrency.acquire(blocking=False)
+        if not acquired:
+            audit("EDGE_BUSY", client_id=client_id, method=method)
+            self.respond_json({"error": "EDGE_BUSY"}, 503)
+            return
+        label = "unknown"
+        try:
+            self.validate_host()
+            path = allowed_route(method, self.path)
+            label = route_label(method, path)
+            if path == "/health" and method == "GET":
+                self.respond_json(
+                    {
+                        "status": "ok",
+                        "service": "aios-gpt-https-edge",
+                        "upstream": "loopback-readonly-ingress",
+                    },
+                    200,
+                )
+                audit(
+                    "EDGE_REQUEST",
+                    client_id=client_id,
+                    method=method,
+                    route=label,
+                    status=200,
+                    duration_ms=int((time.monotonic() - started) * 1000),
+                )
+                return
+            if not self.limiter.allowed(self.client_key()):
+                audit(
+                    "EDGE_RATE_LIMIT",
+                    client_id=client_id,
+                    method=method,
+                    route=label,
+                )
+                self.respond_json({"error": "RATE_LIMITED"}, 429)
+                return
+            has_body = method == "POST"
+            headers = forwarded_headers(self.headers, body=has_body)
+            body = self.body() if has_body else None
+            status, raw, content_type = self.proxy.request(
+                method,
+                path,
+                headers,
+                body,
+            )
+            self.send_proxy_response(status, raw, content_type)
+            audit(
+                "EDGE_REQUEST",
+                client_id=client_id,
+                method=method,
+                route=label,
+                status=status,
+                duration_ms=int((time.monotonic() - started) * 1000),
+            )
+        except EdgeError as exc:
+            audit(
+                "EDGE_REJECT",
+                client_id=client_id,
+                method=method,
+                route=label,
+                status=exc.status,
+                error=exc.code,
+            )
+            self.respond_json({"error": exc.code}, exc.status)
+        finally:
+            self.concurrency.release()
+
+    def do_GET(self) -> None:
+        self.handle_proxy("GET")
+
+    def do_POST(self) -> None:
+        self.handle_proxy("POST")
+
+    def do_PUT(self) -> None:
+        self.respond_json({"error": "METHOD_DENY"}, 405)
+
+    def do_PATCH(self) -> None:
+        self.respond_json({"error": "METHOD_DENY"}, 405)
+
+    def do_DELETE(self) -> None:
+        self.respond_json({"error": "METHOD_DENY"}, 405)
+
+    def do_CONNECT(self) -> None:
+        self.respond_json({"error": "METHOD_DENY"}, 405)
+
+    def do_OPTIONS(self) -> None:
+        self.respond_json({"error": "METHOD_DENY"}, 405)
+
+
+def serve() -> None:
+    host, public_domain = validated_edge_identity(
+        os.environ.get("AIOS_GPT_EDGE_HOST", "127.0.0.1"),
+        os.environ.get(PUBLIC_APPROVAL_ENV, "0"),
+        os.environ.get(PUBLIC_DOMAIN_ENV, ""),
+    )
+    port = int(os.environ.get("AIOS_GPT_EDGE_PORT", "18443"))
+    if not 1024 <= port <= 65535:
+        raise RuntimeError("GPT_EDGE_PORT_INVALID")
+    Handler.expected_domain = public_domain
+
+    credentials = Path(os.environ.get("CREDENTIALS_DIRECTORY", "/run/credentials"))
+    cert_path = Path(
+        os.environ.get(
+            "AIOS_GPT_EDGE_TLS_CERT_FILE",
+            str(credentials / "tls_cert"),
+        )
+    )
+    key_path = Path(
+        os.environ.get(
+            "AIOS_GPT_EDGE_TLS_KEY_FILE",
+            str(credentials / "tls_key"),
+        )
+    )
+    context = build_tls_context(cert_path, key_path)
+
+    server = EdgeServer((host, port), Handler)
+    server.socket = context.wrap_socket(server.socket, server_side=True)
+    server.serve_forever()
+
+
+if __name__ == "__main__":
+    serve()

active-runtime/app/aios_gpt_https_edge_activation.py

@@ -0,0 +1,561 @@
+#!/usr/bin/env python3
+"""Owner-approved activation/rollback controller for the AIOS GPT HTTPS edge.
+
+This controller never creates approval, TLS, or ingress credentials. It consumes
+owner-managed inputs, applies one exact public-bind overlay, verifies the live
+edge, and provides an idempotent rollback path.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+import argparse
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import socket
+import ssl
+import stat
+import subprocess
+import tempfile
+import time
+import urllib.request
+from typing import Any, Mapping
+
+APPROVAL_SCHEMA = "aios.gpt-https-edge.public-approval.v1"
+STATE_SCHEMA = "aios.gpt-https-edge.activation-state.v1"
+
+OID_RE = re.compile(r"^[0-9a-f]{40}$")
+SHA_RE = re.compile(r"^[0-9a-f]{64}$")
+APPROVAL_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$")
+DOMAIN_RE = re.compile(
+    r"(?=.{1,253}\Z)"
+    r"(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+"
+    r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?"
+)
+
+EDGE_UNIT = "aios-gpt-https-edge.service"
+INGRESS_UNIT = "aios-gpt-online-task-ingress.service"
+PUBLIC_HOST = "0.0.0.0"
+PUBLIC_PORT = 18443
+INGRESS_HEALTH = "http://127.0.0.1:18945/health"
+DROPIN_NAME = "zzzzzzzzzz-owner-approved-public.conf"
+
+
+class ActivationError(RuntimeError):
+    pass
+
+
+def canonical(value: object) -> bytes:
+    return json.dumps(
+        value,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    ).encode("utf-8")
+
+
+def sha256_bytes(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def sha256_file(path: Path) -> str:
+    return sha256_bytes(path.read_bytes())
+
+
+def parse_time(value: object, *, field: str) -> datetime:
+    if not isinstance(value, str) or not value.strip():
+        raise ActivationError("APPROVAL_" + field.upper() + "_INVALID")
+    text = value.strip()
+    if text.endswith("Z"):
+        text = text[:-1] + "+00:00"
+    try:
+        parsed = datetime.fromisoformat(text)
+    except ValueError as exc:
+        raise ActivationError("APPROVAL_" + field.upper() + "_INVALID") from exc
+    if parsed.tzinfo is None:
+        raise ActivationError("APPROVAL_" + field.upper() + "_INVALID")
+    return parsed.astimezone(timezone.utc)
+
+
+def validate_domain(value: object) -> str:
+    if not isinstance(value, str):
+        raise ActivationError("APPROVAL_DOMAIN_INVALID")
+    domain = value.strip().lower()
+    if DOMAIN_RE.fullmatch(domain) is None:
+        raise ActivationError("APPROVAL_DOMAIN_INVALID")
+    return domain
+
+
+def validate_root_file(
+    path: Path,
+    *,
+    exact_mode: int | None = None,
+    max_size: int = 256 * 1024,
+) -> Path:
+    if path.is_symlink() or not path.is_file():
+        raise ActivationError("OWNER_FILE_INVALID")
+    info = path.stat()
+    if info.st_uid != 0 or info.st_gid != 0:
+        raise ActivationError("OWNER_FILE_OWNERSHIP_INVALID")
+    mode = stat.S_IMODE(info.st_mode)
+    if exact_mode is not None and mode != exact_mode:
+        raise ActivationError("OWNER_FILE_MODE_INVALID")
+    if info.st_size < 16 or info.st_size > max_size:
+        raise ActivationError("OWNER_FILE_SIZE_INVALID")
+    return path
+
+
+def certificate_dns_names(cert_path: Path) -> tuple[str, ...]:
+    try:
+        decoded = ssl._ssl._test_decode_cert(str(cert_path))
+    except Exception as exc:
+        raise ActivationError("TLS_CERT_DECODE_INVALID") from exc
+    names = [
+        value
+        for kind, value in decoded.get("subjectAltName", ())
+        if kind == "DNS"
+    ]
+    if not names:
+        for rdn in decoded.get("subject", ()):
+            for key, value in rdn:
+                if key == "commonName":
+                    names.append(value)
+    return tuple(names)
+
+
+def cert_matches_domain(cert_path: Path, domain: str) -> bool:
+    names = certificate_dns_names(cert_path)
+    try:
+        return bool(names) and any(ssl._dnsname_match(name, domain) for name in names)
+    except ssl.CertificateError:
+        return False
+
+
+def validate_tls_pair(
+    cert_path: Path,
+    key_path: Path,
+    domain: str,
+    *,
+    now_fn=time.time,
+) -> dict[str, Any]:
+    validate_root_file(cert_path)
+    validate_root_file(key_path, exact_mode=0o600)
+    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+    context.minimum_version = ssl.TLSVersion.TLSv1_2
+    context.options |= ssl.OP_NO_COMPRESSION
+    try:
+        context.load_cert_chain(str(cert_path), str(key_path))
+    except Exception as exc:
+        raise ActivationError("TLS_CERT_KEY_PAIR_INVALID") from exc
+    if not cert_matches_domain(cert_path, domain):
+        raise ActivationError("TLS_CERT_DOMAIN_MISMATCH")
+    decoded = ssl._ssl._test_decode_cert(str(cert_path))
+    not_after = decoded.get("notAfter")
+    if not not_after:
+        raise ActivationError("TLS_CERT_EXPIRY_MISSING")
+    remaining = ssl.cert_time_to_seconds(not_after) - now_fn()
+    if remaining < 7 * 86400:
+        raise ActivationError("TLS_CERT_EXPIRY_TOO_CLOSE")
+    return {
+        "cert_sha256": sha256_file(cert_path),
+        "remaining_days": int(remaining // 86400),
+    }
+
+
+@dataclass(frozen=True)
+class Approval:
+    approval_id: str
+    candidate_oid: str
+    domain: str
+    bind_host: str
+    port: int
+    tls_cert_sha256: str
+    owner: str
+    issued_at: datetime
+    not_before: datetime
+    expires_at: datetime
+    raw_sha256: str
+
+    @classmethod
+    def load(
+        cls,
+        path: Path,
+        *,
+        candidate_oid: str,
+        cert_path: Path,
+        now: datetime | None = None,
+    ) -> "Approval":
+        validate_root_file(path, exact_mode=0o600, max_size=64 * 1024)
+        try:
+            value = json.loads(path.read_text(encoding="utf-8"))
+        except Exception as exc:
+            raise ActivationError("APPROVAL_JSON_INVALID") from exc
+        if not isinstance(value, Mapping):
+            raise ActivationError("APPROVAL_JSON_INVALID")
+        allowed = {
+            "schema",
+            "approval_id",
+            "candidate_oid",
+            "domain",
+            "public_bind",
+            "bind_host",
+            "port",
+            "tls_cert_sha256",
+            "issued_at",
+            "not_before",
+            "expires_at",
+            "owner",
+        }
+        if set(value) != allowed or value.get("schema") != APPROVAL_SCHEMA:
+            raise ActivationError("APPROVAL_FIELDS_INVALID")
+
+        approval_id = value.get("approval_id")
+        if not isinstance(approval_id, str) or APPROVAL_ID_RE.fullmatch(approval_id) is None:
+            raise ActivationError("APPROVAL_ID_INVALID")
+        if not OID_RE.fullmatch(str(value.get("candidate_oid") or "")):
+            raise ActivationError("APPROVAL_CANDIDATE_INVALID")
+        if value["candidate_oid"] != candidate_oid:
+            raise ActivationError("APPROVAL_CANDIDATE_MISMATCH")
+        if value.get("public_bind") is not True:
+            raise ActivationError("APPROVAL_PUBLIC_BIND_REQUIRED")
+        if value.get("bind_host") != PUBLIC_HOST or value.get("port") != PUBLIC_PORT:
+            raise ActivationError("APPROVAL_BINDING_INVALID")
+
+        cert_sha = str(value.get("tls_cert_sha256") or "")
+        if SHA_RE.fullmatch(cert_sha) is None:
+            raise ActivationError("APPROVAL_CERT_SHA_INVALID")
+        if not cert_path.is_file() or sha256_file(cert_path) != cert_sha:
+            raise ActivationError("APPROVAL_CERT_SHA_MISMATCH")
+
+        domain = validate_domain(value.get("domain"))
+        owner = value.get("owner")
+        if not isinstance(owner, str) or not owner.strip() or len(owner) > 255:
+            raise ActivationError("APPROVAL_OWNER_INVALID")
+
+        issued_at = parse_time(value.get("issued_at"), field="issued_at")
+        not_before = parse_time(value.get("not_before"), field="not_before")
+        expires_at = parse_time(value.get("expires_at"), field="expires_at")
+        current = now or datetime.now(timezone.utc)
+        if not (issued_at <= not_before <= expires_at):
+            raise ActivationError("APPROVAL_TIME_ORDER_INVALID")
+        if current < not_before:
+            raise ActivationError("APPROVAL_NOT_YET_VALID")
+        if current >= expires_at:
+            raise ActivationError("APPROVAL_EXPIRED")
+        if expires_at - issued_at > __import__("datetime").timedelta(days=7):
+            raise ActivationError("APPROVAL_VALIDITY_TOO_LONG")
+
+        return cls(
+            approval_id=approval_id,
+            candidate_oid=candidate_oid,
+            domain=domain,
+            bind_host=PUBLIC_HOST,
+            port=PUBLIC_PORT,
+            tls_cert_sha256=cert_sha,
+            owner=owner.strip(),
+            issued_at=issued_at,
+            not_before=not_before,
+            expires_at=expires_at,
+            raw_sha256=sha256_file(path),
+        )
+
+
+def validate_release(release_root: Path) -> dict[str, Path]:
+    if release_root.is_symlink() or not release_root.is_dir():
+        raise ActivationError("IMMUTABLE_RELEASE_INVALID")
+    edge = release_root / "active-runtime/app/aios_gpt_https_edge.py"
+    controller = release_root / "active-runtime/app/aios_gpt_https_edge_activation.py"
+    for path in (edge, controller):
+        if path.is_symlink() or not path.is_file():
+            raise ActivationError("IMMUTABLE_RELEASE_INVALID")
+    return {"edge": edge, "controller": controller}
+
+
+def render_dropin(approval: Approval, release_root: Path) -> str:
+    app = release_root / "active-runtime/app/aios_gpt_https_edge.py"
+    return (
+        "[Service]\n"
+        f"Environment=AIOS_GPT_EDGE_HOST={approval.bind_host}\n"
+        f"Environment=AIOS_GPT_EDGE_PORT={approval.port}\n"
+        "Environment=AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=1\n"
+        f"Environment=AIOS_GPT_EDGE_PUBLIC_DOMAIN={approval.domain}\n"
+        "ExecStart=\n"
+        f"ExecStart=/usr/bin/python3 -B {app}\n"
+        f"ReadOnlyPaths={release_root}\n"
+    )
+
+
+def atomic_write(path: Path, data: str, mode: int = 0o600) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    fd, name = tempfile.mkstemp(prefix="." + path.name + ".", dir=path.parent)
+    try:
+        os.fchmod(fd, mode)
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            handle.write(data)
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.replace(name, path)
+    finally:
+        if os.path.exists(name):
+            os.unlink(name)
+
+
+class Systemd:
+    def run(self, *args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
+        return subprocess.run(
+            ["/usr/bin/systemctl", *args],
+            check=check,
+            text=True,
+            capture_output=True,
+            timeout=30,
+        )
+
+    def active(self, unit: str) -> bool:
+        return self.run("is-active", "--quiet", unit, check=False).returncode == 0
+
+    def enabled(self, unit: str) -> bool:
+        return self.run("is-enabled", "--quiet", unit, check=False).returncode == 0
+
+    def show(self, unit: str, prop: str) -> str:
+        return self.run("show", unit, f"--property={prop}", "--value").stdout.strip()
+
+
+def require_ingress_ready(systemd: Systemd) -> None:
+    if not systemd.active(INGRESS_UNIT):
+        raise ActivationError("INGRESS_SERVICE_NOT_ACTIVE")
+    try:
+        with urllib.request.urlopen(INGRESS_HEALTH, timeout=5) as response:
+            if response.status != 200:
+                raise ActivationError("INGRESS_HEALTH_FAILED")
+            value = json.loads(response.read().decode())
+    except ActivationError:
+        raise
+    except Exception as exc:
+        raise ActivationError("INGRESS_HEALTH_FAILED") from exc
+    if value.get("status") != "ok":
+        raise ActivationError("INGRESS_HEALTH_FAILED")
+
+
+def tls_probe(domain: str, cert_path: Path, *, port: int = PUBLIC_PORT) -> None:
+    context = ssl.create_default_context(cafile=str(cert_path))
+    context.check_hostname = True
+    context.verify_mode = ssl.CERT_REQUIRED
+    try:
+        with socket.create_connection(("127.0.0.1", port), timeout=5) as sock:
+            with context.wrap_socket(sock, server_hostname=domain) as tls_sock:
+                if tls_sock.version() not in {"TLSv1.2", "TLSv1.3"}:
+                    raise ActivationError("EDGE_TLS_VERSION_INVALID")
+    except ActivationError:
+        raise
+    except Exception as exc:
+        raise ActivationError("EDGE_TLS_PROBE_FAILED") from exc
+
+
+def receipt_for(
+    approval: Approval,
+    release_root: Path,
+    *,
+    state: str,
+) -> dict[str, Any]:
+    return {
+        "schema": STATE_SCHEMA,
+        "state": state,
+        "approval_id": approval.approval_id,
+        "approval_sha256": approval.raw_sha256,
+        "candidate_oid": approval.candidate_oid,
+        "domain": approval.domain,
+        "bind_host": approval.bind_host,
+        "port": approval.port,
+        "tls_cert_sha256": approval.tls_cert_sha256,
+        "release_root": str(release_root),
+        "updated_at": datetime.now(timezone.utc).isoformat(),
+    }
+
+
+def activate(
+    *,
+    approval: Approval,
+    release_root: Path,
+    cert_path: Path,
+    key_path: Path,
+    dropin_path: Path,
+    state_path: Path,
+    systemd: Systemd,
+) -> dict[str, Any]:
+    if os.geteuid() != 0:
+        raise ActivationError("ROOT_REQUIRED")
+    release = validate_release(release_root)
+    app = release["edge"]
+    require_ingress_ready(systemd)
+    validate_tls_pair(cert_path, key_path, approval.domain)
+
+    if state_path.is_file():
+        try:
+            existing = json.loads(state_path.read_text(encoding="utf-8"))
+        except Exception:
+            existing = {}
+        if (
+            existing.get("schema") == STATE_SCHEMA
+            and existing.get("state") == "ACTIVE"
+            and existing.get("approval_sha256") == approval.raw_sha256
+            and existing.get("candidate_oid") == approval.candidate_oid
+            and systemd.active(EDGE_UNIT)
+        ):
+            return existing
+
+    dropin = render_dropin(approval, release_root)
+    modified = False
+    try:
+        atomic_write(dropin_path, dropin, mode=0o644)
+        modified = True
+        systemd.run("daemon-reload")
+        systemd.run("enable", EDGE_UNIT)
+        systemd.run("restart", EDGE_UNIT)
+        if not systemd.active(EDGE_UNIT):
+            raise ActivationError("EDGE_SERVICE_NOT_ACTIVE")
+        exec_start = systemd.show(EDGE_UNIT, "ExecStart")
+        environment = systemd.show(EDGE_UNIT, "Environment")
+        if str(app) not in exec_start:
+            raise ActivationError("EDGE_EFFECTIVE_EXECSTART_INVALID")
+        for expected in (
+            f"AIOS_GPT_EDGE_HOST={approval.bind_host}",
+            f"AIOS_GPT_EDGE_PORT={approval.port}",
+            "AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=1",
+            f"AIOS_GPT_EDGE_PUBLIC_DOMAIN={approval.domain}",
+        ):
+            if expected not in environment:
+                raise ActivationError("EDGE_EFFECTIVE_ENVIRONMENT_INVALID")
+        tls_probe(approval.domain, cert_path, port=approval.port)
+        receipt = receipt_for(approval, release_root, state="ACTIVE")
+        atomic_write(state_path, json.dumps(receipt, indent=2, sort_keys=True) + "\n")
+        return receipt
+    except Exception:
+        if modified:
+            rollback(
+                dropin_path=dropin_path,
+                state_path=state_path,
+                systemd=systemd,
+                reason="activation_failed",
+            )
+        raise
+
+
+def rollback(
+    *,
+    dropin_path: Path,
+    state_path: Path,
+    systemd: Systemd,
+    reason: str,
+) -> dict[str, Any]:
+    if os.geteuid() != 0:
+        raise ActivationError("ROOT_REQUIRED")
+    systemd.run("stop", EDGE_UNIT, check=False)
+    systemd.run("disable", EDGE_UNIT, check=False)
+    dropin_path.unlink(missing_ok=True)
+    systemd.run("daemon-reload")
+    if systemd.active(EDGE_UNIT):
+        raise ActivationError("EDGE_ROLLBACK_SERVICE_STILL_ACTIVE")
+    previous: dict[str, Any] = {}
+    if state_path.is_file():
+        try:
+            previous = json.loads(state_path.read_text(encoding="utf-8"))
+        except Exception:
+            previous = {}
+    receipt = {
+        "schema": STATE_SCHEMA,
+        "state": "ROLLED_BACK",
+        "reason": reason[:255],
+        "candidate_oid": previous.get("candidate_oid"),
+        "approval_id": previous.get("approval_id"),
+        "domain": previous.get("domain"),
+        "updated_at": datetime.now(timezone.utc).isoformat(),
+    }
+    atomic_write(state_path, json.dumps(receipt, indent=2, sort_keys=True) + "\n")
+    return receipt
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("command", choices=("check", "activate", "rollback"))
+    parser.add_argument("--candidate")
+    parser.add_argument("--approval", default="/etc/aios-gitops/gpt-https-edge.approved")
+    parser.add_argument("--tls-cert", default="/etc/aios-gitops/gpt-https-edge-tls.crt")
+    parser.add_argument("--tls-key", default="/etc/aios-gitops/gpt-https-edge-tls.key")
+    parser.add_argument("--release-root")
+    parser.add_argument(
+        "--dropin",
+        default=f"/etc/systemd/system/{EDGE_UNIT}.d/{DROPIN_NAME}",
+    )
+    parser.add_argument(
+        "--state",
+        default="/var/lib/aios-gpt-https-edge-activation/state.json",
+    )
+    parser.add_argument("--reason", default="owner_requested")
+    args = parser.parse_args()
+
+    systemd = Systemd()
+    if args.command == "rollback":
+        value = rollback(
+            dropin_path=Path(args.dropin),
+            state_path=Path(args.state),
+            systemd=systemd,
+            reason=args.reason,
+        )
+        print(json.dumps(value, indent=2, sort_keys=True))
+        return 0
+
+    candidate = str(args.candidate or "")
+    if OID_RE.fullmatch(candidate) is None:
+        raise SystemExit("CANDIDATE_REQUIRED")
+    release_root = (
+        Path(args.release_root)
+        if args.release_root
+        else Path(f"/opt/aios/releases/{candidate}-gpt-https-edge-public")
+    )
+    cert_path = Path(args.tls_cert)
+    key_path = Path(args.tls_key)
+    approval = Approval.load(
+        Path(args.approval),
+        candidate_oid=candidate,
+        cert_path=cert_path,
+    )
+    tls = validate_tls_pair(cert_path, key_path, approval.domain)
+    release = validate_release(release_root)
+    require_ingress_ready(systemd)
+
+    if args.command == "check":
+        value = {
+            "status": "READY",
+            "candidate_oid": candidate,
+            "approval_id": approval.approval_id,
+            "approval_sha256": approval.raw_sha256,
+            "domain": approval.domain,
+            "bind_host": approval.bind_host,
+            "port": approval.port,
+            "edge_sha256": sha256_file(release["edge"]),
+            "controller_sha256": sha256_file(release["controller"]),
+            **tls,
+            "public_network_mutation": False,
+        }
+        print(json.dumps(value, indent=2, sort_keys=True))
+        return 0
+
+    value = activate(
+        approval=approval,
+        release_root=release_root,
+        cert_path=cert_path,
+        key_path=key_path,
+        dropin_path=Path(args.dropin),
+        state_path=Path(args.state),
+        systemd=systemd,
+    )
+    print(json.dumps(value, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

active-runtime/app/aios_gpt_online_task_ingress.py

@@ -0,0 +1,816 @@
+#!/usr/bin/env python3
+"""Credential-isolated host-side ingress for externally scheduled read-only Online Tasks.
+
+This service is intentionally not a public AIOS control plane. It accepts a tiny,
+authenticated external intent and translates it to the canonical loopback Bridge.
+Callers cannot select repository coordinates, capabilities, execution action,
+worker, branch policy, credentials, verification decisions, or write permissions.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+import hashlib
+import hmac
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+import os
+from pathlib import Path
+import re
+import sqlite3
+import threading
+import urllib.error
+import urllib.request
+from typing import Any, Mapping
+
+SCHEMA = "aios.gpt-online-task-ingress.request.v1"
+RESPONSE_SCHEMA = "aios.gpt-online-task-ingress.response.v1"
+OPENAPI_VERSION = "3.1.0"
+
+REQUEST_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$")
+TASK_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
+JOB_ID_RE = re.compile(r"^job-[0-9a-f]{32}$")
+OID_RE = re.compile(r"^[0-9a-f]{40}$")
+MAX_BODY = 64 * 1024
+MAX_GOAL = 12_000
+MAX_INSTRUCTIONS = 24_000
+MAX_CRITERIA = 16
+MAX_CRITERION_TEXT = 1_000
+
+READONLY_CAPABILITIES = ("git.read", "git.test")
+REPOSITORY = "aios/aios"
+
+
+class IngressError(RuntimeError):
+    def __init__(self, code: str, status: int = 400) -> None:
+        super().__init__(code)
+        self.code = code
+        self.status = status
+
+
+def canonical(value: object) -> bytes:
+    return json.dumps(
+        value,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    ).encode("utf-8")
+
+
+def digest(value: object) -> str:
+    return hashlib.sha256(canonical(value)).hexdigest()
+
+
+def now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def load_secret(path: Path, *, code: str) -> str:
+    if path.is_symlink() or not path.is_file():
+        raise RuntimeError(code + "_UNAVAILABLE")
+    value = path.read_text(encoding="utf-8").strip()
+    if len(value) < 32 or "\n" in value or "\r" in value:
+        raise RuntimeError(code + "_INVALID")
+    return value
+
+
+def normalized_request(value: Mapping[str, Any]) -> dict[str, Any]:
+    allowed = {
+        "schema",
+        "request_id",
+        "goal",
+        "instructions",
+        "success_criteria",
+        "constraints",
+    }
+    if set(value) - allowed:
+        raise IngressError("UNKNOWN_FIELDS")
+    if value.get("schema") != SCHEMA:
+        raise IngressError("SCHEMA_INVALID")
+
+    request_id = value.get("request_id")
+    if not isinstance(request_id, str) or REQUEST_ID_RE.fullmatch(request_id) is None:
+        raise IngressError("REQUEST_ID_INVALID")
+
+    goal = value.get("goal")
+    if not isinstance(goal, str) or not goal.strip() or len(goal) > MAX_GOAL:
+        raise IngressError("GOAL_INVALID")
+    instructions = value.get("instructions", "")
+    if not isinstance(instructions, str) or len(instructions) > MAX_INSTRUCTIONS:
+        raise IngressError("INSTRUCTIONS_INVALID")
+
+    criteria = value.get("success_criteria") or []
+    if not isinstance(criteria, list) or len(criteria) > MAX_CRITERIA:
+        raise IngressError("SUCCESS_CRITERIA_INVALID")
+    normalized_criteria = []
+    for item in criteria:
+        if not isinstance(item, Mapping) or set(item) != {"id", "description"}:
+            raise IngressError("SUCCESS_CRITERIA_INVALID")
+        criterion_id = item.get("id")
+        description = item.get("description")
+        if (
+            not isinstance(criterion_id, str)
+            or not criterion_id.strip()
+            or len(criterion_id) > 120
+            or not isinstance(description, str)
+            or not description.strip()
+            or len(description) > MAX_CRITERION_TEXT
+        ):
+            raise IngressError("SUCCESS_CRITERIA_INVALID")
+        normalized_criteria.append(
+            {"id": criterion_id.strip(), "description": description.strip()}
+        )
+
+    constraints = value.get("constraints") or {}
+    if not isinstance(constraints, Mapping):
+        raise IngressError("CONSTRAINTS_INVALID")
+    allowed_constraints = {
+        "max_files",
+        "focus_paths",
+        "notes",
+    }
+    if set(constraints) - allowed_constraints:
+        raise IngressError("CONSTRAINTS_FIELDS_INVALID")
+    normalized_constraints: dict[str, Any] = {}
+    if "max_files" in constraints:
+        max_files = constraints["max_files"]
+        if (
+            not isinstance(max_files, int)
+            or isinstance(max_files, bool)
+            or not 1 <= max_files <= 500
+        ):
+            raise IngressError("MAX_FILES_INVALID")
+        normalized_constraints["max_files"] = max_files
+    if "focus_paths" in constraints:
+        paths = constraints["focus_paths"]
+        if (
+            not isinstance(paths, list)
+            or len(paths) > 32
+            or not all(
+                isinstance(path, str)
+                and path
+                and len(path) <= 512
+                and not path.startswith("/")
+                and ".." not in Path(path).parts
+                for path in paths
+            )
+        ):
+            raise IngressError("FOCUS_PATHS_INVALID")
+        normalized_constraints["focus_paths"] = paths
+    if "notes" in constraints:
+        notes = constraints["notes"]
+        if not isinstance(notes, str) or len(notes) > 4_000:
+            raise IngressError("CONSTRAINT_NOTES_INVALID")
+        normalized_constraints["notes"] = notes
+
+    return {
+        "schema": SCHEMA,
+        "request_id": request_id,
+        "goal": goal.strip(),
+        "instructions": instructions,
+        "success_criteria": normalized_criteria,
+        "constraints": normalized_constraints,
+    }
+
+
+def validated_bind_host(value: str) -> str:
+    host = value.strip()
+    if host not in {"127.0.0.1", "::1", "localhost"}:
+        raise RuntimeError("GPT_INGRESS_NON_LOOPBACK_BIND_DENY")
+    return host
+
+
+def task_id_for(request_id: str) -> str:
+    suffix = hashlib.sha256(request_id.encode("utf-8")).hexdigest()[:24]
+    return "gpt-readonly-" + suffix
+
+
+def branch_for(task_id: str) -> str:
+    return "agent/external-gpt-readonly/" + task_id[-24:]
+
+
+@dataclass(frozen=True)
+class Upstream:
+    bridge_base: str
+    task_base: str
+    control_token: str
+    timeout: int = 20
+
+    def _request(
+        self,
+        base: str,
+        method: str,
+        path: str,
+        body: Mapping[str, Any] | None = None,
+    ) -> tuple[int, dict[str, Any]]:
+        raw_body = None if body is None else canonical(body)
+        headers = {
+            "Authorization": "Bearer " + self.control_token,
+            "Accept": "application/json",
+        }
+        if raw_body is not None:
+            headers["Content-Type"] = "application/json"
+        req = urllib.request.Request(
+            base.rstrip("/") + path,
+            method=method,
+            data=raw_body,
+            headers=headers,
+        )
+        try:
+            with urllib.request.urlopen(req, timeout=self.timeout) as response:
+                raw = response.read()
+                status = response.status
+        except urllib.error.HTTPError as exc:
+            raw = exc.read()
+            status = exc.code
+        except Exception as exc:
+            raise IngressError("UPSTREAM_TRANSPORT_FAILED", 503) from exc
+        try:
+            value = json.loads(raw.decode("utf-8"))
+        except Exception as exc:
+            raise IngressError("UPSTREAM_RESPONSE_INVALID", 502) from exc
+        if not isinstance(value, Mapping):
+            raise IngressError("UPSTREAM_RESPONSE_INVALID", 502)
+        return status, dict(value)
+
+    def control_snapshot(self) -> dict[str, Any]:
+        status, value = self._request(
+            self.bridge_base,
+            "GET",
+            "/v2/control-plane-snapshot",
+        )
+        if status != 200:
+            raise IngressError("CONTROL_SNAPSHOT_UNAVAILABLE", 503)
+        return value
+
+    def create_governed_task(
+        self,
+        body: Mapping[str, Any],
+    ) -> tuple[int, dict[str, Any]]:
+        return self._request(
+            self.bridge_base,
+            "POST",
+            "/v2/governed-task",
+            body,
+        )
+
+    def execution_envelope(
+        self,
+        body: Mapping[str, Any],
+    ) -> tuple[int, dict[str, Any]]:
+        return self._request(
+            self.bridge_base,
+            "POST",
+            "/v2/execution-envelope",
+            body,
+        )
+
+    def get_task(self, task_id: str) -> tuple[int, dict[str, Any]]:
+        return self._request(
+            self.task_base,
+            "GET",
+            "/v1/tasks/" + task_id,
+        )
+
+
+class Store:
+    def __init__(self, path: Path) -> None:
+        self.path = Path(path)
+        self._lock = threading.RLock()
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        with self._connect() as con:
+            con.execute(
+                """
+                CREATE TABLE IF NOT EXISTS external_task_requests(
+                  request_id TEXT PRIMARY KEY,
+                  request_sha256 TEXT NOT NULL,
+                  task_id TEXT NOT NULL UNIQUE,
+                  job_id TEXT,
+                  base_oid TEXT,
+                  state TEXT NOT NULL,
+                  created_at TEXT NOT NULL,
+                  updated_at TEXT NOT NULL,
+                  response_json TEXT
+                )
+                """
+            )
+            con.commit()
+
+    def _connect(self) -> sqlite3.Connection:
+        con = sqlite3.connect(self.path, timeout=30)
+        con.row_factory = sqlite3.Row
+        con.execute("PRAGMA busy_timeout=30000")
+        return con
+
+    def get(self, request_id: str) -> sqlite3.Row | None:
+        with self._connect() as con:
+            return con.execute(
+                "SELECT * FROM external_task_requests WHERE request_id=?",
+                (request_id,),
+            ).fetchone()
+
+    def reserve(self, request: Mapping[str, Any], task_id: str) -> sqlite3.Row:
+        request_sha = digest(request)
+        timestamp = now()
+        with self._lock, self._connect() as con:
+            con.execute("BEGIN IMMEDIATE")
+            row = con.execute(
+                "SELECT * FROM external_task_requests WHERE request_id=?",
+                (request["request_id"],),
+            ).fetchone()
+            if row is not None:
+                if row["request_sha256"] != request_sha:
+                    raise IngressError("IDEMPOTENCY_CONFLICT", 409)
+                con.commit()
+                return row
+            con.execute(
+                """
+                INSERT INTO external_task_requests(
+                  request_id,request_sha256,task_id,state,created_at,updated_at
+                ) VALUES(?,?,?,?,?,?)
+                """,
+                (
+                    request["request_id"],
+                    request_sha,
+                    task_id,
+                    "PENDING",
+                    timestamp,
+                    timestamp,
+                ),
+            )
+            con.commit()
+        row = self.get(request["request_id"])
+        assert row is not None
+        return row
+
+    def bound(
+        self,
+        request_id: str,
+        *,
+        job_id: str,
+        base_oid: str,
+        response: Mapping[str, Any],
+    ) -> None:
+        with self._lock, self._connect() as con:
+            con.execute(
+                """
+                UPDATE external_task_requests
+                   SET job_id=?,base_oid=?,state='BOUND',response_json=?,updated_at=?
+                 WHERE request_id=?
+                """,
+                (
+                    job_id,
+                    base_oid,
+                    json.dumps(response, sort_keys=True, separators=(",", ":")),
+                    now(),
+                    request_id,
+                ),
+            )
+            con.commit()
+
+
+class IngressService:
+    def __init__(self, upstream: Upstream, store: Store) -> None:
+        self.upstream = upstream
+        self.store = store
+
+    @staticmethod
+    def _main_oid(snapshot: Mapping[str, Any]) -> str:
+        candidates = [
+            snapshot.get("authoritative_head_oid"),
+            snapshot.get("head"),
+            (snapshot.get("aggregate") or {}).get("authoritative_head_oid")
+            if isinstance(snapshot.get("aggregate"), Mapping)
+            else None,
+        ]
+        for candidate in candidates:
+            if isinstance(candidate, str) and OID_RE.fullmatch(candidate):
+                return candidate
+        raise IngressError("AUTHORITATIVE_HEAD_UNAVAILABLE", 503)
+
+    @staticmethod
+    def _public_response(
+        request_id: str,
+        task_id: str,
+        job_id: str,
+        state: str,
+    ) -> dict[str, Any]:
+        return {
+            "schema": RESPONSE_SCHEMA,
+            "request_id": request_id,
+            "task_id": task_id,
+            "job_id": job_id,
+            "state": state,
+            "execution_mode": "source_capture",
+            "capabilities": list(READONLY_CAPABILITIES),
+            "write_access": False,
+            "official_verification_required": True,
+        }
+
+    def _recover_existing(
+        self,
+        request: Mapping[str, Any],
+        row: sqlite3.Row,
+    ) -> dict[str, Any] | None:
+        task_id = str(row["task_id"])
+        status, task = self.upstream.get_task(task_id)
+        if status == 404:
+            return None
+        if status != 200:
+            raise IngressError("TASK_RECOVERY_FAILED", 503)
+        metadata = task.get("metadata")
+        execution = (
+            metadata.get("execution")
+            if isinstance(metadata, Mapping)
+            and isinstance(metadata.get("execution"), Mapping)
+            else {}
+        )
+        job_id = str(execution.get("job_id") or row["job_id"] or "")
+        base_oid = str(execution.get("base_oid") or row["base_oid"] or "")
+        if (
+            task.get("task_id") != task_id
+            or not JOB_ID_RE.fullmatch(job_id)
+            or not OID_RE.fullmatch(base_oid)
+            or execution.get("action") != "source_capture"
+            or sorted(execution.get("capability_scope") or [])
+            != sorted(READONLY_CAPABILITIES)
+            or (execution.get("governance") or {}).get("writer_lease") is not False
+        ):
+            raise IngressError("TASK_RECOVERY_BINDING_INVALID", 409)
+        response = self._public_response(
+            request["request_id"],
+            task_id,
+            job_id,
+            str(task.get("status") or "BOUND"),
+        )
+        self.store.bound(
+            request["request_id"],
+            job_id=job_id,
+            base_oid=base_oid,
+            response=response,
+        )
+        return response
+
+    def create(self, raw: Mapping[str, Any]) -> dict[str, Any]:
+        request = normalized_request(raw)
+        task_id = task_id_for(request["request_id"])
+        row = self.store.reserve(request, task_id)
+        if row["state"] == "BOUND" and row["response_json"]:
+            return json.loads(row["response_json"])
+
+        recovered = self._recover_existing(request, row)
+        if recovered is not None:
+            return recovered
+
+        snapshot = self.upstream.control_snapshot()
+        base_oid = self._main_oid(snapshot)
+        target_branch = branch_for(task_id)
+        binding_key = "gpt-readonly-binding-" + digest(
+            {"request_id": request["request_id"], "base_oid": base_oid}
+        )[:24]
+        constraints = {
+            **request["constraints"],
+            "repository_mutation": "forbidden",
+            "control_plane_side_effects": "forbidden",
+            "credential_access": "forbidden",
+        }
+        body = {
+            "task_id": task_id,
+            "repository": REPOSITORY,
+            "base_oid": base_oid,
+            "expected_branch_oid": base_oid,
+            "target_branch": target_branch,
+            "branch_mode": "CREATE_NEW",
+            "generation": 1,
+            "risk": "low",
+            "description": "Externally requested read-only governed Online Task.",
+            "idempotency_key": binding_key,
+            "priority": 0,
+            "max_attempts": 3,
+            "require_online_task_projection": True,
+            "online_task": {
+                "goal": request["goal"],
+                "instructions": request["instructions"],
+                "success_criteria": request["success_criteria"],
+                "constraints": constraints,
+                "required_capabilities": list(READONLY_CAPABILITIES),
+                "priority": 0,
+                "max_attempts": 3,
+            },
+            "execution": {
+                "action": "source_capture",
+                "adapter": "source_capture",
+                "capability_scope": list(READONLY_CAPABILITIES),
+                "source_signals": [
+                    "external read-only source analysis request",
+                    "no repository mutation no credential access no control-plane side effects",
+                ],
+            },
+        }
+        status, value = self.upstream.create_governed_task(body)
+        if status not in {200, 201, 202}:
+            if status == 409:
+                recovered = self._recover_existing(request, row)
+                if recovered is not None:
+                    return recovered
+            raise IngressError(
+                "GOVERNED_TASK_CREATE_FAILED",
+                409 if status == 409 else 503,
+            )
+        job_id = str(value.get("job_id") or "")
+        if value.get("task_id") != task_id or JOB_ID_RE.fullmatch(job_id) is None:
+            raise IngressError("GOVERNED_TASK_RESPONSE_INVALID", 502)
+
+        envelope_status, envelope = self.upstream.execution_envelope(
+            {
+                "task_id": task_id,
+                "job_id": job_id,
+                "required_capabilities": list(READONLY_CAPABILITIES),
+                "execution": body["execution"],
+            }
+        )
+        if envelope_status != 200:
+            raise IngressError("EXECUTION_PROJECTION_FAILED", 503)
+        if (
+            envelope.get("task_id") != task_id
+            or envelope.get("job_id") != job_id
+            or envelope.get("base_oid") != base_oid
+            or envelope.get("action") != "source_capture"
+            or sorted(envelope.get("capability_scope") or [])
+            != sorted(READONLY_CAPABILITIES)
+            or (envelope.get("governance") or {}).get("writer_lease") is not False
+        ):
+            raise IngressError("EXECUTION_PROJECTION_INVALID", 502)
+
+        response = self._public_response(
+            request["request_id"],
+            task_id,
+            job_id,
+            "BOUND",
+        )
+        self.store.bound(
+            request["request_id"],
+            job_id=job_id,
+            base_oid=base_oid,
+            response=response,
+        )
+        return response
+
+    def status(self, request_id: str) -> dict[str, Any]:
+        if REQUEST_ID_RE.fullmatch(request_id) is None:
+            raise IngressError("REQUEST_ID_INVALID")
+        row = self.store.get(request_id)
+        if row is None:
+            raise IngressError("REQUEST_NOT_FOUND", 404)
+        task_id = str(row["task_id"])
+        status, task = self.upstream.get_task(task_id)
+        if status == 200 and not row["job_id"]:
+            metadata = task.get("metadata")
+            execution = (
+                metadata.get("execution")
+                if isinstance(metadata, Mapping)
+                and isinstance(metadata.get("execution"), Mapping)
+                else {}
+            )
+            job_id = str(execution.get("job_id") or "")
+            base_oid = str(execution.get("base_oid") or "")
+            if (
+                JOB_ID_RE.fullmatch(job_id)
+                and OID_RE.fullmatch(base_oid)
+                and execution.get("action") == "source_capture"
+                and sorted(execution.get("capability_scope") or [])
+                == sorted(READONLY_CAPABILITIES)
+                and (execution.get("governance") or {}).get("writer_lease") is False
+            ):
+                recovered = self._public_response(
+                    request_id,
+                    task_id,
+                    job_id,
+                    str(task.get("status") or "BOUND"),
+                )
+                self.store.bound(
+                    request_id,
+                    job_id=job_id,
+                    base_oid=base_oid,
+                    response=recovered,
+                )
+                row = self.store.get(request_id) or row
+        if status == 404:
+            return {
+                "schema": RESPONSE_SCHEMA,
+                "request_id": request_id,
+                "task_id": task_id,
+                "state": "PENDING",
+                "official_verification_required": True,
+            }
+        if status != 200:
+            raise IngressError("TASK_STATUS_UNAVAILABLE", 503)
+
+        result = task.get("result") if isinstance(task.get("result"), Mapping) else {}
+        verification = (
+            task.get("verification")
+            if isinstance(task.get("verification"), Mapping)
+            else {}
+        )
+        safe = {
+            "schema": RESPONSE_SCHEMA,
+            "request_id": request_id,
+            "task_id": task_id,
+            "job_id": str(row["job_id"] or ""),
+            "state": str(task.get("status") or "UNKNOWN"),
+            "attempt": int(task.get("attempt") or 0),
+            "claimed": bool(task.get("claimed_by")),
+            "result_status": result.get("status"),
+            "result_classification": result.get("classification"),
+            "official_verification": verification.get("decision"),
+            "official_classification": verification.get("classification"),
+            "write_access": False,
+        }
+        if result.get("execution_mode") == "source_capture":
+            snapshot = result.get("source_snapshot")
+            if isinstance(snapshot, Mapping):
+                safe["source_snapshot_sha256"] = snapshot.get("payload_sha256")
+                safe["source_snapshot_file_count"] = snapshot.get("file_count")
+        return safe
+
+
+def openapi_document() -> dict[str, Any]:
+    return {
+        "openapi": OPENAPI_VERSION,
+        "info": {
+            "title": "AIOS Read-Only Online Task Ingress",
+            "version": "1.0.0",
+        },
+        "paths": {
+            "/v1/read-only-tasks": {
+                "post": {
+                    "operationId": "createReadOnlyOnlineTask",
+                    "security": [{"bearerAuth": []}],
+                    "requestBody": {"required": True},
+                    "responses": {"200": {"description": "Task accepted"}},
+                }
+            },
+            "/v1/read-only-tasks/{request_id}": {
+                "get": {
+                    "operationId": "getReadOnlyOnlineTaskStatus",
+                    "security": [{"bearerAuth": []}],
+                    "responses": {"200": {"description": "Sanitized task status"}},
+                }
+            },
+        },
+        "components": {
+            "securitySchemes": {
+                "bearerAuth": {
+                    "type": "http",
+                    "scheme": "bearer",
+                }
+            }
+        },
+    }
+
+
+class Handler(BaseHTTPRequestHandler):
+    service: IngressService | None = None
+    ingress_token: str = ""
+
+    def log_message(self, *_args: Any) -> None:
+        return
+
+    def respond(self, value: Mapping[str, Any], status: int) -> None:
+        raw = canonical(value)
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Cache-Control", "no-store")
+        self.send_header("Content-Length", str(len(raw)))
+        self.end_headers()
+        self.wfile.write(raw)
+
+    def authorized(self) -> bool:
+        header = self.headers.get("Authorization", "")
+        return (
+            header.startswith("Bearer ")
+            and bool(self.ingress_token)
+            and hmac.compare_digest(header[7:], self.ingress_token)
+        )
+
+    def body(self) -> dict[str, Any]:
+        try:
+            length = int(self.headers.get("Content-Length", "-1"))
+        except ValueError as exc:
+            raise IngressError("CONTENT_LENGTH_INVALID") from exc
+        if length < 1 or length > MAX_BODY:
+            raise IngressError("REQUEST_SIZE_INVALID", 413)
+        try:
+            value = json.loads(self.rfile.read(length).decode("utf-8"))
+        except Exception as exc:
+            raise IngressError("REQUEST_JSON_INVALID") from exc
+        if not isinstance(value, Mapping):
+            raise IngressError("REQUEST_OBJECT_REQUIRED")
+        return dict(value)
+
+    def do_GET(self) -> None:
+        if self.path == "/health":
+            self.respond(
+                {
+                    "status": "ok",
+                    "service": "aios-gpt-online-task-ingress",
+                    "mode": "read-only",
+                },
+                200,
+            )
+            return
+        if self.path == "/openapi.json":
+            self.respond(openapi_document(), 200)
+            return
+        if not self.authorized():
+            self.respond({"error": "UNAUTHORIZED"}, 401)
+            return
+        match = re.fullmatch(r"/v1/read-only-tasks/([^/?#]+)", self.path)
+        if match is None:
+            self.respond({"error": "NOT_FOUND"}, 404)
+            return
+        try:
+            if self.service is None:
+                raise IngressError("RUNTIME_UNAVAILABLE", 503)
+            self.respond(self.service.status(match.group(1)), 200)
+        except IngressError as exc:
+            self.respond({"error": exc.code}, exc.status)
+
+    def do_POST(self) -> None:
+        if self.path != "/v1/read-only-tasks":
+            self.respond({"error": "NOT_FOUND"}, 404)
+            return
+        if not self.authorized():
+            self.respond({"error": "UNAUTHORIZED"}, 401)
+            return
+        try:
+            if self.service is None:
+                raise IngressError("RUNTIME_UNAVAILABLE", 503)
+            value = self.service.create(self.body())
+            self.respond(value, 200)
+        except IngressError as exc:
+            self.respond({"error": exc.code}, exc.status)
+
+
+def serve() -> None:
+    ingress_token_path = Path(
+        os.environ.get(
+            "AIOS_GPT_INGRESS_TOKEN_FILE",
+            str(
+                Path(os.environ.get("CREDENTIALS_DIRECTORY", "/run/credentials"))
+                / "gpt_ingress_token"
+            ),
+        )
+    )
+    control_token_path = Path(
+        os.environ.get(
+            "AIOS_AGENT_TASK_CONTROL_TOKEN_FILE",
+            str(
+                Path(os.environ.get("CREDENTIALS_DIRECTORY", "/run/credentials"))
+                / "agent_task_control_token"
+            ),
+        )
+    )
+    ingress_token = load_secret(
+        ingress_token_path,
+        code="GPT_INGRESS_CREDENTIAL",
+    )
+    control_token = load_secret(
+        control_token_path,
+        code="CONTROL_PLANE_CREDENTIAL",
+    )
+
+    state_root = Path(
+        os.environ.get(
+            "AIOS_GPT_INGRESS_STATE_ROOT",
+            "/var/lib/aios-gpt-online-task-ingress",
+        )
+    )
+    host = validated_bind_host(
+        os.environ.get("AIOS_GPT_INGRESS_HOST", "127.0.0.1")
+    )
+    port = int(os.environ.get("AIOS_GPT_INGRESS_PORT", "18945"))
+    if not 1024 <= port <= 65535:
+        raise RuntimeError("GPT_INGRESS_PORT_INVALID")
+
+    Handler.service = IngressService(
+        Upstream(
+            bridge_base=os.environ.get(
+                "AIOS_GOVERNED_WRITE_API_URL",
+                "http://127.0.0.1:18892",
+            ),
+            task_base=os.environ.get(
+                "AIOS_AGENT_TASK_API_URL",
+                "http://127.0.0.1:18940",
+            ),
+            control_token=control_token,
+        ),
+        Store(state_root / "state.sqlite3"),
+    )
+    Handler.ingress_token = ingress_token
+    ThreadingHTTPServer((host, port), Handler).serve_forever()
+
+
+if __name__ == "__main__":
+    serve()

active-runtime/app/aios_release_lineage_contract.py

@@ -0,0 +1,82 @@
+"""Immutable release-lineage contracts used by activation and recovery."""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+from typing import Any, Mapping
+
+
+@dataclass(frozen=True)
+class ReleaseLineage:
+    task_id: str
+    candidate_id: str
+    source_commit: str
+    version: str
+    required_capabilities: tuple[str, ...]
+
+    def as_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+FIXED_V2R2C2 = ReleaseLineage(
+    task_id="r42-release-candidate-v2r2c2-live-20260819",
+    candidate_id="r42-release-candidate-v2r2c2-20260819",
+    source_commit="71e736b05fab13efc06af7a4fe2dc022c1cf9a85",
+    version="V2R2C2",
+    required_capabilities=("release:candidate",),
+)
+
+
+class ReleaseLineageError(RuntimeError):
+    pass
+
+
+def validate_fixed_v2r2c2_task(task: Mapping[str, Any]) -> dict[str, Any]:
+    """Validate identity/source/version without mutating Task state.
+
+    This contract intentionally does not bind status/attempt/revision, because
+    those are lifecycle state and may change through governed retry/recovery.
+    """
+
+    if str(task.get("task_id") or "") != FIXED_V2R2C2.task_id:
+        raise ReleaseLineageError("FIXED_TASK_ID_DRIFT")
+
+    capabilities = task.get("required_capabilities")
+    if (
+        not isinstance(capabilities, list)
+        or tuple(capabilities) != FIXED_V2R2C2.required_capabilities
+    ):
+        raise ReleaseLineageError("FIXED_TASK_CAPABILITY_DRIFT")
+
+    constraints = task.get("constraints")
+    if not isinstance(constraints, Mapping):
+        raise ReleaseLineageError("FIXED_TASK_CONSTRAINTS_MISSING")
+    release = constraints.get("release_candidate")
+    if not isinstance(release, Mapping):
+        raise ReleaseLineageError("FIXED_RELEASE_BINDING_MISSING")
+
+    actual = {
+        "candidate_id": str(release.get("candidate_id") or ""),
+        "source_commit": str(release.get("source_commit") or ""),
+        "version": str(release.get("version") or ""),
+    }
+    expected = {
+        "candidate_id": FIXED_V2R2C2.candidate_id,
+        "source_commit": FIXED_V2R2C2.source_commit,
+        "version": FIXED_V2R2C2.version,
+    }
+    if actual != expected:
+        raise ReleaseLineageError(
+            "FIXED_RELEASE_LINEAGE_DRIFT:"
+            + repr({"expected": expected, "actual": actual})
+        )
+
+    return {
+        "schema": "aios.release-lineage.validation.v1",
+        "status": "VERIFIED",
+        "lineage": FIXED_V2R2C2.as_dict(),
+        "task_status": task.get("status"),
+        "attempt": task.get("attempt"),
+        "max_attempts": task.get("max_attempts"),
+        "revision": task.get("revision"),
+    }

active-runtime/app/aios_release_online_agent_worker.py

@@ -0,0 +1,714 @@
+#!/usr/bin/env python3
+"""Persistent release-preparation Online Agent worker.
+
+The worker is intentionally narrower than the generic engineering worker:
+it can only consume tasks whose required capability set is exactly
+{"release:candidate"} and whose payload asks for immutable release-candidate
+creation. It never receives Owner signing, deployment, tag, baseline or Git
+write credentials.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import socket
+import threading
+import time
+import urllib.error
+import urllib.request
+from typing import Any, Mapping
+from aios_autonomous_control_plane import (
+    classify_failure,
+    repair_for_failure,
+    runtime_admission,
+)
+
+AGENT_ID = os.environ.get(
+    "AIOS_RELEASE_WORKER_AGENT_ID",
+    "online-release-worker-cn-network-mkt-20260819",
+).strip()
+AGENT_VERSION = os.environ.get(
+    "AIOS_RELEASE_WORKER_AGENT_VERSION",
+    "2026-08-19.1-release-candidate",
+).strip()
+TASK_API = os.environ.get(
+    "AIOS_AGENT_TASK_API_URL",
+    "http://127.0.0.1:18940",
+).rstrip("/")
+RELEASE_API = os.environ.get("AIOS_RELEASE_API_URL", "").rstrip("/")
+POLL_SECONDS = int(os.environ.get("AIOS_RELEASE_WORKER_POLL_SECONDS", "3"))
+LEASE_SECONDS = int(os.environ.get("AIOS_RELEASE_WORKER_LEASE_SECONDS", "60"))
+HEARTBEAT_SECONDS = int(
+    os.environ.get("AIOS_RELEASE_WORKER_HEARTBEAT_SECONDS", "15")
+)
+STATE_ROOT = Path(
+    os.environ.get(
+        "AIOS_RELEASE_WORKER_STATE_ROOT",
+        "/var/lib/aios-online-agent-release",
+    )
+)
+EXPECTED_CAPABILITIES = frozenset({"release:candidate"})
+LIVENESS_PATH = Path(
+    os.environ.get(
+        "AIOS_WORKER_LIVENESS_PATH",
+        str(STATE_ROOT / "worker-liveness.json"),
+    )
+)
+CANDIDATE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,255}$")
+OID_RE = re.compile(r"^[0-9a-f]{40}$")
+VERSION_RE = re.compile(r"^V[0-9]+R[0-9]+C[0-9]+$")
+TASK_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
+SECRET_MARKERS = (
+    "authorization",
+    "token",
+    "secret",
+    "credential",
+    "password",
+    "private_key",
+    "signature",
+)
+
+
+class ReleaseWorkerError(RuntimeError):
+    def __init__(self, code: str, *, phase: str) -> None:
+        super().__init__(code)
+        self.code = code
+        self.phase = phase
+
+
+def now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def canonical(value: object) -> bytes:
+    return json.dumps(
+        value,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    ).encode("utf-8")
+
+
+def sha256(value: bytes) -> str:
+    return hashlib.sha256(value).hexdigest()
+
+
+def sanitized(value: Any) -> Any:
+    if isinstance(value, Mapping):
+        result: dict[str, Any] = {}
+        for key, inner in value.items():
+            lowered = str(key).lower()
+            if any(marker in lowered for marker in SECRET_MARKERS):
+                result[str(key)] = "[REDACTED]"
+            else:
+                result[str(key)] = sanitized(inner)
+        return result
+    if isinstance(value, list):
+        return [sanitized(item) for item in value]
+    return value
+
+
+def audit(event: str, **fields: Any) -> None:
+    STATE_ROOT.mkdir(parents=True, exist_ok=True, mode=0o750)
+    path = STATE_ROOT / "events.jsonl"
+    record = {
+        "schema": "aios.release-worker.event.v1",
+        "event": event,
+        "created_at": now(),
+        "agent_id": AGENT_ID,
+        **sanitized(fields),
+    }
+    with path.open("a", encoding="utf-8") as handle:
+        handle.write(json.dumps(record, sort_keys=True) + "\n")
+    os.chmod(path, 0o640)
+
+
+def sd_notify(message: str) -> None:
+    target = os.environ.get("NOTIFY_SOCKET", "")
+    if not target:
+        return
+    if target.startswith("@"):
+        address = "\0" + target[1:]
+    else:
+        address = target
+    sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
+    try:
+        sock.connect(address)
+        sock.sendall(message.encode("utf-8"))
+    except OSError:
+        # Liveness JSON remains the diagnostic source; systemd will enforce
+        # WatchdogSec if notify transport is persistently broken.
+        pass
+    finally:
+        sock.close()
+
+
+def write_liveness(
+    state: str,
+    *,
+    task_id: str | None = None,
+    stage: str | None = None,
+    error: str | None = None,
+    visible_count: int | None = None,
+) -> None:
+    STATE_ROOT.mkdir(parents=True, exist_ok=True, mode=0o750)
+    value = {
+        "schema": "aios.online-agent.worker-liveness.v1",
+        "agent_id": AGENT_ID,
+        "agent_type": "release-preparation-worker",
+        "agent_version": AGENT_VERSION,
+        "pid": os.getpid(),
+        "state": state,
+        "last_seen_epoch": time.time(),
+        "last_seen_at": now(),
+        "poll_seconds": POLL_SECONDS,
+        "heartbeat_seconds": HEARTBEAT_SECONDS,
+        "api_base": TASK_API,
+        "capabilities": sorted(EXPECTED_CAPABILITIES),
+        "task_id": task_id,
+        "stage": stage,
+        "error": error,
+        "visible_count": visible_count,
+    }
+    LIVENESS_PATH.parent.mkdir(parents=True, exist_ok=True, mode=0o750)
+    temporary = LIVENESS_PATH.with_name(
+        "." + LIVENESS_PATH.name + f".{os.getpid()}.tmp"
+    )
+    temporary.write_text(
+        json.dumps(value, ensure_ascii=False, sort_keys=True) + "\n",
+        encoding="utf-8",
+    )
+    os.chmod(temporary, 0o640)
+    os.replace(temporary, LIVENESS_PATH)
+
+
+def load_credential(name: str) -> str:
+    directory = Path(os.environ.get("CREDENTIALS_DIRECTORY", ""))
+    path = directory / name
+    if path.is_symlink() or not path.is_file():
+        raise ReleaseWorkerError(
+            "CREDENTIAL_UNAVAILABLE:" + name,
+            phase="startup",
+        )
+    value = path.read_text(encoding="utf-8").strip()
+    if len(value) < 24 or any(ch.isspace() for ch in value):
+        raise ReleaseWorkerError(
+            "CREDENTIAL_INVALID:" + name,
+            phase="startup",
+        )
+    return value
+
+
+def http_json(
+    base: str,
+    method: str,
+    route: str,
+    token: str,
+    body: Mapping[str, Any] | None = None,
+    *,
+    timeout: int = 30,
+) -> tuple[int, dict[str, Any]]:
+    data = None if body is None else canonical(body)
+    headers = {"Accept": "application/json"}
+    if token:
+        headers["Authorization"] = "Bearer " + token
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    request = urllib.request.Request(
+        base + route,
+        method=method,
+        data=data,
+        headers=headers,
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=timeout) as response:
+            raw, status = response.read(), response.status
+    except urllib.error.HTTPError as exc:
+        raw, status = exc.read(), exc.code
+    except (urllib.error.URLError, TimeoutError) as exc:
+        raise ReleaseWorkerError(
+            "HTTP_TRANSPORT_UNAVAILABLE",
+            phase="network",
+        ) from exc
+    try:
+        value = json.loads(raw.decode())
+    except Exception as exc:
+        raise ReleaseWorkerError(
+            "HTTP_RESPONSE_INVALID",
+            phase="network",
+        ) from exc
+    return status, value if isinstance(value, dict) else {}
+
+
+@dataclass(frozen=True)
+class ReleaseRequest:
+    candidate_id: str
+    source_commit: str
+    version: str
+    projection_id: str | None
+
+    @classmethod
+    def from_task(cls, task: Mapping[str, Any]) -> "ReleaseRequest":
+        if not TASK_RE.fullmatch(str(task.get("task_id") or "")):
+            raise ReleaseWorkerError("TASK_ID_INVALID", phase="validation")
+        caps = task.get("required_capabilities")
+        if (
+            not isinstance(caps, list)
+            or frozenset(caps) != EXPECTED_CAPABILITIES
+            or len(caps) != 1
+        ):
+            raise ReleaseWorkerError(
+                "TASK_CAPABILITY_SCOPE_INVALID",
+                phase="validation",
+            )
+        metadata = task.get("metadata")
+        if not isinstance(metadata, Mapping):
+            metadata = {}
+        constraints = task.get("constraints")
+        if not isinstance(constraints, Mapping):
+            constraints = {}
+        release = constraints.get("release_candidate")
+        if not isinstance(release, Mapping):
+            release = metadata.get("release_candidate")
+        if not isinstance(release, Mapping):
+            release = task.get("release_candidate")
+        if not isinstance(release, Mapping):
+            raise ReleaseWorkerError(
+                "RELEASE_CANDIDATE_PAYLOAD_MISSING",
+                phase="validation",
+            )
+        allowed = {
+            "candidate_id",
+            "source_commit",
+            "version",
+            "projection_id",
+        }
+        if set(release) - allowed:
+            raise ReleaseWorkerError(
+                "RELEASE_CANDIDATE_PAYLOAD_FIELDS_INVALID",
+                phase="validation",
+            )
+        candidate_id = str(release.get("candidate_id") or "")
+        source_commit = str(release.get("source_commit") or "")
+        version = str(release.get("version") or "")
+        projection_id = release.get("projection_id")
+        if CANDIDATE_RE.fullmatch(candidate_id) is None:
+            raise ReleaseWorkerError(
+                "RELEASE_CANDIDATE_ID_INVALID",
+                phase="validation",
+            )
+        if OID_RE.fullmatch(source_commit) is None:
+            raise ReleaseWorkerError(
+                "RELEASE_SOURCE_COMMIT_INVALID",
+                phase="validation",
+            )
+        if VERSION_RE.fullmatch(version) is None:
+            raise ReleaseWorkerError(
+                "RELEASE_VERSION_INVALID",
+                phase="validation",
+            )
+        if projection_id is not None:
+            projection_id = str(projection_id)
+            if not re.fullmatch(r"projection-[0-9a-f]{32}", projection_id):
+                raise ReleaseWorkerError(
+                    "RELEASE_PROJECTION_ID_INVALID",
+                    phase="validation",
+                )
+        return cls(
+            candidate_id=candidate_id,
+            source_commit=source_commit,
+            version=version,
+            projection_id=projection_id,
+        )
+
+    def body(self) -> dict[str, str]:
+        return {
+            "candidate_id": self.candidate_id,
+            "source_commit": self.source_commit,
+            "version": self.version,
+        }
+
+
+class Heartbeat:
+    def __init__(
+        self,
+        agent_token: str,
+        task_id: str,
+        lease_token: str,
+    ) -> None:
+        self.agent_token = agent_token
+        self.task_id = task_id
+        self.lease_token = lease_token
+        self.stage = "claimed"
+        self.failure: str | None = None
+        self.stop = threading.Event()
+        self.thread = threading.Thread(
+            target=self._run,
+            name="release-worker-heartbeat",
+            daemon=True,
+        )
+
+    def first(self) -> None:
+        status, value = http_json(
+            TASK_API,
+            "POST",
+            f"/v1/tasks/{self.task_id}/heartbeat",
+            self.agent_token,
+            {
+                "lease_token": self.lease_token,
+                "lease_seconds": LEASE_SECONDS,
+                "progress": {
+                    "stage": self.stage,
+                    "worker_version": AGENT_VERSION,
+                },
+            },
+        )
+        if status != 200:
+            raise ReleaseWorkerError(
+                "TASK_HEARTBEAT_FAILED",
+                phase="heartbeat",
+            )
+        audit(
+            "TASK_HEARTBEAT",
+            task_id=self.task_id,
+            stage=self.stage,
+            api_status=status,
+        )
+        write_liveness(
+            "BUSY",
+            task_id=self.task_id,
+            stage=self.stage,
+        )
+        sd_notify("WATCHDOG=1")
+
+    def _run(self) -> None:
+        while not self.stop.wait(HEARTBEAT_SECONDS):
+            try:
+                status, value = http_json(
+                    TASK_API,
+                    "POST",
+                    f"/v1/tasks/{self.task_id}/heartbeat",
+                    self.agent_token,
+                    {
+                        "lease_token": self.lease_token,
+                        "lease_seconds": LEASE_SECONDS,
+                        "progress": {
+                            "stage": self.stage,
+                            "worker_version": AGENT_VERSION,
+                        },
+                    },
+                )
+                if status != 200:
+                    self.failure = str(value.get("error") or status)
+                    return
+                audit(
+                    "TASK_HEARTBEAT",
+                    task_id=self.task_id,
+                    stage=self.stage,
+                    api_status=status,
+                )
+                write_liveness(
+                    "BUSY",
+                    task_id=self.task_id,
+                    stage=self.stage,
+                )
+                sd_notify("WATCHDOG=1")
+            except Exception as exc:
+                self.failure = type(exc).__name__
+                return
+
+    def ensure(self) -> None:
+        if self.failure:
+            raise ReleaseWorkerError(
+                "TASK_LEASE_LOST",
+                phase="heartbeat",
+            )
+
+    def close(self) -> None:
+        self.stop.set()
+        if self.thread.ident is not None:
+            self.thread.join(timeout=2)
+
+
+def release_admission(
+    task: Mapping[str, Any],
+    *,
+    release_token: str,
+):
+    required = task.get("required_capabilities")
+    if not isinstance(required, list):
+        required = []
+
+    def release_api_health() -> bool:
+        status, value = http_json(
+            RELEASE_API,
+            "GET",
+            "/health",
+            "",
+            timeout=10,
+        )
+        return status == 200 and str(value.get("status") or "").lower() in {"ok", "healthy"}
+
+    def authenticated_probe() -> bool:
+        status, _ = http_json(
+            RELEASE_API,
+            "POST",
+            "/createReleaseCandidate",
+            release_token,
+            {},
+            timeout=10,
+        )
+        # Authentication + credential projection succeeded; the empty release
+        # request is intentionally rejected by request validation.
+        return status == 400
+
+    probes = {
+        "credential_projection": authenticated_probe,
+        "trusted_builder": release_api_health,
+        "authenticated_probe": authenticated_probe,
+        "release_api_health": release_api_health,
+        "worker_liveness": lambda: True,
+    }
+    return runtime_admission(
+        worker_capabilities=EXPECTED_CAPABILITIES,
+        required_capabilities=required,
+        requested_operations=("prepare_release",),
+        probes=probes,
+    )
+
+
+def result_payload(
+    task_id: str,
+    release: ReleaseRequest,
+    builder: Mapping[str, Any],
+) -> dict[str, Any]:
+    builder_candidate = builder.get("candidate_id")
+    if builder_candidate not in {None, release.candidate_id}:
+        raise ReleaseWorkerError(
+            "BUILDER_CANDIDATE_BINDING_INVALID",
+            phase="release",
+        )
+    immutable = {
+        "candidate_id": release.candidate_id,
+        "source_commit": release.source_commit,
+        "version": release.version,
+    }
+    return {
+        "status": "SUCCEEDED",
+        "classification": "RELEASE_CANDIDATE_CREATED",
+        "execution_mode": "release_candidate",
+        "adapter": "release_candidate",
+        "task_id": task_id,
+        "agent_id": AGENT_ID,
+        "release_candidate": immutable,
+        "release_candidate_request_sha256": sha256(canonical(immutable)),
+        "builder_response": sanitized(dict(builder)),
+        "projection_id": release.projection_id,
+        "writer_lease": "NONE",
+        "changed_files": [],
+        "direct_main_write": False,
+        "production_write": False,
+        "tag_write": False,
+        "baseline_write": False,
+        "runtime_switch": False,
+        "owner_approval_submitted": False,
+        "credential_material_exposed": False,
+    }
+
+
+def execute_task(
+    task: Mapping[str, Any],
+    *,
+    agent_token: str,
+    release_token: str,
+) -> bool:
+    task_id = str(task.get("task_id") or "")
+    release = ReleaseRequest.from_task(task)
+    admission = release_admission(
+        task,
+        release_token=release_token,
+    )
+    if not admission.admitted:
+        audit(
+            "TASK_ADMISSION_BLOCKED",
+            task_id=task_id,
+            admission=admission.as_dict(),
+        )
+        write_liveness(
+            "BLOCKED",
+            task_id=task_id,
+            stage="execution_ready",
+            error=admission.failure_class.value if admission.failure_class else "ADMISSION_FAILED",
+        )
+        return False
+    audit(
+        "TASK_ADMISSION_PASSED",
+        task_id=task_id,
+        admission=admission.as_dict(),
+    )
+    status, claimed = http_json(
+        TASK_API,
+        "POST",
+        f"/v1/tasks/{task_id}/claim",
+        agent_token,
+        {"lease_seconds": LEASE_SECONDS},
+    )
+    if status in {409, 423}:
+        return False
+    if status != 200:
+        raise ReleaseWorkerError(
+            "TASK_CLAIM_FAILED",
+            phase="claim",
+        )
+    lease_token = str(claimed.get("lease_token") or "")
+    if len(lease_token) < 16:
+        raise ReleaseWorkerError(
+            "TASK_LEASE_TOKEN_INVALID",
+            phase="claim",
+        )
+    attempt = int(claimed.get("attempt") or 0)
+    audit(
+        "TASK_CLAIMED",
+        task_id=task_id,
+        attempt=attempt,
+    )
+
+    heartbeat = Heartbeat(agent_token, task_id, lease_token)
+    heartbeat.first()
+    heartbeat.thread.start()
+    try:
+        heartbeat.stage = "release_candidate"
+        heartbeat.ensure()
+        status, builder = http_json(
+            RELEASE_API,
+            "POST",
+            "/createReleaseCandidate",
+            release_token,
+            release.body(),
+            timeout=120,
+        )
+        if status not in {200, 201, 202}:
+            raise ReleaseWorkerError(
+                "RELEASE_CANDIDATE_CREATE_FAILED:"
+                + str(builder.get("error") or status),
+                phase="release",
+            )
+        heartbeat.ensure()
+        result = result_payload(task_id, release, builder)
+        idempotency_key = (
+            f"{task_id}-attempt-{attempt}-release-result-v1"
+        )
+        status, submitted = http_json(
+            TASK_API,
+            "POST",
+            f"/v1/tasks/{task_id}/result",
+            agent_token,
+            {
+                "lease_token": lease_token,
+                "idempotency_key": idempotency_key,
+                "result": result,
+            },
+        )
+        if status not in {200, 201, 202}:
+            raise ReleaseWorkerError(
+                "TASK_RESULT_SUBMIT_FAILED",
+                phase="result",
+            )
+        audit(
+            "RESULT_SUBMITTED",
+            task_id=task_id,
+            attempt=attempt,
+            release_candidate_id=release.candidate_id,
+        )
+        return True
+    finally:
+        heartbeat.close()
+
+
+def main() -> int:
+    if not RELEASE_API:
+        raise SystemExit("FATAL:AIOS_RELEASE_API_URL_REQUIRED")
+    agent_token = load_credential("agent_token")
+    release_token = load_credential("release_candidate_token")
+    STATE_ROOT.mkdir(parents=True, exist_ok=True, mode=0o750)
+    audit(
+        "WORKER_STARTED",
+        capabilities=sorted(EXPECTED_CAPABILITIES),
+        release_api=RELEASE_API,
+    )
+    write_liveness("STARTING", stage="startup")
+    sd_notify("READY=1\nSTATUS=release worker ready")
+    while True:
+        try:
+            status, response = http_json(
+                TASK_API,
+                "POST",
+                "/v1/tasks/poll",
+                agent_token,
+                {"limit": 100},
+            )
+            if status != 200:
+                raise ReleaseWorkerError(
+                    "TASK_POLL_FAILED",
+                    phase="poll",
+                )
+            tasks = response.get("tasks")
+            if not isinstance(tasks, list):
+                tasks = []
+            write_liveness(
+                "IDLE",
+                stage="poll",
+                visible_count=len(tasks),
+            )
+            sd_notify("WATCHDOG=1\nSTATUS=release worker poll ok")
+            handled = False
+            for raw in tasks:
+                if not isinstance(raw, Mapping):
+                    continue
+                caps = raw.get("required_capabilities")
+                if (
+                    not isinstance(caps, list)
+                    or frozenset(caps) != EXPECTED_CAPABILITIES
+                ):
+                    continue
+                try:
+                    handled = execute_task(
+                        raw,
+                        agent_token=agent_token,
+                        release_token=release_token,
+                    ) or handled
+                except ReleaseWorkerError as exc:
+                    failure = classify_failure(exc.code, phase=exc.phase)
+                    repair = repair_for_failure(failure)
+                    audit(
+                        "TASK_EXECUTION_FAILED",
+                        task_id=raw.get("task_id"),
+                        phase=exc.phase,
+                        error=exc.code,
+                        failure_class=failure.value,
+                        repair=repair.as_dict(),
+                    )
+            time.sleep(0 if handled else POLL_SECONDS)
+        except KeyboardInterrupt:
+            return 0
+        except Exception as exc:
+            audit(
+                "WORKER_LOOP_ERROR",
+                error=type(exc).__name__,
+            )
+            write_liveness(
+                "DEGRADED",
+                stage="loop",
+                error=type(exc).__name__,
+            )
+            time.sleep(POLL_SECONDS)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

config/model-governance/model-governance.staging.json

@@ -45,7 +45,7 @@
       "daily_token_budget": 250000,
       "denied_providers": [],
       "fallback_limit": 1,
-      "max_input_tokens_per_request": 16000,
+      "max_input_tokens_per_request": 16384,
       "max_output_tokens_per_request": 512,
       "project_id": null,
       "required_capabilities": [
@@ -63,7 +63,7 @@
       "daily_token_budget": 250000,
       "denied_providers": [],
       "fallback_limit": 1,
-      "max_input_tokens_per_request": 16000,
+      "max_input_tokens_per_request": 16384,
       "max_output_tokens_per_request": 512,
       "project_id": "staging",
       "required_capabilities": [

deploy/systemd/aios-agent-task-initial-execution-projection.service

@@ -0,0 +1,37 @@
+[Unit]
+Description=AIOS owner-controlled Task initial execution projection API
+After=aios-agent-task-api.service aios-chatgpt-bridge-v2-staging.service
+Wants=aios-agent-task-api.service aios-chatgpt-bridge-v2-staging.service
+
+[Service]
+Type=simple
+User=aios-agent-task
+Group=aios-agent-task
+LoadCredential=agent_task_control_token:/etc/aios-gitops/agent-task-control.token
+Environment=AIOS_AGENT_TASK_DB=/var/lib/aios-agent-task/state.sqlite3
+Environment=AIOS_AGENT_TASK_CONTROL_TOKEN_FILE=%d/agent_task_control_token
+Environment=AIOS_GOVERNED_WRITE_API_URL=http://127.0.0.1:18892
+Environment=AIOS_AGENT_TASK_INITIAL_EXECUTION_PROJECTION_AUTHORIZATION_ID=owner-initial-execution-projection-recovery-20260820
+Environment=AIOS_TASK_INITIAL_EXECUTION_PROJECTION_HOST=127.0.0.1
+Environment=AIOS_TASK_INITIAL_EXECUTION_PROJECTION_PORT=18943
+ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_agent_task_initial_execution_projection.py
+Restart=on-failure
+RestartSec=2
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectControlGroups=true
+LockPersonality=true
+RestrictSUIDSGID=true
+RestrictRealtime=true
+RestrictNamespaces=true
+MemoryDenyWriteExecute=true
+RestrictAddressFamilies=AF_UNIX AF_INET
+ReadWritePaths=/var/lib/aios-agent-task
+UMask=0077
+
+[Install]
+WantedBy=multi-user.target

deploy/systemd/aios-gpt-https-edge.service

@@ -0,0 +1,41 @@
+[Unit]
+Description=AIOS disabled-by-default HTTPS edge for GPT read-only Online Task ingress
+After=network-online.target aios-gpt-online-task-ingress.service
+Wants=network-online.target aios-gpt-online-task-ingress.service
+ConditionPathExists=/etc/aios-gitops/gpt-https-edge.approved
+
+[Service]
+Type=simple
+DynamicUser=yes
+LoadCredential=tls_cert:/etc/aios-gitops/gpt-https-edge-tls.crt
+LoadCredential=tls_key:/etc/aios-gitops/gpt-https-edge-tls.key
+Environment=AIOS_GPT_EDGE_TLS_CERT_FILE=%d/tls_cert
+Environment=AIOS_GPT_EDGE_TLS_KEY_FILE=%d/tls_key
+Environment=AIOS_GPT_EDGE_HOST=127.0.0.1
+Environment=AIOS_GPT_EDGE_PORT=18443
+Environment=AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=0
+Environment=AIOS_GPT_EDGE_PUBLIC_DOMAIN=
+ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_gpt_https_edge.py
+Restart=on-failure
+RestartSec=2
+NoNewPrivileges=yes
+PrivateTmp=yes
+PrivateDevices=yes
+ProtectSystem=strict
+ProtectHome=yes
+ProtectKernelTunables=yes
+ProtectKernelModules=yes
+ProtectControlGroups=yes
+ProtectKernelLogs=yes
+ProtectClock=yes
+LockPersonality=yes
+RestrictSUIDSGID=yes
+RestrictRealtime=yes
+RestrictNamespaces=yes
+MemoryDenyWriteExecute=yes
+RestrictAddressFamilies=AF_INET AF_INET6
+InaccessiblePaths=/data/aios-gitops /etc/aios-gitops /home /var/lib/aios-agent-task
+UMask=0077
+
+[Install]
+WantedBy=multi-user.target

deploy/systemd/aios-gpt-online-task-ingress.service

@@ -0,0 +1,45 @@
+[Unit]
+Description=AIOS external GPT read-only Online Task ingress
+After=network.target aios-chatgpt-bridge-v2-staging.service aios-agent-task-api.service
+Wants=aios-chatgpt-bridge-v2-staging.service aios-agent-task-api.service
+
+[Service]
+Type=simple
+DynamicUser=yes
+StateDirectory=aios-gpt-online-task-ingress
+StateDirectoryMode=0750
+LoadCredential=gpt_ingress_token:/etc/aios-gitops/gpt-online-task-ingress.token
+LoadCredential=agent_task_control_token:/etc/aios-gitops/agent-task-control.token
+Environment=AIOS_GPT_INGRESS_TOKEN_FILE=%d/gpt_ingress_token
+Environment=AIOS_AGENT_TASK_CONTROL_TOKEN_FILE=%d/agent_task_control_token
+Environment=AIOS_GPT_INGRESS_STATE_ROOT=%S/aios-gpt-online-task-ingress
+Environment=AIOS_GPT_INGRESS_HOST=127.0.0.1
+Environment=AIOS_GPT_INGRESS_PORT=18945
+Environment=AIOS_GOVERNED_WRITE_API_URL=http://127.0.0.1:18892
+Environment=AIOS_AGENT_TASK_API_URL=http://127.0.0.1:18940
+ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_gpt_online_task_ingress.py
+Restart=on-failure
+RestartSec=2
+NoNewPrivileges=yes
+PrivateTmp=yes
+PrivateDevices=yes
+ProtectSystem=strict
+ProtectHome=yes
+ProtectKernelTunables=yes
+ProtectKernelModules=yes
+ProtectControlGroups=yes
+ProtectKernelLogs=yes
+ProtectClock=yes
+LockPersonality=yes
+RestrictSUIDSGID=yes
+RestrictRealtime=yes
+RestrictNamespaces=yes
+MemoryDenyWriteExecute=yes
+RestrictAddressFamilies=AF_UNIX AF_INET
+IPAddressDeny=any
+IPAddressAllow=127.0.0.0/8
+InaccessiblePaths=/data/aios-gitops /etc/aios-gitops /home
+UMask=0077
+
+[Install]
+WantedBy=multi-user.target

deploy/systemd/aios-owner-durable-authorization-broker.service

@@ -0,0 +1,47 @@
+[Unit]
+Description=AIOS durable Owner authorization broker
+After=network.target aios-chatgpt-bridge-v2-staging.service
+Wants=aios-chatgpt-bridge-v2-staging.service
+
+[Service]
+Type=simple
+User=root
+Group=aios-executor-exchange
+RuntimeDirectory=aios-owner-authorization-broker
+RuntimeDirectoryMode=0750
+StateDirectory=aios-owner-authorization-broker
+StateDirectoryMode=0750
+LoadCredential=agent_task_control_token:/etc/aios-gitops/agent-task-control.token
+Environment=AIOS_AGENT_TASK_CONTROL_TOKEN_FILE=%d/agent_task_control_token
+Environment=AIOS_OWNER_DURABLE_AUTHORIZATION_POLICY=/etc/aios-gitops/owner-durable-authorization.json
+Environment=AIOS_OWNER_DURABLE_AUTHORIZATION_REVOCATIONS=/etc/aios-gitops/owner-durable-authorization-revocations.json
+Environment=AIOS_OWNER_AUTH_BROKER_ACCOUNT=chatgpt-governed-implementer-singlehost
+Environment=AIOS_OWNER_AUTH_BROKER_SOCKET=%t/aios-owner-authorization-broker/broker.sock
+ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_durable_owner_authorization.py
+Restart=on-failure
+RestartSec=2
+NoNewPrivileges=yes
+PrivateTmp=yes
+PrivateDevices=yes
+ProtectSystem=strict
+ProtectHome=yes
+ProtectKernelTunables=yes
+ProtectKernelModules=yes
+ProtectControlGroups=yes
+ProtectKernelLogs=yes
+ProtectClock=yes
+LockPersonality=yes
+RestrictSUIDSGID=yes
+RestrictRealtime=yes
+RestrictNamespaces=yes
+MemoryDenyWriteExecute=yes
+RestrictAddressFamilies=AF_UNIX AF_INET
+IPAddressDeny=any
+IPAddressAllow=127.0.0.0/8
+ReadOnlyPaths=/etc/aios-gitops/owner-durable-authorization.json
+ReadOnlyPaths=/etc/aios-gitops/owner-durable-authorization-revocations.json
+InaccessiblePaths=/data/aios-gitops /home
+UMask=0007
+
+[Install]
+WantedBy=multi-user.target

deploy/systemd/aios-release-online-agent-worker.service

@@ -0,0 +1,50 @@
+[Unit]
+Description=AIOS release-candidate Online Agent worker
+After=network.target aios-agent-task-api.service
+Wants=aios-agent-task-api.service
+
+[Service]
+Type=notify
+NotifyAccess=main
+WatchdogSec=90
+User=root
+Group=root
+StateDirectory=aios-online-agent-release
+StateDirectoryMode=0750
+LoadCredentialEncrypted=agent_token:/etc/credstore.encrypted/aios-release-online-agent-worker/agent_token
+LoadCredentialEncrypted=release_candidate_token:/etc/credstore.encrypted/aios-release-online-agent-worker/release_candidate_token
+Environment=AIOS_RELEASE_WORKER_AGENT_ID=online-release-worker-cn-network-mkt-20260819
+Environment=AIOS_RELEASE_WORKER_AGENT_VERSION=2026-08-19.1-release-candidate
+Environment=AIOS_AGENT_TASK_API_URL=http://127.0.0.1:18940
+Environment=AIOS_RELEASE_WORKER_STATE_ROOT=/var/lib/aios-online-agent-release
+Environment=AIOS_WORKER_STATE_ROOT=/var/lib/aios-online-agent-release
+Environment=AIOS_RELEASE_WORKER_POLL_SECONDS=3
+Environment=AIOS_RELEASE_WORKER_LEASE_SECONDS=60
+Environment=AIOS_RELEASE_WORKER_HEARTBEAT_SECONDS=15
+Environment=AIOS_RELEASE_API_URL=http://127.0.0.1:18890
+ExecStart=/usr/bin/python3 -B /opt/aios/current/active-runtime/app/aios_release_online_agent_worker.py
+Restart=always
+RestartSec=2
+NoNewPrivileges=yes
+PrivateTmp=yes
+PrivateDevices=yes
+ProtectSystem=strict
+ProtectHome=yes
+ProtectKernelTunables=yes
+ProtectKernelModules=yes
+ProtectControlGroups=yes
+ProtectKernelLogs=yes
+ProtectClock=yes
+LockPersonality=yes
+RestrictSUIDSGID=yes
+RestrictRealtime=yes
+RestrictNamespaces=yes
+MemoryDenyWriteExecute=yes
+RestrictAddressFamilies=AF_UNIX AF_INET
+IPAddressDeny=any
+IPAddressAllow=127.0.0.0/8
+InaccessiblePaths=/data/aios-gitops /home
+UMask=0077
+
+[Install]
+WantedBy=multi-user.target

docs/superpowers/specs/2026-08-18-model-governance-policy-profile-floor-compatibility.md

@@ -0,0 +1,109 @@
+# Model Governance Policy / Profile Floor Compatibility
+
+## Problem
+
+The root-owned OpenClaw runtime profile reserves a minimum of 16,384 input tokens so Model
+Governance can cover OpenClaw-injected workspace, skills and system context before provider
+execution.
+
+The staging scope policies still limited input to 16,000 tokens per request. Route selection
+therefore failed before reservation with:
+
+```text
+MODEL_INPUT_TOKEN_LIMIT_DENY
+```
+
+This was a configuration invariant violation:
+
+```text
+scope max_input_tokens_per_request
+<
+root-owned profile input_reservation_floor_tokens
+```
+
+The registered OpenClaw model itself supports 32,768 input tokens, so the model capability was not
+the blocker.
+
+## Invariant
+
+For every Model Governance scope that allows a model/runtime whose root-owned profile declares an
+input reservation floor:
+
+```text
+scope.max_input_tokens_per_request
+>=
+profile.model_governance.input_reservation_floor_tokens
+```
+
+The profile floor remains root-owned. Task payloads cannot alter it.
+
+## Staging policy
+
+The two staging policies that allow `openai/deepseek-v4-flash` are updated from:
+
+```text
+16000
+```
+
+to:
+
+```text
+16384
+```
+
+This is the minimum compatible value. It intentionally remains below the model's 32,768-token
+technical input ceiling.
+
+The daily token budget is unchanged.
+
+## Governance path
+
+The live SQLite policy must be changed only through the repository bootstrap and
+`aios_model_governance_admin.py`.
+
+The admin reconciler:
+
+- reads the governed repository config;
+- uses current policy revision as CAS;
+- calls `ModelGovernanceStore.set_scope_policy`;
+- appends normal Model Governance audit evidence;
+- verifies the audit chain.
+
+Direct SQL updates remain forbidden.
+
+## Exhausted OpenClaw recovery
+
+The fixed OpenClaw live Task has exhausted its normal three attempts.
+
+After the governed policy candidate is officially verified and the live policy is reconciled:
+
+```text
+attempt 3 VERIFYING/FAILED result
+→ official NEEDS_REWORK
+→ Task becomes FAILED 3/3
+→ official bounded retry extension 3 → 4
+→ same Task / same Job returns QUEUED
+→ persistent OpenClaw worker auto-claims attempt 4
+```
+
+AR1 attempt-scoped external request identity ensures attempt 4 receives a new external request and a
+new Model Governance reservation.
+
+No historical reservation is deleted or rewritten.
+
+## Acceptance
+
+Attempt 4 must prove:
+
+```text
+scope max_input_tokens_per_request=16384
+profile input_reservation_floor_tokens=16384
+reserved_estimated_input_tokens>=16384
+actual input <= reserved input
+settlement=SETTLED
+usage_source=runtime
+audit_chain_valid=true
+```
+
+If attempt 4 exposes a new failure, stop. No additional retry extension is authorized by this
+repair.

docs/superpowers/specs/2026-08-19-all-modified-files-packager-secret-scan.md

@@ -0,0 +1,96 @@
+# Canonical all-modified-files packager and secret scanning
+
+## Problem
+
+The historical all-modified-files evidence packager lived only as a copied handoff script.
+
+Its private-key scanner matched the fragment:
+
+```text
+BEGIN PRIVATE KEY
+```
+
+instead of a complete PEM header. A regression test containing that literal therefore caused a
+false secret finding even though the candidate contained no credential.
+
+A task had to use a temporary evidence-only compatibility packager to complete its bundle. That is
+acceptable as an emergency workaround but not as a durable unattended path.
+
+## Repository source of truth
+
+The canonical implementation is now:
+
+```text
+active-runtime/app/aios_all_modified_files_packager.py
+```
+
+It preserves the historical output/bundle schemas:
+
+```text
+aios.all-modified-files.bundle.v1
+aios.all-modified-files.bundle-summary.v1
+aios.modified-documents.index.v1
+```
+
+and the host-visible status lines:
+
+```text
+ALL_MODIFIED_FILES_BUNDLE=PASS
+ALL_MODIFIED_FILES_BUNDLE_PATH=...
+ALL_MODIFIED_FILES_BUNDLE_SHA256=...
+```
+
+Future governed tasks should use this repository version instead of copying an unversioned scanner.
+
+## Secret scanning invariant
+
+Private-key detection requires a complete PEM header:
+
+```text
+-----BEGIN [RSA|EC|DSA|OPENSSH ]PRIVATE KEY-----
+```
+
+The following source/test/document text is not secret material and must remain packageable:
+
+```text
+BEGIN PRIVATE KEY
+PRIVATE KEY
+PEM private-key header
+```
+
+Bearer and key/value credential detection remains fail-closed:
+
+```text
+Authorization: Bearer <20+ token chars>
+password=<12+ non-space chars>
+passwd=<12+ non-space chars>
+secret=<12+ non-space chars>
+api_key=<12+ non-space chars>
+api-key=<12+ non-space chars>
+```
+
+Tests construct the real PEM header dynamically so the scanner's own test source does not contain a
+literal credential-looking header.
+
+## Bundle integrity
+
+The packager continues to enforce:
+
+- exact base→candidate ancestry;
+- before/after file capture;
+- rename/copy/delete handling;
+- path traversal denial;
+- execution evidence symlink denial;
+- secret scan over repository before/after and execution evidence;
+- document index;
+- SHA-256 closure;
+- ZIP without symlinks.
+
+No secret finding is downgraded to a warning.
+
+## Migration
+
+The old handoff script may remain for historical reproducibility, but new handoffs should call the
+repository packager from the verified candidate lineage.
+
+A temporary compatibility scanner is no longer needed once this candidate is in the lineage.

docs/superpowers/specs/2026-08-19-autonomous-control-plane.md

@@ -0,0 +1,210 @@
+# AIOS Autonomous Control Plane — Admission + Invariant + Repair
+
+## Why this exists
+
+Recent Agent registration, Task claim, release-worker, credential and retry failures shared one
+failure mode:
+
+> The control plane accepted an intent before the runtime had proved it could execute the intent.
+
+This design does not add another fleet of daemons. It adds one shared library contract that Task API,
+workers and governed host recovery consume.
+
+## Three layers
+
+```text
+Intent
+  ↓
+Runtime Admission
+  ↓
+System Invariants
+  ↓
+Repair Policy
+  ↓
+Execution
+```
+
+### Runtime Admission
+
+Admission occurs before Task claim.
+
+A capability such as `release:candidate` is not considered executable merely because its name
+matches. The capability contract also binds:
+
+- supported contract version;
+- allowed permission;
+- forbidden operations;
+- runtime dependencies;
+- health requirements.
+
+For `release:candidate`:
+
+```text
+permission:
+  prepare_release
+
+forbidden:
+  production_deploy
+  submit_owner_approval
+  tag_publish
+  baseline_registry_finalize
+  runtime_switch
+
+dependencies:
+  credential_projection
+  trusted_builder
+
+health:
+  authenticated_probe
+  release_api_health
+  worker_liveness
+```
+
+Only a fully passing decision reaches:
+
+```text
+EXECUTION_READY
+```
+
+A failed decision remains:
+
+```text
+EXECUTION_READY_FAILED
+```
+
+and the worker does not call `/claim`.
+
+Therefore an infrastructure defect discovered by admission does not consume a business attempt.
+
+### System Invariants
+
+The same admission checks are exposed as a machine-readable invariant snapshot.
+
+Examples:
+
+```text
+capability_contract_registry
+task_capabilities_registered
+worker_capabilities_registered
+capability_match
+forbidden_operations_absent
+credential_projection
+trusted_builder
+authenticated_probe
+release_api_health
+worker_liveness
+```
+
+A runtime can expose:
+
+```text
+HEALTHY
+DEGRADED
+```
+
+without relying on a human reconstructing state from logs.
+
+### Repair Policy
+
+Failure classification maps to bounded recovery:
+
+```text
+DEPENDENCY_UNAVAILABLE       → level 0 / auto local
+RESOURCE_UNAVAILABLE         → level 0 / auto local
+WORKER_CRASH                 → level 0 / auto local
+LEASE_LOST                   → level 0 / auto local
+
+CAPABILITY_CONTRACT_ERROR    → level 1 / same-lineage repair task
+CREDENTIAL_PROJECTION_ERROR  → level 1 / same-lineage repair task
+UNKNOWN                      → level 1 / diagnostic repair task
+
+POLICY_DENIED                → level 2 / Owner authorization
+BUSINESS_PERMANENT           → level 3 / terminal/manual decision
+```
+
+Only business-permanent failure consumes the business retry budget in this first contract.
+
+The current Task API does not yet have a first-class `EXECUTION_READY_FAILED` durable state. Until it
+does, the Release Worker enforces the most important invariant by refusing to claim a Task whose
+admission fails.
+
+## Capability Contract Registry
+
+`aios_capability_contract.py` remains backward-compatible with existing aliases but now validates
+every alias target against a versioned semantic contract registry.
+
+Unknown capabilities and wildcard capabilities remain fail-closed.
+
+Agent registration and Task creation should next be changed to call `registered_capability()` so a
+capability cannot be accepted at registration and rejected only later by the matcher.
+
+## Credential rule
+
+Admission probes credential usability, not merely credential existence.
+
+For release candidate creation the authenticated probe intentionally sends an invalid empty
+`/createReleaseCandidate` request.
+
+Expected:
+
+```text
+HTTP 400
+```
+
+This proves:
+
+```text
+credential readable
+→ authentication succeeded
+→ scope accepted
+→ request validation rejected empty business input
+```
+
+A permission failure/401/403/5xx is `EXECUTION_READY_FAILED`.
+
+## Lineage
+
+Repair actions are descriptive, not permission grants. Any repair Task must retain the existing
+root/parent/attempt/authority chain and use governed same-lineage recovery.
+
+No repair action may silently create a replacement business Task.
+
+## Authority separation
+
+Capability answers:
+
+```text
+can this runtime perform the operation?
+```
+
+Authority answers:
+
+```text
+is this operation currently authorized?
+```
+
+`release:candidate` never implies Owner approval, deployment, tag publication, baseline finalization
+or runtime switch.
+
+## Deployment sequence
+
+Phase 1 (this candidate):
+
+- Capability Contract Registry;
+- shared Autonomous Control Plane library;
+- Release Worker pre-claim admission;
+- failure classification + repair evidence.
+
+Phase 2:
+
+- Task API uses the same library before returning Tasks from poll and again during claim CAS;
+- Task schema gains durable `ADMITTED/EXECUTION_READY/EXECUTION_READY_FAILED` evidence without
+  counting pre-side-effect infrastructure failures against business attempts.
+
+Phase 3:
+
+- Agent registration validates against the same Capability Contract Registry;
+- external connector writes only durable Intent/Command Inbox records and leaves execution to the
+  local control plane.
+
+This sequencing avoids a second controller stack while converging every decision onto one contract.

docs/superpowers/specs/2026-08-19-durable-owner-authorization.md

@@ -0,0 +1,228 @@
+# Durable Owner Authorization and capability projection
+
+## Problem
+
+AIOS already supports unattended worker execution after a Task is executable, but Owner intent was
+not durable at the capability layer. The same Owner repeatedly had to authorize or repair:
+
+- `governed-task:project`;
+- `release:candidate`.
+
+This caused valid one-time Owner intent to be lost between conversations and forced repeated human
+interaction.
+
+## Design
+
+Durable authorization is represented by a root-owned mode `0600` policy outside Git. The policy is
+persistent until explicitly revoked.
+
+The first persistable capabilities are deliberately narrow:
+
+```text
+governed-task:project
+release:candidate
+```
+
+They cover preparation work only.
+
+The following capabilities can never be inherited from this durable grant:
+
+```text
+submitOwnerApproval
+startDeployment
+tag_publish
+baseline_registry_finalize
+runtime_switch
+main_write
+production_deploy
+force_push
+credential_admin
+```
+
+Those remain separate high-risk release/Owner-signature gates.
+
+## Policy
+
+Example schema:
+
+```json
+{
+  "schema": "aios.owner-durable-authorization.v1",
+  "grant_id": "owner-permanent-preparation-20260819",
+  "owner_identity": "owner@example",
+  "account": "chatgpt-governed-implementer-singlehost",
+  "capabilities": [
+    "governed-task:project",
+    "release:candidate"
+  ],
+  "issued_at": "2026-08-19T05:00:00Z",
+  "not_before": "2026-08-19T05:00:00Z",
+  "persistent_until_revoked": true,
+  "constraints": {
+    "repository": "aios/aios",
+    "allowed_target_prefix": "agent/",
+    "release_version_prefix": "V2R2C",
+    "max_task_risk": "engineering"
+  }
+}
+```
+
+The repository never stores the live Owner policy or release credential.
+
+## Revocation
+
+A root-owned mode `0600` revocation file lists `revoked_grant_ids`.
+
+Revocation is checked on every operation, not only at service startup. A running broker therefore
+fails closed immediately after revocation without needing a new conversation.
+
+## Broker boundary
+
+The broker exposes only an AF_UNIX socket:
+
+```text
+/run/aios-owner-authorization-broker/broker.sock
+```
+
+The socket is mode `0660` and owned by the service's `aios-executor-exchange` group. This avoids introducing another long-lived bearer secret solely for local broker access.
+
+It exposes only:
+
+```text
+GET  /v1/capabilities
+POST /v1/project-governed-task
+POST /v1/authorize-release-candidate
+```
+
+`project-governed-task` calls only the canonical Bridge.
+
+`authorize-release-candidate` persists an Owner authorization decision only. It deliberately does
+**not** invent a local trusted-builder endpoint. The external release connector remains responsible
+for projecting the account's real `release:candidate` scope. Until that external projection exists,
+the decision state is `AUTHORIZED_PENDING_EXTERNAL_CAPABILITY_PROJECTION`.
+
+The broker holds only the existing host control credential needed for canonical governed-task
+projection. It never holds an Owner signing key or a release/deployment credential. The caller also supplies the expected account identifier over the Unix transport; the OS socket permission is the local access boundary.
+
+
+## Canonical Online Task projection
+
+A durable `governed-task:project` grant may create a complete canonical Online Task rather than a
+bare Task. It requires:
+
+```text
+require_online_task_projection=true
+generation=1
+branch_mode=CREATE_NEW
+max_attempts<=3
+risk<=engineering
+initial action=source_capture
+```
+
+Projected task capabilities are limited to:
+
+```text
+git.read
+git.test
+governed.write.branch
+governed.write.file
+```
+
+A write-capable task still begins as `source_capture`. Existing placeholder-deferral prevents the
+generic worker from claiming it until a deterministic plan is registered. Repository mutation
+therefore still requires the normal deterministic plan, WriterLease and governed commit path.
+
+This fixes the previous split-brain behavior where a bare Task could exist without the canonical
+governed execution projection.
+
+
+## External capability projection outbox
+
+A `release:candidate` authorization now creates a durable machine-readable outbox record:
+
+```text
+schema=aios.external-capability-projection.request.v1
+state=PENDING_EXTERNAL_CONNECTOR
+capability=release:candidate
+```
+
+The projection identifier is deterministic from:
+
+```text
+grant_id
+account
+capability
+request_sha256
+```
+
+The outbox stores no credentials or Owner signing material. It retains only the already-authorized
+release-candidate request, policy/request hashes, account, grant id and timestamps.
+
+The broker exposes read-only Unix-socket endpoints:
+
+```text
+GET /v1/projections
+GET /v1/projections/{projection_id}
+```
+
+These endpoints re-check the current Owner policy on every read. If the durable grant has been
+revoked, enumeration fails closed with `OWNER_GRANT_REVOKED`. The historical outbox record is
+retained for audit and is not silently deleted.
+
+The repository intentionally does **not** provide a connector acknowledgement/write endpoint yet.
+No in-repository code can truthfully assert that the external ChatGPT connector account has gained
+`release:candidate`. Until the platform consumes the outbox and performs its own identity-scope
+projection, the only valid state is:
+
+```text
+AUTHORIZED_PENDING_EXTERNAL_CAPABILITY_PROJECTION
+PENDING_EXTERNAL_CONNECTOR
+```
+
+A future external projector must preserve:
+
+```text
+release:candidate
+!= submitOwnerApproval
+!= startDeployment
+!= tag_publish
+!= baseline_registry_finalize
+!= runtime_switch
+```
+
+## Idempotency and audit
+
+Every broker operation requires an idempotency key. A receipt SQLite database stores only:
+
+- idempotency key;
+- capability;
+- request SHA-256;
+- response SHA-256;
+- sanitized upstream response;
+- timestamp.
+
+An exact retry returns the same receipt. Reusing an idempotency key with different content fails.
+
+## Unattended invariant
+
+Once the Owner policy and narrow service credentials are provisioned:
+
+```text
+Owner durable grant
+→ broker decision
+→ canonical governed task projection / release candidate preparation
+→ worker / trusted builder
+```
+
+No repeated Owner interaction is required for those preparation capabilities.
+
+Production execution remains:
+
+```text
+release candidate
+→ immutable approval envelope
+→ Owner signature
+→ deployment
+```
+
+The durable grant never substitutes for the Owner signature.

docs/superpowers/specs/2026-08-19-gpt-https-edge-activation.md

@@ -0,0 +1,114 @@
+# Owner-approved GPT HTTPS edge activation and rollback
+
+## Scope
+
+This controller is the final host-side control mechanism before the disabled-by-default GPT HTTPS
+edge may bind publicly.
+
+Repository implementation and public activation remain separate states:
+
+```text
+IMPLEMENTED
+≠
+OWNER_APPROVED
+≠
+ACTIVATED
+≠
+RUNTIME_VERIFIED
+```
+
+The controller never creates owner approval, TLS assets, DNS, firewall rules or ingress bearer
+credentials.
+
+## Approval binding
+
+The owner approval file is root-owned mode `0600` and has schema:
+
+```json
+{
+  "schema": "aios.gpt-https-edge.public-approval.v1",
+  "approval_id": "owner-public-edge-...",
+  "candidate_oid": "<exact governed edge candidate>",
+  "domain": "tasks.example.com",
+  "public_bind": true,
+  "bind_host": "0.0.0.0",
+  "port": 18443,
+  "tls_cert_sha256": "<exact certificate sha256>",
+  "issued_at": "2026-08-19T00:00:00Z",
+  "not_before": "2026-08-19T00:00:00Z",
+  "expires_at": "2026-08-20T00:00:00Z",
+  "owner": "<owner identity>"
+}
+```
+
+Approval lifetime is at most seven days. Candidate/domain/bind/port/certificate are immutable within
+one approval.
+
+Changing any of those requires a new owner approval.
+
+## Check
+
+```text
+aios_gpt_https_edge_activation.py check --candidate <GE>
+```
+
+Checks:
+
+- approval ownership/mode/schema/time window;
+- exact candidate binding;
+- exact certificate hash;
+- certificate/private-key pair;
+- certificate hostname;
+- certificate remaining lifetime;
+- loopback GPT ingress service health.
+
+It performs no public network mutation.
+
+## Activate
+
+`activate` additionally:
+
+- writes one exact systemd public-bind drop-in atomically;
+- enables/restarts only the HTTPS edge unit;
+- verifies effective ExecStart points at the immutable candidate release;
+- verifies effective public-bind/domain environment;
+- performs a TLS handshake to `127.0.0.1:18443` using the approved public hostname;
+- writes a durable activation receipt.
+
+Any failure after drop-in mutation automatically invokes rollback.
+
+Exact replay of an already-active matching approval returns the durable receipt without restarting
+or creating duplicate state.
+
+## Rollback
+
+`rollback`:
+
+- stops and disables only the HTTPS edge;
+- removes only the controller-owned public-bind drop-in;
+- daemon-reloads systemd;
+- verifies edge inactive;
+- writes a durable rollback receipt.
+
+It does not delete:
+
+- owner approval;
+- TLS certificate/private key;
+- ingress bearer token;
+- internal loopback ingress;
+- Task/Job state;
+- Git branches.
+
+Rollback is therefore replay-safe and does not require reconstructing secrets.
+
+## Not covered
+
+This controller intentionally does not own:
+
+- DNS records;
+- host firewall/NAT/security-group changes;
+- certificate issuance/renewal;
+- OpenAI Custom GPT domain verification;
+- public Internet acceptance from outside the host.
+
+Those remain separate owner/environment gates.

docs/superpowers/specs/2026-08-19-gpt-https-edge.md

@@ -0,0 +1,120 @@
+# Disabled-by-default HTTPS edge for external GPT Online Task access
+
+## Purpose
+
+The internal ingress (`127.0.0.1:18945`) remains loopback-only. This edge is the only component
+allowed to become Internet-facing, and only after a separate owner approval.
+
+It forwards only:
+
+```text
+POST /v1/read-only-tasks
+GET  /v1/read-only-tasks/{request_id}
+GET  /openapi.json
+GET  /health
+```
+
+It never proxies:
+
+```text
+18892 canonical Bridge
+18940 Task API
+WriterLease
+commit
+verification
+release
+tag
+baseline
+arbitrary URLs
+```
+
+## Default state
+
+The systemd unit is disabled by policy unless the owner creates:
+
+```text
+/etc/aios-gitops/gpt-https-edge.approved
+```
+
+The base unit still binds only to:
+
+```text
+127.0.0.1:18443
+AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=0
+```
+
+Non-loopback bind is rejected by code unless an owner-controlled drop-in sets:
+
+```text
+AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=1
+AIOS_GPT_EDGE_HOST=0.0.0.0
+AIOS_GPT_EDGE_PUBLIC_DOMAIN=<approved hostname>
+```
+
+This repository candidate does not create that drop-in or approval marker.
+
+## TLS
+
+TLS certificate/key are owner-managed systemd credentials:
+
+```text
+/etc/aios-gitops/gpt-https-edge-tls.crt
+/etc/aios-gitops/gpt-https-edge-tls.key
+```
+
+The service requires TLS 1.2 or newer, disables TLS compression, and never logs or returns key
+material.
+
+Certificate/key issuance and renewal are external owner responsibilities until a separately
+governed ACME workflow is implemented.
+
+## Request protections
+
+The edge provides:
+
+- 64 KiB request cap;
+- 256 KiB upstream response cap;
+- JSON validation before proxying;
+- fixed upstream `http://127.0.0.1:18945`;
+- Authorization-only header forwarding;
+- no Cookie/X-Forwarded-* trust;
+- route/method allowlist;
+- per-source-IP in-memory rate limiting;
+- bounded concurrent request handling;
+- optional exact Host-header enforcement;
+- HSTS / no-store / nosniff / no-referrer response headers.
+
+The rate limiter is intentionally simple and local. Multi-node/global abuse protection is outside
+the current one-host architecture.
+
+## Authentication
+
+The HTTPS edge does not create a new authentication scheme. It forwards the external Bearer
+credential to the loopback GPT ingress, which verifies `gpt_ingress_token`.
+
+The edge never receives the AIOS control-plane token.
+
+## Deployment boundary
+
+Public activation is a high-risk network operation and requires explicit owner approval. Before
+activation verify:
+
+```text
+host-side GPT ingress E2E=VERIFIED
+owner ingress token provisioned
+TLS certificate/key provisioned
+public hostname owned and DNS configured
+firewall/network exposure reviewed
+explicit public-bind approval recorded
+rollback plan available
+```
+
+Only after those gates may an owner-authorized runtime drop-in bind this service publicly.
+
+Until then:
+
+```text
+GPT_HTTPS_EDGE_IMPLEMENTATION=candidate
+PUBLIC_NETWORK_EXPOSURE=NONE
+PUBLIC_CUSTOM_GPT_DIRECT_AI0S_ACCESS=NOT_VERIFIED
+```

docs/superpowers/specs/2026-08-19-gpt-online-task-ingress.md

@@ -0,0 +1,138 @@
+# External GPT → Online Task ingress
+
+## Scope
+
+This component closes the current architectural gap between an external Custom GPT and AIOS
+without exposing the internal Task API (`18940`) or canonical Bridge (`18892`) directly.
+
+The first supported public-facing capability is intentionally read-only:
+
+```text
+authenticated external intent
+→ host-side ingress
+→ canonical /v2/governed-task
+→ canonical /v2/execution-envelope
+→ persistent generic worker
+→ source_capture result
+→ official control-plane verification
+```
+
+The ingress itself never claims, heartbeats, verifies, commits, acquires WriterLease, or writes
+protected refs.
+
+## Security boundary
+
+The service binds only to loopback (`127.0.0.1:18945`). A future owner-approved TLS reverse proxy
+or tunnel may expose this service externally. Direct non-loopback binding is denied by code.
+
+Two independent credentials are projected by systemd:
+
+- `gpt_ingress_token`: authenticates the external caller to this ingress.
+- `agent_task_control_token`: lets the ingress call the loopback AIOS control plane.
+
+Neither credential is stored in Git, SQLite, task payloads, results, logs, or API responses.
+
+The ingress has no repository filesystem access.
+
+## Caller-controlled fields
+
+Allowed:
+
+```text
+request_id
+goal
+instructions
+success_criteria
+constraints.max_files
+constraints.focus_paths
+constraints.notes
+```
+
+Not caller-controlled:
+
+```text
+repository
+base_oid
+expected_branch_oid
+target_branch
+branch_mode
+generation
+risk
+priority
+max_attempts
+capabilities
+action/adapter
+WriterLease
+worker identity
+credentials
+verification decision
+main/production/tag/baseline permissions
+```
+
+All external tasks are hard-bound to:
+
+```text
+repository=aios/aios
+capabilities=[git.read, git.test]
+action=source_capture
+writer_lease=false
+repository_mutation=forbidden
+control_plane_side_effects=forbidden
+credential_access=forbidden
+```
+
+## Idempotency
+
+`request_id` is the durable external idempotency boundary.
+
+The ingress stores only:
+
+```text
+request hash
+derived AIOS task id
+canonical job id
+base oid
+sanitized receipt
+state
+```
+
+An exact retry returns the same receipt. Reusing a `request_id` with different content returns
+`IDEMPOTENCY_CONFLICT`.
+
+A crash after canonical task creation can recover the same binding from the Task API rather than
+creating a replacement Task.
+
+## Status
+
+`GET /v1/read-only-tasks/{request_id}` returns only sanitized state:
+
+```text
+task/job ids
+status
+attempt
+whether claimed
+result classification
+official verification decision/classification
+source snapshot hash/count when available
+```
+
+It does not proxy raw Task metadata, execution envelopes, result text, credentials, or source files.
+
+## Public exposure
+
+This candidate does **not** make AIOS publicly reachable. Public exposure requires a separately
+approved HTTPS ingress/reverse proxy with:
+
+- TLS;
+- external authentication lifecycle;
+- rate limiting / abuse controls;
+- request size limits;
+- observability;
+- explicit owner approval for the network exposure.
+
+Until that step is verified:
+
+```text
+HOST_SIDE_EXTERNAL_GPT_ADAPTER=implemented/staging-candidate
+PUBLIC_CUSTOM_GPT_DIRECT_AI0S_ACCESS=not verified
+```

docs/superpowers/specs/2026-08-19-release-candidate-capability-contract.md

@@ -0,0 +1,87 @@
+# Release candidate capability contract remediation
+
+## Problem
+
+The release-preparation worker can be correctly registered with the raw capability:
+
+```text
+release:candidate
+```
+
+while the canonical Task API matcher still rejects the worker.
+
+Before this repair, the capability matrix is:
+
+```text
+worker raw      = ["release:candidate"]
+worker effective= []
+task required   = ["exact:release:candidate"]
+eligible        = false
+```
+
+The Task therefore remains `QUEUED`, `attempt=0`, and cannot be claimed even though both the Task and
+worker use the same literal capability.
+
+## Root cause
+
+`aios_capability_contract.py` is deliberately fail-closed:
+
+- task-side unknown names become `exact:<name>`;
+- worker-side unknown names are discarded;
+- wildcard worker capabilities are ignored.
+
+`release:candidate` was absent from both canonical alias tables.
+
+## Repair
+
+Add exactly one semantic capability to both vocabularies:
+
+```text
+release:candidate → release:candidate
+```
+
+Do not add:
+
+```text
+release:*
+submitOwnerApproval
+startDeployment
+tag_publish
+baseline_registry_finalize
+runtime_switch
+```
+
+Unknown names and wildcards must continue to fail closed.
+
+## Runtime recovery
+
+After the governed remediation candidate is officially verified:
+
+1. install the candidate capability-contract module as a higher-precedence Task API overlay;
+2. restart only `aios-agent-task-api.service`;
+3. require `/health` success;
+4. require the fixed existing release Task to remain the same Task/idempotency;
+5. let the already-active dedicated release worker poll and claim it automatically;
+6. require heartbeat, trusted-builder result and official verification;
+7. do not manually claim or create a replacement live Task.
+
+The fixed V2R2C2 live Task may complete as the acceptance candidate it was originally bound to.
+Because this remediation is a later source commit, the first release candidate that includes the
+matcher repair should use the next immutable version (for example `V2R2C3`) rather than rewriting an
+already-created V2R2C2 candidate.
+
+## Governance boundary
+
+This repair affects only Task eligibility. It grants no credential and performs no release action by
+itself.
+
+The dedicated release worker still requires its independent `release:candidate` credential, and the
+following remain separate:
+
+```text
+Owner signed approval
+deployment
+tag publication
+baseline finalization
+runtime switch
+```

docs/superpowers/specs/2026-08-19-release-lineage-contract.md

@@ -0,0 +1,82 @@
+# Fixed V2R2C2 release lineage contract
+
+## Why this contract exists
+
+AIOS has multiple engineering commits in the same ancestry, and they serve different purposes.
+
+The fixed V2R2C2 live Task was created by the dedicated Release Worker activation. Its immutable
+release source is therefore the Release Worker implementation candidate:
+
+```text
+71e736b05fab13efc06af7a4fe2dc022c1cf9a85
+```
+
+Later engineering commits are descendants but are not retroactive V2R2C2 release sources:
+
+```text
+17deb91ca7a187dc81766945b1174e854f86bd25
+  capability-contract remediation
+
+b4b4416c93fedbe451d655d49e28f915b8fd38e6
+  Autonomous Control Plane engineering candidate
+```
+
+Conflating "current engineering base" with "immutable release source" caused ACP activation to
+fail closed with `FIXED_RELEASE_SOURCE_DRIFT`.
+
+## Canonical fixed lineage
+
+```text
+Task:
+r42-release-candidate-v2r2c2-live-20260819
+
+candidate_id:
+r42-release-candidate-v2r2c2-20260819
+
+source_commit:
+71e736b05fab13efc06af7a4fe2dc022c1cf9a85
+
+version:
+V2R2C2
+
+required_capabilities:
+["release:candidate"]
+```
+
+This contract is immutable for V2R2C2.
+
+## What may change
+
+The following are lifecycle state, not release identity:
+
+```text
+status
+attempt
+max_attempts
+revision
+claimed_by
+lease
+verification
+```
+
+They may change through official retry/claim/result/verification without changing the release
+lineage.
+
+## What must never be "repaired"
+
+Do not rewrite the fixed V2R2C2 Task source to a later engineering commit.
+
+A later engineering source that must itself be released receives a later immutable release version
+(for example V2R2C3), not a mutation of V2R2C2.
+
+## Activation invariant
+
+ACP activation must:
+
+1. verify its own governed engineering candidate separately;
+2. read the live fixed V2R2C2 Task;
+3. validate that Task through `validate_fixed_v2r2c2_task()`;
+4. never compare the fixed release source to the ACP engineering base/candidate;
+5. never mutate the fixed Task merely to satisfy activation.
+
+This makes engineering lineage and release lineage explicit and independently verifiable.

docs/superpowers/specs/2026-08-19-release-preparation-online-agent.md

@@ -0,0 +1,142 @@
+# Release Preparation Online Agent
+
+## Purpose
+
+This worker closes the release-preparation execution gap without widening the ChatGPT connector
+identity.
+
+The dedicated Agent identity is:
+
+```text
+online-release-worker-cn-network-mkt-20260819
+```
+
+Its only advertised task capability is:
+
+```text
+release:candidate
+```
+
+The worker may create an immutable release candidate, but it never submits Owner approval and never
+starts deployment, publishes a tag, finalizes a baseline, switches runtime or writes Git.
+
+## Credentials
+
+The worker uses two independently scoped credentials:
+
+```text
+agent_token
+release_candidate_token
+```
+
+`agent_token` is the Online Agent registration credential. It must be encrypted into systemd's
+credential store and must never be written to Git, Task metadata, evidence or logs.
+
+`release_candidate_token` is provisioned during Owner-authorized runtime activation. Only its SHA-256 record is stored in the release API token registry; the plaintext is immediately encrypted into systemd credstore and the temporary plaintext is removed. Its only allowed scope is:
+
+```text
+release:candidate
+```
+
+It is explicitly not an Owner signing key and must not grant:
+
+```text
+submitOwnerApproval
+startDeployment
+tag_publish
+baseline_registry_finalize
+runtime_switch
+```
+
+The unit reads both credentials through systemd `LoadCredentialEncrypted`; neither plaintext token is kept in the unit, repository, Task metadata, evidence, or logs.
+
+## Task contract
+
+The worker will only claim Tasks whose exact required capability set is:
+
+```json
+["release:candidate"]
+```
+
+The Task must provide a bounded `constraints.release_candidate` payload (metadata is accepted only for compatibility):
+
+```json
+{
+  "candidate_id": "...",
+  "source_commit": "<40 hex>",
+  "version": "V2R2C2",
+  "projection_id": "projection-..."
+}
+```
+
+No caller-supplied URL, method, headers, command, shell, environment, credential path, deployment
+operation or arbitrary release route is accepted.
+
+The release API base is host configuration, not Task input. The only release mutation invoked by the
+worker is:
+
+```text
+POST /createReleaseCandidate
+```
+
+## Lifecycle
+
+```text
+poll
+→ claim
+→ heartbeat
+→ createReleaseCandidate
+→ result submit
+→ official verifier
+```
+
+The result declares:
+
+```text
+writer_lease=NONE
+changed_files=[]
+direct_main_write=false
+production_write=false
+tag_write=false
+baseline_write=false
+runtime_switch=false
+owner_approval_submitted=false
+```
+
+## Remaining high-risk gate
+
+Successful release-candidate creation does not authorize Production.
+
+The next stage remains:
+
+```text
+release candidate
+→ immutable approval envelope
+→ Owner signature
+→ startDeployment
+```
+
+That separation is mandatory even when release preparation is fully unattended.
+
+
+## Unattended availability
+
+The release worker has its own systemd watchdog rather than modifying the shared worker-availability
+guard contract.
+
+Every successful Task poll and Task heartbeat emits:
+
+```text
+WATCHDOG=1
+```
+
+and the service uses:
+
+```text
+Type=notify
+WatchdogSec=90
+Restart=always
+```
+
+A process that is technically alive but no longer polling/heartbeating therefore gets restarted by
+systemd. The worker also writes `worker-liveness.json` for operator diagnostics.

tests/test_agent_task_initial_execution_projection.py

@@ -0,0 +1,452 @@
+from __future__ import annotations
+
+from pathlib import Path
+import hashlib
+import json
+import sqlite3
+import sys
+
+import pytest
+
+ROOT=Path(__file__).resolve().parents[1]
+sys.path.insert(0,str(ROOT/"active-runtime/app"))
+
+from aios_agent_task_initial_execution_projection import (
+    InitialExecutionProjectionError,
+    InitialExecutionProjectionRequest,
+    InitialExecutionProjectionStore,
+    SourceCaptureProjectionRecoveryRequest,
+)
+from aios_online_agent_execution import (
+    GovernedExecutionBinding,
+    project_execution_envelope,
+)
+
+AUTHORIZATION="owner-initial-execution-projection-recovery-20260820"
+TASK_ID="task-initial-projection-a"
+JOB_ID="job-initial-projection-a"
+BASE="a"*40
+BRANCH="agent/task-initial-projection-a"
+IDEMPOTENCY="task-initial-projection-binding-v1"
+CAPABILITY="agent.runtime.openclaw"
+
+
+@pytest.fixture(autouse=True)
+def _auth(monkeypatch):
+    monkeypatch.setenv(
+        "AIOS_AGENT_TASK_INITIAL_EXECUTION_PROJECTION_AUTHORIZATION_ID",
+        AUTHORIZATION,
+    )
+
+
+def canon(value):
+    return hashlib.sha256(
+        json.dumps(value,sort_keys=True,separators=(",",":")).encode()
+    ).hexdigest()
+
+
+def binding():
+    return GovernedExecutionBinding.from_mapping({
+        "task_id":TASK_ID,
+        "job_id":JOB_ID,
+        "repository":"aios/aios",
+        "base_oid":BASE,
+        "expected_branch_oid":BASE,
+        "target_branch":BRANCH,
+        "branch_mode":"CREATE_NEW",
+        "generation":1,
+        "idempotency_key":IDEMPOTENCY,
+    })
+
+
+def runtime_input():
+    return {
+        "goal":"Supervise release preparation",
+        "instructions":"Observe and resume only the same lineage.",
+        "success_criteria":[{"id":"heartbeat","description":"Heartbeat observed"}],
+        "constraints":{"production_write":False},
+        "scope":{"scope_type":"legacy","tenant_id":None,"project_id":None},
+        "trace_id":TASK_ID,
+    }
+
+
+def envelope(*, job_id=JOB_ID, action="agent_runtime", capability=CAPABILITY):
+    bound=binding()
+    if job_id != JOB_ID:
+        bound=GovernedExecutionBinding.from_mapping({**bound.public(),"job_id":job_id})
+    return project_execution_envelope(
+        bound,
+        {
+            "action":action,
+            "adapter":"openclaw" if action=="agent_runtime" else "source_capture",
+            "capability_scope":[capability],
+            "runtime_input":runtime_input() if action=="agent_runtime" else None,
+        },
+    )
+
+
+def make_db(tmp_path, *, claimed_by=None, lease=False, metadata=None, caps=None, event="TASK_CREATED"):
+    path=tmp_path/"state.sqlite3"
+    con=sqlite3.connect(path)
+    con.executescript("""
+    CREATE TABLE tasks(
+      task_id TEXT PRIMARY KEY,
+      goal TEXT NOT NULL,
+      instructions TEXT NOT NULL,
+      success_criteria_json TEXT NOT NULL,
+      required_capabilities_json TEXT NOT NULL,
+      constraints_json TEXT NOT NULL,
+      metadata_json TEXT NOT NULL,
+      deadline TEXT,
+      priority INTEGER NOT NULL,
+      max_attempts INTEGER NOT NULL,
+      attempt INTEGER NOT NULL DEFAULT 0,
+      status TEXT NOT NULL,
+      claimed_by TEXT,
+      lease_token_hash TEXT,
+      lease_expires_at TEXT,
+      result_json TEXT,
+      result_digest TEXT,
+      result_idempotency_key TEXT,
+      verification_json TEXT,
+      revision INTEGER NOT NULL DEFAULT 1,
+      created_at TEXT NOT NULL,
+      updated_at TEXT NOT NULL
+    );
+    CREATE TABLE events(
+      event_id INTEGER PRIMARY KEY AUTOINCREMENT,
+      task_id TEXT,
+      agent_id TEXT,
+      event_type TEXT NOT NULL,
+      payload_json TEXT NOT NULL,
+      created_at TEXT NOT NULL
+    );
+    """)
+    metadata={} if metadata is None else metadata
+    caps=[] if caps is None else caps
+    con.execute(
+        "INSERT INTO tasks VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
+        (
+            TASK_ID,"Supervise release preparation",
+            "Observe and resume only the same lineage.",
+            json.dumps([{"id":"heartbeat","description":"Heartbeat observed"}]),
+            json.dumps(caps),json.dumps({"production_write":False}),json.dumps(metadata),
+            None,100,20,0,"QUEUED",claimed_by,
+            "lease-hash" if lease else None,
+            "2099-01-01T00:00:00Z" if lease else None,
+            None,None,None,None,1,"2026-08-20T00:00:00Z","2026-08-20T00:00:00Z",
+        ),
+    )
+    con.execute(
+        "INSERT INTO events(task_id,agent_id,event_type,payload_json,created_at) VALUES (?,?,?,?,?)",
+        (TASK_ID,None,event,"{}","2026-08-20T00:00:00Z"),
+    )
+    con.commit();con.close()
+    return path
+
+
+def request(*, digest=None, key="initial-projection-a-v1"):
+    return InitialExecutionProjectionRequest.from_mapping({
+        "expected_status":"QUEUED",
+        "expected_revision":1,
+        "expected_attempt":0,
+        "expected_metadata_sha256":canon({}),
+        "expected_required_capabilities_sha256":canon([]),
+        "expected_job_id":JOB_ID,
+        "expected_execution_digest":digest or envelope()["execution_digest"],
+        "required_capability":CAPABILITY,
+        "adapter":"openclaw",
+        "authorization_id":AUTHORIZATION,
+        "idempotency_key":key,
+        "reason":"initial governed supervisor runtime projection",
+    })
+
+
+def resolver(_task_id,_job_id,_capability,_adapter,received_runtime_input):
+    assert received_runtime_input==runtime_input()
+    return envelope()
+
+
+def source_capture_result():
+    old=envelope(action="source_capture",capability="git.read")
+    return {
+        "task_id":TASK_ID,
+        "job_id":JOB_ID,
+        "repository":"aios/aios",
+        "base_oid":BASE,
+        "expected_branch_oid":BASE,
+        "target_branch":BRANCH,
+        "generation":1,
+        "status":"SUCCEEDED",
+        "classification":"NEEDS_REWORK_SOURCE_CAPTURE",
+        "execution_mode":"source_capture",
+        "execution_digest":old["execution_digest"],
+        "changed_files":[],
+        "writer_lease":"NONE",
+        "verification_state":"NEEDS_REWORK",
+        "direct_main_write":False,
+        "production_write":False,
+        "force_push":False,
+        "protected_ref_bypass":False,
+        "secret_material_exposed":False,
+        "source_snapshot":{"schema":"test.snapshot.v1","payload_sha256":"d"*64},
+    }
+
+
+def make_source_capture_rework_db(tmp_path, *, mutate=None, events=None):
+    path=make_db(tmp_path)
+    result=source_capture_result()
+    verification={
+        "decision":"NEEDS_REWORK",
+        "reason":"SOURCE_CAPTURE_COMPLETE_RUNTIME_PROJECTION_REQUIRED",
+    }
+    if mutate:
+        mutate(result,verification)
+    event_types=events or [
+        "TASK_CLAIMED","TASK_HEARTBEAT","RESULT_SUBMITTED","TASK_VERIFIED"
+    ]
+    con=sqlite3.connect(path)
+    con.execute(
+        "UPDATE tasks SET attempt=1,status='QUEUED',claimed_by=NULL,revision=5,"
+        "result_json=?,result_digest=?,result_idempotency_key=?,verification_json=?",
+        (
+            json.dumps(result,sort_keys=True,separators=(",",":")),
+            canon(result),"source-capture-result-v1",
+            json.dumps(verification,sort_keys=True,separators=(",",":")),
+        ),
+    )
+    for event_type in event_types:
+        payload={}
+        if event_type=="TASK_VERIFIED":
+            payload={"decision":"NEEDS_REWORK","next_state":"QUEUED"}
+        con.execute(
+            "INSERT INTO events(task_id,agent_id,event_type,payload_json,created_at) "
+            "VALUES (?,?,?,?,?)",
+            (TASK_ID,"generic-worker",event_type,json.dumps(payload),"2026-08-20T00:01:00Z"),
+        )
+    con.commit();con.close()
+    return path,result,verification
+
+
+def recovery_request(result,verification,*,digest=None,key="source-capture-recovery-a-v1"):
+    return SourceCaptureProjectionRecoveryRequest.from_mapping({
+        "expected_status":"QUEUED",
+        "expected_revision":5,
+        "expected_attempt":1,
+        "expected_metadata_sha256":canon({}),
+        "expected_required_capabilities_sha256":canon([]),
+        "previous_verification_sha256":canon(verification),
+        "expected_source_capture_result_sha256":canon(result),
+        "expected_source_capture_execution_digest":result["execution_digest"],
+        "expected_job_id":JOB_ID,
+        "expected_execution_digest":digest or envelope()["execution_digest"],
+        "required_capability":CAPABILITY,
+        "adapter":"openclaw",
+        "authorization_id":AUTHORIZATION,
+        "idempotency_key":key,
+        "reason":"first durable runtime projection after governed source capture",
+    })
+
+
+def test_initial_projection_cas_persists_envelope_capability_and_event(tmp_path):
+    path=make_db(tmp_path)
+    result=InitialExecutionProjectionStore(path,resolver).project(TASK_ID,request())
+    assert result["revision"]==2
+    assert result["attempt"]==0
+    assert result["idempotent"] is False
+    con=sqlite3.connect(path)
+    row=con.execute(
+        "SELECT status,attempt,revision,claimed_by,metadata_json,required_capabilities_json FROM tasks"
+    ).fetchone()
+    assert row[:4]==("QUEUED",0,2,None)
+    metadata=json.loads(row[4])
+    assert metadata["execution"]["action"]=="agent_runtime"
+    assert metadata["execution"]["adapter"]=="openclaw"
+    assert metadata["execution"]["governance"]["writer_lease"] is False
+    assert json.loads(row[5])==[CAPABILITY]
+    events=con.execute("SELECT event_type,payload_json FROM events ORDER BY event_id").fetchall()
+    assert [event[0] for event in events]==[
+        "TASK_CREATED","TASK_INITIAL_EXECUTION_PROJECTION_AUTHORIZED"
+    ]
+    payload=json.loads(events[-1][1])
+    assert payload["job_id"]==JOB_ID
+    assert payload["main_write"] is False
+    assert payload["production_write"] is False
+
+
+def test_initial_projection_idempotent_exact_replay(tmp_path):
+    path=make_db(tmp_path)
+    store=InitialExecutionProjectionStore(path,resolver)
+    first=store.project(TASK_ID,request())
+    second=store.project(TASK_ID,request())
+    assert first["idempotent"] is False
+    assert second["idempotent"] is True
+    assert second["execution_digest"]==first["execution_digest"]
+    con=sqlite3.connect(path)
+    assert con.execute("SELECT COUNT(*) FROM task_initial_execution_projections").fetchone()[0]==1
+    assert con.execute("SELECT COUNT(*) FROM events").fetchone()[0]==2
+
+
+def test_source_capture_recovery_projects_first_runtime_envelope(tmp_path):
+    path,result,verification=make_source_capture_rework_db(tmp_path)
+    response=InitialExecutionProjectionStore(path,resolver).project_from_source_capture(
+        TASK_ID,recovery_request(result,verification)
+    )
+    assert response["revision"]==6
+    assert response["attempt"]==1
+    assert response["recovery_from_source_capture"] is True
+    assert response["idempotent"] is False
+    con=sqlite3.connect(path)
+    row=con.execute(
+        "SELECT status,attempt,revision,claimed_by,metadata_json,"
+        "required_capabilities_json,result_json,verification_json FROM tasks"
+    ).fetchone()
+    assert row[:4]==("QUEUED",1,6,None)
+    metadata=json.loads(row[4])
+    assert metadata["execution"]["action"]=="agent_runtime"
+    assert metadata["execution"]["adapter"]=="openclaw"
+    assert metadata["execution"]["governance"]["writer_lease"] is False
+    assert json.loads(row[5])==[CAPABILITY]
+    assert json.loads(row[6])==result
+    assert json.loads(row[7])==verification
+    events=con.execute("SELECT event_type,payload_json FROM events ORDER BY event_id").fetchall()
+    assert [event[0] for event in events]==[
+        "TASK_CREATED","TASK_CLAIMED","TASK_HEARTBEAT","RESULT_SUBMITTED",
+        "TASK_VERIFIED","TASK_INITIAL_EXECUTION_PROJECTION_AUTHORIZED",
+    ]
+    payload=json.loads(events[-1][1])
+    assert payload["recovery_from_source_capture"] is True
+    assert payload["source_capture_execution_digest"]==result["execution_digest"]
+    assert payload["main_write"] is False
+    assert payload["production_write"] is False
+
+
+def test_source_capture_recovery_idempotent_exact_replay(tmp_path):
+    path,result,verification=make_source_capture_rework_db(tmp_path)
+    store=InitialExecutionProjectionStore(path,resolver)
+    first=store.project_from_source_capture(TASK_ID,recovery_request(result,verification))
+    second=store.project_from_source_capture(TASK_ID,recovery_request(result,verification))
+    assert first["idempotent"] is False
+    assert second["idempotent"] is True
+    assert second["execution_digest"]==first["execution_digest"]
+    con=sqlite3.connect(path)
+    assert con.execute(
+        "SELECT COUNT(*) FROM task_source_capture_projection_recoveries"
+    ).fetchone()[0]==1
+    assert con.execute("SELECT COUNT(*) FROM events").fetchone()[0]==6
+
+
+@pytest.mark.parametrize(
+    "mutate,error",
+    [
+        (lambda result,verification: result.__setitem__("changed_files",["x"]),
+         "SOURCE_CAPTURE_RESULT_CONTRACT_INVALID"),
+        (lambda result,verification: result.__setitem__("job_id","job-other"),
+         "SOURCE_CAPTURE_RESULT_CONTRACT_INVALID"),
+        (lambda result,verification: verification.__setitem__("decision","SUCCEEDED"),
+         "SOURCE_CAPTURE_NEEDS_REWORK_VERIFICATION_REQUIRED"),
+    ],
+)
+def test_source_capture_recovery_rejects_unsafe_result_or_verification(
+    tmp_path,mutate,error
+):
+    path,result,verification=make_source_capture_rework_db(tmp_path,mutate=mutate)
+    with pytest.raises(InitialExecutionProjectionError,match=error):
+        InitialExecutionProjectionStore(path,resolver).project_from_source_capture(
+            TASK_ID,recovery_request(result,verification)
+        )
+
+
+@pytest.mark.parametrize(
+    "events",
+    [
+        ["TASK_CLAIMED","RESULT_SUBMITTED","TASK_VERIFIED"],
+        ["TASK_CLAIMED","TASK_HEARTBEAT","RESULT_SUBMITTED","TASK_VERIFIED","TASK_CLAIMED"],
+        ["TASK_CLAIMED","TASK_HEARTBEAT","TASK_VERIFIED","RESULT_SUBMITTED"],
+    ],
+)
+def test_source_capture_recovery_rejects_invalid_event_chain(tmp_path,events):
+    path,result,verification=make_source_capture_rework_db(tmp_path,events=events)
+    with pytest.raises(
+        InitialExecutionProjectionError,match="SOURCE_CAPTURE_RECOVERY_EVENT_CHAIN_INVALID"
+    ):
+        InitialExecutionProjectionStore(path,resolver).project_from_source_capture(
+            TASK_ID,recovery_request(result,verification)
+        )
+
+
+def test_source_capture_recovery_rejects_execution_digest_drift(tmp_path):
+    path,result,verification=make_source_capture_rework_db(tmp_path)
+    with pytest.raises(InitialExecutionProjectionError,match="EXECUTION_DIGEST_DRIFT"):
+        InitialExecutionProjectionStore(path,resolver).project_from_source_capture(
+            TASK_ID,recovery_request(result,verification,digest="f"*64)
+        )
+
+
+def test_source_capture_recovery_request_is_strict(tmp_path):
+    _path,result,verification=make_source_capture_rework_db(tmp_path)
+    data=recovery_request(result,verification).public()
+    data["metadata_patch"]={"execution":"forbidden"}
+    with pytest.raises(InitialExecutionProjectionError,match="FIELDS_INVALID"):
+        SourceCaptureProjectionRecoveryRequest.from_mapping(data)
+
+
+@pytest.mark.parametrize(
+    "kwargs,error",
+    [
+        ({"claimed_by":"worker-a"},"TASK_ALREADY_CLAIMED"),
+        ({"lease":True},"ACTIVE_LEASE_DENY"),
+        ({"metadata":{"execution":{}}},"TASK_EXECUTION_ALREADY_PROJECTED"),
+        ({"caps":["git.read"]},"TASK_INITIAL_CAPABILITIES_MUST_BE_EMPTY"),
+        ({"event":"TASK_CLAIMED"},"TASK_INITIAL_EVENT_CHAIN_REQUIRED"),
+    ],
+)
+def test_initial_projection_rejects_noninitial_state(tmp_path,kwargs,error):
+    path=make_db(tmp_path,**kwargs)
+    with pytest.raises(InitialExecutionProjectionError,match=error):
+        InitialExecutionProjectionStore(path,resolver).project(TASK_ID,request())
+
+
+def test_initial_projection_rejects_binding_drift(tmp_path):
+    path=make_db(tmp_path)
+    def wrong(*_args):
+        return envelope(job_id="job-other")
+    with pytest.raises(InitialExecutionProjectionError,match="GOVERNED_BINDING_CONFLICT"):
+        InitialExecutionProjectionStore(path,wrong).project(TASK_ID,request())
+
+
+def test_initial_projection_rejects_nonruntime_envelope(tmp_path):
+    path=make_db(tmp_path)
+    def wrong(*_args):
+        return envelope(action="source_capture",capability="git.read")
+    with pytest.raises(InitialExecutionProjectionError):
+        InitialExecutionProjectionStore(path,wrong).project(TASK_ID,request())
+
+
+def test_initial_projection_rejects_execution_digest_drift(tmp_path):
+    path=make_db(tmp_path)
+    with pytest.raises(InitialExecutionProjectionError,match="EXECUTION_DIGEST_DRIFT"):
+        InitialExecutionProjectionStore(path,resolver).project(
+            TASK_ID,request(digest="f"*64)
+        )
+
+
+def test_request_is_narrow_and_initial_state_bound():
+    data=request().public()
+    data["arbitrary_metadata"]={"x":1}
+    with pytest.raises(InitialExecutionProjectionError,match="FIELDS_INVALID"):
+        InitialExecutionProjectionRequest.from_mapping(data)
+    data=request().public();data["expected_attempt"]=1
+    with pytest.raises(InitialExecutionProjectionError,match="INITIAL_STATE_REQUIRED"):
+        InitialExecutionProjectionRequest.from_mapping(data)
+
+
+def test_initial_projection_unit_is_unprivileged_and_owner_credential_bound():
+    unit=(ROOT/"deploy/systemd/aios-agent-task-initial-execution-projection.service").read_text()
+    assert "User=aios-agent-task" in unit
+    assert "Group=aios-agent-task" in unit
+    assert "LoadCredential=agent_task_control_token:" in unit
+    assert "Environment=AIOS_GOVERNED_WRITE_API_URL=http://127.0.0.1:18892" in unit
+    assert "RestrictAddressFamilies=AF_UNIX AF_INET" in unit
+    assert "ReadWritePaths=/var/lib/aios-agent-task" in unit
+    assert "User=root" not in unit

tests/test_all_modified_files_packager.py

@@ -0,0 +1,252 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+import json
+from pathlib import Path
+import subprocess
+import sys
+import zipfile
+
+import pytest
+
+ROOT=Path(__file__).resolve().parents[1]
+APP=ROOT/"active-runtime/app"
+sys.path.insert(0,str(APP))
+
+import aios_all_modified_files_packager as packager
+
+
+def init_repo(tmp_path: Path) -> tuple[Path, Path, str]:
+    bare=tmp_path/"repo.git"
+    work=tmp_path/"work"
+    subprocess.run(["git","init","--bare",str(bare)],check=True,capture_output=True)
+    subprocess.run(["git","clone",str(bare),str(work)],check=True,capture_output=True)
+    subprocess.run(["git","-C",str(work),"config","user.email","test@example.invalid"],check=True)
+    subprocess.run(["git","-C",str(work),"config","user.name","AIOS Test"],check=True)
+    (work/"README.md").write_text("base\n",encoding="utf-8")
+    subprocess.run(["git","-C",str(work),"add","README.md"],check=True)
+    subprocess.run(["git","-C",str(work),"commit","-m","base"],check=True,capture_output=True)
+    subprocess.run(["git","-C",str(work),"push","origin","HEAD:refs/heads/main"],check=True,capture_output=True)
+    base=subprocess.check_output(["git","-C",str(work),"rev-parse","HEAD"],text=True).strip()
+    return bare,work,base
+
+
+def commit(work: Path, message: str) -> str:
+    subprocess.run(["git","-C",str(work),"add","-A"],check=True)
+    subprocess.run(["git","-C",str(work),"commit","-m",message],check=True,capture_output=True)
+    candidate=subprocess.check_output(["git","-C",str(work),"rev-parse","HEAD"],text=True).strip()
+    subprocess.run(["git","-C",str(work),"push","origin","HEAD:refs/heads/main"],check=True,capture_output=True)
+    return candidate
+
+
+def real_pem_header() -> bytes:
+    return b"-----" + b"BEGIN " + b"PRIVATE KEY" + b"-----"
+
+
+def bearer_value() -> bytes:
+    return b"Authorization" + b": Bearer " + (b"a"*40)
+
+
+def key_value_secret() -> bytes:
+    return b"api_key" + b"=" + (b"x"*24)
+
+
+def test_plain_private_key_words_are_not_secret_findings():
+    for value in (
+        b"BEGIN PRIVATE KEY",
+        b"PRIVATE KEY",
+        b"PEM private-key header",
+        b"assert 'BEGIN PRIVATE KEY' in message",
+    ):
+        assert packager.secret_finding(value) is None
+
+
+def test_complete_pem_private_key_header_is_rejected():
+    assert packager.secret_finding(real_pem_header())=="PEM_PRIVATE_KEY"
+    assert packager.secret_finding(
+        b"-----BEGIN OPENSSH " + b"PRIVATE KEY-----"
+    )=="PEM_PRIVATE_KEY"
+
+
+def test_bearer_and_key_value_secret_detection_remains_fail_closed():
+    assert packager.secret_finding(bearer_value())=="AUTHORIZATION_BEARER"
+    assert packager.secret_finding(key_value_secret())=="KEY_VALUE_SECRET"
+
+
+def test_short_or_descriptive_secret_words_do_not_false_positive():
+    assert packager.secret_finding(b"secret=short") is None
+    assert packager.secret_finding(b"password field is validated") is None
+    assert packager.secret_finding(b"api_key example placeholder") is None
+
+
+def test_end_to_end_bundle_accepts_literal_begin_private_key_text(tmp_path):
+    bare,work,base=init_repo(tmp_path)
+    (work/"tests").mkdir()
+    (work/"tests"/"test_text.py").write_text(
+        'ERROR = "BEGIN PRIVATE KEY is not a complete PEM header"\n',
+        encoding="utf-8",
+    )
+    (work/"docs").mkdir()
+    (work/"docs"/"note.md").write_text("Scanner regression.\n",encoding="utf-8")
+    candidate=commit(work,"candidate")
+
+    extra=tmp_path/"execution"
+    extra.mkdir()
+    (extra/"evidence.json").write_text('{"status":"PASS"}\n',encoding="utf-8")
+    out=tmp_path/"out"
+
+    summary=packager.build_bundle(
+        git_dir=bare,
+        base=base,
+        candidate=candidate,
+        task_id="r42-packager-test",
+        extra_dir=extra,
+        output_root=out,
+        now=datetime(2026,8,19,tzinfo=timezone.utc),
+    )
+
+    assert summary["changed_file_count"]==2
+    assert summary["execution_file_count"]==1
+    assert summary["repository_document_count"]==1
+    assert summary["execution_document_count"]==0
+    bundle_path=Path(summary["bundle_path"])
+    assert bundle_path.is_file()
+    assert packager.sha256(bundle_path.read_bytes())==summary["bundle_sha256"]
+
+    with zipfile.ZipFile(bundle_path) as archive:
+        names=set(archive.namelist())
+        assert "repo/after/tests/test_text.py" in names
+        assert "repo/after/docs/note.md" in names
+        assert "execution/evidence.json" in names
+        assert "documents-index.json" in names
+        assert "all-modified-files-manifest.json" in names
+        assert "checksums.sha256" in names
+        manifest=json.loads(archive.read("all-modified-files-manifest.json"))
+        assert manifest["schema"]=="aios.all-modified-files.bundle.v1"
+        assert manifest["candidate_oid"]==candidate
+
+    packager.verify_checksums(out/"bundle")
+
+
+def test_end_to_end_bundle_rejects_real_pem_header(tmp_path):
+    bare,work,base=init_repo(tmp_path)
+    (work/"credential.txt").write_bytes(
+        real_pem_header()+b"\nnot-a-real-key-body\n"
+    )
+    candidate=commit(work,"credential-like candidate")
+
+    with pytest.raises(
+        packager.PackagerError,
+        match="PEM_PRIVATE_KEY:after:credential.txt",
+    ):
+        packager.build_bundle(
+            git_dir=bare,
+            base=base,
+            candidate=candidate,
+            task_id="r42-packager-pem-reject",
+            output_root=tmp_path/"out",
+        )
+
+
+def test_execution_evidence_bearer_is_rejected(tmp_path):
+    bare,work,base=init_repo(tmp_path)
+    (work/"normal.txt").write_text("normal\n",encoding="utf-8")
+    candidate=commit(work,"normal")
+    extra=tmp_path/"execution"
+    extra.mkdir()
+    (extra/"unsafe.txt").write_bytes(bearer_value()+b"\n")
+
+    with pytest.raises(
+        packager.PackagerError,
+        match="AUTHORIZATION_BEARER:execution:unsafe.txt",
+    ):
+        packager.build_bundle(
+            git_dir=bare,
+            base=base,
+            candidate=candidate,
+            task_id="r42-packager-bearer-reject",
+            extra_dir=extra,
+            output_root=tmp_path/"out",
+        )
+
+
+def test_execution_symlink_is_rejected(tmp_path):
+    bare,work,base=init_repo(tmp_path)
+    (work/"normal.txt").write_text("normal\n",encoding="utf-8")
+    candidate=commit(work,"normal")
+    extra=tmp_path/"execution"
+    extra.mkdir()
+    target=extra/"target.txt"
+    target.write_text("safe\n",encoding="utf-8")
+    (extra/"link.txt").symlink_to(target)
+
+    with pytest.raises(
+        packager.PackagerError,
+        match="EXTRA_FILE_SYMLINK_REJECTED",
+    ):
+        packager.build_bundle(
+            git_dir=bare,
+            base=base,
+            candidate=candidate,
+            task_id="r42-packager-symlink-reject",
+            extra_dir=extra,
+            output_root=tmp_path/"out",
+        )
+
+
+def test_non_ancestor_lineage_is_rejected(tmp_path):
+    bare,work,base=init_repo(tmp_path)
+    (work/"one.txt").write_text("one\n",encoding="utf-8")
+    first=commit(work,"first")
+
+    subprocess.run(["git","-C",str(work),"checkout","--detach",base],check=True,capture_output=True)
+    (work/"other.txt").write_text("other\n",encoding="utf-8")
+    subprocess.run(["git","-C",str(work),"add","other.txt"],check=True)
+    subprocess.run(["git","-C",str(work),"commit","-m","other"],check=True,capture_output=True)
+    other=subprocess.check_output(["git","-C",str(work),"rev-parse","HEAD"],text=True).strip()
+    subprocess.run(
+        ["git","-C",str(work),"push","origin","HEAD:refs/heads/other"],
+        check=True,
+        capture_output=True,
+    )
+
+    with pytest.raises(packager.PackagerError,match="LINEAGE_INVALID"):
+        packager.validate_lineage(bare,first,other)
+
+
+def test_bundle_checksum_closure_detects_tamper(tmp_path):
+    bare,work,base=init_repo(tmp_path)
+    (work/"normal.txt").write_text("normal\n",encoding="utf-8")
+    candidate=commit(work,"normal")
+    out=tmp_path/"out"
+    packager.build_bundle(
+        git_dir=bare,
+        base=base,
+        candidate=candidate,
+        task_id="r42-packager-checksum",
+        output_root=out,
+    )
+    (out/"bundle"/"repo"/"after"/"normal.txt").write_text(
+        "tampered\n",
+        encoding="utf-8",
+    )
+    with pytest.raises(
+        packager.PackagerError,
+        match="CHECKSUM_CLOSURE_INVALID",
+    ):
+        packager.verify_checksums(out/"bundle")
+
+
+def test_source_does_not_contain_complete_pem_header_literal():
+    source=(ROOT/"active-runtime/app/aios_all_modified_files_packager.py").read_bytes()
+    tests=(ROOT/"tests/test_all_modified_files_packager.py").read_bytes()
+    complete=real_pem_header()
+    assert complete not in source
+    assert complete not in tests
+
+
+def test_document_path_contract():
+    assert packager.is_document_path("docs/spec.md")
+    assert packager.is_document_path("README")
+    assert packager.is_document_path("CHANGELOG.md")
+    assert not packager.is_document_path("active-runtime/app/main.py")

tests/test_autonomous_control_plane.py

@@ -0,0 +1,157 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT=Path(__file__).resolve().parents[1]
+APP=ROOT/"active-runtime/app"
+sys.path.insert(0,str(APP))
+
+import aios_autonomous_control_plane as acp
+import aios_capability_contract as contract
+import aios_release_online_agent_worker as worker
+
+
+def _ready_probes():
+    return {
+        "credential_projection": lambda: True,
+        "trusted_builder": lambda: True,
+        "authenticated_probe": lambda: True,
+        "release_api_health": lambda: True,
+        "worker_liveness": lambda: True,
+    }
+
+
+def _task():
+    return {
+        "task_id":"r42-autonomous-release-admission-test",
+        "required_capabilities":["release:candidate"],
+        "constraints":{
+            "release_candidate":{
+                "candidate_id":"r42-release-candidate-v2r2c9-test",
+                "source_commit":"1"*40,
+                "version":"V2R2C9",
+                "projection_id":"projection-"+"a"*32,
+            }
+        },
+    }
+
+
+def test_capability_contract_registry_is_closed_and_release_is_versioned():
+    status=contract.validate_capability_contract_registry()
+    assert status["valid"] is True
+    release=contract.capability_contract("release:candidate")
+    assert release is not None
+    assert release.min_version==1
+    assert release.max_version==1
+    assert release.permissions==("prepare_release",)
+    assert "credential_projection" in release.dependencies
+    assert "trusted_builder" in release.dependencies
+    assert "authenticated_probe" in release.health_requirements
+    assert "production_deploy" in release.forbidden
+    assert "runtime_switch" in release.forbidden
+
+
+def test_unknown_agent_capability_cannot_register_or_match():
+    assert contract.registered_capability("release:future",worker=True) is False
+    assert contract.registered_capability("release:*",worker=True) is False
+    assert contract.has_capabilities(["release:future"],["release:future"]) is False
+
+
+def test_runtime_admission_requires_all_release_execution_readiness():
+    decision=acp.runtime_admission(
+        worker_capabilities=["release:candidate"],
+        required_capabilities=["release:candidate"],
+        requested_operations=["prepare_release"],
+        probes=_ready_probes(),
+    )
+    assert decision.admitted is True
+    assert decision.state=="EXECUTION_READY"
+
+
+def test_runtime_admission_blocks_credential_failure_before_execution():
+    probes=_ready_probes()
+    probes["credential_projection"]=lambda:False
+    probes["authenticated_probe"]=lambda:False
+    decision=acp.runtime_admission(
+        worker_capabilities=["release:candidate"],
+        required_capabilities=["release:candidate"],
+        requested_operations=["prepare_release"],
+        probes=probes,
+    )
+    assert decision.admitted is False
+    assert decision.state=="EXECUTION_READY_FAILED"
+    assert decision.failure_class==acp.FailureClass.CREDENTIAL_PROJECTION_ERROR
+    assert decision.repair_level==acp.RepairLevel.AUTO_REPAIR_TASK
+
+
+def test_forbidden_release_operation_requires_owner_boundary():
+    decision=acp.runtime_admission(
+        worker_capabilities=["release:candidate"],
+        required_capabilities=["release:candidate"],
+        requested_operations=["production_deploy"],
+        probes=_ready_probes(),
+    )
+    assert decision.admitted is False
+    assert decision.failure_class==acp.FailureClass.POLICY_DENIED
+    assert decision.repair_level==acp.RepairLevel.OWNER_APPROVAL
+
+
+@pytest.mark.parametrize(
+    "error,phase,expected,level,consumes",
+    [
+        ("Permission denied: gitops.tokens","release",acp.FailureClass.CREDENTIAL_PROJECTION_ERROR,acp.RepairLevel.AUTO_REPAIR_TASK,False),
+        ("HTTP_TRANSPORT_UNAVAILABLE","network",acp.FailureClass.DEPENDENCY_UNAVAILABLE,acp.RepairLevel.AUTO_LOCAL,False),
+        ("TASK_CAPABILITY_SCOPE_INVALID","validation",acp.FailureClass.CAPABILITY_CONTRACT_ERROR,acp.RepairLevel.AUTO_REPAIR_TASK,False),
+        ("lease lost","heartbeat",acp.FailureClass.LEASE_LOST,acp.RepairLevel.AUTO_LOCAL,False),
+        ("business permanent invalid release","release",acp.FailureClass.BUSINESS_PERMANENT,acp.RepairLevel.MANUAL,True),
+    ],
+)
+def test_failure_classification_drives_repair_policy(error,phase,expected,level,consumes):
+    failure=acp.classify_failure(error,phase=phase)
+    assert failure==expected
+    repair=acp.repair_for_failure(failure)
+    assert repair.level==level
+    assert repair.consumes_business_attempt is consumes
+
+
+def test_release_worker_does_not_claim_when_admission_is_not_ready(monkeypatch):
+    calls=[]
+    def fake_http(base,method,route,token,body=None,timeout=30):
+        calls.append((base,method,route,body))
+        if route=="/health":
+            return 200,{"status":"ok"}
+        if route=="/createReleaseCandidate" and body=={}:
+            return 500,{"error":"Permission denied: gitops.tokens"}
+        if route.endswith("/claim"):
+            raise AssertionError("claim must not happen before EXECUTION_READY")
+        raise AssertionError(route)
+
+    events=[]
+    monkeypatch.setattr(worker,"http_json",fake_http)
+    monkeypatch.setattr(worker,"audit",lambda event,**fields:events.append((event,fields)))
+    monkeypatch.setattr(worker,"write_liveness",lambda *args,**kwargs:None)
+
+    assert worker.execute_task(
+        _task(),
+        agent_token="a"*32,
+        release_token="r"*32,
+    ) is False
+    assert not any(route.endswith("/claim") for _,_,route,_ in calls)
+    blocked=[item for item in events if item[0]=="TASK_ADMISSION_BLOCKED"]
+    assert len(blocked)==1
+    assert blocked[0][1]["admission"]["state"]=="EXECUTION_READY_FAILED"
+
+
+def test_system_invariant_snapshot_is_degraded_when_dependency_is_missing():
+    probes=_ready_probes()
+    probes["trusted_builder"]=lambda:False
+    snapshot=acp.system_invariant_snapshot(
+        worker_capabilities=["release:candidate"],
+        required_capabilities=["release:candidate"],
+        probes=probes,
+    )
+    assert snapshot["status"]=="DEGRADED"
+    assert snapshot["execution_ready"] is False

tests/test_durable_owner_authorization.py

@@ -0,0 +1,627 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+import json
+from pathlib import Path
+import sqlite3
+import sys
+
+import pytest
+
+ROOT=Path(__file__).resolve().parents[1]
+APP=ROOT/"active-runtime/app"
+sys.path.insert(0,str(APP))
+
+import aios_durable_owner_authorization as auth
+
+
+NOW=datetime(2026,8,19,6,0,0,tzinfo=timezone.utc)
+ACCOUNT="chatgpt-governed-implementer-singlehost"
+
+
+def write_root_file(path: Path, value: object) -> Path:
+    path.parent.mkdir(parents=True,exist_ok=True)
+    path.write_text(json.dumps(value,sort_keys=True)+"\n",encoding="utf-8")
+    path.chmod(0o600)
+    return path
+
+
+def policy_value(**overrides):
+    value={
+        "schema":auth.POLICY_SCHEMA,
+        "grant_id":"owner-permanent-preparation-20260819",
+        "owner_identity":"owner-singlehost",
+        "account":ACCOUNT,
+        "capabilities":[
+            "governed-task:project",
+            "release:candidate",
+        ],
+        "issued_at":"2026-08-19T05:00:00Z",
+        "not_before":"2026-08-19T05:00:00Z",
+        "persistent_until_revoked":True,
+        "constraints":{
+            "repository":"aios/aios",
+            "allowed_target_prefix":"agent/",
+            "release_version_prefix":"V2R2C",
+            "max_task_risk":"engineering",
+        },
+    }
+    value.update(overrides)
+    return value
+
+
+def load_policy(tmp_path,monkeypatch,**overrides):
+    path=write_root_file(tmp_path/"policy.json",policy_value(**overrides))
+    revocations=write_root_file(
+        tmp_path/"revocations.json",
+        {"revoked_grant_ids":[]},
+    )
+    monkeypatch.setattr(auth,"validate_root_file",lambda path,**kwargs:path)
+    return auth.DurablePolicy.load(
+        path,
+        revocations,
+        now=NOW,
+    ),path,revocations
+
+
+def governed_request(**overrides):
+    value={
+        "task_id":"r42-owner-auth-projection-test-20260819",
+        "repository":"aios/aios",
+        "base_oid":"8"*40,
+        "expected_branch_oid":"8"*40,
+        "target_branch":"agent/r42-owner-auth-projection-test-20260819",
+        "generation":1,
+        "risk":"engineering",
+        "description":"project existing governed task",
+        "idempotency_key":"binding-owner-auth-projection-v1",
+        "branch_mode":"CREATE_NEW",
+        "priority":100,
+        "max_attempts":3,
+        "require_online_task_projection":True,
+        "online_task":{
+            "goal":"Analyze source read-only.",
+            "instructions":"Use source capture first.",
+            "success_criteria":[
+                {"id":"source","description":"Return source snapshot."}
+            ],
+            "constraints":{"repository_mutation":"forbidden"},
+            "required_capabilities":["git.read","git.test"],
+            "priority":100,
+            "max_attempts":3,
+        },
+        "execution":{
+            "action":"source_capture",
+            "adapter":"source_capture",
+            "capability_scope":["git.read","git.test"],
+            "source_signals":["durable authorization projection"],
+        },
+    }
+    value.update(overrides)
+    return value
+
+
+def release_request(**overrides):
+    value={
+        "candidate_id":"r42-pf1-release-candidate-20260819",
+        "source_commit":"8"*40,
+        "version":"V2R2C2",
+    }
+    value.update(overrides)
+    return value
+
+
+class FakeUpstream:
+    def __init__(self):
+        self.project_calls=[]
+
+    def project_governed_task(self,body):
+        self.project_calls.append(dict(body))
+        return 200,{
+            "task_id":body["task_id"],
+            "job_id":"job-"+"a"*32,
+            "state":"BOUND",
+        }
+
+
+def broker(tmp_path,monkeypatch):
+    policy,path,revocations=load_policy(tmp_path,monkeypatch)
+    store=auth.ReceiptStore(tmp_path/"receipts.sqlite3")
+    upstream=FakeUpstream()
+    instance=auth.Broker(
+        policy_path=path,
+        revocation_path=revocations,
+        store=store,
+        upstream=upstream,
+        expected_account=ACCOUNT,
+    )
+    monkeypatch.setattr(
+        auth.DurablePolicy,
+        "load",
+        classmethod(lambda cls,*args,**kwargs:policy),
+    )
+    return instance,upstream,store,policy
+
+
+def test_allowed_persistent_capabilities_are_only_preparation_scopes():
+    assert auth.ALLOWED_PERSISTENT_CAPABILITIES=={
+        "governed-task:project",
+        "release:candidate",
+    }
+    assert auth.ALLOWED_PERSISTENT_CAPABILITIES.isdisjoint(
+        auth.FORBIDDEN_PERSISTENT_CAPABILITIES
+    )
+
+
+@pytest.mark.parametrize(
+    "capability",
+    sorted(auth.FORBIDDEN_PERSISTENT_CAPABILITIES),
+)
+def test_policy_rejects_high_risk_persistent_capability(
+    tmp_path,monkeypatch,capability
+):
+    bad=policy_value(
+        capabilities=["governed-task:project",capability]
+    )
+    path=write_root_file(tmp_path/"policy.json",bad)
+    revocations=write_root_file(
+        tmp_path/"revocations.json",
+        {"revoked_grant_ids":[]},
+    )
+    monkeypatch.setattr(auth,"validate_root_file",lambda path,**kwargs:path)
+    with pytest.raises(
+        auth.AuthorizationError,
+        match="POLICY_CAPABILITY_NOT_PERSISTABLE",
+    ):
+        auth.DurablePolicy.load(path,revocations,now=NOW)
+
+
+def test_policy_is_persistent_until_revoked(tmp_path,monkeypatch):
+    policy,_,_=load_policy(tmp_path,monkeypatch)
+    assert policy.persistent_until_revoked is True
+    assert policy.account==ACCOUNT
+    assert policy.capabilities=={
+        "governed-task:project",
+        "release:candidate",
+    }
+
+
+def test_revocation_fails_closed(tmp_path,monkeypatch):
+    path=write_root_file(tmp_path/"policy.json",policy_value())
+    revocations=write_root_file(
+        tmp_path/"revocations.json",
+        {"revoked_grant_ids":["owner-permanent-preparation-20260819"]},
+    )
+    monkeypatch.setattr(auth,"validate_root_file",lambda path,**kwargs:path)
+    with pytest.raises(auth.AuthorizationError,match="OWNER_GRANT_REVOKED"):
+        auth.DurablePolicy.load(path,revocations,now=NOW)
+
+
+def test_future_policy_is_not_active(tmp_path,monkeypatch):
+    with pytest.raises(auth.AuthorizationError,match="POLICY_NOT_ACTIVE"):
+        load_policy(
+            tmp_path,
+            monkeypatch,
+            issued_at="2026-08-20T05:00:00Z",
+            not_before="2026-08-20T05:00:00Z",
+        )
+
+
+def test_governed_projection_request_is_tightly_constrained(
+    tmp_path,monkeypatch
+):
+    policy,_,_=load_policy(tmp_path,monkeypatch)
+    decision=policy.authorize(
+        "governed-task:project",
+        governed_request(),
+    )
+    assert decision["decision"]=="ALLOW"
+    assert decision["persistent_until_revoked"] is True
+
+    bad_caps=governed_request()
+    bad_caps["online_task"]={
+        **bad_caps["online_task"],
+        "required_capabilities":["release:candidate"],
+    }
+    bad_action=governed_request()
+    bad_action["execution"]={
+        **bad_action["execution"],
+        "action":"deterministic",
+        "adapter":"deterministic",
+    }
+    bad_scope=governed_request()
+    bad_scope["execution"]={
+        **bad_scope["execution"],
+        "capability_scope":["git.read"],
+    }
+    for bad in (
+        governed_request(repository="other/repo"),
+        governed_request(target_branch="main"),
+        governed_request(generation=2),
+        governed_request(risk="high"),
+        governed_request(expected_branch_oid="9"*40),
+        governed_request(branch_mode="CONTINUE_EXISTING"),
+        governed_request(priority=101),
+        governed_request(max_attempts=4),
+        governed_request(require_online_task_projection=False),
+        bad_caps,
+        bad_action,
+        bad_scope,
+    ):
+        with pytest.raises(auth.AuthorizationError):
+            policy.authorize("governed-task:project",bad)
+
+
+
+def test_write_capable_projection_is_allowed_only_as_source_capture_placeholder(
+    tmp_path,monkeypatch
+):
+    policy,_,_=load_policy(tmp_path,monkeypatch)
+    request=governed_request()
+    request["online_task"]={
+        **request["online_task"],
+        "required_capabilities":[
+            "git.read",
+            "git.test",
+            "governed.write.branch",
+            "governed.write.file",
+        ],
+    }
+    request["execution"]={
+        **request["execution"],
+        "capability_scope":[
+            "git.read",
+            "git.test",
+            "governed.write.branch",
+            "governed.write.file",
+        ],
+    }
+    decision=policy.authorize("governed-task:project",request)
+    assert decision["decision"]=="ALLOW"
+    assert request["execution"]["action"]=="source_capture"
+
+def test_release_candidate_request_is_tightly_constrained(
+    tmp_path,monkeypatch
+):
+    policy,_,_=load_policy(tmp_path,monkeypatch)
+    decision=policy.authorize("release:candidate",release_request())
+    assert decision["decision"]=="ALLOW"
+
+    with pytest.raises(
+        auth.AuthorizationError,
+        match="RELEASE_VERSION_NOT_AUTHORIZED",
+    ):
+        policy.authorize(
+            "release:candidate",
+            release_request(version="V3R1C1"),
+        )
+    with pytest.raises(
+        auth.AuthorizationError,
+        match="RELEASE_CANDIDATE_FIELDS_INVALID",
+    ):
+        policy.authorize(
+            "release:candidate",
+            {**release_request(),"execute":True},
+        )
+
+
+def test_governed_task_projection_executes_canonical_upstream(
+    tmp_path,monkeypatch
+):
+    instance,upstream,store,_=broker(tmp_path,monkeypatch)
+    request=governed_request()
+    response=instance.execute(
+        "governed-task:project",
+        request,
+        idempotency_key="owner-auth-projection-operation-0001",
+    )
+    assert len(upstream.project_calls)==1
+    assert upstream.project_calls[0]==request
+    assert response["operation_state"]=="EXECUTED"
+    assert response["external_capability_projection_required"] is False
+    assert response["separate_owner_release_approval_still_required"] is True
+
+
+def test_release_candidate_authorization_does_not_invent_builder_upstream(
+    tmp_path,monkeypatch
+):
+    instance,upstream,store,_=broker(tmp_path,monkeypatch)
+    response=instance.execute(
+        "release:candidate",
+        release_request(),
+        idempotency_key="owner-auth-release-candidate-operation-0001",
+    )
+    assert upstream.project_calls==[]
+    assert response["operation_state"]==(
+        "AUTHORIZED_PENDING_EXTERNAL_CAPABILITY_PROJECTION"
+    )
+    assert response["upstream_status"] is None
+    assert response["upstream_response"] is None
+    assert response["external_capability_projection_required"] is True
+    assert response["projection_id"].startswith("projection-")
+    assert response["projection_state"]=="PENDING_EXTERNAL_CONNECTOR"
+    assert response["separate_owner_release_approval_still_required"] is True
+
+
+
+def test_release_authorization_creates_durable_projection_outbox(
+    tmp_path,monkeypatch
+):
+    instance,_,store,policy=broker(tmp_path,monkeypatch)
+    request=release_request()
+    response=instance.execute(
+        "release:candidate",
+        request,
+        idempotency_key="owner-auth-release-outbox-0001",
+    )
+    projection=store.get_projection(response["projection_id"])
+    assert projection is not None
+    assert projection["schema"]==auth.PROJECTION_SCHEMA
+    assert projection["state"]=="PENDING_EXTERNAL_CONNECTOR"
+    assert projection["grant_id"]==policy.grant_id
+    assert projection["account"]==ACCOUNT
+    assert projection["capability"]=="release:candidate"
+    assert projection["request"]==request
+    assert projection["separate_owner_release_approval_still_required"] is True
+
+    listed=instance.projections()
+    assert listed["schema"]==auth.PROJECTION_LIST_SCHEMA
+    assert listed["count"]==1
+    assert listed["items"][0]["projection_id"]==response["projection_id"]
+    assert listed["external_connector_ack_supported"] is False
+
+
+
+def test_existing_pre_outbox_release_receipt_is_backfilled_on_exact_retry(
+    tmp_path,monkeypatch
+):
+    instance,_,store,policy=broker(tmp_path,monkeypatch)
+    request=release_request()
+    decision=policy.authorize("release:candidate",request)
+    key="owner-auth-release-pre-outbox-0001"
+    legacy_response={
+        "schema":auth.RECEIPT_SCHEMA,
+        "grant_id":policy.grant_id,
+        "owner_identity":policy.owner_identity,
+        "account":policy.account,
+        "capability":"release:candidate",
+        "policy_sha256":policy.policy_sha256,
+        "request_sha256":decision["request_sha256"],
+        "idempotency_key":key,
+        "operation_state":"AUTHORIZED_PENDING_EXTERNAL_CAPABILITY_PROJECTION",
+        "upstream_status":None,
+        "upstream_response":None,
+        "external_capability_projection_required":True,
+        "separate_owner_release_approval_still_required":True,
+    }
+    store.put(
+        key,
+        "release:candidate",
+        decision["request_sha256"],
+        legacy_response,
+    )
+    response=instance.execute(
+        "release:candidate",
+        request,
+        idempotency_key=key,
+    )
+    assert response["projection_id"].startswith("projection-")
+    assert response["projection_state"]=="PENDING_EXTERNAL_CONNECTOR"
+    assert store.get_projection(response["projection_id"]) is not None
+    persisted=store.get(key)["response"]
+    assert persisted["projection_id"]==response["projection_id"]
+
+def test_projection_id_is_deterministic_and_exact_retry_does_not_duplicate(
+    tmp_path,monkeypatch
+):
+    instance,_,store,_=broker(tmp_path,monkeypatch)
+    key="owner-auth-release-outbox-0002"
+    first=instance.execute("release:candidate",release_request(),idempotency_key=key)
+    second=instance.execute("release:candidate",release_request(),idempotency_key=key)
+    assert second==first
+    con=sqlite3.connect(store.path)
+    rows=con.execute("select projection_id from projection_outbox").fetchall()
+    assert rows==[(first["projection_id"],)]
+
+
+def test_projection_enqueue_recovers_crash_between_outbox_and_receipt(
+    tmp_path,monkeypatch
+):
+    instance,_,store,policy=broker(tmp_path,monkeypatch)
+    request=release_request()
+    decision=policy.authorize("release:candidate",request)
+    key="owner-auth-release-outbox-crash-0001"
+    first=store.enqueue_projection(
+        idempotency_key=key,
+        grant_id=policy.grant_id,
+        account=policy.account,
+        capability="release:candidate",
+        policy_sha256=policy.policy_sha256,
+        request_sha256=decision["request_sha256"],
+        request=request,
+    )
+    # Simulate a later retry after the process crashed before receipt creation.
+    second=store.enqueue_projection(
+        idempotency_key=key,
+        grant_id=policy.grant_id,
+        account=policy.account,
+        capability="release:candidate",
+        policy_sha256=policy.policy_sha256,
+        request_sha256=decision["request_sha256"],
+        request=request,
+    )
+    assert second==first
+
+
+def test_projection_idempotency_conflict_is_fail_closed(
+    tmp_path,monkeypatch
+):
+    instance,_,store,policy=broker(tmp_path,monkeypatch)
+    first=release_request()
+    first_decision=policy.authorize("release:candidate",first)
+    key="owner-auth-release-outbox-conflict-0001"
+    store.enqueue_projection(
+        idempotency_key=key,
+        grant_id=policy.grant_id,
+        account=policy.account,
+        capability="release:candidate",
+        policy_sha256=policy.policy_sha256,
+        request_sha256=first_decision["request_sha256"],
+        request=first,
+    )
+    second=release_request(version="V2R2C3")
+    second_decision=policy.authorize("release:candidate",second)
+    with pytest.raises(auth.AuthorizationError,match="PROJECTION_IDEMPOTENCY_CONFLICT"):
+        store.enqueue_projection(
+            idempotency_key=key,
+            grant_id=policy.grant_id,
+            account=policy.account,
+            capability="release:candidate",
+            policy_sha256=policy.policy_sha256,
+            request_sha256=second_decision["request_sha256"],
+            request=second,
+        )
+
+
+def test_revocation_blocks_projection_enumeration(tmp_path,monkeypatch):
+    instance,_,store,policy=broker(tmp_path,monkeypatch)
+    response=instance.execute(
+        "release:candidate",
+        release_request(),
+        idempotency_key="owner-auth-release-outbox-revoke-0001",
+    )
+    # Restore real policy loading and write a revocation for this grant.
+    monkeypatch.undo()
+    revocations=instance.revocation_path
+    revocations.write_text(
+        json.dumps({"revoked_grant_ids":[policy.grant_id]})+"\n",
+        encoding="utf-8",
+    )
+    revocations.chmod(0o600)
+    monkeypatch.setattr(auth,"validate_root_file",lambda path,**kwargs:path)
+    with pytest.raises(auth.AuthorizationError,match="OWNER_GRANT_REVOKED"):
+        instance.projections()
+    # Evidence is retained; revocation does not erase history.
+    assert store.get_projection(response["projection_id"]) is not None
+
+
+def test_projection_outbox_contains_no_credentials(tmp_path,monkeypatch):
+    instance,_,store,_=broker(tmp_path,monkeypatch)
+    instance.execute(
+        "release:candidate",
+        release_request(),
+        idempotency_key="owner-auth-release-outbox-schema-0001",
+    )
+    con=sqlite3.connect(store.path)
+    columns=[row[1] for row in con.execute("pragma table_info(projection_outbox)")]
+    joined=" ".join(columns).lower()
+    for forbidden in ("token","credential","private_key","secret"):
+        assert forbidden not in joined
+    payload=con.execute("select payload_json from projection_outbox").fetchone()[0]
+    assert "Bearer " not in payload
+    assert "submitOwnerApproval" not in payload
+    assert "startDeployment" not in payload
+
+def test_exact_retry_returns_same_receipt_without_second_projection(
+    tmp_path,monkeypatch
+):
+    instance,upstream,_,_=broker(tmp_path,monkeypatch)
+    request=governed_request()
+    first=instance.execute(
+        "governed-task:project",
+        request,
+        idempotency_key="owner-auth-projection-operation-0002",
+    )
+    second=instance.execute(
+        "governed-task:project",
+        request,
+        idempotency_key="owner-auth-projection-operation-0002",
+    )
+    assert second==first
+    assert len(upstream.project_calls)==1
+
+
+def test_idempotency_conflict_fails_closed(tmp_path,monkeypatch):
+    instance,_,_,_=broker(tmp_path,monkeypatch)
+    instance.execute(
+        "release:candidate",
+        release_request(),
+        idempotency_key="owner-auth-release-operation-0002",
+    )
+    with pytest.raises(auth.AuthorizationError,match="IDEMPOTENCY_CONFLICT"):
+        instance.execute(
+            "release:candidate",
+            release_request(version="V2R2C3"),
+            idempotency_key="owner-auth-release-operation-0002",
+        )
+
+
+def test_receipt_store_contains_no_owner_policy_or_credentials(
+    tmp_path,monkeypatch
+):
+    instance,_,store,_=broker(tmp_path,monkeypatch)
+    instance.execute(
+        "release:candidate",
+        release_request(),
+        idempotency_key="owner-auth-release-operation-0003",
+    )
+    con=sqlite3.connect(store.path)
+    columns=[row[1] for row in con.execute("pragma table_info(receipts)")]
+    joined=" ".join(columns).lower()
+    assert "token" not in joined
+    assert "credential" not in joined
+    assert "policy_json" not in columns
+
+
+def test_systemd_broker_uses_unix_socket_and_only_host_control_credential():
+    unit=(
+        ROOT/"deploy/systemd/aios-owner-durable-authorization-broker.service"
+    ).read_text()
+    assert "LoadCredential=agent_task_control_token:" in unit
+    assert "release_candidate_token" not in unit
+    assert "18947" not in unit
+    assert "AIOS_RELEASE_CANDIDATE_API_URL" not in unit
+    assert "AIOS_OWNER_AUTH_BROKER_SOCKET=%t/aios-owner-authorization-broker/broker.sock" in unit
+    assert "Group=aios-executor-exchange" in unit
+    assert "RestrictAddressFamilies=AF_UNIX AF_INET" in unit
+    assert "IPAddressDeny=any" in unit
+    assert "IPAddressAllow=127.0.0.0/8" in unit
+    assert "InaccessiblePaths=/data/aios-gitops /home" in unit
+
+
+def test_source_contains_no_release_builder_endpoint_or_high_risk_execution():
+    source=(
+        ROOT/"active-runtime/app/aios_durable_owner_authorization.py"
+    ).read_text()
+    assert "/createReleaseCandidate" not in source
+    assert "18947" not in source
+    assert "ThreadingHTTPServer" not in source
+    assert "ThreadingUnixStreamServer" in source
+    assert "startDeployment" in source
+    assert "tag_publish" in source
+    assert "baseline_registry_finalize" in source
+    assert "runtime_switch" in source
+    assert "AUTHORIZED_PENDING_EXTERNAL_CAPABILITY_PROJECTION" in source
+    assert "PENDING_EXTERNAL_CONNECTOR" in source
+    assert "projection_outbox" in source
+    assert "external_connector_ack_supported" in source
+    assert "SAFE_PROJECTED_TASK_CAPABILITIES" in source
+    assert "INITIAL_EXECUTION_ACTION_NOT_AUTHORIZED" in source
+
+
+
+def test_docs_avoid_new_broker_bearer_secret():
+    doc=(
+        ROOT/"docs/superpowers/specs/2026-08-19-durable-owner-authorization.md"
+    ).read_text()
+    assert "AF_UNIX" in doc
+    assert "avoids introducing another long-lived bearer secret" in doc
+
+def test_docs_preserve_separate_owner_release_approval():
+    doc=(
+        ROOT/"docs/superpowers/specs/2026-08-19-durable-owner-authorization.md"
+    ).read_text()
+    assert "persistent until explicitly revoked" in doc.lower()
+    assert "Owner signature" in doc
+    assert "AUTHORIZED_PENDING_EXTERNAL_CAPABILITY_PROJECTION" in doc
+    assert "PENDING_EXTERNAL_CONNECTOR" in doc

tests/test_gpt_https_edge.py

@@ -0,0 +1,205 @@
+from __future__ import annotations
+
+from email.message import Message
+import json
+from pathlib import Path
+import stat
+import sys
+
+import pytest
+
+ROOT=Path(__file__).resolve().parents[1]
+APP=ROOT/"active-runtime/app"
+sys.path.insert(0,str(APP))
+
+import aios_gpt_https_edge as edge
+
+
+class Headers(dict):
+    def get(self,key,default=None):
+        return super().get(key,default)
+
+
+def test_default_and_public_bind_require_explicit_owner_approval():
+    assert edge.public_bind_allowed("127.0.0.1","0")=="127.0.0.1"
+    assert edge.public_bind_allowed("::1","0")=="::1"
+    with pytest.raises(RuntimeError,match="OWNER_APPROVAL_REQUIRED"):
+        edge.public_bind_allowed("0.0.0.0","0")
+    with pytest.raises(RuntimeError,match="OWNER_APPROVAL_REQUIRED"):
+        edge.public_bind_allowed("::","")
+    assert edge.public_bind_allowed("0.0.0.0","1")=="0.0.0.0"
+
+
+def test_arbitrary_public_bind_address_is_denied_even_with_approval():
+    with pytest.raises(RuntimeError,match="BIND_ADDRESS_INVALID"):
+        edge.public_bind_allowed("192.0.2.10","1")
+
+
+def test_public_domain_validation():
+    assert edge.validated_public_domain("tasks.example.com")=="tasks.example.com"
+    assert edge.validated_public_domain("")==""
+    with pytest.raises(RuntimeError,match="PUBLIC_DOMAIN_INVALID"):
+        edge.validated_public_domain("https://example.com")
+    with pytest.raises(RuntimeError,match="PUBLIC_DOMAIN_INVALID"):
+        edge.validated_public_domain("localhost")
+
+
+def test_route_allowlist_accepts_only_readonly_api():
+    assert edge.allowed_route("POST","/v1/read-only-tasks")=="/v1/read-only-tasks"
+    assert edge.allowed_route("GET","/v1/read-only-tasks/external-request-0001")==(
+        "/v1/read-only-tasks/external-request-0001"
+    )
+    assert edge.allowed_route("GET","/health")=="/health"
+    assert edge.allowed_route("GET","/openapi.json")=="/openapi.json"
+    for method,path in [
+        ("POST","/v2/governed-task"),
+        ("GET","/v1/tasks/x"),
+        ("POST","/v1/read-only-tasks/x"),
+        ("DELETE","/v1/read-only-tasks/x"),
+    ]:
+        with pytest.raises(edge.EdgeError,match="ROUTE_DENY"):
+            edge.allowed_route(method,path)
+
+
+def test_query_and_fragment_are_denied():
+    with pytest.raises(edge.EdgeError,match="QUERY_OR_FRAGMENT_DENY"):
+        edge.allowed_route("GET","/health?x=1")
+    with pytest.raises(edge.EdgeError,match="QUERY_OR_FRAGMENT_DENY"):
+        edge.allowed_route("GET","/openapi.json#x")
+
+
+def test_forwarded_headers_are_minimal_and_require_bearer():
+    incoming=Headers({
+        "Authorization":"Bearer "+"a"*40,
+        "Content-Type":"application/json; charset=utf-8",
+        "Cookie":"secret=1",
+        "X-Forwarded-For":"203.0.113.10",
+        "X-Api-Key":"do-not-forward",
+    })
+    forwarded=edge.forwarded_headers(incoming,body=True)
+    assert forwarded=={
+        "Accept":"application/json",
+        "Authorization":"Bearer "+"a"*40,
+        "Content-Type":"application/json",
+    }
+    with pytest.raises(edge.EdgeError,match="AUTHORIZATION_REQUIRED"):
+        edge.forwarded_headers(Headers({}),body=False)
+
+
+def test_invalid_content_type_is_denied():
+    with pytest.raises(edge.EdgeError,match="CONTENT_TYPE_INVALID"):
+        edge.forwarded_headers(
+            Headers({
+                "Authorization":"Bearer "+"a"*40,
+                "Content-Type":"text/plain",
+            }),
+            body=True,
+        )
+
+
+def test_rate_limiter_caps_window_and_allows_expiry():
+    now=[100.0]
+    limiter=edge.RateLimiter(window_seconds=60,limit=2,burst=1,now_fn=lambda:now[0])
+    assert limiter.allowed("client")
+    assert limiter.allowed("client")
+    assert limiter.allowed("client")
+    assert not limiter.allowed("client")
+    now[0]=161.0
+    assert limiter.allowed("client")
+
+
+
+def test_rate_limiter_client_table_is_bounded():
+    now=[100.0]
+    limiter=edge.RateLimiter(
+        window_seconds=60,
+        limit=10,
+        burst=0,
+        max_clients=2,
+        now_fn=lambda:now[0],
+    )
+    assert limiter.allowed("a")
+    assert limiter.allowed("b")
+    assert not limiter.allowed("c")
+    now[0]=161.0
+    assert limiter.allowed("c")
+    assert len(limiter._events) <= 2
+
+
+def test_audit_source_does_not_log_sensitive_request_material():
+    source=(ROOT/"active-runtime/app/aios_gpt_https_edge.py").read_text()
+    assert '"authorization", "body", "token", "cookie", "client_ip"' in source
+    assert "_client_pseudonym" in source
+
+def test_proxy_upstream_is_fixed():
+    proxy=edge.Proxy()
+    assert proxy.upstream=="http://127.0.0.1:18945"
+    bad=edge.Proxy(upstream="http://127.0.0.1:18892")
+    with pytest.raises(edge.EdgeError,match="UPSTREAM_DRIFT_DENY"):
+        bad.request("GET","/health",{"Authorization":"Bearer "+"a"*40},None)
+
+
+
+def test_public_bind_requires_approved_domain():
+    assert edge.validated_edge_identity("127.0.0.1","0","")==("127.0.0.1","")
+    with pytest.raises(RuntimeError,match="PUBLIC_DOMAIN_REQUIRED"):
+        edge.validated_edge_identity("0.0.0.0","1","")
+    assert edge.validated_edge_identity(
+        "0.0.0.0","1","tasks.example.com"
+    )==("0.0.0.0","tasks.example.com")
+
+def test_tls_key_requires_no_group_or_other_permissions(tmp_path):
+    key=tmp_path/"key.pem"
+    key.write_text("x"*64,encoding="utf-8")
+    key.chmod(0o600)
+    assert edge.validate_tls_file(key,key=True)==key
+    key.chmod(0o644)
+    with pytest.raises(RuntimeError,match="TLS_KEY_MODE_INVALID"):
+        edge.validate_tls_file(key,key=True)
+
+
+def test_tls_symlink_is_denied(tmp_path):
+    real=tmp_path/"cert.pem"
+    real.write_text("x"*64,encoding="utf-8")
+    link=tmp_path/"link.pem"
+    link.symlink_to(real)
+    with pytest.raises(RuntimeError,match="TLS_FILE_INVALID"):
+        edge.validate_tls_file(link,key=False)
+
+
+def test_systemd_unit_is_disabled_by_owner_approval_condition_and_loopback_default():
+    unit=(ROOT/"deploy/systemd/aios-gpt-https-edge.service").read_text()
+    assert "ConditionPathExists=/etc/aios-gitops/gpt-https-edge.approved" in unit
+    assert "AIOS_GPT_EDGE_HOST=127.0.0.1" in unit
+    assert "AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=0" in unit
+    assert "AIOS_GPT_EDGE_PORT=18443" in unit
+    assert "LoadCredential=tls_cert:" in unit
+    assert "LoadCredential=tls_key:" in unit
+    assert "InaccessiblePaths=/data/aios-gitops" in unit
+    assert "InaccessiblePaths=" in unit and "/var/lib/aios-agent-task" in unit
+    assert "NoNewPrivileges=yes" in unit
+    assert "ProtectSystem=strict" in unit
+
+
+def test_systemd_unit_does_not_reference_internal_control_credentials():
+    unit=(ROOT/"deploy/systemd/aios-gpt-https-edge.service").read_text()
+    assert "agent_task_control" not in unit
+    assert "gitea" not in unit.lower()
+    assert "18892" not in unit
+    assert "18940" not in unit
+
+
+def test_edge_source_does_not_proxy_internal_bridge_or_task_api():
+    source=(ROOT/"active-runtime/app/aios_gpt_https_edge.py").read_text()
+    assert 'UPSTREAM = "http://127.0.0.1:18945"' in source
+    assert "18892" not in source
+    assert "18940" not in source
+    assert "/v2/governed-task" not in source
+    assert "/v1/tasks/" not in source
+
+
+def test_docs_keep_public_activation_separately_approved():
+    doc=(ROOT/"docs/superpowers/specs/2026-08-19-gpt-https-edge.md").read_text()
+    assert "explicit owner approval" in doc.lower()
+    assert "PUBLIC_NETWORK_EXPOSURE=NONE" in doc
+    assert "PUBLIC_CUSTOM_GPT_DIRECT_AI0S_ACCESS=NOT_VERIFIED" in doc

tests/test_gpt_https_edge_activation.py

@@ -0,0 +1,362 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+import hashlib
+import json
+from pathlib import Path
+import sys
+import subprocess
+
+import pytest
+
+ROOT=Path(__file__).resolve().parents[1]
+APP=ROOT/"active-runtime/app"
+sys.path.insert(0,str(APP))
+
+import aios_gpt_https_edge_activation as activation
+
+
+CANDIDATE="69a51d50c7f6cd9045346d41e338acebe4b6f1e8"
+NOW=datetime(2026,8,19,2,0,0,tzinfo=timezone.utc)
+
+
+def approval_value(cert_sha: str, **overrides):
+    value={
+        "schema":"aios.gpt-https-edge.public-approval.v1",
+        "approval_id":"owner-edge-approval-0001",
+        "candidate_oid":CANDIDATE,
+        "domain":"tasks.example.com",
+        "public_bind":True,
+        "bind_host":"0.0.0.0",
+        "port":18443,
+        "tls_cert_sha256":cert_sha,
+        "issued_at":"2026-08-19T01:00:00Z",
+        "not_before":"2026-08-19T01:00:00Z",
+        "expires_at":"2026-08-20T01:00:00Z",
+        "owner":"owner@example",
+    }
+    value.update(overrides)
+    return value
+
+
+def write_owner_file(path: Path, data: bytes, mode=0o600):
+    path.write_bytes(data)
+    path.chmod(mode)
+    return path
+
+
+def load_approval(tmp_path, monkeypatch, **overrides):
+    tmp_path.mkdir(parents=True,exist_ok=True)
+    cert=write_owner_file(tmp_path/"cert.pem",b"certificate-material-"*4,0o600)
+    value=approval_value(hashlib.sha256(cert.read_bytes()).hexdigest(),**overrides)
+    approval=write_owner_file(
+        tmp_path/"approval.json",
+        (json.dumps(value,sort_keys=True)+"\n").encode(),
+        0o600,
+    )
+    monkeypatch.setattr(
+        activation,
+        "validate_root_file",
+        lambda path, **kwargs: path,
+    )
+    return activation.Approval.load(
+        approval,
+        candidate_oid=CANDIDATE,
+        cert_path=cert,
+        now=NOW,
+    ),approval,cert
+
+
+class FakeSystemd:
+    def __init__(self):
+        self.active_units={activation.INGRESS_UNIT}
+        self.enabled_units=set()
+        self.calls=[]
+        self.exec_start=""
+        self.environment=""
+        self.fail_restart=False
+
+    def run(self,*args,check=True):
+        self.calls.append(args)
+        class Result:
+            returncode=0
+            stdout=""
+        if args[:2]==("restart",activation.EDGE_UNIT):
+            if self.fail_restart:
+                raise subprocess.CalledProcessError(1,args)
+            self.active_units.add(activation.EDGE_UNIT)
+        elif args[:2]==("stop",activation.EDGE_UNIT):
+            self.active_units.discard(activation.EDGE_UNIT)
+        elif args[:2]==("enable",activation.EDGE_UNIT):
+            self.enabled_units.add(activation.EDGE_UNIT)
+        elif args[:2]==("disable",activation.EDGE_UNIT):
+            self.enabled_units.discard(activation.EDGE_UNIT)
+        return Result()
+
+    def active(self,unit):
+        return unit in self.active_units
+
+    def enabled(self,unit):
+        return unit in self.enabled_units
+
+    def show(self,unit,prop):
+        if prop=="ExecStart":
+            return self.exec_start
+        if prop=="Environment":
+            return self.environment
+        return ""
+
+
+def test_approval_binds_candidate_domain_bind_port_and_cert(tmp_path,monkeypatch):
+    approval,_,cert=load_approval(tmp_path,monkeypatch)
+    assert approval.candidate_oid==CANDIDATE
+    assert approval.domain=="tasks.example.com"
+    assert approval.bind_host=="0.0.0.0"
+    assert approval.port==18443
+    assert approval.tls_cert_sha256==hashlib.sha256(cert.read_bytes()).hexdigest()
+    assert approval.expires_at>approval.not_before
+
+
+def test_approval_rejects_candidate_mismatch(tmp_path,monkeypatch):
+    cert=write_owner_file(tmp_path/"cert.pem",b"certificate-material-"*4)
+    value=approval_value(hashlib.sha256(cert.read_bytes()).hexdigest())
+    approval=write_owner_file(tmp_path/"approval.json",json.dumps(value).encode())
+    monkeypatch.setattr(activation,"validate_root_file",lambda path,**kwargs:path)
+    with pytest.raises(activation.ActivationError,match="CANDIDATE_MISMATCH"):
+        activation.Approval.load(
+            approval,
+            candidate_oid="0"*40,
+            cert_path=cert,
+            now=NOW,
+        )
+
+
+def test_approval_rejects_cert_hash_mismatch(tmp_path,monkeypatch):
+    cert=write_owner_file(tmp_path/"cert.pem",b"certificate-material-"*4)
+    value=approval_value("a"*64)
+    approval=write_owner_file(tmp_path/"approval.json",json.dumps(value).encode())
+    monkeypatch.setattr(activation,"validate_root_file",lambda path,**kwargs:path)
+    with pytest.raises(activation.ActivationError,match="CERT_SHA_MISMATCH"):
+        activation.Approval.load(
+            approval,
+            candidate_oid=CANDIDATE,
+            cert_path=cert,
+            now=NOW,
+        )
+
+
+def test_approval_rejects_expired_not_yet_valid_and_long_lived(tmp_path,monkeypatch):
+    with pytest.raises(activation.ActivationError,match="APPROVAL_EXPIRED"):
+        load_approval(
+            tmp_path/"expired",
+            monkeypatch,
+            issued_at="2026-08-17T01:00:00Z",
+            not_before="2026-08-17T01:00:00Z",
+            expires_at="2026-08-18T01:00:00Z",
+        )
+    # Use distinct temp roots because helper writes fixed names.
+
+    with pytest.raises(activation.ActivationError,match="NOT_YET_VALID"):
+        load_approval(
+            tmp_path/"future",
+            monkeypatch,
+            issued_at="2026-08-19T03:00:00Z",
+            not_before="2026-08-19T03:00:00Z",
+            expires_at="2026-08-20T03:00:00Z",
+        )
+    with pytest.raises(activation.ActivationError,match="VALIDITY_TOO_LONG"):
+        load_approval(
+            tmp_path/"long",
+            monkeypatch,
+            issued_at="2026-08-18T01:00:00Z",
+            not_before="2026-08-18T01:00:00Z",
+            expires_at="2026-08-26T01:00:00Z",
+        )
+
+
+
+def test_validate_release_requires_edge_and_controller_and_denies_symlink(tmp_path):
+    release=tmp_path/"release"
+    app=release/"active-runtime/app"
+    app.mkdir(parents=True)
+    edge_file=app/"aios_gpt_https_edge.py"
+    controller=app/"aios_gpt_https_edge_activation.py"
+    edge_file.write_text("edge\n")
+    controller.write_text("controller\n")
+    value=activation.validate_release(release)
+    assert value["edge"]==edge_file
+    assert value["controller"]==controller
+    controller.unlink()
+    with pytest.raises(activation.ActivationError,match="IMMUTABLE_RELEASE_INVALID"):
+        activation.validate_release(release)
+
+def test_render_dropin_is_exact_and_contains_no_secret(tmp_path,monkeypatch):
+    approval,_,_=load_approval(tmp_path,monkeypatch)
+    release=Path("/opt/aios/releases")/(CANDIDATE+"-gpt-https-edge-public")
+    text=activation.render_dropin(approval,release)
+    assert "AIOS_GPT_EDGE_HOST=0.0.0.0" in text
+    assert "AIOS_GPT_EDGE_PORT=18443" in text
+    assert "AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=1" in text
+    assert "AIOS_GPT_EDGE_PUBLIC_DOMAIN=tasks.example.com" in text
+    assert CANDIDATE in text
+    for forbidden in ("Bearer ","agent_task_control","tls_key","gpt_ingress_token"):
+        assert forbidden not in text
+
+
+def test_receipt_contains_binding_but_no_owner_secret_material(tmp_path,monkeypatch):
+    approval,_,_=load_approval(tmp_path,monkeypatch)
+    receipt=activation.receipt_for(
+        approval,
+        Path("/opt/aios/releases/test"),
+        state="ACTIVE",
+    )
+    assert receipt["approval_id"]==approval.approval_id
+    assert receipt["candidate_oid"]==CANDIDATE
+    assert receipt["domain"]=="tasks.example.com"
+    serialized=json.dumps(receipt)
+    assert "Bearer " not in serialized
+    assert "private" not in serialized.lower()
+
+
+def test_activate_is_idempotent_for_same_active_approval(tmp_path,monkeypatch):
+    approval,_,cert=load_approval(tmp_path,monkeypatch)
+    release=tmp_path/"release"
+    app=release/"active-runtime/app/aios_gpt_https_edge.py"
+    controller=release/"active-runtime/app/aios_gpt_https_edge_activation.py"
+    app.parent.mkdir(parents=True)
+    app.write_text("print('x')\n")
+    controller.write_text("print('controller')\n")
+    key=write_owner_file(tmp_path/"key.pem",b"private-key-material-"*4,0o600)
+    dropin=tmp_path/"edge.d"/activation.DROPIN_NAME
+    state=tmp_path/"state.json"
+    systemd=FakeSystemd()
+
+    monkeypatch.setattr(activation.os,"geteuid",lambda:0)
+    monkeypatch.setattr(activation,"require_ingress_ready",lambda _:None)
+    monkeypatch.setattr(activation,"validate_tls_pair",lambda *a,**k:{"cert_sha256":approval.tls_cert_sha256})
+    monkeypatch.setattr(activation,"tls_probe",lambda *a,**k:None)
+
+    systemd.exec_start=str(app)
+    systemd.environment=(
+        "AIOS_GPT_EDGE_HOST=0.0.0.0 "
+        "AIOS_GPT_EDGE_PORT=18443 "
+        "AIOS_GPT_EDGE_PUBLIC_BIND_APPROVED=1 "
+        "AIOS_GPT_EDGE_PUBLIC_DOMAIN=tasks.example.com"
+    )
+    first=activation.activate(
+        approval=approval,
+        release_root=release,
+        cert_path=cert,
+        key_path=key,
+        dropin_path=dropin,
+        state_path=state,
+        systemd=systemd,
+    )
+    calls_after_first=len(systemd.calls)
+    second=activation.activate(
+        approval=approval,
+        release_root=release,
+        cert_path=cert,
+        key_path=key,
+        dropin_path=dropin,
+        state_path=state,
+        systemd=systemd,
+    )
+    assert first["state"]=="ACTIVE"
+    assert second==json.loads(state.read_text())
+    assert len(systemd.calls)==calls_after_first
+
+
+def test_activate_failure_after_mutation_rolls_back(tmp_path,monkeypatch):
+    approval,_,cert=load_approval(tmp_path,monkeypatch)
+    release=tmp_path/"release"
+    app=release/"active-runtime/app/aios_gpt_https_edge.py"
+    controller=release/"active-runtime/app/aios_gpt_https_edge_activation.py"
+    app.parent.mkdir(parents=True)
+    app.write_text("print('x')\n")
+    controller.write_text("print('controller')\n")
+    key=write_owner_file(tmp_path/"key.pem",b"private-key-material-"*4,0o600)
+    dropin=tmp_path/"edge.d"/activation.DROPIN_NAME
+    state=tmp_path/"state.json"
+    systemd=FakeSystemd()
+    systemd.fail_restart=True
+
+    monkeypatch.setattr(activation.os,"geteuid",lambda:0)
+    monkeypatch.setattr(activation,"require_ingress_ready",lambda _:None)
+    monkeypatch.setattr(activation,"validate_tls_pair",lambda *a,**k:{"cert_sha256":approval.tls_cert_sha256})
+    monkeypatch.setattr(activation,"tls_probe",lambda *a,**k:None)
+
+    with pytest.raises(subprocess.CalledProcessError):
+        activation.activate(
+            approval=approval,
+            release_root=release,
+            cert_path=cert,
+            key_path=key,
+            dropin_path=dropin,
+            state_path=state,
+            systemd=systemd,
+        )
+    assert not dropin.exists()
+    assert not systemd.active(activation.EDGE_UNIT)
+    assert json.loads(state.read_text())["state"]=="ROLLED_BACK"
+
+
+def test_rollback_is_replay_safe_and_leaves_owner_assets_alone(tmp_path,monkeypatch):
+    approval_file=tmp_path/"approval.json"
+    cert=tmp_path/"cert.pem"
+    key=tmp_path/"key.pem"
+    for path in (approval_file,cert,key):
+        path.write_text("owner-asset"*4)
+    dropin=tmp_path/"edge.d"/activation.DROPIN_NAME
+    dropin.parent.mkdir(parents=True)
+    dropin.write_text("[Service]\n")
+    state=tmp_path/"state.json"
+    state.write_text(json.dumps({
+        "schema":activation.STATE_SCHEMA,
+        "state":"ACTIVE",
+        "candidate_oid":CANDIDATE,
+        "approval_id":"owner-edge-approval-0001",
+        "domain":"tasks.example.com",
+    }))
+    systemd=FakeSystemd()
+    systemd.active_units.add(activation.EDGE_UNIT)
+    systemd.enabled_units.add(activation.EDGE_UNIT)
+    monkeypatch.setattr(activation.os,"geteuid",lambda:0)
+
+    first=activation.rollback(
+        dropin_path=dropin,
+        state_path=state,
+        systemd=systemd,
+        reason="owner_requested",
+    )
+    second=activation.rollback(
+        dropin_path=dropin,
+        state_path=state,
+        systemd=systemd,
+        reason="owner_requested",
+    )
+    assert first["state"]=="ROLLED_BACK"
+    assert second["state"]=="ROLLED_BACK"
+    assert not dropin.exists()
+    assert not systemd.active(activation.EDGE_UNIT)
+    assert not systemd.enabled(activation.EDGE_UNIT)
+    for path in (approval_file,cert,key):
+        assert path.exists()
+
+
+def test_activation_source_never_creates_owner_approval_or_tls_secret():
+    source=(ROOT/"active-runtime/app/aios_gpt_https_edge_activation.py").read_text()
+    assert "secrets.token" not in source
+    assert "openssl" not in source.lower()
+    assert "BEGIN PRIVATE KEY" not in source
+    assert "gpt-https-edge.approved" in source
+    assert "gpt-https-edge-tls.key" in source
+
+
+def test_docs_separate_implementation_activation_and_runtime_verification():
+    doc=(ROOT/"docs/superpowers/specs/2026-08-19-gpt-https-edge-activation.md").read_text()
+    assert "IMPLEMENTED" in doc
+    assert "OWNER_APPROVED" in doc
+    assert "ACTIVATED" in doc
+    assert "RUNTIME_VERIFIED" in doc
+    assert "does not own" in doc.lower()

tests/test_gpt_online_task_ingress.py

@@ -0,0 +1,342 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+import sqlite3
+import sys
+
+import pytest
+
+ROOT=Path(__file__).resolve().parents[1]
+APP=ROOT/"active-runtime/app"
+sys.path.insert(0,str(APP))
+
+from aios_gpt_online_task_ingress import (
+    IngressError,
+    IngressService,
+    Store,
+    READONLY_CAPABILITIES,
+    branch_for,
+    normalized_request,
+    openapi_document,
+    task_id_for,
+    validated_bind_host,
+)
+
+
+MAIN="3244f0f17f8845592ac94d585f5ebd65693c9f29"
+JOB="job-0123456789abcdef0123456789abcdef"
+
+
+def request(request_id="external-request-0001", **overrides):
+    value={
+        "schema":"aios.gpt-online-task-ingress.request.v1",
+        "request_id":request_id,
+        "goal":"Inspect the repository for durable worker state.",
+        "instructions":"Read only. Return a concise source analysis.",
+        "success_criteria":[
+            {"id":"analysis","description":"Return a source-backed analysis."}
+        ],
+        "constraints":{
+            "max_files":50,
+            "focus_paths":["active-runtime/app","tests"],
+            "notes":"Do not mutate repository state.",
+        },
+    }
+    value.update(overrides)
+    return value
+
+
+class FakeUpstream:
+    def __init__(self):
+        self.created=[]
+        self.projected=[]
+        self.tasks={}
+        self.snapshot={"authoritative_head_oid":MAIN}
+
+    def control_snapshot(self):
+        return dict(self.snapshot)
+
+    def create_governed_task(self, body):
+        self.created.append(dict(body))
+        task_id=body["task_id"]
+        job_id=JOB
+        self.tasks[task_id]={
+            "task_id":task_id,
+            "status":"PROJECTED",
+            "attempt":0,
+            "metadata":{
+                "execution":{
+                    "task_id":task_id,
+                    "job_id":job_id,
+                    "base_oid":body["base_oid"],
+                    "action":"source_capture",
+                    "capability_scope":list(READONLY_CAPABILITIES),
+                    "governance":{"writer_lease":False},
+                }
+            },
+        }
+        return 201,{"task_id":task_id,"job_id":job_id,"state":"BOUND"}
+
+    def execution_envelope(self, body):
+        self.projected.append(dict(body))
+        task_id=body["task_id"]
+        return 200,{
+            "schema":"aios.online-agent.execution.v2",
+            "task_id":task_id,
+            "job_id":body["job_id"],
+            "repository":"aios/aios",
+            "base_oid":MAIN,
+            "action":"source_capture",
+            "capability_scope":list(READONLY_CAPABILITIES),
+            "governance":{"writer_lease":False},
+        }
+
+    def get_task(self, task_id):
+        value=self.tasks.get(task_id)
+        return (404,{}) if value is None else (200,dict(value))
+
+
+def service(tmp_path):
+    upstream=FakeUpstream()
+    store=Store(tmp_path/"state.sqlite3")
+    return IngressService(upstream,store),upstream,store
+
+
+def test_request_schema_is_narrow_and_rejects_capability_or_execution_control():
+    value=normalized_request(request())
+    assert value["goal"].startswith("Inspect")
+    for field in ("capability_scope","required_capabilities","action","adapter","repository","base_oid"):
+        bad=request()
+        bad[field]="evil"
+        with pytest.raises(IngressError,match="UNKNOWN_FIELDS"):
+            normalized_request(bad)
+
+
+def test_constraints_are_bounded_and_path_traversal_is_denied():
+    bad=request(constraints={"focus_paths":["../etc/passwd"]})
+    with pytest.raises(IngressError,match="FOCUS_PATHS_INVALID"):
+        normalized_request(bad)
+    bad=request(constraints={"max_files":1000})
+    with pytest.raises(IngressError,match="MAX_FILES_INVALID"):
+        normalized_request(bad)
+
+
+def test_non_loopback_bind_is_denied():
+    assert validated_bind_host("127.0.0.1")=="127.0.0.1"
+    with pytest.raises(RuntimeError,match="GPT_INGRESS_NON_LOOPBACK_BIND_DENY"):
+        validated_bind_host("0.0.0.0")
+    with pytest.raises(RuntimeError,match="GPT_INGRESS_NON_LOOPBACK_BIND_DENY"):
+        validated_bind_host("192.0.2.10")
+
+
+def test_task_and_branch_ids_are_deterministic_and_bounded():
+    a=task_id_for("external-request-0001")
+    b=task_id_for("external-request-0001")
+    c=task_id_for("external-request-0002")
+    assert a==b
+    assert a!=c
+    assert a.startswith("gpt-readonly-")
+    branch=branch_for(a)
+    assert branch.startswith("agent/external-gpt-readonly/")
+    assert len(a)<180
+
+
+def test_create_translates_to_canonical_readonly_governed_task(tmp_path):
+    svc,upstream,store=service(tmp_path)
+    response=svc.create(request())
+    assert response["state"]=="BOUND"
+    assert response["write_access"] is False
+    assert response["official_verification_required"] is True
+    assert response["capabilities"]==list(READONLY_CAPABILITIES)
+
+    assert len(upstream.created)==1
+    body=upstream.created[0]
+    assert body["repository"]=="aios/aios"
+    assert body["base_oid"]==MAIN
+    assert body["expected_branch_oid"]==MAIN
+    assert body["branch_mode"]=="CREATE_NEW"
+    assert body["risk"]=="low"
+    assert body["priority"]==0
+    assert body["max_attempts"]==3
+    assert body["require_online_task_projection"] is True
+    assert body["online_task"]["required_capabilities"]==list(READONLY_CAPABILITIES)
+    assert body["online_task"]["constraints"]["repository_mutation"]=="forbidden"
+    assert body["online_task"]["constraints"]["credential_access"]=="forbidden"
+    assert body["execution"]["action"]=="source_capture"
+    assert body["execution"]["adapter"]=="source_capture"
+    assert body["execution"]["capability_scope"]==list(READONLY_CAPABILITIES)
+    assert "writer_lease" not in body["execution"]
+
+    assert len(upstream.projected)==1
+    projection=upstream.projected[0]
+    assert projection["task_id"]==response["task_id"]
+    assert projection["job_id"]==JOB
+    assert projection["required_capabilities"]==list(READONLY_CAPABILITIES)
+
+    row=store.get("external-request-0001")
+    assert row is not None
+    assert row["state"]=="BOUND"
+    assert row["job_id"]==JOB
+    assert row["base_oid"]==MAIN
+
+
+def test_exact_retry_is_idempotent_without_duplicate_bridge_call(tmp_path):
+    svc,upstream,_=service(tmp_path)
+    first=svc.create(request())
+    second=svc.create(request())
+    assert second==first
+    assert len(upstream.created)==1
+    assert len(upstream.projected)==1
+
+
+def test_same_request_id_with_different_content_is_conflict(tmp_path):
+    svc,upstream,_=service(tmp_path)
+    svc.create(request())
+    with pytest.raises(IngressError,match="IDEMPOTENCY_CONFLICT"):
+        svc.create(request(goal="Different goal."))
+    assert len(upstream.created)==1
+
+
+def test_pending_request_recovers_existing_canonical_task_after_crash(tmp_path):
+    svc,upstream,store=service(tmp_path)
+    raw=normalized_request(request())
+    task_id=task_id_for(raw["request_id"])
+    store.reserve(raw,task_id)
+
+    upstream.tasks[task_id]={
+        "task_id":task_id,
+        "status":"PROJECTED",
+        "attempt":0,
+        "metadata":{
+            "execution":{
+                "task_id":task_id,
+                "job_id":JOB,
+                "base_oid":MAIN,
+                "action":"source_capture",
+                "capability_scope":list(READONLY_CAPABILITIES),
+                "governance":{"writer_lease":False},
+            }
+        },
+    }
+    result=svc.create(request())
+    assert result["job_id"]==JOB
+    assert result["task_id"]==task_id
+    assert len(upstream.created)==0
+    row=store.get(raw["request_id"])
+    assert row["state"]=="BOUND"
+
+
+def test_status_is_sanitized_and_never_proxies_raw_execution_or_result_text(tmp_path):
+    svc,upstream,store=service(tmp_path)
+    created=svc.create(request())
+    task_id=created["task_id"]
+    upstream.tasks[task_id]={
+        "task_id":task_id,
+        "status":"SUCCEEDED",
+        "attempt":1,
+        "claimed_by":"online-engineering-worker-cn-network-mkt-generic-20260815",
+        "metadata":{
+            "execution":{
+                "job_id":JOB,
+                "base_oid":MAIN,
+                "action":"source_capture",
+                "capability_scope":list(READONLY_CAPABILITIES),
+                "governance":{"writer_lease":False},
+                "secret_marker":"do-not-return",
+            }
+        },
+        "result":{
+            "status":"SUCCEEDED",
+            "classification":"NEEDS_REWORK_SOURCE_CAPTURE",
+            "execution_mode":"source_capture",
+            "output_text":"raw source text that must not be proxied",
+            "source_snapshot":{
+                "payload_sha256":"a"*64,
+                "file_count":42,
+                "files":["secret/path"],
+            },
+        },
+        "verification":{
+            "decision":"SUCCEEDED",
+            "classification":"GENERIC_ONLINE_TASK_AUTO_CONSUMPTION_VERIFIED",
+            "private":"do-not-return",
+        },
+    }
+    status=svc.status("external-request-0001")
+    assert status["state"]=="SUCCEEDED"
+    assert status["attempt"]==1
+    assert status["claimed"] is True
+    assert status["official_verification"]=="SUCCEEDED"
+    assert status["source_snapshot_sha256"]=="a"*64
+    assert status["source_snapshot_file_count"]==42
+    serialized=json.dumps(status)
+    assert "raw source text" not in serialized
+    assert "secret_marker" not in serialized
+    assert "secret/path" not in serialized
+    assert "do-not-return" not in serialized
+
+
+def test_pending_status_can_recover_job_binding(tmp_path):
+    svc,upstream,store=service(tmp_path)
+    raw=normalized_request(request())
+    task_id=task_id_for(raw["request_id"])
+    store.reserve(raw,task_id)
+    upstream.tasks[task_id]={
+        "task_id":task_id,
+        "status":"PROJECTED",
+        "attempt":0,
+        "metadata":{
+            "execution":{
+                "job_id":JOB,
+                "base_oid":MAIN,
+                "action":"source_capture",
+                "capability_scope":list(READONLY_CAPABILITIES),
+                "governance":{"writer_lease":False},
+            }
+        },
+    }
+    value=svc.status(raw["request_id"])
+    assert value["job_id"]==JOB
+    assert store.get(raw["request_id"])["state"]=="BOUND"
+
+
+def test_openapi_exposes_only_create_readonly_and_status_operations():
+    doc=openapi_document()
+    assert doc["openapi"]=="3.1.0"
+    paths=doc["paths"]
+    assert set(paths)=={
+        "/v1/read-only-tasks",
+        "/v1/read-only-tasks/{request_id}",
+    }
+    rendered=json.dumps(doc)
+    for forbidden in ("writer-lease","commit","verify","production","tag","baseline"):
+        assert forbidden not in rendered.lower()
+
+
+def test_systemd_unit_is_loopback_credential_isolated_and_has_no_repository_access():
+    unit=(ROOT/"deploy/systemd/aios-gpt-online-task-ingress.service").read_text()
+    assert "DynamicUser=yes" in unit
+    assert "LoadCredential=gpt_ingress_token:" in unit
+    assert "LoadCredential=agent_task_control_token:" in unit
+    assert "AIOS_GPT_INGRESS_HOST=127.0.0.1" in unit
+    assert "AIOS_GOVERNED_WRITE_API_URL=http://127.0.0.1:18892" in unit
+    assert "AIOS_AGENT_TASK_API_URL=http://127.0.0.1:18940" in unit
+    assert "IPAddressDeny=any" in unit
+    assert "IPAddressAllow=127.0.0.0/8" in unit
+    assert "InaccessiblePaths=/data/aios-gitops /etc/aios-gitops /home" in unit
+    assert "ProtectSystem=strict" in unit
+    assert "NoNewPrivileges=yes" in unit
+    assert "MemoryDenyWriteExecute=yes" in unit
+    assert "0.0.0.0" not in unit
+
+
+def test_ingress_sqlite_contains_no_request_body_or_credentials(tmp_path):
+    svc,upstream,store=service(tmp_path)
+    svc.create(request(instructions="sensitive-looking but non-secret user text"))
+    con=sqlite3.connect(store.path)
+    columns=[row[1] for row in con.execute("pragma table_info(external_task_requests)")]
+    assert "request_json" not in columns
+    assert "goal" not in columns
+    assert "instructions" not in columns
+    assert "token" not in " ".join(columns).lower()

tests/test_model_governance_runtime_config.py

@@ -65,3 +65,35 @@ def test_staging_bootstrap_has_explicit_tenant_live_policy():
     }
     assert ("legacy",None,None) in keys
     assert ("tenant","model-governance-live","staging") in keys
+
+
+def test_staging_policy_input_limit_covers_root_owned_openclaw_reservation_floor():
+    config=json.loads(
+        (ROOT/"config/model-governance/model-governance.staging.json").read_text()
+    )
+    profile=json.loads(
+        (ROOT/"config/external-agent-runtime.d/openclaw.profile.json").read_text()
+    )
+    floor=int(profile["model_governance"]["input_reservation_floor_tokens"])
+    assert floor==16384
+    openclaw_model="openai/deepseek-v4-flash"
+    relevant=[
+        policy for policy in config["policies"]
+        if openclaw_model in policy["allowed_models"]
+    ]
+    assert relevant
+    for policy in relevant:
+        assert int(policy["max_input_tokens_per_request"]) >= floor
+
+
+def test_staging_openclaw_scope_limit_is_least_privilege_floor_not_model_ceiling():
+    config=json.loads(
+        (ROOT/"config/model-governance/model-governance.staging.json").read_text()
+    )
+    models={item["model_id"]:item for item in config["models"]}
+    model=models["openai/deepseek-v4-flash"]
+    assert int(model["max_input_tokens"])==32768
+    for policy in config["policies"]:
+        if "openai/deepseek-v4-flash" in policy["allowed_models"]:
+            assert int(policy["max_input_tokens_per_request"])==16384
+            assert int(policy["max_input_tokens_per_request"]) < int(model["max_input_tokens"])

tests/test_release_candidate_capability_contract.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+
+ROOT=Path(__file__).resolve().parents[1]
+sys.path.insert(0,str(ROOT/"active-runtime/app"))
+
+from aios_capability_contract import (  # noqa: E402
+    has_capabilities,
+    normalize_required_capabilities,
+    normalize_worker_capabilities,
+    normalized_capability_matrix,
+    required_capabilities_subset,
+)
+
+
+def test_release_candidate_is_known_to_task_and_worker_vocabularies():
+    capability="release:candidate"
+    assert normalize_required_capabilities([capability])==frozenset({capability})
+    assert normalize_worker_capabilities([capability])==frozenset({capability})
+    assert has_capabilities([capability],[capability]) is True
+    assert normalized_capability_matrix([capability],[capability])=={
+        "worker_effective":["release:candidate"],
+        "task_required":["release:candidate"],
+        "eligible":True,
+    }
+
+
+def test_release_candidate_capability_does_not_imply_high_risk_release_operations():
+    worker=["release:candidate"]
+    for forbidden in (
+        "submitOwnerApproval",
+        "startDeployment",
+        "tag_publish",
+        "baseline_registry_finalize",
+        "runtime_switch",
+        "deploy.production",
+    ):
+        assert has_capabilities(worker,[forbidden]) is False
+
+
+def test_release_wildcards_and_nearby_unknown_names_remain_fail_closed():
+    assert has_capabilities(["release:candidate"],["release:*"]) is False
+    assert has_capabilities(["release:candidate"],["release:candidate.extra"]) is False
+    assert has_capabilities(["release:*"],["release:candidate"]) is False
+
+
+def test_release_candidate_delegation_subset_is_exact():
+    assert required_capabilities_subset(
+        ["release:candidate"],
+        ["release:candidate"],
+    ) is True
+    assert required_capabilities_subset(
+        ["release:candidate"],
+        ["startDeployment"],
+    ) is False
+
+
+@pytest.mark.parametrize(
+    "worker,required",
+    [
+        (["agent.runtime.hermes"],["agent.runtime.hermes"]),
+        (["agent.runtime.openclaw"],["agent.runtime.openclaw"]),
+        (["git.read"],["repository_read"]),
+        (["git.test"],["test_execution"]),
+        (
+            ["governed.write.branch","governed.write.file"],
+            ["governed.git.write"],
+        ),
+    ],
+)
+def test_existing_known_capability_semantics_are_non_regressed(worker,required):
+    assert has_capabilities(worker,required) is True
+
+
+def test_release_candidate_worker_does_not_match_empty_or_unrelated_task_by_escalation():
+    assert has_capabilities(["release:candidate"],[]) is True
+    assert has_capabilities(["release:candidate"],["git.read"]) is False
+    assert has_capabilities(["release:candidate"],["agent.runtime.hermes"]) is False

tests/test_release_lineage_contract.py

@@ -0,0 +1,95 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT=Path(__file__).resolve().parents[1]
+sys.path.insert(0,str(ROOT/"active-runtime/app"))
+
+from aios_release_lineage_contract import (  # noqa: E402
+    FIXED_V2R2C2,
+    ReleaseLineageError,
+    validate_fixed_v2r2c2_task,
+)
+
+
+def fixed_task(source=FIXED_V2R2C2.source_commit, *, status="FAILED", attempt=1, max_attempts=1, revision=4):
+    return {
+        "task_id":FIXED_V2R2C2.task_id,
+        "status":status,
+        "attempt":attempt,
+        "max_attempts":max_attempts,
+        "revision":revision,
+        "required_capabilities":["release:candidate"],
+        "constraints":{
+            "release_candidate":{
+                "candidate_id":FIXED_V2R2C2.candidate_id,
+                "source_commit":source,
+                "version":FIXED_V2R2C2.version,
+                "projection_id":"projection-"+"a"*32,
+            }
+        },
+    }
+
+
+def test_fixed_v2r2c2_lineage_is_the_original_release_worker_candidate():
+    assert FIXED_V2R2C2.source_commit=="71e736b05fab13efc06af7a4fe2dc022c1cf9a85"
+    result=validate_fixed_v2r2c2_task(fixed_task())
+    assert result["status"]=="VERIFIED"
+    assert result["lineage"]["version"]=="V2R2C2"
+
+
+def test_later_capability_remediation_is_not_retroactively_v2r2c2_source():
+    with pytest.raises(ReleaseLineageError,match="FIXED_RELEASE_LINEAGE_DRIFT"):
+        validate_fixed_v2r2c2_task(
+            fixed_task("17deb91ca7a187dc81766945b1174e854f86bd25")
+        )
+
+
+def test_acp_engineering_candidate_is_not_retroactively_v2r2c2_source():
+    with pytest.raises(ReleaseLineageError,match="FIXED_RELEASE_LINEAGE_DRIFT"):
+        validate_fixed_v2r2c2_task(
+            fixed_task("b4b4416c93fedbe451d655d49e28f915b8fd38e6")
+        )
+
+
+@pytest.mark.parametrize(
+    "status,attempt,max_attempts,revision",
+    [
+        ("FAILED",1,1,4),
+        ("QUEUED",1,2,5),
+        ("CLAIMED",2,2,6),
+        ("VERIFYING",2,2,9),
+        ("SUCCEEDED",2,2,10),
+    ],
+)
+def test_lineage_contract_does_not_confuse_lifecycle_state_with_release_identity(
+    status,attempt,max_attempts,revision
+):
+    result=validate_fixed_v2r2c2_task(
+        fixed_task(
+            status=status,
+            attempt=attempt,
+            max_attempts=max_attempts,
+            revision=revision,
+        )
+    )
+    assert result["status"]=="VERIFIED"
+    assert result["task_status"]==status
+    assert result["attempt"]==attempt
+    assert result["max_attempts"]==max_attempts
+    assert result["revision"]==revision
+
+
+def test_capability_and_task_identity_are_bound():
+    value=fixed_task()
+    value["required_capabilities"]=[]
+    with pytest.raises(ReleaseLineageError,match="FIXED_TASK_CAPABILITY_DRIFT"):
+        validate_fixed_v2r2c2_task(value)
+
+    value=fixed_task()
+    value["task_id"]="replacement-v2r2c2"
+    with pytest.raises(ReleaseLineageError,match="FIXED_TASK_ID_DRIFT"):
+        validate_fixed_v2r2c2_task(value)

tests/test_release_online_agent_worker.py

@@ -0,0 +1,226 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+import sys
+
+import pytest
+
+ROOT=Path(__file__).resolve().parents[1]
+APP=ROOT/"active-runtime/app"
+sys.path.insert(0,str(APP))
+
+import aios_release_online_agent_worker as worker
+
+
+def task(**overrides):
+    value={
+        "task_id":"r42-release-candidate-live-acceptance-20260819",
+        "required_capabilities":["release:candidate"],
+        "constraints":{
+            "release_candidate":{
+                "candidate_id":"r42-release-candidate-v2r2c2-20260819",
+                "source_commit":"5"*40,
+                "version":"V2R2C2",
+                "projection_id":"projection-"+"a"*32,
+            }
+        }
+    }
+    value.update(overrides)
+    return value
+
+
+def test_release_request_accepts_only_exact_release_capability():
+    req=worker.ReleaseRequest.from_task(task())
+    assert req.candidate_id=="r42-release-candidate-v2r2c2-20260819"
+    assert req.source_commit=="5"*40
+    assert req.version=="V2R2C2"
+    assert req.projection_id=="projection-"+"a"*32
+
+    with pytest.raises(worker.ReleaseWorkerError,match="TASK_CAPABILITY_SCOPE_INVALID"):
+        worker.ReleaseRequest.from_task(task(required_capabilities=["release:candidate","git.read"]))
+
+
+@pytest.mark.parametrize(
+    "payload",
+    [
+        {
+            "candidate_id":"r42-release-candidate-v2r2c2-20260819",
+            "source_commit":"short",
+            "version":"V2R2C2",
+            "projection_id":"projection-"+"a"*32,
+        },
+        {
+            "candidate_id":"bad",
+            "source_commit":"5"*40,
+            "version":"V2R2C2",
+            "projection_id":"projection-"+"a"*32,
+        },
+        {
+            "candidate_id":"r42-release-candidate-v2r2c2-20260819",
+            "source_commit":"5"*40,
+            "version":"bad",
+            "projection_id":"projection-"+"a"*32,
+        },
+        {
+            "candidate_id":"r42-release-candidate-v2r2c2-20260819",
+            "source_commit":"5"*40,
+            "version":"V2R2C2",
+            "projection_id":"bad",
+        },
+        {
+            "candidate_id":"r42-release-candidate-v2r2c2-20260819",
+            "source_commit":"5"*40,
+            "version":"V2R2C2",
+            "projection_id":"projection-"+"a"*32,
+            "execute":True,
+        },
+    ],
+)
+def test_release_request_rejects_invalid_or_extra_fields(payload):
+    value=task()
+    value["constraints"]={"release_candidate":payload}
+    with pytest.raises(worker.ReleaseWorkerError):
+        worker.ReleaseRequest.from_task(value)
+
+
+def test_result_payload_preserves_high_risk_denials():
+    release=worker.ReleaseRequest.from_task(task())
+    result=worker.result_payload(
+        "r42-release-candidate-live-acceptance-20260819",
+        release,
+        {"candidate_id":release.candidate_id,"status":"BUILT"},
+    )
+    assert result["status"]=="SUCCEEDED"
+    assert result["classification"]=="RELEASE_CANDIDATE_CREATED"
+    assert result["writer_lease"]=="NONE"
+    assert result["changed_files"]==[]
+    assert result["direct_main_write"] is False
+    assert result["production_write"] is False
+    assert result["tag_write"] is False
+    assert result["baseline_write"] is False
+    assert result["runtime_switch"] is False
+    assert result["owner_approval_submitted"] is False
+    assert result["credential_material_exposed"] is False
+
+
+def test_result_payload_rejects_builder_candidate_mismatch():
+    release=worker.ReleaseRequest.from_task(task())
+    with pytest.raises(worker.ReleaseWorkerError,match="BUILDER_CANDIDATE_BINDING_INVALID"):
+        worker.result_payload(
+            "r42-release-candidate-live-acceptance-20260819",
+            release,
+            {"candidate_id":"other-release-candidate"},
+        )
+
+
+def test_worker_calls_only_fixed_release_candidate_route(monkeypatch):
+    release=worker.ReleaseRequest.from_task(task())
+    calls=[]
+
+    def fake_http(base,method,route,token,body=None,timeout=30):
+        calls.append((base,method,route,token,body))
+        if route=="/health":
+            return 200,{"status":"ok"}
+        if route=="/createReleaseCandidate" and body=={}:
+            return 400,{"error":"request_invalid"}
+        if route.endswith("/claim"):
+            return 200,{"lease_token":"l"*32,"attempt":1}
+        if route.endswith("/heartbeat"):
+            return 200,{}
+        if route=="/createReleaseCandidate":
+            return 200,{
+                "candidate_id":release.candidate_id,
+                "artifact_sha256":"f"*64,
+            }
+        if route.endswith("/result"):
+            return 200,{"status":"VERIFYING"}
+        raise AssertionError(route)
+
+    monkeypatch.setattr(worker,"http_json",fake_http)
+    monkeypatch.setattr(worker,"audit",lambda *args,**kwargs:None)
+    monkeypatch.setattr(worker,"write_liveness",lambda *args,**kwargs:None)
+    monkeypatch.setattr(worker.Heartbeat,"first",lambda self:None)
+    monkeypatch.setattr(worker.Heartbeat,"close",lambda self:None)
+    monkeypatch.setattr(worker.Heartbeat,"ensure",lambda self:None)
+    monkeypatch.setattr(worker.threading.Thread,"start",lambda self:None)
+
+    assert worker.execute_task(
+        task(),
+        agent_token="a"*32,
+        release_token="r"*32,
+    ) is True
+
+    release_calls=[
+        item for item in calls
+        if item[0]==worker.RELEASE_API
+        and item[2]=="/createReleaseCandidate"
+        and item[4]!={}
+    ]
+    assert len(release_calls)==1
+    assert release_calls[0][1]=="POST"
+    assert release_calls[0][2]=="/createReleaseCandidate"
+    assert release_calls[0][4]=={
+        "candidate_id":release.candidate_id,
+        "source_commit":release.source_commit,
+        "version":release.version,
+    }
+
+
+
+def test_write_liveness_records_release_capability(tmp_path,monkeypatch):
+    path=tmp_path/"worker-liveness.json"
+    monkeypatch.setattr(worker,"STATE_ROOT",tmp_path)
+    monkeypatch.setattr(worker,"LIVENESS_PATH",path)
+    worker.write_liveness(
+        "IDLE",
+        stage="poll",
+        visible_count=2,
+    )
+    value=json.loads(path.read_text())
+    assert value["schema"]=="aios.online-agent.worker-liveness.v1"
+    assert value["agent_type"]=="release-preparation-worker"
+    assert value["capabilities"]==["release:candidate"]
+    assert value["state"]=="IDLE"
+    assert value["visible_count"]==2
+
+
+def test_sd_notify_is_noop_without_systemd_socket(monkeypatch):
+    monkeypatch.delenv("NOTIFY_SOCKET",raising=False)
+    worker.sd_notify("WATCHDOG=1")
+
+def test_systemd_unit_uses_two_encrypted_credentials_and_loopback_release_api():
+    unit=(ROOT/"deploy/systemd/aios-release-online-agent-worker.service").read_text()
+    assert "LoadCredentialEncrypted=agent_token:" in unit
+    assert "LoadCredentialEncrypted=release_candidate_token:" in unit
+    assert "AIOS_RELEASE_API_URL=http://127.0.0.1:18890" in unit
+    assert "AIOS_RELEASE_WORKER_AGENT_ID=online-release-worker-cn-network-mkt-20260819" in unit
+    assert "Type=notify" in unit
+    assert "WatchdogSec=90" in unit
+    assert "Restart=always" in unit
+    assert "IPAddressDeny=any" in unit
+    assert "IPAddressAllow=127.0.0.0/8" in unit
+    assert "InaccessiblePaths=/data/aios-gitops /home" in unit
+    assert "submitOwnerApproval" not in unit
+    assert "startDeployment" not in unit
+
+
+def test_source_contains_no_high_risk_release_routes():
+    source=(ROOT/"active-runtime/app/aios_release_online_agent_worker.py").read_text()
+    assert '"/createReleaseCandidate"' in source
+    for forbidden in (
+        "/submitOwnerApproval",
+        "/startDeployment",
+        "/requestOrCreateTag",
+        "/buildProductionArtifact",
+        "/executeApprovedProductionRelease",
+    ):
+        assert forbidden not in source
+
+
+def test_docs_keep_release_candidate_distinct_from_production():
+    doc=(ROOT/"docs/superpowers/specs/2026-08-19-release-preparation-online-agent.md").read_text()
+    assert "release:candidate" in doc
+    assert "Owner signature" in doc
+    assert "never\nstarts deployment" in doc or "never starts deployment" in doc
+    assert "LoadCredentialEncrypted" in doc