feat: prepare disabled fixed-policy Custodian snapshot endpoint

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a15374-f59f-7a31-86d2-d23b5d2ed181

SCOPE.md

@@ -115,3 +115,10 @@ active repository set using the existing Forge/Hub verification and receipts.
 Missing active repositories or limits smaller than the set fail closed; partial
 budget/failure results still require caller coverage checks. This does not
 install the disabled schedule or add any credential, transport or producer.
+
+The runtime has a default-disabled fixed-policy Custodian snapshot endpoint,
+reusing existing Forge read settings and targeted native freshness receipts.
+It prepares bounded exact-commit workplan/decision bundles, not a general file
+service. It is not deployed or admitted to Activity Core. Producer cancellation,
+combined daily timing, receiver integration and reviewed downstream private-source
+access remain unfinished; see `docs/custodian-fleet-input-contract.md`.

docs/custodian-fleet-input-contract.md

@@ -129,3 +129,33 @@ Local tests use temporary Git repositories and prove committed-byte export despi
 dirty files, strict path selection, hashes/completion marker, immutable names,
 invalid-request refusal, selected symlink refusal and size bounds. They perform no
 network, production reads, schedule changes or Hub writes.
+
+## Wind-down checkpoint: disabled remote capture source
+
+`custodian_snapshot.py` and the existing runtime now prepare a fixed-policy
+`POST /ports/repositories/custodian-snapshot` endpoint. Default is404/disabled.
+Configuration requires `REPO_MANAGER_CUSTODIAN_SNAPSHOT_ENABLED=true`,
+`REPO_MANAGER_CUSTODIAN_SNAPSHOT_POLICY` and
+`REPO_MANAGER_CUSTODIAN_SNAPSHOT_HUB`; existing PublisherSettings supplies the
+authorized Forge reader. Existing optional API-token admission is preserved.
+No caller paths/refs/repos/commands are used. No source endpoint is deployed.
+
+Capture targets the complete configured set through native freshness, then reads
+allowlisted files at those exact commits through Forge contents API. It verifies
+Git blob identity, refuses unexpected/symlink files and returns a content-
+addressed manifest/receipt/file bundle. Default hard payload bound8MiB and
+per-response bound3MiB fail closed. Publisher state has no new persistent volume.
+
+Validation: 47 publisher/capture/freshness/exporter tests passed with mocked HTTP;
+one existing Starlette/httpx deprecation warning. Source lint review remains a
+restart check. The source endpoint uses a process-local concurrency lock and
+240-second native budget; it does NOT cancel producer work on client disconnect
+or retain durable in-progress receipt state. Do not treat it as ready for the
+proposed45-second combined daily pre-run phase. Existing15-repo capture timing
+has not been measured. Activity Core receiver is standalone and unintegrated;
+see its `k8s/custodian-release/RESTART.md` for exact restart steps.
+
+The core-hub publisher's existing CCR0031 reader remains in that namespace. New
+Activity Core access to private workplan/decision content and publisher-to-Hub
+egress require the explicit source-reviewed data/network delta before live
+promotion. No credential route, access scope or schedule was changed here.

src/repo_manager/custodian_snapshot.py

@@ -0,0 +1,112 @@
+"""Opt-in, bounded fixed-policy capture using the publisher's existing Forge reader."""
+from __future__ import annotations
+
+import base64
+from dataclasses import replace
+import hashlib
+import json
+from pathlib import Path, PurePosixPath
+import tempfile
+import time
+from urllib.parse import quote
+
+import httpx
+
+from repo_manager.freshness_receipts import load_receipts
+from repo_manager.freshness_runner import RunConfig, run
+
+
+def capture(policy: dict, cfg: RunConfig, *, hub_transport=None, forge_transport=None) -> dict:
+    if policy.get("schema") != "activity-core.custodian-input-policy.v1" or not cfg.forge_token:
+        raise ValueError("fixed capture policy and existing authenticated reader required")
+    repos = tuple(policy["repositories"])
+    if not repos or len(repos) > 15:
+        raise ValueError("capture policy repository bound exceeded")
+    deadline = time.monotonic() + cfg.max_seconds
+    # Cap in-memory capture independently of the consumer's broader retention bound.
+    cap = min(int(policy["max_snapshot_bytes"]), 8 * 1024 * 1024)
+    total = 0
+    files = {}
+    manifest = {"schema": "repo-manager.fleet-input.v1", "freshness_verified": False,
+                "repositories": []}
+    with tempfile.TemporaryDirectory(prefix="rmgr-custodian-capture-") as temporary:
+        selected_cfg = replace(cfg, repo_slugs=repos, limit=len(repos), dry_run=False,
+                               receipts_path=Path(temporary) / "receipts.jsonl")
+        summary = run(selected_cfg, hub_transport=hub_transport, forge_transport=forge_transport)
+        receipts = load_receipts(selected_cfg.receipts_path)
+        accepted = {r["repo_slug"]: r for r in receipts if r["outcome"] in {"applied", "noop"}}
+        if summary["aborted"] or summary["deferred"] or set(accepted) != set(repos):
+            raise ValueError("incomplete targeted freshness capture")
+        manifest["observed_at"] = summary["finished_at"]
+        with httpx.Client(base_url=cfg.forge_base.rstrip("/"), follow_redirects=False,
+                          headers={"Authorization": f"token {cfg.forge_token}"},
+                          transport=forge_transport) as forge:
+            def fetch(slug, relative, commit, optional=False):
+                remaining = deadline - time.monotonic()
+                if remaining <= 0:
+                    raise ValueError("capture deadline exceeded")
+                url = f"/api/v1/repos/{quote(cfg.forge_owner, safe='')}/{quote(slug, safe='')}/contents/{quote(relative, safe='/')}"
+                # Streaming bound applies even if upstream lies about Content-Length.
+                with forge.stream("GET", url, params={"ref": commit}, timeout=min(15, remaining)) as response:
+                    if response.status_code == 404 and optional:
+                        return []
+                    if response.status_code != 200:
+                        raise ValueError("pinned Forge content unavailable")
+                    raw = bytearray()
+                    for chunk in response.iter_bytes():
+                        raw.extend(chunk)
+                        if len(raw) > 3 * 1024 * 1024:
+                            raise ValueError("Forge content response exceeds limit")
+                    return json.loads(raw)
+
+            for slug in sorted(repos):
+                commit = accepted[slug]["forge_commit"]
+                records = []
+                queue = [("workplans", False), ("decisions", True)]
+                visited = set()
+                while queue:
+                    directory, optional = queue.pop()
+                    if directory in visited or len(visited) >= 256:
+                        raise ValueError("duplicate or excessive source directories")
+                    visited.add(directory)
+                    entries = fetch(slug, directory, commit, optional)
+                    if not isinstance(entries, list) or len(entries) > 10000:
+                        raise ValueError("invalid source directory")
+                    for entry in entries:
+                        relative = entry.get("path", "")
+                        path = PurePosixPath(relative)
+                        if path.is_absolute() or ".." in path.parts or str(path.parent) != directory:
+                            raise ValueError("source path escapes selected directory")
+                        if entry.get("type") == "dir" and directory.startswith("workplans"):
+                            queue.append((relative, False))
+                        elif path.suffix == ".md":
+                            if entry.get("type") != "file":
+                                raise ValueError("source record is not regular file")
+                            item = fetch(slug, relative, commit)
+                            if item.get("type") != "file" or item.get("encoding") != "base64":
+                                raise ValueError("invalid source file representation")
+                            data = base64.b64decode(item["content"], validate=True)
+                            total += len(data)
+                            if len(data) > 2 * 1024 * 1024 or total > cap:
+                                raise ValueError("capture content limit exceeded")
+                            blob = hashlib.sha1(b"blob " + str(len(data)).encode() + b"\0" + data).hexdigest()
+                            if blob != entry.get("sha") or blob != item.get("sha"):
+                                raise ValueError("source blob identity mismatch")
+                            key = slug + "/" + relative
+                            if key in files:
+                                raise ValueError("duplicate source record")
+                            files[key] = base64.b64encode(data).decode()
+                            records.append({"path": relative, "bytes": len(data),
+                                            "sha256": hashlib.sha256(data).hexdigest()})
+                if not any(r["path"].startswith("workplans/") for r in records):
+                    raise ValueError("required workplan records absent")
+                manifest["repositories"].append({"slug": slug, "commit": commit,
+                                                 "files": sorted(records, key=lambda r: r["path"])})
+    raw_manifest = json.dumps(manifest, sort_keys=True, indent=2) + "\n"
+    raw_receipts = json.dumps(receipts, sort_keys=True, indent=2) + "\n"
+    manifest_hash = hashlib.sha256(raw_manifest.encode()).hexdigest()
+    receipts_hash = hashlib.sha256(raw_receipts.encode()).hexdigest()
+    return {"schema": "repo-manager.custodian-capture.v1",
+            "snapshot_id": hashlib.sha256((manifest_hash + receipts_hash).encode()).hexdigest(),
+            "manifest": raw_manifest, "manifest_sha256": manifest_hash,
+            "receipts": raw_receipts, "receipts_sha256": receipts_hash, "files": files}

src/repo_manager/runtime.py

@@ -5,6 +5,8 @@ from __future__ import annotations
 import hmac
 import os
 import secrets
+import threading
+import json
 from contextlib import asynccontextmanager
 from dataclasses import dataclass
 from pathlib import Path
@@ -32,6 +34,8 @@ class PublisherSettings:
     forgejo_timeout_seconds: float = 10.0
     forgejo_retries: int = 2
     remote_workers: int = 8
+    custodian_snapshot_policy: Path | None = None
+    custodian_snapshot_hub: str | None = None
 
     @classmethod
     def from_env(cls) -> PublisherSettings:
@@ -51,6 +55,8 @@ class PublisherSettings:
                 raise ValueError("Forgejo token file is empty")
         return cls(
             registry_path=Path(registry) if registry else None,
+            custodian_snapshot_policy=Path(os.environ["REPO_MANAGER_CUSTODIAN_SNAPSHOT_POLICY"]) if os.getenv("REPO_MANAGER_CUSTODIAN_SNAPSHOT_ENABLED") == "true" else None,
+            custodian_snapshot_hub=os.getenv("REPO_MANAGER_CUSTODIAN_SNAPSHOT_HUB"),
             cursor_secret=os.getenv("REPO_MANAGER_CURSOR_SECRET") or secrets.token_urlsafe(32),
             api_token=os.getenv("REPO_MANAGER_API_TOKEN"),
             page_size=int(os.getenv("REPO_MANAGER_PROJECTION_PAGE_SIZE", "100")),
@@ -149,6 +155,34 @@ def create_app(*, settings: PublisherSettings | None = None) -> FastAPI:
                 status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(exc)
             ) from exc
 
+    capture_lock = threading.Lock()
+
+    @app.post("/ports/repositories/custodian-snapshot", tags=["port.repo"])
+    def custodian_snapshot(authorization: str | None = Header(default=None)) -> dict:
+        # No caller paths, refs, repo names, URLs or commands are accepted.
+        if resolved.custodian_snapshot_policy is None or not resolved.custodian_snapshot_hub:
+            raise HTTPException(status_code=404, detail="snapshot publication disabled")
+        if resolved.api_token is not None and not hmac.compare_digest(
+            authorization or "", f"Bearer {resolved.api_token}"
+        ):
+            raise HTTPException(status_code=401, detail="unauthorized")
+        if not capture_lock.acquire(blocking=False):
+            raise HTTPException(status_code=409, detail="capture already running")
+        try:
+            from repo_manager.custodian_snapshot import capture
+            from repo_manager.freshness_runner import RunConfig
+            policy = json.loads(resolved.custodian_snapshot_policy.read_text())
+            return capture(policy, RunConfig(
+                limit=len(policy["repositories"]), max_seconds=240,
+                api_base=resolved.custodian_snapshot_hub,
+                forge_base=resolved.forgejo_base_url or "https://forgejo.coulomb.social",
+                forge_token=resolved.forgejo_token,
+            ))
+        except Exception as exc:
+            raise HTTPException(status_code=503, detail="bounded snapshot capture failed") from exc
+        finally:
+            capture_lock.release()
+
     return app
 
 

tests/test_custodian_snapshot.py

@@ -0,0 +1,65 @@
+import base64
+import hashlib
+from pathlib import Path
+
+import httpx
+import pytest
+from fastapi.testclient import TestClient
+
+from repo_manager.custodian_snapshot import capture
+from repo_manager.freshness_runner import RunConfig
+from repo_manager.runtime import PublisherSettings, create_app
+from test_freshness_runner import A, FakeHub
+
+
+def fixture_transport(tamper=False):
+    def handler(request):
+        path = request.url.path
+        if path.endswith('/example'):
+            return httpx.Response(200, json={'default_branch': 'main'})
+        if '/branches/' in path:
+            return httpx.Response(200, json={'commit': {'id': A}})
+        assert request.url.params['ref'] == A
+        data = b'workplan'
+        sha = hashlib.sha1(b'blob 8\0' + data).hexdigest()
+        if path.endswith('/contents/workplans'):
+            return httpx.Response(200, json=[{'path': 'workplans/task.md', 'type': 'file', 'sha': sha}])
+        if path.endswith('/contents/decisions'):
+            return httpx.Response(404)
+        return httpx.Response(200, json={'type': 'file', 'encoding': 'base64', 'sha': sha,
+                                       'content': base64.b64encode(b'changed' if tamper else data).decode()})
+    return httpx.MockTransport(handler)
+
+
+def test_capture_uses_exact_verified_commits_and_no_local_checkout():
+    policy = {'schema': 'activity-core.custodian-input-policy.v1', 'repositories': ['example'], 'max_snapshot_bytes': 1024}
+    result = capture(policy, RunConfig(limit=1, max_seconds=30, forge_token='fixture'),
+                     hub_transport=httpx.MockTransport(FakeHub([('example', 'active')])),
+                     forge_transport=fixture_transport())
+    assert result['schema'] == 'repo-manager.custodian-capture.v1'
+    assert list(result['files']) == ['example/workplans/task.md']
+    assert base64.b64decode(result['files']['example/workplans/task.md']) == b'workplan'
+    assert hashlib.sha256(result['manifest'].encode()).hexdigest() == result['manifest_sha256']
+
+
+def test_capture_refuses_blob_mismatch():
+    policy = {'schema': 'activity-core.custodian-input-policy.v1', 'repositories': ['example'], 'max_snapshot_bytes': 1024}
+    with pytest.raises(ValueError, match='blob identity'):
+        capture(policy, RunConfig(limit=1, max_seconds=30, forge_token='fixture'),
+                hub_transport=httpx.MockTransport(FakeHub([('example', 'active')])),
+                forge_transport=fixture_transport(tamper=True))
+
+
+def test_snapshot_endpoint_disabled_by_default():
+    with TestClient(create_app(settings=PublisherSettings())) as client:
+        assert client.post('/ports/repositories/custodian-snapshot').status_code == 404
+
+
+def test_snapshot_endpoint_uses_existing_auth_before_capture():
+    settings = PublisherSettings(api_token='fixture', custodian_snapshot_policy=Path('/missing'),
+                                 custodian_snapshot_hub='http://fixture')
+    with TestClient(create_app(settings=settings)) as client:
+        assert client.post('/ports/repositories/custodian-snapshot').status_code == 401
+        result = client.post('/ports/repositories/custodian-snapshot', headers={'Authorization': 'Bearer fixture'})
+        assert result.status_code == 503
+        assert result.json()['detail'] == 'bounded snapshot capture failed'