coordinator.graph

coordinator.graph

Task graph for multi-agent coordination.

In-memory directed acyclic graph where nodes are tasks and edges are dependencies. The graph drives wave-based execution: tasks become ready only when all their predecessors are complete. Multiple tasks in the same wave run in parallel.

SoftCapReached

Bases: Exception

Raised by :meth:TaskGraph.extend_graph when the soft extension cap is hit.

The soft cap is an escalation gate, NOT a hard wall: the caller must either escalate to the user (a higher cap / a fresh graph) or consciously override by re-calling with force=True and a reason (recorded and surfaced). A separate, larger hard ceiling still raises ValueError to stop runaway loops even when forced. Carries the counts so the caller can build a structured response without re-reading graph internals.

Source code in coordinator/graph.py

class SoftCapReached(Exception):
    """Raised by :meth:`TaskGraph.extend_graph` when the soft extension cap is hit.

    The soft cap is an escalation gate, NOT a hard wall: the caller must either
    escalate to the user (a higher cap / a fresh graph) or consciously override
    by re-calling with ``force=True`` and a ``reason`` (recorded and surfaced).
    A separate, larger hard ceiling still raises ``ValueError`` to stop runaway
    loops even when forced. Carries the counts so the caller can build a
    structured response without re-reading graph internals.
    """

    def __init__(self, extensions_used: int, max_extensions: int, hard_max: int) -> None:
        self.extensions_used = extensions_used
        self.max_extensions = max_extensions
        self.hard_max = hard_max
        super().__init__(
            f"Soft extension cap ({max_extensions}) reached after "
            f"{extensions_used} extensions. Escalate to the user, or re-call "
            f"extend_graph(force=True, reason=...) to continue (hard ceiling: "
            f"{hard_max})."
        )

TaskGraph

In-memory task graph backed by a plain dict.

Source code in coordinator/graph.py

class TaskGraph:
    """In-memory task graph backed by a plain dict."""

    def __init__(
        self,
        max_extensions: int = 3,
        agent_roles: dict[str, str] | None = None,
        hard_max_extensions: int | None = None,
    ) -> None:
        self._tasks: dict[str, Task] = {}
        self._alias_to_id: dict[str, str] = {}
        self._extensions_count: int = 0
        self._max_extensions: int = max_extensions
        # Forced extensions are those granted PAST the soft cap via
        # ``extend_graph(force=True)``. Tracked separately so the override is
        # always loud (surfaced in get_report + the dashboard summary), never a
        # silent bypass of the escalation gate.
        self._forced_extensions: int = 0
        # Hard ceiling: a real runaway wall that even force cannot pass. Derived
        # from the soft cap (generous headroom) unless explicitly overridden.
        self._hard_max_extensions: int = (
            hard_max_extensions
            if (hard_max_extensions is not None and hard_max_extensions >= max_extensions)
            else max(max_extensions * 3, max_extensions + 2, 3)
        )
        self._agent_roles: dict[str, str] = dict(agent_roles or {})

    @property
    def tasks(self) -> dict[str, Task]:
        return self._tasks

    def create_graph(
        self,
        tasks: list[dict],
        goal: str = "",
        agent_models: dict[str, str] | None = None,
        agent_passes: dict[str, int] | None = None,
        agent_schemas: dict[str, str] | None = None,
        schema_resolver: Callable[[str], object | None] | None = None,
        schema_enabled: bool = True,
    ) -> list[str]:
        """Create a task graph from a list of task definitions.

        Each task dict has:
            - alias: str -- local reference name (e.g. "eng", "rev")
            - agent: str -- agent type (e.g. "engineer", "reviewer")
            - description: str -- what the agent should do
            - depends_on: list[str] -- aliases this task depends on (empty = root)
            - model: str (optional) -- model override for this specific task
            - passes: int (optional) -- self-refinement pass override for this task
            - schema: str (optional) -- extraction-schema name for this task
            - writes: list[str] -- declared write scope for path-scoped
              coordination. The coordinator protocol requires this to be filled
              explicitly for implementer tasks: pass disjoint paths to run
              implementers in parallel, an empty list for dynamic mode (the
              agent leases paths at write time via manage_paths action acquire/release),
              or ``["."]`` for a deliberate whole-workspace lease. Omitting it
              falls back to a whole-workspace lease as a safety backstop. See
              ``lease_keys``.

        Args:
            agent_models: Default model per agent type (from agents.yaml).
                Task-level ``model`` overrides these defaults.
            agent_passes: Default self-refinement pass count per agent type
                (from agents.yaml). Task-level ``passes`` overrides these.
            agent_schemas: Default extraction-schema name per agent type. Task
                level ``schema`` overrides these defaults.
            schema_resolver: Callable ``name -> expanded object | None`` used to
                expand a resolved schema name (e.g. ``contrib.expand_schema``).
            schema_enabled: Whether the grounded-extraction capability is
                enabled for this project. When False, a task carrying a schema
                raises (spec §8.3).

        Returns:
            List of generated task IDs.
        """
        if not tasks:
            raise ValueError("Graph must have at least one task.")

        # create_graph builds a fresh graph: drop any tasks/aliases from a prior
        # run so a reused TaskGraph instance cannot contaminate the new run with
        # stale tasks (which would leak into get_ready_tasks, get_report, etc.).
        self._tasks.clear()
        self._alias_to_id.clear()
        self._extensions_count = 0
        self._forced_extensions = 0
        defaults = agent_models or {}
        pass_defaults = agent_passes or {}
        schema_defaults = agent_schemas or {}

        violations = validate_graph(tasks, agent_roles=self._agent_roles)
        if violations:
            raise ValueError(
                f"Graph validation failed: {'; '.join(violations)}"
            )

        alias_map: dict[str, str] = {}
        task_ids: list[str] = []

        for t in tasks:
            task_id = f"task-{uuid.uuid4().hex[:8]}"
            alias = t.get("alias", task_id)
            alias_map[alias] = task_id

        for t in tasks:
            alias = t.get("alias", "")
            task_id = alias_map[alias]
            depends_on = [alias_map[dep] for dep in t.get("depends_on", [])]
            model = t.get("model") or defaults.get(t["agent"])
            schema_name, schema_obj = _resolve_schema(
                t, schema_defaults, schema_resolver, schema_enabled
            )

            task = Task(
                id=task_id,
                description=t["description"],
                agent=t["agent"],
                alias=alias,
                model=model,
                schema=schema_name,
                schema_obj=schema_obj,
                passes_total=_resolve_passes(t, pass_defaults),
                depends_on=depends_on,
                writes=t.get("writes"),
            )
            self._tasks[task_id] = task
            self._alias_to_id[alias] = task_id
            task_ids.append(task_id)

        return task_ids

    def create_pipeline(
        self,
        steps: list[tuple[str, str] | tuple[str, str, str | None]],
        goal: str = "",
        agent_models: dict[str, str] | None = None,
        agent_passes: dict[str, int] | None = None,
    ) -> list[str]:
        """Create a linear pipeline (convenience wrapper around create_graph).

        Args:
            steps: Ordered list of (agent, description) or
                (agent, description, model) tuples.
            agent_models: Default model per agent type (from agents.yaml).

        Returns:
            List of task IDs in execution order.
        """
        tasks = []
        for i, step in enumerate(steps):
            agent, description = step[0], step[1]
            model = step[2] if len(step) > 2 else None
            alias = f"step_{i}"
            depends_on = [f"step_{i - 1}"] if i > 0 else []
            task: dict = {
                "alias": alias,
                "agent": agent,
                "description": description,
                "depends_on": depends_on,
            }
            if model:
                task["model"] = model
            tasks.append(task)
        return self.create_graph(
            tasks, goal=goal, agent_models=agent_models, agent_passes=agent_passes
        )

    def extend_graph(
        self,
        tasks: list[dict],
        after_task_ids: list[str],
        context: str = "",
        agent_models: dict[str, str] | None = None,
        agent_passes: dict[str, int] | None = None,
        agent_schemas: dict[str, str] | None = None,
        schema_resolver: Callable[[str], object | None] | None = None,
        schema_enabled: bool = True,
        force: bool = False,
        reason: str = "",
    ) -> list[str]:
        """Append new tasks to the graph after one or more existing tasks.

        Used when tasks fail and the coordinator decides to add corrective
        steps. The graph only grows forward.

        Extension cap is tiered (soft cap + hard ceiling):
            * below the soft cap -> proceed normally;
            * at/above the soft cap and not ``force`` -> raise
              :class:`SoftCapReached` (an escalation gate, not a wall);
            * forced and below the hard ceiling -> proceed and count it as a
              FORCED extension (recorded so the override is visible);
            * at/above the hard ceiling -> raise ``ValueError`` (runaway wall),
              even when forced.

        Args:
            tasks: List of task dicts (same format as create_graph).
            after_task_ids: Task IDs that the first new task depends on.
                            Typically the failed tasks from a wave.
            context: Failure context to inject into the first new task's
                     description.
            agent_models: Default model per agent type (from agents.yaml).
            force: Override the soft cap (still bounded by the hard ceiling).
            reason: Why the override is justified (logged; the caller records it
                    on the run for dashboard visibility).

        Returns:
            List of new task IDs.

        Raises:
            SoftCapReached: Soft cap hit and not forced.
            ValueError: Hard ceiling reached, or extension validation failed.
        """
        for tid in after_task_ids:
            self._get_task(tid)

        # Hard ceiling is a true wall -- force cannot pass it.
        if self._extensions_count >= self._hard_max_extensions:
            raise ValueError(
                f"Hard extension ceiling ({self._hard_max_extensions}) reached "
                f"after {self._extensions_count} extensions. Stop and escalate "
                f"to the user -- a fresh graph or a human decision is required."
            )
        # Soft cap is an escalation gate: proceed only when explicitly forced.
        forcing_past_soft_cap = self._extensions_count >= self._max_extensions
        if forcing_past_soft_cap and not force:
            raise SoftCapReached(
                self._extensions_count, self._max_extensions, self._hard_max_extensions
            )

        existing_ids = set(self._tasks.keys()) | set(self._alias_to_id.keys())
        violations = validate_graph(
            tasks, existing_task_ids=existing_ids, agent_roles=self._agent_roles
        )
        if violations:
            raise ValueError(
                f"Extension validation failed: {'; '.join(violations)}"
            )

        self._extensions_count += 1
        if forcing_past_soft_cap:
            self._forced_extensions += 1
            logger.warning(
                "Forced extension #%d past soft cap %d (now %d/%d; reason: %s)",
                self._forced_extensions,
                self._max_extensions,
                self._extensions_count,
                self._hard_max_extensions,
                reason or "<none given>",
            )
        defaults = agent_models or {}
        pass_defaults = agent_passes or {}
        schema_defaults = agent_schemas or {}

        alias_map: dict[str, str] = {}
        for t in tasks:
            task_id = f"task-{uuid.uuid4().hex[:8]}"
            alias = t.get("alias", task_id)
            alias_map[alias] = task_id

        task_ids: list[str] = []
        for i, t in enumerate(tasks):
            alias = t.get("alias", "")
            task_id = alias_map[alias]
            description = t["description"]

            raw_deps = t.get("depends_on", [])
            depends_on = []
            for dep in raw_deps:
                if dep in alias_map:
                    depends_on.append(alias_map[dep])
                elif dep in self._alias_to_id:
                    depends_on.append(self._alias_to_id[dep])
                elif dep in self._tasks:
                    depends_on.append(dep)

            if i == 0 and not depends_on:
                depends_on = list(after_task_ids)

            if i == 0 and context:
                description = f"{description}\n\nFEEDBACK FROM PRIOR FAILURE:\n{context}"

            model = t.get("model") or defaults.get(t["agent"])
            schema_name, schema_obj = _resolve_schema(
                t, schema_defaults, schema_resolver, schema_enabled
            )

            task = Task(
                id=task_id,
                description=description,
                agent=t["agent"],
                alias=alias,
                model=model,
                schema=schema_name,
                schema_obj=schema_obj,
                passes_total=_resolve_passes(t, pass_defaults),
                depends_on=depends_on,
                writes=t.get("writes"),
            )
            self._tasks[task_id] = task
            self._alias_to_id[alias] = task_id
            task_ids.append(task_id)

        return task_ids

    def restore_tasks(
        self, entries: list[dict], extensions_used: int = 0,
        forced_extensions: int = 0,
    ) -> list[str]:
        """Rehydrate persisted tasks into the graph (for resuming a run).

        Used when adopting a run from the state file after the coordinator
        process that owned it died. Tasks keep their original IDs, aliases,
        dependencies, and terminal statuses. Tasks that were RUNNING when
        the owning process died are reset to PENDING so they can be re-run
        (their results were lost with the process).

        Args:
            entries: Persisted task dicts with keys task_id, alias, agent,
                description, depends_on (task IDs), status, and optionally
                model, result, parsed_result, started_at, completed_at.
            extensions_used: The extension count from the persisted run so a
                resumed run cannot exceed ``max_extensions``. Defaults to 0.

        Returns:
            List of restored task IDs.

        Raises:
            ValueError: A task ID already exists in the graph, or a
                dependency does not resolve.
        """
        known_ids = set(self._tasks.keys()) | {e["task_id"] for e in entries}
        for e in entries:
            if e["task_id"] in self._tasks:
                raise ValueError(f"Task {e['task_id']} already exists in the graph.")
            for dep in e.get("depends_on", []):
                if dep not in known_ids:
                    raise ValueError(
                        f"Task {e['task_id']}: depends_on '{dep}' does not resolve."
                    )

        def parse_time(value) -> datetime | None:
            if not value:
                return None
            try:
                return datetime.fromisoformat(str(value))
            except ValueError:
                return None

        task_ids: list[str] = []
        for e in entries:
            status = TaskStatus(e.get("status", "pending"))
            if status == TaskStatus.RUNNING:
                status = TaskStatus.PENDING
            task = Task(
                id=e["task_id"],
                description=e["description"],
                agent=e["agent"],
                alias=e.get("alias", ""),
                model=e.get("model"),
                schema=e.get("schema"),
                schema_obj=e.get("schema_obj"),
                status=status,
                depends_on=list(e.get("depends_on", [])),
                writes=e.get("writes"),
                result=e.get("result"),
                parsed_result=e.get("parsed_result"),
                passes_total=max(1, int(e.get("passes_total", 1) or 1)),
                passes_done=max(0, int(e.get("passes_done", 0) or 0)),
                pass_notes=list(e.get("pass_notes", []) or []),
                started_at=parse_time(e.get("started_at")),
                completed_at=parse_time(e.get("completed_at")),
            )
            self._tasks[task.id] = task
            if task.alias:
                self._alias_to_id[task.alias] = task.id
            task_ids.append(task.id)
        self._extensions_count = max(0, int(extensions_used))
        self._forced_extensions = max(0, int(forced_extensions or 0))
        return task_ids

    @property
    def extensions_count(self) -> int:
        return self._extensions_count

    @property
    def forced_extensions(self) -> int:
        return self._forced_extensions

    @property
    def hard_max_extensions(self) -> int:
        return self._hard_max_extensions

    def resolve_alias(self, alias: str) -> str | None:
        """Resolve an alias to a task ID, or None if not found."""
        return self._alias_to_id.get(alias)

    def get_ready_tasks(self) -> list[Task]:
        """Return tasks that are pending and have all dependencies satisfied.

        A dependency is satisfied if it is in a terminal state (DONE, FAILED,
        or CANCELLED) -- extensions chain after failed tasks, and a finalized
        run's cancelled tasks never block (though finalize cancels dependents
        too, so this is mostly belt-and-suspenders).
        """
        ready = []
        for task in self._tasks.values():
            if task.status != TaskStatus.PENDING:
                continue
            if all(
                self._tasks[dep].status in TERMINAL_STATUSES
                for dep in task.depends_on
            ):
                ready.append(task)
        return ready

    def claim_task(self, task_id: str) -> Task:
        """Mark a task as running.

        Raises:
            KeyError: Task does not exist.
            ValueError: Task is not in PENDING state.
        """
        task = self._get_task(task_id)
        if task.status != TaskStatus.PENDING:
            raise ValueError(
                f"Cannot claim task {task_id}: status is {task.status.value}, expected pending."
            )
        task.status = TaskStatus.RUNNING
        task.started_at = datetime.now(timezone.utc)
        return task

    def submit_result(
        self, task_id: str, result: str, success: bool = True
    ) -> Task:
        """Record the result of a task and mark it done or failed."""
        task = self._get_task(task_id)
        if task.status != TaskStatus.RUNNING:
            raise ValueError(
                f"Cannot submit result for task {task_id}: status is {task.status.value}, expected running."
            )
        task.result = result
        task.status = TaskStatus.DONE if success else TaskStatus.FAILED
        task.completed_at = datetime.now(timezone.utc)
        return task

    def finalize(self) -> list[str]:
        """Mark the graph complete by cancelling any non-terminal tasks.

        Converts every PENDING/RUNNING task to CANCELLED (stamping
        ``completed_at``) so the run reads as terminally complete. Used when a
        stranded run is "finalized" instead of resumed or trashed -- its
        already-finished work is preserved while the unrun tail is closed out.

        Returns:
            The IDs of the tasks that were cancelled (was pending/running).
        """
        cancelled: list[str] = []
        now = datetime.now(timezone.utc)
        for task in self._tasks.values():
            if task.status in (TaskStatus.PENDING, TaskStatus.RUNNING):
                task.status = TaskStatus.CANCELLED
                task.completed_at = now
                cancelled.append(task.id)
        return cancelled

    def record_pass(self, task_id: str, summary: str | None = None) -> Task:
        """Record completion of one self-refinement pass for a running task.

        Increments ``passes_done`` and appends an optional one-line ``summary``
        to ``pass_notes`` (for the dashboard tooltip). The task must be RUNNING
        -- passes happen between :meth:`claim_task` and :meth:`submit_result`.
        ``passes_done`` is allowed to reach ``passes_total``; if it would exceed
        it, ``passes_total`` is bumped up so the displayed fraction stays sane.

        Raises:
            KeyError: Task does not exist.
            ValueError: Task is not RUNNING.
        """
        task = self._get_task(task_id)
        if task.status != TaskStatus.RUNNING:
            raise ValueError(
                f"Cannot record pass for task {task_id}: status is "
                f"{task.status.value}, expected running."
            )
        task.passes_done += 1
        if task.passes_done > task.passes_total:
            task.passes_total = task.passes_done
        if summary:
            task.pass_notes.append(summary.strip())
        return task

    def get_task_context(self, task_id: str) -> list[dict]:
        """Return outputs from all predecessor tasks.

        Each entry is a dict with keys: task_id, agent, description, result,
        status. Includes predecessors that are DONE or FAILED (so fix agents
        see failures).
        """
        task = self._get_task(task_id)
        context = []
        for dep_id in task.depends_on:
            dep = self._tasks[dep_id]
            if dep.status in (TaskStatus.DONE, TaskStatus.FAILED) and dep.result is not None:
                context.append(
                    {
                        "task_id": dep.id,
                        "agent": dep.agent,
                        "description": dep.description,
                        "result": dep.result,
                        "status": dep.status.value,
                    }
                )
        return context

    def get_status(self) -> list[dict]:
        """Return current state of all tasks."""
        result = []
        for t in self._tasks.values():
            entry: dict = {
                "task_id": t.id,
                "description": t.description,
                "agent": t.agent,
                "alias": t.alias,
                "status": t.status.value,
                "depends_on": t.depends_on,
                "has_result": t.result is not None,
            }
            if t.model:
                entry["model"] = t.model
            if t.schema:
                entry["schema"] = t.schema
            if t.writes is not None:
                entry["writes"] = list(t.writes)
            result.append(entry)
        return result

    def get_report(self) -> dict:
        """Generate a summary report of the graph execution."""
        tasks = list(self._tasks.values())
        total = len(tasks)
        done = sum(1 for t in tasks if t.status == TaskStatus.DONE)
        failed = sum(1 for t in tasks if t.status == TaskStatus.FAILED)
        cancelled = sum(1 for t in tasks if t.status == TaskStatus.CANCELLED)
        pending = sum(1 for t in tasks if t.status == TaskStatus.PENDING)
        running = sum(1 for t in tasks if t.status == TaskStatus.RUNNING)

        timeline = []
        for t in tasks:
            entry = {
                "task_id": t.id,
                "alias": t.alias,
                "agent": t.agent,
                "description": t.description,
                "status": t.status.value,
                "depends_on": t.depends_on,
            }
            if t.model:
                entry["model"] = t.model
            if t.schema:
                entry["schema"] = t.schema
            if t.writes is not None:
                entry["writes"] = list(t.writes)
            if t.passes_total > 1:
                entry["passes_total"] = t.passes_total
                entry["passes_done"] = t.passes_done
                if t.pass_notes:
                    entry["pass_notes"] = list(t.pass_notes)
            if t.started_at:
                entry["started_at"] = t.started_at.isoformat()
            if t.completed_at:
                entry["completed_at"] = t.completed_at.isoformat()
            if t.result:
                entry["result_preview"] = t.result[:200]
            if t.parsed_result:
                entry["parsed_result"] = t.parsed_result
            timeline.append(entry)

        return {
            "summary": {
                "total": total,
                "done": done,
                "failed": failed,
                "cancelled": cancelled,
                "pending": pending,
                "running": running,
                "extensions_used": self._extensions_count,
                "max_extensions": self._max_extensions,
                "forced_extensions": self._forced_extensions,
                "hard_max_extensions": self._hard_max_extensions,
                "all_complete": done + failed + cancelled == total,
            },
            "timeline": timeline,
        }

    def _get_task(self, task_id: str) -> Task:
        """Retrieve a task by ID or raise KeyError."""
        if task_id not in self._tasks:
            raise KeyError(f"Task {task_id} not found.")
        return self._tasks[task_id]

    def to_state(self) -> dict:
        """Full-fidelity serialization for process-boundary persistence.

        Unlike :meth:`export_trace` (telemetry/debugging), this round-trips with
        :meth:`from_state` exactly — it preserves ``created_at`` and the live
        ``RUNNING`` status (no reset), so a CLI that persists between command
        invocations can claim a task in one call and submit it in the next.
        """
        return {
            "version": 1,
            "max_extensions": self._max_extensions,
            "extensions_used": self._extensions_count,
            "forced_extensions": self._forced_extensions,
            "hard_max_extensions": self._hard_max_extensions,
            "tasks": [
                {
                    "id": t.id,
                    "alias": t.alias,
                    "agent": t.agent,
                    "description": t.description,
                    "model": t.model,
                    "schema": t.schema,
                    "schema_obj": t.schema_obj,
                    "status": t.status.value,
                    "depends_on": list(t.depends_on),
                    "writes": t.writes,
                    "result": t.result,
                    "parsed_result": t.parsed_result,
                    "passes_total": t.passes_total,
                    "passes_done": t.passes_done,
                    "pass_notes": list(t.pass_notes),
                    "created_at": t.created_at.isoformat() if t.created_at else None,
                    "started_at": t.started_at.isoformat() if t.started_at else None,
                    "completed_at": t.completed_at.isoformat() if t.completed_at else None,
                }
                for t in self._tasks.values()
            ],
        }

    @classmethod
    def from_state(
        cls, data: dict, agent_roles: dict[str, str] | None = None
    ) -> "TaskGraph":
        """Reconstruct a graph from :meth:`to_state` output, preserving status.

        Args:
            data: A dict produced by :meth:`to_state`.
            agent_roles: Role overrides to attach to the rebuilt graph (e.g.
                project-local roles). Roles are not serialized because they are
                resolved from configuration, not persisted run state.
        """

        def parse_time(value) -> datetime | None:
            if not value:
                return None
            try:
                return datetime.fromisoformat(str(value))
            except (ValueError, TypeError):
                return None

        graph = cls(
            max_extensions=int(data.get("max_extensions", 3)),
            agent_roles=agent_roles,
            hard_max_extensions=(
                int(data["hard_max_extensions"])
                if data.get("hard_max_extensions") is not None
                else None
            ),
        )
        for td in data.get("tasks", []):
            task = Task(
                id=td["id"],
                description=td.get("description", ""),
                agent=td.get("agent", ""),
                alias=td.get("alias", ""),
                model=td.get("model"),
                schema=td.get("schema"),
                schema_obj=td.get("schema_obj"),
                status=TaskStatus(td.get("status", "pending")),
                depends_on=list(td.get("depends_on", [])),
                writes=td.get("writes"),
                result=td.get("result"),
                parsed_result=td.get("parsed_result"),
                passes_total=max(1, int(td.get("passes_total", 1) or 1)),
                passes_done=max(0, int(td.get("passes_done", 0) or 0)),
                pass_notes=list(td.get("pass_notes", []) or []),
                started_at=parse_time(td.get("started_at")),
                completed_at=parse_time(td.get("completed_at")),
            )
            created = parse_time(td.get("created_at"))
            if created is not None:
                task.created_at = created
            graph._tasks[task.id] = task
            if task.alias:
                graph._alias_to_id[task.alias] = task.id
        graph._extensions_count = max(0, int(data.get("extensions_used", 0)))
        graph._forced_extensions = max(0, int(data.get("forced_extensions", 0) or 0))
        return graph

    def export_trace(self) -> dict:
        """Serialize the full graph state for telemetry/debugging."""
        tasks = []
        for t in self._tasks.values():
            entry = {
                "task_id": t.id,
                "alias": t.alias,
                "agent": t.agent,
                "description": t.description,
                "status": t.status.value,
                "depends_on": t.depends_on,
                "result": t.result,
                "parsed_result": t.parsed_result,
            }
            if t.model:
                entry["model"] = t.model
            if t.schema:
                entry["schema"] = t.schema
            if t.writes is not None:
                entry["writes"] = list(t.writes)
            if t.passes_total > 1:
                entry["passes_total"] = t.passes_total
                entry["passes_done"] = t.passes_done
                if t.pass_notes:
                    entry["pass_notes"] = list(t.pass_notes)
            if t.started_at:
                entry["started_at"] = t.started_at.isoformat()
            if t.completed_at:
                entry["completed_at"] = t.completed_at.isoformat()
            if t.started_at and t.completed_at:
                entry["duration_seconds"] = (t.completed_at - t.started_at).total_seconds()
            tasks.append(entry)

        total_duration = 0.0
        started_times = [t.started_at for t in self._tasks.values() if t.started_at]
        completed_times = [t.completed_at for t in self._tasks.values() if t.completed_at]
        if started_times and completed_times:
            total_duration = (max(completed_times) - min(started_times)).total_seconds()

        return {
            "tasks": tasks,
            "extensions_used": self._extensions_count,
            "max_extensions": self._max_extensions,
            "total_tasks": len(self._tasks),
            "total_duration_seconds": total_duration,
        }

create_graph

create_graph(tasks: list[dict], goal: str = '', agent_models: dict[str, str] | None = None, agent_passes: dict[str, int] | None = None, agent_schemas: dict[str, str] | None = None, schema_resolver: Callable[[str], object | None] | None = None, schema_enabled: bool = True) -> list[str]

Create a task graph from a list of task definitions.

Each task dict has

• alias: str -- local reference name (e.g. "eng", "rev")
• agent: str -- agent type (e.g. "engineer", "reviewer")
• description: str -- what the agent should do
• depends_on: list[str] -- aliases this task depends on (empty = root)
• model: str (optional) -- model override for this specific task
• passes: int (optional) -- self-refinement pass override for this task
• schema: str (optional) -- extraction-schema name for this task
• writes: list[str] -- declared write scope for path-scoped coordination. The coordinator protocol requires this to be filled explicitly for implementer tasks: pass disjoint paths to run implementers in parallel, an empty list for dynamic mode (the agent leases paths at write time via manage_paths action acquire/release), or ["."] for a deliberate whole-workspace lease. Omitting it falls back to a whole-workspace lease as a safety backstop. See lease_keys.

Parameters:

Name	Type	Description	Default
agent_models	dict[str, str] | None	Default model per agent type (from agents.yaml). Task-level model overrides these defaults.	None
agent_passes	dict[str, int] | None	Default self-refinement pass count per agent type (from agents.yaml). Task-level passes overrides these.	None
agent_schemas	dict[str, str] | None	Default extraction-schema name per agent type. Task level schema overrides these defaults.	None
schema_resolver	Callable[[str], object | None] | None	Callable name -> expanded object | None used to expand a resolved schema name (e.g. contrib.expand_schema).	None
schema_enabled	bool	Whether the grounded-extraction capability is enabled for this project. When False, a task carrying a schema raises (spec §8.3).	True

Returns:

Type	Description
list[str]	List of generated task IDs.

Source code in coordinator/graph.py

def create_graph(
    self,
    tasks: list[dict],
    goal: str = "",
    agent_models: dict[str, str] | None = None,
    agent_passes: dict[str, int] | None = None,
    agent_schemas: dict[str, str] | None = None,
    schema_resolver: Callable[[str], object | None] | None = None,
    schema_enabled: bool = True,
) -> list[str]:
    """Create a task graph from a list of task definitions.

    Each task dict has:
        - alias: str -- local reference name (e.g. "eng", "rev")
        - agent: str -- agent type (e.g. "engineer", "reviewer")
        - description: str -- what the agent should do
        - depends_on: list[str] -- aliases this task depends on (empty = root)
        - model: str (optional) -- model override for this specific task
        - passes: int (optional) -- self-refinement pass override for this task
        - schema: str (optional) -- extraction-schema name for this task
        - writes: list[str] -- declared write scope for path-scoped
          coordination. The coordinator protocol requires this to be filled
          explicitly for implementer tasks: pass disjoint paths to run
          implementers in parallel, an empty list for dynamic mode (the
          agent leases paths at write time via manage_paths action acquire/release),
          or ``["."]`` for a deliberate whole-workspace lease. Omitting it
          falls back to a whole-workspace lease as a safety backstop. See
          ``lease_keys``.

    Args:
        agent_models: Default model per agent type (from agents.yaml).
            Task-level ``model`` overrides these defaults.
        agent_passes: Default self-refinement pass count per agent type
            (from agents.yaml). Task-level ``passes`` overrides these.
        agent_schemas: Default extraction-schema name per agent type. Task
            level ``schema`` overrides these defaults.
        schema_resolver: Callable ``name -> expanded object | None`` used to
            expand a resolved schema name (e.g. ``contrib.expand_schema``).
        schema_enabled: Whether the grounded-extraction capability is
            enabled for this project. When False, a task carrying a schema
            raises (spec §8.3).

    Returns:
        List of generated task IDs.
    """
    if not tasks:
        raise ValueError("Graph must have at least one task.")

    # create_graph builds a fresh graph: drop any tasks/aliases from a prior
    # run so a reused TaskGraph instance cannot contaminate the new run with
    # stale tasks (which would leak into get_ready_tasks, get_report, etc.).
    self._tasks.clear()
    self._alias_to_id.clear()
    self._extensions_count = 0
    self._forced_extensions = 0
    defaults = agent_models or {}
    pass_defaults = agent_passes or {}
    schema_defaults = agent_schemas or {}

    violations = validate_graph(tasks, agent_roles=self._agent_roles)
    if violations:
        raise ValueError(
            f"Graph validation failed: {'; '.join(violations)}"
        )

    alias_map: dict[str, str] = {}
    task_ids: list[str] = []

    for t in tasks:
        task_id = f"task-{uuid.uuid4().hex[:8]}"
        alias = t.get("alias", task_id)
        alias_map[alias] = task_id

    for t in tasks:
        alias = t.get("alias", "")
        task_id = alias_map[alias]
        depends_on = [alias_map[dep] for dep in t.get("depends_on", [])]
        model = t.get("model") or defaults.get(t["agent"])
        schema_name, schema_obj = _resolve_schema(
            t, schema_defaults, schema_resolver, schema_enabled
        )

        task = Task(
            id=task_id,
            description=t["description"],
            agent=t["agent"],
            alias=alias,
            model=model,
            schema=schema_name,
            schema_obj=schema_obj,
            passes_total=_resolve_passes(t, pass_defaults),
            depends_on=depends_on,
            writes=t.get("writes"),
        )
        self._tasks[task_id] = task
        self._alias_to_id[alias] = task_id
        task_ids.append(task_id)

    return task_ids

create_pipeline

create_pipeline(steps: list[tuple[str, str] | tuple[str, str, str | None]], goal: str = '', agent_models: dict[str, str] | None = None, agent_passes: dict[str, int] | None = None) -> list[str]

Create a linear pipeline (convenience wrapper around create_graph).

Parameters:

Name	Type	Description	Default
steps	list[tuple[str, str] | tuple[str, str, str | None]]	Ordered list of (agent, description) or (agent, description, model) tuples.	required
agent_models	dict[str, str] | None	Default model per agent type (from agents.yaml).	None

Returns:

Type	Description
list[str]	List of task IDs in execution order.

Source code in coordinator/graph.py

def create_pipeline(
    self,
    steps: list[tuple[str, str] | tuple[str, str, str | None]],
    goal: str = "",
    agent_models: dict[str, str] | None = None,
    agent_passes: dict[str, int] | None = None,
) -> list[str]:
    """Create a linear pipeline (convenience wrapper around create_graph).

    Args:
        steps: Ordered list of (agent, description) or
            (agent, description, model) tuples.
        agent_models: Default model per agent type (from agents.yaml).

    Returns:
        List of task IDs in execution order.
    """
    tasks = []
    for i, step in enumerate(steps):
        agent, description = step[0], step[1]
        model = step[2] if len(step) > 2 else None
        alias = f"step_{i}"
        depends_on = [f"step_{i - 1}"] if i > 0 else []
        task: dict = {
            "alias": alias,
            "agent": agent,
            "description": description,
            "depends_on": depends_on,
        }
        if model:
            task["model"] = model
        tasks.append(task)
    return self.create_graph(
        tasks, goal=goal, agent_models=agent_models, agent_passes=agent_passes
    )

extend_graph

extend_graph(tasks: list[dict], after_task_ids: list[str], context: str = '', agent_models: dict[str, str] | None = None, agent_passes: dict[str, int] | None = None, agent_schemas: dict[str, str] | None = None, schema_resolver: Callable[[str], object | None] | None = None, schema_enabled: bool = True, force: bool = False, reason: str = '') -> list[str]

Append new tasks to the graph after one or more existing tasks.

Used when tasks fail and the coordinator decides to add corrective steps. The graph only grows forward.

Extension cap is tiered (soft cap + hard ceiling): * below the soft cap -> proceed normally; * at/above the soft cap and not force -> raise :class:SoftCapReached (an escalation gate, not a wall); * forced and below the hard ceiling -> proceed and count it as a FORCED extension (recorded so the override is visible); * at/above the hard ceiling -> raise ValueError (runaway wall), even when forced.

Parameters:

Name	Type	Description	Default
tasks	list[dict]	List of task dicts (same format as create_graph).	required
after_task_ids	list[str]	Task IDs that the first new task depends on. Typically the failed tasks from a wave.	required
context	str	Failure context to inject into the first new task's description.	''
agent_models	dict[str, str] | None	Default model per agent type (from agents.yaml).	None
force	bool	Override the soft cap (still bounded by the hard ceiling).	False
reason	str	Why the override is justified (logged; the caller records it on the run for dashboard visibility).	''

Returns:

Type	Description
list[str]	List of new task IDs.

Raises:

Type	Description
SoftCapReached	Soft cap hit and not forced.
ValueError	Hard ceiling reached, or extension validation failed.

Source code in coordinator/graph.py

def extend_graph(
    self,
    tasks: list[dict],
    after_task_ids: list[str],
    context: str = "",
    agent_models: dict[str, str] | None = None,
    agent_passes: dict[str, int] | None = None,
    agent_schemas: dict[str, str] | None = None,
    schema_resolver: Callable[[str], object | None] | None = None,
    schema_enabled: bool = True,
    force: bool = False,
    reason: str = "",
) -> list[str]:
    """Append new tasks to the graph after one or more existing tasks.

    Used when tasks fail and the coordinator decides to add corrective
    steps. The graph only grows forward.

    Extension cap is tiered (soft cap + hard ceiling):
        * below the soft cap -> proceed normally;
        * at/above the soft cap and not ``force`` -> raise
          :class:`SoftCapReached` (an escalation gate, not a wall);
        * forced and below the hard ceiling -> proceed and count it as a
          FORCED extension (recorded so the override is visible);
        * at/above the hard ceiling -> raise ``ValueError`` (runaway wall),
          even when forced.

    Args:
        tasks: List of task dicts (same format as create_graph).
        after_task_ids: Task IDs that the first new task depends on.
                        Typically the failed tasks from a wave.
        context: Failure context to inject into the first new task's
                 description.
        agent_models: Default model per agent type (from agents.yaml).
        force: Override the soft cap (still bounded by the hard ceiling).
        reason: Why the override is justified (logged; the caller records it
                on the run for dashboard visibility).

    Returns:
        List of new task IDs.

    Raises:
        SoftCapReached: Soft cap hit and not forced.
        ValueError: Hard ceiling reached, or extension validation failed.
    """
    for tid in after_task_ids:
        self._get_task(tid)

    # Hard ceiling is a true wall -- force cannot pass it.
    if self._extensions_count >= self._hard_max_extensions:
        raise ValueError(
            f"Hard extension ceiling ({self._hard_max_extensions}) reached "
            f"after {self._extensions_count} extensions. Stop and escalate "
            f"to the user -- a fresh graph or a human decision is required."
        )
    # Soft cap is an escalation gate: proceed only when explicitly forced.
    forcing_past_soft_cap = self._extensions_count >= self._max_extensions
    if forcing_past_soft_cap and not force:
        raise SoftCapReached(
            self._extensions_count, self._max_extensions, self._hard_max_extensions
        )

    existing_ids = set(self._tasks.keys()) | set(self._alias_to_id.keys())
    violations = validate_graph(
        tasks, existing_task_ids=existing_ids, agent_roles=self._agent_roles
    )
    if violations:
        raise ValueError(
            f"Extension validation failed: {'; '.join(violations)}"
        )

    self._extensions_count += 1
    if forcing_past_soft_cap:
        self._forced_extensions += 1
        logger.warning(
            "Forced extension #%d past soft cap %d (now %d/%d; reason: %s)",
            self._forced_extensions,
            self._max_extensions,
            self._extensions_count,
            self._hard_max_extensions,
            reason or "<none given>",
        )
    defaults = agent_models or {}
    pass_defaults = agent_passes or {}
    schema_defaults = agent_schemas or {}

    alias_map: dict[str, str] = {}
    for t in tasks:
        task_id = f"task-{uuid.uuid4().hex[:8]}"
        alias = t.get("alias", task_id)
        alias_map[alias] = task_id

    task_ids: list[str] = []
    for i, t in enumerate(tasks):
        alias = t.get("alias", "")
        task_id = alias_map[alias]
        description = t["description"]

        raw_deps = t.get("depends_on", [])
        depends_on = []
        for dep in raw_deps:
            if dep in alias_map:
                depends_on.append(alias_map[dep])
            elif dep in self._alias_to_id:
                depends_on.append(self._alias_to_id[dep])
            elif dep in self._tasks:
                depends_on.append(dep)

        if i == 0 and not depends_on:
            depends_on = list(after_task_ids)

        if i == 0 and context:
            description = f"{description}\n\nFEEDBACK FROM PRIOR FAILURE:\n{context}"

        model = t.get("model") or defaults.get(t["agent"])
        schema_name, schema_obj = _resolve_schema(
            t, schema_defaults, schema_resolver, schema_enabled
        )

        task = Task(
            id=task_id,
            description=description,
            agent=t["agent"],
            alias=alias,
            model=model,
            schema=schema_name,
            schema_obj=schema_obj,
            passes_total=_resolve_passes(t, pass_defaults),
            depends_on=depends_on,
            writes=t.get("writes"),
        )
        self._tasks[task_id] = task
        self._alias_to_id[alias] = task_id
        task_ids.append(task_id)

    return task_ids

restore_tasks

restore_tasks(entries: list[dict], extensions_used: int = 0, forced_extensions: int = 0) -> list[str]

Rehydrate persisted tasks into the graph (for resuming a run).

Used when adopting a run from the state file after the coordinator process that owned it died. Tasks keep their original IDs, aliases, dependencies, and terminal statuses. Tasks that were RUNNING when the owning process died are reset to PENDING so they can be re-run (their results were lost with the process).

Parameters:

Name	Type	Description	Default
entries	list[dict]	Persisted task dicts with keys task_id, alias, agent, description, depends_on (task IDs), status, and optionally model, result, parsed_result, started_at, completed_at.	required
extensions_used	int	The extension count from the persisted run so a resumed run cannot exceed max_extensions. Defaults to 0.	0

Returns:

Type	Description
list[str]	List of restored task IDs.

Raises:

Type	Description
ValueError	A task ID already exists in the graph, or a dependency does not resolve.

Source code in coordinator/graph.py

def restore_tasks(
    self, entries: list[dict], extensions_used: int = 0,
    forced_extensions: int = 0,
) -> list[str]:
    """Rehydrate persisted tasks into the graph (for resuming a run).

    Used when adopting a run from the state file after the coordinator
    process that owned it died. Tasks keep their original IDs, aliases,
    dependencies, and terminal statuses. Tasks that were RUNNING when
    the owning process died are reset to PENDING so they can be re-run
    (their results were lost with the process).

    Args:
        entries: Persisted task dicts with keys task_id, alias, agent,
            description, depends_on (task IDs), status, and optionally
            model, result, parsed_result, started_at, completed_at.
        extensions_used: The extension count from the persisted run so a
            resumed run cannot exceed ``max_extensions``. Defaults to 0.

    Returns:
        List of restored task IDs.

    Raises:
        ValueError: A task ID already exists in the graph, or a
            dependency does not resolve.
    """
    known_ids = set(self._tasks.keys()) | {e["task_id"] for e in entries}
    for e in entries:
        if e["task_id"] in self._tasks:
            raise ValueError(f"Task {e['task_id']} already exists in the graph.")
        for dep in e.get("depends_on", []):
            if dep not in known_ids:
                raise ValueError(
                    f"Task {e['task_id']}: depends_on '{dep}' does not resolve."
                )

    def parse_time(value) -> datetime | None:
        if not value:
            return None
        try:
            return datetime.fromisoformat(str(value))
        except ValueError:
            return None

    task_ids: list[str] = []
    for e in entries:
        status = TaskStatus(e.get("status", "pending"))
        if status == TaskStatus.RUNNING:
            status = TaskStatus.PENDING
        task = Task(
            id=e["task_id"],
            description=e["description"],
            agent=e["agent"],
            alias=e.get("alias", ""),
            model=e.get("model"),
            schema=e.get("schema"),
            schema_obj=e.get("schema_obj"),
            status=status,
            depends_on=list(e.get("depends_on", [])),
            writes=e.get("writes"),
            result=e.get("result"),
            parsed_result=e.get("parsed_result"),
            passes_total=max(1, int(e.get("passes_total", 1) or 1)),
            passes_done=max(0, int(e.get("passes_done", 0) or 0)),
            pass_notes=list(e.get("pass_notes", []) or []),
            started_at=parse_time(e.get("started_at")),
            completed_at=parse_time(e.get("completed_at")),
        )
        self._tasks[task.id] = task
        if task.alias:
            self._alias_to_id[task.alias] = task.id
        task_ids.append(task.id)
    self._extensions_count = max(0, int(extensions_used))
    self._forced_extensions = max(0, int(forced_extensions or 0))
    return task_ids

resolve_alias

resolve_alias(alias: str) -> str | None

Resolve an alias to a task ID, or None if not found.

Source code in coordinator/graph.py

def resolve_alias(self, alias: str) -> str | None:
    """Resolve an alias to a task ID, or None if not found."""
    return self._alias_to_id.get(alias)

get_ready_tasks

get_ready_tasks() -> list[Task]

Return tasks that are pending and have all dependencies satisfied.

A dependency is satisfied if it is in a terminal state (DONE, FAILED, or CANCELLED) -- extensions chain after failed tasks, and a finalized run's cancelled tasks never block (though finalize cancels dependents too, so this is mostly belt-and-suspenders).

Source code in coordinator/graph.py

def get_ready_tasks(self) -> list[Task]:
    """Return tasks that are pending and have all dependencies satisfied.

    A dependency is satisfied if it is in a terminal state (DONE, FAILED,
    or CANCELLED) -- extensions chain after failed tasks, and a finalized
    run's cancelled tasks never block (though finalize cancels dependents
    too, so this is mostly belt-and-suspenders).
    """
    ready = []
    for task in self._tasks.values():
        if task.status != TaskStatus.PENDING:
            continue
        if all(
            self._tasks[dep].status in TERMINAL_STATUSES
            for dep in task.depends_on
        ):
            ready.append(task)
    return ready

claim_task

claim_task(task_id: str) -> Task

Mark a task as running.

Raises:

Type	Description
KeyError	Task does not exist.
ValueError	Task is not in PENDING state.

Source code in coordinator/graph.py

def claim_task(self, task_id: str) -> Task:
    """Mark a task as running.

    Raises:
        KeyError: Task does not exist.
        ValueError: Task is not in PENDING state.
    """
    task = self._get_task(task_id)
    if task.status != TaskStatus.PENDING:
        raise ValueError(
            f"Cannot claim task {task_id}: status is {task.status.value}, expected pending."
        )
    task.status = TaskStatus.RUNNING
    task.started_at = datetime.now(timezone.utc)
    return task

submit_result

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

Record the result of a task and mark it done or failed.

Source code in coordinator/graph.py

def submit_result(
    self, task_id: str, result: str, success: bool = True
) -> Task:
    """Record the result of a task and mark it done or failed."""
    task = self._get_task(task_id)
    if task.status != TaskStatus.RUNNING:
        raise ValueError(
            f"Cannot submit result for task {task_id}: status is {task.status.value}, expected running."
        )
    task.result = result
    task.status = TaskStatus.DONE if success else TaskStatus.FAILED
    task.completed_at = datetime.now(timezone.utc)
    return task

finalize

finalize() -> list[str]

Mark the graph complete by cancelling any non-terminal tasks.

Converts every PENDING/RUNNING task to CANCELLED (stamping completed_at) so the run reads as terminally complete. Used when a stranded run is "finalized" instead of resumed or trashed -- its already-finished work is preserved while the unrun tail is closed out.

Returns:

Type	Description
list[str]	The IDs of the tasks that were cancelled (was pending/running).

Source code in coordinator/graph.py

def finalize(self) -> list[str]:
    """Mark the graph complete by cancelling any non-terminal tasks.

    Converts every PENDING/RUNNING task to CANCELLED (stamping
    ``completed_at``) so the run reads as terminally complete. Used when a
    stranded run is "finalized" instead of resumed or trashed -- its
    already-finished work is preserved while the unrun tail is closed out.

    Returns:
        The IDs of the tasks that were cancelled (was pending/running).
    """
    cancelled: list[str] = []
    now = datetime.now(timezone.utc)
    for task in self._tasks.values():
        if task.status in (TaskStatus.PENDING, TaskStatus.RUNNING):
            task.status = TaskStatus.CANCELLED
            task.completed_at = now
            cancelled.append(task.id)
    return cancelled

record_pass

record_pass(task_id: str, summary: str | None = None) -> Task

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

Increments passes_done and appends an optional one-line summary to pass_notes (for the dashboard tooltip). The task must be RUNNING -- passes happen between :meth:claim_task and :meth:submit_result. passes_done is allowed to reach passes_total; if it would exceed it, passes_total is bumped up so the displayed fraction stays sane.

Raises:

Type	Description
KeyError	Task does not exist.
ValueError	Task is not RUNNING.

Source code in coordinator/graph.py

def record_pass(self, task_id: str, summary: str | None = None) -> Task:
    """Record completion of one self-refinement pass for a running task.

    Increments ``passes_done`` and appends an optional one-line ``summary``
    to ``pass_notes`` (for the dashboard tooltip). The task must be RUNNING
    -- passes happen between :meth:`claim_task` and :meth:`submit_result`.
    ``passes_done`` is allowed to reach ``passes_total``; if it would exceed
    it, ``passes_total`` is bumped up so the displayed fraction stays sane.

    Raises:
        KeyError: Task does not exist.
        ValueError: Task is not RUNNING.
    """
    task = self._get_task(task_id)
    if task.status != TaskStatus.RUNNING:
        raise ValueError(
            f"Cannot record pass for task {task_id}: status is "
            f"{task.status.value}, expected running."
        )
    task.passes_done += 1
    if task.passes_done > task.passes_total:
        task.passes_total = task.passes_done
    if summary:
        task.pass_notes.append(summary.strip())
    return task

get_task_context

get_task_context(task_id: str) -> list[dict]

Return outputs from all predecessor tasks.

Each entry is a dict with keys: task_id, agent, description, result, status. Includes predecessors that are DONE or FAILED (so fix agents see failures).

Source code in coordinator/graph.py

def get_task_context(self, task_id: str) -> list[dict]:
    """Return outputs from all predecessor tasks.

    Each entry is a dict with keys: task_id, agent, description, result,
    status. Includes predecessors that are DONE or FAILED (so fix agents
    see failures).
    """
    task = self._get_task(task_id)
    context = []
    for dep_id in task.depends_on:
        dep = self._tasks[dep_id]
        if dep.status in (TaskStatus.DONE, TaskStatus.FAILED) and dep.result is not None:
            context.append(
                {
                    "task_id": dep.id,
                    "agent": dep.agent,
                    "description": dep.description,
                    "result": dep.result,
                    "status": dep.status.value,
                }
            )
    return context

get_status

get_status() -> list[dict]

Return current state of all tasks.

Source code in coordinator/graph.py

def get_status(self) -> list[dict]:
    """Return current state of all tasks."""
    result = []
    for t in self._tasks.values():
        entry: dict = {
            "task_id": t.id,
            "description": t.description,
            "agent": t.agent,
            "alias": t.alias,
            "status": t.status.value,
            "depends_on": t.depends_on,
            "has_result": t.result is not None,
        }
        if t.model:
            entry["model"] = t.model
        if t.schema:
            entry["schema"] = t.schema
        if t.writes is not None:
            entry["writes"] = list(t.writes)
        result.append(entry)
    return result

get_report

get_report() -> dict

Generate a summary report of the graph execution.

Source code in coordinator/graph.py

def get_report(self) -> dict:
    """Generate a summary report of the graph execution."""
    tasks = list(self._tasks.values())
    total = len(tasks)
    done = sum(1 for t in tasks if t.status == TaskStatus.DONE)
    failed = sum(1 for t in tasks if t.status == TaskStatus.FAILED)
    cancelled = sum(1 for t in tasks if t.status == TaskStatus.CANCELLED)
    pending = sum(1 for t in tasks if t.status == TaskStatus.PENDING)
    running = sum(1 for t in tasks if t.status == TaskStatus.RUNNING)

    timeline = []
    for t in tasks:
        entry = {
            "task_id": t.id,
            "alias": t.alias,
            "agent": t.agent,
            "description": t.description,
            "status": t.status.value,
            "depends_on": t.depends_on,
        }
        if t.model:
            entry["model"] = t.model
        if t.schema:
            entry["schema"] = t.schema
        if t.writes is not None:
            entry["writes"] = list(t.writes)
        if t.passes_total > 1:
            entry["passes_total"] = t.passes_total
            entry["passes_done"] = t.passes_done
            if t.pass_notes:
                entry["pass_notes"] = list(t.pass_notes)
        if t.started_at:
            entry["started_at"] = t.started_at.isoformat()
        if t.completed_at:
            entry["completed_at"] = t.completed_at.isoformat()
        if t.result:
            entry["result_preview"] = t.result[:200]
        if t.parsed_result:
            entry["parsed_result"] = t.parsed_result
        timeline.append(entry)

    return {
        "summary": {
            "total": total,
            "done": done,
            "failed": failed,
            "cancelled": cancelled,
            "pending": pending,
            "running": running,
            "extensions_used": self._extensions_count,
            "max_extensions": self._max_extensions,
            "forced_extensions": self._forced_extensions,
            "hard_max_extensions": self._hard_max_extensions,
            "all_complete": done + failed + cancelled == total,
        },
        "timeline": timeline,
    }

to_state

to_state() -> dict

Full-fidelity serialization for process-boundary persistence.

Unlike :meth:export_trace (telemetry/debugging), this round-trips with :meth:from_state exactly — it preserves created_at and the live RUNNING status (no reset), so a CLI that persists between command invocations can claim a task in one call and submit it in the next.

Source code in coordinator/graph.py

def to_state(self) -> dict:
    """Full-fidelity serialization for process-boundary persistence.

    Unlike :meth:`export_trace` (telemetry/debugging), this round-trips with
    :meth:`from_state` exactly — it preserves ``created_at`` and the live
    ``RUNNING`` status (no reset), so a CLI that persists between command
    invocations can claim a task in one call and submit it in the next.
    """
    return {
        "version": 1,
        "max_extensions": self._max_extensions,
        "extensions_used": self._extensions_count,
        "forced_extensions": self._forced_extensions,
        "hard_max_extensions": self._hard_max_extensions,
        "tasks": [
            {
                "id": t.id,
                "alias": t.alias,
                "agent": t.agent,
                "description": t.description,
                "model": t.model,
                "schema": t.schema,
                "schema_obj": t.schema_obj,
                "status": t.status.value,
                "depends_on": list(t.depends_on),
                "writes": t.writes,
                "result": t.result,
                "parsed_result": t.parsed_result,
                "passes_total": t.passes_total,
                "passes_done": t.passes_done,
                "pass_notes": list(t.pass_notes),
                "created_at": t.created_at.isoformat() if t.created_at else None,
                "started_at": t.started_at.isoformat() if t.started_at else None,
                "completed_at": t.completed_at.isoformat() if t.completed_at else None,
            }
            for t in self._tasks.values()
        ],
    }

from_state classmethod

from_state(data: dict, agent_roles: dict[str, str] | None = None) -> 'TaskGraph'

Reconstruct a graph from :meth:to_state output, preserving status.

Parameters:

Name	Type	Description	Default
data	dict	A dict produced by :meth:to_state.	required
agent_roles	dict[str, str] | None	Role overrides to attach to the rebuilt graph (e.g. project-local roles). Roles are not serialized because they are resolved from configuration, not persisted run state.	None

Source code in coordinator/graph.py

@classmethod
def from_state(
    cls, data: dict, agent_roles: dict[str, str] | None = None
) -> "TaskGraph":
    """Reconstruct a graph from :meth:`to_state` output, preserving status.

    Args:
        data: A dict produced by :meth:`to_state`.
        agent_roles: Role overrides to attach to the rebuilt graph (e.g.
            project-local roles). Roles are not serialized because they are
            resolved from configuration, not persisted run state.
    """

    def parse_time(value) -> datetime | None:
        if not value:
            return None
        try:
            return datetime.fromisoformat(str(value))
        except (ValueError, TypeError):
            return None

    graph = cls(
        max_extensions=int(data.get("max_extensions", 3)),
        agent_roles=agent_roles,
        hard_max_extensions=(
            int(data["hard_max_extensions"])
            if data.get("hard_max_extensions") is not None
            else None
        ),
    )
    for td in data.get("tasks", []):
        task = Task(
            id=td["id"],
            description=td.get("description", ""),
            agent=td.get("agent", ""),
            alias=td.get("alias", ""),
            model=td.get("model"),
            schema=td.get("schema"),
            schema_obj=td.get("schema_obj"),
            status=TaskStatus(td.get("status", "pending")),
            depends_on=list(td.get("depends_on", [])),
            writes=td.get("writes"),
            result=td.get("result"),
            parsed_result=td.get("parsed_result"),
            passes_total=max(1, int(td.get("passes_total", 1) or 1)),
            passes_done=max(0, int(td.get("passes_done", 0) or 0)),
            pass_notes=list(td.get("pass_notes", []) or []),
            started_at=parse_time(td.get("started_at")),
            completed_at=parse_time(td.get("completed_at")),
        )
        created = parse_time(td.get("created_at"))
        if created is not None:
            task.created_at = created
        graph._tasks[task.id] = task
        if task.alias:
            graph._alias_to_id[task.alias] = task.id
    graph._extensions_count = max(0, int(data.get("extensions_used", 0)))
    graph._forced_extensions = max(0, int(data.get("forced_extensions", 0) or 0))
    return graph

export_trace

export_trace() -> dict

Serialize the full graph state for telemetry/debugging.

Source code in coordinator/graph.py

def export_trace(self) -> dict:
    """Serialize the full graph state for telemetry/debugging."""
    tasks = []
    for t in self._tasks.values():
        entry = {
            "task_id": t.id,
            "alias": t.alias,
            "agent": t.agent,
            "description": t.description,
            "status": t.status.value,
            "depends_on": t.depends_on,
            "result": t.result,
            "parsed_result": t.parsed_result,
        }
        if t.model:
            entry["model"] = t.model
        if t.schema:
            entry["schema"] = t.schema
        if t.writes is not None:
            entry["writes"] = list(t.writes)
        if t.passes_total > 1:
            entry["passes_total"] = t.passes_total
            entry["passes_done"] = t.passes_done
            if t.pass_notes:
                entry["pass_notes"] = list(t.pass_notes)
        if t.started_at:
            entry["started_at"] = t.started_at.isoformat()
        if t.completed_at:
            entry["completed_at"] = t.completed_at.isoformat()
        if t.started_at and t.completed_at:
            entry["duration_seconds"] = (t.completed_at - t.started_at).total_seconds()
        tasks.append(entry)

    total_duration = 0.0
    started_times = [t.started_at for t in self._tasks.values() if t.started_at]
    completed_times = [t.completed_at for t in self._tasks.values() if t.completed_at]
    if started_times and completed_times:
        total_duration = (max(completed_times) - min(started_times)).total_seconds()

    return {
        "tasks": tasks,
        "extensions_used": self._extensions_count,
        "max_extensions": self._max_extensions,
        "total_tasks": len(self._tasks),
        "total_duration_seconds": total_duration,
    }

normalize_write_path

normalize_write_path(path: str | None) -> str

Normalize a declared write path into a comparable lock key.

Returns "" (the whole-workspace sentinel) for an empty path, ., /, or a path whose first segment is a glob. A glob is otherwise reduced to its longest non-magic leading directory prefix (notes/A/*.md -> notes/A), so a lease conservatively covers the glob's containing dir.

Source code in coordinator/graph.py

def normalize_write_path(path: str | None) -> str:
    """Normalize a declared write path into a comparable lock key.

    Returns ``""`` (the whole-workspace sentinel) for an empty path, ``.``,
    ``/``, or a path whose first segment is a glob. A glob is otherwise reduced
    to its longest non-magic leading directory prefix (``notes/A/*.md`` ->
    ``notes/A``), so a lease conservatively covers the glob's containing dir.
    """
    if path is None:
        return ""
    p = str(path).strip().replace("\\", "/")
    while p.startswith("./"):
        p = p[2:]
    p = p.strip("/")
    if not p or p == ".":
        return ""
    segments: list[str] = []
    for seg in p.split("/"):
        if seg in ("", "."):
            continue
        if any(c in _GLOB_MAGIC for c in seg):
            break
        segments.append(seg)
    return "/".join(segments)

keys_overlap

keys_overlap(a: str, b: str) -> bool

Two normalized keys overlap if one contains the other (or either is the whole-workspace sentinel "").

Source code in coordinator/graph.py

def keys_overlap(a: str, b: str) -> bool:
    """Two normalized keys overlap if one contains the other (or either is the
    whole-workspace sentinel ``""``)."""
    if a == "" or b == "":
        return True
    if a == b:
        return True
    return a.startswith(b + "/") or b.startswith(a + "/")

lease_keys

lease_keys(writes: list[str] | None) -> list[str]

Resolve a task's declared writes to claim-time lock keys.

None -> [""] (whole-workspace lease; conservative default that reproduces the legacy one-implementer-at-a-time behavior). [] -> [] (dynamic mode: no claim-time lease). A concrete list is normalized to unique keys; any element that normalizes to the whole-workspace sentinel collapses the whole set to [""].

Source code in coordinator/graph.py

def lease_keys(writes: list[str] | None) -> list[str]:
    """Resolve a task's declared ``writes`` to claim-time lock keys.

    ``None`` -> ``[""]`` (whole-workspace lease; conservative default that
    reproduces the legacy one-implementer-at-a-time behavior). ``[]`` -> ``[]``
    (dynamic mode: no claim-time lease). A concrete list is normalized to unique
    keys; any element that normalizes to the whole-workspace sentinel collapses
    the whole set to ``[""]``.
    """
    if writes is None:
        return [""]
    keys: list[str] = []
    for w in writes:
        k = normalize_write_path(w)
        if k == "":
            return [""]
        if k not in keys:
            keys.append(k)
    return keys

write_sets_conflict

write_sets_conflict(a: list[str] | None, b: list[str] | None) -> bool

Whether two declared write scopes would contend at claim time.

A side holding nothing (dynamic mode, []) never conflicts at claim.

Source code in coordinator/graph.py

def write_sets_conflict(a: list[str] | None, b: list[str] | None) -> bool:
    """Whether two declared write scopes would contend at claim time.

    A side holding nothing (dynamic mode, ``[]``) never conflicts at claim.
    """
    ka, kb = lease_keys(a), lease_keys(b)
    if not ka or not kb:
        return False
    return any(keys_overlap(x, y) for x in ka for y in kb)

validate_graph

validate_graph(tasks: list[dict], existing_task_ids: set[str] | None = None, agent_roles: dict[str, str] | None = None) -> list[str]

Validate a DAG definition. Returns list of violations (empty = valid).

Parameters:

Name	Type	Description	Default
tasks	list[dict]	List of task dicts with keys: alias, agent, depends_on.	required
existing_task_ids	set[str] | None	Task IDs already in the graph (for extension validation).	None
agent_roles	dict[str, str] | None	Extra agent->role mappings (e.g. project-local agents), merged over the built-in AGENT_ROLES. Roles: "implementer", "planner", "checker", "meta". Unknown agents default to "checker".	None

Rules

• Acyclic: no cycles in the dependency graph.
• At least one implementer or planner: a graph must have at least one agent that does real work (writes code or designs a plan). A planner-only graph (no implementer) is valid.
• Implementer ancestry: every checker task must have at least one implementer in its transitive dependency chain. Meta and planner agents are exempt.
• Planner ancestry: planner tasks must NOT have an implementer in their transitive dependency chain (planners run before engineers, not after them).
• No orphan references: every depends_on alias must resolve to a task in the graph or in existing_task_ids.

Source code in coordinator/graph.py

def validate_graph(
    tasks: list[dict],
    existing_task_ids: set[str] | None = None,
    agent_roles: dict[str, str] | None = None,
) -> list[str]:
    """Validate a DAG definition. Returns list of violations (empty = valid).

    Args:
        tasks: List of task dicts with keys: alias, agent, depends_on.
        existing_task_ids: Task IDs already in the graph (for extension validation).
        agent_roles: Extra agent->role mappings (e.g. project-local agents),
            merged over the built-in AGENT_ROLES. Roles: "implementer",
            "planner", "checker", "meta". Unknown agents default to "checker".

    Rules:
        - Acyclic: no cycles in the dependency graph.
        - At least one implementer or planner: a graph must have at least
          one agent that does real work (writes code or designs a plan).
          A planner-only graph (no implementer) is valid.
        - Implementer ancestry: every checker task must have at least one
          implementer in its transitive dependency chain. Meta and planner
          agents are exempt.
        - Planner ancestry: planner tasks must NOT have an implementer in
          their transitive dependency chain (planners run before engineers,
          not after them).
        - No orphan references: every depends_on alias must resolve to a
          task in the graph or in existing_task_ids.
    """
    if not tasks:
        return ["Graph must have at least one task."]

    roles = {**AGENT_ROLES, **(agent_roles or {})}

    def role_of(agent: str) -> str:
        return roles.get(agent, "checker")

    violations = []
    alias_list = [t["alias"] for t in tasks if t.get("alias")]
    aliases = set(alias_list)
    for dup in sorted({a for a in alias_list if alias_list.count(a) > 1}):
        violations.append(
            f"Duplicate alias '{dup}': each task must have a unique alias."
        )
    valid_refs = aliases | (existing_task_ids or set())

    agents_by_alias: dict[str, str] = {}
    deps_by_alias: dict[str, list[str]] = {}

    for t in tasks:
        alias = t.get("alias", "")
        agent = t.get("agent", "")
        depends_on = t.get("depends_on", [])

        if alias:
            agents_by_alias[alias] = agent
            deps_by_alias[alias] = depends_on

        for dep in depends_on:
            if dep not in valid_refs:
                violations.append(f"Task '{alias}': depends_on '{dep}' does not exist.")

    task_agents = [t.get("agent", "") for t in tasks]
    has_implementer = any(role_of(a) == "implementer" for a in task_agents)
    has_planner = any(role_of(a) == "planner" for a in task_agents)
    if not has_implementer and not has_planner:
        violations.append(
            "Graph must contain at least one implementer (e.g. engineer) "
            "or planner task."
        )

    if _has_cycle(deps_by_alias):
        violations.append("Graph contains a cycle.")

    for t in tasks:
        alias = t.get("alias", "")
        agent = t.get("agent", "")
        role = role_of(agent)
        if role == "checker" and alias:
            if not _has_implementer_ancestor(
                alias, deps_by_alias, agents_by_alias, role_of
            ):
                violations.append(
                    f"Task '{alias}' ({agent}): no engineer/implementer "
                    "in dependency ancestry."
                )
        elif role == "planner" and alias:
            if _has_implementer_ancestor(
                alias, deps_by_alias, agents_by_alias, role_of
            ):
                violations.append(
                    f"Task '{alias}' ({agent}): planner must not depend "
                    "on an implementer (planners run before engineers)."
                )

    return violations