angelo_cli

angelo_cli

Angelo CLI - scaffolds MCP config into target projects.

cmd_init

cmd_init(args: Namespace) -> None

Scaffold editor config into the current project.

Source code in angelo_cli/__init__.py

def cmd_init(args: argparse.Namespace) -> None:
    """Scaffold editor config into the current project."""
    project_dir = Path.cwd()
    force = args.force
    with_zk = args.with_zettelkasten
    for_cursor = args.cursor
    for_claude = args.claude
    local_paths = getattr(args, "local_paths", False)

    if not for_cursor and not for_claude:
        for_cursor, for_claude = _prompt_editor_choice()

    if for_cursor:
        _init_cursor(project_dir, force=force, with_zk=with_zk, local_paths=local_paths)
    if for_claude:
        if for_cursor:
            print()
        _init_claude(project_dir, force=force, with_zk=with_zk, local_paths=local_paths)

    # Seamlessly relocate any legacy zettelkasten data (.angelo/zettelkasten/ ->
    # committed .zettelkasten/) when re-running init on an older repo.
    zk_migration = _migrate_zettelkasten_layout(project_dir)
    if zk_migration != "unchanged":
        print(f"\nzettelkasten layout: {zk_migration}")

    if local_paths:
        print("\nNote: --local-paths pinned absolute, machine-specific command paths into "
              "mcp.json. Great for a solo repo; for a shared one, don't commit those paths "
              "- teammates run 'angelo init --local-paths' (or 'angelo doctor --fix') in "
              "their own environment.")

    # Prime the embedding-model cache now so the first semantic search isn't a
    # silent ~120MB download (and so an offline first run degrades predictably).
    # Resilient + skippable (MEMORY_SKIP_EMBEDDINGS=1); never fails init.
    print(f"\nEmbedding model cache: {_prefetch_embedding_model()}")

cmd_update

cmd_update(args: Namespace) -> None

Upgrade the angelo package (when newer) and refresh managed files.

The package upgrade reinstalls from git when a newer version exists (copied installs only; editable installs get guidance instead). The config refresh rewrites the managed rule/hook/skill files and reconciles mcp.json non-destructively: any angelo-managed MCP server added in a newer package version (e.g. memory-artifacts) is merged into the existing config without touching the user's other servers or env tweaks.

Source code in angelo_cli/__init__.py

def cmd_update(args: argparse.Namespace) -> None:
    """Upgrade the angelo package (when newer) and refresh managed files.

    The package upgrade reinstalls from git when a newer version exists (copied
    installs only; editable installs get guidance instead). The config refresh
    rewrites the managed rule/hook/skill files and reconciles mcp.json
    non-destructively: any angelo-managed MCP server added in a newer package
    version (e.g. memory-artifacts) is merged into the existing config without
    touching the user's other servers or env tweaks.
    """
    project_dir = Path.cwd()
    has_cursor = _has_cursor_layer(project_dir)
    has_claude = _has_claude_layer(project_dir)

    if not has_cursor and not has_claude:
        print("No .cursor/ or .claude/ config found. Run 'angelo init' or "
              "'angelo init --claude' first.", file=sys.stderr)
        sys.exit(1)

    # Package upgrade first: on a successful copied-install upgrade this re-execs
    # a fresh `angelo update --no-upgrade` and exits, so the config refresh below
    # runs against the newly installed bundled data rather than the stale data in
    # this process's memory.
    if not getattr(args, "no_upgrade", False):
        with_zk_any = (
            _detect_zettelkasten_cursor(project_dir)
            or _detect_zettelkasten_claude(project_dir)
        )
        edit_flags = []
        if getattr(args, "migrate_edits", False):
            edit_flags.append("--migrate-edits")
        elif getattr(args, "keep_edits", False):
            edit_flags.append("--keep-edits")
        _maybe_upgrade_package(
            with_zk_any, getattr(args, "force_upgrade", False), edit_flags
        )

    print(f"Updating angelo config in {project_dir}\n")
    changed = 0
    preserved = 0
    migrated = 0

    # Relocate legacy zettelkasten data (.angelo/zettelkasten/ -> .zettelkasten/)
    # and repoint ZETTELKASTEN_PATH before reconciling mcp.json below, so an
    # upgrade from the old gitignored layout is seamless.
    zk_migration = _migrate_zettelkasten_layout(project_dir)
    if zk_migration != "unchanged":
        print(f"  zettelkasten layout: {zk_migration}")
        changed += 1

    if has_cursor:
        with_zk = _detect_zettelkasten_cursor(project_dir)
        print("Cursor layer:")
        manifest_path = project_dir / CURSOR_MANIFEST_REL
        manifest = _load_manifest(manifest_path)
        managed = _get_cursor_files(with_zk)
        for src_rel, tgt_rel in managed.items():
            src = DATA_DIR / src_rel
            tgt = project_dir / tgt_rel
            content = src.read_text(encoding="utf-8")
            status, bucket = _apply_managed_update(tgt, content, manifest, tgt_rel, tgt_rel, args)
            print(f"  {tgt_rel}: {status}")
            if bucket == "preserved":
                preserved += 1
            elif bucket == "migrated":
                migrated += 1
            elif bucket == "changed":
                changed += 1
        _save_manifest(manifest_path, manifest)
        mcp_template = "mcp-zettelkasten.json" if with_zk else "mcp.json"
        status = _reconcile_mcp_json(project_dir / ".cursor" / "mcp.json", mcp_template)
        print(f"  .cursor/mcp.json: {status}")
        if status.startswith("updated"):
            changed += 1

    if has_claude:
        with_zk_claude = _detect_zettelkasten_claude(project_dir)
        print("Claude Code layer:")
        manifest_path = project_dir / CLAUDE_MANIFEST_REL
        manifest = _load_manifest(manifest_path)
        managed = _get_claude_files(with_zk_claude)
        for src_rel, tgt_rel in managed.items():
            src = DATA_DIR / src_rel
            tgt = project_dir / tgt_rel
            content = src.read_text(encoding="utf-8")
            status, bucket = _apply_managed_update(tgt, content, manifest, tgt_rel, tgt_rel, args)
            print(f"  {tgt_rel}: {status}")
            if bucket == "preserved":
                preserved += 1
            elif bucket == "migrated":
                migrated += 1
            elif bucket == "changed":
                changed += 1
        _save_manifest(manifest_path, manifest)
        mcp_template = "claude/mcp-zettelkasten.json" if with_zk_claude else "claude/mcp.json"
        status = _reconcile_mcp_json(project_dir / ".mcp.json", mcp_template)
        print(f"  .mcp.json: {status}")
        if status.startswith("updated"):
            changed += 1
        status = _update_claude_md(project_dir, with_zk_claude)
        print(f"  CLAUDE.md: {status}")
        if status != "unchanged":
            changed += 1

    status = _update_gitignore(project_dir)
    print(f"  .gitignore: {status}")
    if status != "unchanged":
        changed += 1

    if changed == 0 and preserved == 0 and migrated == 0:
        print("\nAll files are up to date.")
    else:
        editors = []
        if has_cursor:
            editors.append("Cursor")
        if has_claude:
            editors.append("Claude Code")
        if changed or migrated:
            n = changed + migrated
            print(f"\n{n} file(s) updated. Restart {' / '.join(editors)} to pick up changes.")
        else:
            print()
    if migrated:
        print(f"{migrated} edited rule(s) had their changes migrated into a matching "
              "'*-local' rule (un-managed; survives updates) and the managed rule was "
              "refreshed. Review the migrated block(s) and trim to just your customizations.")
    if preserved:
        print(f"{preserved} file(s) had local edits and were preserved. The new upstream "
              "version is alongside each as '<file>.new' - reconcile your changes and "
              "delete the .new file. To keep customizations that survive updates, put them "
              "in a separate un-managed rule (e.g. .cursor/rules/coordinator-local.mdc), "
              "or re-run 'angelo update --migrate-edits' to move them automatically.")

    # Recycle any dashboard still serving pre-update code so the next open is
    # fresh. Skipped with --no-restart-dashboards.
    if getattr(args, "restart_dashboards", True):
        _recycle_stale_dashboards(project_dir)

cmd_doctor

cmd_doctor(args: Namespace) -> None

Verify the install: PATH commands, dependencies, config files, onboarding state.

Source code in angelo_cli/__init__.py

def cmd_doctor(args: argparse.Namespace) -> None:
    """Verify the install: PATH commands, dependencies, config files, onboarding state."""
    project_dir = Path.cwd()
    has_cursor = _has_cursor_layer(project_dir)
    has_claude = _has_claude_layer(project_dir)
    with_zk = _detect_zettelkasten_cursor(project_dir) or _detect_zettelkasten_claude(project_dir)

    failures = 0
    warnings = 0

    def ok(msg: str) -> None:
        print(f"  [ok]   {msg}")

    def warn(msg: str) -> None:
        nonlocal warnings
        warnings += 1
        print(f"  [warn] {msg}")

    def fail(msg: str) -> None:
        nonlocal failures
        failures += 1
        print(f"  [FAIL] {msg}")

    layers = []
    if has_cursor:
        layers.append("Cursor")
    if has_claude:
        layers.append("Claude Code")
    layer_label = " + ".join(layers) if layers else "no editor config"

    label = "angelo" + (" + zettelkasten" if with_zk else "")
    print(f"angelo doctor - checking {label} ({layer_label}) in {project_dir}\n")

    # --- Optional repair pass (--fix) ---
    # Runs before the checks so the report below reflects the post-fix state.
    if getattr(args, "fix", False):
        print("Applying fixes (--fix)")
        any_mcp = False
        for mcp_json in (project_dir / ".cursor" / "mcp.json", project_dir / ".mcp.json"):
            if mcp_json.exists():
                any_mcp = True
                rel = mcp_json.relative_to(project_dir)
                print(f"  {rel}: {_rewrite_mcp_commands_absolute(mcp_json)}")
        if not any_mcp:
            print("  no mcp.json found - run 'angelo init' first")
        if not _dedupe_kglite():
            print("  kglite: single backend (no duplicate to migrate)")
        print(f"  embedding model: {_prefetch_embedding_model()}")
        print()

    # --- Python ---
    print("Python")
    if sys.version_info >= (3, 10):
        ok(f"Python {sys.version_info.major}.{sys.version_info.minor} (>= 3.10)")
    else:
        fail(f"Python {sys.version_info.major}.{sys.version_info.minor} - 3.10+ required")

    # --- Package version / currency ---
    # The MCP servers run from the installed package, NOT the project config, so
    # a stale package is invisible to the config-file checks below. Surface it
    # here: a copied install behind the repo is the "ran update but the servers
    # are still old" footgun (e.g. canvas-only, pre-dashboard builds).
    print("\nPackage version")
    editable, src = _install_origin()
    ok(f"angelo {__version__}" + (" (editable)" if editable else ""))
    if editable:
        src_ver = None
        if src and (src / "pyproject.toml").exists():
            try:
                src_ver = _read_pyproject_version((src / "pyproject.toml").read_text(encoding="utf-8"))
            except OSError:
                src_ver = None
        if src_ver and _version_newer(src_ver, __version__):
            warn(f"source tree is at {src_ver} but installed metadata is {__version__} "
                 f"- re-run 'pip install -e' and restart the MCP servers")
        else:
            print("  [note] editable install - 'git pull' then restart the MCP servers "
                  "('angelo kill', reopen) to load new server code")
    else:
        latest = _fetch_remote_version()
        if latest is None:
            print("  [note] could not check for a newer version (offline?)")
        elif _version_newer(latest, __version__):
            warn(f"newer version available ({latest}) - run 'angelo update' to upgrade "
                 f"the package + refresh config, then restart the MCP servers")
        else:
            ok(f"up to date (latest {latest})")

    # --- Commands on PATH ---
    print("\nCommands on PATH")
    required_cmds = ["angelo-coordinator", "angelo-memory"]
    if with_zk:
        required_cmds.append("angelo-zettelkasten")
    if has_claude:
        required_cmds.append("angelo-claude-hook")
    for cmd in required_cmds:
        path = shutil.which(cmd)
        if path:
            ok(f"{cmd} -> {path}")
        else:
            fail(f"{cmd} not found on PATH - is the package installed in the active environment?")

    # --- Dependencies ---
    print("\nDependencies importable")
    # Import names of the base dependencies. The dashboard web runtime
    # (fastapi/uvicorn/httpx/sklearn) is part of the base install, so a missing
    # one means a broken/partial install, not a skipped extra.
    # model2vec backs semantic search (memory + dashboard + zettelkasten) and
    # is a base dependency, so a missing one means a broken/partial install.
    required_deps = ["mcp", "dulwich", "kglite", "pandas", "yaml", "model2vec",
                     "fastapi", "uvicorn", "httpx", "sklearn"]
    for dep in required_deps:
        if importlib.util.find_spec(dep) is not None:
            ok(dep)
        else:
            fail(f"{dep} missing - reinstall with: pip install angelo")

    if with_zk:
        if importlib.util.find_spec("numpy") is not None:
            ok("numpy (zettelkasten)")
        else:
            fail("numpy missing - install the extra: pip install \"angelo[zettelkasten]\"")

    # --- Native backend integrity (kglite) ---
    # kglite was renamed from distribution `kglite` to `kglite-angelo` (both
    # import as `kglite`). pip does NOT remove the old one on upgrade, so both
    # can end up owning the same kglite.abi3.so - an overlap that SIGSEGVs
    # semantic search (a hard crash that kills the MCP server mid-call, not a
    # catchable error). The plain import check above cannot see this because
    # both dist-infos still import fine; verify exactly one is installed.
    print("\nNative backend (kglite)")
    _check_kglite_backend(ok, warn, fail)
    # pyarrow (if present) shares the process with kglite.abi3.so; an incompatible
    # major (>=24) SIGSEGVs semantic search. Not a kglite dep - only pandas pulls
    # it opportunistically - so this is a separate, silent crash source.
    _check_pyarrow_kglite(ok, warn, fail)

    # --- Embedding model cache ---
    # Semantic search needs the model2vec model (~120MB). It's fetched from our
    # mirror on first use, but until then search degrades to keyword matching -
    # so report whether it's already cached and how to prime it.
    print("\nEmbedding model (semantic search)")
    cached, cache_dir = _embedding_model_cached()
    if cached:
        ok(f"model cached ({cache_dir})")
    elif cache_dir is None:
        warn("could not check the model cache (memory.embedders not importable)")
    else:
        warn("model not cached yet - it downloads (~120MB) on first search, or run "
             "'angelo doctor --fix' to prime it now. Search falls back to keyword "
             "matching until then.")

    # --- Cursor layer ---
    if has_cursor:
        print("\nProject config (Cursor: .cursor/)")
        cursor_dir = project_dir / ".cursor"
        mcp_json = cursor_dir / "mcp.json"
        if mcp_json.exists():
            ok(".cursor/mcp.json present")
            _check_mcp_commands(mcp_json, ok, warn, fail)
        else:
            fail(".cursor/mcp.json missing - run 'angelo init'")

        manifest = _load_manifest(project_dir / CURSOR_MANIFEST_REL)
        stale, modified = [], []
        for src_rel, tgt_rel in _get_cursor_files(with_zk).items():
            tgt = project_dir / tgt_rel
            bundled = (DATA_DIR / src_rel).read_text(encoding="utf-8")
            status = _classify_managed_file(tgt, bundled, manifest, tgt_rel)
            if status == "missing":
                fail(f"{tgt_rel} missing - run 'angelo init' (or 'angelo update')")
            elif status == "ok":
                ok(f"{tgt_rel} up to date")
            elif status == "stale":
                stale.append(tgt_rel)
            else:
                modified.append(tgt_rel)
        for tgt_rel in stale:
            warn(f"{tgt_rel} differs from the installed version - run 'angelo update' to refresh")
        for tgt_rel in modified:
            if _is_cursor_rule(tgt_rel):
                warn(f"{tgt_rel} has local edits - 'angelo update' will offer to migrate them "
                     f"into a '*-local' rule (or '--migrate-edits' to do it non-interactively); "
                     f"otherwise it keeps them and writes {tgt_rel}.new")
            else:
                warn(f"{tgt_rel} has local edits - 'angelo update' will preserve them and "
                     f"write the new version to {tgt_rel}.new")

    # --- Claude Code layer ---
    if has_claude:
        print("\nProject config (Claude Code: .claude/ + .mcp.json)")
        mcp_json = project_dir / ".mcp.json"
        if mcp_json.exists():
            ok(".mcp.json present")
            _check_mcp_commands(mcp_json, ok, warn, fail)
        else:
            fail(".mcp.json missing - run 'angelo init --claude'")

        with_zk_claude = _detect_zettelkasten_claude(project_dir)
        manifest = _load_manifest(project_dir / CLAUDE_MANIFEST_REL)
        stale, modified = [], []
        for src_rel, tgt_rel in _get_claude_files(with_zk_claude).items():
            tgt = project_dir / tgt_rel
            bundled = (DATA_DIR / src_rel).read_text(encoding="utf-8")
            status = _classify_managed_file(tgt, bundled, manifest, tgt_rel)
            if status == "missing":
                fail(f"{tgt_rel} missing - run 'angelo init --claude' (or 'angelo update')")
            elif status == "ok":
                ok(f"{tgt_rel} up to date")
            elif status == "stale":
                stale.append(tgt_rel)
            else:
                modified.append(tgt_rel)
        for tgt_rel in stale:
            warn(f"{tgt_rel} differs from the installed version - run 'angelo update' to refresh")
        for tgt_rel in modified:
            warn(f"{tgt_rel} has local edits - 'angelo update' will preserve them and "
                 f"write the new version to {tgt_rel}.new")

        claude_md = project_dir / "CLAUDE.md"
        if claude_md.exists() and CLAUDE_MD_BEGIN in claude_md.read_text(encoding="utf-8"):
            ok("CLAUDE.md has the angelo import block")
        else:
            warn("CLAUDE.md missing the angelo import block - run 'angelo init --claude' or 'angelo update'")

    # --- Dashboard assets (bundled UI + agent sidecar) ---
    print("\nDashboard assets")
    _check_dashboard_assets("memory.dashboard", "memory dashboard", ok, warn)
    if with_zk:
        _check_dashboard_assets("zettelkasten.dashboard", "zettelkasten dashboard", ok, warn)

    # --- Node.js (optional: dashboard chat sidecar + Vite dev frontend) ---
    # Node can't be pip-installed; the dashboard serves its static prod UI via
    # FastAPI without it, so a missing Node only disables the chat sidecar /
    # source-mode frontend. Report it as a warning, not a hard failure.
    print("\nNode.js (optional, dashboard chat)")
    node = shutil.which("node")
    npm = shutil.which("npm")
    if node:
        try:
            ver = subprocess.run(
                [node, "--version"], capture_output=True, text=True, timeout=10
            ).stdout.strip()
        except (OSError, subprocess.SubprocessError):
            ver = ""
        ok(f"node{(' ' + ver) if ver else ''} -> {node}")
        if npm:
            ok(f"npm -> {npm}")
        else:
            warn("npm not found on PATH - dashboard can't install the chat sidecar's "
                 "node_modules. Install Node.js (npm ships with it) from https://nodejs.org.")
    else:
        warn("node not found on PATH - the dashboard chat sidecar is disabled (the rest "
             "of the dashboard still works). Install Node.js from https://nodejs.org to "
             "enable it.")

    # --- Shared checks ---
    gitignore = project_dir / ".gitignore"
    if gitignore.exists() and GITIGNORE_BEGIN in gitignore.read_text(encoding="utf-8"):
        ok(".gitignore has the angelo block")
    else:
        warn(".gitignore missing the angelo block - run 'angelo init' to add it "
             "(keeps derived caches out of git)")

    print("\nMemory onboarding")
    memory_dir = project_dir / ".memory"
    if (memory_dir / "project.md").exists():
        ok(".memory/ project exists - repo is onboarded")
    else:
        editor_hint = "Cursor or Claude Code" if (has_cursor and has_claude) else (
            "Cursor" if has_cursor else "Claude Code" if has_claude else "your editor"
        )
        warn(f"no .memory/ project yet - open the repo in {editor_hint} and ask the agent to "
             "onboard the project (it calls the memory MCP's create_project tool)")

    if not has_cursor and not has_claude:
        print("\n  [warn] No editor config found. Run 'angelo init' (Cursor) or "
              "'angelo init --claude' (Claude Code).")
        warnings += 1

    print()
    if failures:
        print(f"{failures} problem(s) found, {warnings} warning(s). Fix the [FAIL] items above.")
        sys.exit(1)
    if warnings:
        print(f"No blocking problems. {warnings} warning(s) above are worth a look.")
    else:
        print("Everything looks good.")

cmd_kill

cmd_kill(args: Namespace) -> None

Kill all running angelo MCP processes and their descendant process trees.

The MCP servers lock the installed Scripts/angelo-*.exe wrappers and their children (dashboards, runner daemon, node/uvicorn subprocesses) can hold other locks, which makes a pip install -e reinstall fail. Run this first to clear the way, then reinstall and restart the servers in your editor.

In addition to command-line matching, anything holding a known dashboard port is reaped — this catches orphaned uvicorn --reload spawn workers whose command line (python -c "from multiprocessing.spawn ...") matches no pattern and whose reloader parent has already died.

Source code in angelo_cli/__init__.py

def cmd_kill(args: argparse.Namespace) -> None:
    """Kill all running angelo MCP processes and their descendant process trees.

    The MCP servers lock the installed `Scripts/angelo-*.exe` wrappers and their
    children (dashboards, runner daemon, node/uvicorn subprocesses) can hold
    other locks, which makes a `pip install -e` reinstall fail. Run this first to
    clear the way, then reinstall and restart the servers in your editor.

    In addition to command-line matching, anything holding a known dashboard
    port is reaped — this catches orphaned uvicorn --reload spawn workers whose
    command line (`python -c "from multiprocessing.spawn ..."`) matches no
    pattern and whose reloader parent has already died.
    """
    procs = _enumerate_processes()
    if not procs:
        print("Could not enumerate processes (no psutil, and the shell fallback "
              "returned nothing). Nothing killed.", file=sys.stderr)
        sys.exit(1)

    # Command-pattern matches get their full descendant tree reaped; port-only
    # seeds are gated on a dashboard-worker pattern so an unrelated process that
    # merely shares one of our ports (and its children) is not killed.
    port_pids = _listening_pids_for_ports(_dashboard_ports())
    targets = _kill_targets(procs, port_pids)

    targets -= _ancestor_pids(os.getpid(), procs)

    if not targets:
        print("No angelo processes running.")
        return

    cmd_by_pid = {pid: cmd for pid, _, cmd in procs}

    def _short(cmd: str) -> str:
        cmd = (cmd or "").strip()
        return cmd if len(cmd) <= 100 else cmd[:97] + "..."

    if args.dry_run:
        print(f"Would kill {len(targets)} angelo process(es):")
        for pid in sorted(targets):
            print(f"  [{pid}] {_short(cmd_by_pid.get(pid, '?'))}")
        return

    killed = 0
    for pid in sorted(targets):
        if _kill_pid(pid):
            killed += 1
            print(f"  killed [{pid}] {_short(cmd_by_pid.get(pid, '?'))}")
        else:
            print(f"  could not kill [{pid}] {_short(cmd_by_pid.get(pid, '?'))}",
                  file=sys.stderr)

    print(f"\nKilled {killed}/{len(targets)} angelo process(es). "
          "Now reinstall (pip install -e \".[zettelkasten,dashboard]\") and "
          "restart the MCP servers in your editor.")

commit_dirty_memory

commit_dirty_memory(*, no_push: bool = False, quiet: bool = False, dry_run: bool = False) -> int

Commit dirty .memory/ files in isolation. Returns the number committed.

Discovers modified/untracked files under .memory/ and commits them as the memory-mcp author without touching the user's staged changes, then safe-auto-pushes (only when every unpushed commit is memory-authored). This lets a code commit/push also flush any memory entries a separate MCP-server process hasn't committed yet. Shared by the angelo memory commit CLI and the Claude Code Bash hook. Never raises — safe to call from any hook.

Source code in angelo_cli/__init__.py

def commit_dirty_memory(*, no_push: bool = False, quiet: bool = False, dry_run: bool = False) -> int:
    """Commit dirty .memory/ files in isolation. Returns the number committed.

    Discovers modified/untracked files under ``.memory/`` and commits them as
    the ``memory-mcp`` author without touching the user's staged changes, then
    safe-auto-pushes (only when every unpushed commit is memory-authored). This
    lets a code commit/push also flush any memory entries a separate MCP-server
    process hasn't committed yet. Shared by the ``angelo memory commit`` CLI and
    the Claude Code Bash hook. Never raises — safe to call from any hook.
    """
    try:
        root_out = subprocess.run(
            ["git", "rev-parse", "--show-toplevel"],
            capture_output=True, text=True, check=True,
        ).stdout.strip()
    except Exception:
        return 0  # not a git repo / git unavailable — nothing to do
    root = Path(root_out)
    paths = _dirty_memory_paths(root)
    if not paths:
        return 0
    if dry_run:
        print(f"[memory] would commit {len(paths)} file(s):")
        for p in paths:
            print(f"  {p}")
        return 0
    try:
        from memory.commit import MemoryCommitter
    except Exception:
        return 0
    committer = MemoryCommitter(auto_push=not no_push)
    # commit_now opens DulwichRepo(".") and stages repo-relative paths, so run
    # from the repo root regardless of where the hook fired.
    cwd = os.getcwd()
    try:
        os.chdir(root)
        n = committer.commit_now(paths)
    except Exception:
        return 0
    finally:
        os.chdir(cwd)
    if n and not quiet:
        print(f"[memory] committed {n} file(s)")
    return n

commit_dirty_zettel

commit_dirty_zettel(*, quiet: bool = False, dry_run: bool = False) -> int

Commit dirty .zettelkasten/ files in isolation. Returns the number committed.

Parallel to :func:commit_dirty_memory: discovers modified/untracked files under .zettelkasten/ and commits them as the zettelkasten-mcp author via :meth:ZettelCommitter.commit_now (no push — local history is enough for the zettelkasten). This lets a user code commit/push (or the memory git hook) also flush any review/note files a separate MCP-server process scheduled but hasn't committed yet — the debounce timer + atexit alone are lost on a hard kill (SIGKILL). Never raises — safe to call from any hook.

Source code in angelo_cli/__init__.py

def commit_dirty_zettel(*, quiet: bool = False, dry_run: bool = False) -> int:
    """Commit dirty .zettelkasten/ files in isolation. Returns the number committed.

    Parallel to :func:`commit_dirty_memory`: discovers modified/untracked files
    under ``.zettelkasten/`` and commits them as the ``zettelkasten-mcp`` author
    via :meth:`ZettelCommitter.commit_now` (no push — local history is enough for
    the zettelkasten). This lets a user code commit/push (or the memory git hook)
    also flush any review/note files a separate MCP-server process scheduled but
    hasn't committed yet — the debounce timer + ``atexit`` alone are lost on a
    hard kill (SIGKILL). Never raises — safe to call from any hook.
    """
    try:
        root_out = subprocess.run(
            ["git", "rev-parse", "--show-toplevel"],
            capture_output=True, text=True, check=True,
        ).stdout.strip()
    except Exception:
        return 0  # not a git repo / git unavailable — nothing to do
    root = Path(root_out)
    paths = _dirty_zettel_paths(root)
    if not paths:
        return 0
    if dry_run:
        print(f"[zettelkasten] would commit {len(paths)} file(s):")
        for p in paths:
            print(f"  {p}")
        return 0
    try:
        from zettelkasten.commit import ZettelCommitter
    except Exception:
        return 0
    committer = ZettelCommitter()
    # commit_now discovers each file's repo and stages its resolved path, so pass
    # absolute paths and run from the repo root regardless of where the hook fired.
    cwd = os.getcwd()
    try:
        os.chdir(root)
        n = committer.commit_now([str(root / p) for p in paths])
    except Exception:
        return 0
    finally:
        os.chdir(cwd)
    if n and not quiet:
        print(f"[zettelkasten] committed {n} file(s)")
    return n

cmd_memory_commit

cmd_memory_commit(args: Namespace) -> None

CLI wrapper around :func:commit_dirty_memory (the git-commit hook target).

Also flushes pending .zettelkasten/ files on the same hook so weeks of review/note authoring ride up with a user git commit/push, not just on the committer's atexit (which is skipped on a hard kill). Always exits 0 so it can never break the triggering shell hook.

Source code in angelo_cli/__init__.py

def cmd_memory_commit(args: argparse.Namespace) -> None:
    """CLI wrapper around :func:`commit_dirty_memory` (the git-commit hook target).

    Also flushes pending ``.zettelkasten/`` files on the same hook so weeks of
    review/note authoring ride up with a user git commit/push, not just on the
    committer's ``atexit`` (which is skipped on a hard kill). Always exits 0 so
    it can never break the triggering shell hook.
    """
    commit_dirty_memory(
        no_push=getattr(args, "no_push", False),
        quiet=getattr(args, "quiet", False),
        dry_run=getattr(args, "dry_run", False),
    )
    commit_dirty_zettel(
        quiet=getattr(args, "quiet", False),
        dry_run=getattr(args, "dry_run", False),
    )

cmd_memory_tether_hint

cmd_memory_tether_hint(args: Namespace) -> None

Cursor postToolUse hook target: emit a non-blocking tether reminder.

Cursor's preToolUse has no advisory-on-allow channel (a message reaches the agent only on deny), so the code-aware-editing reflex cannot be a non-blocking pre-edit nudge there. Instead this fires on postToolUse for Write and prints Cursor's {"additional_context": ...} shape — reminding the agent to record(..., files=...) after a code edit so the symbol tether advances. The decision (code file vs not) is deterministic, made in :func:claude_hooks.decide_post_edit_context; it never blocks and always exits 0 so a malformed payload can never break the triggering tool call.

Source code in angelo_cli/__init__.py

def cmd_memory_tether_hint(args: argparse.Namespace) -> None:
    """Cursor postToolUse hook target: emit a non-blocking tether reminder.

    Cursor's ``preToolUse`` has no advisory-on-allow channel (a message reaches
    the agent only on ``deny``), so the code-aware-editing reflex cannot be a
    non-blocking *pre*-edit nudge there. Instead this fires on ``postToolUse`` for
    ``Write`` and prints Cursor's ``{"additional_context": ...}`` shape — reminding
    the agent to ``record(..., files=...)`` after a code edit so the symbol tether
    advances. The decision (code file vs not) is deterministic, made in
    :func:`claude_hooks.decide_post_edit_context`; it never blocks and always exits
    0 so a malformed payload can never break the triggering tool call.
    """
    try:
        from angelo_cli.claude_hooks import decide_post_edit_context

        raw = sys.stdin.read()
        hook_input = json.loads(raw) if raw.strip() else {}
        if not isinstance(hook_input, dict):
            hook_input = {}
        context = decide_post_edit_context(hook_input)
        if context:
            json.dump({"additional_context": context}, sys.stdout)
            sys.stdout.write("\n")
    except Exception:
        # Fail open: a hook must never raise into the agent's tool pipeline.
        pass

cmd_memory_publish

cmd_memory_publish(args: Namespace) -> None

Build and commit a filtered, per-recipient-encrypted memory bundle.

Reads the owner-authored sharing policy (sharing/grants in .memory/config.yaml) and the public-key identity registry, then calls :func:memory.sharing.build_bundle to write the ciphertext bundle into a producer-owned bundle repo. The bundle files are committed in that repo (as the memory-mcp author, isolated from the user's working tree); push is off unless --push is given.

Source code in angelo_cli/__init__.py

def cmd_memory_publish(args: argparse.Namespace) -> None:
    """Build and commit a filtered, per-recipient-encrypted memory bundle.

    Reads the owner-authored sharing policy (``sharing``/``grants`` in
    ``.memory/config.yaml``) and the public-key identity registry, then calls
    :func:`memory.sharing.build_bundle` to write the ciphertext bundle into a
    producer-owned bundle repo. The bundle files are committed in that repo (as
    the ``memory-mcp`` author, isolated from the user's working tree); push is
    off unless ``--push`` is given.
    """
    from memory import storage

    config = storage.read_config()
    sharing_block = config.get("sharing") if isinstance(config.get("sharing"), dict) else {}

    bundle_dir = _memory_bundle_dir(args, sharing_block)
    if bundle_dir is None:
        print(
            "error: no bundle repo given; pass --bundle-repo PATH or set "
            "sharing.bundle_repo in .memory/config.yaml",
            file=sys.stderr,
        )
        sys.exit(1)

    result, repo_root, committed = _memory_publish_bundle(
        args, config, sharing_block, bundle_dir
    )

    print(
        f"[publish] wrote {result.entry_unit_count} encrypted entr"
        f"{'y' if result.entry_unit_count == 1 else 'ies'} to {bundle_dir}"
    )
    if result.recipient_ids:
        print(f"[publish] recipients: {', '.join(result.recipient_ids)}")
    if result.skipped_unregistered_ids:
        print(
            "[publish] warning: skipped unregistered recipient id(s): "
            f"{', '.join(result.skipped_unregistered_ids)}"
        )
    if committed:
        print(f"[publish] committed {committed} bundle file(s) in {repo_root}")

cmd_memory_identity_init

cmd_memory_identity_init(args: Namespace) -> None

Create this machine's local identity (and optionally register it).

Minimal bootstrap so publish/sync are usable end-to-end: generates a keypair, writes the PRIVATE identity to ~/.angelo/identity.yaml (chmod 0600, never committed), and — with --register — adds the PUBLIC key to the shared registry (.memory/identities.yaml, safe to commit).

Source code in angelo_cli/__init__.py

def cmd_memory_identity_init(args: argparse.Namespace) -> None:
    """Create this machine's local identity (and optionally register it).

    Minimal bootstrap so ``publish``/``sync`` are usable end-to-end: generates a
    keypair, writes the PRIVATE identity to ``~/.angelo/identity.yaml`` (chmod
    0600, never committed), and — with ``--register`` — adds the PUBLIC key to
    the shared registry (``.memory/identities.yaml``, safe to commit).
    """
    from memory import sharing

    identity_path = getattr(args, "identity_path", None)
    try:
        identity = sharing.create_identity(
            args.id,
            display_name=getattr(args, "display_name", None),
            github=getattr(args, "github", None),
            path=identity_path,
            overwrite=getattr(args, "overwrite", False),
        )
    except FileExistsError as exc:
        print(f"error: {exc}", file=sys.stderr)
        sys.exit(1)

    dest = Path(identity_path) if identity_path else sharing.default_identity_path()
    print(f"[identity] created identity '{identity.id}' at {dest}")
    print(f"[identity] public key: {identity.public_key}")
    if identity.signing_public_key:
        print(f"[identity] signing public key: {identity.signing_public_key}")

    if getattr(args, "register", False):
        registry_path = getattr(args, "registry_path", None)
        registry = sharing.load_registry(registry_path)
        registry = registry.with_entry(identity.to_registry_entry())
        saved = sharing.save_registry(registry, registry_path)
        print(f"[identity] registered public key for '{identity.id}' in {saved}")

cmd_memory_revoke

cmd_memory_revoke(args: Namespace) -> None

Revoke a recipient id from the memory sharing policy and republish.

Edits .memory/config.yaml to drop --id (optionally confined to a --scope subtree) via :func:memory.sharing.grants_edit.revoke_in_store, then — if anything changed — re-publishes the bundle so the new ciphertext excludes them (forward revocation). With --purge-history it also rewrites the bundle repo history (and force-pushes when a remote exists). The source config edit is deliberately NOT auto-committed — the report reminds the owner to review + commit it.

Source code in angelo_cli/__init__.py

def cmd_memory_revoke(args: argparse.Namespace) -> None:
    """Revoke a recipient id from the memory sharing policy and republish.

    Edits ``.memory/config.yaml`` to drop ``--id`` (optionally confined to a
    ``--scope`` subtree) via :func:`memory.sharing.grants_edit.revoke_in_store`,
    then — if anything changed — re-publishes the bundle so the new ciphertext
    excludes them (forward revocation). With ``--purge-history`` it also rewrites
    the bundle repo history (and force-pushes when a remote exists). The source
    config edit is deliberately NOT auto-committed — the report reminds the owner
    to review + commit it.
    """
    from memory import sharing, storage
    from memory.sharing.grants_edit import RevocationError, revoke_in_store

    prefix = "[memory revoke]"
    revoked_id = args.id
    scope = getattr(args, "scope", None)

    # Snapshot the pre-edit config + resolve the bundle repo BEFORE mutating
    # anything, so a later republish failure can roll the config back cleanly.
    config_before = storage.read_config()
    sharing_block = config_before.get("sharing") if isinstance(config_before.get("sharing"), dict) else {}
    bundle_dir = _memory_bundle_dir(args, sharing_block)

    registry = sharing.load_registry()
    try:
        change = revoke_in_store(revoked_id, scope=scope, registry=registry)
    except RevocationError as exc:
        print(f"{prefix} error: {exc}", file=sys.stderr)
        sys.exit(1)

    scope_desc = f" within scope '{scope}'" if scope else ""
    if not change.changed:
        if change.residual_inherited:
            # Nothing to edit in-scope, but the id still reaches the subtree via
            # an ancestor grant/scope — say so instead of "nothing to revoke".
            print(
                f"{prefix} nothing to edit in scope '{scope}', but '{revoked_id}' "
                "STILL has access there, inherited from ancestor "
                f"grant/scope key(s): {', '.join(change.residual_inherited)}."
            )
            print(
                f"{prefix} to fully cut them off, revoke at that ancestor node "
                "(--scope <ancestor>) or globally (no --scope)."
            )
        else:
            print(
                f"{prefix} '{revoked_id}' is not granted anything{scope_desc}; "
                "nothing to revoke (no config change, no republish)."
            )
        return

    print(f"{prefix} revoked '{revoked_id}'{scope_desc} from the sharing policy")
    if change.grants_removed_from:
        print(f"{prefix}   grants dropped from: {', '.join(change.grants_removed_from)}")
    if change.emptied_grant_keys:
        print(f"{prefix}   grant keys emptied + removed: {', '.join(change.emptied_grant_keys)}")
    if change.firm_converted:
        print(f"{prefix}   firm -> bilateral (minus id): {', '.join(change.firm_converted)}")
        print(f"{prefix}   note: firm->bilateral FREEZES the audience — future firm "
              "members will NOT receive these nodes until you re-widen them")
    if change.bilateral_dropped_from:
        print(f"{prefix}   dropped from bilateral audiences: {', '.join(change.bilateral_dropped_from)}")
    if change.emptied_scope_keys:
        print(f"{prefix}   scope keys emptied + removed (fall back to private): "
              f"{', '.join(change.emptied_scope_keys)}")
    if change.skills_default_touched:
        print(f"{prefix}   sharing.skills default edited")
    if change.residual_inherited:
        print(f"{prefix}   WARNING: '{revoked_id}' STILL has inherited access within "
              f"scope '{scope}' via ancestor key(s): {', '.join(change.residual_inherited)}")
        print(f"{prefix}            revoke at that ancestor node or globally to fully cut them off")

    republished = False
    purged = False
    if bundle_dir is None:
        print(
            f"{prefix} policy updated but no bundle repo is configured "
            "(pass --bundle-repo PATH or set sharing.bundle_repo) — bundle NOT "
            "republished; run 'angelo memory publish' once a bundle repo is set.",
            file=sys.stderr,
        )
    else:
        config_after = storage.read_config()
        try:
            result, repo_root, committed = _memory_publish_bundle(
                args, config_after, sharing_block, bundle_dir
            )
        except Exception as exc:
            # Atomicity: republish failed, so restore the pre-edit config rather
            # than leaving a dropped grant on disk while the bundle still holds
            # the old ciphertext. No traceback — an honest one-line error.
            storage.write_config(config_before)
            print(
                f"{prefix} error: republish failed, so the config edit was REVERTED "
                f"(no change applied): {exc}",
                file=sys.stderr,
            )
            sys.exit(1)
        republished = True
        print(
            f"{prefix} republished bundle to {bundle_dir}: "
            f"{result.entry_unit_count} encrypted entr"
            f"{'y' if result.entry_unit_count == 1 else 'ies'}"
            + (f" for {', '.join(result.recipient_ids)}" if result.recipient_ids else "")
        )
        if committed:
            print(f"{prefix} committed {committed} bundle file(s) in {repo_root}")
        if getattr(args, "purge_history", False):
            purged = _purge_bundle_and_push(
                prefix, bundle_dir, source_repo_root=storage._workspace_root()
            )

    _print_memory_cutoff(prefix, revoked_id, change, republished, purged)
    _print_revoke_footer(prefix, ".memory/config.yaml")

cmd_zettelkasten_publish

cmd_zettelkasten_publish(args: Namespace) -> None

Build and commit a filtered, per-recipient-encrypted zettelkasten bundle.

Reads the owner-authored sharing policy (sharing/grants in .zettelkasten/config.yaml) and the SHARED public-key identity registry (.memory/identities.yaml — the same firm registry memory uses), then calls :func:zettelkasten.sharing.build_bundle to write the ciphertext bundle into a producer-owned bundle repo. The bundle files are committed in that repo (as the memory-mcp author, isolated from the user's working tree); push is off unless --push is given.

Source code in angelo_cli/__init__.py

def cmd_zettelkasten_publish(args: argparse.Namespace) -> None:
    """Build and commit a filtered, per-recipient-encrypted zettelkasten bundle.

    Reads the owner-authored sharing policy (``sharing``/``grants`` in
    ``.zettelkasten/config.yaml``) and the SHARED public-key identity registry
    (``.memory/identities.yaml`` — the same firm registry memory uses), then
    calls :func:`zettelkasten.sharing.build_bundle` to write the ciphertext
    bundle into a producer-owned bundle repo. The bundle files are committed in
    that repo (as the ``memory-mcp`` author, isolated from the user's working
    tree); push is off unless ``--push`` is given.
    """
    from zettelkasten import config as zk_config

    config = zk_config.read_config()
    sharing_block = config.get("sharing") if isinstance(config.get("sharing"), dict) else {}

    bundle_repo = getattr(args, "bundle_repo", None) or sharing_block.get("bundle_repo")
    if not bundle_repo:
        print(
            "error: no bundle repo given; pass --bundle-repo PATH or set "
            "sharing.bundle_repo in .zettelkasten/config.yaml",
            file=sys.stderr,
        )
        sys.exit(1)
    bundle_dir = _resolve_workspace_path(str(bundle_repo))
    bundle_dir.mkdir(parents=True, exist_ok=True)

    result, repo_root, committed = _zettelkasten_publish_bundle(
        args, config, sharing_block, bundle_dir
    )

    total = (
        result.note_unit_count
        + result.source_unit_count
        + result.project_unit_count
        + result.citation_unit_count
        + result.review_unit_count
        + result.org_unit_count
        + result.table_unit_count
        + result.outline_unit_count
        + result.sidecar_unit_count
    )
    print(
        f"[zk publish] wrote {total} encrypted unit{'' if total == 1 else 's'} "
        f"({result.note_unit_count} note, {result.source_unit_count} source, "
        f"{result.project_unit_count} project, {result.citation_unit_count} citation, "
        f"{result.review_unit_count} review, {result.org_unit_count} organization, "
        f"{result.table_unit_count} table, {result.outline_unit_count} outline, "
        f"{result.sidecar_unit_count} sidecar) "
        f"to {bundle_dir}"
    )
    if result.recipient_ids:
        print(f"[zk publish] recipients: {', '.join(result.recipient_ids)}")
    if result.skipped_unregistered_ids:
        print(
            "[zk publish] warning: skipped unregistered recipient id(s): "
            f"{', '.join(result.skipped_unregistered_ids)}"
        )
    ri_n = result.recipient_index_note_count
    if ri_n:
        print(
            f"[zk publish] recipient-index export: {ri_n} note"
            f"{'' if ri_n == 1 else 's'} (firm-wide, enables cross-producer edges)"
        )
    if committed:
        print(f"[zk publish] committed {committed} bundle file(s) in {repo_root}")

cmd_zettelkasten_revoke

cmd_zettelkasten_revoke(args: Namespace) -> None

Revoke a recipient id from the zettelkasten sharing policy and republish.

Edits .zettelkasten/config.yaml to drop --id (optionally confined to a --project closure) via :func:zettelkasten.sharing.grants_edit.revoke_in_store, then — if anything changed — re-publishes the bundle so the new ciphertext excludes them. With --purge-history it also rewrites the bundle repo history (and force-pushes when a remote exists). The source config edit is deliberately NOT auto-committed — the report reminds the owner to review + commit it.

Source code in angelo_cli/__init__.py

def cmd_zettelkasten_revoke(args: argparse.Namespace) -> None:
    """Revoke a recipient id from the zettelkasten sharing policy and republish.

    Edits ``.zettelkasten/config.yaml`` to drop ``--id`` (optionally confined to
    a ``--project`` closure) via
    :func:`zettelkasten.sharing.grants_edit.revoke_in_store`, then — if anything
    changed — re-publishes the bundle so the new ciphertext excludes them. With
    ``--purge-history`` it also rewrites the bundle repo history (and force-pushes
    when a remote exists). The source config edit is deliberately NOT
    auto-committed — the report reminds the owner to review + commit it.
    """
    from memory import sharing as mem_sharing
    from zettelkasten import config as zk_config
    from zettelkasten.sharing.grants_edit import RevocationError, revoke_in_store

    prefix = "[zk revoke]"
    revoked_id = args.id
    project = getattr(args, "project", None)

    # Snapshot the pre-edit config + resolve the bundle repo BEFORE mutating
    # anything, so a later republish failure can roll the config back cleanly.
    config_before = zk_config.read_config()
    sharing_block = config_before.get("sharing") if isinstance(config_before.get("sharing"), dict) else {}
    bundle_repo = getattr(args, "bundle_repo", None) or sharing_block.get("bundle_repo")

    registry = mem_sharing.load_registry()
    try:
        change = revoke_in_store(revoked_id, project=project, registry=registry)
    except RevocationError as exc:
        print(f"{prefix} error: {exc}", file=sys.stderr)
        sys.exit(1)

    scope_desc = f" within project '{project}'" if project else ""
    if not change.changed:
        print(
            f"{prefix} '{revoked_id}' is not granted anything{scope_desc}; "
            "nothing to revoke (no config change, no republish)."
        )
        if change.cross_project_shared:
            print(f"{prefix} note: '{revoked_id}' retains access via shared key(s) "
                  f"{', '.join(change.cross_project_shared)} (also used by another "
                  "project); revoke globally or from the sharing project to cut that off.")
        if change.residual:
            print(f"{prefix} note: '{revoked_id}' remains granted on key(s) outside "
                  f"project '{project}': {', '.join(change.residual)}; revoke globally "
                  "or from those projects to cut that off.")
        if change.residual_undetermined:
            print(f"{prefix} note: residual access UNDETERMINED (identity registry "
                  f"empty/unavailable) for out-of-project firm-scoped key(s): "
                  f"{', '.join(change.residual_undetermined)}; if '{revoked_id}' is a "
                  "firm member they would be re-granted on the next publish. Fix the "
                  "registry (.memory/identities.yaml) or revoke globally to be sure.")
        if change.closure_incomplete:
            print(f"{prefix} warning: project '{project}' closure is INCOMPLETE — a note "
                  "could not be parsed, so its keys were not checked; '{}' may still be "
                  "granted on them. Fix the note(s) (see 'project-closure' warnings) and "
                  "re-run, or revoke globally to be sure.".format(revoked_id))
        return

    print(f"{prefix} revoked '{revoked_id}'{scope_desc} from the sharing policy")
    if change.grants_removed_from:
        print(f"{prefix}   grants dropped from: {', '.join(change.grants_removed_from)}")
    if change.firm_converted:
        print(f"{prefix}   firm -> bilateral (minus id): {', '.join(change.firm_converted)}")
        print(f"{prefix}   note: firm->bilateral FREEZES the audience — future firm "
              "members will NOT receive these units until you re-widen them")
    if change.audiences_dropped_from:
        print(f"{prefix}   dropped from bilateral audiences: {', '.join(change.audiences_dropped_from)}")
    if change.keys_removed:
        print(f"{prefix}   keys emptied + removed: {', '.join(change.keys_removed)}")
    if change.cross_project_shared:
        print(f"{prefix}   NOT revoked from cross-project shared key(s) (another project "
              f"grants them): {', '.join(change.cross_project_shared)}")
        print(f"{prefix}            '{revoked_id}' retains access via these until revoked "
              "globally or from the sharing project")
    if change.residual:
        print(f"{prefix}   STILL granted on key(s) outside project '{project}': "
              f"{', '.join(change.residual)}")
        print(f"{prefix}            '{revoked_id}' remains an effective bundle recipient "
              "via these until revoked globally or from those projects")
    if change.residual_undetermined:
        print(f"{prefix}   residual UNDETERMINED (identity registry empty/unavailable) "
              f"for out-of-project firm-scoped key(s): {', '.join(change.residual_undetermined)}")
        print(f"{prefix}            cannot confirm '{revoked_id}' is cut off — a firm "
              "member would be re-granted on the next publish; fix the registry "
              "(.memory/identities.yaml) or revoke globally")

    republished = False
    purged = False
    if not bundle_repo:
        print(
            f"{prefix} policy updated but no bundle repo is configured "
            "(pass --bundle-repo PATH or set sharing.bundle_repo) — bundle NOT "
            "republished; run 'angelo zettelkasten publish' once a bundle repo is set.",
            file=sys.stderr,
        )
    else:
        bundle_dir = _resolve_workspace_path(str(bundle_repo))
        bundle_dir.mkdir(parents=True, exist_ok=True)
        config_after = zk_config.read_config()
        try:
            result, repo_root, committed = _zettelkasten_publish_bundle(
                args, config_after, sharing_block, bundle_dir
            )
        except Exception as exc:
            # Atomicity: republish failed → restore the pre-edit config so a
            # dropped grant is never left on disk against stale ciphertext.
            zk_config.write_config(config_before)
            print(
                f"{prefix} error: republish failed, so the config edit was REVERTED "
                f"(no change applied): {exc}",
                file=sys.stderr,
            )
            sys.exit(1)
        republished = True
        total = (
            result.note_unit_count
            + result.source_unit_count
            + result.project_unit_count
            + result.citation_unit_count
            + result.review_unit_count
            + result.org_unit_count
            + result.table_unit_count
            + result.outline_unit_count
            + result.sidecar_unit_count
        )
        print(
            f"{prefix} republished bundle to {bundle_dir}: "
            f"{total} encrypted unit{'' if total == 1 else 's'}"
            + (f" for {', '.join(result.recipient_ids)}" if result.recipient_ids else "")
        )
        if committed:
            print(f"{prefix} committed {committed} bundle file(s) in {repo_root}")
        if getattr(args, "purge_history", False):
            from memory import storage as _mem_storage

            purged = _purge_bundle_and_push(
                prefix, bundle_dir, source_repo_root=_mem_storage._workspace_root()
            )

    if (change.cross_project_shared or change.residual
            or change.residual_undetermined or change.closure_incomplete):
        # Residual access preserved (or undetermined) OUTSIDE the revoked
        # project's closure — via an in-closure shared key (cross_project_shared),
        # an out-of-project grant/scope (residual), or an out-of-project firm
        # scope we cannot evaluate because the registry is empty/unavailable
        # (residual_undetermined). Or the closure itself was INCOMPLETE
        # (unparseable note) so some in-project keys were never edited
        # (closure_incomplete). Mirror memory's residual handling and do NOT
        # claim a clean cut-off: the id may still be an effective bundle recipient.
        print(f"{prefix} '{revoked_id}' was removed from project '{project}', but is "
              "NOT fully cut off — it remains (or may remain) an effective bundle recipient:")
        if change.cross_project_shared:
            print(f"{prefix}   - via cross-project shared key(s): "
                  f"{', '.join(change.cross_project_shared)}")
        if change.residual:
            print(f"{prefix}   - via key(s) outside project '{project}': "
                  f"{', '.join(change.residual)}")
        if change.residual_undetermined:
            print(f"{prefix}   - UNDETERMINED (identity registry empty/unavailable) via "
                  f"out-of-project firm-scoped key(s): "
                  f"{', '.join(change.residual_undetermined)} — a firm member would be "
                  "re-granted on the next publish")
        if change.closure_incomplete:
            print(f"{prefix}   - INCOMPLETE closure: a note in project '{project}' could "
                  "not be parsed, so its keys were NOT edited; '{}' may still be granted "
                  "on them. Fix the note(s) (see the 'project-closure' warnings) and "
                  "re-run.".format(revoked_id))
        print(f"{prefix}            revoke globally (no --project) or from the sharing "
              "project(s) to fully cut them off"
              + (" (and fix the identity registry)." if change.residual_undetermined else "."))
    elif republished:
        print(f"{prefix} '{revoked_id}' is now CUT OFF FROM:")
        print(f"{prefix}   - the current bundle and all future bundles (new ciphertext excludes them)")
        if purged:
            print(f"{prefix}   - the remote's superseded ciphertext (purged + force-pushed where a remote exists)")
    else:
        print(f"{prefix} policy updated; bundle NOT republished — '{revoked_id}' will be cut "
              "off from FUTURE bundles once you run 'angelo zettelkasten publish'.")
    _print_revoke_footer(prefix, ".zettelkasten/config.yaml")

cmd_synapse_publish

cmd_synapse_publish(args: Namespace) -> None

Build and commit a filtered, per-recipient-encrypted synapse-edge bundle.

Reads the synapse sharing plumbing (sharing.bundle_repo/producer_id in .synapse/config.yaml) and the SHARED public-key identity registry (.memory/identities.yaml — the same firm registry memory/zk use), then calls :func:zettelkasten.synapse.sharing.build_bundle to write the ciphertext bundle into a producer-owned bundle repo. Each cross-store edge ships only to recipients who can open BOTH endpoints (the intersection of the memory + ZK grants). The bundle files are committed in that repo (as the memory-mcp author, isolated from the user's working tree); push is off unless --push.

Source code in angelo_cli/__init__.py

def cmd_synapse_publish(args: argparse.Namespace) -> None:
    """Build and commit a filtered, per-recipient-encrypted synapse-edge bundle.

    Reads the synapse sharing plumbing (``sharing.bundle_repo``/``producer_id`` in
    ``.synapse/config.yaml``) and the SHARED public-key identity registry
    (``.memory/identities.yaml`` — the same firm registry memory/zk use), then
    calls :func:`zettelkasten.synapse.sharing.build_bundle` to write the ciphertext
    bundle into a producer-owned bundle repo. Each cross-store edge ships only to
    recipients who can open BOTH endpoints (the intersection of the memory + ZK
    grants). The bundle files are committed in that repo (as the ``memory-mcp``
    author, isolated from the user's working tree); push is off unless ``--push``.
    """
    from memory.commit import MemoryCommitter
    from memory import sharing as mem_sharing
    from zettelkasten.synapse import config as syn_config
    from zettelkasten.synapse import sharing as syn_sharing

    config = syn_config.read_config()
    sharing_block = config.get("sharing") if isinstance(config.get("sharing"), dict) else {}

    bundle_repo = getattr(args, "bundle_repo", None) or sharing_block.get("bundle_repo")
    if not bundle_repo:
        print(
            "error: no bundle repo given; pass --bundle-repo PATH or set "
            "sharing.bundle_repo in .synapse/config.yaml",
            file=sys.stderr,
        )
        sys.exit(1)
    bundle_dir = _resolve_workspace_path(str(bundle_repo))
    bundle_dir.mkdir(parents=True, exist_ok=True)
    repo_root = _ensure_git_repo(bundle_dir)

    producer_id = getattr(args, "producer_id", None) or sharing_block.get("producer_id")
    if not producer_id:
        local = mem_sharing.load_local_identity()
        producer_id = local.id if local is not None else None

    # Opt-in fail-closed staleness TTL: --max-policy-age-hours overrides
    # sharing.recipient_index_max_age_hours; unset on both = default warn-only.
    sharing_cfg = syn_sharing.parse_sharing_config(config)
    max_age_hours = getattr(args, "max_policy_age_hours", None)
    if max_age_hours is None:
        max_age_hours = sharing_cfg.recipient_index_max_age_hours

    # The identity registry is shared with memory + zk (one firm registry).
    registry = mem_sharing.load_registry()
    try:
        result = syn_sharing.build_bundle(
            bundle_dir,
            registry=registry,
            producer_id=producer_id,
            max_policy_age_hours=max_age_hours,
        )
    except syn_sharing.StalePeerPolicyError as exc:
        # Fail-closed: refuse the publish, leaving any existing bundle untouched.
        print(str(exc), file=sys.stderr)
        sys.exit(1)

    # Commit exactly the bundle artefacts (manifest + units dir). The units
    # directory is committed as a pathspec so removed/revoked units are staged
    # as deletions too.
    paths = [
        str(bundle_dir / "manifest.json"),
        str(bundle_dir / "units"),
    ]
    committer = MemoryCommitter(auto_push=getattr(args, "push", False))
    cwd = os.getcwd()
    committed = 0
    try:
        os.chdir(repo_root)
        committed = committer.commit_now(paths)
    finally:
        os.chdir(cwd)

    n = result.edge_unit_count
    print(
        f"[synapse publish] wrote {n} encrypted edge{'' if n == 1 else 's'} "
        f"to {bundle_dir}"
    )
    if result.recipient_ids:
        print(f"[synapse publish] recipients: {', '.join(result.recipient_ids)}")
    if result.skipped_unregistered_ids:
        print(
            "[synapse publish] warning: skipped unregistered recipient id(s): "
            f"{', '.join(result.skipped_unregistered_ids)}"
        )
    if committed:
        print(f"[synapse publish] committed {committed} bundle file(s) in {repo_root}")

cmd_federation_sync

cmd_federation_sync(args: Namespace) -> None

Fetch and decrypt subscribed bundle repos into the federation cache.

For each bundle_subscriptions entry in .memory/config.yaml (a local path or a git remote), fetch/clone the bundle repo if needed, then decrypt with the local identity into .angelo/federation-cache/<peer_id>/.memory so it renders like any other federated peer. Zettelkasten bundle subscriptions (bundle_subscriptions in .zettelkasten/config.yaml) are ALSO synced, decrypted into .angelo/federation-cache/<peer_id>/.zettelkasten; synapse bundle subscriptions (bundle_subscriptions in .synapse/config.yaml) are ALSO synced, decrypted into .angelo/federation-cache/<peer_id>/.synapse. Absent config = nothing to sync.

Source code in angelo_cli/__init__.py

def cmd_federation_sync(args: argparse.Namespace) -> None:
    """Fetch and decrypt subscribed bundle repos into the federation cache.

    For each ``bundle_subscriptions`` entry in ``.memory/config.yaml`` (a local
    ``path`` or a git ``remote``), fetch/clone the bundle repo if needed, then
    decrypt with the local identity into
    ``.angelo/federation-cache/<peer_id>/.memory`` so it renders like any other
    federated peer. Zettelkasten bundle subscriptions (``bundle_subscriptions``
    in ``.zettelkasten/config.yaml``) are ALSO synced, decrypted into
    ``.angelo/federation-cache/<peer_id>/.zettelkasten``; synapse bundle
    subscriptions (``bundle_subscriptions`` in ``.synapse/config.yaml``) are ALSO
    synced, decrypted into ``.angelo/federation-cache/<peer_id>/.synapse``. Absent
    config = nothing to sync.
    """
    from memory import federation, sharing, storage
    from zettelkasten import config as zk_config
    from zettelkasten.synapse import config as syn_config

    config = storage.read_config()
    subs = config.get("bundle_subscriptions")
    has_memory_subs = isinstance(subs, list) and bool(subs)

    # Zettelkasten + synapse bundle subscriptions are synced INDEPENDENTLY of
    # memory ones — a user with only one kind must still sync. Peek at both
    # configs so a truly empty setup (no kind) can still exit cleanly without
    # demanding an identity.
    zk_subs = zk_config.read_config().get("bundle_subscriptions")
    has_zettel_subs = isinstance(zk_subs, list) and bool(zk_subs)
    syn_subs = syn_config.read_config().get("bundle_subscriptions")
    has_synapse_subs = isinstance(syn_subs, list) and bool(syn_subs)

    # Stranded-policy cleanup runs on EVERY `federation sync`, INDEPENDENT of the
    # local identity gate below. It only reads config + removes stranded policy
    # files (never decrypts), so it needs no identity — and running it before the
    # gate ensures a stranded/unsubscribed peer's stale (revoked) policy is still
    # revoked even when the local identity is temporarily missing/broken. The
    # identity-gated work below (decrypt/refresh) is unchanged.
    cache_root = federation.federation_cache_root()
    _cleanup_stranded_recipient_indexes(cache_root)

    identity = sharing.load_local_identity()
    if identity is None:
        if has_memory_subs or has_zettel_subs or has_synapse_subs:
            print(
                "error: no local identity at ~/.angelo/identity.yaml; create one "
                "with 'angelo memory identity init --id <id>'",
                file=sys.stderr,
            )
            sys.exit(1)
        print("[federation] no bundle subscriptions configured; nothing to sync")
        return

    fetch_root = storage.ANGELO_DIR / "federation-bundles"

    if has_memory_subs:
        synced = 0
        for sub in subs:
            if not isinstance(sub, dict):
                print(f"[federation] skipping malformed subscription: {sub!r}", file=sys.stderr)
                continue
            peer_id = str(sub.get("id") or "").strip()
            if not peer_id or not federation.ID_RE.match(peer_id):
                print(f"[federation] skipping subscription with invalid id: {sub.get('id')!r}", file=sys.stderr)
                continue
            remote = str(sub.get("remote") or "").strip()
            raw_path = str(sub.get("path") or "").strip()

            if remote:
                local_bundle = fetch_root / peer_id
                if not _sync_remote_bundle(remote, local_bundle):
                    print(f"[federation] {peer_id}: failed to fetch remote {remote}", file=sys.stderr)
                    continue
            elif raw_path:
                local_bundle = _resolve_workspace_path(raw_path)
            else:
                print(f"[federation] {peer_id}: no path or remote configured; skipping", file=sys.stderr)
                continue

            out_memory = cache_root / peer_id / ".memory"
            try:
                result = sharing.decrypt_bundle(local_bundle, out_memory, identity=identity)
            except FileNotFoundError:
                print(f"[federation] {peer_id}: no bundle manifest at {local_bundle}; skipping", file=sys.stderr)
                continue
            except Exception as exc:  # pragma: no cover - defensive
                print(f"[federation] {peer_id}: failed to decrypt bundle: {exc}", file=sys.stderr)
                continue

            synced += 1
            n = len(result.entry_ids)
            print(
                f"[federation] {peer_id}: decrypted {n} entr"
                f"{'y' if n == 1 else 'ies'}"
                f"{' + project' if result.project_written else ''}"
                f" -> {cache_root / peer_id}"
            )
        print(f"[federation] synced {synced} subscription(s)")
    else:
        print("[federation] no memory bundle subscriptions configured")

    # The firm identity registry (shared with memory + zk) turns on fail-closed
    # recipient-index signature verification in the ZK sync path below. Loaded
    # once here and threaded down so a tampered/downgraded/relabelled peer index
    # is REJECTED in production (mirrors the load in `synapse publish`).
    registry = sharing.load_registry()
    _sync_zettelkasten_bundles(identity, cache_root, fetch_root, registry=registry)
    _sync_synapse_bundles(identity, cache_root, fetch_root)

cmd_iceberg_contributions

cmd_iceberg_contributions(args: Namespace) -> None

List staged contribution branches + their manifests (purely read-only).

Source code in angelo_cli/__init__.py

def cmd_iceberg_contributions(args: argparse.Namespace) -> None:
    """List staged contribution branches + their manifests (purely read-only)."""
    from iceberg import writeback

    owner_repo = _iceberg_owner_repo(args)
    registry = _iceberg_registry(args, owner_repo)
    try:
        branches = writeback.list_contribution_branches(owner_repo)
    except writeback.WritebackGitError as exc:
        print(f"error: {exc}", file=sys.stderr)
        sys.exit(1)

    rows: list[tuple[str, str, str, str, str]] = []
    for b in branches:
        caller_id = b["caller_id"]
        registered = registry.signing_public_key_for(caller_id) is not None
        try:
            manifests = writeback.read_contribution_manifests(owner_repo, caller_id)
        except writeback.WritebackGitError:
            manifests = []
        for man in manifests:
            rows.append((
                caller_id,
                str(man.get("contribution_id") or ""),
                str(man.get("kind") or ""),
                str(man.get("created_at") or ""),
                "yes" if registered else "NO",
            ))

    if not rows:
        print("[iceberg] no staged contributions")
        return

    headers = ("CALLER", "CONTRIBUTION_ID", "KIND", "CREATED_AT", "SIGNING_KEY")
    widths = [len(h) for h in headers]
    for row in rows:
        for i, cell in enumerate(row):
            widths[i] = max(widths[i], len(cell))
    fmt = "  ".join(f"{{:<{w}}}" for w in widths)
    print(fmt.format(*headers))
    print(fmt.format(*("-" * w for w in widths)))
    for row in rows:
        print(fmt.format(*row))

cmd_iceberg_review

cmd_iceberg_review(args: Namespace) -> None

Decrypt + verify a caller's staged contributions and present them.

Fail-closed presentation: a contribution whose signature cannot be verified against the caller's registered signing key is shown under a LOUD UNVERIFIED — DO NOT TRUST banner and marked not-mergeable. A verified contribution prints VERIFIED (signed by <caller_id>) plus its decrypted content and authorship metadata.

Source code in angelo_cli/__init__.py

def cmd_iceberg_review(args: argparse.Namespace) -> None:
    """Decrypt + verify a caller's staged contributions and present them.

    Fail-closed presentation: a contribution whose signature cannot be verified
    against the caller's registered signing key is shown under a LOUD
    ``UNVERIFIED — DO NOT TRUST`` banner and marked not-mergeable. A verified
    contribution prints ``VERIFIED (signed by <caller_id>)`` plus its decrypted
    content and authorship metadata.
    """
    from iceberg import writeback

    owner_repo = _iceberg_owner_repo(args)
    registry = _iceberg_registry(args, owner_repo)
    owner = _iceberg_owner_identity(args)
    if owner is None:
        print(
            "error: no owner identity available (looked at "
            f"{getattr(args, 'identity_path', None) or '~/.angelo/identity.yaml'}); "
            "the owner private key is required to decrypt contributions",
            file=sys.stderr,
        )
        sys.exit(1)

    caller_id = args.caller_id
    try:
        manifests = writeback.read_contribution_manifests(owner_repo, caller_id)
    except writeback.WritebackGitError as exc:
        print(f"error: {exc}", file=sys.stderr)
        sys.exit(1)
    wanted = getattr(args, "id", None)
    if wanted:
        manifests = [m for m in manifests if m.get("contribution_id") == wanted]
    if not manifests:
        print(f"[iceberg] no staged contributions for {caller_id!r}")
        return

    bar = "!" * 60
    for man in manifests:
        res = _iceberg_verify_contribution(owner_repo, owner, registry, caller_id, man)
        cid = res["contribution_id"]
        print("")
        print(f"contribution {cid} (kind={res['kind']}, target={res['target']})")
        if res["verified"]:
            print(f"  VERIFIED (signed by {caller_id})")
        else:
            print(f"  {bar}")
            print("  UNVERIFIED — DO NOT TRUST")
            print(f"  reason: {res['error'] or 'signature could not be verified'}")
            print("  status: NOT MERGEABLE")
            print(f"  {bar}")
        print(
            f"  authored-by: {man.get('caller_id')}"
            f"   created-at: {man.get('created_at')}"
        )
        if res["content"] is not None:
            body = json.dumps(res["content"], indent=2, sort_keys=True)
            print("  content:")
            for line in body.splitlines():
                print(f"    {line}")

cmd_iceberg_merge

cmd_iceberg_merge(args: Namespace) -> None

Re-verify, merge (fail-closed), then materialize a caller's contributions.

Order is strict: (1) RE-VERIFY every staged contribution's signature — if ANY fails to verify against the caller's registered signing key, the WHOLE merge is refused (no git mutation, no materialization); (2) :func:iceberg.writeback.merge_contribution brings the branch into the default branch, fail-closed on git conflict; (3) each accepted contribution is MATERIALIZED into the owner's live store, provenance-stamped with the contributor id. A materialization failure after the merge is surfaced as a partial state (the git merge already happened).

Source code in angelo_cli/__init__.py

def cmd_iceberg_merge(args: argparse.Namespace) -> None:
    """Re-verify, merge (fail-closed), then materialize a caller's contributions.

    Order is strict: (1) RE-VERIFY every staged contribution's signature — if
    ANY fails to verify against the caller's registered signing key, the WHOLE
    merge is refused (no git mutation, no materialization); (2)
    :func:`iceberg.writeback.merge_contribution` brings the branch into the
    default branch, fail-closed on git conflict; (3) each accepted contribution
    is MATERIALIZED into the owner's live store, provenance-stamped with the
    contributor id. A materialization failure after the merge is surfaced as a
    partial state (the git merge already happened).
    """
    from iceberg import writeback

    owner_repo = _iceberg_owner_repo(args)
    registry = _iceberg_registry(args, owner_repo)
    owner = _iceberg_owner_identity(args)
    if owner is None:
        print(
            "error: no owner identity available (looked at "
            f"{getattr(args, 'identity_path', None) or '~/.angelo/identity.yaml'}); "
            "the owner private key is required to decrypt + merge contributions",
            file=sys.stderr,
        )
        sys.exit(1)

    caller_id = args.caller_id
    try:
        manifests = writeback.read_contribution_manifests(owner_repo, caller_id)
    except writeback.WritebackGitError as exc:
        print(f"error: {exc}", file=sys.stderr)
        sys.exit(1)
    if not manifests:
        print(f"[iceberg] no staged contributions for {caller_id!r}", file=sys.stderr)
        sys.exit(1)

    # (1) RE-VERIFY every contribution BEFORE any git or store mutation.
    results = [
        _iceberg_verify_contribution(owner_repo, owner, registry, caller_id, man)
        for man in manifests
    ]
    blockers = [r for r in results if not r["verified"]]
    if blockers:
        print(
            f"[iceberg] merge REFUSED for {caller_id!r}: "
            f"{len(blockers)} of {len(results)} contribution(s) failed "
            "signature re-verification.",
            file=sys.stderr,
        )
        for r in blockers:
            print(
                f"  - {r['contribution_id']}: {r['error'] or 'unverified'}",
                file=sys.stderr,
            )
        print(
            "[iceberg] nothing was merged and nothing was materialized "
            "(fail-closed).",
            file=sys.stderr,
        )
        sys.exit(1)

    # (1b) Re-derive the WRITE id from the decrypted content and re-run the
    # guard/write binding + transport guard against the OWNER store BEFORE any
    # git or store mutation (defense in depth, fail-closed). The broker enforces
    # this at ingest, but merge must not TRUST that a staged contribution came
    # through a current broker — an older/bypassed broker (or a hand-staged
    # branch) could carry a benign ``target`` while ``content['id']`` names an
    # existing OFFLINE owner unit. Re-checking here provably guards the unit that
    # materialize will actually write. Also verify ``content_sha256`` matches the
    # decrypted bytes (catches any manifest/ciphertext desync) in the SAME
    # pre-mutation phase, so nothing merges if ANY contribution fails.
    from hashlib import sha256

    from iceberg.broker.app import (
        _memory_target_transport,
        _zk_target_transport,
        derive_write_id,
        split_zk_target,
    )

    mem_dir = owner_repo / ".memory"
    zk_root = owner_repo / ".zettelkasten"
    mem_cfg = mem_dir / "config.yaml"
    zk_cfg = zk_root / "config.yaml"

    guard_errors: list[str] = []
    overwrite_warnings: list[str] = []
    for r in results:
        cid = r["contribution_id"]
        man = r["manifest"]
        content = r["content"]
        target = str(r["target"] or "")
        kind = r["kind"]

        # content_sha256 == sha256(decrypted canonical bytes).
        expected_sha = str(man.get("content_sha256") or "")
        actual_sha = (
            sha256(r["decrypted"]).hexdigest() if r["decrypted"] is not None else ""
        )
        if not expected_sha or expected_sha != actual_sha:
            guard_errors.append(
                f"{cid}: content_sha256 mismatch "
                "(manifest/ciphertext desync — refusing)"
            )
            continue

        # Guard/write binding: the id materialize will write must equal target,
        # AND the (box, note_id)/id shape must be valid. This validates the shape
        # HERE, in the pre-mutation phase, so ANY malformed contribution aborts
        # the WHOLE merge fail-closed BEFORE ``merge_contribution`` runs (never
        # crashing mid-materialize after the git merge already happened).
        try:
            write_id = derive_write_id(kind, content, target)
            if kind == "zk-note":
                # Parse ONCE through the shared helper and stash the result so
                # materialize reuses this exact (box, note_id) — no second
                # interpretation of the raw target after the merge.
                r["_zk_box"], r["_zk_note_id"] = split_zk_target(target)
        except (ValueError, TypeError) as exc:
            guard_errors.append(f"{cid}: {exc}")
            continue

        # Transport re-guard against the OWNER store: refuse an existing OFFLINE
        # unit; warn (loudly, never silent) if it overwrites an existing HOSTED
        # unit; a brand-new id is fine.
        if kind == "memory-entry":
            transport = _memory_target_transport(mem_dir, mem_cfg, write_id)
        elif kind == "zk-note":
            transport = _zk_target_transport(zk_root, zk_cfg, write_id)
        else:
            guard_errors.append(f"{cid}: unknown kind {kind!r}")
            continue

        if transport == "offline":
            guard_errors.append(
                f"{cid}: refuses to materialize onto existing OFFLINE unit "
                f"{write_id!r} (offline units are never reachable via write-back)"
            )
        elif transport == "hosted":
            overwrite_warnings.append(
                f"{cid}: OVERWRITES existing hosted unit {write_id!r}"
            )

    if guard_errors:
        print(
            f"[iceberg] merge REFUSED for {caller_id!r}: "
            f"{len(guard_errors)} contribution(s) failed the guard/write-id or "
            "transport re-check.",
            file=sys.stderr,
        )
        for e in guard_errors:
            print(f"  - {e}", file=sys.stderr)
        print(
            "[iceberg] nothing was merged and nothing was materialized "
            "(fail-closed).",
            file=sys.stderr,
        )
        sys.exit(1)

    for w in overwrite_warnings:
        print(f"[iceberg] WARNING: {w}", file=sys.stderr)

    # (2) Merge the branch into the default branch — fail-closed on conflict.
    try:
        merge_result = writeback.merge_contribution(
            owner_repo, caller_id, into=getattr(args, "into", None)
        )
    except writeback.WritebackGitError as exc:
        print(
            f"[iceberg] merge failed (fail-closed): {exc}\n"
            "[iceberg] the default branch and working tree were left untouched; "
            "nothing was materialized.",
            file=sys.stderr,
        )
        sys.exit(1)
    print(
        f"[iceberg] merged {merge_result.branch} into {merge_result.into} "
        f"at {merge_result.merged}"
    )

    # (3) Materialize each accepted contribution into the owner's live store.
    materialized: list[tuple[str, str]] = []
    errors: list[str] = []
    for r in results:
        try:
            if r["kind"] == "memory-entry":
                mid, _ = _iceberg_materialize_memory(owner_repo, r["content"], caller_id)
                materialized.append(("memory-entry", mid))
            elif r["kind"] == "zk-note":
                mid, _ = _iceberg_materialize_zk(
                    owner_repo,
                    r["_zk_box"],
                    r["_zk_note_id"],
                    r["content"],
                    caller_id,
                )
                materialized.append(("zk-note", mid))
            else:
                errors.append(
                    f"{r['contribution_id']}: unknown kind {r['kind']!r}"
                )
        except Exception as exc:  # materialization is best-effort AFTER merge
            errors.append(f"{r['contribution_id']}: materialize failed: {exc}")

    for kind, mid in materialized:
        print(
            f"[iceberg] materialized {kind} {mid!r} (contributed-by: {caller_id})"
        )

    if errors:
        print(
            "[iceberg] WARNING: the git merge SUCCEEDED but some contributions "
            "could not be materialized into the live store:",
            file=sys.stderr,
        )
        for e in errors:
            print(f"  - {e}", file=sys.stderr)
        print(
            f"[iceberg] partial state: branch merged at {merge_result.merged}, "
            "but the owner store was not fully updated (re-run materialization "
            "or inspect manually).",
            file=sys.stderr,
        )
        sys.exit(1)

cmd_iceberg_serve

cmd_iceberg_serve(args: Namespace) -> None

Run the Iceberg broker under uvicorn (owner-hosted read + write-back server).

Resolves a :class:~iceberg.broker.config.BrokerConfig by layering the provided flags over BrokerConfig.from_env() (flags win over env), then serves build_app(config) on --host/--port with NO reload/debug. Binds 127.0.0.1 by default (loopback only). Write-back and rate limiting stay OFF unless explicitly enabled by flag or a truthy ICEBERG_BROKER_* env var (fail-closed).

Source code in angelo_cli/__init__.py

def cmd_iceberg_serve(args: argparse.Namespace) -> None:
    """Run the Iceberg broker under uvicorn (owner-hosted read + write-back server).

    Resolves a :class:`~iceberg.broker.config.BrokerConfig` by layering the
    provided flags over ``BrokerConfig.from_env()`` (flags win over env), then
    serves ``build_app(config)`` on ``--host``/``--port`` with NO reload/debug.
    Binds ``127.0.0.1`` by default (loopback only). Write-back and rate limiting
    stay OFF unless explicitly enabled by flag or a truthy ``ICEBERG_BROKER_*``
    env var (fail-closed).
    """
    import dataclasses as _dataclasses

    from iceberg.broker.app import build_app
    from iceberg.broker.config import BrokerConfig

    # If --registry-path is not given, default to the owner repo's registry so
    # `angelo iceberg serve` works from a checkout with no env configured (same
    # convention as the review/merge handlers). Env still wins if it set a path.
    base = BrokerConfig.from_env()
    overrides: dict = {}

    reg = getattr(args, "registry_path", None)
    if reg is None and not os.environ.get("ICEBERG_BROKER_REGISTRY_PATH"):
        top = _git_toplevel(Path.cwd())
        if top is not None:
            overrides["registry_path"] = str(top / ".memory" / "identities.yaml")
    elif reg is not None:
        overrides["registry_path"] = reg

    for field in ("cache_root", "memory_source", "zk_store_root"):
        val = getattr(args, field, None)
        if val is not None:
            overrides[field] = val
    if getattr(args, "enable_writeback", False):
        overrides["enable_writeback"] = True
    if getattr(args, "rate_limit_enabled", False):
        overrides["rate_limit_enabled"] = True
    if getattr(args, "trust_forwarded_for", False):
        overrides["trust_forwarded_for"] = True

    config = _dataclasses.replace(base, **overrides) if overrides else base

    host = getattr(args, "host", "127.0.0.1")
    port = int(getattr(args, "port", 8779))
    print(
        f"[iceberg] serving broker on http://{host}:{port} "
        f"(writeback={'on' if config.enable_writeback else 'off'}, "
        f"rate_limit={'on' if config.rate_limit_enabled else 'off'})",
        file=sys.stderr,
    )
    try:
        import uvicorn
    except ImportError:
        print(
            "error: uvicorn is required to serve the broker "
            "(pip install uvicorn)",
            file=sys.stderr,
        )
        sys.exit(1)
    uvicorn.run(build_app(config), host=host, port=port)