coulomb/repo-manager
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: export pinned offline fleet work-record inputs

Browse source 
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a15374-f59f-7a31-86d2-d23b5d2ed181
This commit is contained in:
tegwick 2026-10-19 14:45:11 +02:00
parent f758e277ab
commit 3e2726f5f1
4 changed files with 344 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

11
SCOPE.md
View file

@ -98,3 +98,14 @@ retain their own generators. It makes no schedule or public-release change.
disk for offline use and is root-ignored rather than versioned. Regenerate it
with the State Hub brief renderer; workplans and other authored repository
records remain authoritative. Brief refreshes must not create Git commits.
## Offline pinned work-record export
`python -m repo_manager.fleet_snapshot` exports only workplan and decision
Markdown from explicitly pinned local Git commits into a new completion-marked
directory with per-file hashes. It performs no fetch, checkout, Hub write,
scheduling or pruning and does not assert freshness. See
`docs/custodian-fleet-input-contract.md` for the proposed owner publication,
consumer admission and retention contract. Runtime publication, freshness
enforcement and consumer acceptance remain unimplemented.

127
docs/custodian-fleet-input-contract.md Normal file
View file

@ -0,0 +1,127 @@
# Custodian fleet input publication contract
Owner recommendation, 2026-10-19; source preparation only. Repo Manager owns
commit selection and work-record publication. Activity Core owns scheduling,
private consumer storage, admission and run evidence. Custodian owns daily report
semantics. These are proposed defaults, not runtime acceptance or an activation.
## Reuse and exact missing boundary
`rmgr freshness run --limit N --max-seconds S` already observes Forge default HEAD,
verifies the primary Hub projection and writes durable
`repo-manager.freshness-receipt.v1` receipts. It explicitly makes no checkout.
The classification publisher's snapshots contain metadata, not work-record files.
Neither command currently supplies a filesystem tree to Custodian commands.
Do not reinterpret Hub rows or a classification snapshot as those input files.
The new offline prerequisite is:
```
python -m repo_manager.fleet_snapshot --request selected-commits.json --output snapshots/<unique-run-id>
```
It reads existing local Git objects, never fetches, checks out, changes a repo,
contacts Hub/Forge, or reads a credential. This is a module CLI rather than a
new daemon or scheduler. The publication owner supplies object stores using its
existing approved Git delivery path. Choosing that runtime source/access path
is still required; the exporter itself needs no network or authentication.
Request contract:
```json
{
"observed_at": "2026-10-19T07:00:00+02:00",
"repositories": [
{"slug": "the-custodian", "path": "/accepted/local/object-store/the-custodian", "commit": "<full-40-character-commit>"}
]
}
```
The example commit is a placeholder, deliberately not runnable. The publisher
must retain each matching Forge observation/receipt ID separately; `observed_at`
is supplied evidence, not a freshness assertion made by the offline exporter.
Only `workplans/**/*.md` (including archived dependencies) and `decisions/*.md`
are copied. Git symlinks, submodules at selected record paths, invalid slugs,
non-full commit IDs, absent workplans and oversize records are refused. Dirty
working files, .git, credentials and unrelated files are not copied. Limits are
2MiB per record and 64MiB total per export; exceeding either is an explicit
publication failure, never partial accepted coverage.
Each new unique output contains `<slug>/workplans`, optional `<slug>/decisions`,
`manifest.json` and finally `COMPLETE`. Manifest schema
`repo-manager.fleet-input.v1` records the requested observation time, all commits,
file byte lengths and SHA-256 values. `freshness_verified: false` is intentional.
`COMPLETE` contains the manifest SHA-256. Consumers must validate this marker,
manifest and every file before accepting a publication. Failed promotion leaves
no COMPLETE and must never be consumed. Names are never overwritten, even empty
existing directories. The owner protects the output parent against concurrent
untrusted writers. This is completion-marked publication, not a crash-durable
transaction or automatically enforced read-only store. Do not change published
files; mount the selected immutable directory read-only for the consumer.
## Recommended default daily policy
Initial scope is the existing report estate: the-custodian, net-kingdom,
railiance-platform, activity-core, access-engine, secrets-engine, ops-mason,
ops-warden, key-cape, user-engine, tenant-engine, gate-house, kings-guard,
policy-nexus and rapp-policy-nexus. This is a proposed explicit allowlist, not
permission to publish every registered repo. The same selected set bounds
both workplan reporting and decision synchronization. Review additions for
cross-repo dependency completeness; missing dependencies remain unknown.
Recommend one pre-run publication step in the existing native daily workflow,
not a separate timer. Reuse Repo Manager's Forge HEAD/primary receipt verification
for the selected set, then export those exact commits from accepted object stores.
Existing `freshness run` rotates the stalest fleet members and has no explicit
repo-set option: its successful invocation alone does not establish this set's
coverage. The publisher must join per-repo receipts and require every selected
repo; no fabricated receipt or broad forced reconciliation is acceptable.
Recommended admission: observations at most 2 hours old at daily start, no future
observation beyond 5 minutes of clock tolerance, all required repos present,
full commits matching the verified observations, exact file hashes, and COMPLETE
valid. Old commit age is not failure when its HEAD was recently re-observed.
Never rewrite an old snapshot's timestamp to make it fresh. On missing/stale
inputs, fail before `security_flow_daily.py --apply`, retain a sanitized missing
repo/reason receipt and notify through the existing Activity Core failure path.
Do not silently reuse yesterday's tree. These consumer freshness checks and the
publication step are **not yet integrated** into the bounded executor; its current
path existence checks are insufficient for activating this contract.
Hold a specific snapshot path throughout a run and its retries. Avoid `current`
symlinks (executor rejects them) and in-place directory replacement. Placement
must supply an immutable per-run path or a serialized binding update while
consumers are quiescent. A static worker env mount cannot advance per-run inputs
by itself; runtime owner must implement this handoff before activation.
## Retention and weekly boundary
Recommend retaining 35 days of daily publications plus every snapshot referenced
by an unresolved failure, checkpoint, audit or accepted recommendation. Capacity
planning must use measured export sizes, not assume the shared 1Gi claim suffices.
Pruning is owner-reviewed initially; this exporter performs no deletion. Keep
manifest/receipt hashes with run evidence and back up private state separately.
Restore testing and capacity alarms are delivery gates, not established here.
Weekly sizing uses the existing cluster, resource-control, telemetry, inventory,
provenance and optional previous-report/review inputs. Repo Manager must not
manufacture telemetry or impersonate those collectors. Recommend retaining at
least 13 weekly bundles plus all referenced/open recommendation evidence, with
evaluation windows preserved. Custodian owns the weekly request and collector
coverage validation; Activity Core owns immutable request handoff and retained
report storage. Daily Git snapshots do not satisfy weekly measurement coverage.
## Owner handoff and validation
Repo Manager recommendation: use this bounded offline export after the existing
Forge receipt checks, with the explicit 15-repo set and 2-hour admission default.
Activity Core should add the manifest/freshness gate and per-run immutable binding
when implementing the chosen worker placement; never loosen symlink validation.
No public API, broad clone service, extra schedule or new credential path is
needed. Production receipt verification, object-store provisioning, consumer gate,
retention/storage acceptance and first scheduled evidence remain unfinished.
Local tests use temporary Git repositories and prove committed-byte export despite
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.

114
src/repo_manager/fleet_snapshot.py Normal file
View file

@ -0,0 +1,114 @@
"""Offline, immutable work-record export from explicitly pinned local Git objects.
No fetch, checkout, Hub calls, scheduling, or freshness assertion. See
``docs/custodian-fleet-input-contract.md`` for publisher/consumer obligations.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import os
import re
import shutil
import subprocess
import tempfile
from datetime import datetime
from pathlib import Path, PurePosixPath
SCHEMA = "repo-manager.fleet-input.v1"
MAX_FILE_BYTES = 2 * 1024 * 1024
MAX_TOTAL_BYTES = 64 * 1024 * 1024
def _git(repo: Path, *args: str) -> bytes:
result = subprocess.run(
["git", "-C", str(repo), *args], capture_output=True, timeout=30, check=False,
env={**os.environ, "GIT_NO_REPLACE_OBJECTS": "1"},
)
if result.returncode:
raise ValueError("pinned Git object unavailable")
return result.stdout
def export_snapshot(request: dict, output: Path) -> dict:
"""Caller owns/protects parent directory; output must be a new immutable path."""
stamp = datetime.fromisoformat(request["observed_at"])
if stamp.tzinfo is None:
raise ValueError("observation must have timezone")
repos = request["repositories"]
if not isinstance(repos, list) or not repos:
raise ValueError("explicit nonempty repository selection required")
output = output.absolute()
if output.exists() or output.is_symlink():
raise ValueError("output already exists")
if any(p.is_symlink() for p in [output.parent, *output.parents]):
raise ValueError("output parent must not traverse symlinks")
manifest = {"schema": SCHEMA, "observed_at": request["observed_at"],
"freshness_verified": False, "repositories": []}
total = 0
seen = set()
temporary = Path(tempfile.mkdtemp(prefix=".fleet-input-", dir=output.parent))
try:
for item in repos:
slug, commit = item["slug"], item["commit"]
if not re.fullmatch(r"[a-z0-9]+(?:-[a-z0-9]+)*", slug) or slug in seen:
raise ValueError("invalid or repeated repository slug")
if not re.fullmatch(r"[0-9a-f]{40}", commit):
raise ValueError("full commit required")
seen.add(slug)
repo = Path(item["path"])
if _git(repo, "rev-parse", commit + "^{commit}").decode().strip() != commit:
raise ValueError("object is not the specified commit")
records = []
(temporary / slug / "workplans").mkdir(parents=True)
for entry in _git(repo, "ls-tree", "-rz", commit).split(b"\0"):
if not entry:
continue
metadata, raw_path = entry.split(b"\t", 1)
mode, kind, oid = metadata.decode().split()
name = raw_path.decode("utf-8")
parts = PurePosixPath(name).parts
allowed = (len(parts) >= 2 and parts[0] == "workplans" or
len(parts) == 2 and parts[0] == "decisions") and name.endswith(".md")
if not allowed:
continue
if ".." in parts or name.startswith("/") or mode not in {"100644", "100755"} or kind != "blob":
raise ValueError("unsafe record entry")
size = int(_git(repo, "cat-file", "-s", oid))
total += size
if size > MAX_FILE_BYTES or total > MAX_TOTAL_BYTES:
raise ValueError("snapshot size limit exceeded")
data = _git(repo, "cat-file", "blob", oid)
target = temporary / slug / name
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(data)
records.append({"path": name, "sha256": hashlib.sha256(data).hexdigest(),
"bytes": len(data)})
if not any(r["path"].startswith("workplans/") for r in records):
raise ValueError("selected repository has no workplan records")
manifest["repositories"].append({"slug": slug, "commit": commit, "files": records})
manifest["repositories"].sort(key=lambda row: row["slug"])
encoded = json.dumps(manifest, sort_keys=True, indent=2) + "\n"
(temporary / "manifest.json").write_text(encoded)
# Refuse existing names, including empty directories; reserve before moving files.
output.mkdir(mode=0o700)
# No COMPLETE means unpublished; preserve incomplete evidence for owner recovery.
for child in temporary.iterdir():
child.rename(output / child.name)
(output / "COMPLETE").write_text(hashlib.sha256(encoded.encode()).hexdigest() + "\n")
return manifest
finally:
shutil.rmtree(temporary)
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--request", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
args = parser.parse_args()
export_snapshot(json.loads(args.request.read_text()), args.output)
if __name__ == "__main__":
main()

92
tests/test_fleet_snapshot.py Normal file
View file

@ -0,0 +1,92 @@
import hashlib
import json
import subprocess
import pytest
from repo_manager.fleet_snapshot import export_snapshot
def git(repo, *args):
return subprocess.check_output(["git", "-C", str(repo), *args]).decode().strip()
@pytest.fixture
def source(tmp_path):
repo = tmp_path / "source"
repo.mkdir()
git(repo, "init", "-q")
git(repo, "config", "user.name", "Fixture")
git(repo, "config", "user.email", "fixture@example.invalid")
for path, content in {"workplans/active.md": "committed work", "decisions/one.md": "accepted",
"private.txt": "excluded", "workplans/archived/old.md": "done"}.items():
file = repo / path
file.parent.mkdir(parents=True, exist_ok=True)
file.write_text(content)
git(repo, "add", ".")
git(repo, "commit", "-qm", "fixture")
return repo
def request(source):
return {"observed_at": "2026-10-19T10:00:00Z", "repositories": [
{"slug": "example", "path": str(source), "commit": git(source, "rev-parse", "HEAD")}]}
def test_exact_committed_bytes_allowlist_and_manifest(source, tmp_path):
req = request(source)
(source / "workplans/active.md").write_text("dirty excluded")
out = tmp_path / "snapshot"
manifest = export_snapshot(req, out)
assert (out / "example/workplans/active.md").read_text() == "committed work"
assert not (out / "example/private.txt").exists()
assert not (out / "example/.git").exists()
assert len(manifest["repositories"][0]["files"]) == 3
assert manifest["freshness_verified"] is False
assert (out / "COMPLETE").read_text().strip() == hashlib.sha256((out / "manifest.json").read_bytes()).hexdigest()
assert json.loads((out / "manifest.json").read_text()) == manifest
def test_no_overwrite(source, tmp_path):
out = tmp_path / "snapshot"
out.mkdir()
with pytest.raises(ValueError, match="exists"):
export_snapshot(request(source), out)
assert not list(out.iterdir())
@pytest.mark.parametrize("change", ["slug", "commit", "timestamp", "duplicate"])
def test_invalid_request_is_not_published(source, tmp_path, change):
req = request(source)
if change == "slug":
req["repositories"][0]["slug"] = "../outside"
elif change == "commit":
req["repositories"][0]["commit"] = "HEAD"
elif change == "timestamp":
req["observed_at"] = "2026-10-19T10:00:00"
else:
req["repositories"].append(dict(req["repositories"][0]))
with pytest.raises(ValueError):
export_snapshot(req, tmp_path / "snapshot")
assert not (tmp_path / "snapshot").exists()
def test_record_symlink_refused(source, tmp_path):
(source / "workplans/link.md").symlink_to("../private.txt")
git(source, "add", ".")
git(source, "commit", "-qm", "symlink")
with pytest.raises(ValueError, match="unsafe"):
export_snapshot(request(source), tmp_path / "snapshot")
assert not (tmp_path / "snapshot").exists()
def test_symlink_parent_refused(source, tmp_path):
(tmp_path / "alias").symlink_to(source, target_is_directory=True)
with pytest.raises(ValueError, match="symlinks"):
export_snapshot(request(source), tmp_path / "alias/snapshot")
def test_size_limit_refused(source, tmp_path, monkeypatch):
monkeypatch.setattr("repo_manager.fleet_snapshot.MAX_FILE_BYTES", 1)
with pytest.raises(ValueError, match="size"):
export_snapshot(request(source), tmp_path / "snapshot")