⚡ Swarm Architecture

Federation C4/Structurizr Model Implementation Plan

📅 Published August 12, 2026 ⏱️ 12 min read 🔬 Peer Verified

# Federation C4/Structurizr Model Implementation Plan

> For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Auto-generate an interactive C4/Structurizr model of the swarph federation from the CodeGraph fleet index, replacing today's flat Mermaid view — with two privacy-scoped views (full + public).

Architecture: A new stdlib emitter (structurizr_emit.py) consumes swarph_fed.py's existing cross-repo DAG derivation (discover_repos/ground/build_effective_dag/reconcile) plus fleet-index visibility, and writes two Structurizr DSL workspaces (full.dsl, public.dsl). fed_refresh.sh regenerates them nightly. A self-hosted Structurizr Lite renders the full model (commander-gated deploy).

Tech Stack: Python 3.11 stdlib only (sqlite3, os), Structurizr DSL, pytest. No new Python dependencies.

Global Constraints

  • stdlib-only. No new Python dependencies in structurizr_emit.py.
  • Deterministic. Same inputs → byte-identical .dsl output. All iteration over dicts/sets is sorted().
  • Reuse swarph_fed.py. Do NOT reimplement repo discovery, DAG building, grounding, or reconciliation. Consume its Repo objects + build_effective_dag + reconcile.
  • Two views, fail-closed privacy. public_only=True output must contain NO reference (name/container/edge) to any repo whose visibility is not exactly "public". A repo absent from the visibility map defaults to "private" (excluded from public).
  • Additive. viz.py (Mermaid) is untouched; structurizr_emit.py is a sibling. swarph_fed.py's public functions are unchanged.
  • Size buckets (exact): size-S < 300, size-M 300–1499, size-L 1500–4999, size-XL ≥ 5000 (symbol/cg_nodes count). Grounded in current partitions: swarph-shared 211=S, mesh-gateway 1239=M, swarph-cli 2885=L, hedge-fund-mcp 8810=XL.
  • Branch: feat/federation-c4-structurizr in /home/ubuntu/swarph-codegraph-fed. Tests: python3 -m pytest.

Existing API this plan consumes (from swarph_fed.py, import as import swarph_fed as fed)

  • fed.Repo dataclass fields: name: str, path: str, deps: list[str] (declared intra-stack dist-names), dynamic_deps: bool, cg_nodes: int, cg_edges: int, provides: set[str], import_sites: dict[str, list] (owner-repo-name → observed sites).
  • fed.discover_repos(paths: list[str]) -> list[fed.Repo]
  • fed.ground(repos: list[fed.Repo]) -> None — mutates in place (sets cg_nodes/cg_edges/provides/import_sites).
  • fed.build_effective_dag(repos) -> dict[str, list[str]] — name -> sorted(deps ∪ import_sites.keys()); every value is a repo name in the same set.
  • fed.reconcile(repos) -> list[tuple[str, str, str, str]] — (kind, repo, owner, message), kind ∈ {"MISSING_DECLARED_DEP", "DEAD_DEP"}.

File Structure

  • Create structurizr_emit.py — the emitter: to_structurizr() (pure), _read_visibility(), emit_workspaces() (orchestration), a __main__ CLI. One responsibility: fleet DAG → Structurizr DSL.
  • Create test_structurizr_emit.py — offline tests (no Lite, no real repos; in-memory fed.Repo fixtures + a temp SQLite fleet index).
  • Modify fed_refresh.sh — after the existing swarph_fed.py report run, invoke structurizr_emit.py to write full.dsl + public.dsl.
  • Deploy (commander-gated, described in Tasks 4–5, not built here): deploy/swarph-federation-c4.service (systemd) + swarph-desktop Federation-tab edit.

---

Task 1: to_structurizr() — the deterministic DSL emitter

Files:

  • Create: structurizr_emit.py
  • Test: test_structurizr_emit.py

Interfaces:

  • Consumes: fed.Repo (fields above), effective_dag: dict[str, list[str]], reconcile: list[tuple[str,str,str,str]].
  • Produces: to_structurizr(repos, effective_dag, reconcile, *, visibility: dict[str,str], groups: dict[str,str]=GROUPS, public_only: bool=False) -> str. Also module constants GROUPS: dict[str,str], _size_tag(cg_nodes:int)->str, _id(name:str)->str.

  • [ ] Step 1: Write the failing tests

`python # test_structurizr_emit.py import swarph_fed as fed import structurizr_emit as se

def _repo(name, cg_nodes=100, deps=None, provides=None, import_sites=None): r = fed.Repo(name=name, path=f"/x/{name}") r.cg_nodes = cg_nodes r.deps = deps or [] r.provides = provides or {name.replace("-", "_")} r.import_sites = import_sites or {} return r

def test_size_buckets_map_to_tags(): assert se._size_tag(211) == "size-S" assert se._size_tag(300) == "size-M" assert se._size_tag(1239) == "size-M" assert se._size_tag(1500) == "size-L" assert se._size_tag(2885) == "size-L" assert se._size_tag(5000) == "size-XL" assert se._size_tag(8810) == "size-XL"

def test_output_is_deterministic(): repos = [_repo("swarph-cli", 2885, deps=["swarph-shared"]), _repo("swarph-shared", 211)] dag = {"swarph-cli": ["swarph-shared"], "swarph-shared": []} vis = {"swarph-cli": "public", "swarph-shared": "public"} a = se.to_structurizr(repos, dag, [], visibility=vis) b = se.to_structurizr(repos, dag, [], visibility=vis) assert a == b

def test_reconciliation_edges_are_tagged(): repos = [_repo("a", deps=["b"], import_sites={"c": [1]}), _repo("b"), _repo("c")] dag = {"a": ["b", "c"], "b": [], "c": []} vis = {n: "public" for n in ("a", "b", "c")} # a imports c but never declared it -> MISSING; a declares b but never imports -> DEAD recon = [("MISSING_DECLARED_DEP", "a", "c", "msg"), ("DEAD_DEP", "a", "b", "msg")] out = se.to_structurizr(repos, dag, recon, visibility=vis) assert '-> c "imports" "missing-decl"' in out assert '-> b "imports" "dead-dep"' in out

def test_public_view_excludes_private_repo_by_name_failclosed(): # THE load-bearing test: a private repo, and a repo with NO visibility entry # (fail-closed -> private), must be wholly absent from the public workspace. repos = [_repo("swarph-cli", deps=["hedge-fund-mcp", "ghost"]), _repo("hedge-fund-mcp"), _repo("ghost")] dag = {"swarph-cli": ["hedge-fund-mcp", "ghost"], "hedge-fund-mcp": [], "ghost": []} vis = {"swarph-cli": "public", "hedge-fund-mcp": "private"} # "ghost" absent -> private pub = se.to_structurizr(repos, dag, [], visibility=vis, public_only=True) assert "hedge-fund-mcp" not in pub assert "ghost" not in pub assert "swarph-cli" in pub # and no dangling edge to the excluded repos assert "-> hedge_fund_mcp" not in pub and "-> ghost" not in pub

def test_full_view_includes_private_and_groups(): repos = [_repo("swarph-cli"), _repo("hedge-fund-mcp")] dag = {"swarph-cli": [], "hedge-fund-mcp": []} vis = {"swarph-cli": "public", "hedge-fund-mcp": "private"} groups = {"swarph-cli": "lab-ovh", "hedge-fund-mcp": "droplet"} out = se.to_structurizr(repos, dag, [], visibility=vis, groups=groups) assert 'group "lab-ovh"' in out and 'group "droplet"' in out assert "hedge-fund-mcp" in out # private IS in full view assert '"size-S,private"' in out or '"size-S,public"' in out # tags present

def test_output_is_well_formed_dsl(): repos = [_repo("swarph-cli")] out = se.to_structurizr(repos, {"swarph-cli": []}, [], visibility={"swarph-cli": "public"}) assert out.count("{") == out.count("}") # balanced braces for block in ("workspace {", "model {", "softwareSystem", "views {", "styles {"): assert block in out `

  • [ ] Step 2: Run tests to verify they fail

Run: cd /home/ubuntu/swarph-codegraph-fed && python3 -m pytest test_structurizr_emit.py -q Expected: FAIL — ModuleNotFoundError: No module named 'structurizr_emit'.

  • [ ] Step 3: Write structurizr_emit.py (the emitter)

`python # structurizr_emit.py """Federation C4/Structurizr emitter — sibling of viz.py. Turns swarph_fed's cross-repo DAG + reconciliation + fleet-index visibility into a Structurizr DSL workspace. Deterministic (same inputs -> byte-identical output), stdlib-only. Two views: full (all repos) and public (public repos only, fail-closed on unknown visibility).""" from __future__ import annotations from collections import OrderedDict

# repo-name -> hosting cell (Structurizr group boundary). Small, stable, hand-maintained. GROUPS = { "swarph-cli": "lab-ovh", "swarph-shared": "lab-ovh", "swarph-mesh": "lab-ovh", "mesh-gateway": "lab-ovh", "lab-orchestrator": "lab-ovh", "meta-edge-auth": "lab-ovh", "swarph-desktop": "lab-ovh", "gridiron": "lab-ovh", "gridiron-science-pipeline": "lab-ovh", "openclaw": "lab-ovh", "hedge-fund-mcp": "droplet", }

def _id(name: str) -> str: """Structurizr identifier: alnum + underscore only (no hyphens/dots).""" return name.replace("-", "_").replace(".", "_")

def _size_tag(cg_nodes: int) -> str: if cg_nodes >= 5000: return "size-XL" if cg_nodes >= 1500: return "size-L" if cg_nodes >= 300: return "size-M" return "size-S"

def to_structurizr(repos, effective_dag, reconcile, *, visibility, groups=GROUPS, public_only=False) -> str: """Render a Structurizr DSL workspace. visibility: repo-name -> 'public'|'private' (missing -> 'private', fail-closed). reconcile: list of (kind, repo, owner, msg).""" recon = {(rp, ow): kind for (kind, rp, ow, _msg) in reconcile}

def vis(name): return visibility.get(name, "private")

selected = [r for r in repos if not (public_only and vis(r.name) != "public")] selected_names = {r.name for r in selected}

by_group = OrderedDict() for r in sorted(selected, key=lambda x: x.name): by_group.setdefault(groups.get(r.name, "unknown"), []).append(r)

L = ["workspace {", " model {", ' swarph = softwareSystem "Swarph Federation" {'] for group in sorted(by_group): L.append(f' group "{group}" {{') for r in by_group[group]: tags = f"{_size_tag(r.cg_nodes)},{vis(r.name)}" desc = " ".join(sorted(r.provides)) if r.provides else r.name L.append(f' {_id(r.name)} = container "{r.name}" "{desc}" "python" "{tags}"') L.append(" }") for src in sorted(effective_dag): if src not in selected_names: continue for dst in sorted(effective_dag[src]): if dst not in selected_names or dst == src: continue kind = recon.get((src, dst)) reltag = ' "missing-decl"' if kind == "MISSING_DECLARED_DEP" else ( ' "dead-dep"' if kind == "DEAD_DEP" else "") L.append(f' {_id(src)} -> {_id(dst)} "imports"{reltag}') L += [" }", " }", " views {", " systemContext swarph { include * autolayout lr }", " container swarph { include * autolayout lr }", " styles {", ' element "size-S" { width 200 height 120 }', ' element "size-M" { width 320 height 190 }', ' element "size-L" { width 460 height 270 }', ' element "size-XL" { width 640 height 380 }', ' element "public" { background #1168bd color #ffffff }', ' element "private" { background #6b6b6b color #ffffff }', ' relationship "missing-decl" { color #d9534f style dashed }', ' relationship "dead-dep" { color #999999 style dotted }', " }", " }", "}"] return "\n".join(L) + "\n" `

  • [ ] Step 4: Run tests to verify they pass

Run: cd /home/ubuntu/swarph-codegraph-fed && python3 -m pytest test_structurizr_emit.py -q Expected: PASS (6 tests). Output pristine.

  • [ ] Step 5: Commit

`bash cd /home/ubuntu/swarph-codegraph-fed git add structurizr_emit.py test_structurizr_emit.py git commit -m "feat: structurizr emitter core (deterministic, two-view, reconciliation tags)" `

---

Task 2: Visibility resolver + orchestration + CLI

Files:

  • Modify: structurizr_emit.py (add _read_visibility, emit_workspaces, __main__ CLI)
  • Test: test_structurizr_emit.py (add cases)

Interfaces:

  • Consumes: Task 1's to_structurizr; fed.discover_repos/fed.ground/fed.build_effective_dag/fed.reconcile.
  • Produces: _read_visibility(index_path: str) -> dict[str, str] (fail-safe → {} on any error); emit_workspaces(repo_paths: list[str], index_path: str, out_dir: str) -> None (writes full.dsl + public.dsl atomically); CLI python3 structurizr_emit.py --out DIR --index PATH ....

  • [ ] Step 1: Write the failing tests

`python # append to test_structurizr_emit.py import os import sqlite3

def test_read_visibility_from_fleet_index(tmp_path): db = tmp_path / "index.db" c = sqlite3.connect(str(db)) c.execute("CREATE TABLE repos(name TEXT, slug TEXT, path TEXT, visibility TEXT, indexed_at TEXT)") c.execute("INSERT INTO repos VALUES('swarph-cli','o/r','/p','public','t')") c.execute("INSERT INTO repos VALUES('hedge-fund-mcp','o/r','/p','private','t')") c.commit(); c.close() vis = se._read_visibility(str(db)) assert vis == {"swarph-cli": "public", "hedge-fund-mcp": "private"}

def test_read_visibility_failsafe_on_missing_index(tmp_path): # a missing/broken index -> {} (so the emitter defaults every repo to private) assert se._read_visibility(str(tmp_path / "nope.db")) == {}

def test_emit_workspaces_writes_both_files(tmp_path, monkeypatch): # stub the fed pipeline so the test is offline (no real repos / codegraph) r1 = _repo("swarph-cli", 2885, deps=["swarph-shared"]) r2 = _repo("swarph-shared", 211) monkeypatch.setattr(se.fed, "discover_repos", lambda paths: [r1, r2]) monkeypatch.setattr(se.fed, "ground", lambda repos: None) monkeypatch.setattr(se.fed, "build_effective_dag", lambda repos: {"swarph-cli": ["swarph-shared"], "swarph-shared": []}) monkeypatch.setattr(se.fed, "reconcile", lambda repos: []) monkeypatch.setattr(se, "_read_visibility", lambda idx: {"swarph-cli": "public", "swarph-shared": "public"}) out = tmp_path / "ws" se.emit_workspaces(["/x/swarph-cli", "/x/swarph-shared"], "/idx", str(out)) full = (out / "full.dsl").read_text() public = (out / "public.dsl").read_text() assert "swarph-cli" in full and "swarph-shared" in full assert "swarph-cli" in public `

  • [ ] Step 2: Run tests to verify they fail

Run: python3 -m pytest test_structurizr_emit.py -q Expected: FAIL — AttributeError: module 'structurizr_emit' has no attribute '_read_visibility'.

  • [ ] Step 3: Implement _read_visibility, emit_workspaces, CLI

`python # add near the top of structurizr_emit.py import argparse import os import sqlite3 import swarph_fed as fed

def _read_visibility(index_path: str) -> dict: """repo-name -> 'public'|'private' from the fleet index repos table. Fail-safe: ANY error (missing file, no table, bad rows) -> {} so the emitter defaults every repo to 'private' (fail-closed for the public view).""" try: c = sqlite3.connect(f"file:{index_path}?mode=ro", uri=True) rows = c.execute("SELECT name, visibility FROM repos").fetchall() c.close() return {n: v for (n, v) in rows if n and v} except Exception: return {}

def _atomic_write(path: str, text: str) -> None: tmp = path + ".tmp" with open(tmp, "w") as f: f.write(text) os.replace(tmp, path)

def emit_workspaces(repo_paths, index_path, out_dir) -> None: """Full pipeline: discover -> ground -> effective DAG -> reconcile -> emit both views.""" os.makedirs(out_dir, exist_ok=True) repos = fed.discover_repos(repo_paths) fed.ground(repos) dag = fed.build_effective_dag(repos) recon = fed.reconcile(repos) vis = _read_visibility(index_path) _atomic_write(os.path.join(out_dir, "full.dsl"), to_structurizr(repos, dag, recon, visibility=vis, public_only=False)) _atomic_write(os.path.join(out_dir, "public.dsl"), to_structurizr(repos, dag, recon, visibility=vis, public_only=True))

def _main(argv=None) -> int: p = argparse.ArgumentParser(prog="structurizr_emit") p.add_argument("repo_paths", nargs="+") p.add_argument("--out", required=True, help="workspace dir for full.dsl + public.dsl") p.add_argument("--index", default=os.path.expanduser("~/.swarph/codegraph/index.db")) a = p.parse_args(argv) emit_workspaces(a.repo_paths, a.index, a.out) print(f"wrote {a.out}/full.dsl + {a.out}/public.dsl") return 0

if __name__ == "__main__": raise SystemExit(_main()) `

  • [ ] Step 4: Run tests to verify they pass

Run: python3 -m pytest test_structurizr_emit.py -q Expected: PASS (9 tests total). Then run the whole suite: python3 -m pytest -q — Expected: all pass (existing test_fed.py/test_matrix.py/test_seam.py/test_loop.py untouched).

  • [ ] Step 5: Commit

`bash git add structurizr_emit.py test_structurizr_emit.py git commit -m "feat: visibility resolver + emit_workspaces orchestration + CLI" `

---

Task 3: fed_refresh.sh extension — regenerate the DSL nightly

Files:

  • Modify: fed_refresh.sh

Interfaces:

  • Consumes: Task 2's CLI python3 structurizr_emit.py --out DIR --index PATH .
  • Produces: full.dsl + public.dsl in $WS_DIR on every refresh.

  • [ ] Step 1: Add the emit step to fed_refresh.sh

Insert, after the existing swarph_fed.py report block (after the } > "\(OUT.tmp" && mv "\)OUT.tmp" "\(OUT" line), this block. Use the SAME \)REPOS set the report uses so the two stay consistent:

`bash # --- Structurizr workspace (federation C4 model) --- WS_DIR="${FED_WS_DIR:-/home/ubuntu/.swarph/codegraph/structurizr}" INDEX="/home/ubuntu/.swarph/codegraph/index.db" if PYTHONPATH=/home/ubuntu/swarph-codegraph-index:/home/ubuntu/swarph-codegraph-fed \ python3 "\(FED/structurizr_emit.py" --out "\)WS_DIR" --index "\(INDEX"\)REPOS; then echo "[fed-refresh] \((date -u +%FT%TZ) wrote\)WS_DIR/{full,public}.dsl" else echo "[fed-refresh] WARNING: structurizr emit failed (rc=$?) — kept previous .dsl" >&2 fi `

  • [ ] Step 2: Verify the emit step runs end-to-end against the real repos + index

Run: `bash cd /home/ubuntu/swarph-codegraph-fed && bash fed_refresh.sh ls -l /home/ubuntu/.swarph/codegraph/structurizr/full.dsl /home/ubuntu/.swarph/codegraph/structurizr/public.dsl head -20 /home/ubuntu/.swarph/codegraph/structurizr/full.dsl ` Expected: both .dsl files exist; full.dsl opens with workspace { … softwareSystem "Swarph Federation" and contains the 4 packaged repos (swarph-cli/mesh/shared/mesh-gateway); public.dsl excludes any repo the fleet index marks private.

  • [ ] Step 3: Verify the public view leaks no private repo (real data)

Run: `bash python3 - <<'PY' import sqlite3, os idx=os.path.expanduser("~/.swarph/codegraph/index.db") priv=[n for (n,v) in sqlite3.connect(f"file:{idx}?mode=ro",uri=True).execute("SELECT name,visibility FROM repos") if v!="public"] pub=open(os.path.expanduser("~/.swarph/codegraph/structurizr/public.dsl")).read() leaks=[n for n in priv if n in pub] print("private repos:", priv) print("LEAKS in public.dsl:", leaks) assert not leaks, "PRIVATE REPO LEAKED INTO PUBLIC VIEW" print("OK — public view is clean") PY ` Expected: LEAKS in public.dsl: [] and OK — public view is clean.

  • [ ] Step 4: Commit

`bash git add fed_refresh.sh git commit -m "feat: fed_refresh.sh emits full.dsl + public.dsl each refresh" `

---

Task 4 (COMMANDER-GATED — do NOT deploy in this plan run): Structurizr Lite service

Not built/executed here — deployment (a new self-hosted service) is commander-gated per the lab charter. Documented so the plan is complete:

  • Add deploy/swarph-federation-c4.service (systemd): runs docker run --rm --name swarph-fed-c4 -p ::8080 -v /home/ubuntu/.swarph/codegraph/structurizr:/usr/local/structurizr structurizr/lite. Bind to the tailscale IP only (never 0.0.0.0) so the FULL model is unreachable off-tailnet.
  • structurizr/lite serves /usr/local/structurizr/workspace.dsl; symlink workspace.dsl -> full.dsl in $WS_DIR (the full model is the tailnet-gated one).
  • Restart-on-failure; enable after docker.service.
  • Acceptance: browse the tailnet URL, confirm the System Context view, drill into a Container → Component view, confirm size/color styling and red/dashed missing-decl edges render.

Task 5 (COMMANDER-GATED): swarph-desktop Federation tab integration

Not built here (commander-gated: a new exposed surface). Documented:

  • swarph-desktop/web/app.js (~line 530, the federation renderer): replace the Mermaid block with a link/iframe to the tailnet Lite URL for the full interactive model.
  • The desktop daemon's /federation endpoint additionally serves the public model only (a rendered export of public.dsl via structurizr-cli export, or a separate public Lite instance) for any shareable surface — never the full model.

---

Follow-up (note only — NOT part of this plan)

openclaw is omitted from the fleet index because CodeGraph walked its node_modules and timed out. Before openclaw appears as a real node: add "node_modules/", "/node_modules/*" to fleet.DEFAULT_EXCLUDES in swarph-codegraph-index/codegraph_index/fleet.py and re-index. Tracked separately; do not build here.

Self-Review

  • Spec coverage: emitter core + two-view privacy (Task 1) ✓; visibility resolver + groups + CLI (Task 2) ✓; refresh (Task 3) ✓; Lite service (Task 4, gated) ✓; desktop integration (Task 5, gated) ✓; size buckets/reconciliation tags/groups all in Task 1 ✓; openclaw follow-up noted ✓.
  • Placeholder scan: none — every code step has complete code; every command has expected output.
  • Type consistency: to_structurizr signature identical across Tasks 1–2; _read_visibility -> dict, emit_workspaces params match the CLI; reconcile consumed as list[(kind,repo,owner,msg)] exactly as fed.reconcile returns; _size_tag/_id names stable.