Coverage for cli / commands / tasks_cmd.py: 100.00%
159 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Long-running task observability commands.
3Mirrors the ``task_status`` / ``task_tail`` MCP tools so operators can
4inspect the same on-disk status records from a terminal:
6* ``gco tasks list`` — newest-first table of recent invocations
7* ``gco tasks show TASK_ID`` — full record for one task
8* ``gco tasks tail TASK_ID -n 100 [-f]`` — last N lines of raw output,
9 with ``-f`` polling like ``tail -f``
10* ``gco tasks prune`` — drop all but the most recent N records
12Long-running MCP tools (``deploy_all``, ``destroy_all``,
13``bootstrap_cdk``, ``deploy_stack``, ``destroy_stack``,
14``images_build``, ``images_push``) record progress to
15``~/.gco/tasks/{task_id}.json`` and the raw subprocess output to
16``~/.gco/tasks/{task_id}.log`` on every line. This module reads
17those artifacts and never writes them — the writer lives in
18``gco_mcp/tools/_task_status.py``.
19"""
21import contextlib
22import json
23import sys
24import time
25from pathlib import Path
26from typing import Any
28import click
30from ..output import confirm, emit_structured_document
33def _status_dir() -> Path:
34 """Honour ``GCO_TASK_STATUS_DIR`` for tests, fall back to ``~/.gco/tasks``.
36 Mirrors ``mcp.tools._task_status.status_dir`` so the CLI and the MCP
37 server always read from the same place. We don't import the MCP
38 helper directly because ``cli/`` and ``gco_mcp/`` are separate top-level
39 packages and we want this command to work without the MCP install.
40 """
41 import os
43 override = os.environ.get("GCO_TASK_STATUS_DIR")
44 if override:
45 return Path(override)
46 return Path.home() / ".gco" / "tasks"
49def _is_pid_alive(pid: int | None) -> bool:
50 """Best-effort liveness check via ``os.kill(pid, 0)``.
52 Returns ``False`` when the PID is missing/zero or the OS reports
53 the process gone. Returns ``True`` for live processes including
54 those owned by other users (``PermissionError``). Anything else is
55 treated as not-alive so an unexpected ``OSError`` can't strand a
56 task in ``running`` forever.
57 """
58 import os
60 if pid is None or pid <= 0:
61 return False
62 try:
63 os.kill(pid, 0)
64 return True
65 except ProcessLookupError:
66 return False
67 except PermissionError:
68 return True
69 except OSError:
70 return False
73def _read_status(path: Path) -> dict[str, Any] | None:
74 """Load one status JSON, applying the orphan rewrite.
76 Identical semantics to ``mcp.tools._task_status._read_status_file``:
77 re-checks the PID and rewrites ``state=running`` to ``orphaned``
78 when the recorded process is dead. Kept as a local copy so the
79 CLI doesn't need ``gco_mcp/`` on the import path.
80 """
81 try:
82 record = json.loads(path.read_text(encoding="utf-8"))
83 except OSError, ValueError:
84 return None
85 if not isinstance(record, dict):
86 return None
87 pid = record.get("pid")
88 is_alive = _is_pid_alive(pid if isinstance(pid, int) else None)
89 record["is_alive"] = is_alive
90 if record.get("state") == "running" and not is_alive:
91 record["state"] = "orphaned"
92 return record
95def _list_records(directory: Path) -> list[dict[str, Any]]:
96 """Return all records newest-first."""
97 if not directory.exists():
98 return []
99 out: list[dict[str, Any]] = []
100 for path in sorted(directory.glob("*.json"), key=lambda p: p.stat().st_mtime, reverse=True):
101 record = _read_status(path)
102 if record is not None:
103 out.append(record)
104 return out
107def _format_state(state: str) -> str:
108 """Colour-code state for terminal output. Plain text when not a TTY."""
109 if not sys.stdout.isatty():
110 return state
111 palette = {
112 "running": "\x1b[36mrunning\x1b[0m", # cyan
113 "succeeded": "\x1b[32msucceeded\x1b[0m", # green
114 "failed": "\x1b[31mfailed\x1b[0m", # red
115 "cancelled": "\x1b[33mcancelled\x1b[0m", # yellow
116 "orphaned": "\x1b[35morphaned\x1b[0m", # magenta
117 }
118 return palette.get(state, state)
121def _format_elapsed(seconds: int | None) -> str:
122 """Render an integer second count compactly: ``s`` / ``MmSs`` / ``HhMm``."""
123 if seconds is None:
124 return "-"
125 if seconds < 60:
126 return f"{seconds}s"
127 minutes, sec = divmod(seconds, 60)
128 if minutes < 60:
129 return f"{minutes}m{sec:02d}s"
130 hours, mins = divmod(minutes, 60)
131 return f"{hours}h{mins:02d}m"
134@click.group()
135def tasks() -> None:
136 """Inspect long-running MCP / CLI task status.
138 Commands like ``gco stacks deploy-all`` and ``gco images push`` write
139 progress records and raw subprocess logs to ``~/.gco/tasks/`` so you
140 can observe them without parsing terminal scrollback. ``gco tasks
141 list/show/tail`` read those files; the writer is in the MCP tool
142 runner.
143 """
146@tasks.command("list")
147@click.option(
148 "-n",
149 "--limit",
150 type=int,
151 default=20,
152 show_default=True,
153 help="Maximum records to display (newest first).",
154)
155@click.option(
156 "--json",
157 "as_json",
158 is_flag=True,
159 help="Emit raw JSON instead of the table.",
160)
161def tasks_list(limit: int, as_json: bool) -> None:
162 """List recent task invocations newest-first.
164 Shows tool, state (with orphan rewriting for dead PIDs), elapsed
165 wall-clock, stacks completed, and the last stack name observed.
166 Pass ``--json`` for machine-readable output you can pipe to ``jq``.
167 """
168 records = _list_records(_status_dir())[:limit] if limit > 0 else _list_records(_status_dir())
170 if as_json:
171 emit_structured_document(
172 {"tasks": records},
173 output_format="json",
174 rendered=json.dumps({"tasks": records}, indent=2, sort_keys=True),
175 )
176 return
178 if not records:
179 click.echo(
180 "No tasks recorded yet. Run a long-running command (e.g. 'gco stacks deploy-all') to populate ~/.gco/tasks/."
181 )
182 return
184 header = (
185 f"{'TASK ID':<40} {'TOOL':<18} {'STATE':<11} {'ELAPSED':<8} {'STACKS':<10} LAST STACK"
186 )
187 click.echo(header)
188 click.echo("-" * len(header))
189 for r in records:
190 task_id = (r.get("task_id") or "")[:40]
191 tool = (r.get("tool") or "")[:18]
192 state = _format_state(r.get("state") or "?")
193 elapsed = _format_elapsed(r.get("elapsed_seconds"))
194 stacks_completed = r.get("stacks_completed") or 0
195 stacks_total = r.get("stacks_total")
196 stacks = f"{stacks_completed}/{stacks_total}" if stacks_total else f"{stacks_completed}"
197 last_stack = r.get("last_stack") or "-"
198 # State string may include ANSI codes — pad on the visible width.
199 visible_state = r.get("state") or "?"
200 state_pad = " " * max(0, 11 - len(visible_state))
201 click.echo(
202 f"{task_id:<40} {tool:<18} {state}{state_pad} {elapsed:<8} {stacks:<10} {last_stack}"
203 )
206@tasks.command("show")
207@click.argument("task_id")
208def tasks_show(task_id: str) -> None:
209 """Print the full JSON record for one task.
211 Useful when ``gco tasks list`` shows a task that needs deeper
212 inspection — argv, exit code, stderr tail, etc.
213 """
214 path = _status_dir() / f"{task_id}.json"
215 record = _read_status(path)
216 if record is None:
217 click.echo(f"Task not found: {task_id}", err=True)
218 sys.exit(1)
219 emit_structured_document(
220 record,
221 output_format="json",
222 rendered=json.dumps(record, indent=2, sort_keys=True),
223 )
226@tasks.command("tail")
227@click.argument("task_id")
228@click.option(
229 "-n",
230 "--lines",
231 type=int,
232 default=100,
233 show_default=True,
234 help="Lines to show.",
235)
236@click.option(
237 "-f",
238 "--follow",
239 is_flag=True,
240 help="Follow the log file (poll for new lines, like 'tail -f').",
241)
242def tasks_tail(task_id: str, lines: int, follow: bool) -> None:
243 """Print the last N lines of a task's raw output log.
245 Each line is prefixed with ``[stdout]`` or ``[stderr]`` so you can
246 tell which stream produced it. ``--follow`` keeps polling the file
247 until interrupted, mirroring ``tail -f``.
248 """
249 log_path = _status_dir() / f"{task_id}.log"
250 if not log_path.exists():
251 click.echo(f"No log for task: {task_id}", err=True)
252 sys.exit(1)
254 # Initial tail.
255 from collections import deque
257 try:
258 with open(log_path, encoding="utf-8", errors="replace") as fp:
259 buf: deque[str] = deque(fp, maxlen=lines if lines > 0 else 0)
260 except OSError as e:
261 click.echo(f"Failed to read log: {e}", err=True)
262 sys.exit(1)
264 for line in buf:
265 click.echo(line.rstrip("\n"))
267 if not follow:
268 return
270 # Follow mode: poll the file every 500ms.
271 try:
272 with open(log_path, encoding="utf-8", errors="replace") as fp:
273 fp.seek(0, 2) # end of file
274 while True:
275 chunk = fp.read()
276 if chunk:
277 click.echo(chunk, nl=False)
278 else:
279 # Stop following once the task is no longer running.
280 record = _read_status(_status_dir() / f"{task_id}.json")
281 if record is not None and record.get("state") not in {"running"}:
282 return
283 time.sleep(0.5)
284 except KeyboardInterrupt:
285 return
288@tasks.command("prune")
289@click.option(
290 "-k",
291 "--keep",
292 type=int,
293 default=50,
294 show_default=True,
295 help="Keep the N most recent tasks. Older are deleted.",
296)
297@click.option("-y", "--yes", is_flag=True, help="Skip confirmation.")
298def tasks_prune(keep: int, yes: bool) -> None:
299 """Delete old task records, keeping the most recent N.
301 Useful if ``~/.gco/tasks/`` has accumulated stale records and you
302 want a manual sweep. The MCP runner also auto-prunes on every new
303 task start, so this is purely for ad-hoc cleanup.
304 """
305 directory = _status_dir()
306 if not directory.exists():
307 click.echo("No task directory yet — nothing to prune.")
308 return
310 json_files = sorted(directory.glob("*.json"), key=lambda p: p.stat().st_mtime, reverse=True)
311 stale = json_files[keep:]
312 if not stale:
313 click.echo(f"Already at or below {keep} task(s). Nothing to do.")
314 return
316 if not yes:
317 confirm(
318 f"Delete {len(stale)} task record(s) older than the {keep} most recent?",
319 abort=True,
320 )
322 removed = 0
323 for path in stale:
324 with contextlib.suppress(OSError):
325 path.unlink()
326 removed += 1
327 log_path = path.with_suffix(".log")
328 if log_path.exists():
329 with contextlib.suppress(OSError):
330 log_path.unlink()
332 click.echo(f"Removed {removed} task record(s).")