Add assembly-checkpoint approach (ASSEMBLY-WP-0010)

Executes activity-core assembly-checkpoint-* runs: fast-forward the
assembly-economics checkout, run its deterministic checkpoints/run_checkpoint.py
(data via web-fetcher, optional llm-connect assessment), commit only
checkpoints/, push fast-forward (rebase once, never force; publish grant
limited to checkpoints/), post assembly_checkpoint. Out-of-window fires are
no-ops (checkpoints/schedule.md is authoritative). Selected by label
'assembly-checkpoint'. Founder-approved 2026-10-12.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 1494246@bnt-lap001
Assistant-Session: caf6de6f-f9e4-47ce-a584-49baa271fa4b

rein_aharness/approaches.py

@@ -30,6 +30,7 @@ from rein_aharness.taskspec import TaskSpecError
 
 # Approach command names (stable; used in metrics + ops_run.result)
 APPROACH_FI_RESEARCH_BRIEF = "fi-research-brief"
+APPROACH_ASSEMBLY_CHECKPOINT = "assembly-checkpoint"
 APPROACH_BRIEF_DAILY = "brief-daily"
 APPROACH_BRIEF_WEEKLY = "brief-weekly"
 APPROACH_MAIL_SCAN = "mail-scan"
@@ -69,6 +70,12 @@ class ApproachRule:
 
 # Ordered: first match wins. Cheapest / most specific first.
 APPROACH_RULES: tuple[ApproachRule, ...] = (
+    ApproachRule(
+        name=APPROACH_ASSEMBLY_CHECKPOINT,
+        labels_any=frozenset({"assembly-checkpoint"}),
+        blob_contains=frozenset({"assembly-checkpoint"}),
+        hint_name=APPROACH_ASSEMBLY_CHECKPOINT,
+    ),
     ApproachRule(
         name=APPROACH_FI_RESEARCH_BRIEF,
         labels_any=frozenset({"research-brief", "freedom-intelligence"}),
@@ -144,6 +151,7 @@ def select_approach(run: OpsRun) -> str:
                 return rule.name
         # Explicit known command as hint
         known = {
+            APPROACH_ASSEMBLY_CHECKPOINT,
             APPROACH_FI_RESEARCH_BRIEF,
             APPROACH_BRIEF_DAILY,
             APPROACH_BRIEF_WEEKLY,
@@ -226,6 +234,10 @@ def execute_approach(
         )
 
     def _dispatch(tx: RepositoryTransaction | None) -> ApproachResult:
+        if name == APPROACH_ASSEMBLY_CHECKPOINT:
+            return _run_assembly_checkpoint(
+                target, title=run.title, report_to_hub=report_to_hub, commit=commit
+            )
         if name == APPROACH_FI_RESEARCH_BRIEF:
             return _run_fi(target, report_to_hub=report_to_hub, commit=commit)
         if name == APPROACH_BRIEF_DAILY:
@@ -297,6 +309,34 @@ def execute_approach(
         )
 
 
+def _run_assembly_checkpoint(
+    target: Path, *, title: str | None, report_to_hub: bool, commit: bool
+) -> ApproachResult:
+    from rein_aharness.assembly_checkpoint import kind_from_title, run_assembly_checkpoint
+
+    r = run_assembly_checkpoint(
+        target, kind=kind_from_title(title), report_to_hub=report_to_hub, commit=commit
+    )
+    return ApproachResult(
+        ok=r.ok,
+        approach=APPROACH_ASSEMBLY_CHECKPOINT,
+        result={
+            "date": r.date,
+            "kind": r.kind,
+            "path": r.path,
+            "out_of_window": r.out_of_window,
+            "committed": r.committed,
+            "pushed": r.pushed,
+            "origin_sha": r.origin_sha,
+            "head_after": r.head_after,
+            "counts": r.counts,
+            "target_repo": "assembly-economics",
+        },
+        reason=r.reason,
+        reopen=not r.ok,
+    )
+
+
 def _run_fi(target: Path, *, report_to_hub: bool, commit: bool) -> ApproachResult:
     from rein_aharness.fi_research_brief import run_fi_research_brief
 

rein_aharness/assembly_checkpoint.py

@@ -0,0 +1,156 @@
+"""Assembly Economics reality checkpoint (ASSEMBLY-WP-0010).
+
+activity-core owns *when* (assembly-checkpoint-* definitions). This approach
+owns *execution*: run the repo's deterministic checkpoint runner
+(`checkpoints/run_checkpoint.py`, data via web-fetcher, optional llm-connect
+assessment), commit the record under `checkpoints/`, push to origin
+(fast-forward only; publish grant limited to `checkpoints/`), then post
+`assembly_checkpoint`. A local-only commit is a failed run.
+
+Out-of-window fires (cron cannot express years; `checkpoints/schedule.md` is
+authoritative) are successful no-ops: nothing is written or pushed.
+
+No Claude Code / host coding agent; network access stays in this process.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import subprocess
+import sys
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+from rein_aharness import hub
+
+PUBLISH_PREFIX = "checkpoints/"
+RUNNER = "checkpoints/run_checkpoint.py"
+EXIT_OUT_OF_WINDOW = 3
+
+
+class AssemblyCheckpointError(RuntimeError):
+    pass
+
+
+@dataclass
+class AssemblyCheckpointResult:
+    ok: bool
+    kind: str
+    date: str = ""
+    path: str = ""
+    out_of_window: bool = False
+    committed: bool = False
+    pushed: bool = False
+    origin_sha: str = ""
+    head_after: str = ""
+    counts: dict[str, Any] = field(default_factory=dict)
+    reason: str | None = None
+
+
+def kind_from_title(title: str | None) -> str:
+    m = re.search(r"\((light|full)\)", title or "")
+    return m.group(1) if m else "light"
+
+
+def _git(repo: Path, *args: str, timeout: int = 60) -> str:
+    r = subprocess.run(["git", "-C", str(repo), *args], capture_output=True, text=True, timeout=timeout)
+    if r.returncode != 0:
+        raise AssemblyCheckpointError(f"git {' '.join(args)} failed: {r.stderr.strip()[:200]}")
+    return r.stdout.strip()
+
+
+def _branch(repo: Path) -> str:
+    b = _git(repo, "rev-parse", "--abbrev-ref", "HEAD")
+    return "main" if not b or b == "HEAD" else b
+
+
+def _sync_from_origin(repo: Path) -> None:
+    """Fast-forward to origin before running, so the runner sees current schedule/sources."""
+    branch = _branch(repo)
+    _git(repo, "fetch", "origin", branch, timeout=120)
+    _git(repo, "merge", "--ff-only", f"origin/{branch}", timeout=60)
+
+
+def _publish(repo: Path) -> str:
+    """Push HEAD; on rejection rebase checkpoint-only commits once and retry. Never force."""
+    branch = _branch(repo)
+    try:
+        _git(repo, "push", "origin", f"HEAD:{branch}", timeout=120)
+    except AssemblyCheckpointError:
+        _git(repo, "fetch", "origin", branch, timeout=120)
+        changed = _git(repo, "diff", "--name-only", f"origin/{branch}...HEAD").splitlines()
+        outside = [p for p in changed if not p.startswith(PUBLISH_PREFIX)]
+        if outside:
+            raise AssemblyCheckpointError(f"unpublished commits touch paths outside {PUBLISH_PREFIX}: {outside[:5]}")
+        try:
+            _git(repo, "rebase", f"origin/{branch}", timeout=120)
+        except AssemblyCheckpointError:
+            subprocess.run(["git", "-C", str(repo), "rebase", "--abort"], capture_output=True, timeout=60)
+            raise
+        _git(repo, "push", "origin", f"HEAD:{branch}", timeout=120)
+    return _git(repo, "rev-parse", "HEAD")
+
+
+def run_assembly_checkpoint(target_repo: Path, *, kind: str = "light", report_to_hub: bool = True,
+                            commit: bool = True, sync: bool = True) -> AssemblyCheckpointResult:
+    repo = target_repo.expanduser().resolve()
+    result = AssemblyCheckpointResult(ok=False, kind=kind)
+    try:
+        if sync and commit:
+            _sync_from_origin(repo)
+        env = dict(os.environ)
+        env.setdefault("WEB_FETCHER_PATH", str(Path(os.environ.get("ASSEMBLY_WEB_FETCHER_PATH",
+                                                                    "~/web-fetcher")).expanduser()))
+        proc = subprocess.run([sys.executable, "-B", RUNNER, "--kind", kind], cwd=repo, env=env,
+                              capture_output=True, text=True, timeout=900)
+        summary = {}
+        for line in reversed(proc.stdout.strip().splitlines()):
+            try:
+                summary = json.loads(line)
+                break
+            except ValueError:
+                continue
+        result.date = str(summary.get("date", ""))
+        if proc.returncode == EXIT_OUT_OF_WINDOW:
+            result.ok, result.out_of_window, result.reason = True, True, "out of window (checkpoints/schedule.md)"
+        elif proc.returncode != 0:
+            raise AssemblyCheckpointError(f"runner exit {proc.returncode}: {(proc.stderr or proc.stdout)[-300:]}")
+        else:
+            result.path = str(summary.get("path", ""))
+            result.counts = {k: summary.get(k) for k in ("ok", "not_found", "error", "assessment", "phase")}
+            if commit:
+                _git(repo, "add", PUBLISH_PREFIX)
+                staged = _git(repo, "diff", "--cached", "--name-only").splitlines()
+                outside = [p for p in staged if not p.startswith(PUBLISH_PREFIX)]
+                if outside:
+                    raise AssemblyCheckpointError(f"refusing to commit paths outside {PUBLISH_PREFIX}: {outside[:5]}")
+                if staged:
+                    _git(repo, "commit", "-m", f"Checkpoint {result.date} ({kind}) — activity-core")
+                    result.committed = True
+                result.origin_sha = _publish(repo)
+                result.pushed = True
+                result.head_after = result.origin_sha
+            result.ok = True
+    except (AssemblyCheckpointError, subprocess.TimeoutExpired, OSError) as exc:
+        result.ok = False
+        result.reason = str(exc)[:300]
+    _hub(result, report_to_hub)
+    return result
+
+
+def _hub(result: AssemblyCheckpointResult, report_to_hub: bool) -> None:
+    if not report_to_hub:
+        return
+    detail = {"repo": "assembly-economics", "date": result.date, "kind": result.kind, "path": result.path,
+              "out_of_window": result.out_of_window, "committed": result.committed, "pushed": result.pushed,
+              "origin_sha": result.origin_sha, **result.counts, "executor": "rein-aharness"}
+    if result.ok:
+        what = "out of window (no-op)" if result.out_of_window else f"recorded {result.path}"
+        hub.post_progress_event(summary=f"Assembly checkpoint {result.date} ({result.kind}): {what}",
+                                event_type="assembly_checkpoint", detail=detail)
+    else:
+        hub.post_progress_event(summary=f"Assembly checkpoint failed: {result.reason}",
+                                event_type="executor_run", detail={**detail, "ok": False, "reason": result.reason})

tests/test_assembly_checkpoint.py

@@ -0,0 +1,86 @@
+"""Tests for the assembly-checkpoint approach (offline; local bare origin)."""
+
+import subprocess
+import textwrap
+from pathlib import Path
+
+import pytest
+
+from rein_aharness import approaches as ap
+from rein_aharness.assembly_checkpoint import kind_from_title, run_assembly_checkpoint
+
+RUNNER = textwrap.dedent('''
+    import json, sys, pathlib
+    kind = sys.argv[sys.argv.index("--kind") + 1]
+    mode = pathlib.Path("checkpoints/MODE").read_text().strip()
+    if mode == "out":
+        print(json.dumps({"status": "out-of-window", "date": "2026-10-12", "kind": kind})); sys.exit(3)
+    if mode == "outside":
+        pathlib.Path("README.md").write_text("tampered")
+    p = pathlib.Path("checkpoints/2026/2026-10-14-" + kind + ".md")
+    p.parent.mkdir(parents=True, exist_ok=True); p.write_text("# record")
+    print(json.dumps({"status": "ok", "date": "2026-10-14", "kind": kind, "path": str(p), "ok": 5,
+                      "not_found": 4, "error": 0, "assessment": False, "phase": "setup"}))
+''')
+
+
+def _git(cwd, *args):
+    subprocess.run(["git", "-C", str(cwd), *args], check=True, capture_output=True)
+
+
+@pytest.fixture
+def repo(tmp_path: Path) -> Path:
+    origin = tmp_path / "origin.git"
+    _git(tmp_path, "init", "-q", "--bare", "-b", "main", str(origin))
+    work = tmp_path / "work"
+    _git(tmp_path, "clone", "-q", str(origin), str(work))
+    _git(work, "config", "user.email", "t@t"), _git(work, "config", "user.name", "t")
+    (work / "checkpoints").mkdir()
+    (work / "checkpoints" / "run_checkpoint.py").write_text(RUNNER)
+    (work / "checkpoints" / "MODE").write_text("in")
+    (work / "README.md").write_text("x")
+    _git(work, "add", "-A"), _git(work, "commit", "-qm", "init"), _git(work, "push", "-q", "origin", "HEAD:main")
+    return work
+
+
+def test_kind_from_title():
+    assert kind_from_title("Assembly Economics reality checkpoint (full) — steady phase") == "full"
+    assert kind_from_title("anything") == "light"
+
+
+def test_in_window_commits_and_pushes(repo):
+    r = run_assembly_checkpoint(repo, kind="light", report_to_hub=False)
+    assert r.ok and r.committed and r.pushed and r.path.endswith("2026-10-14-light.md")
+    origin_head = subprocess.run(["git", "-C", str(repo), "rev-parse", "origin/main"],
+                                 capture_output=True, text=True).stdout.strip()
+    assert origin_head == r.origin_sha
+
+
+def test_out_of_window_is_noop(repo):
+    (repo / "checkpoints" / "MODE").write_text("out")
+    _git(repo, "commit", "-qam", "mode out"), _git(repo, "push", "-q", "origin", "HEAD:main")
+    r = run_assembly_checkpoint(repo, kind="light", report_to_hub=False)
+    assert r.ok and r.out_of_window and not r.committed and not r.pushed
+
+
+def test_refuses_paths_outside_checkpoints(repo):
+    (repo / "checkpoints" / "MODE").write_text("outside")
+    _git(repo, "commit", "-qam", "mode outside"), _git(repo, "push", "-q", "origin", "HEAD:main")
+    r = run_assembly_checkpoint(repo, kind="light", report_to_hub=False)
+    assert r.ok  # README change is not staged: only checkpoints/ is added
+    changed = subprocess.run(["git", "-C", str(repo), "show", "--name-only", "--format=", "HEAD"],
+                             capture_output=True, text=True).stdout.split()
+    assert all(p.startswith("checkpoints/") for p in changed)
+
+
+def test_label_selects_assembly_checkpoint():
+    from rein_aharness.ops_run_client import OpsRun
+    import inspect
+    kw = {p: None for p in inspect.signature(OpsRun).parameters}
+    for k, p in inspect.signature(OpsRun).parameters.items():
+        if p.default is not inspect.Parameter.empty:
+            kw[k] = p.default
+    kw.update(id="x", title="Assembly Economics reality checkpoint (light) — setup phase", description="d",
+              labels=["assembly-checkpoint", "research", "automated"], target_repo="assembly-economics",
+              activity_definition_id="assembly-checkpoint-setup-daily")
+    assert ap.select_approach(OpsRun(**kw)) == ap.APPROACH_ASSEMBLY_CHECKPOINT