Coverage for gco_mcp/tools/_task_status.py: 97.87%

1""" 

2Disk-backed status reporting for long-running MCP tools. 

3 

4The MCP spec lets tools stream progress and log notifications back to the 

5calling client, but client-side rendering of those notifications is 

6inconsistent. Some clients drop them, some bury them in a debug panel, 

7some never surface them at all. The result: a 30-60 minute deploy_all 

8looks "wedged" to the user even though the underlying CLI is producing 

9output every few seconds. 

10 

11This module gives every long-running tool a parallel observability 

12channel that doesn't depend on the MCP wire at all: 

13 

14* A JSON ``status`` file at ``~/.gco/tasks/{task_id}.json`` updated on 

15 every progress event (atomic via tempfile + ``os.replace``). 

16* A private, size-bounded ``log`` file at 

17 ``~/.gco/tasks/{task_id}.log`` containing interleaved stdout+stderr 

18 records from the subprocess. 

19* Orphan detection on read: when the status reports ``state=running`` 

20 but the recorded PID is no longer alive, the returned dict is 

21 re-stamped to ``state=orphaned`` so callers see honest data even 

22 when the MCP wrapper crashed without a final write. 

23 

24Two MCP tools (``task_status`` / ``task_tail``) and one CLI group 

25(``gco tasks list/tail/prune``) read these files. Both surfaces are 

26read-only — the writer always lives in ``_run_long_task``. 

27 

28The status directory is configurable via ``GCO_TASK_STATUS_DIR`` so 

29unit tests can isolate to ``tmp_path``. ``GCO_DISABLE_TASK_STATUS=1`` 

30skips file emission entirely (kept as an escape hatch for sandboxed 

31environments where ``~/.gco`` isn't writable). 

32""" 

33 

34from __future__ import annotations 

35 

36import json 

37import os 

38import re 

39import stat 

40import tempfile 

41import threading 

42import time 

43from collections import deque 

44from collections.abc import Iterable 

45from datetime import UTC, datetime 

46from pathlib import Path 

47from typing import IO, Any 

48 

49# Task IDs are used as filenames. Keep the accepted alphabet deliberately 

50# narrower than a generic filesystem name: generated IDs and FastMCP's UUID-like 

51# IDs fit, while path separators, control characters, drive prefixes, and 

52# special ``.`` / ``..`` components cannot reach a filesystem API. 

53_TASK_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$") 

54 

55# Disk observability must remain bounded even when a child emits a giant line or 

56# never stops logging. The status JSON keeps a smaller per-line projection; the 

57# raw log has independent per-record and whole-file ceilings; tail readers cap 

58# both caller-requested line count and bytes read from disk. 

59_STATUS_LINE_MAX_BYTES = 8 * 1024 

60_STATUS_FILE_MAX_BYTES = 1 * 1024 * 1024 

61_STATUS_TOOL_MAX_BYTES = 256 

62_STATUS_ARG_MAX_ITEMS = 128 

63_STATUS_ARG_MAX_BYTES = 4 * 1024 

64_TASK_LOG_LINE_MAX_BYTES = 64 * 1024 

65_TASK_LOG_MAX_BYTES = 10 * 1024 * 1024 

66_TASK_TAIL_MAX_LINES = 500 

67_TASK_TAIL_MAX_BYTES = 1 * 1024 * 1024 

68_TRUNCATED_TEXT = "...[truncated]" 

69_LOG_TRUNCATED_RECORD = b"[gco-mcp] log truncated at configured byte limit\n" 

70 

71# Keep at most this many task files (status + log) in the directory. 

72# When a new task starts, anything older than the most recent N gets 

73# pruned. 50 is enough for a couple of full deploy cycles plus a 

74# handful of one-off image pushes. 

75_TASK_RETENTION = 50 

76 

77# Tail buffer kept in the status file. Larger than what any single 

78# message line will need, but capped so the JSON file stays small. 

79_STATUS_TAIL_LINES = 20 

80 

81# Debounce window for atomic status writes. Per-line writes would do 

82# hundreds of fsyncs during a noisy CDK phase; this batches them so 

83# we write at most ~2 times per second under sustained output. 

84_STATUS_WRITE_DEBOUNCE_SECONDS = 0.5 

85 

86# How long to consider a process "alive" — really just a guard so a 

87# stale PID that's been recycled by another unrelated process isn't 

88# falsely reported as still running. We don't try to be clever about 

89# PID recycling beyond this; ``ps``-style verification would need 

90# command-line matching and is out of scope. 

91_PID_ALIVE_SIGNAL = 0 

92 

93# Serialize task creation/pruning inside one MCP process. Atomic file writes 

94# already protect readers; this lock additionally prevents two local writers 

95# from pruning between each other's status and log creation windows. 

96_TASK_ARTIFACT_LOCK = threading.RLock() 

97 

98 

99def status_dir() -> Path: 

100 """Resolve the status directory honouring the env override. 

101 

102 Tests set ``GCO_TASK_STATUS_DIR`` to a ``tmp_path`` so they don't 

103 write to the developer's real home dir. 

104 """ 

105 override = os.environ.get("GCO_TASK_STATUS_DIR") 

106 if override: 

107 return Path(override) 

108 return Path.home() / ".gco" / "tasks" 

109 

110 

111def task_status_enabled() -> bool: 

112 """``True`` unless the operator explicitly opted out. 

113 

114 The opt-out is a defensive escape hatch — sandboxed CI runs and 

115 container builds that mount a read-only home directory can set 

116 ``GCO_DISABLE_TASK_STATUS=1`` to skip the disk writes without 

117 losing any of the MCP wire-side observability. 

118 """ 

119 return os.environ.get("GCO_DISABLE_TASK_STATUS", "").lower() not in {"1", "true", "yes"} 

120 

121 

122def _now_iso() -> str: 

123 """ISO-8601 UTC timestamp with second precision.""" 

124 return datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") 

125 

126 

127def is_valid_task_id(task_id: object) -> bool: 

128 """Return whether ``task_id`` is safe to use as one flat filename stem.""" 

129 return ( 

130 isinstance(task_id, str) 

131 and task_id not in {".", ".."} 

132 and _TASK_ID_RE.fullmatch(task_id) is not None 

133 ) 

134 

135 

136def _bounded_text(value: str, max_bytes: int) -> str: 

137 """Return UTF-8 text no larger than ``max_bytes``, with a marker on truncation.""" 

138 encoded = value.encode("utf-8", errors="replace") 

139 if len(encoded) <= max_bytes: 

140 return value 

141 marker = _TRUNCATED_TEXT.encode() 

142 if max_bytes <= len(marker): 

143 return marker[:max_bytes].decode("utf-8", errors="ignore") 

144 prefix = encoded[: max_bytes - len(marker)] 

145 return prefix.decode("utf-8", errors="ignore") + _TRUNCATED_TEXT 

146 

147 

148def _open_status_directory(directory: Path) -> int: 

149 """Open the canonical status root and verify its identity without links.""" 

150 root = directory.expanduser().resolve(strict=True) 

151 before = root.lstat() 

152 if not stat.S_ISDIR(before.st_mode): 

153 raise OSError("task status root is not a directory") 

154 flags = ( 

155 os.O_RDONLY 

156 | getattr(os, "O_DIRECTORY", 0) 

157 | getattr(os, "O_CLOEXEC", 0) 

158 | getattr(os, "O_NOFOLLOW", 0) 

159 ) 

160 fd = os.open(root, flags) 

161 try: 

162 after = os.fstat(fd) 

163 if not stat.S_ISDIR(after.st_mode) or (before.st_dev, before.st_ino) != ( 

164 after.st_dev, 

165 after.st_ino, 

166 ): 

167 raise OSError("task status root changed while opening") 

168 return fd 

169 except Exception: 

170 os.close(fd) 

171 raise 

172 

173 

174def _open_task_artifact( 

175 task_id: object, 

176 suffix: str, 

177 directory: Path, 

178) -> int | None: 

179 """Open one flat task artifact relative to a verified root descriptor. 

180 

181 Flat task-ID validation blocks lexical traversal. Descriptor-relative 

182 no-follow opening then prevents final-link and status-root replacement 

183 races from redirecting reads outside the configured directory. Nonblocking 

184 open plus regular/single-link identity checks reject FIFOs, devices, 

185 sockets, hard links, and raced replacements before any read occurs. 

186 """ 

187 if not is_valid_task_id(task_id) or suffix not in {".json", ".log"}: 

188 return None 

189 try: 

190 root_fd = _open_status_directory(directory) 

191 except OSError, RuntimeError: 

192 return None 

193 try: 

194 name = f"{task_id}{suffix}" 

195 before = os.stat(name, dir_fd=root_fd, follow_symlinks=False) 

196 if not stat.S_ISREG(before.st_mode) or before.st_nlink != 1: 

197 return None 

198 flags = ( 

199 os.O_RDONLY 

200 | getattr(os, "O_CLOEXEC", 0) 

201 | getattr(os, "O_NOFOLLOW", 0) 

202 | getattr(os, "O_NONBLOCK", 0) 

203 ) 

204 fd = os.open(name, flags, dir_fd=root_fd) 

205 try: 

206 after = os.fstat(fd) 

207 if ( 

208 not stat.S_ISREG(after.st_mode) 

209 or after.st_nlink != 1 

210 or (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino) 

211 ): 

212 raise OSError("task artifact changed while opening") 

213 return fd 

214 except Exception: 

215 os.close(fd) 

216 raise 

217 except OSError, NotImplementedError, RuntimeError, TypeError, ValueError: 

218 return None 

219 finally: 

220 os.close(root_fd) 

221 

222 

223def _open_regular_nofollow(path: Path) -> int: 

224 """Open an existing regular file without following a final symlink. 

225 

226 ``lstat`` plus post-open identity comparison provides a portable guard; on 

227 platforms exposing ``O_NOFOLLOW`` the kernel also rejects a swapped symlink 

228 atomically. Non-regular artifacts (directories, devices, FIFOs, sockets) 

229 are rejected so a status read can never block on an attacker-controlled 

230 special file. 

231 """ 

232 before = path.lstat() 

233 if not stat.S_ISREG(before.st_mode) or before.st_nlink != 1: 233 ↛ 234line 233 didn't jump to line 234 because the condition on line 233 was never true

234 raise OSError("task artifact is not a private regular file") 

235 flags = ( 

236 os.O_RDONLY 

237 | getattr(os, "O_CLOEXEC", 0) 

238 | getattr(os, "O_NOFOLLOW", 0) 

239 | getattr(os, "O_NONBLOCK", 0) 

240 ) 

241 fd = os.open(path, flags) 

242 try: 

243 after = os.fstat(fd) 

244 if ( 

245 not stat.S_ISREG(after.st_mode) 

246 or after.st_nlink != 1 

247 or (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino) 

248 ): 

249 raise OSError("task artifact changed while opening") 

250 return fd 

251 except Exception: 

252 os.close(fd) 

253 raise 

254 

255 

256def _open_private_log(path: Path) -> int: 

257 """Create or safely reset one private regular log file. 

258 

259 ``O_TRUNC`` is deliberately avoided until an existing path has passed 

260 no-follow, regular-file, single-link, and identity checks. This prevents a 

261 planted symlink or hard link from turning task startup into an arbitrary 

262 host-file truncation primitive, including on platforms without 

263 ``O_NOFOLLOW``. 

264 """ 

265 base_flags = ( 

266 os.O_WRONLY 

267 | getattr(os, "O_CLOEXEC", 0) 

268 | getattr(os, "O_NOFOLLOW", 0) 

269 | getattr(os, "O_NONBLOCK", 0) 

270 ) 

271 try: 

272 fd = os.open(path, base_flags | os.O_CREAT | os.O_EXCL, 0o600) 

273 except FileExistsError as exc: 

274 before = path.lstat() 

275 if not stat.S_ISREG(before.st_mode) or before.st_nlink != 1: 

276 raise OSError("task log is not a private regular file") from exc 

277 fd = os.open(path, base_flags) 

278 try: 

279 after = os.fstat(fd) 

280 if ( 

281 not stat.S_ISREG(after.st_mode) 

282 or after.st_nlink != 1 

283 or (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino) 

284 ): 

285 raise OSError("task log changed while opening") 

286 os.ftruncate(fd, 0) 

287 except Exception: 

288 os.close(fd) 

289 raise 

290 

291 try: 

292 opened = os.fstat(fd) 

293 if not stat.S_ISREG(opened.st_mode) or opened.st_nlink != 1: 293 ↛ 294line 293 didn't jump to line 294 because the condition on line 293 was never true

294 raise OSError("task log is not a private regular file") 

295 os.fchmod(fd, 0o600) 

296 return fd 

297 except Exception: 

298 os.close(fd) 

299 raise 

300 

301 

302def _sorted_regular_json_files(directory: Path) -> list[Path]: 

303 """Return private regular ``*.json`` children sorted newest first.""" 

304 candidates: list[tuple[float, Path]] = [] 

305 try: 

306 paths = directory.glob("*.json") 

307 for path in paths: 

308 try: 

309 metadata = path.lstat() 

310 except OSError: 

311 continue 

312 if ( 

313 is_valid_task_id(path.stem) 

314 and stat.S_ISREG(metadata.st_mode) 

315 and metadata.st_nlink == 1 

316 ): 

317 candidates.append((metadata.st_mtime, path)) 

318 except OSError: 

319 return [] 

320 candidates.sort(key=lambda item: item[0], reverse=True) 

321 return [path for _, path in candidates] 

322 

323 

324def _read_open_regular_fd(fd: int, max_bytes: int) -> bytes | None: 

325 """Read a bounded already-verified regular-file descriptor.""" 

326 try: 

327 size = os.fstat(fd).st_size 

328 if size > max_bytes: 

329 return None 

330 chunks: list[bytes] = [] 

331 remaining = max_bytes + 1 

332 while remaining > 0: 

333 chunk = os.read(fd, min(remaining, 64 * 1024)) 

334 if not chunk: 

335 break 

336 chunks.append(chunk) 

337 remaining -= len(chunk) 

338 data = b"".join(chunks) 

339 return data if len(data) <= max_bytes else None 

340 except OSError: 

341 return None 

342 

343 

344def _read_regular_file(path: Path, max_bytes: int) -> bytes | None: 

345 """Read at most ``max_bytes`` from one no-follow regular file.""" 

346 try: 

347 fd = _open_regular_nofollow(path) 

348 except OSError: 

349 return None 

350 try: 

351 return _read_open_regular_fd(fd, max_bytes) 

352 finally: 

353 os.close(fd) 

354 

355 

356def _atomic_write_json(target: Path, payload: dict[str, Any]) -> None: 

357 """Write ``payload`` to ``target`` atomically. 

358 

359 Uses ``tempfile.NamedTemporaryFile`` in the same directory so 

360 ``os.replace`` is a same-filesystem rename (atomic on POSIX). 

361 Readers always see either the previous file or the new one, 

362 never a partial write. 

363 """ 

364 target.parent.mkdir(parents=True, exist_ok=True) 

365 # delete=False so we can rename it before the context manager closes. 

366 fd = tempfile.NamedTemporaryFile( # noqa: SIM115 - explicit close+replace below 

367 mode="w", 

368 encoding="utf-8", 

369 dir=str(target.parent), 

370 prefix=f".{target.name}.", 

371 suffix=".tmp", 

372 delete=False, 

373 ) 

374 temporary_name = fd.name 

375 try: 

376 json.dump(payload, fd, indent=2, sort_keys=True) 

377 fd.write("\n") 

378 fd.flush() 

379 os.fsync(fd.fileno()) 

380 fd.close() 

381 os.replace(temporary_name, target) 

382 except BaseException: 

383 fd.close() 

384 with _suppress_oserror(): 

385 Path(temporary_name).unlink() 

386 raise 

387 with _suppress_oserror(): 

388 os.chmod(target, 0o600) 

389 

390 

391def _is_pid_alive(pid: int | None) -> bool: 

392 """Best-effort liveness check via signal 0. 

393 

394 ``os.kill(pid, 0)`` raises ``ProcessLookupError`` when the PID is 

395 free, ``PermissionError`` when the PID belongs to a process we 

396 don't own (still alive), and returns ``None`` on success. Any 

397 other ``OSError`` we treat conservatively as "not alive" so an 

398 edge case can't strand a task in ``running`` forever. 

399 """ 

400 if pid is None or pid <= 0: 

401 return False 

402 try: 

403 os.kill(pid, _PID_ALIVE_SIGNAL) 

404 return True 

405 except ProcessLookupError: 

406 return False 

407 except PermissionError: 

408 return True 

409 except OSError: 

410 return False 

411 

412 

413def _unlink_private_regular(path: Path) -> bool: 

414 """Unlink one single-link regular artifact without following links.""" 

415 try: 

416 metadata = path.lstat() 

417 if not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1: 

418 return False 

419 path.unlink() 

420 return True 

421 except OSError: 

422 return False 

423 

424 

425def _sweep_orphan_logs(directory: Path, anchored_stems: set[str]) -> None: 

426 """Remove private task logs that have no retained status anchor.""" 

427 try: 

428 candidates = list(directory.glob("*.log")) 

429 except OSError: 

430 return 

431 for candidate in candidates: 

432 if not is_valid_task_id(candidate.stem) or candidate.stem in anchored_stems: 

433 continue 

434 _unlink_private_regular(candidate) 

435 

436 

437def _prune_old_tasks( 

438 directory: Path, 

439 keep: int = _TASK_RETENTION, 

440 *, 

441 preserve_stem: str | None = None, 

442) -> int: 

443 """Drop old task pairs and private orphan logs, returning pair count. 

444 

445 Creation calls this only after the current status and optional log exist. 

446 ``preserve_stem`` makes the just-created task count toward retention even 

447 when filesystem timestamp resolution causes ties. Pruning is best-effort 

448 and must never break a live task's status emission. 

449 """ 

450 removed = 0 

451 try: 

452 if not directory.exists(): 

453 return 0 

454 with _TASK_ARTIFACT_LOCK: 

455 json_files = _sorted_regular_json_files(directory) 

456 limit = max(0, int(keep)) 

457 if preserve_stem and limit > 0: 

458 current = [path for path in json_files if path.stem == preserve_stem] 

459 others = [path for path in json_files if path.stem != preserve_stem] 

460 json_files = current[:1] + others 

461 for stale in json_files[limit:]: 

462 if not _unlink_private_regular(stale): 

463 continue 

464 removed += 1 

465 _unlink_private_regular(stale.with_suffix(".log")) 

466 

467 anchored_stems = {path.stem for path in _sorted_regular_json_files(directory)} 

468 _sweep_orphan_logs(directory, anchored_stems) 

469 except OSError, RuntimeError, ValueError: 

470 # Pruning is best-effort. Never raise from here. 

471 return removed 

472 return removed 

473 

474 

475class _suppress_oserror: 

476 """Compact context manager that swallows OSError only. 

477 

478 ``contextlib.suppress(OSError)`` would do, but we use the dedicated 

479 type so future readers can grep for the intentional swallows. 

480 """ 

481 

482 def __enter__(self) -> _suppress_oserror: 

483 return self 

484 

485 def __exit__( 

486 self, 

487 exc_type: type[BaseException] | None, 

488 exc: BaseException | None, 

489 tb: object, 

490 ) -> bool: 

491 return exc_type is not None and issubclass(exc_type, OSError) 

492 

493 

494def make_task_id(tool_name: str) -> str: 

495 """Generate a sortable, collision-resistant, filesystem-safe task ID. 

496 

497 Format: ``{tool_name}-{millis_since_epoch}-{counter}``. Executable paths 

498 used by generic long tasks are collapsed to a safe stem before the ID is 

499 assembled; already-safe tool names retain their historical spelling. 

500 """ 

501 raw_stem = re.split(r"[/\\]", str(tool_name))[-1] 

502 safe_stem = re.sub(r"[^A-Za-z0-9._-]+", "_", raw_stem).strip("._-") or "task" 

503 safe_stem = safe_stem[:64] 

504 counter = _next_task_counter() 

505 return f"{safe_stem}-{int(time.time() * 1000)}-{counter}" 

506 

507 

508_TASK_COUNTER_LOCK = threading.Lock() 

509_TASK_COUNTER = 0 

510 

511 

512def _next_task_counter() -> int: 

513 """Return a monotonically-increasing counter, thread-safe.""" 

514 global _TASK_COUNTER 

515 with _TASK_COUNTER_LOCK: 

516 _TASK_COUNTER += 1 

517 return _TASK_COUNTER 

518 

519 

520class TaskStatusWriter: 

521 """Disk-backed status emitter for one long-running tool invocation. 

522 

523 Owns the lifecycle of a ``{task_id}.json`` + ``{task_id}.log`` 

524 pair. Use as a context manager — the ``__exit__`` flushes a 

525 final ``state=succeeded|failed|cancelled`` write so observers 

526 always see a terminal record. 

527 

528 Thread-safety: the lock guards the in-memory tail buffer and 

529 debounce timestamps so it's safe to call ``record_line`` from 

530 the stdout and stderr drain coroutines concurrently. The 

531 underlying file ops are single-writer per task by construction. 

532 """ 

533 

534 def __init__( 

535 self, 

536 task_id: str, 

537 tool: str, 

538 argv: list[str], 

539 *, 

540 pid: int | None, 

541 total_units: int | None = None, 

542 ) -> None: 

543 if not is_valid_task_id(task_id): 

544 raise ValueError("task_id must be a safe filename stem") 

545 self.task_id = task_id 

546 self.tool = _bounded_text(str(tool), _STATUS_TOOL_MAX_BYTES) 

547 raw_argv = list(argv) 

548 self._argv = [ 

549 _bounded_text(str(argument), _STATUS_ARG_MAX_BYTES) 

550 for argument in raw_argv[:_STATUS_ARG_MAX_ITEMS] 

551 ] 

552 if len(raw_argv) > _STATUS_ARG_MAX_ITEMS: 

553 self._argv.append(f"...[truncated {len(raw_argv) - _STATUS_ARG_MAX_ITEMS} arguments]") 

554 self._pid = pid 

555 self._total_units = total_units 

556 self._enabled = task_status_enabled() 

557 

558 # Resolve the configured root once. Observability is optional: an 

559 # invalid home directory, inaccessible override, or resolution cycle 

560 # disables disk emission rather than preventing the tool from running. 

561 self._dir = Path() 

562 if self._enabled: 

563 try: 

564 self._dir = status_dir().expanduser().resolve(strict=False) 

565 except OSError, RuntimeError: 

566 self._enabled = False 

567 self._status_path = self._dir / f"{task_id}.json" 

568 self._log_path = self._dir / f"{task_id}.log" 

569 

570 self._started_at_iso = _now_iso() 

571 self._started_monotonic = time.monotonic() 

572 self._stacks_completed = 0 

573 self._last_stack: str | None = None 

574 self._last_message: str | None = None 

575 self._tail: deque[str] = deque(maxlen=_STATUS_TAIL_LINES) 

576 self._state = "running" 

577 self._exit_code: int | None = None 

578 self._error: str | None = None 

579 

580 self._lock = threading.Lock() 

581 self._last_write_ts = 0.0 

582 self._log_fp: IO[bytes] | None = None 

583 self._log_bytes_written = 0 

584 self._log_truncated = False 

585 

586 if self._enabled: 

587 try: 

588 with _TASK_ARTIFACT_LOCK: 

589 self._dir.mkdir(mode=stat.S_IRWXU, parents=True, exist_ok=True) 

590 with _suppress_oserror(): 

591 os.chmod(self._dir, stat.S_IRWXU) 

592 

593 # Establish the JSON anchor before creating a log. If the 

594 # initial status cannot be written, disable all disk output 

595 # so no untracked orphan log is left behind. 

596 if not self._write_status_now(): 

597 self._enabled = False 

598 else: 

599 try: 

600 fd = _open_private_log(self._log_path) 

601 try: 

602 self._log_fp = os.fdopen(fd, "wb", buffering=0) 

603 fd = -1 

604 finally: 

605 if fd >= 0: 

606 os.close(fd) 

607 except OSError: 

608 # A planted or unavailable log degrades to 

609 # status-only observability; never weaken no-follow. 

610 self._log_fp = None 

611 

612 # Prune after current creation so the retained count 

613 # converges to the configured bound, and sweep private 

614 # logs whose status anchor no longer exists. 

615 _prune_old_tasks( 

616 self._dir, 

617 _TASK_RETENTION, 

618 preserve_stem=self.task_id, 

619 ) 

620 except OSError, RuntimeError, ValueError: 

621 self._enabled = False 

622 if self._log_fp is not None: 

623 with _suppress_oserror(): 

624 self._log_fp.close() 

625 self._log_fp = None 

626 

627 # --- recording ------------------------------------------------------ 

628 

629 def set_pid(self, pid: int | None) -> None: 

630 """Attach the spawned process ID and publish it immediately.""" 

631 self._pid = pid 

632 if not self._enabled: 

633 return 

634 with self._lock: 

635 self._last_write_ts = time.monotonic() 

636 self._write_status_now() 

637 

638 def record_line(self, line: str, *, stream: str) -> None: 

639 """Append a single output line and refresh the status file. 

640 

641 ``stream`` is "stdout" or "stderr" — used as a prefix in the 

642 log file so readers can tell them apart in interleaved order. 

643 Status writes are debounced to avoid hundreds of fsyncs on 

644 noisy phases; the log file is unbuffered. 

645 """ 

646 if not self._enabled: 

647 return 

648 with self._lock: 

649 status_line = _bounded_text(str(line), _STATUS_LINE_MAX_BYTES) 

650 self._tail.append(status_line) 

651 self._last_message = status_line 

652 now = time.monotonic() 

653 if now - self._last_write_ts >= _STATUS_WRITE_DEBOUNCE_SECONDS: 

654 self._last_write_ts = now 

655 self._write_status_now() 

656 if self._log_fp is not None and not self._log_truncated: 

657 label = stream if stream in {"stdout", "stderr"} else "output" 

658 record = f"[{label}] {line}\n".encode("utf-8", errors="replace") 

659 if len(record) > _TASK_LOG_LINE_MAX_BYTES: 

660 marker = _TRUNCATED_TEXT.encode() + b"\n" 

661 record = record[: _TASK_LOG_LINE_MAX_BYTES - len(marker)] + marker 

662 remaining = _TASK_LOG_MAX_BYTES - self._log_bytes_written 

663 if len(record) <= remaining: 

664 try: 

665 written = self._log_fp.write(record) 

666 self._log_bytes_written += int(written or 0) 

667 except OSError: 

668 with _suppress_oserror(): 

669 self._log_fp.close() 

670 self._log_fp = None 

671 else: 

672 if len(_LOG_TRUNCATED_RECORD) <= remaining: 

673 with _suppress_oserror(): 

674 written = self._log_fp.write(_LOG_TRUNCATED_RECORD) 

675 self._log_bytes_written += int(written or 0) 

676 self._log_truncated = True 

677 

678 def increment_stacks(self, stack_name: str) -> None: 

679 """Record that one more stack finished. 

680 

681 Triggers an immediate (un-debounced) write so the 

682 ``stacks_completed`` counter is fresh for any reader polling 

683 between stack milestones. 

684 """ 

685 if not self._enabled: 

686 return 

687 with self._lock: 

688 self._stacks_completed += 1 

689 self._last_stack = stack_name 

690 self._last_write_ts = time.monotonic() 

691 self._write_status_now() 

692 

693 def set_last_stack(self, stack_name: str) -> None: 

694 """Update the last-seen stack name without bumping the counter.""" 

695 if not self._enabled: 

696 return 

697 with self._lock: 

698 self._last_stack = stack_name 

699 

700 def finish( 

701 self, 

702 *, 

703 state: str, 

704 exit_code: int | None = None, 

705 error: str | None = None, 

706 ) -> None: 

707 """Stamp a terminal state and flush. 

708 

709 ``state`` is one of "succeeded", "failed", "cancelled". Once 

710 finished, the status file is no longer touched — readers 

711 rely on the timestamp + state to know they have the final 

712 record. 

713 """ 

714 if not self._enabled: 

715 return 

716 with self._lock: 

717 self._state = state 

718 self._exit_code = exit_code 

719 self._error = error 

720 self._write_status_now() 

721 if self._log_fp is not None: 

722 with _suppress_oserror(): 

723 self._log_fp.flush() 

724 self._log_fp.close() 

725 self._log_fp = None 

726 

727 # --- internals ------------------------------------------------------ 

728 

729 def _build_payload(self) -> dict[str, Any]: 

730 elapsed = int(time.monotonic() - self._started_monotonic) 

731 payload: dict[str, Any] = { 

732 "task_id": self.task_id, 

733 "tool": self.tool, 

734 "argv": self._argv, 

735 "pid": self._pid, 

736 "started_at": self._started_at_iso, 

737 "updated_at": _now_iso(), 

738 "elapsed_seconds": elapsed, 

739 "state": self._state, 

740 "stacks_completed": self._stacks_completed, 

741 "last_stack": self._last_stack, 

742 "last_message": self._last_message, 

743 "tail": list(self._tail), 

744 "log_path": str(self._log_path), 

745 } 

746 if self._total_units is not None and self._total_units > 0: 

747 payload["stacks_total"] = self._total_units 

748 if self._exit_code is not None: 

749 payload["exit_code"] = self._exit_code 

750 if self._error is not None: 

751 payload["error"] = _bounded_text(self._error, _STATUS_LINE_MAX_BYTES) 

752 if self._log_truncated: 

753 payload["log_truncated"] = True 

754 return payload 

755 

756 def _write_status_now(self) -> bool: 

757 try: 

758 _atomic_write_json(self._status_path, self._build_payload()) 

759 except Exception: 

760 # Disk emission is best-effort — never let an unavailable status 

761 # surface crash or orphan the live tool invocation. 

762 return False 

763 return True 

764 

765 

766# --------------------------------------------------------------------------- 

767# Read-side helpers for the task_status / task_tail tools and the CLI. 

768# --------------------------------------------------------------------------- 

769 

770 

771def list_tasks(directory: Path | None = None) -> list[dict[str, Any]]: 

772 """Return all known task status records, newest first. 

773 

774 Each record gets ``is_alive`` re-computed from the recorded PID, 

775 and ``state`` is rewritten to ``"orphaned"`` when a record claims 

776 ``running`` but the PID is dead. This is the canonical way for 

777 callers to detect tasks whose MCP wrapper exited unexpectedly 

778 while the underlying CDK kept going. 

779 """ 

780 directory = directory or status_dir() 

781 if not directory.exists(): 

782 return [] 

783 records: list[dict[str, Any]] = [] 

784 for path in _sorted_regular_json_files(directory): 

785 record = _read_status_file(path) 

786 if record is not None: 

787 records.append(record) 

788 return records 

789 

790 

791def get_task(task_id: str, directory: Path | None = None) -> dict[str, Any] | None: 

792 """Return one task record by ID, with ``is_alive`` / orphan rewriting. 

793 

794 Invalid IDs, traversal attempts, symlinks, non-regular files, and missing 

795 records all fail closed as ``None`` so callers retain the established 

796 not-found contract without exposing host filesystem details. 

797 """ 

798 directory = directory or status_dir() 

799 fd = _open_task_artifact(task_id, ".json", directory) 

800 if fd is None: 

801 return None 

802 try: 

803 return _read_status_fd(fd) 

804 finally: 

805 os.close(fd) 

806 

807 

808def tail_log(task_id: str, lines: int = 100, directory: Path | None = None) -> list[str]: 

809 """Return the last ``lines`` lines of the task's raw log. 

810 

811 Empty list when the log file is missing, the task hasn't emitted 

812 anything yet, or the directory is unreadable. Lines do NOT include 

813 the trailing newline so callers don't have to strip them. 

814 """ 

815 if lines <= 0: 

816 return [] 

817 directory = directory or status_dir() 

818 fd = _open_task_artifact(task_id, ".log", directory) 

819 if fd is None: 

820 return [] 

821 try: 

822 requested = min(int(lines), _TASK_TAIL_MAX_LINES) 

823 size = os.fstat(fd).st_size 

824 start = max(0, size - _TASK_TAIL_MAX_BYTES) 

825 os.lseek(fd, start, os.SEEK_SET) 

826 remaining = min(size - start, _TASK_TAIL_MAX_BYTES) 

827 chunks: list[bytes] = [] 

828 while remaining > 0: 

829 chunk = os.read(fd, min(remaining, 64 * 1024)) 

830 if not chunk: 830 ↛ 831line 830 didn't jump to line 831 because the condition on line 830 was never true

831 break 

832 chunks.append(chunk) 

833 remaining -= len(chunk) 

834 data = b"".join(chunks) 

835 if start > 0: 

836 # Drop the first partial line. If the entire byte window is one 

837 # giant line, return its bounded suffix with an explicit marker. 

838 boundary = data.find(b"\n") 

839 if boundary >= 0: 

840 data = data[boundary + 1 :] 

841 elif data: 841 ↛ 847line 841 didn't jump to line 847 because the condition on line 841 was always true