fix: rank Done retention correctly and prepare guarded fdev archive schedule

.github/workflows/coord-board-archive.yml

@@ -9,7 +9,8 @@ name: Done retention dry run for the Coordination board
 #
 # THIS WORKFLOW ONLY EVER RUNS A DRY RUN. It reports the board size against the 500-item limit, the Done count and
 # the cards the policy would archive. A real run is a separate, owner-run invocation with the private enrollment
-# binding and an exact project id. The schedule gains --execute only after Root accepts the first real run.
+# binding and an exact project id. After Root accepts the first real run, the sibling fdev user timer is the
+# scheduled writer (deployment/fdev-board-tend); this hosted job remains read-only and counts-only.
 #
 # WHY THE PREVIOUS WORKFLOW AT THIS PATH WAS RETIRED (Root, 2026-10-31)
 # =====================================================================

deployment/fdev-board-tend/fsgg-board-archive-run.py

@@ -0,0 +1,259 @@
+#!/usr/bin/python3 -I
+"""Private fdev Done-retention adapter (#5170). Never enables itself or retries an unresolved run.
+
+The existing archiver owns native reads, per-card writer exclusion, intent/send/readback and policy.
+This adapter pins its activation inputs and preserves a cross-run latch before invoking it.
+"""
+from __future__ import annotations
+
+import argparse
+import datetime as dt
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import signal
+import stat
+import subprocess
+import sys
+import tempfile
+
+PROJECT = "PVT_kwDOEYAWY84Bldpa"
+SCHEMA = "fsgg.board-archive-config/1"
+FIELDS = {"schema", "enabled", "maxArchive", "script", "gh", "binding", "commitment", "readerVersion",
+          "readerFiles", "acceptedReceipt", "stateDir", "outDir", "workdir"}
+
+
+class Refused(Exception):
+    pass
+
+
+def private_file(path: Path) -> bytes:
+    info = path.lstat()
+    if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o600:
+        raise Refused("private input must be an owner-owned regular 0600 file")
+    return path.read_bytes()
+
+
+def absolute(text: object) -> Path:
+    if not isinstance(text, str) or not Path(text).is_absolute():
+        raise Refused("absolute path required")
+    return Path(text)
+
+
+def pin(value: object) -> tuple[Path, str]:
+    if not isinstance(value, dict) or set(value) != {"path", "sha256"}:
+        raise Refused("invalid file pin")
+    path = absolute(value["path"])
+    digest = value["sha256"]
+    if not isinstance(digest, str) or not re.fullmatch(r"[0-9a-f]{64}", digest):
+        raise Refused("invalid file digest")
+    info = path.lstat()
+    if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or info.st_mode & 0o022:
+        raise Refused("pinned file must be owner-owned, regular and not group/world writable")
+    if hashlib.sha256(path.read_bytes()).hexdigest() != digest:
+        raise Refused("pinned file changed; Root must refresh activation")
+    return path, digest
+
+
+def private_dir(path: Path) -> Path:
+    # Anchor every ancestor, including entries left by an earlier failed fsync. Creating parents=True and
+    # syncing only the leaf would not make the leaf's entry in its parent durable after a first-run crash.
+    for directory in reversed((path, *path.parents)):
+        directory.mkdir(exist_ok=True, mode=0o700)
+        if not stat.S_ISDIR(directory.lstat().st_mode):
+            raise Refused("directory ancestry must not contain symlinks or non-directories")
+        sync_dir(directory)
+        if directory.parent != directory:
+            sync_dir(directory.parent)
+    info = path.lstat()
+    if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid() or stat.S_IMODE(info.st_mode) != 0o700:
+        raise Refused("output/state directory must be owner-owned, regular and 0700")
+    return path
+
+
+def sync_dir(path: Path) -> None:
+    fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY)
+    try:
+        os.fsync(fd)
+    finally:
+        os.close(fd)
+
+
+def exclusive_json(path: Path, data: dict) -> None:
+    fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY | os.O_NOFOLLOW, 0o600)
+    with os.fdopen(fd, "w") as handle:
+        json.dump(data, handle, sort_keys=True)
+        handle.write("\n")
+        handle.flush()
+        os.fsync(handle.fileno())
+    sync_dir(path.parent)
+
+
+def configuration(path: Path) -> tuple[dict, dict[str, Path]]:
+    c = json.loads(private_file(path))
+    if not isinstance(c, dict) or set(c) != FIELDS or c["schema"] != SCHEMA or c["enabled"] is not True:
+        raise Refused("missing, disabled or invalid archive activation")
+    if type(c["maxArchive"]) is not int or not 1 <= c["maxArchive"] <= 500:
+        raise Refused("Root must supply a positive archive bound at most 500")
+    version = c["readerVersion"]
+    if not isinstance(version, str) or not re.fullmatch(r"\d+\.\d+\.\d+", version) or tuple(map(int, version.split("."))) < (0, 114, 1):
+        raise Refused("qualified archived-membership reader required")
+    paths = {name: pin(c[name])[0] for name in ("script", "gh", "binding", "commitment", "acceptedReceipt")}
+    private_file(paths["binding"])
+    binding = json.loads(paths["binding"].read_bytes())
+    selected = binding.get("selectedIssues") if isinstance(binding, dict) else None
+    if not isinstance(selected, dict) or not selected:
+        raise Refused("invalid enrollment binding")
+    if not isinstance(c["readerFiles"], list) or not c["readerFiles"]:
+        raise Refused("qualified installed reader file pins required")
+    for value in c["readerFiles"]:
+        pin(value)
+    receipt = json.loads(private_file(paths["acceptedReceipt"]))
+    expected = {"projectId": PROJECT, "keep": 200, "maxArchive": c["maxArchive"], "readerVersion": version,
+                "readerFiles": c["readerFiles"], "scriptSha256": c["script"]["sha256"],
+                "bindingSha256": c["binding"]["sha256"], "commitmentSha256": c["commitment"]["sha256"]}
+    if (not isinstance(receipt, dict) or receipt.get("schema") != "fsgg.board-archive-accepted/1"
+            or receipt.get("accepted") is not True or receipt.get("inputs") != expected
+            or not re.fullmatch(r"https://github\.com/FS-GG/\.github/issues/5170#issuecomment-\d+", str(receipt.get("evidenceUrl")))):
+        raise Refused("Root's accepted first-real execution must match activation inputs")
+    for name in ("stateDir", "outDir", "workdir"):
+        paths[name] = absolute(c[name])
+    if Path.cwd().resolve() != paths["workdir"].resolve() or (paths["workdir"] / ".git").exists():
+        raise Refused("fixed non-checkout working directory required")
+    runtime = os.environ.get("XDG_RUNTIME_DIR", "")
+    if not runtime or not Path(runtime).is_absolute():
+        raise Refused("shared XDG_RUNTIME_DIR required")
+    return c, paths
+
+
+def settled(run: Path, max_archive: int) -> bool:
+    custody = run / "custody" / "custody.jsonl"
+    if not custody.exists() or custody.stat().st_size == 0:
+        return (run / "stdout").read_text().rstrip().endswith("nothing to archive.")
+    rows = [json.loads(line) for line in custody.read_text().splitlines()]
+    common = {"schema", "at", "step"}
+    identity = {"itemId", "contentId", "repo", "number", "closedAt", "enrolled"}
+    def timestamp(value: object) -> bool:
+        try:
+            return isinstance(value, str) and dt.datetime.fromisoformat(value).tzinfo is not None
+        except ValueError:
+            return False
+    def integer(value: object) -> bool:
+        return type(value) is int and value >= 0
+    for row in rows:
+        if (not isinstance(row, dict) or not common <= row.keys()
+                or row.get("schema") != "fsgg.board-done-retention-custody/1"
+                or not timestamp(row.get("at"))):
+            return False
+    if len(rows) < 2:
+        return False
+    start, end = rows[0], rows[-1]
+    if (set(start) != common | {"project", "board", "done", "keep", "planned", "maxArchive"}
+            or start["step"] != "run" or start["project"] != PROJECT or start["keep"] != 200
+            or type(start["keep"]) is not int or start["maxArchive"] != max_archive
+            or not all(integer(start[k]) for k in ("board", "done", "planned", "maxArchive"))
+            or not 0 < start["planned"] <= start["done"] - 200 or not 200 < start["done"] <= start["board"] <= 500
+            or not 1 <= max_archive <= 500
+            or set(end) != common | {"archived", "skipped"} or end["step"] != "finished"
+            or not all(integer(end[k]) for k in ("archived", "skipped"))):
+        return False
+    pending = None
+    seen_items, seen_content = set(), set()
+    counts = {"archived": 0, "skipped": 0}
+    for row in rows[1:-1]:
+        step = row["step"]
+        keys = common | identity | ({"reason"} if step == "skipped" else set())
+        if (step not in ("intent", "archived", "skipped") or set(row) != keys
+                or any(not isinstance(row[k], str) or not row[k].strip() for k in ("itemId", "contentId", "repo"))
+                or not re.fullmatch(r"[^/\s]+/[^/\s]+", row["repo"])
+                or type(row["number"]) is not int or row["number"] < 1
+                or type(row["enrolled"]) is not bool or not timestamp(row["closedAt"])
+                or (step == "skipped" and (not isinstance(row["reason"], str) or not row["reason"].strip()))):
+            return False
+        ref = {key: row[key] for key in identity}
+        if pending is not None:
+            if step != "archived" or ref != pending:
+                return False
+            counts["archived"] += 1
+            pending = None
+        else:
+            if step == "archived" or row["itemId"] in seen_items or row["contentId"] in seen_content:
+                return False
+            seen_items.add(row["itemId"]); seen_content.add(row["contentId"])
+            if step == "intent":
+                pending = ref
+            else:
+                counts["skipped"] += 1
+    return (pending is None and sum(counts.values()) == min(start["planned"], max_archive)
+            and all(end[k] == count for k, count in counts.items()))
+
+
+def run(config: Path, credentials: Path, timeout: float = 720) -> int:
+    try:
+        c, p = configuration(config)
+        secrets = json.loads(private_file(credentials))
+        token = secrets.get("GH_TOKEN") if isinstance(secrets, dict) else None
+        if not isinstance(token, str) or not token.strip():
+            raise Refused("credential file has no GH_TOKEN")
+        state = private_dir(p["stateDir"])
+        latch = state / "in-progress.json"
+        if latch.exists() or latch.is_symlink():
+            raise Refused("unresolved prior run; Root readback required before another invocation")
+        out = private_dir(p["outDir"])
+        stamp = dt.datetime.now(dt.timezone.utc).strftime("%Y%m%dT%H%M%SZ-")
+        directory = Path(tempfile.mkdtemp(prefix=stamp, dir=out))
+        sync_dir(directory)
+        sync_dir(out)
+        exclusive_json(latch, {"run": str(directory), "configSha256": hashlib.sha256(config.read_bytes()).hexdigest()})
+        # Preprovision the child's append-only file, including its directory entry, before any possible send.
+        custody = private_dir(directory / "custody")
+        fd = os.open(custody / "custody.jsonl", os.O_CREAT | os.O_EXCL | os.O_WRONLY | os.O_NOFOLLOW, 0o600)
+        try:
+            os.fsync(fd)
+        finally:
+            os.close(fd)
+        sync_dir(custody)
+        env = {key: os.environ[key] for key in ("HOME", "LANG", "LC_ALL", "XDG_RUNTIME_DIR", "DOTNET_ROOT") if key in os.environ}
+        env.update(PATH="/usr/local/bin:/usr/bin:/bin", GH_TOKEN=token.strip(), COORD_ARCHIVE_GH=str(p["gh"]))
+        command = ["/usr/bin/python3", "-I", str(p["script"]), "--owner", "FS-GG", "--project", "3",
+                   "--expect-project-id", PROJECT, "--keep", "200", "--max-archive", str(c["maxArchive"]),
+                   "--enrollment-binding", str(p["binding"]), "--reader-version", c["readerVersion"],
+                   "--custody-dir", str(directory / "custody"), "--execute"]
+        with os.fdopen(os.open(directory / "stdout", os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "w") as stdout, \
+                os.fdopen(os.open(directory / "stderr", os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "w") as stderr:
+            os.fsync(stdout.fileno())
+            os.fsync(stderr.fileno())
+            sync_dir(directory)
+            process = subprocess.Popen(command, env=env, stdout=stdout, stderr=stderr, start_new_session=True, umask=0o077)
+            try:
+                code = process.wait(timeout=timeout)
+            except subprocess.TimeoutExpired:
+                os.killpg(process.pid, signal.SIGKILL)
+                process.wait()
+                code = 124
+        if code != 0 or not settled(directory, c["maxArchive"]):
+            print("board-archive: stopped; private custody retained, Root readback required", file=sys.stderr)
+            return 3
+        exclusive_json(directory / "completed.json", {"exitCode": code, "settled": True})
+        latch.unlink()
+        sync_dir(state)
+        print("board-archive: completed; private custody retained")
+        return 0
+    except (Refused, OSError, ValueError, KeyError, TypeError):
+        # Never emit paths, card identities, tokens or exception text to the journal.
+        print("board-archive: refused; check private activation inputs or unresolved custody", file=sys.stderr)
+        return 2
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--config", type=Path, required=True)
+    parser.add_argument("--credentials", type=Path, required=True)
+    args = parser.parse_args()
+    return run(args.config, args.credentials)
+
+
+if __name__ == "__main__":
+    sys.exit(main())

deployment/fdev-board-tend/fsgg-board-archive.service.in

@@ -0,0 +1,24 @@
+# Rendered by install.py archive; installation does not enable or start this unit (#5170).
+[Unit]
+Description=FS-GG Project3 guarded Done retention
+
+[Service]
+Type=exec
+WorkingDirectory=@WORKDIR@
+Environment=XDG_RUNTIME_DIR=%t
+Environment=XDG_CONFIG_HOME=%h/.config
+Environment=XDG_STATE_HOME=%h/.local/state
+RuntimeDirectory=fsgg
+RuntimeDirectoryMode=0700
+RuntimeDirectoryPreserve=yes
+UMask=0077
+NoNewPrivileges=yes
+# No PrivateTmp: the archiver shares the literal /tmp legacy writer locks with the installed engine.
+CPUQuota=100%
+MemoryMax=1G
+MemorySwapMax=0
+RuntimeMaxSec=900
+TimeoutStopSec=30
+KillMode=control-group
+# No Restart: an incomplete run retains the adapter's durable latch for Root readback.
+ExecStart=/usr/bin/python3 -I @RELEASE_DIR@/fsgg-board-archive-run.py --config @CONFIG@ --credentials @CREDENTIALS@

deployment/fdev-board-tend/fsgg-board-archive.timer.in

@@ -0,0 +1,10 @@
+[Unit]
+Description=Daily FS-GG Project3 guarded Done retention
+
+[Timer]
+OnCalendar=*-*-* 04:43:00 UTC
+Persistent=true
+Unit=fsgg-board-archive.service
+
+[Install]
+WantedBy=timers.target

deployment/fdev-board-tend/install.py

@@ -1,7 +1,7 @@
 #!/usr/bin/python3 -I
 """Install the fdev board-tend units (board-reconciler design section 8, #5136 PR 8). User-level only.
 
-Five subcommands, each writing one kind of thing and nothing else. Every one has --dry-run, which prints what it
+Six subcommands, each writing one kind of thing and nothing else. Every one has --dry-run, which prints what it
 would write and changes nothing. None needs or uses root; every destination must be under $HOME.
 
   units        Renders fsgg-board-tend.{service,timer} and fsgg-board-tend-daily.{service,timer} into the user
@@ -16,6 +16,8 @@
                the script with --apply. Without --enable it does not touch systemd.
   chromium-env Copies deployment/fdev-chromium/fdev-chromium-env to ~/.local/bin (the source of truth for the
                installed wrapper, which sets the short disk-backed TMPDIR default, .github#5240).
+  archive      Stages an immutable archive adapter/script and two daily user units. Never enables or starts
+               them; Root's accepted first execution and private activation configuration remain mandatory.
 
 The engine and the kit root are pinned to one release: the engine must live in .../coord-<version>/ and the kit
 root must be named kit-<version>.
@@ -286,6 +288,29 @@ def command_enable_file(args: argparse.Namespace) -> int:
     return 0
 
 
+def command_archive(args: argparse.Namespace) -> int:
+    if os.geteuid() == 0:
+        raise Refusal("run as the fdev user, not root")
+    release = user_path(args.release_dir, "--release-dir")
+    workdir = user_path(args.workdir, "--workdir")
+    values = {"RELEASE_DIR": str(release), "WORKDIR": str(workdir),
+              "CONFIG": str(user_path(args.config, "--config")),
+              "CREDENTIALS": str(user_path(args.credentials, "--credentials"))}
+    unit_dir = user_path(args.unit_dir, "--unit-dir")
+    plan = Plan(args.dry_run)
+    # Release files are immutable: --force can replace rendered units, never a pinned release's bytes.
+    for name, source in (("fsgg-board-archive-run.py", HERE / "fsgg-board-archive-run.py"),
+                         ("coord-board-archive.py", HERE.parents[1] / "scripts/coord-board-archive.py")):
+        plan.add(release / name, source.read_bytes(), 0o400, False)
+    for name in ("fsgg-board-archive.service", "fsgg-board-archive.timer"):
+        plan.add(unit_dir / name, render((HERE / (name + ".in")).read_text(), values).encode(), 0o644, args.force)
+    plan.apply()
+    if not args.dry_run:
+        workdir.mkdir(parents=True, exist_ok=True, mode=0o700)
+    print("archive staged disabled; systemd not touched; Root owns first-real acceptance and activation")
+    return 0
+
+
 def parser() -> argparse.ArgumentParser:
     h = home()
     root = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
@@ -324,14 +349,23 @@ def parser() -> argparse.ArgumentParser:
     chromium.add_argument("--dest", default=str(h / ".local/bin/fdev-chromium-env"))
     chromium.add_argument("--force", action="store_true", help="replace the installed wrapper when it differs")
     chromium.add_argument("--dry-run", action="store_true")
+    archive = sub.add_parser("archive")
+    archive.add_argument("--release-dir", required=True, help="a new immutable, versioned directory under HOME")
+    archive.add_argument("--workdir", default=str(h / ".local/share/fs-gg/board-archive"))
+    archive.add_argument("--config", default=str(h / ".config/fsgg/board-archive.json"))
+    archive.add_argument("--credentials", default=str(h / ".config/fdev/credentials.json"))
+    archive.add_argument("--unit-dir", default=str(h / ".config/systemd/user"))
+    archive.add_argument("--force", action="store_true")
+    archive.add_argument("--dry-run", action="store_true")
     return root
 
 
 def main(argv: list[str] | None = None) -> int:
     args = parser().parse_args(argv)
     try:
         return {"units": command_units, "kit-root": command_kit_root, "enable-file": command_enable_file,
-                "tmp-cleanup": command_tmp_cleanup, "chromium-env": command_chromium_env}[args.command](args)
+                "tmp-cleanup": command_tmp_cleanup, "chromium-env": command_chromium_env,
+                "archive": command_archive}[args.command](args)
     except Refusal as error:
         print(f"install.py: refused: {error}", file=sys.stderr)
         return 2