Coverage for src/footman/_app.py: 92%

1"""The execution path: load tasks, refresh the manifest, run the chain. 

2 

3This is everything that happens for a real `fm ...` invocation (as opposed to 

4the completion hot path). It imports the user's tasks file — paying that cost is 

5fine here — resolves the command line against the freshly-built manifest, and 

6runs the resulting segments, honouring the global options. 

7""" 

8 

9from __future__ import annotations 

10 

11import contextlib 

12import json 

13import os 

14import shutil 

15import subprocess 

16import sys 

17import time 

18import tomllib 

19from collections.abc import Callable 

20from pathlib import Path 

21 

22from footman import ( 

23 _describe, 

24 _paths, 

25 _progress, 

26 config, 

27 context, 

28 discover, 

29 executor, 

30 manifest, 

31 registry, 

32 schedule, 

33 split, 

34) 

35from footman.app import DEFAULT_BRAND, Brand 

36from footman.split import Segment 

37 

38# The brand (names + version) in effect for the current invocation. Set at the 

39# top of `run()`; a CLI is one invocation per process, so a module global is 

40# the simplest way to reach it from the error/version helpers. The colour 

41# flags follow the same pattern: one per stream, resolved once from the 

42# stream's tty-ness, --no-color, NO_COLOR, and TERM. 

43_brand: Brand = DEFAULT_BRAND 

44_color_out: bool = False 

45_color_err: bool = False 

46 

47 

48_COLOR_MODES = ("auto", "always", "never") 

49 

50 

51def _resolve_color(g: dict[str, object], cfg: dict[str, object] | None = None) -> str: 

52 """The run-wide colour mode: `auto` | `always` | `never`. 

53 

54 Precedence, highest first: an explicit `--color=…` / `--no-color` on the 

55 command line, then `[tool.footman] color`, then the environment 

56 (`NO_COLOR` → never, `FORCE_COLOR` → always), else `auto`. An unrecognised 

57 value is ignored at that rung (validated loudly on the run path); `cfg` is 

58 absent for the pre-run colouring of `--version`/errors, present for the run. 

59 """ 

60 cli = g.get("color") 

61 if isinstance(cli, str) and cli in _COLOR_MODES: 

62 return cli 

63 if g.get("no_color"): 

64 return "never" 

65 if cfg is not None: 

66 cfg_color = cfg.get("color") 

67 if isinstance(cfg_color, str) and cfg_color in _COLOR_MODES: 

68 return cfg_color 

69 if "NO_COLOR" in os.environ: 

70 return "never" 

71 forced = os.environ.get("FORCE_COLOR") 

72 if forced not in (None, "", "0"): 

73 return "always" 

74 return "auto" 

75 

76 

77def _set_colors(mode: str) -> None: 

78 global _color_out, _color_err 

79 _color_out = _describe.wants_color(sys.stdout, mode) 

80 _color_err = _describe.wants_color(sys.stderr, mode) 

81 

82 

83def _error(message: str) -> None: 

84 prog = _describe.red(_brand.prog, _color_err) 

85 sys.stderr.write(f"{prog}: {message}\n") 

86 

87 

88def _refuse(json_mode: bool, message: str, code: int = 2) -> int: 

89 """Report a refusal on stderr — and when `--json` promised an envelope, 

90 keep stdout a single JSON document describing the same refusal, so a 

91 machine consumer never has to parse two formats.""" 

92 _error(message) 

93 if json_mode: 

94 envelope = {"schema": 1, "error": {"code": code, "message": message}} 

95 print(json.dumps({**envelope, "results": []}, indent=2)) 

96 return code 

97 

98 

99def _wants_json(argv: list[str]) -> bool: 

100 """Whether the leading globals include `--json`, tolerant of a malformed 

101 line — the refusal for `fm --json --nope` must still honour the envelope 

102 `--json` already promised. Mirrors `_parse_globals`' walk, minus raising. 

103 """ 

104 i = 0 

105 while i < len(argv) and argv[i].startswith("-") and argv[i] != "--": 

106 name = argv[i].split("=", 1)[0] 

107 if name == "--json": 

108 return True 

109 kind = split._GLOBAL_KIND.get(name) 

110 i += 1 

111 if kind == "option" and "=" not in argv[i - 1] and i < len(argv): 111 ↛ 112line 111 didn't jump to line 112 because the condition on line 111 was never true

112 i += 1 # skip the option's value 

113 elif ( 113 ↛ 119line 113 didn't jump to line 119 because the condition on line 113 was never true

114 kind == "option?" 

115 and "=" not in argv[i - 1] 

116 and i < len(argv) 

117 and not argv[i].startswith("-") 

118 ): 

119 i += 1 

120 return False 

121 

122 

123def _print_version(json_mode: bool) -> int: 

124 if json_mode: 

125 payload = {"schema": 1, "name": _brand.name, "version": _brand.version} 

126 print(json.dumps(payload, indent=2)) 

127 else: 

128 print(f"{_brand.name} {_brand.version}") 

129 return 0 

130 

131 

132def _globals_to_dict(tokens: list[str]) -> dict[str, object]: 

133 """Interpret the splitter's canonical global tokens into a flat mapping.""" 

134 result: dict[str, object] = {} 

135 i = 0 

136 while i < len(tokens): 

137 tok = tokens[i] 

138 name = tok.split("=", 1)[0] 

139 key = name.lstrip("-").replace("-", "_") 

140 if "=" in tok: # a value attached by the splitter (--name=value) 

141 result[key] = tok.split("=", 1)[1] 

142 i += 1 

143 elif split._GLOBAL_KIND.get(name) == "option": 

144 result[key] = tokens[i + 1] if i + 1 < len(tokens) else "" 

145 i += 2 

146 else: # a flag, or an option? given bare 

147 result[key] = True 

148 i += 1 

149 return result 

150 

151 

152def resolve_task_files( 

153 g: dict[str, object], 

154 *, 

155 on_warning: Callable[[str], None] | None = None, 

156 on_note: Callable[[str], None] | None = None, 

157) -> tuple[list[Path], dict[str, object]]: 

158 """The task files and merged config for the cwd + globals — the pure core of 

159 `_discover`, shared with the completion subprocess (`_suggest`) so both 

160 discover exactly the same tasks. 

161 

162 `-f/--tasks-file` loads exactly one file, no cascade; otherwise every 

163 `tasks.py` from the repo root down to the cwd. Raises `config.ConfigError` 

164 on a bad `--config`; an empty file list means nothing matched. The caller 

165 owns how either outcome is surfaced. 

166 """ 

167 cwd = Path.cwd() 

168 ceiling = _paths.find_repo_root(cwd) 

169 cfg = config.load_config( 

170 cwd, 

171 ceiling, 

172 g.get("config"), # type: ignore[arg-type] 

173 on_warning=on_warning, 

174 on_note=on_note, 

175 ) 

176 override = g.get("tasks_file") 

177 if override: 

178 one = Path(str(override)).expanduser() 

179 files = [one] if one.is_file() else [] 

180 else: 

181 filename = cfg.get("tasks") 

182 name = filename if isinstance(filename, str) else _brand.tasks_file 

183 files = _paths.task_files(cwd, ceiling, name) 

184 return files, cfg 

185 

186 

187def _discover( 

188 g: dict[str, object], wants_help: bool, bare: bool 

189) -> tuple[list[Path], dict[str, object]] | int: 

190 """Resolve the task files to load and the merged config for this cwd. 

191 

192 `-f/--tasks-file` is the escape hatch: it loads exactly one file, no 

193 cascade. Otherwise footman collects every `tasks.py` from the repo root 

194 (the `.git` ceiling) down to the cwd. Returns `(files, config)` or, when 

195 nothing was found, the exit code to return (0 for a listing, 2 otherwise). 

196 """ 

197 try: 

198 files, cfg = resolve_task_files( 

199 g, 

200 on_warning=_error, 

201 on_note=_error if g.get("verbose") else None, 

202 ) 

203 except config.ConfigError as exc: 

204 return _refuse(bool(g.get("json")), f"--config: {exc}") 

205 

206 if files: 

207 return files, cfg 

208 

209 looked = g.get("tasks_file") or cfg.get("tasks") or _brand.tasks_file 

210 if wants_help: 

211 # A stuck newcomer asking for help should see the globals (-f/-C are the 

212 # way out) — not a bare one-liner. Global help over an empty tree, then 

213 # the "where did I look" note. 

214 _print_global_help(manifest.build_manifest(registry.Group("root"))["tree"]) 

215 print(f"\n(no tasks file found — looked for {looked})") 

216 return 0 

217 if bare or g.get("list") or g.get("tree"): 

218 # A bare `fm` (like `--list`) is a warm empty state, not a hard error. 

219 if g.get("json"): # the catalog envelope, honestly empty 

220 tree = manifest.build_manifest(registry.Group("root"))["tree"] 

221 print(json.dumps({"schema": 1, "tree": tree}, indent=2)) 

222 else: 

223 print(f"No tasks file found (looked for {looked}).") 

224 return 0 

225 return _refuse( 

226 bool(g.get("json")), 

227 f"no tasks file found (looked for {looked}); " 

228 f"create one or pass -f/--tasks-file.", 

229 ) 

230 

231 

232# --- rendering --------------------------------------------------------------- 

233 

234 

235def _format_value(value: object) -> str: 

236 if value is True: 236 ↛ 237line 236 didn't jump to line 237 because the condition on line 236 was never true

237 return "" 

238 if isinstance(value, list): 238 ↛ 239line 238 didn't jump to line 239 because the condition on line 238 was never true

239 return "[" + ", ".join(str(v) for v in value) + "]" 

240 return str(value) 

241 

242 

243def _plan_line(seg: Segment) -> str: 

244 parts = [] 

245 for name, value in seg.values.items(): 

246 if value is True: 

247 parts.append(f"--{name}") 

248 elif value is False: 

249 parts.append(f"--no-{name}") 

250 else: 

251 parts.append(f"{name}={_format_value(value)}") 

252 if seg.variadic: 

253 parts.append("*" + " ".join(seg.variadic)) 

254 arrow = _describe.dim("->", _color_out) 

255 task = _describe.bold(seg.task, _color_out) 

256 line = f" {arrow} {task} " + " ".join(parts) 

257 if seg.passthrough is not None: 

258 line += _describe.dim(f" [-- {' '.join(seg.passthrough)}]", _color_out) 

259 return line.rstrip() 

260 

261 

262def _print_plan(globals_: list[str], segments: list[Segment]) -> None: 

263 if globals_: 263 ↛ 266line 263 didn't jump to line 266 because the condition on line 263 was always true

264 label = _describe.dim("globals:", _color_out) 

265 print(f" {label} {' '.join(globals_)}") 

266 for seg in segments: 

267 print(_plan_line(seg)) 

268 

269 

270def _print_footer() -> None: 

271 footer = f"Run `{_brand.prog} --help <task>` for a task's options." 

272 print(f"\n{_describe.dim(footer, _color_out)}") 

273 

274 

275def _styled_name(name: str, width: int) -> str: 

276 """A task name for a listing: dim group prefix, bold leaf, padded.""" 

277 pad = " " * (width - len(name)) 

278 prefix, _, leaf = name.rpartition(" ") 

279 lead = _describe.dim(f"{prefix} ", _color_out) if prefix else "" 

280 return f"{lead}{_describe.bold(leaf, _color_out)}{pad}" 

281 

282 

283def _styled_help(help_text: str) -> str: 

284 """A help line for a listing: trailing status notes dimmed.""" 

285 for marker in ("(runs until", "(unavailable:"): 

286 head, sep, note = help_text.partition(marker) 

287 if sep: 

288 return f"{head}{_describe.dim(f'{marker}{note}', _color_out)}" 

289 return help_text 

290 

291 

292def _print_list(tree: dict) -> None: 

293 rows = list(_describe.iter_tasks(tree)) 

294 if not rows: 

295 print("No tasks defined.") 

296 return 

297 width = max(len(name) for name, _ in rows) 

298 print(_describe.bold("Tasks:", _color_out)) 

299 for name, help_text in rows: 

300 line = f" {_styled_name(name, width)} {_styled_help(help_text)}" 

301 print(line.rstrip()) 

302 

303 

304def _print_tree(node: dict, indent: str = "") -> None: 

305 # Top-level empty tree (indent sentinel) → mirror _print_list rather than 

306 # printing zero bytes and exiting 0. 

307 if not indent and not node["tasks"] and not node["groups"]: 

308 print("No tasks defined.") 

309 return 

310 dash = _describe.dim("—", _color_out) 

311 for name, task in node["tasks"].items(): 

312 line = _describe.task_line(task) 

313 help_text = f" {dash} {_styled_help(line)}" if line else "" 

314 print(f"{indent}{_describe.bold(name, _color_out)}{help_text}") 

315 for name, sub in node["groups"].items(): 

316 label = f" {dash} {sub['help']}" if sub["help"] else "" 

317 print(f"{indent}{_describe.bold_cyan(f'{name}/', _color_out)}{label}") 

318 _print_tree(sub, indent + " ") 

319 

320 

321def _print_task_help(tree: dict, path: list[str]) -> None: 

322 # All phrasing (labels, details, examples) lives in `_describe`, shared 

323 # with the markdown exporter so help text and pages can never drift. 

324 node = tree 

325 for name in path[:-1]: 

326 node = node["groups"][name] 

327 task = node["tasks"][path[-1]] 

328 on = _color_out 

329 usage = _describe.paint_cli(_describe.usage_parts(_brand.prog, path, task), on) 

330 print(f"usage: {usage}") 

331 if task["help"]: 331 ↛ 333line 331 didn't jump to line 333 because the condition on line 331 was always true

332 print(f"\n {task['help']}") 

333 if task.get("long"): # the docstring's body, structure preserved 

334 body = "\n".join(f" {ln}".rstrip() for ln in task["long"].splitlines()) 

335 print(f"\n{body}") 

336 if task.get("infinite"): 336 ↛ 337line 336 didn't jump to line 337 because the condition on line 336 was never true

337 print(_describe.dim("\n runs until you stop it — Ctrl-C", on)) 

338 if task.get("disabled"): 

339 print(_describe.dim(f"\n unavailable here: {task['disabled']}", on)) 

340 positionals = [p for p in task["params"] if p["kind"] in ("argument", "variadic")] 

341 options = [p for p in task["params"] if p["kind"] in ("flag", "option")] 

342 for title, params in (("positionals", positionals), ("options", options)): 

343 if not params: 

344 continue 

345 rows = [] 

346 for p in params: 

347 doc, mech = _describe.param_detail_parts(p) 

348 # The author's words stay bright; the mechanics dim beneath them. 

349 mech = _describe.dim(mech, on) if mech else "" 

350 detail = "; ".join(bit for bit in (doc, mech) if bit) 

351 rows.append((_describe.param_label(p), detail)) 

352 width = max(len(label) for label, _ in rows) 

353 print(f"\n{_describe.bold(f'{title}:', on)}") 

354 for label, detail in rows: 

355 pad = " " * (width - len(label)) 

356 print(f" {_describe.bold(label, on)}{pad} {detail}".rstrip()) 

357 example = _describe.paint_cli(_describe.example_parts(path, task, _brand.prog), on) 

358 print(f"\n{_describe.dim('Example:', on)} {example}") 

359 if (shadows := task.get("shadows")) is not None: 

360 # This task overrides one further up the cascade — show the call 

361 # `inherited()` makes, so the forwarding line can be read off it. 

362 where = shadows.get("where") or "the cascade" 

363 print(_describe.dim(f"\nshadows {where} — inherited() calls it", on)) 

364 usage = _describe.paint_cli( 

365 _describe.usage_parts(_brand.prog, path, shadows), on 

366 ) 

367 print(f" {usage}") 

368 

369 

370def _print_group_help(tree: dict, path: list[str]) -> None: 

371 node = tree 

372 for name in path: 

373 node = node["groups"][name] 

374 on = _color_out 

375 default = node.get("default") 

376 parts = [("prog", _brand.prog), *[("group", name) for name in path]] 

377 # A runnable group (one with `@group.default`) can run bare — its default — 

378 # so the task becomes optional. 

379 parts += [("opt", "[<task>]")] if default else [("req", "<task>")] 

380 parts += [("opt", "[options]")] 

381 print(f"usage: {_describe.paint_cli(parts, on)}") 

382 if node["help"]: 382 ↛ 384line 382 didn't jump to line 384 because the condition on line 382 was always true

383 print(f"\n {node['help']}") 

384 if default: 384 ↛ 385line 384 didn't jump to line 385 because the condition on line 384 was never true

385 print(_describe.dim("\n runs its default when no task is named", on)) 

386 rows = list(_describe.iter_tasks(node)) 

387 if rows: 387 ↛ 393line 387 didn't jump to line 393 because the condition on line 387 was always true

388 width = max(len(name) for name, _ in rows) 

389 print(f"\n{_describe.bold('tasks:', on)}") 

390 for name, help_text in rows: 

391 line = f" {_styled_name(name, width)} {_styled_help(help_text)}" 

392 print(line.rstrip()) 

393 params = default["params"] if default else [] 

394 options = [p for p in params if p["kind"] in ("flag", "option")] 

395 if options: 395 ↛ 396line 395 didn't jump to line 396 because the condition on line 395 was never true

396 rows2 = [] 

397 for p in options: 

398 doc, mech = _describe.param_detail_parts(p) 

399 mech = _describe.dim(mech, on) if mech else "" 

400 detail = "; ".join(bit for bit in (doc, mech) if bit) 

401 rows2.append((_describe.param_label(p), detail)) 

402 width2 = max(len(label) for label, _ in rows2) 

403 print(f"\n{_describe.bold('options:', on)}") 

404 for label, detail in rows2: 

405 pad = " " * (width2 - len(label)) 

406 print(f" {_describe.bold(label, on)}{pad} {detail}".rstrip()) 

407 

408 

409def _print_global_help(tree: dict) -> None: 

410 prog = _brand.prog 

411 parts = [ 

412 ("prog", prog), 

413 ("opt", "[globals]"), 

414 ("req", "<task>"), 

415 ("opt", "[options]"), 

416 ("opt", "[<task> ...]"), 

417 ] 

418 print(f"usage: {_describe.paint_cli(parts, _color_out)}") 

419 print(f"\n{_describe.bold('globals (before the first task):', _color_out)}") 

420 rows = [] 

421 for name, alias, _kind, hint, help_text in split.GLOBALS: 

422 label = f"{alias}, {name}" if alias else f" {name}" 

423 if hint: 

424 label += f" {hint}" 

425 # `.replace` (not `.format`) so a help string containing braces can 

426 # never crash help output. 

427 rows.append((label, help_text.replace("{prog}", prog))) 

428 width = max(len(label) for label, _ in rows) 

429 for label, help_text in rows: 

430 pad = " " * (width - len(label)) 

431 print(f" {_describe.bold(label, _color_out)}{pad} {help_text}") 

432 print() 

433 _print_list(tree) 

434 _print_footer() 

435 

436 

437def _wants_help(argv: list[str]) -> bool: 

438 """`-h`/`--help` anywhere before `--` turns the whole line into a help 

439 request — asking for help must never execute anything, wherever it lands 

440 on the line. After `--` it belongs to the passthrough.""" 

441 for tok in argv: 

442 if tok == "--": 

443 return False 

444 if tok in ("-h", "--help"): 

445 return True 

446 return False 

447 

448 

449def _expand_help_path(argv: list[str], start: int) -> list[str]: 

450 """Split a quoted multi-word or dotted path token into its components. 

451 

452 `fm --help "dict add"` (the shell hands the path as one token) and 

453 `fm --help dict.add` (a user dots the path) both name the task at 

454 `dict` → `add`. No group or task name contains a space or a dot, so 

455 splitting a path token on either only ever un-quotes a path the walk can 

456 then resolve — turning a self-referential "did you mean 'dict add'?" into a 

457 real lookup. Option-shaped tokens and everything past `--` are left verbatim. 

458 """ 

459 out = argv[:start] 

460 for j in range(start, len(argv)): 

461 tok = argv[j] 

462 if tok == "--": 

463 out.extend(argv[j:]) # passthrough boundary — leave the rest verbatim 

464 break 

465 if not tok.startswith("-") and (" " in tok or "." in tok): 

466 out.extend(tok.replace(".", " ").split()) 

467 else: 

468 out.append(tok) 

469 return out 

470 

471 

472def _help_targets( 

473 tree: dict, argv: list[str] 

474) -> tuple[list[tuple[str, list[str]]], list[str]]: 

475 """Group/task paths mentioned on a `--help` line, resolved leniently — 

476 plus the bare words that resolved to nothing, so the caller can refuse a 

477 `--help typo` instead of shrugging. 

478 

479 The real splitter enforces arity — `--help add` must work even though 

480 `add` alone would be "missing required argument(s)" — so this walks group 

481 and task names only and skips every other token (option-shaped tokens and, 

482 once a target is found, its argument values). 

483 """ 

484 _, i = split._parse_globals(argv, 0) 

485 argv = _expand_help_path(argv, i) 

486 targets: list[tuple[str, list[str]]] = [] 

487 strays: list[str] = [] 

488 while i < len(argv): 

489 if argv[i] == "--": 

490 break 

491 node, path = tree, [] 

492 while i < len(argv) and argv[i] in node["groups"]: 

493 path.append(argv[i]) 

494 node = node["groups"][argv[i]] 

495 i += 1 

496 if i < len(argv) and argv[i] in node["tasks"]: 

497 targets.append(("task", [*path, argv[i]])) 

498 elif path: 

499 targets.append(("group", path)) 

500 continue # the walk already consumed the group name(s) 

501 elif i < len(argv) and not argv[i].startswith("-"): 

502 strays.append(argv[i]) 

503 i += 1 

504 return targets, strays 

505 

506 

507def _print_help(tree: dict, argv: list[str]) -> int: 

508 """`--help` alone covers fm itself; with names, the named groups/tasks. 

509 

510 A name that matches nothing is a refusal (exit 2) with a suggestion — 

511 silently degrading to the global listing looked like an answer while 

512 teaching nothing. With at least one real target found, extra bare words 

513 stay lenient: they are argument values, not typos. 

514 """ 

515 targets, strays = _help_targets(tree, argv) 

516 if not targets: 

517 if strays: 

518 known = [name for name, _ in _describe.iter_tasks(tree)] 

519 known += _describe.iter_group_paths(tree) 

520 # Help's *success* output is the one human-only surface; a refusal 

521 # still honours the envelope `--json` promised. 

522 return _refuse( 

523 _wants_json(argv), 

524 f"--help: unknown task or group {strays[0]!r}" 

525 f"{split._did_you_mean(strays[0], known)}", 

526 ) 

527 _print_global_help(tree) 

528 return 0 

529 for index, (kind, path) in enumerate(targets): 

530 if index: 530 ↛ 531line 530 didn't jump to line 531 because the condition on line 530 was never true

531 print() 

532 if kind == "task": 

533 _print_task_help(tree, path) 

534 else: 

535 _print_group_help(tree, path) 

536 return 0 

537 

538 

539def _where(root: registry.Group, tree: dict, dotted: str) -> int: 

540 path = dotted.replace(".", " ").split() 

541 try: 

542 fn = executor.resolve(root, path) 

543 except (KeyError, IndexError): 

544 names = [name.replace(" ", ".") for name, _ in _describe.iter_tasks(tree)] 

545 _error(f"--where: unknown task {dotted!r}{split._did_you_mean(dotted, names)}") 

546 return 2 

547 chain = discover.shadow_chain(fn) 

548 lines = [] 

549 for index, member in enumerate(chain): 

550 code = getattr(member, "__code__", None) 

551 if code is None: 551 ↛ 552line 551 didn't jump to line 552 because the condition on line 551 was never true

552 continue 

553 where = f"{code.co_filename}:{code.co_firstlineno}" 

554 # The winner first; anything it shadows follows, marked — so 

555 # "am I overriding something, and where is it?" is one command. 

556 lines.append(where if index == 0 else f"{where} (shadowed)") 

557 if not lines: 557 ↛ 558line 557 didn't jump to line 558 because the condition on line 557 was never true

558 _error(f"--where: cannot locate source for {dotted!r}") 

559 return 2 

560 print("\n".join(lines)) 

561 return 0 

562 

563 

564def _print_summary( 

565 results: list[executor.TaskResult], 

566 *, 

567 timings: bool, 

568 total: float, 

569) -> None: 

570 # The summary is commentary about the run, not the run's output — it goes 

571 # to stderr so `fm task > file` captures exactly what the task produced. 

572 # Each receipt is task-shaped (mark · name · time), the same grid as the 

573 # step lines above it, with the name in cyan — same family, one rank up. 

574 color = _color_err 

575 width = max((len(r.task) for r in results), default=0) 

576 for result in results: 

577 ok = result.ok 

578 cancelled = result.cancelled 

579 if color: 

580 if ok: 580 ↛ 582line 580 didn't jump to line 582 because the condition on line 580 was always true

581 mark = "\033[32m✓\033[0m" 

582 elif cancelled: 

583 mark = "\033[33m○\033[0m" # cut off by fail-fast, not a failure 

584 else: 

585 mark = "\033[31m✗\033[0m" 

586 name = f"\033[1;36m{result.task:<{width}}\033[0m" 

587 else: 

588 word = "ok" if ok else ("cut" if cancelled else "FAIL") 

589 mark = f"{word:<4}" 

590 name = f"{result.task:<{width}}" 

591 timing = ( 

592 f"({result.duration * 1000:.0f} ms)" 

593 if timings 

594 else f"({_progress.fmt_secs(result.duration)})" 

595 ) 

596 if color: 

597 timing = f"\033[36m{timing}\033[0m" 

598 print(f"{mark} {name} {timing}", file=sys.stderr) 

599 if cancelled: 599 ↛ 600line 599 didn't jump to line 600 because the condition on line 599 was never true

600 _error(f"{result.task}: cancelled — fail-fast stopped the run") 

601 elif result.error is not None: 

602 # A deliberate stop with a reason (`fail("…")` / `sys.exit("…")`) 

603 # renders verbatim, the way Python prints it — not "Failed: …"; a bare 

604 # `fail()` with no reason falls back to the code line. A real exception 

605 # keeps its type, which signals a crash, not a chosen stop. 

606 err = result.error 

607 if context._is_deliberate_stop(err): 

608 detail = str(err) or f"exited with code {result.code}" 

609 else: 

610 detail = f"{type(err).__name__}: {err}" 

611 _error(f"{result.task}: {detail}") 

612 elif not result.ok: 

613 _error(f"{result.task}: exited with code {result.code}") 

614 if len(results) > 1: # one task's receipt already carries the total 

615 took = f"took {_progress.fmt_secs(total)}" 

616 if color: 

617 took = f"\033[2m{took}\033[0m" 

618 print(took, file=sys.stderr) 

619 

620 

621def _print_json(results: list[executor.TaskResult], *, total: float) -> None: 

622 payload = [] 

623 for r in results: 

624 entry: dict[str, object] = { 

625 "task": r.task, 

626 "ok": r.ok, 

627 "cancelled": r.cancelled, 

628 "code": r.code, 

629 "duration_ms": round(r.duration * 1000, 3), 

630 "output": r.output, 

631 "steps": [ 

632 { 

633 "command": s.command, 

634 "code": s.code, 

635 "duration_ms": round(s.duration * 1000, 3), 

636 "stdout": s.stdout, 

637 "stderr": s.stderr, 

638 } 

639 for s in r.steps 

640 ], 

641 "error": None if r.error is None else str(r.error), 

642 } 

643 value = r.returned 

644 # An int return is the exit-code channel (duty's contract), not data; 

645 # None is "nothing to say". Everything else — bools included — is data. 

646 if value is not None and not ( 

647 isinstance(value, int) and not isinstance(value, bool) 

648 ): 

649 try: 

650 json.dumps(value, default=_describe.json_default) 

651 except (TypeError, ValueError) as exc: # ValueError: circular refs 

652 entry["returned_error"] = str(exc) 

653 _error(f"{r.task}: --json: return value dropped — {exc}") 

654 else: 

655 entry["returned"] = value 

656 payload.append(entry) 

657 # The stable machine surface: an envelope so post-1.0 additions (metadata, 

658 # summaries) never have to break consumers of the results list. 

659 print( 

660 json.dumps( 

661 {"schema": 1, "total_ms": round(total * 1000, 3), "results": payload}, 

662 indent=2, 

663 default=_describe.json_default, 

664 ) 

665 ) 

666 

667 

668def _resolve_shell(shell: object, flag: str) -> str | None: 

669 """Resolve *shell* to a supported name for *flag*, or None after `_error`. 

670 

671 A bare flag (`shell is True`) detects the invoking shell; an explicit value 

672 is lowercased and de-aliased (`nu`→`nushell`, `powershell`→`pwsh`). 

673 """ 

674 from footman import _shellcomp 

675 

676 supported = "|".join(_shellcomp.SHELLS) 

677 if shell is True: 

678 name = _shellcomp.detect_shell() 

679 if name is None: 

680 _error( 

681 f"{flag}: could not detect your shell — " 

682 f"name it explicitly: one of {supported}" 

683 ) 

684 return None