coordinator.server

coordinator.server

MCP server exposing the task graph as tools to Cursor.

Tools

agents -- Introspect the agent roster (action="list"|"get") create_graph -- Build a DAG of tasks with dependencies get_ready_tasks -- Return unblocked tasks (current wave) claim_task -- Mark a task as running (acquires its path-scoped lease) manage_paths -- Acquire/release fine-grained write leases (action dispatch) submit_result -- Record output, mark done/failed (releases active leases) record_pass -- Advance a multi-pass task's round counter (live progress) get_task_context -- Get predecessor outputs + current task info extend_graph -- Append corrective tasks after failures get_report -- Live status + summary report of the run manage_runs -- Resume, finalize, or clean up interrupted runs

PathLockManager

Process-wide active leases plus run-lifetime path reservations.

Implementers no longer serialize globally — they coordinate on the actual paths they write. A lease on a path covers that subtree, so two implementers with disjoint write scopes run concurrently while two that touch the same path serialize. The whole-workspace key "" conflicts with everything, which is the conservative default for a task that declares no writes.

Runs with a final memory task retain every acquired path after the active task lease is released. This prevents another run from changing a path before the memory task performs its synchronous code-pin commit. A same-run fix engineer may reacquire a retained path; active task leases still prevent simultaneous implementers within that run.

Holds its OWN lock (a threading.Condition), separate from the server's _state_lock. Lock ordering is always _state_lock -> this manager and NEVER the reverse, and the only blocking wait (acquire) is invoked WITHOUT _state_lock held, so the two locks cannot deadlock.

Source code in coordinator/server.py

class PathLockManager:
    """Process-wide active leases plus run-lifetime path reservations.

    Implementers no longer serialize globally — they coordinate on the actual
    paths they write. A lease on a path covers that subtree, so two implementers
    with disjoint write scopes run concurrently while two that touch the same
    path serialize. The whole-workspace key ``""`` conflicts with everything,
    which is the conservative default for a task that declares no ``writes``.

    Runs with a final memory task retain every acquired path after the active
    task lease is released. This prevents another run from changing a path
    before the memory task performs its synchronous code-pin commit. A same-run
    fix engineer may reacquire a retained path; active task leases still prevent
    simultaneous implementers within that run.

    Holds its OWN lock (a ``threading.Condition``), separate from the server's
    ``_state_lock``. Lock ordering is always ``_state_lock`` -> this manager and
    NEVER the reverse, and the only blocking wait (``acquire``) is invoked
    WITHOUT ``_state_lock`` held, so the two locks cannot deadlock.
    """

    def __init__(self) -> None:
        self._cond = threading.Condition()
        self._owners: dict[str, set[str]] = {}  # task_id -> held keys
        self._meta: dict[str, dict] = {}         # task_id -> {run_id, agent}
        self._reservations: dict[str, set[str]] = {}  # run_id -> retained keys

    def _conflict_locked(
        self, keys: list[str], exclude_task_id: str, requesting_run_id: str = ""
    ) -> dict | None:
        for tid, held in self._owners.items():
            if tid == exclude_task_id:
                continue
            for hk in held:
                for rk in keys:
                    if keys_overlap(hk, rk):
                        meta = self._meta.get(tid, {})
                        return {
                            "task_id": tid,
                            "run_id": meta.get("run_id", ""),
                            "agent": meta.get("agent", ""),
                            "held_path": _display_key(hk),
                            "requested_path": _display_key(rk),
                        }
        for rid, held in self._reservations.items():
            if rid == requesting_run_id:
                continue
            for hk in held:
                for rk in keys:
                    if keys_overlap(hk, rk):
                        return {
                            "task_id": "",
                            "run_id": rid,
                            "agent": "run-reservation",
                            "held_path": _display_key(hk),
                            "requested_path": _display_key(rk),
                            "reserved": True,
                        }
        return None

    def _grant_locked(
        self,
        task_id: str,
        keys: list[str],
        run_id: str,
        agent: str,
        *,
        reserve_for_run: bool = False,
    ) -> None:
        self._owners.setdefault(task_id, set()).update(keys)
        self._meta[task_id] = {"run_id": run_id, "agent": agent}
        if reserve_for_run and run_id:
            self._reservations.setdefault(run_id, set()).update(keys)

    def try_acquire(
        self,
        task_id: str,
        keys: list[str],
        *,
        run_id: str = "",
        agent: str = "",
        reserve_for_run: bool = False,
    ) -> dict | None:
        """Non-blocking grant. Returns None on success, or holder info on conflict."""
        if not keys:
            return None
        with self._cond:
            conflict = self._conflict_locked(keys, task_id, run_id)
            if conflict is not None:
                return conflict
            self._grant_locked(
                task_id,
                keys,
                run_id,
                agent,
                reserve_for_run=reserve_for_run,
            )
            return None

    def acquire(
        self,
        task_id: str,
        keys: list[str],
        *,
        run_id: str = "",
        agent: str = "",
        reserve_for_run: bool = False,
        timeout: float = 30.0,
    ) -> dict | None:
        """Blocking grant with a timeout. Returns None on success, or holder
        info if the lease could not be acquired within ``timeout`` seconds.

        Must NOT be called while holding ``_state_lock`` (it can block)."""
        if not keys:
            return None
        deadline = time.monotonic() + max(0.0, timeout)
        with self._cond:
            while True:
                conflict = self._conflict_locked(keys, task_id, run_id)
                if conflict is None:
                    self._grant_locked(
                        task_id,
                        keys,
                        run_id,
                        agent,
                        reserve_for_run=reserve_for_run,
                    )
                    return None
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    return conflict
                self._cond.wait(timeout=remaining)

    def release(self, task_id: str, keys: list[str] | None = None) -> None:
        """Release some (or, when ``keys`` is None, all) leases held by a task."""
        with self._cond:
            held = self._owners.get(task_id)
            if held is None:
                return
            if keys is None:
                self._owners.pop(task_id, None)
                self._meta.pop(task_id, None)
            else:
                held.difference_update(keys)
                if not held:
                    self._owners.pop(task_id, None)
                    self._meta.pop(task_id, None)
            self._cond.notify_all()

    def restore_run(self, run_id: str, keys: list[str]) -> None:
        """Restore persisted run reservations after resume or adoption."""
        if not run_id or not keys:
            return
        with self._cond:
            self._reservations.setdefault(run_id, set()).update(keys)
            self._cond.notify_all()

    def release_run(self, run_id: str) -> None:
        """Release every path retained for a completed or cleared run."""
        if not run_id:
            return
        with self._cond:
            self._reservations.pop(run_id, None)
            self._cond.notify_all()

    def reserved_for(self, run_id: str) -> list[str]:
        """Return sorted retained paths for persistence and diagnostics."""
        with self._cond:
            return sorted(self._reservations.get(run_id, set()))

    def held(self) -> dict[str, list[str]]:
        """Snapshot of held leases (task_id -> sorted keys) for diagnostics."""
        with self._cond:
            return {tid: sorted(keys) for tid, keys in self._owners.items()}

try_acquire

try_acquire(task_id: str, keys: list[str], *, run_id: str = '', agent: str = '', reserve_for_run: bool = False) -> dict | None

Non-blocking grant. Returns None on success, or holder info on conflict.

Source code in coordinator/server.py

def try_acquire(
    self,
    task_id: str,
    keys: list[str],
    *,
    run_id: str = "",
    agent: str = "",
    reserve_for_run: bool = False,
) -> dict | None:
    """Non-blocking grant. Returns None on success, or holder info on conflict."""
    if not keys:
        return None
    with self._cond:
        conflict = self._conflict_locked(keys, task_id, run_id)
        if conflict is not None:
            return conflict
        self._grant_locked(
            task_id,
            keys,
            run_id,
            agent,
            reserve_for_run=reserve_for_run,
        )
        return None

acquire

acquire(task_id: str, keys: list[str], *, run_id: str = '', agent: str = '', reserve_for_run: bool = False, timeout: float = 30.0) -> dict | None

Blocking grant with a timeout. Returns None on success, or holder info if the lease could not be acquired within timeout seconds.

Must NOT be called while holding _state_lock (it can block).

Source code in coordinator/server.py

def acquire(
    self,
    task_id: str,
    keys: list[str],
    *,
    run_id: str = "",
    agent: str = "",
    reserve_for_run: bool = False,
    timeout: float = 30.0,
) -> dict | None:
    """Blocking grant with a timeout. Returns None on success, or holder
    info if the lease could not be acquired within ``timeout`` seconds.

    Must NOT be called while holding ``_state_lock`` (it can block)."""
    if not keys:
        return None
    deadline = time.monotonic() + max(0.0, timeout)
    with self._cond:
        while True:
            conflict = self._conflict_locked(keys, task_id, run_id)
            if conflict is None:
                self._grant_locked(
                    task_id,
                    keys,
                    run_id,
                    agent,
                    reserve_for_run=reserve_for_run,
                )
                return None
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                return conflict
            self._cond.wait(timeout=remaining)

release

release(task_id: str, keys: list[str] | None = None) -> None

Release some (or, when keys is None, all) leases held by a task.

Source code in coordinator/server.py

def release(self, task_id: str, keys: list[str] | None = None) -> None:
    """Release some (or, when ``keys`` is None, all) leases held by a task."""
    with self._cond:
        held = self._owners.get(task_id)
        if held is None:
            return
        if keys is None:
            self._owners.pop(task_id, None)
            self._meta.pop(task_id, None)
        else:
            held.difference_update(keys)
            if not held:
                self._owners.pop(task_id, None)
                self._meta.pop(task_id, None)
        self._cond.notify_all()

restore_run

restore_run(run_id: str, keys: list[str]) -> None

Restore persisted run reservations after resume or adoption.

Source code in coordinator/server.py

def restore_run(self, run_id: str, keys: list[str]) -> None:
    """Restore persisted run reservations after resume or adoption."""
    if not run_id or not keys:
        return
    with self._cond:
        self._reservations.setdefault(run_id, set()).update(keys)
        self._cond.notify_all()

release_run

release_run(run_id: str) -> None

Release every path retained for a completed or cleared run.

Source code in coordinator/server.py

def release_run(self, run_id: str) -> None:
    """Release every path retained for a completed or cleared run."""
    if not run_id:
        return
    with self._cond:
        self._reservations.pop(run_id, None)
        self._cond.notify_all()

reserved_for

reserved_for(run_id: str) -> list[str]

Return sorted retained paths for persistence and diagnostics.

Source code in coordinator/server.py

def reserved_for(self, run_id: str) -> list[str]:
    """Return sorted retained paths for persistence and diagnostics."""
    with self._cond:
        return sorted(self._reservations.get(run_id, set()))

held

held() -> dict[str, list[str]]

Snapshot of held leases (task_id -> sorted keys) for diagnostics.

Source code in coordinator/server.py

def held(self) -> dict[str, list[str]]:
    """Snapshot of held leases (task_id -> sorted keys) for diagnostics."""
    with self._cond:
        return {tid: sorted(keys) for tid, keys in self._owners.items()}

list_agents

list_agents() -> str

Return available agent types with one-line descriptions.

Includes built-in agents and any project-local overrides from .cursor/agents.yaml. Each description is suffixed with the agent's workflow role ("implementer", "checker", or "meta"), which determines where the agent may sit in the graph (see create_graph validation rules).

Returns:

Type	Description
str	JSON object mapping agent names to their descriptions.

Source code in coordinator/server.py

def list_agents() -> str:
    """Return available agent types with one-line descriptions.

    Includes built-in agents and any project-local overrides from .cursor/agents.yaml.
    Each description is suffixed with the agent's workflow role
    ("implementer", "checker", or "meta"), which determines where the agent
    may sit in the graph (see create_graph validation rules).

    Returns:
        JSON object mapping agent names to their descriptions.
    """
    merged = {**AGENT_DESCRIPTIONS, **_contrib_descriptions, **_local_descriptions}
    roles = {**AGENT_ROLES, **_contrib_roles, **_local_roles}
    models = _merged_models()
    result = {}
    for name, desc in merged.items():
        suffix = f"role: {roles.get(name, 'checker')}"
        model = models.get(name)
        if model:
            suffix += f", model: {model}"
        passes = _resolve_passes(name)
        if passes != DEFAULT_PASSES:
            suffix += f", passes: {passes}"
        result[name] = f"{desc} ({suffix})"
    return json.dumps(result)

get_persona

get_persona(agent: str) -> str

Return the full persona prompt text for a given agent type.

Checks project-local overrides first, then falls back to built-in personas.

Parameters:

Name	Type	Description	Default
agent	str	Agent type name (e.g. "engineer", "reviewer").	required

Returns:

Type	Description
str	JSON object with the agent name, its workflow role, its full persona
str	text, and passes (the number of self-refinement passes the
str	coordinator should drive for this agent via Task resume; 1 = single
str	pass / no re-prompt). May also include model and schema when
str	configured, and native_subagent -- the host-native subagent type
str	(e.g. Cursor's bugbot) the coordinator should spawn instead of the
str	persona when the host provides it, falling back to the persona otherwise.

Source code in coordinator/server.py

def get_persona(agent: str) -> str:
    """Return the full persona prompt text for a given agent type.

    Checks project-local overrides first, then falls back to built-in personas.

    Args:
        agent: Agent type name (e.g. "engineer", "reviewer").

    Returns:
        JSON object with the agent name, its workflow role, its full persona
        text, and ``passes`` (the number of self-refinement passes the
        coordinator should drive for this agent via Task `resume`; 1 = single
        pass / no re-prompt). May also include ``model`` and ``schema`` when
        configured, and ``native_subagent`` -- the host-native subagent type
        (e.g. Cursor's ``bugbot``) the coordinator should spawn instead of the
        persona when the host provides it, falling back to the persona otherwise.
    """
    all_personas = {**PERSONAS, **_contrib_personas, **_local_personas}
    if agent not in all_personas:
        raise ValueError(
            f"Unknown agent type: {agent!r}. "
            f"Available: {sorted(all_personas.keys())}"
        )
    roles = {**AGENT_ROLES, **_contrib_roles, **_local_roles}
    result = {
        "agent": agent,
        "role": roles.get(agent, "checker"),
        "persona": all_personas[agent],
        "passes": _resolve_passes(agent),
    }
    model = _merged_models().get(agent)
    if model:
        result["model"] = model
    schema = _merged_agent_schemas().get(agent)
    if schema:
        result["schema"] = schema
    native = _merged_native().get(agent)
    if native:
        result["native_subagent"] = native
    return json.dumps(result)

overview

overview() -> str

Use this first when you're unsure how to run a multi-agent task: a compact orientation map.

Read-only. Returns the coordinator's purpose, the standard wave-execution loop, and a tool -> action index. Authoritative action values also live on each tool's own action schema enum.

Source code in coordinator/server.py

@mcp_server.tool()
def overview() -> str:
    """Use this first when you're unsure how to run a multi-agent task: a compact orientation map.

    Read-only. Returns the coordinator's purpose, the standard wave-execution
    loop, and a tool -> action index. Authoritative action values also live on
    each tool's own ``action`` schema enum.
    """
    return (
        "# Coordinator server — orientation\n\n"
        "Runs multi-agent tasks as a wave-based task graph (DAG). You are the\n"
        "coordinator: scope the task, design a graph, then execute it by spawning\n"
        "subagents wave by wave.\n\n"
        "## Typical flow\n"
        "1. `agents(action=\"list\")` / `agents(action=\"get\", agent=...)` — pick agents + fetch personas.\n"
        "2. `create_graph(tasks=[...])` — build the DAG (returns a run_id).\n"
        "3. Loop until no ready tasks:\n"
        "   `get_ready_tasks` -> `claim_task` -> spawn subagent -> `submit_result`.\n"
        "4. `get_report` — final results + extension history.\n\n"
        "## Tools\n"
        "- `agents(action=list|get)` — agent roster + personas.\n"
        "- `create_graph` — build the executable task graph.\n"
        "- `get_ready_tasks` — the next parallel wave.\n"
        "- `claim_task` — mark running + take write lease (call before spawning).\n"
        "- `submit_result` — record a finished task's outcome.\n"
        "- `get_task_context` — task description + predecessor outputs.\n"
        "- `record_pass` — log a self-refinement pass.\n"
        "- `extend_graph` — append a corrective fix cycle after a checker FAIL.\n"
        "- `manage_paths(action=acquire|release)` — dynamic write leases at write time.\n"
        "- `manage_runs(action=resume|cleanup|finalize)` — recover/clean up interrupted runs.\n"
        "- `track_subtask` — register ad-hoc helper subagents for visibility.\n"
        "  When the zettelkasten schema capability is enabled, extraction-pipeline\n"
        "  tools (`create_extraction_graph`, `prep_synthesis_contexts`,\n"
        "  `prep_connection_candidates`) also appear.\n\n"
        "## Resources you can pull\n"
        "- `coordinator://overview` (this map).\n"
    )

agents

agents(action: Literal['list', 'get'], agent: str = '') -> str

Use this when you need the agent roster: list every agent, or fetch one agent's full persona before spawning it.

Actions — name(required, optional?): list(): available agents (built-ins + .cursor/agents.yaml overrides) with one-line descriptions, each suffixed with the agent's workflow role (implementer/checker/planner/meta), which gates where it may sit in the graph. Call while designing the graph. get(agent): full persona text for one agent plus its role, passes, and — when configured — model, schema, and native_subagent (a host-native subagent, e.g. Cursor's bugbot, spawned instead of the persona and falling back to it). Call right before spawning each subagent.

Returns:

Type	Description
str	JSON — a name→description map for list; a persona object for get.

Source code in coordinator/server.py

@mcp_server.tool()
def agents(action: Literal["list", "get"], agent: str = "") -> str:
    """Use this when you need the agent roster: list every agent, or fetch one agent's full persona before spawning it.

    Actions — name(required, optional?):
        list(): available agents (built-ins + .cursor/agents.yaml overrides) with
            one-line descriptions, each suffixed with the agent's workflow role
            (implementer/checker/planner/meta), which gates where it may sit in the
            graph. Call while designing the graph.
        get(agent): full persona text for one agent plus its role, ``passes``, and —
            when configured — ``model``, ``schema``, and ``native_subagent`` (a
            host-native subagent, e.g. Cursor's ``bugbot``, spawned instead of the
            persona and falling back to it). Call right before spawning each subagent.

    Returns:
        JSON — a name→description map for ``list``; a persona object for ``get``.
    """
    if action == "list":
        return list_agents()
    if action == "get":
        if not agent:
            raise ValueError("agents(action='get') requires an 'agent' name.")
        return get_persona(agent)
    raise ValueError(f"Unknown action {action!r}. Use 'list' or 'get'.")

create_graph

create_graph(tasks: list[dict], goal: str = '', target_entry_id: str = '', run_id: str = '', rigor: str = '', max_extensions: int = 0) -> str

Use this when you've scoped a multi-agent task and are ready to launch it: build the executable task graph (DAG) from a list of task definitions.

Each task dict must have

• alias: str -- local reference name (e.g. "eng", "rev")
• agent: str -- agent type (e.g. "engineer", "reviewer", "critic")
• description: str -- what the agent should do
• depends_on: list[str] -- aliases this task depends on (empty = root)

Optional per-task attributes: model (model override), passes (self-refinement passes), schema (grounded-extraction rubric name; requires an enabled capability -- otherwise a hard validation error), and writes (the implementer task's write scope -- honored here, NOT on claim_task; see the dedicated section just below). writes is a first-class per-task field on the SAME footing as model/passes/ schema: set it in each task dict and this tool applies it. It is REQUIRED in practice for every engineer/scribe task -- omitting it silently falls back to a whole-workspace lease that blocks every other implementer.

writes declares an implementer (engineer/scribe) task's write scope so implementers coordinate on a path-scoped lease instead of serializing globally: tasks with DISJOINT scopes run in parallel; tasks touching the SAME path serialize at claim time. It is a per-task attribute (set it here, not on claim_task).

PROTOCOL: you MUST fill writes for every implementer (engineer/scribe) task -- do not omit it. A whole-workspace lease blocks every other implementer, so taking one must be a deliberate, explicit choice rather than an accident of leaving the field blank. Accepted values:

- a list of workspace-relative paths/subtrees (e.g.
  ``"writes": ["src/api", "docs/api.md"]``) -> leases just those
  subtrees; declare disjoint scopes across a wave to run scribes/
  engineers concurrently. PREFER this -- keep it as narrow as the task
  actually needs.
- an empty list ``[]`` -> dynamic mode: no claim-time lease; the agent
  leases paths at write time via
  ``manage_paths(action="acquire"/"release")``.
- ``["."]`` -> the explicit whole-workspace lease (the conservative,
  serialize-everything option). Choose this ONLY when the task may touch
  unpredictable paths across the tree; it is the explicit spelling of
  the legacy one-implementer-at-a-time behavior.

Omitting writes falls back to a whole-workspace lease for safety, but that is a backstop, not the intended path -- always pick one of the values above explicitly. writes is ignored for read-only roles (planner/checker/meta), which never take a lease. Note: zettelkasten scribes do not need writes for store writes -- the zettelkasten MCP enforces per-box locking itself.

The graph is validated before creation

• Must be acyclic
• Must contain at least one implementer (e.g. engineer)
• Every checker task must have an implementer in its dependency ancestry; meta agents (memory, custom role=meta) are exempt
• All depends_on references must resolve

Each call creates an isolated run with its own graph, so multiple graphs can run concurrently in one process. The returned run_id identifies the run; pass it to the other run-scoped tools (get_ready_tasks, get_report) when more than one run is active. Task-scoped tools (claim_task, submit_result, ...) locate the owning run from the task ID automatically.

Parameters:

Name	Type	Description	Default
tasks	list[dict]	List of task definitions.	required
goal	str	Optional high-level goal for the graph.	''
target_entry_id	str	Optional memory tree entry ID where results will be recorded. Persisted for the dashboard to position the execution monitor at the correct tree node.	''
run_id	str	Optional explicit run identifier. Auto-generated when omitted. Must be unique among currently registered runs.	''
rigor	str	Optional run rigor: "low", "medium" (default), or "high". Sets how many corrective extension cycles are allowed and the default self-refinement passes for checker agents. Per-task passes and agents.yaml passes still override the profile defaults. Falls back to the agents.yaml top-level rigor: setting, then "medium".	''
max_extensions	int	Optional explicit cap on corrective extension cycles for this run. Overrides the rigor profile's cap when > 0.	0

Returns:

Type	Description
str	JSON with the run_id, created task IDs, resolved rigor/max_extensions,
str	and a graph summary. Each summary entry echoes the resolved writes
str	scope (when declared) so you can confirm the per-task lease this tool
str	registered before spawning; an entry with no writes key defaulted to
str	a whole-workspace lease.

Source code in coordinator/server.py

@mcp_server.tool()
def create_graph(
    tasks: list[dict],
    goal: str = "",
    target_entry_id: str = "",
    run_id: str = "",
    rigor: str = "",
    max_extensions: int = 0,
) -> str:
    """Use this when you've scoped a multi-agent task and are ready to launch it: build the executable task graph (DAG) from a list of task definitions.

    Each task dict must have:
        - alias: str -- local reference name (e.g. "eng", "rev")
        - agent: str -- agent type (e.g. "engineer", "reviewer", "critic")
        - description: str -- what the agent should do
        - depends_on: list[str] -- aliases this task depends on (empty = root)

    Optional per-task attributes: ``model`` (model override), ``passes``
    (self-refinement passes), ``schema`` (grounded-extraction rubric name;
    requires an enabled capability -- otherwise a hard validation error), and
    ``writes`` (the implementer task's write scope -- honored here, NOT on
    ``claim_task``; see the dedicated section just below). ``writes`` is a
    first-class per-task field on the SAME footing as ``model``/``passes``/
    ``schema``: set it in each task dict and this tool applies it. It is
    REQUIRED in practice for every engineer/scribe task -- omitting it silently
    falls back to a whole-workspace lease that blocks every other implementer.

    ``writes`` declares an implementer (engineer/scribe) task's write scope so
    implementers coordinate on a path-scoped lease instead of serializing
    globally: tasks with DISJOINT scopes run in parallel; tasks touching the
    SAME path serialize at claim time. It is a per-task attribute (set it here,
    not on ``claim_task``).

    PROTOCOL: you MUST fill ``writes`` for every implementer (engineer/scribe)
    task -- do not omit it. A whole-workspace lease blocks every other
    implementer, so taking one must be a deliberate, explicit choice rather than
    an accident of leaving the field blank. Accepted values:

        - a list of workspace-relative paths/subtrees (e.g.
          ``"writes": ["src/api", "docs/api.md"]``) -> leases just those
          subtrees; declare disjoint scopes across a wave to run scribes/
          engineers concurrently. PREFER this -- keep it as narrow as the task
          actually needs.
        - an empty list ``[]`` -> dynamic mode: no claim-time lease; the agent
          leases paths at write time via
          ``manage_paths(action="acquire"/"release")``.
        - ``["."]`` -> the explicit whole-workspace lease (the conservative,
          serialize-everything option). Choose this ONLY when the task may touch
          unpredictable paths across the tree; it is the explicit spelling of
          the legacy one-implementer-at-a-time behavior.

    Omitting ``writes`` falls back to a whole-workspace lease for safety, but
    that is a backstop, not the intended path -- always pick one of the values
    above explicitly. ``writes`` is ignored for read-only roles
    (planner/checker/meta), which never take a lease. Note: zettelkasten scribes
    do not need ``writes`` for store writes -- the zettelkasten MCP enforces
    per-box locking itself.

    The graph is validated before creation:
        - Must be acyclic
        - Must contain at least one implementer (e.g. engineer)
        - Every checker task must have an implementer in its dependency
          ancestry; meta agents (memory, custom role=meta) are
          exempt
        - All depends_on references must resolve

    Each call creates an isolated run with its own graph, so multiple graphs
    can run concurrently in one process. The returned ``run_id`` identifies the
    run; pass it to the other run-scoped tools (get_ready_tasks, get_report)
    when more than one run is active. Task-scoped tools (claim_task,
    submit_result, ...) locate the owning run from the task ID automatically.

    Args:
        tasks: List of task definitions.
        goal: Optional high-level goal for the graph.
        target_entry_id: Optional memory tree entry ID where results will be
            recorded. Persisted for the dashboard to position the execution
            monitor at the correct tree node.
        run_id: Optional explicit run identifier. Auto-generated when omitted.
            Must be unique among currently registered runs.
        rigor: Optional run rigor: "low", "medium" (default), or "high". Sets
            how many corrective extension cycles are allowed and the default
            self-refinement passes for checker agents. Per-task `passes` and
            agents.yaml `passes` still override the profile defaults. Falls back
            to the agents.yaml top-level `rigor:` setting, then "medium".
        max_extensions: Optional explicit cap on corrective extension cycles for
            this run. Overrides the rigor profile's cap when > 0.

    Returns:
        JSON with the run_id, created task IDs, resolved rigor/max_extensions,
        and a graph summary. Each summary entry echoes the resolved ``writes``
        scope (when declared) so you can confirm the per-task lease this tool
        registered before spawning; an entry with no ``writes`` key defaulted to
        a whole-workspace lease.
    """
    return json.dumps(
        _create_graph_impl(tasks, goal, target_entry_id, run_id, rigor, max_extensions)
    )

get_ready_tasks

get_ready_tasks(run_id: str = '') -> str

Use this to get the next wave: the pending tasks whose dependencies are all satisfied (they can run in parallel).

These are the current "wave" -- all returned tasks can run in parallel.

Parameters:

Name	Type	Description	Default
run_id	str	Which run to query. Optional when a single run is active; required to disambiguate when multiple runs are running in parallel.	''

Returns:

Type	Description
str	JSON list of ready tasks with their id, alias, agent, and description.

Source code in coordinator/server.py

@mcp_server.tool()
def get_ready_tasks(run_id: str = "") -> str:
    """Use this to get the next wave: the pending tasks whose dependencies are all satisfied (they can run in parallel).

    These are the current "wave" -- all returned tasks can run in parallel.

    Args:
        run_id: Which run to query. Optional when a single run is active;
            required to disambiguate when multiple runs are running in parallel.

    Returns:
        JSON list of ready tasks with their id, alias, agent, and description.
    """
    _, run_graph = _resolve_graph(run_id)
    ready = run_graph.get_ready_tasks()
    result = []
    for t in ready:
        entry: dict = {
            "task_id": t.id,
            "alias": t.alias,
            "agent": t.agent,
            "description": t.description,
        }
        if t.model:
            entry["model"] = t.model
        if t.schema_obj:
            entry["schema"] = t.schema_obj
        if t.passes_total > 1:
            entry["passes_total"] = t.passes_total
        # Surface the declared write scope so the coordinator sees which
        # implementers can be claimed in parallel (disjoint) vs. will serialize
        # (overlapping / whole-workspace).
        if t.writes is not None:
            entry["writes"] = list(t.writes)
        result.append(entry)
    return json.dumps(result)

claim_task

claim_task(task_id: str, run_id: str = '') -> str

Use this right before spawning a subagent: mark the task as running and take its write lease.

Implementer (engineer/scribe) tasks coordinate on a process-wide path-scoped write lease shared across all runs, so two implementers with DISJOINT write scopes run concurrently while two that touch the SAME path serialize. The lease scope comes from the task's writes:

• omitted -> a whole-workspace lease (the conservative default; reproduces the legacy one-implementer-at-a-time behavior).
• a list of paths -> leases on just those subtrees (disjoint declarations run in parallel).
• an empty list -> dynamic mode: no claim-time lease; the agent leases paths at write time via manage_paths(action="acquire"/"release").

If the claim-time lease conflicts with one already held, this returns {"blocked": true, "held_by": {...}} WITHOUT claiming the task. Spawn the other ready tasks meanwhile. Active task leases release on submit_result or manage_paths(action="release"). For a graph with a final memory task, acquired paths also remain reserved to the run until that memory task submits; held_by.reserved identifies this case. Read-only tasks (planner/checker/meta) never take an active lease, so review/critique/test waves run fully in parallel; the lease is also how the zettelkasten store's per-box locking is complemented for cooperative engineers.

Parameters:

Name	Type	Description	Default
task_id	str	The ID of the task to claim.	required
run_id	str	Optional owning run. Inferred from the task ID when omitted.	''

Returns:

Type	Description
str	JSON confirmation with task details, or a blocked notice when the
str	write lease conflicts with another implementer's.

Source code in coordinator/server.py

@mcp_server.tool()
def claim_task(task_id: str, run_id: str = "") -> str:
    """Use this right before spawning a subagent: mark the task as running and take its write lease.

    Implementer (engineer/scribe) tasks coordinate on a process-wide
    path-scoped write lease shared across all runs, so two implementers with
    DISJOINT write scopes run concurrently while two that touch the SAME path
    serialize. The lease scope comes from the task's ``writes``:

    - omitted -> a whole-workspace lease (the conservative default; reproduces
      the legacy one-implementer-at-a-time behavior).
    - a list of paths -> leases on just those subtrees (disjoint declarations
      run in parallel).
    - an empty list -> dynamic mode: no claim-time lease; the agent leases
      paths at write time via ``manage_paths(action="acquire"/"release")``.

    If the claim-time lease conflicts with one already held, this returns
    ``{"blocked": true, "held_by": {...}}`` WITHOUT claiming the task. Spawn the
    other ready tasks meanwhile. Active task leases release on ``submit_result``
    or ``manage_paths(action="release")``. For a graph with a final memory task,
    acquired paths also remain reserved to the run until that memory task
    submits; ``held_by.reserved`` identifies this case. Read-only tasks
    (planner/checker/meta) never take an active lease, so review/critique/test
    waves run fully in parallel; the lease is also how the zettelkasten store's
    per-box locking is complemented for cooperative engineers.

    Args:
        task_id: The ID of the task to claim.
        run_id: Optional owning run. Inferred from the task ID when omitted.

    Returns:
        JSON confirmation with task details, or a ``blocked`` notice when the
        write lease conflicts with another implementer's.
    """
    with _state_lock:
        rid, run_graph = _resolve_graph_for_task(task_id, run_id)
        task = run_graph._get_task(task_id)
        if _role_for_agent(task.agent) == "implementer":
            keys = lease_keys(task.writes)
            retain = _run_retains_write_reservations(
                run_graph, _get_run_meta(rid)
            )
            holder = _lock_mgr.try_acquire(
                task.id,
                keys,
                run_id=rid,
                agent=task.agent,
                reserve_for_run=retain,
            )
            if holder is not None:
                conflicting = holder.get("held_path", "")
                return json.dumps({
                    "blocked": True,
                    "task_id": task_id,
                    "agent": task.agent,
                    "held_by": holder,
                    "reason": (
                        "Another implementer holds a write lease overlapping "
                        f"{holder.get('requested_path', '')!r} (held: "
                        f"{conflicting!r}). Implementers serialize only on "
                        "OVERLAPPING write paths now. Spawn other ready tasks "
                        "meanwhile, then retry claim_task once the holder "
                        "releases; a run reservation lasts through its final "
                        "memory task. "
                        "To run in parallel, declare disjoint per-task `writes`."
                    ),
                })
        task = run_graph.claim_task(task_id)
        _persist_state_unlocked()
    result = {
        "task_id": task.id,
        "alias": task.alias,
        "agent": task.agent,
        "description": task.description,
        "status": task.status.value,
        "started_at": task.started_at.isoformat() if task.started_at else None,
    }
    if task.model:
        result["model"] = task.model
    if task.schema_obj:
        result["schema"] = task.schema_obj
    if task.passes_total > 1:
        result["passes_total"] = task.passes_total
    # Report the write lease this claim just took so a silent whole-workspace
    # lease (declared no `writes`) is visible rather than a surprise block on
    # the next implementer.
    if _role_for_agent(task.agent) == "implementer":
        if task.writes is not None:
            result["writes"] = list(task.writes)
        result["lease"] = _describe_lease(lease_keys(task.writes))
    return json.dumps(result)

manage_paths

manage_paths(action: Literal['acquire', 'release'], task_id: str, paths: list[str] | None = None, timeout_seconds: float = 30.0, run_id: str = '') -> str

Use this when a running task needs to lease or release specific file paths at write time (dynamic write coordination).

Pair an acquire immediately BEFORE writing a shared file with a release right after, so two agents in the same run never write the SAME file at once while still progressing on different files. In a memory-backed run, release drops the active task lease but preserves the run reservation; overlapping implementers in another run unblock after the memory task submits.

Actions — name(required, optional?): acquire(task_id, paths, timeout_seconds?, run_id?): take write leases on paths — subtree-covering, granted atomically (all-or-nothing) and deadlock-free regardless of order. BLOCKS until free or timeout_seconds elapses (waits instead of polling). Returns {"granted": true, "paths": [...]} or {"granted": false, "blocked": true, "held_by": {...}} on timeout. release(task_id, paths?, run_id?): release the task's active leases — pass paths for just those, omit to release ALL. Active leases also release automatically on submit_result; a memory-backed run's reservation remains until its memory task submits.

run_id is inferred from the task ID when omitted.

Returns:

Type	Description
str	JSON result of the chosen action.

Source code in coordinator/server.py

@mcp_server.tool()
def manage_paths(
    action: Literal["acquire", "release"],
    task_id: str,
    paths: list[str] | None = None,
    timeout_seconds: float = 30.0,
    run_id: str = "",
) -> str:
    """Use this when a running task needs to lease or release specific file paths at write time (dynamic write coordination).

    Pair an ``acquire`` immediately BEFORE writing a shared file with a
    ``release`` right after, so two agents in the same run never write the SAME
    file at once while still progressing on different files. In a memory-backed
    run, release drops the active task lease but preserves the run reservation;
    overlapping implementers in another run unblock after the memory task
    submits.

    Actions — name(required, optional?):
        acquire(task_id, paths, timeout_seconds?, run_id?): take write leases on
            ``paths`` — subtree-covering, granted atomically (all-or-nothing) and
            deadlock-free regardless of order. BLOCKS until free or
            ``timeout_seconds`` elapses (waits instead of polling). Returns
            ``{"granted": true, "paths": [...]}`` or
            ``{"granted": false, "blocked": true, "held_by": {...}}`` on timeout.
        release(task_id, paths?, run_id?): release the task's active leases —
            pass ``paths`` for just those, omit to release ALL. Active leases
            also release automatically on ``submit_result``; a memory-backed
            run's reservation remains until its memory task submits.

    ``run_id`` is inferred from the task ID when omitted.

    Returns:
        JSON result of the chosen action.
    """
    if action == "acquire":
        return _acquire_paths(
            task_id, paths or [], timeout_seconds=timeout_seconds, run_id=run_id
        )
    if action == "release":
        return _release_paths(task_id, paths, run_id=run_id)
    return _bad_action(manage_paths, action)

submit_result

submit_result(task_id: str, result: str, success: bool = True, run_id: str = '') -> str

Record the result of a completed task and advance the graph.

Parameters:

Name	Type	Description	Default
task_id	str	The ID of the task to submit results for.	required
result	str	The agent's output text.	required
success	bool	Whether the task succeeded (True) or failed (False).	True
run_id	str	Optional owning run. Inferred from the task ID when omitted.	''

Returns:

Type	Description
str	JSON confirmation with updated task state.

Source code in coordinator/server.py

@mcp_server.tool()
def submit_result(task_id: str, result: str, success: bool = True, run_id: str = "") -> str:
    """Record the result of a completed task and advance the graph.

    Args:
        task_id: The ID of the task to submit results for.
        result: The agent's output text.
        success: Whether the task succeeded (True) or failed (False).
        run_id: Optional owning run. Inferred from the task ID when omitted.

    Returns:
        JSON confirmation with updated task state.
    """
    rid, run_graph = _resolve_graph_for_task(task_id, run_id)
    task = run_graph.submit_result(task_id, result, success=success)
    try:
        task.parsed_result = parse_agent_output(result, agent=task.agent) or None
    except Exception as e:
        logger.warning("Failed to parse output for task %s: %s", task_id, e)
        task.parsed_result = None
    # Release every write lease still held by this task (claim-time lease plus
    # any dynamic manage_paths(action="acquire") leases the agent forgot to release).
    _lock_mgr.release(task_id)
    # A memory task's synchronous record(files=...) calls have now completed, so
    # another run may safely edit the retained paths. Release on failure too:
    # the run is terminal and keeping a reservation would deadlock future work.
    if task.agent == "memory":
        _lock_mgr.release_run(rid)
    _persist_state()
    return json.dumps({
        "task_id": task.id,
        "status": task.status.value,
        "completed_at": task.completed_at.isoformat() if task.completed_at else None,
    })

record_pass

record_pass(task_id: str, summary: str = '', run_id: str = '') -> str

Record completion of one self-refinement pass for a running task.

Call this after each pass of a multi-pass agent (reviewer/critic with passes > 1) -- i.e. after the initial spawn and after each resume. It advances the task's pass counter so the dashboard shows live round progress (e.g. "2/3"). This does NOT complete the task; call submit_result once at the end with the union of findings.

Parameters:

Name	Type	Description	Default
task_id	str	The ID of the running task.	required
summary	str	Optional one-line summary of what this pass found (shown in the dashboard tooltip).	''
run_id	str	Optional owning run. Inferred from the task ID when omitted.	''

Returns:

Type	Description
str	JSON with the updated pass progress.

Source code in coordinator/server.py

@mcp_server.tool()
def record_pass(task_id: str, summary: str = "", run_id: str = "") -> str:
    """Record completion of one self-refinement pass for a running task.

    Call this after each pass of a multi-pass agent (reviewer/critic with
    ``passes > 1``) -- i.e. after the initial spawn and after each ``resume``.
    It advances the task's pass counter so the dashboard shows live round
    progress (e.g. "2/3"). This does NOT complete the task; call
    ``submit_result`` once at the end with the union of findings.

    Args:
        task_id: The ID of the running task.
        summary: Optional one-line summary of what this pass found (shown in
            the dashboard tooltip).
        run_id: Optional owning run. Inferred from the task ID when omitted.

    Returns:
        JSON with the updated pass progress.
    """
    _, run_graph = _resolve_graph_for_task(task_id, run_id)
    task = run_graph.record_pass(task_id, summary=summary or None)
    _persist_state()
    return json.dumps({
        "task_id": task.id,
        "passes_done": task.passes_done,
        "passes_total": task.passes_total,
        "pass_notes": list(task.pass_notes),
    })

get_task_context

get_task_context(task_id: str, run_id: str = '') -> str

Get the full context needed to build a subagent prompt.

Returns the current task's info plus all predecessor outputs.

Parameters:

Name	Type	Description	Default
task_id	str	The ID of the task to get context for.	required
run_id	str	Optional owning run. Inferred from the task ID when omitted.	''

Returns:

Type	Description
str	JSON with "task" (current task info) and "predecessor_outputs" (list).

Source code in coordinator/server.py

@mcp_server.tool()
def get_task_context(task_id: str, run_id: str = "") -> str:
    """Get the full context needed to build a subagent prompt.

    Returns the current task's info plus all predecessor outputs.

    Args:
        task_id: The ID of the task to get context for.
        run_id: Optional owning run. Inferred from the task ID when omitted.

    Returns:
        JSON with "task" (current task info) and "predecessor_outputs" (list).
    """
    _, run_graph = _resolve_graph_for_task(task_id, run_id)
    task = run_graph._get_task(task_id)
    predecessors = run_graph.get_task_context(task_id)
    task_info: dict = {
        "task_id": task.id,
        "alias": task.alias,
        "agent": task.agent,
        "description": task.description,
    }
    if task.model:
        task_info["model"] = task.model
    if task.schema_obj:
        task_info["schema"] = task.schema_obj
    return json.dumps({
        "task": task_info,
        "predecessor_outputs": predecessors,
    })

extend_graph

extend_graph(tasks: list[dict], after_task_ids: list[str], context: str = '', run_id: str = '', force: bool = False, reason: str = '') -> str

Append corrective tasks to the graph after one or more failed tasks.

Use this when tasks fail and the coordinator decides to add a fix cycle. The graph grows forward -- nothing is reset or replayed.

The first new task depends on ALL after_task_ids, so it receives context from every predecessor in the failed wave.

OVERSIGHT -- stay on the graph. The extension cap is TIERED, not a single hard wall: * Below the soft cap -> the fix cycle is appended normally. * AT the soft cap -> this returns {"soft_cap_reached": true, ...} instead of appending. That is an escalation gate, NOT a cue to keep iterating off-graph. Prefer escalating to the user (a fresh graph or a human decision). If continuing is clearly right, re-call with force=true and a reason -- the override is recorded and shown on the dashboard (forced_extensions), so it is visible, never silent. * AT the hard ceiling -> a hard error (runaway wall); force cannot pass. Do NOT fall back to driving raw fix subagents (the Task tool) yourself: every fix cycle MUST run through the graph so the coordinator records task activity. A run driven off-graph logs no claim_task/submit_result progress, so the dashboard loses visibility and, after the idle window, the run is released as a stale/interrupted record (see manage_runs -- this is by design). To continue legitimately, force-with-reason here, escalate for a fresh graph, or resume an interrupted run with manage_runs(action="resume"). Never blindly carry on without a graph. (If you legitimately spawn ad-hoc helper subagents outside the wave structure, register them with track_subtask so they stay visible and the run does not go stale.)

Parameters:

Name	Type	Description	Default
tasks	list[dict]	List of task dicts (same format as create_graph). As in create_graph, every implementer (engineer/scribe) task MUST fill writes explicitly -- a narrow path list, [] for dynamic mode, or ["."] for a deliberate whole-workspace lease. Prefer the narrowest scope the fix actually needs.	required
after_task_ids	list[str]	Task IDs to chain after (typically the failed tasks from a wave).	required
context	str	Failure context to inject into the first new task.	''
run_id	str	Optional owning run. Inferred from after_task_ids when omitted.	''
force	bool	Override the soft cap (still bounded by the hard ceiling). Pass a reason whenever forcing.	False
reason	str	Why the soft-cap override is justified. Recorded on the run and surfaced on the dashboard.	''

Returns:

Type	Description
str	JSON with new task IDs and extension metadata, OR a structured
str	{"soft_cap_reached": true, ...} payload when the soft cap is hit
str	without force.

Source code in coordinator/server.py

@mcp_server.tool()
def extend_graph(
    tasks: list[dict],
    after_task_ids: list[str],
    context: str = "",
    run_id: str = "",
    force: bool = False,
    reason: str = "",
) -> str:
    """Append corrective tasks to the graph after one or more failed tasks.

    Use this when tasks fail and the coordinator decides to add a fix cycle.
    The graph grows forward -- nothing is reset or replayed.

    The first new task depends on ALL after_task_ids, so it receives context
    from every predecessor in the failed wave.

    OVERSIGHT -- stay on the graph. The extension cap is TIERED, not a single
    hard wall:
      * Below the soft cap -> the fix cycle is appended normally.
      * AT the soft cap -> this returns ``{"soft_cap_reached": true, ...}``
        instead of appending. That is an escalation gate, NOT a cue to keep
        iterating off-graph. Prefer escalating to the user (a fresh graph or a
        human decision). If continuing is clearly right, re-call with
        ``force=true`` and a ``reason`` -- the override is recorded and shown on
        the dashboard (``forced_extensions``), so it is visible, never silent.
      * AT the hard ceiling -> a hard error (runaway wall); force cannot pass.
    Do NOT fall back to driving raw fix subagents (the Task tool) yourself:
    every fix cycle MUST run through the graph so the coordinator records task
    activity. A run driven off-graph logs no claim_task/submit_result progress,
    so the dashboard loses visibility and, after the idle window, the run is
    released as a stale/interrupted record (see manage_runs -- this is by
    design). To continue legitimately, force-with-reason here, escalate for a
    fresh graph, or resume an interrupted run with manage_runs(action="resume").
    Never blindly carry on without a graph. (If you legitimately spawn ad-hoc
    helper subagents outside the wave structure, register them with
    track_subtask so they stay visible and the run does not go stale.)

    Args:
        tasks: List of task dicts (same format as create_graph). As in
            create_graph, every implementer (engineer/scribe) task MUST fill
            ``writes`` explicitly -- a narrow path list, ``[]`` for dynamic
            mode, or ``["."]`` for a deliberate whole-workspace lease. Prefer
            the narrowest scope the fix actually needs.
        after_task_ids: Task IDs to chain after (typically the failed tasks
                        from a wave).
        context: Failure context to inject into the first new task.
        run_id: Optional owning run. Inferred from after_task_ids when omitted.
        force: Override the soft cap (still bounded by the hard ceiling). Pass
            a ``reason`` whenever forcing.
        reason: Why the soft-cap override is justified. Recorded on the run and
            surfaced on the dashboard.

    Returns:
        JSON with new task IDs and extension metadata, OR a structured
        ``{"soft_cap_reached": true, ...}`` payload when the soft cap is hit
        without ``force``.
    """
    resolved_run_id, run_graph = _resolve_graph_for_tasks(after_task_ids, run_id)
    run = _get_run_meta(resolved_run_id)
    if run is None:
        run = next(
            (r for r in _runs
             if any(tid in r["task_ids"] for tid in after_task_ids)),
            None,
        )
    # New fix tasks follow the run's rigor profile for pass defaults.
    rigor_passes = resolve_rigor(
        run.get("rigor") if run else None, config=_local_config
    )["passes"]
    forced_before = run_graph.forced_extensions
    try:
        task_ids = run_graph.extend_graph(
            tasks,
            after_task_ids,
            context=context,
            agent_models=_merged_models(),
            agent_passes=_merged_passes(rigor_passes=rigor_passes),
            agent_schemas=_merged_agent_schemas(),
            schema_resolver=contrib.expand_schema,
            schema_enabled=_schema_enabled,
            force=force,
            reason=reason,
        )
    except SoftCapReached as e:
        # Escalation gate, not a wall: tell the caller exactly how to proceed.
        return json.dumps({
            "soft_cap_reached": True,
            "can_force": True,
            "extensions_used": e.extensions_used,
            "max_extensions": e.max_extensions,
            "hard_max_extensions": e.hard_max,
            "run_id": resolved_run_id,
            "message": (
                "Soft extension cap reached. Escalate to the user (a fresh "
                "graph or a human decision), or re-call extend_graph with "
                "force=true and a reason to continue under oversight. Do not "
                "drive fix cycles off-graph."
            ),
        })
    if run is not None:
        run["task_ids"].update(task_ids)
        run["extensions"] = run.get("extensions", 0) + 1
        # Record a forced override loudly so the dashboard shows it.
        if run_graph.forced_extensions > forced_before:
            run["forced_extensions"] = run_graph.forced_extensions
            run.setdefault("force_reasons", []).append({
                "at": _utc_now(),
                "extension": run_graph.extensions_count,
                "reason": reason or "",
            })
    _persist_state()
    return json.dumps({
        "new_task_ids": task_ids,
        "after_task_ids": after_task_ids,
        "extensions_used": run_graph.extensions_count,
        "max_extensions": run_graph._max_extensions,
        "forced_extensions": run_graph.forced_extensions,
        "hard_max_extensions": run_graph.hard_max_extensions,
    })

track_subtask

track_subtask(run_id: str, label: str, status: str = 'started', agent: str = '', note: str = '', subtask_id: str = '') -> str

Register an ad-hoc "loose" subtask against a run for visibility/recovery.

Use this when an agent legitimately spawns its OWN helper subagents outside the wave/DAG structure (work the coordinator did not create via create_graph / extend_graph). A loose task is OBSERVED, NOT GOVERNED: it gets dashboard visibility and keeps the run's activity clock fresh -- so the run is not released as stale while real work is happening -- but it gets NONE of the structural guarantees of a graph task: no DAG dependencies, no role validation, no write-lease (so concurrent loose writers can still collide), and it does not count against the extension cap.

IMPORTANT: ping a real subtask BOUNDARY -- "started" when you spawn the helper, "completed"/"failed" when it finishes -- NEVER as a periodic keepalive. Faking liveness re-creates the zombie-run problem that release-on-abandon exists to prevent (heartbeat != run progress).

Parameters:

Name	Type	Description	Default
run_id	str	The run this subtask belongs to (required). Must be a run owned by this coordinator process (resume it first if it is interrupted).	required
label	str	Short human description of the subtask.	required
status	str	"started" (default), "completed", or "failed".	'started'
agent	str	Optional label for the spawned helper (e.g. "engineer").	''
note	str	Optional free-form note (e.g. what it produced).	''
subtask_id	str	Stable id to update an existing loose task across pings. Auto-generated on the first "started" ping if omitted; pass it back on the closing ping. If omitted when closing, the most recently started OPEN loose task with the same label is closed.	''

Returns:

Type	Description
str	JSON with the subtask_id and its recorded state.

Source code in coordinator/server.py

@mcp_server.tool()
def track_subtask(
    run_id: str,
    label: str,
    status: str = "started",
    agent: str = "",
    note: str = "",
    subtask_id: str = "",
) -> str:
    """Register an ad-hoc "loose" subtask against a run for visibility/recovery.

    Use this when an agent legitimately spawns its OWN helper subagents outside
    the wave/DAG structure (work the coordinator did not create via create_graph
    / extend_graph). A loose task is OBSERVED, NOT GOVERNED: it gets dashboard
    visibility and keeps the run's activity clock fresh -- so the run is not
    released as stale while real work is happening -- but it gets NONE of the
    structural guarantees of a graph task: no DAG dependencies, no role
    validation, no write-lease (so concurrent loose writers can still collide),
    and it does not count against the extension cap.

    IMPORTANT: ping a real subtask BOUNDARY -- "started" when you spawn the
    helper, "completed"/"failed" when it finishes -- NEVER as a periodic
    keepalive. Faking liveness re-creates the zombie-run problem that
    release-on-abandon exists to prevent (heartbeat != run progress).

    Args:
        run_id: The run this subtask belongs to (required). Must be a run owned
            by this coordinator process (resume it first if it is interrupted).
        label: Short human description of the subtask.
        status: "started" (default), "completed", or "failed".
        agent: Optional label for the spawned helper (e.g. "engineer").
        note: Optional free-form note (e.g. what it produced).
        subtask_id: Stable id to update an existing loose task across pings.
            Auto-generated on the first "started" ping if omitted; pass it back
            on the closing ping. If omitted when closing, the most recently
            started OPEN loose task with the same label is closed.

    Returns:
        JSON with the subtask_id and its recorded state.
    """
    status = (status or "started").strip().lower()
    if status not in _LOOSE_STATUSES:
        return _err(f"Invalid status '{status}'. Use one of {list(_LOOSE_STATUSES)}.", "BadRequest")
    if not run_id:
        return _err("track_subtask requires run_id.", "BadRequest")
    if not label:
        return _err("track_subtask requires a label.", "BadRequest")

    with _state_lock:
        run = _get_run_meta(run_id)
        if run is None:
            return _err(
                f"Run {run_id} is not owned by this process. Resume it with "
                "manage_runs(action='resume') before tracking subtasks.",
                "BadRequest",
                owned_runs=[r["run_id"] for r in _runs],
            )
        loose = run.setdefault("loose_tasks", {})
        now = _utc_now()

        sid = subtask_id.strip()
        if not sid and status != "started":
            # Closing ping with no id: close the newest still-open loose task
            # with this label, else fall through to a fresh (already-closed) record.
            open_matches = [
                lt for lt in loose.values()
                if lt.get("label") == label and lt.get("status") == "started"
            ]
            if open_matches:
                sid = max(
                    open_matches, key=lambda lt: lt.get("started_at") or ""
                )["subtask_id"]
        if not sid:
            sid = f"lt-{uuid.uuid4().hex[:8]}"

        entry = loose.get(sid) or {
            "subtask_id": sid,
            "label": label,
            "agent": agent,
            "status": status,
            "note": note,
            "started_at": "",
            "completed_at": "",
        }
        entry["label"] = label or entry["label"]
        if agent:
            entry["agent"] = agent
        if note:
            entry["note"] = note
        if status == "started":
            entry["started_at"] = entry.get("started_at") or now
            entry["status"] = "started"
        else:
            entry["started_at"] = entry.get("started_at") or now
            entry["completed_at"] = now
            entry["status"] = status
        entry["updated_at"] = now
        loose[sid] = entry

        # A real subtask boundary is genuine progress: refresh the activity
        # floor so the run is not released while loose work is in flight.
        run["activity_at"] = now
        _persist_state_unlocked()

    return json.dumps({
        "run_id": run_id,
        "subtask_id": sid,
        "status": entry["status"],
        "label": entry["label"],
        "loose": True,
    })

get_report

get_report(run_id: str = '') -> str

Report graph state: summary counts, extension info, and a per-task timeline.

Serves both live status checks (per-task status and dependencies in the timeline) and the final execution report. Includes any ad-hoc loose_tasks (registered via track_subtask) separately from the governed timeline.

Parameters:

Name	Type	Description	Default
run_id	str	Which run to report on. Optional when a single run is active; required to disambiguate when multiple runs are running in parallel.	''

Returns:

Type	Description
str	JSON report with summary counts, extension info, timeline, and loose tasks.

Source code in coordinator/server.py

@mcp_server.tool()
def get_report(run_id: str = "") -> str:
    """Report graph state: summary counts, extension info, and a per-task timeline.

    Serves both live status checks (per-task status and dependencies in the
    timeline) and the final execution report. Includes any ad-hoc ``loose_tasks``
    (registered via track_subtask) separately from the governed timeline.

    Args:
        run_id: Which run to report on. Optional when a single run is active;
            required to disambiguate when multiple runs are running in parallel.

    Returns:
        JSON report with summary counts, extension info, timeline, and loose tasks.
    """
    resolved_run_id, run_graph = _resolve_graph(run_id)
    report = run_graph.get_report()
    run = _get_run_meta(resolved_run_id)
    if run is not None:
        loose = _loose_task_entries(run)
        report.setdefault("summary", {})["loose"] = len(loose)
        if loose:
            report["loose_tasks"] = loose
        if run.get("force_reasons"):
            report["force_reasons"] = run["force_reasons"]
    return json.dumps(report, indent=2)

manage_runs

manage_runs(action: Literal['resume', 'cleanup', 'finalize'], run_id: str = '', max_age_seconds: int = 1800, dry_run: bool = False, repo: str = '') -> str

Use this when a coordinator run was interrupted or went stale: resume, clean up, or finalize runs no longer being driven.

A run becomes resumable when its owning process dies (chat closed, MCP restart, crash) or it goes idle (no claim_task/submit_result for COORDINATOR_ABANDON_AFTER_SECONDS, default 2h — its heartbeat freezes and it stops showing "Live"). Going idle is BY DESIGN; the failure mode to avoid is driving work off-graph so the graph goes idle WHILE work continues unsupervised — RESUME (or create a new graph) instead.

Actions — name(required, optional?): resume(run_id): adopt an interrupted run into this process, then continue the wave loop (get_ready_tasks → claim_task → spawn → submit_result). Done/failed tasks keep their results; "running" tasks reset to pending. No repo — a run resumes only into its own repo's process. cleanup(run_id?, max_age_seconds?, dry_run?, repo?): remove abandoned runs from state; no run_id = sweep all incomplete foreign runs whose heartbeat is older than max_age_seconds. dry_run previews. finalize(run_id, max_age_seconds?, repo?): mark a stranded run terminally complete — unfinished (pending/running) tasks become "cancelled", already-done work and results are preserved.

repo targets a federated peer repo (memory-federation id, or a filesystem path) for cleanup/finalize only; a peer run still heartbeating is refused.

Returns:

Type	Description
str	JSON result of the chosen action.

Source code in coordinator/server.py

@mcp_server.tool()
def manage_runs(
    action: Literal["resume", "cleanup", "finalize"],
    run_id: str = "",
    max_age_seconds: int = 1800,
    dry_run: bool = False,
    repo: str = "",
) -> str:
    """Use this when a coordinator run was interrupted or went stale: resume, clean up, or finalize runs no longer being driven.

    A run becomes resumable when its owning process dies (chat closed, MCP
    restart, crash) or it goes idle (no claim_task/submit_result for
    COORDINATOR_ABANDON_AFTER_SECONDS, default 2h — its heartbeat freezes and it
    stops showing "Live"). Going idle is BY DESIGN; the failure mode to avoid is
    driving work off-graph so the graph goes idle WHILE work continues
    unsupervised — RESUME (or create a new graph) instead.

    Actions — name(required, optional?):
        resume(run_id): adopt an interrupted run into this process, then continue
            the wave loop (get_ready_tasks → claim_task → spawn → submit_result).
            Done/failed tasks keep their results; "running" tasks reset to pending.
            No ``repo`` — a run resumes only into its own repo's process.
        cleanup(run_id?, max_age_seconds?, dry_run?, repo?): remove abandoned runs
            from state; no run_id = sweep all incomplete foreign runs whose
            heartbeat is older than ``max_age_seconds``. ``dry_run`` previews.
        finalize(run_id, max_age_seconds?, repo?): mark a stranded run terminally
            complete — unfinished (pending/running) tasks become "cancelled",
            already-done work and results are preserved.

    ``repo`` targets a federated peer repo (memory-federation id, or a filesystem
    path) for cleanup/finalize only; a peer run still heartbeating is refused.

    Returns:
        JSON result of the chosen action.
    """
    if action == "resume":
        if repo:
            return _err(
                "action='resume' cannot target a federated repo — a run is "
                "only resumable into the coordinator process bound to its own "
                "repo. Use 'cleanup' or 'finalize' for peer runs.",
                "BadRequest",
            )
        if not run_id:
            return _needs(manage_runs, "resume", "action='resume' requires run_id.")
        return _resume_run(run_id)
    if action == "cleanup":
        if repo:
            return _cleanup_runs_federated(
                repo=repo, run_id=run_id, max_age_seconds=max_age_seconds, dry_run=dry_run
            )
        return _cleanup_runs(run_id=run_id, max_age_seconds=max_age_seconds, dry_run=dry_run)
    if action == "finalize":
        if not run_id:
            return _needs(manage_runs, "finalize", "action='finalize' requires run_id.")
        if repo:
            return _finalize_run_federated(
                repo=repo, run_id=run_id, max_age_seconds=max_age_seconds
            )
        return _finalize_run(run_id)
    return _bad_action(manage_runs, action)

create_extraction_graph

create_extraction_graph(sources: list[dict], schema: str, project: str, target_entry_id: str = '', hub_per_source: bool = True, batch_size: int = 0, synthesis_label: str = '', parent_label: str = '', spines: 'list[str] | None' = None, multi_spine: bool = False, synthesize: bool = True, slice_window: int = 0, slice_overlap: int = 0, slice_threshold: int = 0, reextract: bool = False) -> str

Build a grounded-extraction run over a corpus, parameterized by a schema.

Pattern-encodes the fixed pipeline so nobody hand-builds 3xN tasks. For each source it ingests the document into the zettelkasten and seeds a per-source hub note (synchronous, tool-side -- this is why the extractor, a planner, can run first). It then emits per-source extractor -> scribe -> auditor tasks, each carrying the named schema (the coordinator flows the expanded rubric into task context), plus a trailing memory task. Drive the returned run with the normal wave loop (get_ready_tasks -> claim_task -> spawn -> submit_result).

Parameters:

Name	Type	Description	Default
sources	list[dict]	List of source dicts. A PROSE source needs name (kebab-case graph id) and either path (the tool ingests it) or a precomputed content_hash (+ optional source_path). Optional metadata: title, authors, year, venue, doi, doc_type. A DATA source instead sets kind: "data" and carries graph (an EXISTING dataset box holding type=dataset notes) plus a list of derivations: prep SKIPS document ingest for it (there is nothing to slice -- it references a stored dataset), and it builds a data-extractor -> data-scribe -> auditor trio instead of the prose trio. Prose and data sources may be mixed in one run; sharing a synthesis_label attaches both to the SAME spine dimension nodes. BOOK-SCALE: a large source (long PDF / book) is auto-sliced into per-slice extractor->scribe trios (char windows, snapped to PDF chapter bookmarks when present); a paper stays a single trio. Pass an explicit slices list on a source to override the auto plan -- each entry is a page range ({"page_start": a, "page_end": b}), a char range ({"char_start": s, "char_end": e}), or a chapter ({"chapter": "<bookmark title>"}).	required
schema	str	Registry schema name (see the enabled capability's schemas).	required
project	str	Memory/zettelkasten project the run belongs to.	required
target_entry_id	str	Optional memory tree node for the run's records.	''
hub_per_source	bool	Seed a per-source hub note during prep (default True).	True
batch_size	int	0 = one wave of all extractors; N = fan out N at a time.	0
synthesis_label	str	Label for the materialized structure when the schema declares a synthesis block (feeds the {label} in the synthesis graph name + node titles). Defaults to the project name.	''
parent_label	str	Optional PARENT spine label for build-time sub-spine wiring: declares this (named-schema) spine a sub-spine of another, writing a primary child-apex --component-of--> parent-apex edge as the child materializes. The parent label mints the parent spine's synthesis graph the same way synthesis_label does, so the parent tier must already exist. A hand-authored parent edge is never overwritten (stability rule). Ignored on the spine-sourced path.	''
spines	'list[str] | None'	Optional list of project spine org ids to SOURCE the run's structure from (the spine directory, design §6-7). The project's default spine is ALWAYS auto-included; these are additional opt-ins, de-duped by org id. When the resolved scope yields a promoted spine the run attaches extracted claims onto that spine's EXISTING dimension nodes (via the spine's embedded spine-member relation) instead of minting a fresh scaffold from schema. With no default spine and no opt-ins the run keeps today's named-schema behavior exactly.	None
multi_spine	bool	Opt-in full multi-spine fill (default False = "first wins", byte-identical to today). When True, the run fills EVERY in-scope spine that resolves: each claim attaches to the matching dimension node of every spine whose dimension tag it fits (a spine whose dimensions it does not fit simply gets no edge), each spine gets its own keyed (project, synthesis_graph) extraction-meta, and the trio tasks carry one numbered STRUCTURE block per spine.	False
synthesize	bool	Append a trailing synthesizer task (default True) that runs ONCE after every source's trio. It places the run's claims onto the OTHER spine layouts of EVERY project the sources touch (the per-source scribe only attaches to the one spine its run chose, by tag-match; the synthesizer adds the cross-project / cross-vocabulary placements by JUDGEMENT) and wires cross-source themes into the shared _cross hubs. Set False to keep the legacy trio-only graph (extractor -> scribe -> auditor + memory).	True
slice_window	int	Target chars per slice for the book-scale fan-out (0 = default ~180k). Only affects sources above the threshold.	0
slice_overlap	int	Char overlap between consecutive slices so a quote straddling a boundary stays whole (0 = default ~2k).	0
slice_threshold	int	Char size above which a source is sliced (0 = default ~200k; sources at/below stay a single full trio).	0
reextract	bool	When True, bypass the per-slice incremental-skip filter and RE-DO everything -- every planned slice is extracted again even if the source's completed-slice ledger marks it done (default False: a re-run only mines the not-yet-extracted regions, and a source whose every slice is already done emits no trio). The per-source done vs todo split is surfaced in the returned graph_summary['slices'].	False

Returns:

Type	Description
str	JSON with run_id, task_ids, the per-task graph, the resolved schema
str	name, the prepared sources (name + hub_id), a graph_summary, and --
str	when the schema declares a synthesis block -- a structure
str	record (synthesis graph + apex/spec ids + dimension tag->node map,
str	plus missing_dims/partial for a half-covered chosen spine)
str	pre-created during prep and threaded into every task's context. Always
str	includes the scope decision: ignored_spines (also-resolved, deferred
str	to resync), broken_spines (scoped promoted spine that didn't
str	resolve), and unknown_spines (explicit opt-ins that didn't resolve).

Source code in coordinator/server.py

@mcp_server.tool()
def create_extraction_graph(
    sources: list[dict],
    schema: str,
    project: str,
    target_entry_id: str = "",
    hub_per_source: bool = True,
    batch_size: int = 0,
    synthesis_label: str = "",
    parent_label: str = "",
    spines: "list[str] | None" = None,
    multi_spine: bool = False,
    synthesize: bool = True,
    slice_window: int = 0,
    slice_overlap: int = 0,
    slice_threshold: int = 0,
    reextract: bool = False,
) -> str:
    """Build a grounded-extraction run over a corpus, parameterized by a schema.

    Pattern-encodes the fixed pipeline so nobody hand-builds 3xN tasks. For
    each source it ingests the document into the zettelkasten and seeds a
    per-source hub note (synchronous, tool-side -- this is why the extractor,
    a planner, can run first). It then emits per-source
    ``extractor -> scribe -> auditor`` tasks, each carrying the named
    ``schema`` (the coordinator flows the expanded rubric into task context),
    plus a trailing ``memory`` task. Drive the returned run with the normal
    wave loop (get_ready_tasks -> claim_task -> spawn -> submit_result).

    Args:
        sources: List of source dicts. A PROSE source needs ``name``
            (kebab-case graph id) and either ``path`` (the tool ingests it) or
            a precomputed ``content_hash`` (+ optional ``source_path``).
            Optional metadata: ``title``, ``authors``, ``year``, ``venue``,
            ``doi``, ``doc_type``. A DATA source instead sets ``kind: "data"``
            and carries ``graph`` (an EXISTING dataset box holding
            ``type=dataset`` notes) plus a list of ``derivations``: prep SKIPS
            document ingest for it (there is nothing to slice -- it references a
            stored dataset), and it builds a ``data-extractor -> data-scribe ->
            auditor`` trio instead of the prose trio. Prose and data sources may
            be mixed in one run; sharing a ``synthesis_label`` attaches both to
            the SAME spine dimension nodes.
            BOOK-SCALE: a large source (long PDF / book) is auto-sliced into
            per-slice extractor->scribe trios (char windows, snapped to PDF
            chapter bookmarks when present); a paper stays a single trio. Pass
            an explicit ``slices`` list on a source to override the auto plan
            -- each entry is a page range (``{"page_start": a, "page_end": b}``),
            a char range (``{"char_start": s, "char_end": e}``), or a chapter
            (``{"chapter": "<bookmark title>"}``).
        schema: Registry schema name (see the enabled capability's schemas).
        project: Memory/zettelkasten project the run belongs to.
        target_entry_id: Optional memory tree node for the run's records.
        hub_per_source: Seed a per-source hub note during prep (default True).
        batch_size: 0 = one wave of all extractors; N = fan out N at a time.
        synthesis_label: Label for the materialized structure when the schema
            declares a ``synthesis`` block (feeds the ``{label}`` in the
            synthesis graph name + node titles). Defaults to the project name.
        parent_label: Optional PARENT spine label for build-time sub-spine
            wiring: declares this (named-schema) spine a sub-spine of another,
            writing a primary ``child-apex --component-of--> parent-apex`` edge
            as the child materializes. The parent label mints the parent spine's
            synthesis graph the same way ``synthesis_label`` does, so the parent
            tier must already exist. A hand-authored parent edge is never
            overwritten (stability rule). Ignored on the spine-sourced path.
        spines: Optional list of project spine org ids to SOURCE the run's
            structure from (the spine directory, design §6-7). The project's
            default spine is ALWAYS auto-included; these are additional
            opt-ins, de-duped by org id. When the resolved scope yields a
            promoted spine the run attaches extracted claims onto that spine's
            EXISTING dimension nodes (via the spine's embedded ``spine-member``
            relation) instead of minting a fresh scaffold from ``schema``. With
            no default spine and no opt-ins the run keeps today's named-schema
            behavior exactly.
        multi_spine: Opt-in full multi-spine fill (default ``False`` = "first
            wins", byte-identical to today). When ``True``, the run fills EVERY
            in-scope spine that resolves: each claim attaches to the matching
            dimension node of every spine whose dimension tag it fits (a spine
            whose dimensions it does not fit simply gets no edge), each spine
            gets its own keyed ``(project, synthesis_graph)`` extraction-meta,
            and the trio tasks carry one numbered STRUCTURE block per spine.
        synthesize: Append a trailing ``synthesizer`` task (default ``True``)
            that runs ONCE after every source's trio. It places the run's
            claims onto the OTHER spine layouts of EVERY project the sources
            touch (the per-source scribe only attaches to the one spine its run
            chose, by tag-match; the synthesizer adds the cross-project /
            cross-vocabulary placements by JUDGEMENT) and wires cross-source
            themes into the shared ``_cross`` hubs. Set ``False`` to keep the
            legacy trio-only graph (extractor -> scribe -> auditor + memory).
        slice_window: Target chars per slice for the book-scale fan-out
            (``0`` = default ~180k). Only affects sources above the threshold.
        slice_overlap: Char overlap between consecutive slices so a quote
            straddling a boundary stays whole (``0`` = default ~2k).
        slice_threshold: Char size above which a source is sliced (``0`` =
            default ~200k; sources at/below stay a single ``full`` trio).
        reextract: When ``True``, bypass the per-slice incremental-skip filter
            and RE-DO everything -- every planned slice is extracted again even
            if the source's completed-slice ledger marks it done (default
            ``False``: a re-run only mines the not-yet-extracted regions, and a
            source whose every slice is already done emits no trio). The
            per-source ``done`` vs ``todo`` split is surfaced in the returned
            ``graph_summary['slices']``.

    Returns:
        JSON with run_id, task_ids, the per-task graph, the resolved schema
        name, the prepared sources (name + hub_id), a graph_summary, and --
        when the schema declares a ``synthesis`` block -- a ``structure``
        record (synthesis graph + apex/spec ids + dimension tag->node map,
        plus ``missing_dims``/``partial`` for a half-covered chosen spine)
        pre-created during prep and threaded into every task's context. Always
        includes the scope decision: ``ignored_spines`` (also-resolved, deferred
        to resync), ``broken_spines`` (scoped promoted spine that didn't
        resolve), and ``unknown_spines`` (explicit opt-ins that didn't resolve).
    """
    from coordinator import extraction

    schema_obj = contrib.expand_schema(schema)
    if schema_obj is None:
        return json.dumps({
            "error": (
                f"Unknown schema '{schema}'. Available: "
                f"{contrib.list_schemas()}"
            ),
            "type": "BadRequest",
        })

    try:
        prepped = extraction.prep_sources(
            sources, schema_obj, project, hub_per_source=hub_per_source
        )
        resolution = _resolve_extraction_structure(
            project, prepped, spines or [], multi=multi_spine
        )
        structure = resolution["structure"]
        # The full materialized list (back-compat: ``[structure]`` or ``[]``
        # when the resolver predates the multi-spine return). ``multi_spine``
        # fills every in-scope spine; the default keeps exactly one.
        structures = resolution.get("structures")
        if structures is None:
            structures = [structure] if structure is not None else []
        ignored_spines = resolution["ignored_spines"]
        broken_spines = resolution["broken_spines"]
        unknown_spines = resolution.get("unknown_spines", [])
        if unknown_spines:
            # An EXPLICIT opt-in that did not resolve to a usable spine must not
            # silently fall to the default -- name it so the caller can fix the
            # id / promote the lens instead of wondering why nothing changed.
            logger.warning(
                "extraction scope: explicit spine opt-in(s) %s did not resolve "
                "to a usable promoted spine (unknown org id / un-promoted lens / "
                "blank spine_ref); they were NOT used",
                unknown_spines,
            )
        if ignored_spines:
            # This run fills exactly ONE structure; name the other in-scope
            # spines that ALSO resolved so the loss is HONEST, not silent.
            logger.warning(
                "extraction scope: chose one in-scope spine; %d other in-scope "
                "spine(s) %s also resolved and are deferred to resync",
                len(ignored_spines), ignored_spines,
            )
        if broken_spines:
            # A scoped spine that points at a missing/empty graph: do not
            # silently mint a parallel scaffold as if nothing was scoped.
            logger.warning(
                "extraction scope: scoped spine(s) %s are BROKEN (spine_ref "
                "graph missing/empty or no materialized dimension matched); the "
                "run does NOT source its structure from them",
                broken_spines,
            )
        # DEMOTE-TO-FLAT (supersedes the old quarantine/exclude model): the
        # flat semantic graph is the always-present substrate and a spine is a
        # SEPARATE additive overlay, so a source whose ENFORCING keyed-meta
        # write fails is NEVER dropped from the run -- it simply does not join
        # that spine and extracts FLAT instead. The run is never aborted for a
        # spine-rubric failure. ``demoted_sources`` records the (source, spine)
        # pairs that fell back to flat; ``unused_structures`` records spines no
        # source could join (their orphaned per-source meta is pruned by
        # reconcile; the spine graph itself stays materialized).
        demoted_sources: list[dict] = []
        unused_structures: list[dict] = []
        # Per-source availability map threaded into build_extraction_tasks. It
        # is ``None`` for the common no-demotion case (every source sees the
        # same full spine set), so the render stays byte-identical to today.
        per_source: "dict[str, list] | None" = None
        if structure is not None:
            # A SPINE-sourced structure won: its dimension_nodes are keyed by
            # the spine's OWN vocabulary, which can differ from the named
            # schema's tags prep_sources persisted. Reconcile each source's
            # persisted strict-tag vocabulary to the chosen spine's tags so a
            # scribe's spine-tagged claims pass add_note strict enforcement.
            # Threaded ``project`` targets the keyed (project, spine
            # synthesis_graph) entry (P0); the run's strict/grounded/schema are
            # carried into that keyed entry too, since it is not seeded from the
            # flat block and the spine path runs no other keyed rubric write.
            # The named ``schema_obj`` values are passed only as a FALLBACK:
            # ``reconcile_structures`` PREFERS the spine's EMBEDDED rubric
            # (carried on each structure by ``prep_spine_from_org``).
            # MULTI-SPINE: it reconciles EACH resolved structure so every spine
            # gets its OWN keyed (project, synthesis_graph) rubric, and builds a
            # PER-SOURCE availability map -- a source whose write failed for a
            # spine is demoted to flat for THAT spine only (never dropped).
            source_structures, demoted_sources, unused_structures = (
                extraction.reconcile_structures(
                    prepped, structures, project=project,
                    schema=str(schema_obj.get("name") or ""),
                    strict=bool(schema_obj.get("strict", True)),
                    grounded=bool(schema_obj.get("grounded", True)),
                )
            )
            # SURVIVING spines (>=1 member) drive the run JSON + the back-compat
            # singular ``structure``; ``per_source`` is set only when sources
            # have heterogeneous available sets (some demotion from a survivor).
            structures, per_source = extraction.finalize_source_structures(
                structures, source_structures
            )
            structure = structures[0] if structures else None
        else:
            # No usable in-scope promoted spine (no default spine / intrinsic
            # lens / un-promoted / empty / broken) -> the named-schema path:
            # mint a fresh scaffold from ``schema``. NOTE: this is NOT
            # byte-identical to the pre-spine legacy run -- ``prep_spine``
            # now persists an ``extraction.structure`` block via
            # ``_persist_structure_meta`` (so ``note(action="attach")`` /
            # ``attach_to_dimension`` can resolve dimension nodes at write
            # time). The behavioral change is the persisted structure meta; the
            # task-graph SHAPE is otherwise the same as the legacy named-schema
            # run. (When no spines are scoped this is the only path taken.)
            try:
                structure = extraction.prep_spine(
                    schema_obj, project, prepped,
                    synthesis_label=synthesis_label,
                    parent_label=parent_label,
                )
            except extraction.StructurePersistError as exc:
                # ONLY a per-source RUBRIC/PERSIST WRITE failure for EVERY
                # source (``_persist_structure_meta`` raised the
                # ``StructurePersistError`` subtype) demotes the WHOLE run to
                # FLAT -- the sources still extract context-less into the flat
                # semantic graph. A genuine scaffold/config ``ExtractionError``
                # (an empty resolved synthesis-graph name, a wrapped
                # ``SpineError``) is NOT caught here: it propagates to the outer
                # handler and surfaces as a visible ``{error, type}`` JSON
                # rather than silently degrading a mis-configured run to flat.
                # (Deferred: a partially-built scaffold graph may be left
                # orphaned -- cleanup deferred.)
                logger.warning(
                    "extraction: named-schema structure could not be persisted "
                    "(%s); demoting the whole run to FLAT (sources still extract "
                    "context-less into the flat graph)", exc,
                )
                demoted_sources = [
                    {"source": p["name"], "synthesis_graph": "", "reason": str(exc)}
                    for p in prepped
                ]
                structure = None
                structures = []
            else:
                # ``prep_spine`` stashes its per-source quarantine on the
                # structure dict; pop it here. Those sources are DEMOTED to flat
                # for the named spine (not excluded): they keep extracting, just
                # context-less / hub-linked.
                q = (
                    structure.pop("quarantined", []) or []
                    if isinstance(structure, dict) else []
                )
                structures = [structure] if structure is not None else []
                if q and structure is not None:
                    sg = structure.get("synthesis_graph", "")
                    q_names = {x.get("source") for x in q}
                    demoted_sources = [
                        {"source": x.get("source"), "synthesis_graph": sg,
                         "reason": x.get("reason", "")}
                        for x in q
                    ]
                    source_structures = {
                        p["name"]: ([] if p["name"] in q_names else [structure])
                        for p in prepped
                    }
                    structures, per_source = (
                        extraction.finalize_source_structures(
                            structures, source_structures
                        )
                    )
                    structure = structures[0] if structures else None
        # Cross-cutting synthesis context: materialize every spine layout of
        # every project the sources touch (so the synthesizer's attaches
        # resolve) and render their dimension maps. Best-effort inside the
        # helper -- it returns "" on any failure, degrading the synthesizer to
        # synthesis-only rather than aborting the run. Gated on ``synthesize``.
        synthesis_context = ""
        connection_context = ""
        if synthesize:
            from zettelkasten.server import _get_graph as zk_get_graph

            synthesis_context = extraction.build_synthesis_context(
                prepped, project, get_graph=zk_get_graph,
                present_fn=_explicit_spine_present,
            )
            # DISCOVERY OFFLOAD: hand the synthesizer a prepared CONNECTIONS
            # worklist (flat cross-graph candidates + islands) so it adjudicates
            # rather than re-discovering cross-links. Best-effort: any failure
            # (or a fresh corpus whose claims are not yet committed) yields "",
            # which the synth task renders as "no prepared connections".
            connection_context = _prep_connections_context(project, prepped)
        # Book-scale slice knobs: 0 => use the extraction module defaults.
        _slice_kwargs: dict = {}
        if slice_window > 0:
            _slice_kwargs["slice_window"] = slice_window
        if slice_overlap > 0:
            _slice_kwargs["slice_overlap"] = slice_overlap
        if slice_threshold > 0:
            _slice_kwargs["slice_threshold"] = slice_threshold
        # Per-source done/todo slice breakdown for the graph summary (populated
        # by build_extraction_tasks; a source skipped entirely has empty todo).
        slice_report: dict[str, dict] = {}
        tasks = extraction.build_extraction_tasks(
            prepped, schema, project, batch_size=batch_size,
            structure=structure, structures=structures,
            source_structures=per_source,
            synthesis_context=synthesis_context,
            include_synthesis=synthesize,
            connection_context=connection_context,
            reextract=reextract,
            slice_report=slice_report,
            **_slice_kwargs,
        )
    except extraction.ExtractionError as e:
        return json.dumps({"error": str(e), "type": "ExtractionError"})

    # INCREMENTAL-SKIP NO-OP: every source was already fully covered by the
    # completed-slice ledger, so ``build_extraction_tasks`` emitted no trios.
    # It STILL appends a trailing ``memory`` task (include_memory defaults on),
    # so ``tasks`` is never empty -- testing ``if not tasks`` would miss this
    # and create a run (and a spurious memory session). Detect "no extraction
    # trios" by ignoring a lone trailing memory task instead, so an all-skipped
    # run short-circuits with a clean no-op result that still carries the skip
    # breakdown and a reextract hint. ``reextract`` can never reach here (it
    # forces every slice into ``todo``, so at least one trio is always built).
    if not any(t.get("agent") != "memory" for t in tasks):
        skipped = [
            name for name, rep in slice_report.items() if not rep.get("todo")
        ]
        return json.dumps({
            "status": "noop",
            "reason": "all sources already fully extracted (incremental skip); "
            "pass reextract=true to force a full re-extraction",
            "schema": schema,
            "sources": [
                {"name": p["name"], "title": p["title"], "hub_id": p["hub_id"]}
                for p in prepped
            ],
            "skipped_sources": skipped,
            "reextract": bool(reextract),
            "graph_summary": {
                "sources": len(prepped),
                "tasks": 0,
                "schema": schema,
                "reextract": bool(reextract),
                "slices": slice_report,
                "skipped_sources": len(skipped),
            },
        })

    _src_desc = extraction.describe_sources(prepped)
    goal = (
        f"Grounded extraction: {_src_desc} → {project} (schema: {schema})"
        if _src_desc
        else f"Grounded extraction: {project} (schema: {schema})"
    )
    result = _create_graph_impl(
        tasks,
        goal=goal,
        target_entry_id=target_entry_id,
    )
    result["schema"] = schema
    result["sources"] = [
        {"name": p["name"], "title": p["title"], "hub_id": p["hub_id"]}
        for p in prepped
    ]
    if structure:
        result["structure"] = {
            "synthesis_graph": structure.get("synthesis_graph", ""),
            "apex_id": structure.get("apex_id", ""),
            "spec_id": structure.get("spec_id", ""),
            "dimension_nodes": structure.get("dimension_nodes", {}),
            # PARTIAL coverage: embedded-schema dimension tags with no
            # materialized node in the chosen spine (empty for a full spine and
            # for the named-schema fallback). Makes a half-covered chosen spine
            # distinguishable from a fully-covered one in the run's JSON.
            "missing_dims": structure.get("missing_dims", []),
            "partial": bool(structure.get("missing_dims")),
        }
    # MULTI-SPINE: surface EVERY filled spine (one entry per in-scope spine).
    # For the default single-spine / named-schema run this is a one-element
    # list echoing ``structure``; under ``multi_spine`` it lists them all so a
    # caller can see each spine the claims were attached to.
    result["structures"] = [
        {
            "synthesis_graph": s.get("synthesis_graph", ""),
            "apex_id": s.get("apex_id", ""),
            "spec_id": s.get("spec_id", ""),
            "dimension_nodes": s.get("dimension_nodes", {}),
            "missing_dims": s.get("missing_dims", []),
            "partial": bool(s.get("missing_dims")),
        }
        for s in structures
        if s and s.get("dimension_nodes")
    ]
    # Surface the scope decision so the single-structure limitation is HONEST:
    # which other in-scope spines also resolved (deferred to resync), which
    # scoped spines were broken, and which explicit opt-ins did not resolve.
    result["ignored_spines"] = ignored_spines
    result["broken_spines"] = broken_spines
    result["unknown_spines"] = unknown_spines
    # DEMOTE-TO-FLAT diagnostics. ``demoted_sources`` lists the
    # (source, spine) pairs that fell back to flat because the source's
    # enforcing keyed extraction-meta write failed -- the source STILL extracts
    # (into the flat graph), it just did not join that spine. Empty for a clean
    # run. ``unused_structures`` lists spines NO source could join (their
    # orphaned per-source meta is pruned by reconcile; the spine graph itself
    # stays materialized).
    result["demoted_sources"] = demoted_sources
    unused_payload = [
        {"synthesis_graph": s.get("synthesis_graph", "")}
        for s in unused_structures if isinstance(s, dict)
    ]
    result["unused_structures"] = unused_payload
    # Back-compat: nothing is EXCLUDED anymore, so ``quarantined_sources`` is
    # always empty; ``dropped_structures`` aliases ``unused_structures``.
    result["quarantined_sources"] = []
    result["dropped_structures"] = unused_payload
    # Book-scale slice fan-out: count extractor tasks per source (alias is
    # ext_{i} for a single-trio source, ext_{i}_{k} for a sliced one) so the
    # caller sees which sources fanned out and by how much.
    per_source_slices: dict[int, int] = {}
    for t in tasks:
        if t.get("agent") != "extractor":
            continue
        parts = str(t.get("alias", "")).split("_")
        if len(parts) >= 2 and parts[1].isdigit():
            si = int(parts[1])
            per_source_slices[si] = per_source_slices.get(si, 0) + 1
    sliced_sources = [
        {"name": prepped[si]["name"], "slices": n}
        for si, n in sorted(per_source_slices.items())
        if n > 1 and si < len(prepped)
    ]
    result["sliced_sources"] = sliced_sources
    # Per-source incremental-skip breakdown: which slices were already done
    # (skipped) vs planned for this run. A source with an empty ``todo`` emitted
    # no trio (fully extracted); ``reextract`` forces every slice into ``todo``.
    result["skipped_sources"] = [
        name for name, rep in slice_report.items() if not rep.get("todo")
    ]
    result["graph_summary"] = {
        "sources": len(prepped),
        "tasks": result["total_tasks"],
        "schema": schema,
        "synthesis_graph": (structure or {}).get("synthesis_graph", ""),
        "extractor_slices": sum(per_source_slices.values()),
        "sliced_sources": len(sliced_sources),
        "reextract": bool(reextract),
        # {source_name: {"done": [labels], "todo": [labels]}}
        "slices": slice_report,
        "skipped_sources": sum(
            1 for rep in slice_report.values() if not rep.get("todo")
        ),
    }
    return json.dumps(result)

prep_synthesis_contexts

prep_synthesis_contexts(project: str, sources: 'list[str] | None' = None) -> str

Register spine extraction contexts so a STANDALONE synthesizer can place.

This is the synthesis-prep half of create_extraction_graph, exposed on its own so the synthesizer can run OUTSIDE a fresh extraction. A per-source scribe only attaches claims to the ONE spine its run chose, so each source ends up with a keyed (project, synthesis_graph) extraction context for that PRIMARY spine only. note(action="attach") for any OTHER spine then fails with ExtractionContextNotFound -- the context was never registered. Inside create_extraction_graph the trailing synthesizer is saved by build_synthesis_context, which materializes EVERY promoted spine's structure meta onto the run's sources first; a standalone synthesizer never hits that path.

Call this BEFORE spawning a standalone synthesizer over an already-extracted corpus. It materializes each promoted spine layout of every project the given sources touch -- writing the per-source keyed (project, synthesis_graph) structure meta (and the apex->hub rollup edges) that lets cross-spine attach resolve -- and returns the rendered SPINES blocks to thread into the synthesizer's task context (job 1). The writes are idempotent (structure-only, non-enforcing: they never set strict, so they add no write-time rejection and re-running is harmless) and this path never prunes contexts, so it is safe to run repeatedly.

Parameters:

Name	Type	Description	Default
project	str	The launch project (placed first; other projects the sources belong to are discovered from the manifests and also materialized).	required
sources	'list[str] | None'	Optional list of source GRAPH names (kebab-case ids) to register contexts for. When omitted, defaults to every source currently registered in project (the full corpus). These must already be ingested + project members (a prior extraction run) -- this tool registers spine contexts onto existing sources, it does NOT ingest. IMPORTANT -- contexts are PER-SOURCE, not global. Passing a subset registers the spine contexts onto ONLY those sources; every other source keeps just its primary-spine context, so a standalone synthesizer trying to cross-spine-place a claim from an un-prepped source still gets ExtractionContextNotFound. For a WHOLE-CORPUS cross-spine pass you MUST prep the whole project (omit sources, or pass all of them) -- prepping only the run's new/changed sources leaves the rest of the corpus (e.g. island papers on their primary spine only) un-placeable. Subsets are for the narrow "I only need these sources cross-placed right now" case.	None

Returns:

Type	Description
str	JSON with synthesis_context (the rendered SPINES string to paste
str	into the synthesizer task -- "" when no touched project has a
str	promoted spine, meaning cross-spine placement has nothing to resolve
str	against), contexts (the [{project, synthesis_graph}] list this
str	call registered), projects (every project the sources touched), and
str	sources (the resolved source graph names).

Source code in coordinator/server.py

@mcp_server.tool()
def prep_synthesis_contexts(
    project: str,
    sources: "list[str] | None" = None,
) -> str:
    """Register spine extraction contexts so a STANDALONE synthesizer can place.

    This is the synthesis-prep half of ``create_extraction_graph``, exposed on
    its own so the synthesizer can run OUTSIDE a fresh extraction. A per-source
    scribe only attaches claims to the ONE spine its run chose, so each source
    ends up with a keyed ``(project, synthesis_graph)`` extraction context for
    that PRIMARY spine only. ``note(action="attach")`` for any OTHER spine then
    fails with ``ExtractionContextNotFound`` -- the context was never
    registered. Inside ``create_extraction_graph`` the trailing synthesizer is
    saved by ``build_synthesis_context``, which materializes EVERY promoted
    spine's structure meta onto the run's sources first; a standalone
    synthesizer never hits that path.

    Call this BEFORE spawning a standalone synthesizer over an already-extracted
    corpus. It materializes each promoted spine layout of every project the
    given sources touch -- writing the per-source keyed ``(project,
    synthesis_graph)`` structure meta (and the apex->hub rollup edges) that lets
    cross-spine ``attach`` resolve -- and returns the rendered ``SPINES`` blocks
    to thread into the synthesizer's task context (job 1). The writes are
    idempotent (structure-only, non-enforcing: they never set ``strict``, so
    they add no write-time rejection and re-running is harmless) and this path
    never prunes contexts, so it is safe to run repeatedly.

    Args:
        project: The launch project (placed first; other projects the sources
            belong to are discovered from the manifests and also materialized).
        sources: Optional list of source GRAPH names (kebab-case ids) to
            register contexts for. When omitted, defaults to every source
            currently registered in ``project`` (the full corpus). These must
            already be ingested + project members (a prior extraction run) --
            this tool registers spine contexts onto existing sources, it does
            NOT ingest.

            IMPORTANT -- contexts are PER-SOURCE, not global. Passing a subset
            registers the spine contexts onto ONLY those sources; every other
            source keeps just its primary-spine context, so a standalone
            synthesizer trying to cross-spine-place a claim from an
            un-prepped source still gets ``ExtractionContextNotFound``. For a
            WHOLE-CORPUS cross-spine pass you MUST prep the whole project
            (omit ``sources``, or pass all of them) -- prepping only the run's
            new/changed sources leaves the rest of the corpus (e.g. island
            papers on their primary spine only) un-placeable. Subsets are for
            the narrow "I only need these sources cross-placed right now" case.

    Returns:
        JSON with ``synthesis_context`` (the rendered ``SPINES`` string to paste
        into the synthesizer task -- ``""`` when no touched project has a
        promoted spine, meaning cross-spine placement has nothing to resolve
        against), ``contexts`` (the ``[{project, synthesis_graph}]`` list this
        call registered), ``projects`` (every project the sources touched), and
        ``sources`` (the resolved source graph names).
    """
    from coordinator import extraction
    from zettelkasten import graph as zk_graph
    from zettelkasten.server import _get_graph as zk_get_graph

    if not (project or "").strip():
        return json.dumps({
            "error": "prep_synthesis_contexts requires a project.",
            "type": "BadRequest",
        })

    # Resolve the source set: explicit list, else the project's full corpus.
    src_names: list[str]
    if sources:
        src_names = [str(s or "").strip() for s in sources if str(s or "").strip()]
    else:
        try:
            pdata = zk_graph.load_project(project)
        except Exception as exc:  # noqa: BLE001 — a corrupt manifest is reported
            return json.dumps({
                "error": (
                    f"could not load project '{project}' to enumerate its "
                    f"sources: {exc}"
                ),
                "type": "ExtractionError",
            })
        src_names = [
            str(s or "").strip()
            for s in (pdata.get("sources") or [])
            if str(s or "").strip()
        ]
    if not src_names:
        return json.dumps({
            "error": (
                f"no sources to register: project '{project}' has no sources "
                f"(pass an explicit `sources` list, or extract first)."
            ),
            "type": "BadRequest",
        })

    # Minimal prep records: ``build_synthesis_context`` only needs each
    # source's graph ``name`` to write the keyed structure meta. ``hub_id`` is
    # left blank -- the apex->hub rollup edges were already written by the
    # original extraction run, and ``link_spine_hubs`` no-ops without a hub id,
    # so omitting it just skips a redundant idempotent re-link (never an error).
    prepped = [{"name": n, "hub_id": ""} for n in src_names]

    context_str, structures = extraction.build_synthesis_context(
        prepped, project, get_graph=zk_get_graph,
        present_fn=_explicit_spine_present, return_structures=True,
    )
    contexts = [
        {
            "project": str(s.get("project") or project),
            "synthesis_graph": str(s.get("synthesis_graph") or ""),
        }
        for s in structures
        if isinstance(s, dict) and s.get("synthesis_graph")
    ]
    projects = extraction.projects_touching(prepped, project)
    return json.dumps({
        "synthesis_context": context_str,
        "contexts": contexts,
        "projects": projects,
        "sources": src_names,
    })

prep_connection_candidates

prep_connection_candidates(project: str, sources: 'list[str] | None' = None, top_k: int = 8, similarity_threshold: float = 0.5, max_candidates: int = 500, min_cross_degree: int = 1) -> str

Render cross-graph connection candidates for a standalone synthesizer.

The job-2 (flat cross-graph connectivity) counterpart of prep_synthesis_contexts: that one hands the synthesizer its SPINES blocks (job 1, cross-spine placement); this one hands it a concrete worklist of CANDIDATE cross-graph edges to adjudicate (job 2) plus the island / under-connected audit -- so a whole-corpus pass does not force the synthesizer to recall the entire corpus from memory (which is why islands like a lone paper stay islands). Thin renderer over the zettelkasten suggest(kind="cross-connections") primitive; READ-ONLY (the synthesizer still writes each edge it judges real via note(action="link", ..., target_graph=...)).

Parameters:

Name	Type	Description	Default
project	str	The project whose sources to analyze.	required
sources	'list[str] | None'	Optional subset of source graph names to propose edges FROM (targets still range over the whole project). When omitted, proposes across every source in the project (the whole-project pass). These must already be project members.	None
top_k	int	Max candidate targets per source note.	8
similarity_threshold	float	Minimum cosine similarity to propose a pair. Cross-graph cosine is compressed (genuine neighbors ~0.5–0.66), so the default is 0.5; raise for precision, lower for recall.	0.5
max_candidates	int	Global cap on returned candidate pairs.	500
min_cross_degree	int	A content note with fewer cross-graph edges than this is reported under_connected.	1

Returns:

Type	Description
str	JSON with connections_context (a prompt-ready CONNECTIONS block to
str	thread into the synthesizer task; "" when nothing is missing),
str	candidates / islands / under_connected (the raw primitive
str	output), stats, and truncated.

Source code in coordinator/server.py

@mcp_server.tool()
def prep_connection_candidates(
    project: str,
    sources: "list[str] | None" = None,
    top_k: int = 8,
    similarity_threshold: float = 0.5,
    max_candidates: int = 500,
    min_cross_degree: int = 1,
) -> str:
    """Render cross-graph connection candidates for a standalone synthesizer.

    The job-2 (flat cross-graph connectivity) counterpart of
    ``prep_synthesis_contexts``: that one hands the synthesizer its SPINES
    blocks (job 1, cross-spine placement); this one hands it a concrete
    worklist of CANDIDATE cross-graph edges to adjudicate (job 2) plus the
    island / under-connected audit -- so a whole-corpus pass does not force the
    synthesizer to recall the entire corpus from memory (which is why islands
    like a lone paper stay islands). Thin renderer over the zettelkasten
    ``suggest(kind="cross-connections")`` primitive; READ-ONLY (the synthesizer
    still writes each edge it judges real via note(action="link", ...,
    target_graph=...)).

    Args:
        project: The project whose sources to analyze.
        sources: Optional subset of source graph names to propose edges FROM
            (targets still range over the whole project). When omitted,
            proposes across every source in the project (the whole-project
            pass). These must already be project members.
        top_k: Max candidate targets per source note.
        similarity_threshold: Minimum cosine similarity to propose a pair.
            Cross-graph cosine is compressed (genuine neighbors ~0.5–0.66), so
            the default is 0.5; raise for precision, lower for recall.
        max_candidates: Global cap on returned candidate pairs.
        min_cross_degree: A content note with fewer cross-graph edges than this
            is reported ``under_connected``.

    Returns:
        JSON with ``connections_context`` (a prompt-ready CONNECTIONS block to
        thread into the synthesizer task; ``""`` when nothing is missing),
        ``candidates`` / ``islands`` / ``under_connected`` (the raw primitive
        output), ``stats``, and ``truncated``.
    """
    from zettelkasten.coverage import suggest_cross_connections

    if not (project or "").strip():
        return json.dumps({
            "error": "prep_connection_candidates requires a project.",
            "type": "BadRequest",
        })

    src_list = [str(s or "").strip() for s in (sources or []) if str(s or "").strip()]

    def _run(graph_scope: str) -> dict:
        return json.loads(suggest_cross_connections(
            project=project, graph=graph_scope, top_k=top_k,
            similarity_threshold=similarity_threshold, target_scope="project",
            min_cross_degree=min_cross_degree, max_candidates=max_candidates,
        ))

    if src_list:
        # Per-source passes (propose FROM each named source, targets across the
        # project), merged + de-duped by unordered pair.
        merged: dict = {}
        islands: list = []
        island_seen: set = set()
        under: list = []
        under_seen: set = set()
        truncated = False
        for src in src_list:
            res = _run(src)
            if res.get("error"):
                return json.dumps(res)
            for c in res.get("candidates", []):
                key = frozenset({
                    (c["a"]["graph"], c["a"]["note_id"]),
                    (c["b"]["graph"], c["b"]["note_id"]),
                })
                if key not in merged or c["similarity"] > merged[key]["similarity"]:
                    merged[key] = c
            for isl in res.get("islands", []):
                if isl["graph"] not in island_seen:
                    island_seen.add(isl["graph"])
                    islands.append(isl)
            for u in res.get("under_connected", []):
                uk = (u["graph"], u["note_id"])
                if uk not in under_seen:
                    under_seen.add(uk)
                    under.append(u)
            truncated = truncated or bool(res.get("truncated"))
        candidates = sorted(merged.values(), key=lambda x: x["similarity"], reverse=True)
        if len(candidates) > max_candidates:
            candidates = candidates[:max_candidates]
            truncated = True
        result = {
            "candidates": candidates, "islands": islands,
            "under_connected": under[:max_candidates],
            "stats": {"pairs_proposed": len(candidates), "graphs_flagged": len(islands)},
            "truncated": truncated,
        }
    else:
        result = _run("")
        if result.get("error"):
            return json.dumps(result)

    result["connections_context"] = _render_connections_block(
        result.get("candidates", []), result.get("islands", []),
    )
    result["project"] = project
    result["sources"] = src_list or "(all project sources)"
    return json.dumps(result)