feat: select retained Custodian snapshot through pinned descriptor

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

k8s/custodian-release/README.md

@@ -9,8 +9,7 @@ infer deployed bytes from Git HEAD. A new image digest is required.
 
 `worker-patch.yaml` is a strategic-merge review artifact preserving existing env,
 volumes and resources. It reuses `actcore-ops-artefacts` with two narrowly scoped
-subPaths: private state and a selected immutable snapshot. Explicit image and
-snapshot placeholders must be replaced with real accepted evidence. Do not apply
+subPaths: private state and a selected immutable snapshot. The explicit image placeholder must be replaced with real accepted evidence. Do not apply
 this file directly. Review its merged diff and render through owner GitOps after
 source/image acceptance. No Kubernetes resources were read or changed here.
 
@@ -35,14 +34,23 @@ object stores, retain the complete matching receipt rows as a JSON array, and
 copy the completed immutable tree to `custodian/snapshots/<id>` on the claim.
 Only workplans/decisions are admitted. Do not fetch credentials into the worker.
 
-The candidate uses a fixed snapshot mount deliberately. It can prove one bounded
-run after acceptance; it is not a recurring publication handoff. Before enabling
-the daily schedule, add the owned pre-run publication/handoff within native
-Activity Core orchestration, or a reviewed serialized mount update with consumers
-quiescent. A static stale snapshot fails the freshness gate; never refresh its
-timestamp. The worker image's packaged Repo Manager version is still the earlier
-freshness CLI pin; the exporter is a publisher-side prerequisite, not falsely
-claimed installed there.
+The candidate mounts the snapshot parent read-only. Each run reads one regular
+`inputs/current.json` descriptor under private state with schema
+`activity-core.custodian-snapshot-selection.v1`, `snapshot_id`,
+`manifest_sha256`, and `receipts_sha256`. It selects only
+`<snapshot-root>/<snapshot_id>` and
+`<state>/inputs/<snapshot_id>.receipts.json`, validates both digests and then the
+full freshness/content contract. Snapshot IDs permit no traversal or symlinks.
+The receiver must finish and retain both artifacts, then atomically replace the
+regular descriptor file. A run keeps its selected immutable paths; subsequent
+runs see the next descriptor without a Deployment change. Never alter selected
+artifacts or prune snapshots referenced by in-flight runs.
+
+Descriptor selection is implemented locally; remote capture/transport is not.
+The packaged Repo Manager version remains the earlier freshness CLI pin; the
+new targeted refresh/export source is a publisher-side prerequisite, not falsely
+claimed installed in that image. See `publisher-access-delta.md` for the exact
+existing producer and proposed network/data-access change.
 
 ## Required preflight and verification
 

k8s/custodian-release/publisher-access-delta.md

@@ -0,0 +1,76 @@
+# Exact publisher and access delta for review
+
+Source inspection, 2026-10-19. No credentials read or runtime changed.
+
+Recommended producer: existing `core-hub-api-repository-publisher:8020` in
+namespace core-hub, owned by Repo Manager source and rapp-core-hub packaging.
+`rapp-core-hub/values/production.yaml` enables it with immutable image
+`sha256:69d00898cfa1ff0a4b7db8cdc86d15f5f4bea4526915019d3ca82b6f5574354c` and
+`repo-manager-forgejo-source` Secret. CCR-2026-0031 is applied: its dedicated
+core-hub-source identity already has read:repository for coulomb repositories,
+restricted to the core-hub workload. Do not copy that Secret or identity into
+Activity Core. The Activity Core Forge reader design explicitly has no admitted
+CCR and is not an existing usable worker credential.
+
+Current Repo Manager `runtime.py` exports classifications only; it has no file
+snapshot endpoint or retained artifact volume. Current rapp-core-hub publisher
+NetworkPolicy admits only candidate pods in core-hub, and egress only to Forgejo
+and DNS. There is no accepted file transport to Activity Core. The owner-return
+pickup adapter consumes operator-supplied C-25 JSON; it is not an artifact
+transfer channel and must not be repurposed by relabelling a snapshot.
+
+## Recommended minimal extension (not implemented or approved by this note)
+
+Add one opt-in endpoint to the existing Repo Manager publisher, for the server's
+fixed `custodian-security-flow` export policy. It must not accept arbitrary
+repository names, paths, URLs, refs, commands or target output locations from
+callers. The configured exact 15-repo policy controls source disclosure. Default
+enabled=false. Require an idempotent per-activity capture ID and bound concurrent
+captures to one, with explicit time/file/byte limits; do not add a scheduler.
+
+For each requested capture, run targeted freshness for every policy repo, then
+read the allowlisted workplan/decision content at those exact Forge commits via
+the existing authenticated Forge transport. Native `rmgr freshness run --repos`
+source now supports exact selection and refuses an undersized limit or missing
+active repo. Rotating disabled limit6/every2h freshness is not sufficient for
+this 15-repo/2h policy. A partial, failed, stale or unknown capture must not be
+published. Existing native receipt trust and no-op issuer semantics are reused.
+
+Return a bounded immutable data bundle with per-file hashes, full existing
+freshness receipt rows and manifest identity. Do not return credentials, source
+archives, unrelated files or arbitrary Forge download URLs. Activity Core receives
+and verifies it under private retained state/snapshot storage, then atomically
+advances its regular descriptor. Use the existing daily workflow's pre-run stage;
+no separate recurrence is introduced. Interrupted/partial reception never updates
+the descriptor. Content transfer and receiver remain required source work, not
+claimed done by offline export or descriptor selection.
+
+Required source/network scope deltas for owner review:
+
+- Add publisher ingress on TCP8020 only from namespace label
+  `kubernetes.io/metadata.name: activity-core` **and** pod label
+  `app.kubernetes.io/name: actcore-worker` in the same peer selector. Preserve
+  existing candidate ingress. Add matching worker egress only if its active
+  policy otherwise denies this destination.
+- Targeted native freshness also requires publisher egress to the existing
+  State Hub service on TCP8000 for primary verification, active registry and
+  per-repo projection reconciliation. This is absent today; admit only that
+  namespace/pod selector after verifying its actual labels. No broad Internet
+  or namespace egress is proposed.
+- NetworkPolicy is L4: it cannot limit a caller to the new endpoint. Endpoint
+  caller admission must use the existing service identity/auth model if available;
+  current publisher API-token setting is optional and not configured in the
+  inspected package. Do not claim a path-scoped grant from port filtering alone.
+  An explicit owner decision must accept worker access to the publisher API or
+  supply an existing authenticated gateway. Do not invent a new token mount.
+- New disclosure is the private **workplan and decision contents of the explicit
+  15 repos** to the existing Activity Core worker, beyond current classification
+  metadata. Existing Forge read scope does not itself approve that downstream
+  disclosure. This is the specific access review, not a generic deployment gate.
+
+No source packaging or mount choice can remove this transport/access prerequisite.
+The existing worker placement remains selected. The next review should accept or
+narrow the exact downstream data/endpoint access above; then implement and test
+capture/receiver before enabling the daily workflow. Private storage capacity,
+source bytes, no extra resource requests, rollback and unattended evidence stay
+as documented in the adjacent release README.

k8s/custodian-release/worker-patch.yaml

@@ -1,5 +1,5 @@
 # REVIEW ONLY: not referenced by k8s/gitops/kustomization.yaml.
-# Replace both explicit placeholders only after image/input evidence is accepted.
+# Replace the explicit image placeholder only after image evidence is accepted.
 apiVersion: apps/v1
 kind: Deployment
 metadata:
@@ -15,7 +15,7 @@ spec:
             - name: ACTIVITY_CORE_CUSTODIAN_ROOT
               value: /opt/custodian-commands
             - name: ACTIVITY_CORE_CUSTODIAN_FLEET_ROOT
-              value: /var/lib/custodian-inputs/fleet
+              value: /var/lib/custodian-inputs/snapshots
             - name: ACTIVITY_CORE_CUSTODIAN_STATE_ROOT
               value: /var/lib/activity-core/custodian
             - name: ACTIVITY_CORE_CUSTODIAN_REPOS
@@ -24,8 +24,8 @@ spec:
               value: http://state-hub.state-hub.svc.cluster.local:8000
             - name: ACTIVITY_CORE_CUSTODIAN_INPUT_POLICY
               value: /opt/custodian-input-policy.json
-            - name: ACTIVITY_CORE_CUSTODIAN_FRESHNESS_RECEIPTS
-              value: /var/lib/activity-core/custodian/inputs/freshness-receipts.json
+            - name: ACTIVITY_CORE_CUSTODIAN_SNAPSHOT_DESCRIPTOR
+              value: /var/lib/activity-core/custodian/inputs/current.json
             - name: ACTIVITY_CORE_CUSTODIAN_ALLOCATION_REQUEST
               value: /var/lib/activity-core/custodian/requests/current.json
           volumeMounts:
@@ -33,8 +33,8 @@ spec:
               mountPath: /var/lib/activity-core/custodian
               subPath: custodian/state
             - name: ops-artefacts
-              mountPath: /var/lib/custodian-inputs/fleet
-              subPath: custodian/snapshots/REQUIRES_ACCEPTED_SNAPSHOT_ID
+              mountPath: /var/lib/custodian-inputs/snapshots
+              subPath: custodian/snapshots
               readOnly: true
 # Existing ops-artefacts volume/claim, UID1000 and resources are preserved.
 # Owner precreates only these private subdirectories; no whole-claim chmod/chown.

src/activity_core/custodian_inputs.py

@@ -1,6 +1,7 @@
 """Fail-closed admission for trusted-owner fleet snapshots and freshness receipts."""
 import hashlib
 import json
+import re
 import stat
 from datetime import UTC, datetime
 from pathlib import Path, PurePosixPath
@@ -79,3 +80,21 @@ def admit(fleet, policy_path, receipts_path, repos, now=None):
     if actual_files != expected_files or total > policy["max_snapshot_bytes"]:
         raise ValueError("snapshot content outside manifest or size policy")
     return hashlib.sha256(raw).hexdigest()
+
+
+def select_snapshot(snapshot_root, state_root, descriptor_path):
+    """Read one protected atomic descriptor; never follow a mutable symlink pointer."""
+    root, state, path = Path(snapshot_root), Path(state_root), Path(descriptor_path)
+    if not path.is_absolute() or ".." in path.parts or state not in path.parents:
+        raise ValueError("snapshot descriptor outside private state")
+    descriptor = json.loads(_read(path))
+    snapshot_id = descriptor.get("snapshot_id", "")
+    if descriptor.get("schema") != "activity-core.custodian-snapshot-selection.v1" or not re.fullmatch(r"[a-zA-Z0-9][a-zA-Z0-9_-]{0,127}", snapshot_id):
+        raise ValueError("invalid snapshot selection")
+    selected = root / snapshot_id
+    manifest_hash = hashlib.sha256(_read(selected / "manifest.json")).hexdigest()
+    receipts = state / "inputs" / (snapshot_id + ".receipts.json")
+    receipts_hash = hashlib.sha256(_read(receipts)).hexdigest()
+    if manifest_hash != descriptor.get("manifest_sha256") or receipts_hash != descriptor.get("receipts_sha256"):
+        raise ValueError("snapshot selection digest mismatch")
+    return selected, receipts

src/activity_core/custodian_operations.py

@@ -84,6 +84,11 @@ def build_command(operation: str, params: dict) -> tuple[list[str], Path, Path,
     argv = [sys.executable, str(script)]
     request = None
     if operation == "custodian_security_flow":
+        from activity_core.custodian_inputs import select_snapshot
+        descriptor = os.environ.get("ACTIVITY_CORE_CUSTODIAN_SNAPSHOT_DESCRIPTOR")
+        selected_receipts = None
+        if descriptor:
+            fleet, selected_receipts = select_snapshot(fleet, state, descriptor)
         repos = os.environ.get("ACTIVITY_CORE_CUSTODIAN_REPOS", "").split(",")
         if not repos or any(re.fullmatch(r"[a-z0-9]+(?:-[a-z0-9]+)*", repo) is None for repo in repos):
             raise ValueError("explicit canonical security-flow repository list required")
@@ -91,7 +96,7 @@ def build_command(operation: str, params: dict) -> tuple[list[str], Path, Path,
             _path(str(fleet / repo / "workplans"), directory=True)
         from activity_core.custodian_inputs import admit
         policy_path = _path(os.environ.get("ACTIVITY_CORE_CUSTODIAN_INPUT_POLICY"))
-        receipts_path = _path(os.environ.get("ACTIVITY_CORE_CUSTODIAN_FRESHNESS_RECEIPTS"))
+        receipts_path = selected_receipts or _path(os.environ.get("ACTIVITY_CORE_CUSTODIAN_FRESHNESS_RECEIPTS"))
         admit(fleet, policy_path, receipts_path, repos)
         hub = os.environ.get("ACTIVITY_CORE_CUSTODIAN_HUB", "")
         parsed = urlsplit(hub)

tests/test_custodian_inputs.py

@@ -91,3 +91,28 @@ def test_fifo_refused_without_open(evidence):
     os.mkfifo(path)
     with pytest.raises(ValueError, match="unsafe"):
         check(evidence)
+
+
+def test_descriptor_pins_snapshot_and_receipts(evidence, tmp_path):
+    from activity_core.custodian_inputs import select_snapshot
+    fleet, policy, receipts = evidence
+    root = tmp_path / "snapshots"
+    root.mkdir()
+    fleet.rename(root / "capture-1")
+    state = tmp_path / "state"
+    (state / "inputs").mkdir(parents=True)
+    retained = state / "inputs/capture-1.receipts.json"
+    retained.write_bytes(receipts.read_bytes())
+    descriptor = state / "inputs/current.json"
+    data = {"schema": "activity-core.custodian-snapshot-selection.v1", "snapshot_id": "capture-1",
+            "manifest_sha256": hashlib.sha256((root / "capture-1/manifest.json").read_bytes()).hexdigest(),
+            "receipts_sha256": hashlib.sha256(retained.read_bytes()).hexdigest()}
+    descriptor.write_text(json.dumps(data))
+    assert select_snapshot(root, state, descriptor) == (root / "capture-1", retained)
+    retained.write_text("[]")
+    with pytest.raises(ValueError, match="digest mismatch"):
+        select_snapshot(root, state, descriptor)
+    data["snapshot_id"] = "../escape"
+    descriptor.write_text(json.dumps(data))
+    with pytest.raises(ValueError, match="invalid snapshot"):
+        select_snapshot(root, state, descriptor)