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
+subPaths: private state and a selected immutable snapshot. The explicit image placeholder must be replaced with real accepted evidence. Do not apply
-snapshot placeholders 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
+The candidate mounts the snapshot parent read-only. Each run reads one regular
-run after acceptance; it is not a recurring publication handoff. Before enabling
+`inputs/current.json` descriptor under private state with schema
-the daily schedule, add the owned pre-run publication/handoff within native
+`activity-core.custodian-snapshot-selection.v1`, `snapshot_id`,
-Activity Core orchestration, or a reviewed serialized mount update with consumers
+`manifest_sha256`, and `receipts_sha256`. It selects only
-quiescent. A static stale snapshot fails the freshness gate; never refresh its
+`<snapshot-root>/<snapshot_id>` and
-timestamp. The worker image's packaged Repo Manager version is still the earlier
+`<state>/inputs/<snapshot_id>.receipts.json`, validates both digests and then the
-freshness CLI pin; the exporter is a publisher-side prerequisite, not falsely
+full freshness/content contract. Snapshot IDs permit no traversal or symlinks.
-claimed installed there.
+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
+            - name: ACTIVITY_CORE_CUSTODIAN_SNAPSHOT_DESCRIPTOR
-              value: /var/lib/activity-core/custodian/inputs/freshness-receipts.json
+              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
+              mountPath: /var/lib/custodian-inputs/snapshots
-              subPath: custodian/snapshots/REQUIRES_ACCEPTED_SNAPSHOT_ID
+              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)