Coverage for src/footman/_app.py: 94%
1220 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-06 02:59 +0000
1"""The execution path: load tasks, refresh the manifest, run the chain.
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"""
9from __future__ import annotations
11import contextlib
12import json
13import os
14import subprocess
15import sys
16import time
17from collections.abc import Callable, Iterator, Sequence
18from pathlib import Path
19from typing import Any, NamedTuple
21from footman import (
22 _coerce,
23 _config,
24 _describe,
25 _discover,
26 _executor,
27 _manifest,
28 _notes,
29 _paths,
30 _progress,
31 _schedule,
32 _script,
33 _signals,
34 _split,
35 context,
36 invocation,
37 registry,
38)
39from footman._executor import EX_USAGE
40from footman.app import DEFAULT_BRAND, Brand
42# The brand (names + version) in effect for the current invocation. Set at the
43# top of `run()`; a CLI is one invocation per process, so a module global is
44# the simplest way to reach it from the error/version helpers. The colour
45# flags follow the same pattern: one per stream, resolved once from the
46# stream's tty-ness, --no-color, NO_COLOR, and TERM.
47_brand: Brand = DEFAULT_BRAND
48_color_out: bool = False
49_docs_url: str | None = None # the run's docs_url template, config-set
52def _builtin() -> tuple[str, ...]:
53 """This invocation's built-in set: the brand's, plus the user's own.
55 Every consumer of the built-in ladder asks here rather than reading
56 `_brand.builtin` — global mode, the base mount, `--plugins`, the
57 "it's built in, mount it" remedy — so a user's declarations are built
58 in wherever the brand's are, which is the whole of the feature.
59 A broken `builtin` key is refused where it can be reported (the
60 execution path); the quieter surfaces fall back to the brand's own.
61 """
62 try:
63 return _config.effective_builtin(_brand.builtin)
64 except _config.ConfigError:
65 return _brand.builtin
68def _task_docs_url(address: str) -> str | None:
69 """*address*'s documentation URL under this run's template, or None."""
70 return _describe.docs_url_for(_docs_url, address)
73_color_err: bool = False
76_COLOR_MODES = ("auto", "always", "never")
79def _resolve_color(g: dict[str, object], *, bound: bool = False) -> str:
80 """The run-wide colour mode: `auto` | `always` | `never`.
82 Precedence, highest first: an explicit `--color=…` / `--no-color` on the
83 command line, then the option's own ladder — the `color` config key, then
84 the declared computed default, which reads the ambient protocol
85 variables (`NO_COLOR` → never, `FORCE_COLOR` → always), else `auto`.
86 *bound* is False for the pre-run colouring of `--version`/errors, where
87 no config exists yet and the declared default answers directly; the run
88 path re-resolves bound and repaints.
89 """
90 cli = g.get("color")
91 if isinstance(cli, str) and cli in _COLOR_MODES:
92 return cli
93 if g.get("no_color"):
94 return "never"
95 if bound:
96 return str(_split.COLOR.value)
97 return _split._color_from_env()
100def _set_colors(mode: str) -> None:
101 global _color_out, _color_err
102 _color_out = _describe.wants_color(sys.stdout, mode)
103 _color_err = _describe.wants_color(sys.stderr, mode)
106def _error(message: str) -> None:
107 prog = _describe.red(_brand.prog, _color_err)
108 sys.stderr.write(f"{prog}: {message}\n")
111def _refuse(json_mode: bool, message: str, code: int = EX_USAGE) -> int:
112 """Report a refusal on stderr — and when `--json` promised an envelope,
113 keep stdout a single JSON document describing the same refusal, so a
114 machine consumer never has to parse two formats. Refusals exit
115 `EX_USAGE` (64), never a code a task could mean on purpose; the
116 non-refusal callers pass 130 for an interrupt and 143/129 for a stop
117 signal."""
118 _error(message)
119 if json_mode:
120 envelope = {"schema": 1, "error": {"code": code, "message": message}}
121 print(json.dumps({**envelope, "items": []}, indent=2))
122 return code
125def _wants_json(argv: list[str]) -> bool:
126 """Whether the leading globals include `--json`, tolerant of a malformed
127 line — the refusal for `fm --json --nope` must still honour the envelope
128 `--json` already promised. Mirrors `_parse_globals`' walk, minus raising.
129 """
130 i = 0
131 while i < len(argv) and argv[i].startswith("-") and argv[i] != "--":
132 if argv[i].split("=", 1)[0] == "--json":
133 return True
134 i += 1 # every global is one self-contained token (values `=`-attach)
135 return False
138def _print_version(json_mode: bool) -> int:
139 if json_mode:
140 payload = {"schema": 1, "name": _brand.name, "version": _brand.version}
141 print(json.dumps(payload, indent=2))
142 else:
143 print(f"{_brand.name} {_brand.version}")
144 return 0
147def _globals_to_dict(tokens: list[str]) -> dict[str, object]:
148 """Interpret the splitter's canonical global tokens into a flat mapping."""
149 result: dict[str, object] = {}
150 for tok in tokens:
151 name = tok.split("=", 1)[0]
152 if not name.startswith("--") and name not in _split._GLOBAL_KIND:
153 # A lenient pre-discovery walk carries unknown dash tokens
154 # through for the authoritative parse to teach — but stripping
155 # the dashes here quietly made the misspelling ACT first:
156 # `-version=1` drove `--version` and exited 0, and
157 # `-install-completion=zsh` would have edited a shell rc, all
158 # before the parse that refuses `-version` ever ran. An unknown
159 # spelling populates nothing; the refusal stays the answer.
160 continue
161 key = name.lstrip("-").replace("-", "_")
162 if "=" in tok: # a value attached by the splitter (--name=value)
163 result[key] = tok.split("=", 1)[1]
164 elif _split._GLOBAL_KIND.get(name) == "flag":
165 result[key] = True
166 else:
167 # A value-taking global named bare. It carries presence and no
168 # value, so it reads as the empty string and the consumer resolves
169 # what that means — `--describe` the whole tree, `--color` the
170 # ambient decision. It used to read as `True`, which made every
171 # consumer test the *type* of the value to learn how it was
172 # spelled; that union is what this removes.
173 result[key] = ""
174 return result
177def _uv_wanted(g: dict[str, object], cfg: dict[str, Any]) -> bool:
178 """`--uv` / `--no-uv` / the `uv` config key, resolved lexically:
179 **CLI > config > on by default**.
181 The one config-backed option read before any bind: the uv handoff runs
182 first and may replace this process, so `_split.UV`'s declared ladder
183 cannot answer yet. Same order, hand-held — the stated early consumer.
184 """
185 if g.get("uv"): 185 ↛ 186line 185 didn't jump to line 186 because the condition on line 185 was never true
186 return True
187 if g.get("no_uv"): 187 ↛ 188line 187 didn't jump to line 188 because the condition on line 187 was never true
188 return False
189 value = cfg.get("uv")
190 return True if not isinstance(value, bool) else value
193def _config_arg(g: dict[str, object]) -> str | None:
194 """The --config global as the string it is (or None) — the globals dict
195 is object-valued, and this is the one place that narrows it."""
196 value = g.get("config")
197 return value if isinstance(value, str) else None
200class Discovery(NamedTuple):
201 """What one discovery walk found — resolved once, shared verbatim by the
202 run, the handoff probe, and the completion child, so they cannot
203 disagree about what loads."""
205 files: list[Path]
206 """Everything `load_tree` mounts, outermost rung first: the user tasks
207 file when present, then the project cascade root-down. Nearest wins on
208 a name, so a project task shadows a same-named user task — the reading
209 the cascade always had, extended one rung outward."""
210 cfg: dict[str, object]
211 root: str
212 """The project cascade's top — and `""` outside a project. The user
213 rung never claims it: a root is the *project's* top, and footman
214 invents none where there is no project ("empty means global mode")."""
215 user: Path | None
216 """The user tasks file, when it joined `files`."""
219def resolve_task_files(
220 g: dict[str, object],
221 *,
222 on_warning: Callable[[str], None] | None = None,
223 on_note: Callable[[str], None] | None = None,
224 cwd: Path | None = None,
225) -> Discovery:
226 """The task files, merged config, and project root for the cwd + globals
227 — the pure core of `_discover_files`, shared with the completion
228 subprocess (`_suggest`) so both discover exactly the same tasks.
230 `-f/--tasks-file` loads exactly one file, no cascade and no user rung
231 (total control); otherwise the user tasks file leads as the cascade's
232 outermost rung, then every `tasks.py` along the walk down to the cwd.
233 The walk's reach is the cascade mode (user-level `cascade` key,
234 `FOOTMAN_CASCADE` override): the cwd alone (`none`), the repo root
235 (`repo`, default), or the whole ancestor path (`filesystem`) — and
236 config search follows the same walk, so the two cascades stay one
237 concept. Raises `_config.ConfigError` on a bad `--config` or an unknown
238 cascade mode (`_config.CascadeError`); an empty `files` means nothing
239 matched. The caller owns how either outcome is surfaced.
241 *cwd* answers from somewhere other than the process directory: the
242 handoff resolves against its `-C` probe *before* the chdir happens, and
243 must see exactly what the run will.
244 """
245 cwd = cwd or Path.cwd()
246 mode = _config.cascade_mode(_config_arg(g))
247 if mode == "none":
248 ceiling = cwd
249 elif mode == "filesystem":
250 ceiling = Path(cwd.anchor)
251 else:
252 ceiling = _paths.find_repo_root(cwd)
253 cfg = _config.load_config(
254 cwd,
255 ceiling,
256 _config_arg(g),
257 on_warning=on_warning,
258 on_note=on_note,
259 )
260 override = g.get("tasks_file")
261 if override:
262 one = Path(str(override)).expanduser()
263 if not one.is_absolute():
264 one = cwd / one # identical to the plain relative read when cwd is the cwd
265 files = [one] if one.is_file() else []
266 return Discovery(files, cfg, str(files[0].parent) if files else "", None)
267 filename = cfg.get("tasks")
268 if filename is not None and not isinstance(filename, str):
269 # The option-backed keys refuse a wrong TOML type loudly; this
270 # config-only key silently fell back to the default instead —
271 # `tasks = 123` behaved as if unset, which reads as footman
272 # ignoring the setting rather than refusing it.
273 raise _config.ConfigError(
274 f"config key 'tasks' expects a filename string (got {filename!r})"
275 )
276 name = filename if isinstance(filename, str) else _brand.tasks_file
277 miscased: list[tuple[Path, str]] = []
278 files = _paths.task_files(cwd, ceiling, name, miscased)
279 if on_warning is not None:
280 for directory, actual in miscased:
281 # Unconditional: a wrong-case file is worth saying whether or not
282 # the walk found a real one elsewhere. The person wrote a tasks
283 # file and footman is not loading it — silence is the bug, and
284 # the builtin base answering in its place is exactly when the
285 # silence is most convincing.
286 on_warning(
287 f"{directory / actual} is not {name} — the name is "
288 f"case-sensitive, so this file is not being loaded. "
289 f"Rename it to {name}."
290 )
291 root = str(files[0].parent) if files else ""
292 if _config_arg(g):
293 # An explicit --config is total control, the user rung included:
294 # the user named exactly what applies.
295 user_name = name
296 else:
297 # The user rung's own name comes from the user's own writing — the
298 # user-level file — never from a project's `tasks` key, which
299 # renames the *project's* file and stops there. Steered by the
300 # project, the walk looked for a personal file the user never
301 # wrote, and the personal rung silently vanished under any project
302 # that renames its tasks file.
303 setting = _config.user_level_value("tasks")
304 user_name = setting if isinstance(setting, str) else _brand.tasks_file
305 user = _paths.user_tasks_file(user_name)
306 if user.is_file():
307 # The cascade's outermost rung: personal tasks ride everywhere, and
308 # anything nearer shadows them — project > user, the nearest-wins
309 # reading the cascade already has, extended one rung outward. A
310 # project that wants a personal task's name owns it; `inherited()`
311 # still reaches what it shadowed.
312 return Discovery([user, *files], cfg, root, user)
313 return Discovery(files, cfg, root, None)
316def _base_tree(names: tuple[str, ...], json_mode: bool) -> registry.Group | int:
317 """The built-in base: each `footman.tasks` entry point mounted into a
318 fresh tree, exactly as a tasks file's `plugin(...)` would mount it — so
319 a project that wants the same set mounts it the ordinary way.
321 Built only when discovery found no project task files. A name that does
322 not mount is a refusal naming *whoever declared it* — the brand for its
323 own, the user's config for theirs, since that is whose install or
324 spelling is what needs fixing — never a crash."""
325 from footman import compose
327 declared_by_user = set(_config.user_builtin())
328 with registry.capture() as base:
329 for name in names:
330 try:
331 compose.plugin(name)
332 except Exception as exc:
333 if name in declared_by_user and name not in _brand.builtin:
334 return _refuse(
335 json_mode,
336 f"builtins.user in {_paths.footman_config_file()} "
337 f"names {name!r}, which did not mount: {exc}",
338 )
339 if name not in _brand.builtin:
340 # Discovery put it there, and discovery can be wrong
341 # about a package that has since been removed by hand.
342 # A note and carry on, never a refusal: this list is a
343 # record footman writes and can always rebuild, which
344 # is already why a missing or malformed one means
345 # "nothing discovered" instead of an error. An entry
346 # that no longer resolves is the same thing one entry
347 # at a time. Refusing also made the remedy the message
348 # names unreachable — `self.add` mounts this same base
349 # on its way in, so the way out refused too.
350 _error(
351 f"a discovered built-in, {name!r}, did not mount: "
352 f"{exc} — skipping it; run {_brand.prog} self.add to "
353 f"bring the discovered list back in step with what "
354 f"is installed"
355 )
356 continue
357 return _refuse(
358 json_mode,
359 f"{_brand.name} declares built-in tasks from {name!r}, "
360 f"which did not mount: {exc}",
361 )
362 # A built-in was defined by no tasks file, so it must carry no folder —
363 # and the stamp lives on the function, which is the same object every
364 # time it is mounted. Without this, an earlier in-process invocation that
365 # mounted the same provider from inside a project leaves that project's
366 # directory on it, and nothing here overwrites it: the base exists only
367 # when discovery found no task files, so no overlay ever runs.
368 _discover.untag(base)
369 # Before the cascade overlays the user's own tasks into this group: after
370 # that, nothing tells the brand's tasks from the person's, and the two
371 # have opposite defaults.
372 registry.seal_expose(base)
373 return base
376def _discover_files(
377 g: dict[str, object], wants_help: bool, bare: bool
378) -> Discovery | int:
379 """Resolve the task files to load and the merged config for this cwd.
381 `-f/--tasks-file` is the escape hatch: it loads exactly one file, no
382 cascade. Otherwise the user rung leads and footman collects every
383 `tasks.py` from the repo root (the `.git` ceiling) down to the cwd.
384 Returns the `Discovery` or, when nothing was found, the exit code to
385 return (0 for a listing, 2 otherwise).
386 """
387 try:
388 found = resolve_task_files(
389 g,
390 on_warning=_error,
391 on_note=_error if g.get("verbose") else None,
392 )
393 except _config.CascadeError as exc:
394 # Self-describing (names FOOTMAN_CASCADE / the `cascade` key): no
395 # `--config:` prefix, which would misattribute an env-var mistake.
396 return _refuse(bool(g.get("json")), str(exc))
397 except _config.ConfigError as exc:
398 return _refuse(bool(g.get("json")), f"--config: {exc}")
400 if found.files:
401 return found
403 if _builtin() and not g.get("tasks_file"):
404 # Global mode: the brand's built-ins answer where nothing else does,
405 # so listings, help, and runs proceed over the mounted base. The
406 # empty-tree teaching below is for a runner with no base at all.
407 # (`-f` names a file that wasn't there — total control includes the
408 # miss, so it keeps today's teaching rather than a surprise base.)
409 return found
411 cfg = found.cfg
412 looked = g.get("tasks_file") or cfg.get("tasks") or _brand.tasks_file
413 # Name the file exactly, and say the name is case-sensitive. That is
414 # the whole of what a `Tasks.py` owner needs and it costs nothing to
415 # say always: on a filesystem that folds case the walk already
416 # complained by name (it had the listing in hand), and on one that does
417 # not, this is the only affordable way to teach the rule — finding the
418 # variant there would mean listing every level of the chain.
419 looked_at = f"{looked} (exactly — the name is case-sensitive)"
420 if wants_help:
421 # A stuck newcomer asking for help should see the globals (-f/-C are the
422 # way out) — not a bare one-liner. Global help over an empty tree, then
423 # the "where did I look" note.
424 _print_global_help(_manifest.build_manifest(registry.Group("root"))["tree"])
425 print(f"\n(no tasks file found — looked for {looked_at})")
426 return 0
427 if bare or g.get("list") or g.get("tree"):
428 # A bare `fm` (like `--list`) is a warm empty state, not a hard error.
429 if g.get("json"): # the catalog envelope, honestly empty
430 tree = _manifest.build_manifest(registry.Group("root"))["tree"]
431 print(json.dumps({"schema": 1, "tree": tree}, indent=2))
432 else:
433 print(f"No tasks file found (looked for {looked_at}).")
434 return 0
435 return _refuse(
436 bool(g.get("json")),
437 f"no tasks file found (looked for {looked_at}); "
438 f"create one or pass -f/--tasks-file.",
439 )
442# --- rendering ---------------------------------------------------------------
445def _print_footer() -> None:
446 footer = f"Run `{_brand.prog} --help <task>` for a task's options."
447 print(f"\n{_describe.dim(footer, _color_out)}")
450def _styled_name(name: str) -> str:
451 """A task address for a listing: dim group prefix, bold leaf — and a
452 terminal hyperlink to its docs page when the project configured a
453 `docs_url` template (zero width, so the band math never notices)."""
454 prefix, _, leaf = name.rpartition(".")
455 lead = _describe.dim(f"{prefix}.", _color_out) if prefix else ""
456 styled = f"{lead}{_describe.bold(leaf, _color_out)}"
457 return _describe.link(styled, _task_docs_url(name), _color_out)
460def _styled_help(help_text: str) -> str:
461 """A help line for a listing: trailing status notes dimmed."""
462 for marker in ("(runs until", "(unavailable:"):
463 head, sep, note = help_text.partition(marker)
464 if sep:
465 return f"{head}{_describe.dim(f'{marker}{note}', _color_out)}"
466 return help_text
469def _print_two_band(rows: list[tuple[str, int, str]]) -> None:
470 """The one two-band layout every listing draws: painted name cells on
471 the left, descriptions aligned into a single wrapped column on the
472 right. Rows are `(cell, cell's plain width, help)` — `--list`, `--tree`
473 and group help differ only in how they paint the cell.
475 Long descriptions wrap with a hanging indent to the description
476 column — the terminal's own wrap would drop continuations to column 0,
477 shearing the two-band layout apart.
478 """
479 import textwrap
481 width = max(plain for _, plain, _ in rows)
482 desc_col = width + 2
483 import shutil # ~1.7 ms of archive codecs for one terminal width
485 avail = max(24, shutil.get_terminal_size().columns - desc_col)
486 for cell, plain, help_text in rows:
487 if not help_text:
488 print(cell.rstrip())
489 continue
490 pieces = textwrap.wrap(help_text, avail) or [""]
491 pad = " " * (width - plain)
492 print(f"{cell}{pad} {_styled_help(pieces[0])}".rstrip())
493 for cont in pieces[1:]:
494 print(f"{' ' * desc_col}{_styled_help(cont)}".rstrip())
497def _address_band(rows: list[tuple[str, str]]) -> list[tuple[str, int, str]]:
498 """Two-band rows for a flat address listing (`--list`, group help)."""
499 return [
500 (f" {_styled_name(name)}", 2 + _describe.display_width(name), help_text)
501 for name, help_text in rows
502 ]
505def _sifted(node: dict[str, Any], want_global: bool) -> dict[str, Any] | None:
506 """*node* keeping only tasks from one side of the cascade, or `None`.
508 `None` when nothing on that side survives, so an empty section is not
509 a heading over blank space. Groups are kept only for what they still
510 hold: a group can straddle the split — a personal `docs.notes` beside
511 a project's `docs.build` — so the same group name can head a branch in
512 both sections, each carrying its own children.
513 """
514 tasks = {
515 name: t
516 for name, t in node.get("tasks", {}).items()
517 if bool(t.get("global")) is want_global
518 }
519 groups = {}
520 for name, sub in node.get("groups", {}).items():
521 if (kept := _sifted(sub, want_global)) is not None:
522 groups[name] = kept
523 if not tasks and not groups:
524 return None
525 sifted = {**node, "tasks": tasks, "groups": groups}
526 # A group's default is one of its tasks, so it goes when that task does.
527 if "default" in sifted and "default" not in tasks: 527 ↛ 528line 527 didn't jump to line 528 because the condition on line 527 was never true
528 sifted.pop("default", None)
529 sifted.pop("default_fanout", None)
530 return sifted
533def _sections(tree: dict[str, Any]) -> list[tuple[str, dict[str, Any]]]:
534 """The listing's sections, project first, each omitted when empty.
536 Project first because it is what the person came for: their own tasks,
537 in the directory they are standing in. The global rung — built-ins and
538 the personal tasks file — rides everywhere and is the backdrop.
539 Outside a project there is only one section, and outside a project
540 with no built-ins there are none, which the callers report as before.
541 """
542 found = [
543 (heading, kept)
544 for heading, want_global in (("Tasks:", False), ("Global tasks:", True))
545 if (kept := _sifted(tree, want_global)) is not None
546 ]
547 # One section needs no label saying which of two it is.
548 if len(found) == 1 and found[0][0] == "Global tasks:":
549 return [("Tasks:", found[0][1])]
550 return found
553def _print_list(tree: dict[str, Any], show_hidden: bool = False) -> None:
554 sections = [
555 (heading, rows)
556 for heading, node in _sections(tree)
557 if (
558 rows := list(
559 _describe.iter_tasks(
560 node, show_hidden=show_hidden, dedupe_defaults=True
561 )
562 )
563 )
564 ]
565 if not sections:
566 print("No tasks defined.")
567 return
568 for i, (heading, rows) in enumerate(sections):
569 if i:
570 print()
571 print(_describe.bold(heading, _color_out))
572 _print_two_band(_address_band(rows))
575def _print_tree(tree: dict[str, Any], show_hidden: bool = False) -> None:
576 sections = [
577 (heading, node)
578 for heading, node in _sections(tree)
579 if list(_describe.walk(node, show_hidden=show_hidden, dedupe_defaults=True))
580 ]
581 if not sections:
582 # Mirror _print_list rather than printing zero bytes and exiting 0.
583 print("No tasks defined.")
584 return
585 for i, (heading, node) in enumerate(sections):
586 if i:
587 print()
588 print(_describe.bold(heading, _color_out))
589 _print_branch(node, show_hidden)
592def _print_branch(node: dict[str, Any], show_hidden: bool = False) -> None:
593 rows = list(_describe.walk(node, show_hidden=show_hidden, dedupe_defaults=True))
594 last = _last_of_each_branch(rows)
595 # Leaf names under a drawn branch, not repeated dotted addresses:
596 # `--list` is the flat, copy-paste view, and a `--tree` that only
597 # indented the same addresses was that listing with worse alignment.
598 # Top level carries no connector: those names *are* the root, and a
599 # branch drawn from nothing reads as a stray glyph.
600 leads: list[str] = [] # per row: stem + joint, plain
601 trunk: list[bool] = [] # per ancestor depth: was it its branch's last child?
602 for i, (depth, _address, _leaf, _help_text, _kind) in enumerate(rows):
603 del trunk[depth:]
604 trunk.append(last[i])
605 stem = "".join(" " if up else "│ " for up in trunk[1:-1])
606 leads.append(stem + ("" if depth == 0 else ("└─ " if last[i] else "├─ ")))
607 band: list[tuple[str, int, str]] = []
608 for i, (_depth, address, leaf, help_text, kind) in enumerate(rows):
609 name = (
610 _describe.bold(leaf, _color_out)
611 if kind == "task"
612 else _describe.bold_cyan(f"{leaf}.", _color_out)
613 )
614 # Groups link too: their docs pages exist under the same scheme
615 # (a directory index, or the group heading's anchor).
616 name = _describe.link(name, _task_docs_url(address), _color_out)
617 cell = f"{_describe.dim(leads[i], _color_out)}{name}"
618 plain = len(leads[i]) + len(leaf) + (kind != "task") # groups carry a dot
619 band.append((cell, plain, help_text))
620 _print_two_band(band)
623def _last_of_each_branch(rows: Sequence[tuple[Any, ...]]) -> list[bool]:
624 """Which rows end their own branch — the `└─` corners.
626 A row is last when no later row sits at its depth before the walk climbs
627 back out above it."""
628 flags = [True] * len(rows)
629 for i, row in enumerate(rows):
630 for later in rows[i + 1 :]:
631 if later[0] < row[0]:
632 break
633 if later[0] == row[0]:
634 flags[i] = False
635 break
636 return flags
639def _print_option_rows(rows: list[tuple[str, str, str]], heading: str) -> None:
640 """One aligned option listing, wrapped to the terminal: labels bold on
641 the left, details in a hanging-indent column on the right — the
642 author's words bright, the mechanics dimmed, and the mechanics moving
643 to their own wrapped lines when the pair cannot share one. The globals
644 table and every parameter listing draw through here, so no help
645 surface leaves the terminal to hard-wrap a column apart.
646 """
647 if not rows:
648 return
649 import shutil
650 import textwrap
652 on = _color_out
653 width = max(_describe.display_width(label) for label, _, _ in rows)
654 desc_col = width + 4 # two leading spaces + two separating
655 avail = max(24, shutil.get_terminal_size().columns - desc_col)
656 print(f"\n{_describe.bold(f'{heading}:', on)}")
657 for label, doc, mech in rows:
658 joined = "; ".join(bit for bit in (doc, mech) if bit)
659 if _describe.display_width(joined) <= avail:
660 dimmed = _describe.dim(mech, on) if mech else ""
661 pieces = ["; ".join(bit for bit in (doc, dimmed) if bit)]
662 else:
663 pieces = textwrap.wrap(doc, avail) if doc else []
664 pieces += [
665 _describe.dim(cont, on)
666 for cont in (textwrap.wrap(mech, avail) if mech else [])
667 ]
668 pieces = pieces or [""]
669 pad = " " * (width - _describe.display_width(label))
670 print(f" {_describe.bold(label, on)}{pad} {pieces[0]}".rstrip())
671 for cont in pieces[1:]:
672 print(f"{' ' * desc_col}{cont}".rstrip())
675def _print_param_rows(params: list[dict[str, Any]], heading: str) -> None:
676 """One two-band parameter listing — task help and group help draw the
677 same rows (author's words bright, mechanics dimmed beneath), and used
678 to carry the loop twice."""
679 if not params:
680 return
681 rows = []
682 for p in params:
683 doc, mech = _describe.param_detail_parts(p)
684 rows.append((_describe.param_label(p), doc, mech))
685 _print_option_rows(rows, heading)
688def _choices_resolver(reg: registry.Group) -> _split.ChoicesFor:
689 """Answer a dynamic parameter's choices live, from the registry.
691 The completers no longer run when the manifest is built — nothing needed
692 them there — so the surfaces that *do* need values ask for them one
693 parameter at a time: the splitter for the value it is validating, help
694 for the options it is about to print. Memoised per invocation, so a
695 parameter asked twice runs its completer once, and a completer nobody's
696 line touches never runs at all.
697 """
698 from footman import _coerce
700 seen: dict[tuple[str, str], list[str] | None] = {}
702 def resolve(address: str, param: str) -> list[str] | None:
703 key = (address, param)
704 if key in seen: 704 ↛ 705line 704 didn't jump to line 705 because the condition on line 704 was never true
705 return seen[key]
706 found: list[str] | None = None
707 try:
708 fn = _executor.resolve(reg, address.split("."))
709 except (KeyError, IndexError):
710 fn = None
711 if fn is not None: 711 ↛ 719line 711 didn't jump to line 719 because the condition on line 711 was always true
712 for declared in _manifest.resolved_signature(fn).parameters.values(): 712 ↛ 719line 712 didn't jump to line 719 because the loop on line 712 didn't complete
713 if registry.cli_name(declared.name) != param: 713 ↛ 714line 713 didn't jump to line 714 because the condition on line 713 was never true
714 continue
715 completer = _coerce.peel(declared.annotation).completer
716 if completer is not None: 716 ↛ 718line 716 didn't jump to line 718 because the condition on line 716 was always true
717 found = _manifest._run_completer(completer, {})
718 break
719 seen[key] = found
720 return found
722 return resolve
725def _print_task_help(
726 tree: dict[str, Any],
727 path: list[str],
728 show_hidden: bool = False,
729 choices_for: _split.ChoicesFor | None = None,
730) -> None:
731 # All phrasing (labels, details, examples) lives in `_describe`, shared
732 # with the markdown exporter so help text and pages can never drift.
733 node = tree
734 for name in path[:-1]:
735 node = node["groups"][name]
736 task = node["tasks"][path[-1]]
737 on = _color_out
738 # Resolve before anything renders — the synopsis shows value shapes too.
739 # Only what this page prints: doing it here, after the listing filter,
740 # is what keeps `fm --help build` from running another task's completer.
741 shown = _describe.listed_params(task, show_hidden=show_hidden)
742 for spec in shown:
743 _split.live_choices(".".join(path), spec, choices_for)
744 usage = _describe.paint_cli(
745 _describe.usage_parts(_brand.prog, path, task, show_hidden=show_hidden), on
746 )
747 print(f"usage: {usage}")
748 if task["help"]:
749 print(f"\n {task['help']}")
750 if task.get("long"): # the docstring's body, structure preserved
751 body = "\n".join(f" {ln}".rstrip() for ln in task["long"].splitlines())
752 print(f"\n{body}")
753 if task.get("infinite"): 753 ↛ 754line 753 didn't jump to line 754 because the condition on line 753 was never true
754 print(_describe.dim("\n runs until you stop it — Ctrl-C", on))
755 if task.get("disabled"):
756 print(_describe.dim(f"\n unavailable here: {task['disabled']}", on))
757 positionals = [p for p in shown if p["kind"] in ("positional", "variadic")]
758 options = [p for p in shown if p["kind"] in ("flag", "option")]
759 _print_param_rows(positionals, "positionals")
760 _print_param_rows(options, "options")
761 returned = task.get("returned")
762 returned_doc = task.get("returned_doc", "")
763 if returned is not None or returned_doc:
764 # The output contract: the author's words bright, the shape dimmed
765 # beneath them — the same split every param row makes.
766 mech = _describe.returns_phrase(returned) if returned is not None else ""
767 detail = " ".join(
768 bit
769 for bit in (returned_doc, _describe.dim(mech, on) if mech else "")
770 if bit
771 )
772 print(f"\n{_describe.bold('returns:', on)} {detail}")
773 if uses := _describe.uses_line(task, tree):
774 # The globals this task declared (`uses=`): mechanics, so dimmed.
775 print(_describe.dim(f"\n {uses}", on))
776 example = _describe.paint_cli(_describe.example_parts(path, task, _brand.prog), on)
777 print(f"\n{_describe.dim('Example:', on)} {example}")
778 if (shadows := task.get("shadows")) is not None:
779 # This task overrides one further up the cascade — show the call
780 # `inherited()` makes, so the forwarding line can be read off it.
781 where = shadows.get("where") or "the cascade"
782 print(_describe.dim(f"\nshadows {where} — inherited() calls it", on))
783 usage = _describe.paint_cli(
784 _describe.usage_parts(_brand.prog, path, shadows, show_hidden=show_hidden),
785 on,
786 )
787 print(f" {usage}")
788 _print_docs_line(".".join(path), on)
791def _print_docs_line(address: str, on: bool) -> None:
792 """The help pages' pointer at the task's own docs page — the URL as
793 visible text (a terminal without hyperlinks still shows something to
794 copy), hyperlinked where the terminal dresses up. Nothing without a
795 configured `docs_url`."""
796 if (url := _task_docs_url(address)) is not None:
797 print(f"\n{_describe.dim('docs:', on)} {_describe.link(url, url, on)}")
800def _print_group_help(
801 tree: dict[str, Any],
802 path: list[str],
803 show_hidden: bool = False,
804 choices_for: _split.ChoicesFor | None = None,
805) -> None:
806 node = tree
807 for name in path:
808 node = node["groups"][name]
809 on = _color_out
810 default = node.get("default")
811 dotted = ".".join(path)
812 # A runnable group (one with `@group.default`) can run bare — its default —
813 # so the task suffix becomes optional. The address stays one dotted token.
814 head = f"{dotted}[.<task>]" if default else f"{dotted}.<task>"
815 parts = [("prog", _brand.prog), ("group", head), ("opt", "[options]")]
816 print(f"usage: {_describe.paint_cli(parts, on)}")
817 if node["help"]:
818 print(f"\n {node['help']}")
819 if default:
820 print(_describe.dim("\n runs its default when no task is named", on))
821 # `covered`: the bare-group row inserted below stands in for `default`.
822 rows = list(
823 _describe.iter_tasks(
824 node,
825 f"{dotted}.",
826 show_hidden=show_hidden,
827 dedupe_defaults=True,
828 covered=bool(default),
829 )
830 )
831 if default:
832 # The bare-group spelling is itself a runnable, listed address —
833 # described by its default action (docstring, or generated).
834 rows.insert(0, (dotted, _describe.default_line(node)))
835 if rows: 835 ↛ 838line 835 didn't jump to line 838 because the condition on line 835 was always true
836 print(f"\n{_describe.bold('tasks:', on)}")
837 _print_two_band(_address_band(rows))
838 params = (
839 _describe.listed_params(default, show_hidden=show_hidden) if default else []
840 )
841 for spec in params: # the default's own options, and only those
842 _split.live_choices(f"{dotted}.default", spec, choices_for)
843 options = [p for p in params if p["kind"] in ("flag", "option")]
844 _print_param_rows(options, "options")
845 _print_docs_line(dotted, on)
848def _print_global_help(tree: dict[str, Any], show_hidden: bool = False) -> None:
849 prog = _brand.prog
850 parts = [
851 ("prog", prog),
852 ("opt", "[globals]"),
853 ("req", "<task>"),
854 ("opt", "[options]"),
855 ("opt", "[<task> ...]"),
856 ]
857 print(f"usage: {_describe.paint_cli(parts, _color_out)}")
858 rows = []
859 for name, alias, _kind, hint, _default, help_text in _split.GLOBALS:
860 label = f"{alias}, {name}" if alias else f" {name}"
861 if hint:
862 label += f"={hint}" # values are always `=`-attached
863 # `.replace` (not `.format`) so a help string containing braces can
864 # never crash help output.
865 detail = help_text.replace("{prog}", prog)
866 # The same thing task parameters gained: say what you get when you say
867 # nothing. Computed defaults resolve here, so `--jobs` reports this
868 # machine's width rather than a number from the author's. One
869 # composition with the docs table (`_describe.global_default_suffix`).
870 detail += _describe.global_default_suffix(name)
871 rows.append((label, detail, ""))
872 _print_option_rows(rows, "globals (before the first task)")
873 # Mounted plugin globals ride the same pre-task position, so a help page
874 # claiming to list the globals must list these too — with provenance,
875 # since "where did --profile come from" is the first question a reader
876 # who didn't mount the plugin asks.
877 prows = []
878 for p in tree.get("globals") or ():
879 doc, mech = _describe.param_detail_parts(p)
880 doc = doc or p.get("help", "") # a global's words ride as `help`
881 if owner := p.get("owner"): 881 ↛ 883line 881 didn't jump to line 883 because the condition on line 881 was always true
882 mech = f"{mech}; from {owner}" if mech else f"from {owner}"
883 prows.append((_describe.param_label(p), doc, mech))
884 _print_option_rows(prows, "plugin globals (before the first task)")
885 print()
886 _print_list(tree, show_hidden)
887 _print_footer()
890def _wants_help(argv: list[str]) -> bool:
891 """`-h`/`--help` anywhere before `--` turns the whole line into a help
892 request — asking for help must never execute anything, wherever it lands
893 on the line. After `--` it belongs to the passthrough."""
894 for tok in argv:
895 if tok == "--":
896 return False
897 if tok in ("-h", "--help"):
898 return True
899 return False
902def _resolve_lenient(tree: dict[str, Any], token: str) -> tuple[str, list[str]] | None:
903 """Walk one dotted address to `("task"|"group", path)`, or None.
905 The help surface's resolver: never raises — a token that isn't an address
906 is someone's argument value or a typo, and the caller decides which.
907 """
908 parts = token.split(".")
909 if "" in parts: 909 ↛ 910line 909 didn't jump to line 910 because the condition on line 909 was never true
910 return None
911 node, path = tree, []
912 for pos, part in enumerate(parts): 912 ↛ 923line 912 didn't jump to line 923 because the loop on line 912 didn't complete
913 last = pos == len(parts) - 1
914 if part in node["groups"]:
915 node = node["groups"][part]
916 path.append(part)
917 if last:
918 return ("group", path)
919 elif part in node["tasks"] and last:
920 return ("task", [*path, part])
921 else:
922 return None
923 return None
926def _help_targets(
927 tree: dict[str, Any], argv: list[str], after: int
928) -> tuple[list[tuple[str, list[str]]], list[str]]:
929 """Group/task addresses mentioned on a `--help` line, resolved leniently —
930 plus the bare words that resolved to nothing, so the caller can refuse a
931 `--help typo` instead of shrugging.
933 The real splitter enforces arity — `--help add` must work even though
934 `add` alone would be "missing required positional(s)" — so this resolves
935 dotted addresses only and skips every other token (option-shaped tokens
936 and, once a target is found, its argument values).
937 """
938 targets: list[tuple[str, list[str]]] = []
939 strays: list[str] = []
940 for tok in argv[after:]:
941 if tok == "--":
942 break
943 if tok.startswith("-"):
944 continue
945 resolved = _resolve_lenient(tree, tok)
946 if resolved is not None:
947 targets.append(resolved)
948 else:
949 strays.append(tok)
950 return targets, strays
953def _print_help(
954 tree: dict[str, Any],
955 argv: list[str],
956 after: int,
957 show_hidden: bool = False,
958 choices_for: _split.ChoicesFor | None = None,
959) -> int:
960 """`--help` alone covers fm itself; with names, the named groups/tasks.
962 A name that matches nothing is a refusal (exit EX_USAGE) with a suggestion —
963 silently degrading to the global listing looked like an answer while
964 teaching nothing. With at least one real target found, extra bare words
965 stay lenient: they are argument values, not typos.
966 """
967 targets, strays = _help_targets(tree, argv, after)
968 if not targets:
969 if strays:
970 # Every address a human can type, hidden included: the index
971 # answers a typo, and a machine-facing task typo'd by hand
972 # deserves the same "did you mean" as any other.
973 known = [name for name, _ in _describe.iter_tasks(tree, show_hidden=True)]
974 known += _describe.iter_group_paths(tree)
975 # Help's *success* output is the one human-only surface; a refusal
976 # still honours the envelope `--json` promised.
977 return _refuse(
978 _wants_json(argv),
979 f"--help: unknown task or group {strays[0]!r}"
980 f"{_split._did_you_mean(strays[0], known)}",
981 )
982 _print_global_help(tree, show_hidden)
983 return 0
984 for index, (kind, path) in enumerate(targets):
985 if index: 985 ↛ 986line 985 didn't jump to line 986 because the condition on line 985 was never true
986 print()
987 if kind == "task":
988 _print_task_help(tree, path, show_hidden, choices_for)
989 else:
990 _print_group_help(tree, path, show_hidden, choices_for)
991 return 0
994def _plugins_report(reg: registry.Group) -> int:
995 """`--plugins`: installed `footman.tasks` entry points, mounted or not,
996 grouped by the distribution that ships them.
998 "Installed but nobody mounted it" becomes visible. The entry-point
999 record itself cannot carry a description (the packaging spec is strictly
1000 `name = "module:attr"`), and a distribution's Summary describes the
1001 *package* — so it prints once, on the package's header line, instead of
1002 repeating identically beside every entry the package ships. An entry
1003 then describes itself only where footman genuinely knows it: a mounted
1004 entry from its landed tree, a declared built-in from its advertised tree
1005 (the brand vouches for importing its own declarations). A plain
1006 unmounted entry shows its state alone — importing unmounted third-party
1007 code could crash a listing, and a repeated package summary taught
1008 nothing.
1009 """
1010 from importlib.metadata import entry_points
1012 from footman import compose
1014 eps = sorted(entry_points(group=compose.ENTRY_POINT_GROUP), key=lambda e: e.name)
1015 if not eps: 1015 ↛ 1016line 1015 didn't jump to line 1016 because the condition on line 1015 was never true
1016 print("No footman.tasks plugins installed.")
1017 return 0
1019 landed: dict[str, list[str]] = {}
1021 def walk(node: registry.Group, prefix: str, inside: bool) -> None:
1022 # Report the top-most mounted node per identity — its whole subtree
1023 # shares the provenance, and the top is the copy-paste address.
1024 for name, fn in node.tasks.items():
1025 ident = registry.mounted_from(fn)
1026 if ident is not None and not inside:
1027 landed.setdefault(ident, []).append(f"{prefix}{name}")
1028 for name, sub in node.groups.items():
1029 ident = sub.mounted_from
1030 if ident is not None and not inside:
1031 landed.setdefault(ident, []).append(f"{prefix}{name}")
1032 walk(sub, f"{prefix}{name}.", inside or ident is not None)
1034 walk(reg, "", False)
1036 # A plugin with no tasks still mounts — its hooks and options ride every
1037 # run — and the tree walk cannot see it. Its contributions carry the
1038 # identity the mount stamped.
1039 riding: set[str] = set()
1040 for kind, bucket in reg.contributions.items():
1041 for item in bucket:
1042 ident = (
1043 item._mounted
1044 if kind == "globals"
1045 else getattr(item, registry._MOUNTED, None)
1046 )
1047 if isinstance(ident, str) and ident != registry._MANY_MOUNTS: 1047 ↛ 1041line 1047 didn't jump to line 1041 because the condition on line 1047 was always true
1048 riding.add(ident)
1050 def described(addresses: list[str]) -> str:
1051 node: object = reg
1052 for part in addresses[0].split("."):
1053 node = (
1054 node.groups.get(part, node.tasks.get(part))
1055 if isinstance(node, registry.Group)
1056 else None
1057 )
1058 if isinstance(node, registry.Group):
1059 return node.help
1060 doc = (getattr(node, "__doc__", "") or "").strip()
1061 return doc.splitlines()[0] if doc else ""
1063 def advertised(entry: str) -> str:
1064 """A declared built-in's own line, from the tree it advertises —
1065 resolved on demand, exactly as the unknown-task remedy resolves it.
1066 Empty on any failure: the report never crashes."""
1067 try:
1068 _ident, node = compose._resolve_plugin(entry)
1069 except Exception:
1070 return ""
1071 if isinstance(node, registry.Group): 1071 ↛ 1080line 1071 didn't jump to line 1080 because the condition on line 1071 was always true
1072 if not node.groups and len(node.tasks) == 1: 1072 ↛ 1075line 1072 didn't jump to line 1075 because the condition on line 1072 was never true
1073 # A single-task plugin: the task's own line beats the
1074 # module docstring an anonymous capture carries as help.
1075 (fn,) = node.tasks.values()
1076 doc = (getattr(fn, "__doc__", "") or "").strip()
1077 if doc:
1078 return doc.splitlines()[0]
1079 return node.help or ""
1080 doc = (getattr(node, "__doc__", "") or "").strip()
1081 return doc.splitlines()[0] if doc else ""
1083 def mount_points(where: list[str]) -> str:
1084 tops = sorted(where)
1085 if len(tops) > 3: 1085 ↛ 1086line 1085 didn't jump to line 1086 because the condition on line 1085 was never true
1086 return f"mounted at {', '.join(tops[:3])} (+{len(tops) - 3} more)"
1087 return f"mounted at {', '.join(tops)}"
1089 dists: dict[str, tuple[str, str]] = {} # name -> (version, summary)
1090 grouped: dict[str, list[tuple[str, str, str]]] = {}
1091 for ep in eps:
1092 meta = getattr(ep.dist, "metadata", None)
1093 dist_name = (meta.get("Name", "") if meta else "") or "(unknown)"
1094 dists.setdefault(
1095 dist_name,
1096 (
1097 getattr(ep.dist, "version", "") or "",
1098 (meta.get("Summary", "") if meta else "") or "",
1099 ),
1100 )
1101 where = landed.get(ep.name)
1102 if ep.name in _builtin():
1103 # Built into this runner: part of the product wherever there is
1104 # no project, and an ordinary mount inside one. Which *rung*
1105 # declared it matters to the reader — the brand's own set is the
1106 # product, the user's is their machine's, and only the second is
1107 # theirs to edit.
1108 desc = described(where) if where else advertised(ep.name)
1109 state = (
1110 "built in" if ep.name in _brand.builtin else "built in (your config)"
1111 )
1112 grouped.setdefault(dist_name, []).append((ep.name, state, desc))
1113 elif where:
1114 # One landed node speaks for itself; a family mounted piecemeal
1115 # speaks with its advertised voice — an arbitrary member's
1116 # docstring must not stand for seven siblings.
1117 desc = described(where) if len(where) == 1 else advertised(ep.name)
1118 grouped.setdefault(dist_name, []).append(
1119 (ep.name, mount_points(where), desc)
1120 )
1121 elif ep.name in riding:
1122 # Mounted with no tasks to land: hooks and options riding every
1123 # run have no address to name, so the state is the plain word.
1124 grouped.setdefault(dist_name, []).append(
1125 (ep.name, "mounted", advertised(ep.name))
1126 )
1127 else:
1128 grouped.setdefault(dist_name, []).append((ep.name, "(not mounted)", ""))
1129 every = [row for rows in grouped.values() for row in rows]
1130 name_w = max(_describe.display_width(name) for name, _, _ in every)
1131 state_w = max(_describe.display_width(state) for _, state, _ in every)
1132 on = _color_out
1133 for dist_name in sorted(grouped):
1134 version, summary = dists[dist_name]
1135 header = f"{_describe.bold(dist_name, on)} {version}".rstrip()
1136 if summary:
1137 header += f" {_describe.dim(f'— {summary}', on)}"
1138 print(header)
1139 for name, state, desc in grouped[dist_name]:
1140 line = (
1141 f" {_describe.pad_to(name, name_w)}"
1142 f" {_describe.pad_to(state, state_w)}"
1143 )
1144 if desc:
1145 line += f" {_describe.dim(f'— {desc}', on)}"
1146 print(line.rstrip())
1147 unmounted = [name for name, state, _ in every if state == "(not mounted)"]
1148 if unmounted: 1148 ↛ 1159line 1148 didn't jump to line 1159 because the condition on line 1148 was always true
1149 # Installed is not asked-for: mounts are authored in the tasks file,
1150 # and a listing that stops at "(not mounted)" leaves the reader with
1151 # a state and no verb. One line names the move.
1152 print(
1153 _describe.dim(
1154 f"not mounted = installed but not asked for — mount one from "
1155 f'your tasks file: plugin("{unmounted[0]}")',
1156 on,
1157 )
1158 )
1159 return 0
1162def _where(root: registry.Group, tree: dict[str, Any], dotted: str) -> int:
1163 # Strict, like every address surface: `docs..serve` or a trailing dot is
1164 # an error, never silently normalised away.
1165 path = dotted.split(".")
1166 try:
1167 if "" in path:
1168 raise KeyError(dotted)
1169 fn = _executor.resolve(root, path)
1170 except (KeyError, IndexError):
1171 names = _split.flat_addresses(tree)
1172 _error(f"--where: unknown task {dotted!r}{_split._did_you_mean(dotted, names)}")
1173 return EX_USAGE
1174 chain = _discover.shadow_chain(fn)
1175 lines = []
1176 for index, member in enumerate(chain):
1177 code = getattr(member, "__code__", None)
1178 if code is None: 1178 ↛ 1179line 1178 didn't jump to line 1179 because the condition on line 1178 was never true
1179 continue
1180 where = f"{code.co_filename}:{code.co_firstlineno}"
1181 # The winner first; anything it shadows follows, marked — so
1182 # "am I overriding something, and where is it?" is one command.
1183 lines.append(where if index == 0 else f"{where} (shadowed)")
1184 if not lines: 1184 ↛ 1185line 1184 didn't jump to line 1185 because the condition on line 1184 was never true
1185 _error(f"--where: cannot locate source for {dotted!r}")
1186 return EX_USAGE
1187 print("\n".join(lines))
1188 return 0
1191def _iter_task_nodes(
1192 node: dict[str, Any], prefix: str
1193) -> Iterator[tuple[str, dict[str, Any]]]:
1194 """Every task node under *node* with its dotted address — a group's
1195 `default` included, because it *is* the child task named `default`."""
1196 for name, task in node["tasks"].items():
1197 yield prefix + name, task
1198 for name, sub in node["groups"].items():
1199 yield from _iter_task_nodes(sub, f"{prefix}{name}.")
1202def _contract_entry(address: str, task: dict[str, Any]) -> dict[str, Any]:
1203 """One task's `--describe` entry: the input surface (the param specs,
1204 volatile completer choices dropped) and the output contract (the declared
1205 schema rendered as JSON Schema, plus the docstring's `Returns:` prose)."""
1206 params = []
1207 for p in task["params"]:
1208 spec = dict(p)
1209 if "dynamic" in spec: 1209 ↛ 1214line 1209 didn't jump to line 1214 because the condition on line 1209 was always true
1210 # A dynamic completer's baked choices are runtime data (whatever
1211 # the completer saw at build time), not contract — a snapshot
1212 # that pinned them would flap between machines.
1213 spec.pop("choices", None)
1214 params.append(spec)
1215 entry: dict[str, Any] = {"task": address, "help": task["help"], "params": params}
1216 if task.get("hidden"):
1217 entry["hidden"] = True
1218 returned = task.get("returned")
1219 doc = task.get("returned_doc")
1220 if returned is not None or doc:
1221 block: dict[str, Any] = {}
1222 if returned is not None: 1222 ↛ 1224line 1222 didn't jump to line 1224 because the condition on line 1222 was always true
1223 block["schema"] = _describe.returns_json_schema(returned)
1224 if doc:
1225 block["doc"] = doc
1226 entry["returns"] = block
1227 return entry
1230def _describe_contract(tree: dict[str, Any], target: object, argv_rest: bool) -> int:
1231 """`--describe[=ADDR]`: the input+output contract as one JSON document.
1233 Bare, it hands an agent the entire API — every task's params and declared
1234 return schema, sorted by address so a checked-in snapshot is invariant to
1235 declaration order. A task address answers with that one entry; a group
1236 address answers for its whole subtree — the prefix-names-a-subtree rule
1237 every address surface speaks — and a runnable group's default alone is
1238 its real `group.default` address, the child the bare group runs. Plain
1239 JSON on stdout either way: like `--where`, the output already is the
1240 machine format, so `--json` adds nothing.
1241 """
1242 if not target: # named bare: the whole tree's contract
1243 if argv_rest:
1244 # `fm --describe check` reads as bare --describe plus a run of
1245 # `check` — surely not what was meant. Teach the `=` spelling
1246 # rather than silently describing everything.
1247 _error("--describe: name the address in the value — --describe=<addr>")
1248 return EX_USAGE
1249 entries = [
1250 _contract_entry(address, node)
1251 for address, node in sorted(_iter_task_nodes(tree, ""))
1252 ]
1253 else:
1254 dotted = str(target)
1255 path = dotted.split(".")
1256 node: dict[str, Any] = tree
1257 try:
1258 if "" in path: 1258 ↛ 1259line 1258 didn't jump to line 1259 because the condition on line 1258 was never true
1259 raise KeyError(dotted)
1260 for name in path[:-1]:
1261 node = node["groups"][name]
1262 last = path[-1]
1263 if last in node["tasks"]:
1264 entries = [_contract_entry(dotted, node["tasks"][last])]
1265 else:
1266 sub = node["groups"][last]
1267 entries = [
1268 _contract_entry(address, task)
1269 for address, task in sorted(_iter_task_nodes(sub, f"{dotted}."))
1270 ]
1271 except (KeyError, IndexError):
1272 names = _split.flat_addresses(tree) + list(_describe.iter_group_paths(tree))
1273 _error(
1274 f"--describe: unknown task or group {dotted!r}"
1275 f"{_split._did_you_mean(dotted, names)}"
1276 )
1277 return EX_USAGE
1278 print(json.dumps({"schema": 1, "tasks": entries}, indent=2))
1279 return 0
1282def _utf8(text: str) -> bytes:
1283 """A document's text as UTF-8 bytes, degrading only what cannot be text.
1285 Strict would trade a silent '?' for a traceback on an already-finished
1286 task, and the only strings that fail are ones holding lone surrogates —
1287 the escape hatch for bytes that were never characters, which no encoding
1288 can carry. Everything a reader would call text survives intact.
1289 """
1290 return text.encode("utf-8", "replace")
1293def _emit_bytes(payload: bytes) -> None:
1294 """Put a document's bytes on stdout, underneath the text layer's codec.
1296 A document is a payload footman defines, so it is UTF-8 always — no
1297 autodetection, no locale. `sys.stdout` encodes with the locale's codec
1298 and a run reconfigures it to `errors="replace"` so a tool's stray glyph
1299 can never crash the run; together those turned "café" into b"caf?" on a
1300 cp1252 console, with exit 0 and nothing on stderr. Writing bytes to the
1301 underlying buffer skips the translation and pins UTF-8; captured stdout
1302 (tests, some wrappers) has no buffer, so fall back.
1303 """
1304 buffer = getattr(sys.stdout, "buffer", None)
1305 if buffer is None: # an embedded runner's text capture has no byte buffer
1306 sys.stdout.write(payload.decode("utf-8", "replace"))
1307 return
1308 # Anything already handed to the text layer belongs ahead of these bytes.
1309 with contextlib.suppress(Exception):
1310 sys.stdout.flush()
1311 buffer.write(payload)
1312 buffer.flush()
1315def _emit_document(value: object, inner: object) -> None:
1316 """The declared document, on stdout: text verbatim (plus a trailing
1317 newline), bytes raw, anything else JSON — pretty-printed on a terminal,
1318 one compact line into a pipe, always a trailing newline, and always UTF-8.
1319 Encoded via `_describe.json_default`, the same encoder `--json` uses, so
1320 dataclasses serialise and `Secret` redacts identically on both surfaces."""
1321 mode = _coerce.emission_mode(inner)
1322 if mode == "bytes" and isinstance(value, (bytes, bytearray)):
1323 _emit_bytes(bytes(value))
1324 return
1325 if mode == "text" and isinstance(value, str):
1326 _emit_bytes(_utf8(value if value.endswith("\n") else value + "\n"))
1327 return
1328 try:
1329 tty = sys.stdout.isatty()
1330 except Exception:
1331 tty = False
1332 text = json.dumps(
1333 # The walk `--json` does, which this surface claimed above to share and
1334 # did not: `json_default` alone cannot redact, because `Secret` is a
1335 # `str` subclass and rides `json.dumps`' fast path without ever
1336 # reaching the `default` hook. A document went out in the clear while
1337 # the envelope for the same task redacted.
1338 _describe.redact(value),
1339 default=_describe.json_default,
1340 ensure_ascii=False,
1341 indent=2 if tty else None,
1342 separators=None if tty else (",", ":"),
1343 )
1344 _emit_bytes(_utf8(text + "\n"))
1347def _stack_wanted(verbose: bool) -> bool:
1348 """`_describe.stack_wanted`, asked about this run's real stderr.
1350 The *real* stream, not `ctx.tty`, which folds in whether output is being
1351 captured: `Runner` captures, and a suite is not a log to be rescued.
1352 """
1353 return _describe.stack_wanted(verbose, context.real_stderr().isatty())
1356def _print_summary(
1357 results: list[_executor.TaskResult],
1358 *,
1359 timings: bool,
1360 verbose: bool = False,
1361 total: float,
1362) -> None:
1363 # The summary is commentary about the run, not the run's output — it goes
1364 # to stderr so `fm task > file` captures exactly what the task produced.
1365 # Each receipt is task-shaped (mark · name · time), the same grid as the
1366 # step lines above it, with the name in cyan — same family, one rank up.
1367 color = _color_err
1368 width = max((_describe.display_width(r.task) for r in results), default=0)
1369 for result in results:
1370 ok = result.ok
1371 cancelled = result.cancelled
1372 state = _executor.reported_state(result)
1373 if color:
1374 if state == "shared": 1374 ↛ 1376line 1374 didn't jump to line 1376 because the condition on line 1374 was never true
1375 # Dimmed, not green: nothing ran, the run already had it.
1376 mark = "\033[2m·\033[0m"
1377 elif state == "skipped": 1377 ↛ 1379line 1377 didn't jump to line 1379 because the condition on line 1377 was never true
1378 # Dimmed too: nothing ran, and the cause is the story.
1379 mark = "\033[2m-\033[0m"
1380 elif ok: 1380 ↛ 1382line 1380 didn't jump to line 1382 because the condition on line 1380 was always true
1381 mark = "\033[32m✓\033[0m"
1382 elif cancelled:
1383 mark = "\033[33m○\033[0m" # cut off by fail-fast, not a failure
1384 else:
1385 mark = "\033[31m✗\033[0m"
1386 name = _describe.bold_cyan(_describe.pad_to(result.task, width), True)
1387 else:
1388 word = "ok" if ok else ("cut" if cancelled else "FAIL")
1389 if state == "shared":
1390 word = "same"
1391 elif state == "skipped":
1392 word = "skip"
1393 mark = f"{word:<4}"
1394 name = _describe.pad_to(result.task, width)
1395 if state == "shared":
1396 timing = "(already run this run)"
1397 elif state == "skipped":
1398 cause = result.blocked_by
1399 timing = f"(blocked by {cause})" if cause else "(skipped)"
1400 elif timings:
1401 timing = f"({result.duration * 1000:.0f} ms)"
1402 else:
1403 timing = f"({_progress.fmt_secs(result.duration)})"
1404 if color:
1405 timing = f"\033[36m{timing}\033[0m"
1406 print(f"{mark} {name} {timing}", file=sys.stderr)
1407 if cancelled: 1407 ↛ 1408line 1407 didn't jump to line 1408 because the condition on line 1407 was never true
1408 _error(f"{result.task}: cancelled — fail-fast stopped the run")
1409 elif result.error is not None:
1410 # A deliberate stop with a reason (`fail("…")` / `sys.exit("…")`)
1411 # renders verbatim, the way Python prints it — not "Failed: …"; a bare
1412 # `fail()` with no reason falls back to the code line. A real exception
1413 # keeps its type, which signals a crash, not a chosen stop.
1414 err = result.error
1415 if context._is_deliberate_stop(err):
1416 detail = str(err) or f"exited with code {result.code}"
1417 else:
1418 detail = f"{type(err).__name__}: {err}"
1419 _error(f"{result.task}: {detail}")
1420 if not context._was_expected(err):
1421 # An exception nobody planned: the reader's own bug, and the
1422 # only question it raises is where. The line used to end above,
1423 # so a task that raised said what happened and never where —
1424 # information destroyed rather than merely unformatted.
1425 #
1426 # The place by default, the whole stack under -v or when
1427 # stderr is not a terminal. Not a terminal means a log — CI, a
1428 # redirect, cron — where there is no second run to add -v to,
1429 # and the same reasoning already sends the progress estimate
1430 # down its one-shot path.
1431 # One place decides, whichever half of the runner the exception
1432 # came out of: a step keeps its stack in the record and leaves
1433 # the placing to here, so a task and a step read the same and
1434 # neither is said twice — including under a quiet capture,
1435 # where a step's receipt is never displayed at all.
1436 if _stack_wanted(verbose) and (stack := _describe.user_traceback(err)): 1436 ↛ 1438line 1436 didn't jump to line 1438 because the condition on line 1436 was always true
1437 print(stack.rstrip("\n"), file=sys.stderr)
1438 elif where := _describe.user_frame(err):
1439 _error(f" at {where}")
1440 elif not result.ok and state != "skipped":
1441 # A skipped row's whole story is its cause, already on the line.
1442 _error(f"{result.task}: exited with code {result.code}")
1443 if len(results) > 1: # one task's receipt already carries the total
1444 took = f"took {_progress.fmt_secs(total)}"
1445 if color:
1446 took = f"\033[2m{took}\033[0m"
1447 print(took, file=sys.stderr)
1450def _check_returns(
1451 root: registry.Group, results: list[_executor.TaskResult]
1452) -> dict[int, tuple[dict[str, Any], str | None]]:
1453 """Per-result output contract: `{index: (native spec, mismatch note)}`.
1455 The producer side of drift protection, paid only by declaring tasks
1456 (~tens of µs per value): every reported return that has a declared
1457 schema is walked against it, and a mismatch warns on stderr — in every
1458 mode, so the rename goes red in the producer's own gate before any
1459 consumer integrates — and rides the envelope as `returned_mismatch`.
1460 A payload problem stays a note; the exit code never moves.
1461 """
1462 memo: dict[int, dict[str, Any] | None] = {}
1463 out: dict[int, tuple[dict[str, Any], str | None]] = {}
1464 for index, r in enumerate(results):
1465 try:
1466 fn = _executor.resolve(root, r.task.split("."))
1467 except (KeyError, IndexError):
1468 continue # a synthetic row (a step-only shared node) has no task
1469 if id(fn) not in memo:
1470 memo[id(fn)] = _manifest.returned_spec(
1471 _manifest.resolved_signature(fn).return_annotation
1472 )
1473 spec = memo[id(fn)]
1474 if spec is None:
1475 continue
1476 value = r.returned
1477 note = None
1478 if value is not None and not (
1479 isinstance(value, int) and not isinstance(value, bool)
1480 ):
1481 note = _describe.returned_mismatch(value, spec)
1482 if note:
1483 _error(f"{r.task}: return value breaks its declared shape — {note}")
1484 out[index] = (spec, note)
1485 return out
1488def _print_json(
1489 results: list[_executor.TaskResult],
1490 *,
1491 total: float,
1492 returns: dict[int, tuple[dict[str, Any], str | None]] | None = None,
1493) -> None:
1494 payload = []
1495 for index, r in enumerate(results):
1496 entry: dict[str, object] = {
1497 "task": r.task,
1498 "address": r.address,
1499 "ok": r.ok,
1500 "state": _executor.reported_state(r),
1501 "cancelled": r.cancelled,
1502 "code": r.code,
1503 "duration_ms": round(r.duration * 1000, 3),
1504 "output": r.output,
1505 "error": None if r.error is None else str(r.error),
1506 }
1507 if r.error is not None and not context._was_expected(r.error):
1508 # An exception nobody planned, so the reader's own bug: the stack
1509 # rides along whatever the terminal was doing. A consumer of this
1510 # envelope is a log or a dashboard, never someone who can re-run
1511 # with -v, and losing the one thing that places the failure is the
1512 # problem this whole rule exists to fix. Trimmed of footman's
1513 # frames like the printed one. Additive to schema 1; absent for a
1514 # command that exited non-zero or a deliberate stop, which have
1515 # nothing to place.
1516 entry["traceback"] = _describe.user_traceback(r.error)
1517 if r.title:
1518 # A reviewer's label for the row; absent when no one set one.
1519 entry["title"] = r.title
1520 if r.audit:
1521 # The row was reviewed: its verdict provenance and the derived
1522 # failing moment, same shape as on steps. Additive to schema 1.
1523 entry["audit"] = [[e.moment, e.actor, e.code] for e in r.audit]
1524 entry["failed_at"] = r.failed_at
1525 if r.blocked_by:
1526 entry["blocked_by"] = r.blocked_by
1527 if r.eligible is not None and r.started is not None:
1528 # Launch latency: how long the node sat ready, waiting for a
1529 # worker, after its last prerequisite finished. Never part of
1530 # `duration_ms` — the task wasn't running.
1531 entry["queued_ms"] = round(max(r.started - r.eligible, 0.0) * 1000, 3)
1532 if r.lane_waits:
1533 # Which lanes serialised this task, for how long — present only
1534 # when the claim actually waited. Additive to schema 1.
1535 entry["lane_waits"] = [
1536 {"lane": lane, "waited_ms": round(seconds * 1000, 3)}
1537 for lane, seconds in r.lane_waits
1538 ]
1539 if r.thread:
1540 # Where it ran: the worker's stable name and OS thread id — the
1541 # correlation keys a profiler's timeline uses. Absent for a row
1542 # that executed nothing (a `shared` row, a refusal).
1543 entry["thread"] = r.thread
1544 entry["thread_id"] = r.thread_id
1545 if r.after:
1546 # The plan's edges into this row, by address — what a profile
1547 # draws dependency arrows from. Additive to schema 1.
1548 entry["after"] = list(r.after)
1549 if (docs := _task_docs_url(r.task)) is not None:
1550 # The same pointer the terminal links carry, for readers that
1551 # cannot click — an agent's next move after a confusing row is
1552 # the task's docs. Additive to schema 1.
1553 entry["docs_url"] = docs
1554 if r.notes:
1555 # What the level system resolved for this execution — every
1556 # fired note, trace included: the machine channel ignores print
1557 # gating. Additive to schema 1.
1558 entry["notes"] = [
1559 {
1560 "kind": n.kind,
1561 "level": n.level,
1562 "site": n.site,
1563 "text": n.text,
1564 }
1565 for n in r.notes
1566 ]
1567 if r.sections and r.started is not None:
1568 # Task-authored profiling: `at_ms` places each section relative
1569 # to the task's own start (negative is legal — a retroactive
1570 # stream window may predate the task). Additive to schema 1.
1571 entry["sections"] = [
1572 {
1573 "name": s.name,
1574 **({"stream": s.stream} if s.stream else {}),
1575 "at_ms": round((s.started - r.started) * 1000, 3),
1576 "duration_ms": round(s.duration * 1000, 3),
1577 }
1578 for s in r.sections
1579 ]
1580 value = r.returned
1581 # An int return is the exit-code channel (duty's contract), not data;
1582 # None is "nothing to say". Everything else — bools included — is data.
1583 if value is not None and not (
1584 isinstance(value, int) and not isinstance(value, bool)
1585 ):
1586 value = _describe.redact(value) # a Secret never serialises
1587 try:
1588 json.dumps(value, default=_describe.json_default)
1589 except (TypeError, ValueError) as exc: # ValueError: circular refs
1590 entry["returned_error"] = str(exc)
1591 _error(f"{r.task}: --json: return value dropped — {exc}")
1592 else:
1593 entry["returned"] = value
1594 if (contract := (returns or {}).get(index)) is not None:
1595 spec, note = contract
1596 # The declared output contract, in the baked native form — data
1597 # and how to read it, one call. Present whenever the task
1598 # declares, value or no value (a failed run still has a shape).
1599 entry["returned_schema"] = spec
1600 if note:
1601 # Loud but local: the value still serialises; the note says
1602 # where it first breaks the declared shape.
1603 entry["returned_mismatch"] = note
1604 payload.append(entry)
1605 # The children stand alone, right after their requester: ONE flat
1606 # list in creation order, the tree carried by every item's address
1607 # (a prefix names a subtree, and a child's entry is self-describing
1608 # before its parent's would even be complete in a stream). A row
1609 # has "task"; a step has "command" — that is the reader's kind
1610 # test, and `[.items[] | select(.task == NAME)]` is the name
1611 # lookup, a LIST by contract: the same label may name distinct
1612 # work, distinct by address.
1613 for s in r.steps:
1614 payload.append(
1615 {
1616 # `shown`, not `command`: a document that leaves the
1617 # process is a display, and a `Secret` argument has no
1618 # business in a CI log. The record keeps the real line.
1619 "command": s.shown,
1620 "address": s.address,
1621 "code": s.code,
1622 "duration_ms": round(s.duration * 1000, 3),
1623 # Where the step sits inside its task's span — absent for
1624 # a record that never ran (dry-run). Additive to schema 1.
1625 **(
1626 {"at_ms": round((s.started - r.started) * 1000, 3)}
1627 if s.started is not None and r.started is not None
1628 else {}
1629 ),
1630 "stdout": s.stdout,
1631 "stderr": s.stderr,
1632 "audit": [[e.moment, e.actor, e.code] for e in s.audit],
1633 "failed_at": s.failed_at,
1634 }
1635 )
1636 # The stable machine surface: an envelope so post-1.0 additions (metadata,
1637 # summaries) never have to break consumers of the items list.
1638 print(
1639 json.dumps(
1640 {"schema": 1, "total_ms": round(total * 1000, 3), "items": payload},
1641 indent=2,
1642 default=_describe.json_default,
1643 )
1644 )
1647def _resolve_shell(shell: object, flag: str) -> str | None:
1648 """Resolve *shell* to a supported name for *flag*, or None after `_error`.
1650 A bare mention (no value) detects the invoking shell; an explicit value
1651 is lowercased and de-aliased (`nu`→`nushell`, `powershell`→`pwsh`).
1652 """
1653 from footman import _shellcomp
1655 supported = "|".join(_shellcomp.SHELLS)
1656 if not shell: # named bare: work out which shell is asking
1657 name = _shellcomp.detect_shell()
1658 if name is None:
1659 _error(
1660 f"{flag}: could not detect your shell — "
1661 f"name it explicitly: one of {supported}"
1662 )
1663 return None
1664 else:
1665 name = str(shell or "").lower()
1666 name = {"powershell": "pwsh", "nu": "nushell"}.get(name, name) # muscle-memory
1667 if name not in _shellcomp.SHELLS:
1668 got = f" (got {name!r})" if name else ""
1669 _error(f"{flag} expects one of {supported}{got}")
1670 return None
1671 return name
1674def _completion_action(key: str, shell: object) -> int:
1675 """The completion trio, one body: install and uninstall touch rc files
1676 and echo what they did; setup prints the hook for `eval` (so its
1677 detection note goes to stderr — stdout must stay clean). A bare flag
1678 detects the invoking shell and says which one it worked out.
1679 """
1680 from footman import _shellcomp
1682 flag = "--" + key.replace("_", "-")
1683 name = _resolve_shell(shell, flag)
1684 if name is None:
1685 return EX_USAGE
1686 setup = key == "setup_completion"
1687 if not shell: # named bare: say which shell we worked out
1688 print(f"detected shell: {name}", file=sys.stderr if setup else sys.stdout)
1689 if setup:
1690 print(_shellcomp.script_for(name, _brand.prog))
1691 return 0
1692 act = _shellcomp.install if key == "install_completion" else _shellcomp.uninstall
1693 try:
1694 lines = act(name, _brand.prog)
1695 except _shellcomp.InstallError as exc:
1696 _error(f"{flag} {name}: {exc}")
1697 return EX_USAGE
1698 for line in lines:
1699 print(line)
1700 return 0
1703def run(
1704 argv: list[str],
1705 brand: Brand = DEFAULT_BRAND,
1706 collect: list[_executor.TaskResult] | None = None,
1707) -> int:
1708 """Run the CLI; when *collect* is given, extend it with the TaskResults.
1710 `collect` exists for `footman.testing.Runner`, which needs the structured
1711 results as well as the exit code and printed output.
1712 """
1713 try:
1714 with _signals.installed():
1715 return _run(argv, brand, collect)
1716 except KeyboardInterrupt:
1717 # In --json mode nothing has reached stdout yet (capture buffers task
1718 # output), so the envelope contract still holds at 130.
1719 return _refuse(_wants_json(argv), "interrupted", 130)
1720 except _signals.Stop as stop:
1721 # A supervisor asked for a stop — `timeout`, `docker stop`, a
1722 # cancelled CI job. It is delivered as an exception in the main
1723 # thread on purpose, so it unwinds through the same abort arms Ctrl-C
1724 # does: the in-flight process trees are already reaped by the time it
1725 # lands here, and the envelope holds for the same reason 130's does.
1726 return _refuse(_wants_json(argv), stop.word, stop.code)
1729_WINDOWS = os.name == "nt" # decided at import; a constant tests can steer
1732GC_INTERVAL_S = 24 * 3600
1735def _maybe_collect(cfg: dict[str, object], skip_stem: str) -> None:
1736 """At most daily, and never on a fresh cache, spawn the collector.
1738 *skip_stem* names this invocation's own manifest — the cwd's, or the
1739 shared global one in global mode — so the sweep never eats the file the
1740 run just wrote.
1742 A missing stamp is *planted*, not acted on — the first run a cache ever
1743 sees schedules collection for tomorrow, so short-lived caches (a test
1744 suite's tmp dirs) never spawn anything. An aged stamp is re-touched
1745 *before* spawning, the refresh idiom: concurrent runs elect one
1746 collector, and a crashed child costs a day, not correctness.
1747 """
1748 if cfg.get("gc") is False or os.environ.get(_paths.env_var("NO_GC")):
1749 return
1750 cache = _paths.footman_cache_dir()
1751 stamp = cache / "gc.stamp"
1752 try:
1753 age = time.time() - stamp.stat().st_mtime
1754 except OSError:
1755 with contextlib.suppress(OSError):
1756 cache.mkdir(parents=True, exist_ok=True)
1757 stamp.touch()
1758 return
1759 if age < GC_INTERVAL_S:
1760 return
1761 with contextlib.suppress(OSError):
1762 stamp.touch()
1763 _spawn_gc(cache, skip_stem)
1766def _spawn_gc(cache: Path, skip_stem: str) -> None:
1767 """Detach the collector child through `_complete.detach` — one copy of
1768 the background-child dance, where its Windows story is pinned by tests,
1769 instead of the drift-prone verbatim twin this used to carry."""
1770 from footman import _complete
1772 _complete.detach(
1773 [
1774 sys.executable,
1775 # `-P`, like the completion children: `-c` would otherwise head
1776 # sys.path with the directory the run started in, where a
1777 # `footman.py` would answer the collector's own import.
1778 "-P",
1779 "-c",
1780 "from footman import _gc; _gc.main()",
1781 str(cache),
1782 skip_stem,
1783 ]
1784 )
1787def _find_uv() -> str | None:
1788 """The uv both handoffs use — this runner's own environment, then PATH.
1790 Lives in `_script` so the completion children, which never import this
1791 module, resolve uv exactly the same way.
1792 """
1793 return _script.find_uv()
1796def _reexec(cmd: list[str]) -> None:
1797 """Replace this invocation with *cmd* — the tail every handoff shares.
1799 On POSIX the process is replaced (`execvp`: tty, signals, stdin and the
1800 exit code all belong to the child). Windows `exec*` lies — the parent
1801 exits while the child runs on — so there it spawns and waits, swallowing
1802 its own Ctrl-C and console break (the console delivered both to the child
1803 as well, which will exit 130 or 143 on its own terms).
1804 """
1805 sys.stdout.flush()
1806 sys.stderr.flush()
1807 if _WINDOWS:
1808 proc = subprocess.Popen(cmd)
1809 while True:
1810 try:
1811 raise SystemExit(proc.wait())
1812 except (KeyboardInterrupt, _signals.Stop):
1813 continue
1814 os.execvp(cmd[0], cmd)
1817def _script_hint(exc: object) -> str:
1818 """The sentence a failed import of a *script* tasks file earns.
1820 A file that declares its own dependencies and still failed to import
1821 is almost always one whose environment never got built — no uv, or a
1822 cascade that made the script rule inapplicable. Say so where the
1823 failure is read, rather than leaving a bare ImportError.
1824 """
1825 path = getattr(exc, "path", None)
1826 if path is None or not isinstance(getattr(exc, "original", None), ImportError):
1827 return ""
1828 meta, _warning = _script.read_block(Path(str(path)))
1829 if meta is None or not meta.get("dependencies"):
1830 return ""
1831 return (
1832 f" — {Path(str(path)).name} declares script dependencies, so it "
1833 f"expects its own environment: install uv, name the file alone with "
1834 f"-f, or add the dependencies to this project"
1835 )
1838def _inside(venv: Path) -> bool:
1839 """Whether this interpreter is already running out of *venv* — the rule
1840 lives in `_script`, shared with the completion children."""
1841 return _script.inside(venv)
1844def _locked_project(probe: Path) -> Path | None:
1845 """The nearest ancestor holding a `uv.lock` — one existence walk, no read.
1846 Lives in `_script`, shared with the completion children."""
1847 return _script.locked_project(probe)
1850def _pins_the_runner(root: Path) -> bool:
1851 """Whether *root*'s lockfile pins this runner — the question that decides
1852 whether the invocation belongs to that project's environment. The read
1853 lives in `_script`, shared with the completion children."""
1854 return _script.pins_dist(root, _brand.dist or "footman")
1857def _note_ignored_block(g: dict[str, object], probe: Path) -> None:
1858 """Under `-v` only: say that a script block is being ignored here.
1860 A tasks file may carry `# /// script` metadata and still live inside a
1861 project that pins the runner — a portable file checked into a repo is
1862 exactly that. The project wins, and the block is a non-event: worth
1863 seeing when you asked to see everything, never worth a warning.
1864 """
1865 if not g.get("verbose"):
1866 return
1867 source = _script_source(g, probe)
1868 if source is None: 1868 ↛ 1869line 1868 didn't jump to line 1869 because the condition on line 1868 was never true
1869 return
1870 meta, _warning = _script.read_block(source)
1871 if meta is None or not meta.get("dependencies"): 1871 ↛ 1872line 1871 didn't jump to line 1872 because the condition on line 1871 was never true
1872 return
1873 print(
1874 f"{_brand.prog}: {source.name} declares script dependencies; this "
1875 f"project pins the runner, so they are ignored here",
1876 file=sys.stderr,
1877 )
1880def _script_source(g: dict[str, object], probe: Path) -> Path | None:
1881 """The single tasks file this invocation would load, or None.
1883 A script's environment can only be *the* environment, so the rule is
1884 single-file: an explicit `-f`, or a cascade that found exactly one
1885 file. Resolved through `resolve_task_files` against the `-C` probe —
1886 the same walk the run will do, so the handoff can never disagree with
1887 it — and quiet, because the real run repeats every warning.
1888 """
1889 try:
1890 found = resolve_task_files(g, on_warning=lambda _: None, cwd=probe)
1891 except (_config.ConfigError, _config.CascadeError):
1892 return None # the real run reports these properly
1893 # The user rung never disables a project's script environment: project
1894 # files answer first, and only a machine with nothing but the user file
1895 # runs THAT file's environment — it is a tasks file like any other.
1896 project = [f for f in found.files if f != found.user]
1897 candidates = project or ([found.user] if found.user else [])
1898 return candidates[0] if len(candidates) == 1 else None
1901def _script_handoff(argv: list[str], g: dict[str, object], probe: Path) -> int | None:
1902 """Hand off to a tasks file's own PEP 723 script environment.
1904 A tasks file that declares `dependencies` carries its own world: uv
1905 materialises it (`uv sync --script`), and this invocation continues
1906 inside it (`uv python find --script` → `python -m footman …`). One
1907 portable file then runs anywhere the runner is installed, no project
1908 needed. Returns `None` to say "not my business, carry on"; an exit code
1909 when this invocation is over — a refusal, or uv's own failure.
1911 Never a hard refusal for anything environmental: no uv, a cascade of
1912 several files, an unreadable block — all fall through to running the
1913 file as-is, exactly as before this rule existed. The import that then
1914 fails carries the teaching (`_execute`). The one refusal is a file
1915 that declares dependencies but not the runner it imports, because that
1916 environment provably cannot run it.
1917 """
1918 if os.environ.get(_paths.env_var("UV_REEXEC")) or os.environ.get( 1918 ↛ 1921line 1918 didn't jump to line 1921 because the condition on line 1918 was never true
1919 _paths.env_var("NO_UV")
1920 ):
1921 return None
1922 source = _script_source(g, probe)
1923 if source is None:
1924 return None
1925 meta, warning = _script.read_block(source)
1926 if warning is not None:
1927 _error(f"warning: {warning}")
1928 return None
1929 if meta is None or not meta.get("dependencies"):
1930 return None # not a script, or a block that asks for no world
1931 try:
1932 cfg = _config.load_config(
1933 probe,
1934 _paths.find_repo_root(probe),
1935 str(g["config"]) if g.get("config") else None,
1936 on_warning=lambda _: None, # the real run repeats any warning
1937 )
1938 except _config.ConfigError:
1939 return None # the real run reports the broken --config properly
1940 if not _uv_wanted(g, cfg):
1941 return None
1942 if _brand.dist is None:
1943 # A branded runner can't know which distribution ships it, so it
1944 # can't tell whether the script's environment would contain it.
1945 return None
1946 if not _script.declares(meta, _brand.dist):
1947 _error(
1948 f"{source.name} declares script dependencies but not "
1949 f"{_brand.dist!r} — the environment it asks for cannot import "
1950 f"the runner. Add {_brand.dist!r} to its dependencies, or drop "
1951 f"the script block and run inside a project."
1952 )
1953 return EX_USAGE
1954 # Resolved here, not by the caller: `shutil.which` is a PATH walk (and
1955 # `shutil` itself ~1.9 ms of archive codecs), and every return above this
1956 # line is a run that never needed uv at all — which is most of them.
1957 uv = _find_uv()
1958 if uv is None:
1959 return None # nothing to build the environment with: run as-is
1960 verbose = bool(g.get("verbose"))
1961 if verbose:
1962 print(
1963 f"{_brand.prog}: handing off to the script environment of "
1964 f"{source.name} (uv)",
1965 file=sys.stderr,
1966 )
1967 # A script environment is deliberately not the active one: an unrelated
1968 # VIRTUAL_ENV (a project shell, a `uv run` above us) would only draw a
1969 # warning from uv and confuse anything the tasks spawn, so it leaves
1970 # the picture — here and in the child.
1971 os.environ.pop("VIRTUAL_ENV", None)
1972 try:
1973 synced = subprocess.run(_script.sync_argv(uv, source, quiet=not verbose))
1974 if synced.returncode != 0:
1975 return synced.returncode # uv already said why; don't paraphrase it
1976 found = subprocess.run(
1977 _script.find_argv(uv, source), capture_output=True, text=True
1978 )
1979 except OSError:
1980 return None # uv wouldn't start: run as-is rather than refuse
1981 python = found.stdout.strip()
1982 if found.returncode != 0 or not python: 1982 ↛ 1983line 1982 didn't jump to line 1983 because the condition on line 1982 was never true
1983 return None
1984 os.environ[_paths.env_var("UV_REEXEC")] = "1"
1985 if _brand.dist == "footman":
1986 _reexec([python, "-m", "footman", *argv])
1987 else:
1988 # `-m footman` is the *stock* CLI, so a branded child re-ran this
1989 # handoff — its belt variable above is scoped to the brand, so the
1990 # one just set didn't count — and then refused, because the script
1991 # declares the brand's dist and not 'footman'. The brand's door in
1992 # the script environment is its own console script; loading the
1993 # entry point by name runs exactly what that script runs, on any
1994 # platform, without guessing at the environment's bin layout.
1995 shim = (
1996 "import sys;"
1997 "from importlib.metadata import entry_points;"
1998 f"[ep] = entry_points(group='console_scripts', name={_brand.prog!r});"
1999 "sys.exit(ep.load()())"
2000 )
2001 _reexec([python, "-c", shim, *argv])
2002 return None # unreachable: _reexec replaces or exits this process
2005def _uv_handoff(argv: list[str], g: dict[str, object]) -> int | None:
2006 """Hand this invocation to the environment that owns it.
2008 Two rules, one order. A project whose `uv.lock` pins the runner has
2009 already declared what `fm` means there, so it wins — and a tasks file's
2010 own script metadata is simply not that run's business. Only where no
2011 project has spoken may a tasks file declare its world for itself
2012 (`_script_handoff`).
2014 Returns `None` when nothing applied and the caller carries on, or an
2015 exit code when the invocation is over.
2017 The lock rule, one sentence: when the project's `uv.lock` pins footman
2018 and this interpreter is not already inside the project's environment,
2019 the invocation belongs to `uv run` — the project has declared what `fm`
2020 means there, version and all. Reached only where tasks would be
2021 imported: `--version`, completion management, and the TAB hot path
2022 never arrive here. Opt out with `uv = false` in `[tool.footman]` or
2023 `FOOTMAN_NO_UV=1`. The child carries `FOOTMAN_UV_REEXEC` as a loop
2024 belt for projects whose environment lives outside `.venv`.
2026 The process is replaced through `_reexec`, which owns the POSIX /
2027 Windows difference.
2028 """
2029 if os.environ.get(_paths.env_var("UV_REEXEC")) or os.environ.get(
2030 _paths.env_var("NO_UV")
2031 ):
2032 return None
2033 try:
2034 probe = Path(str(g.get("directory") or Path.cwd())).resolve(strict=True)
2035 except OSError:
2036 return None # a missing -C target: _run's own error path reports it
2037 root = _locked_project(probe)
2038 if root is not None and _inside(root / ".venv") and not g.get("verbose"):
2039 # Already running out of that project's environment, which is the
2040 # overwhelmingly common case — and the one answer that needs no
2041 # lockfile read at all. Asked first, because reading a real
2042 # project's `uv.lock` costs ~21 ms and every invocation paid it
2043 # just to conclude it was already home. Under `-v` the full walk
2044 # runs anyway: its note about an ignored script block is exactly
2045 # what someone asking to see everything is asking for.
2046 return None
2047 if root is None or not _pins_the_runner(root):
2048 # Nobody has declared what the runner means here, so a tasks file
2049 # may declare it for itself.
2050 return _script_handoff(argv, g, probe)
2051 # A pinned project has already declared what the runner means here, so
2052 # a tasks file's own script block is simply not this run's business —
2053 # not a warning, not a refusal; visible under -v and nowhere else.
2054 _note_ignored_block(g, probe)
2055 if _inside(root / ".venv"):
2056 return None # already the project's environment (the -v path lands here)
2057 uv = _find_uv()
2058 if uv is None: 2058 ↛ 2059line 2058 didn't jump to line 2059 because the condition on line 2058 was never true
2059 return None
2060 try:
2061 cfg = _config.load_config(
2062 probe,
2063 _paths.find_repo_root(probe),
2064 str(g["config"]) if g.get("config") else None,
2065 on_warning=lambda _: None, # the real run repeats any warning
2066 )
2067 except _config.ConfigError:
2068 return None # the real run reports the broken --config properly
2069 if not _uv_wanted(g, cfg):
2070 return None
2071 if g.get("verbose"): 2071 ↛ 2072line 2071 didn't jump to line 2072 because the condition on line 2071 was never true
2072 print(
2073 f"{_brand.prog}: handing off to uv run --project {root}",
2074 file=sys.stderr,
2075 )
2076 os.environ[_paths.env_var("UV_REEXEC")] = "1"
2077 _reexec([uv, "run", "--project", str(root), _brand.prog, *argv])
2078 return None # unreachable: _reexec replaces or exits this process
2081_SYNC_HINT = " — the project's environment may be out of date: run uv sync"
2084def _uv_retry(
2085 argv: list[str], g: dict[str, object], cfg: dict[str, Any], exc: BaseException
2086) -> str:
2087 """A failed import in a pinned project may just be a stale environment.
2089 The lock rule's "already home" answer trusts that the project's venv
2090 matches its lockfile; a dependency locked after the last sync breaks
2091 that trust, and the first symptom is exactly this import failure. So:
2092 hand the invocation to `uv run --project` after all — it syncs, then
2093 retries — and let a failure *after* that surface unchanged (the child
2094 carries the loop belt, which doubles as "sync was already tried").
2096 Re-execs and never returns when the retry applies. Otherwise returns
2097 the sentence to append to the refusal: the `uv sync` hint when a
2098 pinned project exists but the retry cannot run (no uv, opted out), or
2099 nothing at all — a non-import failure, no pinned project, or a run
2100 that already came through `uv run`.
2101 """
2102 if os.environ.get(_paths.env_var("UV_REEXEC")) or os.environ.get(
2103 _paths.env_var("NO_UV")
2104 ):
2105 return "" # uv run already synced (or was told not to): the failure is real
2106 if not _script.import_caused(exc):
2107 return "" # a syntax error or a raise — no sync would mend it
2108 root = _locked_project(Path.cwd())
2109 if root is None or not _pins_the_runner(root):
2110 return ""
2111 if not _uv_wanted(g, cfg):
2112 return _SYNC_HINT # asked to stay out of uv: advise, don't act
2113 uv = _find_uv()
2114 if uv is None:
2115 return _SYNC_HINT
2116 if g.get("verbose"): 2116 ↛ 2117line 2116 didn't jump to line 2117 because the condition on line 2116 was never true
2117 print(
2118 f"{_brand.prog}: import failed; retrying through uv run --project {root}",
2119 file=sys.stderr,
2120 )
2121 os.environ[_paths.env_var("UV_REEXEC")] = "1"
2122 _reexec([uv, "run", "--project", str(root), _brand.prog, *argv])
2123 return "" # unreachable: _reexec replaces or exits this process
2126def _run(
2127 argv: list[str],
2128 brand: Brand,
2129 collect: list[_executor.TaskResult] | None = None,
2130 *,
2131 handoff: bool = True,
2132) -> int:
2133 global _brand
2134 _brand = brand
2135 try:
2136 # Before anything writes: a cache and a data directory that are the
2137 # same place would have the collector deleting durable things.
2138 _paths.check_locations()
2139 except _paths.LocationError as exc:
2140 return _refuse(_wants_json(argv), str(exc))
2141 try:
2142 pre_globals, after = _split._parse_globals(argv, 0, lenient=True)
2143 except _split.ChainError as exc:
2144 return _refuse(_wants_json(argv), str(exc))
2145 g = _globals_to_dict(pre_globals)
2146 _set_colors(_resolve_color(g))
2147 wants_help = _wants_help(argv)
2149 if g.get("version"): # D7: --version wins even over --help
2150 return _print_version(bool(g.get("json")))
2151 # Asking for help must never touch the filesystem: `--install-completion
2152 # fish --help` used to write rc files before printing anything.
2153 for key in ("install_completion", "setup_completion", "uninstall_completion"):
2154 if key in g and not wants_help:
2155 value = g.get(key)
2156 if not value and after < len(argv) and not argv[after].startswith("-"):
2157 # A word behind the bare action has nowhere else to go — these
2158 # end the invocation, so there is no task chain for it to be
2159 # part of. Teach the `=` form rather than acting on the
2160 # detected shell and leaving the word unexplained.
2161 flag = "--" + key.replace("_", "-")
2162 return _refuse(
2163 bool(g.get("json")),
2164 _split._expects_value(None, flag, "[SHELL]", argv[after]),
2165 )
2166 return _completion_action(key, value)
2167 # May replace the process (POSIX) or exit with the child's code
2168 # (Windows); returns quietly whenever the handoff doesn't apply.
2169 # `handoff=False` is the embedded-harness escape: `Runner.invoke` runs
2170 # inside a host process (pytest — possibly a pytest-xdist worker whose
2171 # stdio IS the test-protocol channel), and an execvp there replaces the
2172 # host itself. An embedded invocation must always run in-process.
2173 if handoff and (code := _uv_handoff(argv, g)) is not None:
2174 # A handoff that neither replaced this process nor stayed out of the
2175 # way has ended the invocation: a refusal, or uv's own failure.
2176 return code
2178 if not g.get("directory"):
2179 return _execute(argv, g, pre_globals, after, wants_help, collect, handoff)
2181 # -C must not permanently move the process (a `Runner.invoke` shares the
2182 # host pytest's cwd): chdir, run, then restore in a finally. The original
2183 # dir may have vanished mid-run, so the restore is best-effort.
2184 saved_cwd = os.getcwd()
2185 try:
2186 os.chdir(str(g["directory"]))
2187 except OSError as exc:
2188 return _refuse(bool(g.get("json")), f"-C {g['directory']}: {exc}")
2189 try:
2190 return _execute(argv, g, pre_globals, after, wants_help, collect, handoff)
2191 finally:
2192 with contextlib.suppress(OSError):
2193 os.chdir(saved_cwd)
2196def _execute(
2197 argv: list[str],
2198 g: dict[str, object],
2199 pre_globals: list[str],
2200 after: int,
2201 wants_help: bool,
2202 collect: list[_executor.TaskResult] | None,
2203 handoff: bool,
2204) -> int:
2205 """Discover the cascade, load + sync its manifest, then run the tree.
2207 Everything after globals/`--version`/`--install-completion`/`-C`: the
2208 disk-backed half that `run_group` (in-memory) deliberately skips.
2209 *g* and *after* are the entry point's one lenient parse, passed down —
2210 argv is never re-parsed on the way in. *handoff* rides along for the
2211 stale-environment retry: an embedded `Runner.invoke` must never exec.
2212 """
2213 # Before discovery, because discovery already consults the built-in
2214 # set: a `[builtins]` table footman cannot read would otherwise be
2215 # swallowed into "no tasks file found" — the one answer that hides the
2216 # mistake instead of naming it.
2217 try:
2218 _config.discovery_mode()
2219 _config.user_builtin()
2220 except _config.ConfigError as exc:
2221 return _refuse(bool(g.get("json")), str(exc))
2222 # "Bare" means no chain was asked for — globals-only lines (`fm --json`,
2223 # `fm -k`) are listing-shaped, exactly like they are when tasks exist.
2224 found = _discover_files(g, wants_help, bare=after >= len(argv))
2225 if isinstance(found, int):
2226 return found
2227 files, cfg = found.files, found.cfg
2228 json_mode = bool(g.get("json"))
2230 base = registry.Group("root")
2231 if (base_set := _builtin()) and not g.get("tasks_file"):
2232 # The built-in set is the cascade's outermost rung — under the user
2233 # rung, under the project's own files — so the full ladder is
2234 # project > user > built-in wherever you stand. Everything nearer
2235 # shadows it by name, exactly as the cascade already resolves its
2236 # own rungs, which is what lets a project own a name a built-in
2237 # also uses without either of them being special.
2238 #
2239 # `expose` is what keeps this honest in both directions: a package's
2240 # tasks are `project_only` until one opts out, so joining the
2241 # cascade offers nothing outside a project that did not say it
2242 # belonged there. (`-f` still means total control: one file, no
2243 # cascade, no base.)
2244 built = _base_tree(base_set, json_mode)
2245 if isinstance(built, int):
2246 return built
2247 base = built
2248 plugins_cfg = cfg.get("plugins")
2249 if plugins_cfg and not isinstance(plugins_cfg, dict):
2250 # The old list-valued key died with the composition rework: mounts are
2251 # authored in tasks.py, where placement, filtering, and overrides
2252 # live. A *table* is the reserved sections child instead —
2253 # `[tool.footman.plugins.<section>]` holds a provider's own settings.
2254 return _refuse(
2255 json_mode,
2256 f"the [tool.{_paths.config_table()}] plugins key was removed — "
2257 "mount plugins from "
2258 'tasks.py instead: plugin("footman.docs", into="footman") '
2259 "(footman.compose.plugin; see the composing docs)",
2260 )
2261 # The note-level policy: validated loudly (an ignored policy line is a
2262 # wall someone thinks is up), then installed for the run — always, so a
2263 # previous embedded invocation's table never leaks into this one.
2264 notes_cfg = cfg.get("notes")
2265 if notes_cfg is not None and (error := _notes.validate(notes_cfg)) is not None:
2266 return _refuse(json_mode, error)
2267 _notes.install_policy(notes_cfg)
2268 # The docs-link template: validated loudly (a link scheme that silently
2269 # pointed nowhere would be worse than none), installed for the run —
2270 # always, so a previous embedded invocation's template never leaks.
2271 global _docs_url
2272 docs_cfg = cfg.get("docs_url")
2273 if docs_cfg is not None and (bad := _describe.docs_url_error(docs_cfg)) is not None:
2274 return _refuse(json_mode, bad)
2275 _docs_url = docs_cfg if isinstance(docs_cfg, str) else None
2277 inv = invocation.Invocation(
2278 cli=g,
2279 config=cfg,
2280 # The project cascade's top — never the user rung's directory. `""`
2281 # is global mode: footman invents no root where there is no project.
2282 root=found.root,
2283 cwd=os.getcwd(),
2284 )
2285 try:
2286 reg = _discover.load_tree(files, base=base, inv=inv)
2287 except _discover.HookError as exc:
2288 # A hook that raised is a refusal, named — nothing has run yet.
2289 return _refuse(json_mode, str(exc))
2290 except _discover.TasksImportError as exc:
2291 # A stale project environment produces exactly this failure; when
2292 # the lock rule owns this directory, hand the run to `uv run` (it
2293 # syncs, then retries) instead of reporting a mystery — or at least
2294 # name the fix. Never from an embedded Runner: no exec in a host.
2295 hint = _uv_retry(argv, g, cfg, exc.original) if handoff else ""
2296 if isinstance(exc.original, registry.RegistrationError):
2297 # a user mistake, not a crash
2298 return _refuse(json_mode, f"{exc.path}: {exc.original}{hint}")
2299 return _refuse(
2300 json_mode,
2301 f"failed to import {exc.path}: "
2302 f"{type(exc.original).__name__}: {exc.original}"
2303 f"{_script_hint(exc)}{hint}",
2304 )
2305 except Exception as exc: # report import failures cleanly, don't crash
2306 return _refuse(
2307 json_mode,
2308 f"failed to import the task cascade: {type(exc).__name__}: {exc}",
2309 )
2310 if ( 2310 ↛ 2313line 2310 didn't jump to line 2313 because the condition on line 2310 was never true
2311 clash := registry.validate_global_options(reg.contributions["globals"])
2312 ) is not None:
2313 return _refuse(json_mode, clash)
2314 for orphan in registry.orphan_global_options(reg):
2315 _error(f"warning: {orphan}")
2317 # The personal rung's own directory: what tells a user task from a
2318 # project one when the listings split the two. `None` when the cascade
2319 # mounted no user file, which simply means nothing can match it.
2320 user_dir = str(found.user.parent) if found.user is not None else None
2321 try:
2322 if g.get("tasks_file"):
2323 # -f loads one arbitrary file, not the cwd's cascade. Cache its
2324 # manifest under a (cwd, file) key — separate from the cwd's, so it
2325 # never poisons plain TAB there — so `fm -f <file> <TAB>` completes
2326 # that file's tasks. max_age=0: no background refresh (rebuilt on the
2327 # next -f run); a live refresh is a fast-follow.
2328 override = str(g.get("tasks_file"))
2329 tree = _manifest.sync_manifest(
2330 reg,
2331 Path.cwd(),
2332 user_dir=user_dir,
2333 completion_max_age=0,
2334 tasks_file=override,
2335 path=_paths.source_manifest_path(Path.cwd(), Path(override)),
2336 )["tree"]
2337 elif config_path := _config_arg(g):
2338 # An explicit --config reshapes the tree (its own tasks name, its
2339 # own keys), so its manifest rides a (cwd, config) key — separate
2340 # from the plain cwd's, which must keep answering plain TAB.
2341 # max_age=0, like -f: no background refresh, rebuilt on the next
2342 # such run.
2343 cfg_tasks = cfg.get("tasks")
2344 tree = _manifest.sync_manifest(
2345 reg,
2346 Path.cwd(),
2347 user_dir=user_dir,
2348 completion_max_age=0,
2349 tasks_file=cfg_tasks
2350 if isinstance(cfg_tasks, str)
2351 else _brand.tasks_file,
2352 path=_paths.source_manifest_path(Path.cwd(), Path(config_path)),
2353 )["tree"]
2354 elif not found.root:
2355 # Global mode: one manifest for every project-less directory,
2356 # keyed by the brand rather than the cwd — cold once per brand
2357 # version, not once per directory. No baked cwd (the collector's
2358 # idle sweep owns it), and the builtin names ride so the refresh
2359 # child can rebuild a tree it cannot otherwise know.
2360 cfg_tasks = cfg.get("tasks")
2361 tree = _manifest.sync_manifest(
2362 reg,
2363 Path.cwd(),
2364 user_dir=user_dir,
2365 completion_max_age=_config.completion_max_age(cfg, strict=True),
2366 tasks_file=cfg_tasks
2367 if isinstance(cfg_tasks, str)
2368 else _brand.tasks_file,
2369 path=_paths.global_manifest_path(),
2370 bake_cwd=False,
2371 builtin=_builtin(),
2372 # No project here — the one place the question is asked, so
2373 # every reader downstream just checks `unexposed`.
2374 project=False,
2375 )["tree"]
2376 else:
2377 cfg_tasks = cfg.get("tasks")
2378 tree = _manifest.sync_manifest(
2379 reg,
2380 Path.cwd(),
2381 user_dir=user_dir,
2382 completion_max_age=_config.completion_max_age(cfg, strict=True),
2383 tasks_file=cfg_tasks
2384 if isinstance(cfg_tasks, str)
2385 else _brand.tasks_file,
2386 )["tree"]
2387 except _manifest.ManifestError as exc: # broken completer, bad markers, …
2388 return _refuse(json_mode, str(exc))
2389 except _config.ConfigError as exc: # a mistyped completion.max_age
2390 return _refuse(json_mode, str(exc))
2392 # The `root` policy token's target: the project cascade's top, never the
2393 # user rung's directory — a personal task's `cwd="root"` means the
2394 # project it landed in, and outside one the empty root exhausts the
2395 # ladder to the invocation directory.
2396 root_dir = found.root
2397 # Arm the per-task lifecycle for exactly this run: every execution —
2398 # segment, prerequisite, fan-out member, body call — reaches the same
2399 # ladder through `run_bound`, and the frozen invocation rides along.
2400 _executor.install_lifecycle(inv, reg.contributions)
2401 try:
2402 code = _run_tree(
2403 reg, tree, argv, cfg, collect, g, pre_globals, after, root_dir=root_dir
2404 )
2405 finally:
2406 _executor.clear_lifecycle()
2407 # Core's ladder instances release with the plugins': they are
2408 # module-level singletons, and a Runner drives many runs in one
2409 # process — an outside-a-run read goes back to teaching.
2410 registry.release_global_options(
2411 (*_split.CORE_LADDER, *reg.contributions["globals"])
2412 )
2413 # After the run, so it never adds latency before the user's command —
2414 # and after the uv handoff by construction (the handoff replaced this
2415 # process back in _run), so a pinned project's own footman collects.
2416 _maybe_collect(
2417 cfg,
2418 skip_stem=(
2419 _paths.global_manifest_path().stem
2420 if not found.root and not g.get("tasks_file")
2421 else _paths.manifest_path(Path.cwd()).stem
2422 ),
2423 )
2424 return code
2427def _run_tree(
2428 reg: registry.Group,
2429 tree: dict[str, Any],
2430 argv: list[str],
2431 cfg: dict[str, object],
2432 collect: list[_executor.TaskResult] | None,
2433 g: dict[str, object],
2434 pre_globals: list[str],
2435 after: int,
2436 root_dir: str = "",
2437 record_times: bool = True,
2438) -> int:
2439 """The post-manifest tail: help/where/split/list/tree/dry-run/run/report.
2441 Shared by the disk path (`_execute`) and the in-memory path (`run_group`),
2442 so both honour `--help`/`--version`/`--list`/`--tree`/`--json` identically.
2443 *g* and *after* ride down from the entry point's one lenient parse.
2444 """
2445 json_mode = bool(g.get("json"))
2447 # Core's ladder-bearing options bind first — before the colour repaint,
2448 # the sort, and the listings, which all read them. One machine with the
2449 # plugin bind below; a bad value (`--jobs=abc`, `--color=hi`) or a broken
2450 # config key refuses here, even when a listing would have exited first —
2451 # eager, like every other parse refusal. Broken config teaches on every
2452 # invocation, not only the ones it would steer.
2453 if (
2454 bad := _executor.bind_global_options(
2455 _split.CORE_LADDER, pre_globals, config=cfg
2456 )
2457 ) is not None:
2458 return _refuse(json_mode, bad)
2459 # Config can set the mode too, so re-resolve now that the ladder is in
2460 # hand and repaint footman's own chrome to match (the pre-run call saw
2461 # CLI + environment only).
2462 color_mode = _resolve_color(g, bound=True)
2463 _set_colors(color_mode)
2465 # Presentation only: the sorted copy feeds every human-facing walk
2466 # (--list, --tree, help, the --json catalog). The run resolves through
2467 # the registry, so execution order never follows this setting.
2468 if _split.SORT.value:
2469 tree = _describe.sort_tree(tree)
2471 show_hidden = bool(g.get("all"))
2473 live = _choices_resolver(reg)
2474 if _wants_help(argv):
2475 return _print_help(tree, argv, after, show_hidden, live)
2477 if g.get("plugins"):
2478 return _plugins_report(reg)
2480 if g.get("where"):
2481 # Deliberately plain under --json too: `file:line` already is the
2482 # machine format.
2483 return _where(reg, tree, str(g["where"]))
2485 if (describe := g.get("describe")) is not None:
2486 return _describe_contract(tree, describe, after < len(argv))
2488 try:
2489 globals_, segments = _split.split_chain(tree, argv, live)
2490 except _split.ChainError as exc:
2491 return _refuse(json_mode, str(exc))
2492 except _manifest.CompleterError as exc:
2493 # A strict completer raised while the line was being validated. It
2494 # used to surface from the manifest build, before any line was read;
2495 # now it surfaces where its values were actually needed — same taught
2496 # message, and a broken completer on one task no longer refuses every
2497 # other task's invocation.
2498 return _refuse(json_mode, str(exc))
2500 # A task whose signature claims stdout (`-> Stdout[T]`) makes this a
2501 # *document run*: stdout carries exactly the addressed task's return
2502 # value, and every print and run() line replays on stderr instead. Two
2503 # claimants leave "whose document?" without an answer, so that is a
2504 # plan-time refusal rather than a silent pick. Only the addressed tasks
2505 # count — a declaring task reached as a dependency or through a group
2506 # fan-out is suppressed, not refused, so composing a filter into a
2507 # bigger task stays legal.
2508 emitters: list[tuple[_split.Segment, object]] = []
2509 for seg in segments:
2510 try:
2511 seg_fn = _executor.resolve(reg, seg.task.split("."))
2512 except (KeyError, IndexError):
2513 continue # the splitter validated; never refuse twice
2514 declares, inner = _coerce.emitted(
2515 _manifest.resolved_signature(seg_fn).return_annotation
2516 )
2517 if declares:
2518 emitters.append((seg, inner))
2519 if len(emitters) > 1:
2520 names = " and ".join(s.task for s, _ in emitters)
2521 return _refuse(
2522 json_mode,
2523 f"{names} both declare Stdout[…] — whose document would stdout "
2524 f"carry? Run them as separate invocations.",
2525 )
2527 # Advisory notes from the splitter (a group default's positional value
2528 # that names — or nearly names — a child task): stderr commentary, ahead
2529 # of the run, so the plan stays deterministic but never silent. Skipped
2530 # under --json: the envelope's stdout is the contract and the notes are
2531 # human-facing teaching, not results.
2532 if not json_mode:
2533 for seg in segments:
2534 for note in seg.notes:
2535 print(note.replace("{prog}", _brand.prog), file=sys.stderr)
2537 if not segments:
2538 if json_mode:
2539 # The catalog envelope: the manifest tree, params and all — the
2540 # machine twin of --list/--tree (and of bare `fm`).
2541 print(json.dumps({"schema": 1, "tree": tree}, indent=2))
2542 return 0
2543 if g.get("tree"):
2544 _print_tree(tree, show_hidden)
2545 else:
2546 _print_list(tree, show_hidden)
2547 if tree["tasks"] or tree["groups"]:
2548 _print_footer()
2549 return 0
2551 # --dry-run is a rehearsal, not a parse echo: the run proceeds with
2552 # `dry_run` on the context, bodies run, and everything footman owns —
2553 # recorded run() calls, tools, deferred steps — is faked into honest
2554 # plan-line receipts. The report shapes (--json included) are the plan.
2555 dry_run = bool(g.get("dry_run"))
2556 sequential = bool(_split.SEQUENTIAL.value)
2558 # The parallel width, from the one ladder: -j/--jobs wins, then config
2559 # `jobs`, then the declared computed default (cores minus one) — bound
2560 # above, so what `--help` prints and what the run caps at come from one
2561 # declaration, and a bad width (`--jobs=abc`, `--jobs=0`, `jobs = 0` in
2562 # config) was already a taught refusal. Caps both engines (the
2563 # scheduler's pool and parallel() in task bodies) and is part of the
2564 # timing key — a -j2 run has a genuinely different duration distribution.
2565 jobs = int(_split.JOBS.value)
2567 fetch_cfg = cfg.get("fetch")
2568 backend = fetch_cfg.get("backend") if isinstance(fetch_cfg, dict) else None
2569 shell_cfg = cfg.get("shell")
2570 shell_default = shell_cfg.get("default") if isinstance(shell_cfg, dict) else None
2571 cwd_cfg = cfg.get("cwd")
2572 cwd_policy = cwd_cfg if isinstance(cwd_cfg, str) else ""
2573 if (
2574 cwd_policy
2575 and cwd_policy not in registry.CWD_TOKENS
2576 and not Path(cwd_policy).is_absolute()
2577 ):
2578 return _refuse(
2579 json_mode,
2580 f"config cwd = {cwd_policy!r} is not a policy token "
2581 f"({', '.join(registry.CWD_TOKENS)}) or an absolute path — "
2582 f"a relative suffix belongs on a task's rel=…",
2583 )
2584 ctx_config = {
2585 "fetch_backend": backend if isinstance(backend, str) else "",
2586 "shell_default": shell_default if isinstance(shell_default, str) else "",
2587 # The cwd policy ladder's run-wide rungs: the config default, and the
2588 # two pinned directions tokens resolve against — the highest cascade
2589 # file's directory (`root`) and the launch cwd snapshot (`asinvoked`).
2590 "cwd_policy": cwd_policy,
2591 "root_dir": root_dir,
2592 "invoked_dir": str(Path.cwd()),
2593 "quiet": bool(g.get("quiet")),
2594 # Every step has a row in the envelope, so footman's own receipt
2595 # lines would only arrive twice — once as chrome inside a task's
2596 # `output` string, once as the fields a reader actually parses.
2597 "machine_read": json_mode,
2598 "verbose": bool(g.get("verbose")),
2599 # The resolved tri-state, split into the two Context bits: `never` stops
2600 # all colour, `always` forces it past a non-terminal (the scheduler still
2601 # gates that off under capture). `auto` leaves both false — tty decides.
2602 "no_color": color_mode == "never",
2603 "force_color": color_mode == "always",
2604 # Tasks can know who invoked them (a branded CLI's prog) — the
2605 # taskdocs plugin brands its output with this, for one.
2606 "prog": _brand.prog,
2607 # The user's -s/config request, so parallel() in task bodies
2608 # serialises too — not the scheduler's single-node routing.
2609 "sequential": sequential,
2610 "jobs": jobs,
2611 # Interactivity globals: --yes auto-answers confirm() gates, --no-input
2612 # refuses to prompt (a required prompt errors instead of hanging).
2613 "assume_yes": bool(g.get("yes")),
2614 "no_input": not _split.INPUT.value,
2615 # The rehearsal switch: bodies run, footman's own work is faked.
2616 "dry_run": dry_run,
2617 }
2619 # The timing story: `--no-progress` or `progress = false` in config turns
2620 # the whole apparatus off, and `--progress` turns it back on for one run —
2621 # config is a default, never a one-way door. A run is
2622 # *predictable* when it's on, every task consented, and this is the real
2623 # cascade (-f runs pollute no cache, times included) — only then do we
2624 # estimate from history and record the outcome.
2625 # A rehearsal is near-instant and teaches nothing about durations: no
2626 # bar, no eta, and (below) no recorded timing to pollute the history.
2627 progress_on = bool(_split.PROGRESS.value) and not dry_run
2628 predictable = (
2629 progress_on
2630 # Synthetic runs pollute no cache, times included: `-f` runs and
2631 # in-memory trees (`run_group`) both estimate nothing and record
2632 # nothing, or a consumer's Runner-driven test suite would write
2633 # timing history into the real user cache.
2634 and record_times
2635 and not g.get("tasks_file")
2636 and _schedule.dag_wants_progress(reg, segments)
2637 )
2638 est = times_key = None
2639 context.seed_cmd_width(0) # each run learns (or is seeded) afresh
2640 if predictable:
2641 times_key = _progress.chain_key(segments, sequential=sequential, jobs=jobs)
2642 est = _progress.estimate(_progress.load_runs(Path.cwd(), times_key))
2643 context.seed_cmd_width(_progress.load_cmd_width(Path.cwd(), times_key))
2644 if est is not None and not g.get("quiet") and not sys.stderr.isatty():
2645 # No TTY (CI, a pipe): the one-line version of the bar, up front.
2646 print(f" {'eta':>4} ~{_progress.fmt_secs(est.typical)}", file=sys.stderr)
2648 # Tri-state on the command line: `-k` forces keep-going, `--fail-fast` forces
2649 # fail-fast, neither leaves it to the invoked task's declared policy.
2650 cli_keep_going = True if g.get("keep_going") else None
2651 if g.get("fail_fast"): 2651 ↛ 2652line 2651 didn't jump to line 2652 because the condition on line 2651 was never true
2652 cli_keep_going = False
2654 # A mounted plugin's globals bind now — after every listing exit,
2655 # before anything runs — and freeze for the run; `.value` answers from
2656 # here. Released by the caller's finally, so an outside-a-run read goes
2657 # back to teaching.
2658 if (
2659 bad := _executor.bind_global_options(
2660 reg.contributions["globals"], globals_, config=cfg
2661 )
2662 ) is not None:
2663 return _refuse(json_mode, bad)
2664 start = time.perf_counter()
2665 try:
2666 results = _schedule.run_plan(
2667 reg,
2668 segments,
2669 sequential=sequential,
2670 keep_going=cli_keep_going, # None = unspecified; run_plan scopes per node
2671 # A document run buffers exactly as --json does: stdout belongs
2672 # to the declared return value, so no task print may stream to it.
2673 capture=json_mode or bool(emitters),
2674 ctx_config=ctx_config,
2675 estimate=est,
2676 progress=progress_on,
2677 jobs=jobs,
2678 )
2679 except _split.ChainError as exc: # e.g. passthrough with no *args
2680 return _refuse(json_mode, str(exc))
2681 total = time.perf_counter() - start
2683 if collect is not None:
2684 collect.extend(results)
2685 # The run report's moment: every row is in, nothing has printed. A
2686 # rewrite a hook makes through its result view is what gets reported.
2687 post_error = _executor.run_post_tasks(results, total, json_mode)
2688 if predictable and times_key and results and all(r.ok for r in results):
2689 # Green runs teach: the duration, and the step-alignment width.
2690 _progress.record(Path.cwd(), times_key, total, cmd_width=context.cmd_width())
2692 # After the post hooks: what a `set_returned` rewrite reported is what
2693 # the contract check reads, the same value the envelope carries.
2694 returns_meta = _check_returns(reg, results)
2696 if json_mode:
2697 _print_json(results, total=total, returns=returns_meta)
2698 else:
2699 if emitters:
2700 # The document run's receipts: everything that is not the
2701 # document — prints, run() lines — replays on stderr, where the
2702 # summary already lives, in dependency order.
2703 for r in results:
2704 if r.output:
2705 sys.stderr.write(r.output)
2706 sys.stderr.flush()
2707 if not g.get("quiet"):
2708 _print_summary(
2709 results,
2710 timings=bool(g.get("timings")),
2711 verbose=bool(g.get("verbose")),
2712 total=total,
2713 )
2714 if emitters:
2715 doc_seg, doc_inner = emitters[0]
2716 doc_result = next((r for r in results if r.task == doc_seg.task), None)
2717 # A failed task emits nothing (the exit code talks), and a None
2718 # return means empty stdout: nothing to say, said nothing.
2719 if (
2720 doc_result is not None
2721 and doc_result.ok
2722 and doc_result.returned is not None
2723 ):
2724 _emit_document(doc_result.returned, doc_inner)
2726 # The exit code is the first genuine failure's — a cancelled task carries
2727 # only a kill signal and a skipped node only a cause, so those are the
2728 # fallback, never the headline.
2729 if post_error is not None:
2730 # A reporter that crashed must not pass silently: named, non-zero.
2731 print(f"{_brand.prog}: {post_error}", file=sys.stderr)
2732 # `retried` is recorded but is never the run's verdict — an attempt with
2733 # attempts left has not failed yet, so there is nothing for the exit code
2734 # to report. The terminal attempt carries the outcome, whichever way it
2735 # went. Filtered here, at the source, so the fallback below cannot pick a
2736 # retried row up either.
2737 failed = [r for r in results if not r.ok and r.state != "retried"]
2738 genuine = next(
2739 (r.code or 1 for r in failed if not r.cancelled and r.state != "skipped"),
2740 None,
2741 )
2743 def carriable(code: int) -> int:
2744 # This number's destiny is a process exit status, and POSIX keeps only
2745 # the low byte: a task's 256 would report *success* to the shell, so
2746 # `fm deploy || rollback` never rolls back. A failure the shell cannot
2747 # carry collapses to 1 — the failure survives, and the real number
2748 # stays on the receipt line and the `--json` row.
2749 return code if 0 < code < 256 else 1
2751 if genuine is not None:
2752 return carriable(genuine)
2753 code = next((r.code or 1 for r in failed), 0)
2754 if code: 2754 ↛ 2755line 2754 didn't jump to line 2755 because the condition on line 2754 was never true
2755 return carriable(code)
2756 return 1 if post_error is not None else 0
2759def run_group(
2760 root: registry.Group,
2761 argv: list[str],
2762 brand: Brand = DEFAULT_BRAND,
2763 collect: list[_executor.TaskResult] | None = None,
2764) -> int:
2765 """Drive an in-memory Group tree: globals, `--version`, manifest, run.
2767 The in-memory sibling of `_run`, minus discovery/cascade/config and the
2768 `-C`/`--install-completion` machinery those imply. No KeyboardInterrupt
2769 wrapper (D13): a test runner must let Ctrl-C reach pytest. This is the
2770 single shared surface `footman.testing.Runner` drives, so its Group mode
2771 can never drift from the real CLI's help/version/list/tree/json behaviour.
2772 """
2773 global _brand
2774 _brand = brand
2775 try:
2776 pre_globals, after = _split._parse_globals(argv, 0, lenient=True)
2777 except _split.ChainError as exc:
2778 return _refuse(_wants_json(argv), str(exc))
2779 g = _globals_to_dict(pre_globals)
2780 _set_colors(_resolve_color(g))
2782 if g.get("version"):
2783 return _print_version(bool(g.get("json")))
2785 # No config here (in-memory trees have no cascade), so the note-level
2786 # policy is the defaults and there is no docs-link template — installed
2787 # rather than inherited, or a prior embedded invocation's would leak.
2788 _notes.install_policy(None)
2789 global _docs_url
2790 _docs_url = None
2791 tree = _manifest.build_manifest(root)["tree"]
2792 # An in-memory tree still gets the per-task lifecycle — its hooks live on
2793 # the Group's own contributions. `pre_tasks` stays a discovery-time moment
2794 # (there is no cascade here), so the invocation arrives already frozen.
2795 inv = invocation.Invocation(cli=g, cwd=os.getcwd(), tasks=registry.Tasks(root))
2796 inv.freeze()
2797 if (
2798 clash := registry.validate_global_options(root.contributions["globals"])
2799 ) is not None:
2800 return _refuse(bool(g.get("json")), clash)
2801 for orphan in registry.orphan_global_options(root):
2802 _error(f"warning: {orphan}")
2803 _executor.install_lifecycle(inv, root.contributions)
2804 try:
2805 return _run_tree(
2806 root, tree, argv, {}, collect, g, pre_globals, after, record_times=False
2807 )
2808 finally:
2809 _executor.clear_lifecycle()
2810 registry.release_global_options(
2811 (*_split.CORE_LADDER, *root.contributions["globals"])
2812 )