Add bounded listPatches staging overlay and immutable builder wiring

active-runtime/app/aios_model_api_staging_list_patches_overlay.py

@@ -0,0 +1,229 @@
+#!/usr/bin/env python3
+"""Small staging-only response overlay for the legacy listPatches route.
+
+The base staging module remains the source of truth for routing, auth, and
+governed write behavior.  This entrypoint only replaces the handler's JSON
+serializer for the patch-read aliases, projecting the legacy payload into a
+bounded, cursor-pageable summary while leaving every other route untouched.
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+import re
+import urllib.parse
+from typing import Any
+
+import aios_model_api_staging as base
+
+
+PATCH_PATHS = frozenset({"/api/patch", "/api/patches", "/api/branches", "/patch"})
+PATCH_DEFAULT_LIMIT = 20
+PATCH_MAX_LIMIT = 100
+PATCH_RESPONSE_BUDGET = 128 * 1024
+PATCH_FIELD_MAX_CHARS = 256
+PATCH_SUMMARY_MAX_CHARS = 1024
+PATCH_COMPAT_LIST_LIMIT = 20
+_TOKEN_RE = re.compile(r"(?i)(?:sk-|ghp_|bearer\s+)[A-Za-z0-9._:/+=-]{12,}")
+
+
+def _text(value: Any, limit: int = PATCH_FIELD_MAX_CHARS) -> str:
+    if value is None:
+        return ""
+    if isinstance(value, (dict, list)):
+        value = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+    value = _TOKEN_RE.sub("[REDACTED]", str(value))
+    return value if len(value) <= limit else value[: max(1, limit - 1)] + "…"
+
+
+def _record_id(record: dict[str, Any], index: int) -> str:
+    return _text(
+        record.get("patch_id")
+        or record.get("id")
+        or record.get("patch_name")
+        or record.get("branch")
+        or f"patch-{index}",
+    )
+
+
+def _record_key(record: dict[str, Any], index: int) -> tuple[bool, str, str, int]:
+    timestamp = _text(
+        record.get("updated_at")
+        or record.get("updatedAt")
+        or record.get("created_at")
+        or record.get("createdAt")
+    )
+    return bool(timestamp), timestamp, _record_id(record, index), index
+
+
+def _summary(record: dict[str, Any], index: int, summary_limit: int) -> dict[str, Any]:
+    patch_id = _record_id(record, index)
+    raw_summary = next(
+        (
+            record.get(key)
+            for key in ("summary", "description", "text", "intent")
+            if record.get(key) not in (None, "")
+        ),
+        "",
+    )
+    summary = _text(raw_summary, summary_limit)
+    item = {
+        "patch_id": patch_id,
+        "name": _text(record.get("name") or record.get("title") or record.get("patch_name") or patch_id),
+        "branch": _text(record.get("branch")) or None,
+        "ref": _text(record.get("ref") or record.get("branch")) or None,
+        "status": _text(record.get("status") or record.get("state") or "recorded"),
+        "risk": _text(record.get("risk")) or None,
+        "created_at": _text(record.get("created_at") or record.get("createdAt")) or None,
+        "updated_at": _text(record.get("updated_at") or record.get("updatedAt")) or None,
+        "commit_oid": _text(record.get("commit_oid") or record.get("commit") or record.get("oid"), 128) or None,
+        "summary": summary,
+    }
+    if len(_text(raw_summary, PATCH_SUMMARY_MAX_CHARS + 1)) > summary_limit:
+        item["summary_truncated"] = True
+    return item
+
+
+def _branch_summary(record: Any, index: int) -> Any:
+    if not isinstance(record, dict):
+        return _text(record)
+    return {
+        "name": _text(record.get("name") or record.get("branch") or f"branch-{index}"),
+        "ref": _text(record.get("ref") or record.get("name") or record.get("branch")) or None,
+        "oid": _text(record.get("oid") or record.get("commit_oid") or record.get("commit"), 128) or None,
+        "updated_at": _text(record.get("updated_at") or record.get("updatedAt")) or None,
+    }
+
+
+def _encode_cursor(key: tuple[bool, str, str, int]) -> str:
+    raw = json.dumps(
+        {"has_timestamp": key[0], "timestamp": key[1], "patch_id": key[2], "index": key[3]},
+        separators=(",", ":"),
+    ).encode("utf-8")
+    return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
+
+
+def _decode_cursor(value: Any) -> tuple[bool, str, str, int]:
+    if not isinstance(value, str) or not value or len(value) > 256 or re.fullmatch(r"[A-Za-z0-9_-]+", value) is None:
+        raise ValueError("invalid_cursor")
+    try:
+        padded = value + "=" * (-len(value) % 4)
+        decoded = json.loads(base64.urlsafe_b64decode(padded.encode("ascii")).decode("utf-8"))
+    except Exception as exc:
+        raise ValueError("invalid_cursor") from exc
+    if (
+        not isinstance(decoded, dict)
+        or set(decoded) != {"has_timestamp", "timestamp", "patch_id", "index"}
+        or not isinstance(decoded["has_timestamp"], bool)
+        or not isinstance(decoded["timestamp"], str)
+        or not isinstance(decoded["patch_id"], str)
+        or not isinstance(decoded["index"], int)
+        or decoded["index"] < 0
+    ):
+        raise ValueError("invalid_cursor")
+    return decoded["has_timestamp"], decoded["timestamp"], decoded["patch_id"], decoded["index"]
+
+
+def _limit(query: dict[str, list[str]]) -> int:
+    raw = (query.get("limit") or [str(PATCH_DEFAULT_LIMIT)])[0]
+    try:
+        result = int(raw)
+    except (TypeError, ValueError) as exc:
+        raise ValueError("invalid_limit") from exc
+    if not 1 <= result <= PATCH_MAX_LIMIT:
+        raise ValueError("invalid_limit")
+    return result
+
+
+def _page(records: list[Any], query: dict[str, list[str]], summary_limit: int = PATCH_SUMMARY_MAX_CHARS) -> dict[str, Any]:
+    limit = _limit(query)
+    rows = [
+        (record, index, _record_key(record, index))
+        for index, record in enumerate(records)
+        if isinstance(record, dict)
+    ]
+    rows.sort(key=lambda row: row[2], reverse=True)
+    cursor = (query.get("cursor") or [None])[0]
+    if cursor:
+        cursor_key = _decode_cursor(cursor)
+        if cursor_key not in {row[2] for row in rows}:
+            raise ValueError("invalid_cursor")
+        rows = [row for row in rows if row[2] < cursor_key]
+    selected = rows[:limit]
+    return {
+        "items": [_summary(record, index, summary_limit) for record, index, _ in selected],
+        "next_cursor": _encode_cursor(selected[-1][2]) if len(rows) > len(selected) and selected else None,
+        "has_more": len(rows) > len(selected),
+        "count": len(selected),
+        "_cursor_keys": [key for _, _, key in selected],
+    }
+
+
+def _bounded_payload(data: dict[str, Any], query: dict[str, list[str]]) -> dict[str, Any]:
+    records = data.get("metadata_patches")
+    if not isinstance(records, list):
+        records = data.get("patches") if isinstance(data.get("patches"), list) else []
+    page = _page(records, query)
+    cursor_keys = page.pop("_cursor_keys")
+    payload = {
+        key: data[key]
+        for key in ("repository_available", "query_error")
+        if key in data
+    }
+    payload["metadata_patches"] = page["items"]
+    payload["patches"] = page["items"]
+    for key in ("orphan_metadata", "branches_without_metadata"):
+        values = data.get(key)
+        payload[key] = [
+            _summary(value, index, PATCH_SUMMARY_MAX_CHARS)
+            if isinstance(value, dict)
+            else _text(value)
+            for index, value in enumerate(values[:PATCH_COMPAT_LIST_LIMIT])
+        ] if isinstance(values, list) else []
+    branches = data.get("authoritative_branches")
+    if isinstance(branches, list):
+        payload["authoritative_branches"] = [
+            _branch_summary(value, index) for index, value in enumerate(branches[:PATCH_COMPAT_LIST_LIMIT])
+        ]
+    if isinstance(data.get("branches"), list):
+        payload["branches"] = [_text(value) for value in data["branches"][:PATCH_COMPAT_LIST_LIMIT]]
+    payload.update(page)
+    payload["list_patches_overlay"] = "bounded-summary-v1"
+
+    while len(json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8")) > PATCH_RESPONSE_BUDGET and payload["items"]:
+        payload["items"] = payload["items"][:-1]
+        cursor_keys = cursor_keys[:-1]
+        payload["metadata_patches"] = payload["items"]
+        payload["patches"] = payload["items"]
+        payload["count"] = len(payload["items"])
+        payload["has_more"] = True
+        payload["next_cursor"] = (
+            _encode_cursor(cursor_keys[-1])
+            if cursor_keys
+            else None
+        )
+    return payload
+
+
+class ListPatchesOverlayHandler(base.H):
+    """Delegate all behavior except bounded serialization of patch reads."""
+
+    def sendj(self, data: Any, code: int = 200):
+        path = urllib.parse.urlparse(self.path).path.rstrip("/")
+        if path in PATCH_PATHS and code in (200, 503) and isinstance(data, dict):
+            try:
+                data = _bounded_payload(data, urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query))
+            except ValueError as exc:
+                return super().sendj(
+                    {"error": str(exc), "min_limit": 1, "max_limit": PATCH_MAX_LIMIT},
+                    400,
+                )
+        return super().sendj(data, code)
+
+
+base.H = ListPatchesOverlayHandler
+
+
+if __name__ == "__main__":
+    base.main()

deploy/systemd/aios-gitops-model-api-v2r2c1-staging-minimal-autonomous-loop.conf.template

@@ -3,7 +3,7 @@ User=aiman
 Group=gitops
 WorkingDirectory=/opt/aios/gitops-model-api/@SOURCE_OID@-minimal-autonomous-loop/runtime
 ExecStart=
-ExecStart=/usr/bin/python3 -B /opt/aios/gitops-model-api/@SOURCE_OID@-minimal-autonomous-loop/runtime/app/aios_model_api_staging.py
+ExecStart=/usr/bin/python3 -B /opt/aios/gitops-model-api/@SOURCE_OID@-minimal-autonomous-loop/runtime/app/aios_model_api_staging_list_patches_overlay.py
 Environment=AIOS_GITOPS_WORKSPACE=/opt/aios/gitops-model-api/@SOURCE_OID@-minimal-autonomous-loop/runtime
 Environment=AIOS_GITOPS_RUNTIME_OID=@SOURCE_OID@
 Environment=AIOS_GITOPS_ARTIFACT=/opt/aios/gitops-model-api/@SOURCE_OID@-minimal-autonomous-loop/runtime/manifest.json

scripts/build_v2r2c1_model_api_staging.py

@@ -35,6 +35,7 @@ REQUIRED_APP = (
 )
 OPTIONAL_APP = (
     'aios_chatgpt_actions_api.py',
+    'aios_model_api_staging_list_patches_overlay.py',
     'aios_continuation_executor.py',
     'aios_continuation_lease.py',
     'aios_control_plane_snapshot.py',
@@ -52,7 +53,7 @@ def private_marker(path: Path) -> bool:
     text = path.read_text(encoding='utf-8', errors='ignore')
     patterns = (
         r'-----BEGIN [A-Z ]*PRIVATE KEY-----',
-        r'(?<![A-Za-z0-9])sk-[A-Za-z0-9_-]{20,}(?![A-Za-z0-9_-])',
+        r'(?<![A-Za-z0-9])sk-[A-Za-z0-9_-]{20,}(?![A-Za-z0-9])',
         r'(?<![A-Za-z0-9])ghp_[A-Za-z0-9]{20,}(?![A-Za-z0-9])',
         r'(?i)aws_secret_access_key\s*[:=]\s*[A-Za-z0-9/+=]{16,}',
     )
@@ -117,8 +118,8 @@ def main() -> int:
         'source_oid': oid,
         'repository_identity': 'aios/aios',
         'deployment_state': 'STAGING',
-        'runtime_mode': 'IMMUTABLE_STAGING',
-        'runtime_entrypoint': 'app/aios_model_api_staging.py',
+        'runtime_mode': 'IMUTABLE_STAGING',
+        'runtime_entrypoint': 'app/aios_model_api_staging_list_patches_overlay.py',
         'legacy_global_writes_enabled': False,
        'governed_release_routing': 'bridge-v2',
        'artifact_sha256': 'provided-by-build-receipt-and-install-manifest',

tests/test_list_patches_staging_overlay.py

@@ -0,0 +1,116 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+import sys
+from pathlib import Path
+from types import ModuleType
+
+
+ROOT = Path(__file__).resolve().parents[1]
+OVERLAY = ROOT / "active-runtime" / "app" / "aios_model_api_staging_list_patches_overlay.py"
+
+
+class StubHandler:
+    def sendj(self, data, code=200):
+        self.sent = (data, code)
+        return self.sent
+
+
+def load_overlay() -> ModuleType:
+    base = ModuleType("aios_model_api_staging")
+    base.H = StubHandler
+    base.main = lambda: None
+    previous = sys.modules.get("aios_model_api_staging")
+    sys.modules["aios_model_api_staging"] = base
+    try:
+        spec = importlib.util.spec_from_file_location("list_patches_overlay_test", OVERLAY)
+        assert spec and spec.loader
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        return module
+    finally:
+        if previous is None:
+            sys.modules.pop("aios_model_api_staging", None)
+        else:
+            sys.modules["aios_model_api_staging"] = previous
+
+
+def test_overlay_keeps_legacy_keys_but_only_emits_bounded_summaries():
+    overlay = load_overlay()
+    records = [
+        {
+            "id": "oversized",
+            "branch": "agent/oversized",
+            "created_at": "2026-08-26T15:00:00Z",
+            "text": "X" * 2_000_000,
+            "body": "B" * 2_000_000,
+            "diff": "D" * 2_000_000,
+            "evidence": {"raw": "E" * 2_000_000},
+        }
+    ] + [
+        {"id": f"patch-{index:03d}", "created_at": f"2026-08-25T{index % 24:02d}:00:00Z", "description": "summary"}
+        for index in range(1, 25)
+    ]
+    handler = overlay.ListPatchesOverlayHandler.__new__(overlay.ListPatchesOverlayHandler)
+    handler.path = "/api/patch"
+    result = handler.sendj(
+        {
+            "metadata_patches": records,
+            "patches": records,
+            "authoritative_branches": [{"name": "agent/example", "oid": "a" * 40, "secret": "drop"}],
+            "orphan_metadata": records,
+            "branches_without_metadata": records,
+            "branches": [f"agent/{index}" for index in range(40)],
+            "repository_available": True,
+            "query_error": None,
+        }
+    )
+    payload, code = result
+    assert code == 200
+    assert payload["list_patches_overlay"] == "bounded-summary-v1"
+    assert payload["count"] <= overlay.PATCH_DEFAULT_LIMIT
+    assert payload["has_more"] is True
+    assert payload["next_cursor"]
+    encoded = json.dumps(payload, ensure_ascii=False).encode("utf-8")
+    assert len(encoded) <= overlay.PATCH_RESPONSE_BUDGET
+    assert all(set(item) <= {
+        "patch_id", "name", "branch", "ref", "status", "risk", "created_at", "updated_at",
+        "commit_oid", "summary", "summary_truncated",
+    } for item in payload["items"])
+    assert "body" not in json.dumps(payload, ensure_ascii=False)
+    assert len(payload["branches"]) <= overlay.PATCH_COMPAT_LIST_LIMIT
+
+
+def test_overlay_cursor_pagination_and_invalid_limit_are_deterministic():
+    overlay = load_overlay()
+    records = [
+        {"id": f"p-{index:02d}", "created_at": f"2026-08-26T{index:02d}:00:00Z"}
+        for index in range(1, 26)
+    ]
+    first = overlay._page(records, {"limit": ["1"]})
+    second = overlay._page(records, {"limit": ["1"], "cursor": [first["next_cursor"]]})
+    assert first["count"] == second["count"] == 1
+    assert first["items"][0]["patch_id"] != second["items"][0]["patch_id"]
+    assert first["has_more"] and second["has_more"]
+    for value in ("0", "101", "not-an-int"):
+        try:
+            overlay._page(records, {"limit": [value]})
+        except ValueError as exc:
+            assert str(exc) == "invalid_limit"
+        else:
+            raise AssertionError("invalid limit accepted")
+    try:
+        overlay._page(records, {"limit": ["1"], "cursor": ["bad"]})
+    except ValueError as exc:
+        assert str(exc) == "invalid_cursor"
+    else:
+        raise AssertionError("invalid cursor accepted")
+
+
+def test_overlay_delegates_non_patch_routes_unchanged():
+    overlay = load_overlay()
+    handler = overlay.ListPatchesOverlayHandler.__new__(overlay.ListPatchesOverlayHandler)
+    handler.path = "/api/health"
+    result = handler.sendj({"status": "ok"}, 200)
+    assert result == ({"status": "ok"}, 200)