feat(governance): persist external release capability projection outbox

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_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())

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-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

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-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
+```

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_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()