(self, args: str, context: CommandContext)
| 33 | @dataclass(frozen=True) |
| 34 | class WorkflowsCommand(InteractiveCommand): |
| 35 | async def run(self, args: str, context: CommandContext) -> InteractiveOutcome: |
| 36 | tc = getattr(context, "tool_context", None) |
| 37 | registry = getattr(tc, "runtime_tasks", None) if tc is not None else None |
| 38 | if registry is None: |
| 39 | return InteractiveOutcome( |
| 40 | message="Workflows are unavailable on this surface.", display="system" |
| 41 | ) |
| 42 | try: |
| 43 | runs = [t for t in registry.all() if getattr(t, "type", None) == "local_workflow"] |
| 44 | except Exception: |
| 45 | runs = [] |
| 46 | if not runs: |
| 47 | return InteractiveOutcome( |
| 48 | message="No workflow runs. Start one with /deep-research or by asking for a workflow.", |
| 49 | display="system", |
| 50 | ) |
| 51 | from src.workflow.progress import render_run_lines |
| 52 | |
| 53 | blocks: list[str] = [] |
| 54 | for t in runs: |
| 55 | lines = render_run_lines(t) |
| 56 | run_id = getattr(t, "run_id", "") or "" |
| 57 | if run_id and lines: |
| 58 | lines[0] = f"{lines[0]} (run: {run_id})" |
| 59 | blocks.append("\n".join(lines)) |
| 60 | return InteractiveOutcome(message="\n\n".join(blocks), display="system") |
| 61 | |
| 62 | |
| 63 | WORKFLOWS_COMMAND = WorkflowsCommand( |
nothing calls this directly
no test coverage detected