memory.dashboard.backend.routes.coordinator

memory.dashboard.backend.routes.coordinator

Coordinator-status routes (local + federated).

Split out of the former flat routes.py; behaviour is unchanged.

get_coordinator_agents

get_coordinator_agents() -> dict

Return the merged coordinator agent roster.

Combines the built-in agents (coordinator/agents.py) with any project-local agents and built-in overrides declared in agents.yaml (discovered via AGENTS_FILE env > .cursor/ > .claude/). This is the same roster the coordinator's agents(action="list") tool exposes, so the workflow builder palette stays in sync with what the coordinator can actually spawn.

Returns:

Type	Description
dict	A mapping {"agents": [...]} sorted by name, each agent carrying the keys name, role, description, model, passes, has_persona, and builtin; the agents list is empty on any failure to import or load the roster (so the UI falls back to its built-in defaults rather than erroring).

Source code in memory/dashboard/backend/routes/coordinator.py

@router.get("/coordinator/agents")
def get_coordinator_agents() -> dict:
    """Return the merged coordinator agent roster.

    Combines the built-in agents (``coordinator/agents.py``) with any
    project-local agents and built-in overrides declared in ``agents.yaml``
    (discovered via AGENTS_FILE env > .cursor/ > .claude/). This is the same
    roster the coordinator's ``agents(action="list")`` tool exposes, so the workflow
    builder palette stays in sync with what the coordinator can actually spawn.

    Returns:
        A mapping ``{"agents": [...]}`` sorted by name, each agent carrying the keys name, role, description, model, passes, has_persona, and builtin; the agents list is empty on any failure to import or load the roster (so the UI falls back to its built-in defaults rather than erroring).
    """
    try:
        from coordinator.agents import build_agent_registry
        return {"agents": build_agent_registry()}
    except Exception as e:
        logger.warning("Failed to build coordinator agent registry: %s", e)
        return {"agents": []}

get_coordinator_status

get_coordinator_status(request: Request = None, response: Response = None)

Return coordinator graphs, one per run.

Merges runs from both canonical and legacy state files. Filters out empty ghost runs (total == 0 with no tasks).

Handles two state file formats: - New: {"runs": [...]} with per-run metadata - Legacy: flat task list — split into connected components via union-find

Answers conditional GETs: an ETag from :func:_coordinator_status_etag_key (the state-file signature plus a live-run time bucket) lets an unchanged 2s poll collapse to a bodyless 304, skipping the whole assembly.

Source code in memory/dashboard/backend/routes/coordinator.py

@router.get("/coordinator-status")
def get_coordinator_status(request: Request = None, response: Response = None):
    """Return coordinator graphs, one per run.

    Merges runs from both canonical and legacy state files.
    Filters out empty ghost runs (total == 0 with no tasks).

    Handles two state file formats:
    - New: ``{"runs": [...]}`` with per-run metadata
    - Legacy: flat task list — split into connected components via union-find

    Answers conditional GETs: an ETag from :func:`_coordinator_status_etag_key`
    (the state-file signature plus a live-run time bucket) lets an unchanged 2s
    poll collapse to a bodyless 304, skipping the whole assembly.
    """
    not_modified = _conditional_etag(request, response, _coordinator_status_etag_key())
    if not_modified is not None:
        return not_modified

    runs = _load_coordinator_runs()
    if runs:
        graphs = []
        now = datetime.now(timezone.utc)
        for run in runs:
            summary = run.get("summary", {})
            total = summary.get("total", 0)
            if total == 0 and not run.get("tasks"):
                continue
            stale, age_seconds, heartbeat_at = _coordinator_stale_info(run, summary, now)
            graphs.append({
                "run_id": run.get("run_id", ""),
                "active": total > 0 and not summary.get("all_complete", True) and not stale,
                "stale": stale,
                "goal": run.get("goal", ""),
                "target_entry_id": run.get("target_entry_id", ""),
                "summary": {**summary, "stale": stale},
                "tasks": run.get("tasks", []),
                "created_at": run.get("created_at", ""),
                "updated_at": run.get("updated_at", ""),
                "heartbeat_at": heartbeat_at,
                "completed_at": run.get("completed_at", ""),
                "stale_after_seconds": COORDINATOR_STALE_AFTER_SECONDS,
                "heartbeat_age_seconds": age_seconds,
            })
        return {"graphs": graphs}

    # Fall back to reading the primary file for legacy flat format
    if not COORDINATOR_STATE_FILE.exists():
        return {"graphs": []}

    try:
        data = json.loads(COORDINATOR_STATE_FILE.read_text(encoding="utf-8"))
    except Exception:
        return {"graphs": []}
    try:
        state_updated_at = datetime.fromtimestamp(
            COORDINATOR_STATE_FILE.stat().st_mtime,
            tz=timezone.utc,
        )
    except OSError:
        state_updated_at = datetime.now(timezone.utc)

    # Legacy flat format — split tasks into connected components
    tasks = data.get("tasks", [])
    if not tasks:
        return {"graphs": []}

    parent_uf: dict[str, str] = {}

    def _find(x: str) -> str:
        while parent_uf.setdefault(x, x) != x:
            parent_uf[x] = parent_uf.get(parent_uf[x], parent_uf[x])
            x = parent_uf[x]
        return x

    def _union(a: str, b: str) -> None:
        ra, rb = _find(a), _find(b)
        if ra != rb:
            parent_uf[ra] = rb

    all_ids = {t["task_id"] for t in tasks}
    for t in tasks:
        for dep in t.get("depends_on", []):
            if dep in all_ids:
                _union(t["task_id"], dep)

    groups: dict[str, list] = {}
    for t in tasks:
        groups.setdefault(_find(t["task_id"]), []).append(t)

    goal = data.get("goal", "")
    target = data.get("target_entry_id", "")

    # Fallback: read sidecar metadata written by the coordinator agent
    # when the MCP server hasn't been restarted with run-tracking code yet
    if not goal or not target:
        meta_file = COORDINATOR_STATE_FILE.parent / ".coordinator-meta.json"
        if meta_file.exists():
            try:
                meta = json.loads(meta_file.read_text(encoding="utf-8"))
                goal = goal or meta.get("goal", "")
                target = target or meta.get("target_entry_id", "")
            except Exception:
                pass

    graphs = []
    for comp_tasks in groups.values():
        done = sum(1 for t in comp_tasks if t.get("status") == "done")
        failed = sum(1 for t in comp_tasks if t.get("status") == "failed")
        pend = sum(1 for t in comp_tasks if t.get("status") == "pending")
        run_ = sum(1 for t in comp_tasks if t.get("status") == "running")
        total = len(comp_tasks)
        is_active = run_ > 0 or pend > 0
        age_seconds = (datetime.now(timezone.utc) - state_updated_at).total_seconds()
        stale = is_active and age_seconds > COORDINATOR_STALE_AFTER_SECONDS
        graphs.append({
            "run_id": f"legacy-{comp_tasks[0]['task_id']}",
            "active": is_active and not stale,
            "stale": stale,
            "goal": goal,
            "target_entry_id": target,
            "summary": {
                "total": total, "done": done, "failed": failed,
                "pending": pend, "running": run_,
                "extensions_used": 0,
                "all_complete": total > 0 and done + failed == total,
                "stale": stale,
            },
            "tasks": comp_tasks,
            "updated_at": state_updated_at.isoformat(),
            "heartbeat_at": state_updated_at.isoformat(),
            "stale_after_seconds": COORDINATOR_STALE_AFTER_SECONDS,
            "heartbeat_age_seconds": age_seconds,
        })
    return {"graphs": graphs}

get_federated_coordinator_status

get_federated_coordinator_status()

Report coordinator activity in each configured federated repo.

Federation otherwise only reads committed graph data; this reaches one step further and reads each peer's gitignored .angelo/coordinator-state.json sidecar — a sibling of the .memory dir whose path federation already resolves — so the dashboard can tell when a coordinator is running in another repo. Read-only and best-effort: an unreachable or malformed peer yields runs: [] rather than an error.

Per-task detail is included so the dashboard can expand a federated run's graph inline, but "View in tree" stays disabled for these — a federated run targets entries in the peer's tree, which can't be opened in the local overlay. Liveness reuses the same heartbeat staleness rule as the local endpoint.

Source code in memory/dashboard/backend/routes/coordinator.py

@router.get("/coordinator-status/federated")
def get_federated_coordinator_status():
    """Report coordinator activity in each configured federated repo.

    Federation otherwise only reads committed graph data; this reaches one step
    further and reads each peer's gitignored ``.angelo/coordinator-state.json``
    sidecar — a sibling of the ``.memory`` dir whose path federation already
    resolves — so the dashboard can tell when a coordinator is running *in
    another repo*. Read-only and best-effort: an unreachable or malformed peer
    yields ``runs: []`` rather than an error.

    Per-task detail *is* included so the dashboard can expand a federated run's
    graph inline, but "View in tree" stays disabled for these — a federated run
    targets entries in the peer's tree, which can't be opened in the local
    overlay. Liveness reuses the same heartbeat staleness rule as the local
    endpoint.
    """
    repos, errors = federation.configured_repos()
    now = datetime.now(timezone.utc)
    out_repos: list[dict] = []
    for repo in repos:
        runs_out: list[dict] = []
        for run in federation.read_coordinator_runs(repo):
            summary = run.get("summary", {})
            total = summary.get("total", 0)
            if total == 0 and not run.get("tasks"):
                continue
            stale, age_seconds, heartbeat_at = _coordinator_stale_info(run, summary, now)
            runs_out.append({
                "run_id": run.get("run_id", ""),
                "active": total > 0 and not summary.get("all_complete", True) and not stale,
                "stale": stale,
                "goal": run.get("goal", ""),
                "target_entry_id": run.get("target_entry_id", ""),
                "summary": {**summary, "stale": stale},
                "created_at": run.get("created_at", ""),
                "updated_at": run.get("updated_at", ""),
                "heartbeat_at": heartbeat_at,
                "completed_at": run.get("completed_at", ""),
                "stale_after_seconds": COORDINATOR_STALE_AFTER_SECONDS,
                "heartbeat_age_seconds": age_seconds,
                # Per-task detail so the dashboard can expand a federated run's
                # graph just like a local one. "View in tree" stays disabled for
                # these (the target entry lives in the peer's tree), but the
                # self-contained task graph renders fine from this list alone.
                "tasks": run.get("tasks", []),
            })
        out_repos.append({
            "repo_id": repo.id,
            "repo_name": repo.name,
            "repo_path": str(repo.path),
            "active": any(run["active"] for run in runs_out),
            "active_run_count": sum(1 for run in runs_out if run["active"]),
            "runs": runs_out,
        })
    return {"repos": out_repos, "errors": errors}

update_coordinator_run

update_coordinator_run(run_id: str, updates: CoordinatorRunUpdate)

Rename a stale or completed coordinator run's goal (its card title).

Refuses active runs (a live coordinator owns and would overwrite them).

Source code in memory/dashboard/backend/routes/coordinator.py

@router.patch("/coordinator-status/{run_id}")
def update_coordinator_run(run_id: str, updates: CoordinatorRunUpdate):
    """Rename a stale or completed coordinator run's goal (its card title).

    Refuses active runs (a live coordinator owns and would overwrite them).
    """
    run = _find_local_coordinator_run(run_id)
    if run is None:
        raise HTTPException(status_code=404, detail=f"Coordinator run not found: {run_id}")
    editable, status = _coordinator_run_editable(run)
    if not editable:
        raise HTTPException(
            status_code=409,
            detail="Run is active and owned by a live coordinator; edits would be reverted.",
        )
    if updates.goal is None:
        raise HTTPException(status_code=400, detail="No updates provided")
    goal = updates.goal.strip()
    if not _rewrite_coordinator_run(run_id, goal=goal):
        raise HTTPException(status_code=404, detail=f"Coordinator run not found: {run_id}")
    return {"run_id": run_id, "goal": goal, "status": status}

delete_coordinator_run

delete_coordinator_run(run_id: str)

Delete a stale or completed coordinator run from the state file.

Refuses active runs (a live coordinator owns them and would re-add them).

Source code in memory/dashboard/backend/routes/coordinator.py

@router.delete("/coordinator-status/{run_id}")
def delete_coordinator_run(run_id: str):
    """Delete a stale or completed coordinator run from the state file.

    Refuses active runs (a live coordinator owns them and would re-add them).
    """
    run = _find_local_coordinator_run(run_id)
    if run is None:
        raise HTTPException(status_code=404, detail=f"Coordinator run not found: {run_id}")
    editable, status = _coordinator_run_editable(run)
    if not editable:
        raise HTTPException(
            status_code=409,
            detail="Run is active and owned by a live coordinator; it cannot be deleted from the dashboard.",
        )
    if not _rewrite_coordinator_run(run_id, delete=True):
        raise HTTPException(status_code=404, detail=f"Coordinator run not found: {run_id}")
    return {"deleted": run_id, "status": status}

finalize_coordinator_run

finalize_coordinator_run(run_id: str)

Mark a stale coordinator run as complete instead of deleting it.

Cancels any task that never finished (pending/running -> cancelled), recomputes the summary so the run reads as terminally complete, and moves it into the completed history -- preserving its already-done work and results. Refuses active runs (a live coordinator owns them and would revert the change). Already-completed runs are a no-op success.

Source code in memory/dashboard/backend/routes/coordinator.py

@router.post("/coordinator-status/{run_id}/finalize")
def finalize_coordinator_run(run_id: str):
    """Mark a stale coordinator run as complete instead of deleting it.

    Cancels any task that never finished (pending/running -> cancelled),
    recomputes the summary so the run reads as terminally complete, and moves
    it into the completed history -- preserving its already-done work and
    results. Refuses active runs (a live coordinator owns them and would revert
    the change). Already-completed runs are a no-op success.
    """
    run = _find_local_coordinator_run(run_id)
    if run is None:
        raise HTTPException(status_code=404, detail=f"Coordinator run not found: {run_id}")
    editable, status = _coordinator_run_editable(run)
    if not editable:
        raise HTTPException(
            status_code=409,
            detail="Run is active and owned by a live coordinator; it cannot be finalized from the dashboard.",
        )
    if status == "completed":
        return {"finalized": run_id, "status": "completed", "already_complete": True}
    if not _rewrite_coordinator_run(run_id, finalize=True):
        raise HTTPException(status_code=404, detail=f"Coordinator run not found: {run_id}")
    return {"finalized": run_id, "status": "completed"}

delete_federated_coordinator_run

delete_federated_coordinator_run(repo_id: str, run_id: str)

Delete a stale or completed coordinator run from a federated peer repo.

Cleanup only: refuses active runs (a live coordinator in the peer repo owns them and would re-add them on its next heartbeat).

Source code in memory/dashboard/backend/routes/coordinator.py

@router.delete("/coordinator-status/federated/{repo_id}/{run_id}")
def delete_federated_coordinator_run(repo_id: str, run_id: str):
    """Delete a stale or completed coordinator run from a federated peer repo.

    Cleanup only: refuses active runs (a live coordinator in the peer repo owns
    them and would re-add them on its next heartbeat).
    """
    repo = _find_federated_repo(repo_id)
    if repo is None:
        raise HTTPException(status_code=404, detail=f"Federated repo not found: {repo_id}")
    run = _find_federated_coordinator_run(repo, run_id)
    if run is None:
        raise HTTPException(status_code=404, detail=f"Coordinator run not found: {run_id}")
    editable, status = _coordinator_run_editable(run)
    if not editable:
        raise HTTPException(
            status_code=409,
            detail="Run is active and owned by a live coordinator in the federated repo; it cannot be deleted.",
        )
    if not _rewrite_federated_coordinator_run(repo, run_id, delete=True):
        raise HTTPException(status_code=404, detail=f"Coordinator run not found: {run_id}")
    return {"deleted": run_id, "repo_id": repo_id, "status": status}

finalize_federated_coordinator_run

finalize_federated_coordinator_run(repo_id: str, run_id: str)

Mark a stale coordinator run in a federated peer repo as complete.

Cancels any unrun task and archives the run as complete — the federated analogue of the local finalize endpoint. Refuses active runs; an already-completed run is a no-op success.

Source code in memory/dashboard/backend/routes/coordinator.py

@router.post("/coordinator-status/federated/{repo_id}/{run_id}/finalize")
def finalize_federated_coordinator_run(repo_id: str, run_id: str):
    """Mark a stale coordinator run in a federated peer repo as complete.

    Cancels any unrun task and archives the run as complete — the federated
    analogue of the local finalize endpoint. Refuses active runs; an
    already-completed run is a no-op success.
    """
    repo = _find_federated_repo(repo_id)
    if repo is None:
        raise HTTPException(status_code=404, detail=f"Federated repo not found: {repo_id}")
    run = _find_federated_coordinator_run(repo, run_id)
    if run is None:
        raise HTTPException(status_code=404, detail=f"Coordinator run not found: {run_id}")
    editable, status = _coordinator_run_editable(run)
    if not editable:
        raise HTTPException(
            status_code=409,
            detail="Run is active and owned by a live coordinator in the federated repo; it cannot be finalized.",
        )
    if status == "completed":
        return {"finalized": run_id, "repo_id": repo_id, "status": "completed", "already_complete": True}
    if not _rewrite_federated_coordinator_run(repo, run_id, finalize=True):
        raise HTTPException(status_code=404, detail=f"Coordinator run not found: {run_id}")
    return {"finalized": run_id, "repo_id": repo_id, "status": "completed"}