Class WorkflowProgress

src/workflow/progress.py:66–166

class WorkflowProgress:
    def __init__(
        self,
        phases_meta: Optional[list[dict]] = None,
        *,
        on_change: Callable[["WorkflowProgress"], None] | None = None,
    ) -> None:
        # Seed declared phases (from meta.phases) so the tree shows them before
        # any agent runs; phase() started later by title reuses the same record.
        self._phases: list[PhaseRecord] = [
            PhaseRecord(title=p.get("title", ""), detail=p.get("detail"))
            for p in (phases_meta or [])
            if isinstance(p, dict) and p.get("title")
        ]
        self._logs: list[str] = []
        self._current_phase: Optional[str] = None
        self._on_change = on_change

    # ── mutations ──────────────────────────────────────────────────────────
    def start_phase(self, title: str) -> None:
        self._current_phase = title
        if not any(p.title == title for p in self._phases):
            self._phases.append(PhaseRecord(title=title))
        self._changed()

    def log(self, message: str) -> None:
        self._logs.append(message)
        self._changed()

    def agent_started(
        self,
        index: int,
        label: str,
        phase: Optional[str],
        key: str = "",
        agent_type: str = "",
    ) -> AgentRecord:
        phase = phase or self._current_phase
        record = AgentRecord(
            index=index, label=label, phase=phase, key=key,
            agent_type=agent_type, started_at=time.monotonic(),
        )
        self._phase_for(phase).agents.append(record)
        self._changed()
        return record

    def agent_finished(
        self,
        record: AgentRecord,
        *,
        status: AgentStatus,
        tokens: int = 0,
        error: Optional[str] = None,
        tool_count: int = 0,
    ) -> None:
        record.status = status
        record.tokens = tokens
        record.error = error