feat(brief): generate .custodian-brief.md per repo for offline worker orientation

Adds _write_custodian_brief() to consistency_check.py. After every fix_repo()
run, a .custodian-brief.md is written to the repo root with: domain, last-synced
timestamp, current repo goal, active workstreams with progress (done/total), and
the first 7 open tasks per workstream (blocked → in_progress → todo order) with
task IDs. The file is git-committed when content changes so remote workers (e.g.
CoulombCore) can pull it and orient without a live MCP connection.

Session protocol template and CLAUDE.md updated: read .custodian-brief.md first,
then call get_domain_summary() as an enhancement (skip if MCP unreachable).
This eliminates false "State hub is offline" alarms in subagents and remote workers.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

scripts/consistency_check.py

@@ -858,6 +858,143 @@ def _git_commit_writeback(
         return False
 
 
+# ---------------------------------------------------------------------------
+# Worker orientation brief (.custodian-brief.md)
+# ---------------------------------------------------------------------------
+
+_BRIEF_HEADER = "<!-- custodian-brief: generated by fix-consistency — do not edit manually -->"
+_TASK_STATUS_ICON = {"done": "✓", "cancelled": "✗", "in_progress": "►", "blocked": "!", "todo": "·"}
+_OPEN_STATUSES = {"todo", "in_progress", "blocked"}
+
+
+def _write_custodian_brief(api_base: str, repo_slug: str, repo_path: str) -> bool:
+    """Generate .custodian-brief.md at the repo root and git-commit if changed.
+
+    The brief gives any agent — including subagents without MCP access and
+    workers on remote machines — instant orientation without a live hub
+    connection.  Returns True if the file was written (content changed).
+    """
+    import datetime as _dt
+    from datetime import timezone as _tz
+
+    repo = _api_get(api_base, f"/repos/{repo_slug}")
+    if not repo:
+        return False
+
+    repo_id: str = repo.get("id", "")
+    domain_slug: str = ""
+
+    # Resolve domain slug via the topic linked to any workstream
+    workstreams = _api_get(api_base, "/workstreams", {"repo_id": repo_id, "status": "active"}) or []
+    if isinstance(workstreams, list) and workstreams:
+        topic = _api_get(api_base, f"/topics/{workstreams[0].get('topic_id', '')}")
+        if topic:
+            domain_slug = topic.get("domain_slug", "")
+
+    # Active repo goal (first active one if multiple)
+    goal_text = ""
+    goals = _api_get(api_base, "/repo-goals", {"repo_slug": repo_slug}) or []
+    if isinstance(goals, list):
+        active_goals = [g for g in goals if g.get("status") == "active"]
+        if active_goals:
+            g = active_goals[0]
+            goal_text = g.get("title", "") or g.get("description", "")
+
+    now_utc = _dt.datetime.now(_tz.utc)
+    ts = now_utc.strftime("%Y-%m-%d %H:%M UTC")
+
+    lines = [
+        _BRIEF_HEADER,
+        f"# Custodian Brief — {repo_slug}",
+        "",
+        f"**Domain:** {domain_slug or '(unknown)'}  ",
+        f"**Last synced:** {ts}  ",
+        "**State Hub:** http://127.0.0.1:8000 *(adjust if running on a remote machine)*",
+        "",
+    ]
+
+    if goal_text:
+        lines += ["## Current Goal", "", goal_text, ""]
+
+    if isinstance(workstreams, list) and workstreams:
+        lines.append("## Active Workstreams")
+        for ws in workstreams:
+            ws_title = ws.get("title", ws.get("slug", "?"))
+            ws_id = ws["id"]
+            tasks = _api_get(api_base, "/tasks", {"workstream_id": ws_id}) or []
+            if not isinstance(tasks, list):
+                tasks = []
+
+            done = sum(1 for t in tasks if t.get("status") in ("done", "cancelled"))
+            total = len(tasks)
+            pct = f"{done}/{total}" if total else "no tasks"
+
+            open_tasks = [t for t in tasks if t.get("status") in _OPEN_STATUSES]
+            # Show blocked first, then in_progress, then todo (cap at 5)
+            priority_order = {"blocked": 0, "in_progress": 1, "todo": 2}
+            open_tasks.sort(key=lambda t: priority_order.get(t.get("status", "todo"), 9))
+
+            lines += [
+                "",
+                f"### {ws_title}",
+                f"Progress: {pct} done  |  workstream_id: `{ws_id}`",
+            ]
+
+            if open_tasks:
+                lines.append("")
+                lines.append("**Open tasks:**")
+                for t in open_tasks[:7]:
+                    icon = _TASK_STATUS_ICON.get(t.get("status", "todo"), "·")
+                    title = t.get("title", t["id"])
+                    tid = t["id"]
+                    status = t.get("status", "")
+                    blocker = t.get("blocking_reason", "")
+                    task_line = f"- {icon} {title}  `{tid[:8]}`"
+                    if status == "blocked" and blocker:
+                        task_line += f"\n  *(blocked: {blocker})*"
+                    lines.append(task_line)
+                if len(open_tasks) > 7:
+                    lines.append(f"- … and {len(open_tasks) - 7} more open tasks")
+    else:
+        lines += ["## Active Workstreams", "", "*(none — repo may need first-session setup)*"]
+
+    lines += [
+        "",
+        "---",
+        "## MCP Orientation (when available)",
+        "",
+        "If the state-hub MCP server is reachable, call:",
+        f"`get_domain_summary(\"{domain_slug}\")`",
+        "This provides richer cross-domain context.",
+        "If the MCP call fails, use this file as your orientation source.",
+    ]
+
+    content = "\n".join(lines) + "\n"
+
+    brief_path = Path(repo_path) / ".custodian-brief.md"
+    existing = brief_path.read_text(encoding="utf-8") if brief_path.exists() else ""
+
+    # Strip the timestamp line before comparing to avoid spurious writes
+    def _strip_ts(text: str) -> str:
+        return "\n".join(
+            ln for ln in text.splitlines()
+            if not ln.startswith("**Last synced:**")
+        )
+
+    if _strip_ts(content) == _strip_ts(existing):
+        return False  # no meaningful change
+
+    brief_path.write_text(content, encoding="utf-8")
+
+    # Commit the brief so remote workers can pull it
+    _git_commit_writeback(
+        repo_path,
+        brief_path,
+        [f"update .custodian-brief.md for {repo_slug}"],
+    )
+    return True
+
+
 # ---------------------------------------------------------------------------
 # Fix engine
 # ---------------------------------------------------------------------------
@@ -1094,6 +1231,12 @@ def fix_repo(
     now_iso = _dt.datetime.now(_tz.utc).isoformat()
     _api_patch(api_base, f"/repos/{repo_slug}/", {"last_state_synced_at": now_iso})
 
+    # Write the worker orientation brief (.custodian-brief.md)
+    if repo_path:
+        brief_written = _write_custodian_brief(api_base, repo_slug, repo_path)
+        if brief_written:
+            report.fixes_applied.append("brief: .custodian-brief.md updated")
+
     return report
 
 

scripts/project_rules/session-protocol.template

@@ -3,10 +3,16 @@
 State Hub: http://127.0.0.1:8000
 
 **Step 1 — Orient**
+
+Read the offline-safe brief first — it works without a live hub connection:
+```bash
+cat .custodian-brief.md
+```
+Then call the MCP tool for richer cross-domain context (skip if unreachable):
 ```
 get_domain_summary("{DOMAIN}")
 ```
-If offline: `cd ~/the-custodian/state-hub && make api`
+If the hub is offline: `cd ~/the-custodian/state-hub && make api`
 
 **Step 2 — Check inbox**
 ```