Coverage for src/footman/_manifest.py: 95%
584 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-06 02:59 +0000
1"""Manifest generation and caching — the "cold" path.
3The manifest is a JSON description of the command tree (groups, tasks, and the
4CLI shape of every parameter). The execution path imports the user's tasks
5module anyway, so introspecting the tree and rewriting the cache is effectively
6free. The completion hot path (`footman._complete`) only ever *reads* the
7cached JSON — it never imports this module or the user's code.
9Parameter mapping (function signature -> CLI shape):
11| Signature | CLI shape |
12| ------------------------ | ----------------------------------------- |
13| `fix: bool = False` | flag `--fix` / `--no-fix` |
14| `mode: str = "loose"` | option `--mode=VALUE`, bare `--mode` ok |
15| `env: Literal[...]` | completable, eagerly-validated choices |
16| `count: int = 100` | typed option, validated at parse time |
17| `paths: list[Path] = ()` | repeatable option (`--paths a --paths b`) |
18| `template: Path` | required positional (exact arity) |
19| `*cmd: str` | variadic trailing passthrough |
20"""
22from __future__ import annotations
24import contextlib
25import dataclasses
26import datetime
27import decimal
28import enum
29import hashlib
30import inspect
31import json
32import os
33import sys
34import typing
35import uuid
36from pathlib import Path, PurePath
37from typing import Any
39from footman import _binder, _coerce, _describe, _discover, _paths, registry
40from footman.context import context_param_name
41from footman.params import suggest
42from footman.registry import Group
44SCHEMA_VERSION = 11
46_warned: set[str] = set()
49def _warn(message: str) -> None:
50 """An advisory on stderr, once per message — never a Python UserWarning.
52 `warnings.warn` dressed these as `…/_manifest.py:416: UserWarning:` with
53 footman's own source line quoted underneath — internals nobody asked
54 about, above every `--help` and `--list`, pointing at the framework
55 instead of the user's file the message already names. The message is
56 the whole story; it is delivered as one line, once, like every other
57 advisory footman prints.
58 """
59 if message in _warned: 59 ↛ 60line 59 didn't jump to line 60 because the condition on line 59 was never true
60 return
61 _warned.add(message)
62 print(message, file=sys.stderr)
65class ManifestError(Exception):
66 """A tasks file describes a command surface footman cannot honour.
68 Raised at manifest-build time (the execution path) with a taught message;
69 the app layer reports it and exits 2.
70 """
73class CompleterError(ManifestError):
74 """A strict dynamic completer failed while refreshing its choices.
76 Raised so a broken completer surfaces as a taught error instead of
77 silently baking an empty choice list — which would disable the very
78 validation `strict=True` promises.
79 """
82class SpecError(ManifestError):
83 """A parameter's markers are inconsistent (e.g. `env()` with no default)."""
86_SIGNATURE = "_footman_signature"
87_CALL_SIGNATURE = "_footman_call_signature"
90def _memo(fn: Any, attr: str) -> tuple[Any, inspect.Signature | None]:
91 """The body to stamp a signature memo on, and the memo already there.
93 Keyed on the **body**, through `task_body`, so a handle, an `.opts()`
94 reference and the bare function share one answer — they share one
95 signature by construction (`__wrapped__` runs the whole way down). A
96 callable that refuses attributes (a builtin) simply never memoises.
97 """
98 body = registry.task_body(fn)
99 return body, getattr(body, attr, None)
102def resolved_signature(fn: Any) -> inspect.Signature:
103 """Signature of *fn* with string annotations evaluated to real types.
105 `from __future__ import annotations` (and any PEP 563 usage) turns a
106 tasks file's annotations into strings; `eval_str` turns them back into the
107 types the grammar reasons about. `eval_str` is all-or-nothing — one name
108 that cannot resolve raises for the whole signature — so on failure each
109 annotation is evaluated on its own: one broken parameter degrades to
110 pass-through text, and the rest keep their types, choices and completion.
112 Memoised on the function, the `_footman_*` house pattern: the answer is a
113 pure function of a function object whose annotations are fixed by the
114 time anything asks (discovery imports a module whole before a task in it
115 can run), and the ask is *hot* — every execution path asks about a task
116 around eight times, at ~68 µs a call on a string-annotated file, and the
117 manifest asks once more per task. The memo dies with the function it is
118 stamped on, and discovery's per-file isolation mints fresh functions on
119 every load, so it cannot outlive what it describes.
120 """
121 body, cached = _memo(fn, _SIGNATURE)
122 if cached is not None:
123 return cached
124 try:
125 sig = inspect.signature(fn, eval_str=True)
126 except (NameError, TypeError, AttributeError):
127 sig = _partially_resolved(fn)
128 with contextlib.suppress(AttributeError, TypeError):
129 setattr(body, _SIGNATURE, sig)
130 return sig
133def _partially_resolved(fn: Any) -> inspect.Signature:
134 """The raw signature with every annotation that *can* resolve, resolved.
136 Evaluates each string annotation against the function's module globals —
137 the same environment `eval_str` uses — and leaves only the failing ones
138 as strings, warning once per broken parameter with the underlying error.
139 (`warnings` dedups an identical message per process, so the repeated
140 `resolved_signature` calls of one invocation say this once.)
141 """
142 sig = inspect.signature(fn)
143 module_globals = getattr(fn, "__globals__", {})
144 where = f"{getattr(fn, '__module__', '?')}.{getattr(fn, '__qualname__', fn)}"
146 def resolve(name: str, annotation: Any) -> Any:
147 if not isinstance(annotation, str):
148 return annotation
149 try:
150 # What `eval_str` does per annotation, minus the all-or-nothing.
151 return eval(annotation, module_globals)
152 except Exception as exc: # any failure means "not a usable type"
153 _warn(
154 f"footman: {where} <{name}>: annotation {annotation!r} did "
155 f"not resolve ({exc}); values pass through as text"
156 )
157 return annotation
159 params = [
160 p.replace(annotation=resolve(p.name, p.annotation))
161 for p in sig.parameters.values()
162 ]
163 return sig.replace(
164 parameters=params,
165 return_annotation=resolve("return", sig.return_annotation),
166 )
169def call_signature(fn: Any) -> inspect.Signature:
170 """The signature a Python caller binds against: the declared one minus the
171 context parameter. A body call never passes `ctx` — `run_bound` injects it
172 at the task boundary — so binding a call's arguments against the declared
173 signature would land the first positional value in the `ctx` slot.
175 Memoised beside `resolved_signature`, for the same reason and with the
176 same lifetime: `given()` and the futures memo ask per call, not per run.
177 """
178 body, cached = _memo(fn, _CALL_SIGNATURE)
179 if cached is not None:
180 return cached
181 sig = resolved_signature(fn)
182 name = context_param_name(sig)
183 if name is not None:
184 sig = sig.replace(
185 parameters=[p for p in sig.parameters.values() if p.name != name]
186 )
187 with contextlib.suppress(AttributeError, TypeError):
188 setattr(body, _CALL_SIGNATURE, sig)
189 return sig
192def _unique_globals(root: Group) -> list[Any]:
193 """The tree's plugin globals, deduped by identity, contribution order.
195 The same singleton mounted through two routes is one option; a name clash
196 between two different singletons was refused at discovery, before the
197 manifest could bake either."""
198 seen: list[Any] = []
199 for opt in root.contributions.get("globals", ()):
200 if not any(o is opt for o in seen): 200 ↛ 199line 200 didn't jump to line 199 because the condition on line 200 was always true
201 seen.append(opt)
202 return seen
205def _global_spec(
206 opt: Any, memo: dict[int, list[str]], *, bake: bool = False
207) -> dict[str, Any]:
208 """One manifest entry for a plugin's global option — described by the
209 same machinery as a task parameter, so choices, path-typed file
210 completion and `suggest()` come along by construction."""
211 # The option builds its own synthetic parameter — one construction,
212 # shared with the binder's absent-option ladder. (A bool default is
213 # already normalised at declaration, by GlobalOption itself.)
214 spec = _finish(param_spec(opt._as_parameter()), memo, bake=bake)
215 spec["name"] = opt.name # the cli spelling is the identity
216 if opt.help:
217 spec["help"] = opt.help
218 spec["owner"] = opt.owner
219 return spec
222def param_spec(param: inspect.Parameter) -> dict[str, Any]:
223 """Map one function parameter to its CLI shape (one manifest entry).
225 Dynamic-completer params get a transient `_completer` key that
226 `_finish` replaces with the completer's (cached) choices.
227 """
228 spec: dict[str, Any] = {"name": registry.cli_name(param.name)}
229 ann = param.annotation
230 empty = inspect.Parameter.empty
232 if param.kind is inspect.Parameter.VAR_KEYWORD:
233 raise SpecError(
234 f"**{param.name} is not supported — declare named parameters, or "
235 f"accept KEY=VALUE pairs with a dict[str, str] parameter"
236 )
238 if param.kind is inspect.Parameter.VAR_POSITIONAL:
239 spec["kind"] = "variadic"
240 if ann is not empty:
241 peeled = _coerce.peel(ann) # unwrap Annotated so markers reach the spec
242 tags = _coerce.element_tags(peeled.element)
243 if tags and tags != ["str"]:
244 spec["types"] = tags
245 _marker_keys(spec, peeled, param, has_default=False)
246 return spec
248 has_default = param.default is not empty
249 if has_default:
250 # Bake the default into the manifest when it survives the JSON
251 # coercion mirror (Path → str, Enum → value, …) — an additive key for
252 # help, the catalog, and the markdown exporter. An exotic default is
253 # simply omitted, never an error.
254 ok_default, encoded = _describe.jsonable(param.default)
255 if ok_default:
256 spec["default"] = encoded
257 # A keyword-only parameter (after `*` or `*args`) is an option by
258 # Python's own declaration — defaultless, it is a *required* option,
259 # the same shape defaultless dicts and flags already take.
260 kw_only = param.kind is inspect.Parameter.KEYWORD_ONLY
262 if ann is empty:
263 if isinstance(param.default, bool):
264 spec["kind"] = "flag"
265 elif has_default or kw_only:
266 spec["kind"] = "option"
267 if not has_default:
268 spec["required"] = True
269 else:
270 spec["kind"] = "positional"
271 # A basic default types the parameter as surely as an annotation
272 # would — `port=8000` is an int to Python and to every checker, so
273 # it is an int here. Recording the tags is what buys the eager
274 # refusal, the catalog's `types`, and the coercion at bind time;
275 # `str` is omitted like any other str-tagged spec, since it is the
276 # shape a bare command-line value already has.
277 inferred = _coerce.inferred_type(param.default)
278 if inferred is not None and (tags := _coerce.element_tags(inferred)) != ["str"]:
279 spec["types"] = tags
280 return spec
282 peeled = _coerce.peel(ann)
283 if peeled.mapping:
284 # A dict is always an option (--name KEY=VALUE); when it has no default
285 # it is a *required* option — footman has no positional-mapping syntax.
286 spec["kind"] = "option"
287 spec["mapping"] = True
288 if not has_default:
289 spec["required"] = True
290 _marker_keys(spec, peeled, param, has_default)
291 if peeled.nosplit:
292 spec["nosplit"] = True
293 if (ktags := _coerce.element_tags(peeled.key)) and ktags != ["str"]: 293 ↛ 294line 293 didn't jump to line 294 because the condition on line 293 was never true
294 spec["key_types"] = ktags
295 vchoices = _coerce.all_choices(peeled.element)
296 vtags = _coerce.element_tags(peeled.element)
297 if vchoices is not None: 297 ↛ 298line 297 didn't jump to line 298 because the condition on line 297 was never true
298 spec["value_choices"] = vchoices
299 if vtags and vtags != ["str"] and _coerce.eagerly_checkable(peeled.element):
300 spec["value_types"] = vtags
301 return spec
303 element = peeled.element
304 reads_document = (
305 peeled.stdin is not None
306 and peeled.stdin.field is None
307 and not peeled.stdin.lines
308 and element is not bytes
309 )
310 record = _coerce.fields_of(element) if reads_document else None
311 if record is not None:
312 # What a machine needs to build the JSON this parameter expects: the
313 # shape's name, its fields, each field's coercion tags and whether it
314 # must be given. A name alone — which is all this used to carry — told
315 # a reader the pipe wanted a `Config` and nothing whatever about what
316 # a `Config` is.
317 spec["shape"] = _shape_spec(element)
318 if record is not None and not peeled.multiple and _coerce.group_of(element) is None:
319 # No command-line spelling, so no token spelling: a shape with a
320 # nested record or a collection in it cannot say where a slot ends
321 # inside a comma-separated group, and a TypedDict has named keys
322 # rather than positions. The splitter and completion key on the known
323 # kinds, so `"stdin"` is invisible to them by construction; help still
324 # lists it, with the shape it binds.
325 #
326 # A shape that *does* group keeps its `--opt=a,b` spelling and reads
327 # the pipe too — the command line wins when both are given.
328 spec["kind"] = "stdin"
329 _marker_keys(spec, peeled, param, has_default)
330 return spec
332 if _coerce.is_flag(element) and not peeled.multiple:
333 # Only a *scalar* bool is a --flag; `list[bool]` stays a repeatable
334 # option whose tokens parse as booleans (true/false/1/0/yes/no/on/off).
335 spec["kind"] = "flag"
336 # ask() prompts if absent; stdin fills at the boundary — either one
337 # makes a defaultless flag satisfiable without the command line.
338 if not has_default and peeled.ask is None and peeled.stdin is None:
339 spec["required"] = True # else state it explicitly: --x or --no-x
340 _marker_keys(spec, peeled, param, has_default)
341 return spec
343 if peeled.optional:
344 # Arg[T]: an optional trailing positional — greedy for one token when
345 # present, running on the default when absent (`+` says "absent, next
346 # task"). Deterministic by construction: no name-peeking, cap one.
347 if not has_default:
348 raise SpecError(
349 f"<{param.name}>: Arg[…] needs a default — an optional "
350 f"positional's absence must mean something"
351 )
352 if peeled.multiple: 352 ↛ 353line 352 didn't jump to line 353 because the condition on line 352 was never true
353 raise SpecError(
354 f"<{param.name}>: Arg[…] takes at most one token; a "
355 f"many-valued positional is already optionalable as "
356 f"`Many[…]` with a default"
357 )
358 spec["kind"] = "positional"
359 spec["optional"] = True
360 elif (peeled.ask is not None or peeled.stdin is not None) and not has_default:
361 # ask() and stdin both make a defaultless parameter a CLI-optional
362 # option: absence is filled at the boundary (a prompt, the piped
363 # payload) or refused with a taught message there, so the splitter
364 # must let it be missing rather than enforce it as a required
365 # positional.
366 spec["kind"] = "option"
367 elif has_default or kw_only:
368 spec["kind"] = "option"
369 if not has_default:
370 spec["required"] = True
371 else:
372 spec["kind"] = "positional"
373 _marker_keys(spec, peeled, param, has_default)
374 # A fixed-arity shape accumulates values exactly as a collection does —
375 # commas and repetition both feed one stream — and the declared arity
376 # groups that stream. So it is `multiple` to the splitter, plus a
377 # `group` the splitter and `--help` read for arity and per-position
378 # types. The class itself is not in the manifest (it is not JSON); the
379 # executor reads it back off the annotation when it builds.
380 group = _coerce.group_of(peeled.element)
381 if group is not None:
382 spec["multiple"] = True
383 spec["group"] = {
384 "names": list(group.names) if group.names else None,
385 "types": [_coerce.element_tags(t) for t in group.types],
386 "min": group.required,
387 "max": group.total,
388 "many": peeled.multiple,
389 "label": group.label(),
390 }
391 if peeled.nosplit: 391 ↛ 392line 391 didn't jump to line 392 because the condition on line 391 was never true
392 spec["nosplit"] = True
393 return spec
394 if peeled.multiple:
395 spec["multiple"] = True
396 if peeled.nosplit:
397 spec["nosplit"] = True
398 if peeled.completer is not None:
399 # No `choices` key at all unless someone bakes one: *absent* means
400 # "nobody has asked yet", which is what lets validation and help
401 # resolve live, while a baked `[]` keeps its old meaning — the
402 # completer ran and genuinely offered nothing.
403 spec["dynamic"] = {"strict": peeled.completer.strict}
404 spec["_completer"] = peeled.completer
405 return spec
407 try:
408 choices = _coerce.all_choices(element)
409 accepts = _coerce.enum_accepts(element) if choices is not None else []
410 except _coerce.EnumContractError as exc:
411 # Declaration-time refusal, loudly, through the taught door: the
412 # enum spells an ambiguous or unprojectable surface, and refusing
413 # here — where the contract is first read — is what keeps every
414 # later door free of precedence rules.
415 raise ManifestError(f"parameter {param.name!r}: {exc}") from exc
416 tags = _coerce.element_tags(element)
417 if choices is not None:
418 spec["choices"] = choices
419 # The accepted synonym set beyond the choices — an enum's aliases
420 # and value faces. Baked, so every frontend's eager gate accepts
421 # the same inputs by contract: authored and enumerated, never
422 # leniency. Completion and help read `choices` alone; a synonym is
423 # accepted, not taught.
424 if accepts:
425 spec["accepts"] = accepts
426 # Emit `types` only when the element is eagerly checkable — a union with a
427 # custom member (`UUID | int`) can't be accept/rejected up front, so leave
428 # it to binding rather than eagerly rejecting valid values.
429 if tags and tags != ["str"] and _coerce.eagerly_checkable(element):
430 spec["types"] = tags
431 elif choices is None and not tags and not isinstance(element, type):
432 if isinstance(element, str) and "stdin" in element:
433 # `eval_str` failed for the whole signature, so the marker never
434 # even peeled — but the annotation names it, and a silently
435 # text-degraded stdin parameter would be a mystery at bind time.
436 raise SpecError(
437 f"<{param.name}>: annotation {element!r} did not resolve — "
438 f"a stdin payload type and everything it names must be "
439 f"module-level, where `eval_str` can see them"
440 )
441 # The annotation resolves to nothing footman can coerce (a value, an
442 # exotic generic): values will pass through as plain text. Silent
443 # degrade is a debugging tax — say so. A *string* landing here was
444 # already reported by `_partially_resolved`, which knows the task and
445 # the underlying error; repeating it per spec build would turn one
446 # broken name back into a warning block.
447 if not isinstance(element, str): 447 ↛ 448line 447 didn't jump to line 448 because the condition on line 447 was never true
448 _warn(
449 f"footman: parameter {param.name!r}: annotation {element!r} "
450 f"is not a usable type; values are passed through as text"
451 )
452 return spec
455def _shape_spec(target: Any, seen: tuple[Any, ...] = ()) -> dict[str, Any]:
456 """The JSON a document parameter expects, as data a machine can act on:
457 the shape's name and, for each field, its name, its type, whether it must
458 be given, and — when the field is itself a record — that field's shape in
459 turn.
461 A field with no `types`, `choices` or `shape` is one footman does not
462 coerce: whatever JSON holds arrives as it is. That is a description, not
463 an omission.
465 A shape already on the way down is emitted by name alone. A
466 self-referential record has no finite expansion, and every shape appears
467 in full somewhere above, so the name resolves.
468 """
469 spec: dict[str, Any] = {"name": getattr(target, "__name__", str(target))}
470 if target in seen:
471 return spec # recursive: named above, expanded there
472 fields = _coerce.fields_of(target)
473 if fields is None: 473 ↛ 474line 473 didn't jump to line 474 because the condition on line 473 was never true
474 return spec
475 spec["fields"] = [_field_spec(f, (*seen, target)) for f in fields]
476 return spec
479def _field_spec(field: _coerce.Field, seen: tuple[Any, ...]) -> dict[str, Any]:
480 out: dict[str, Any] = {"name": field.name}
481 inner = field.type
482 if (collection := _coerce.collection_of(inner)) is not None:
483 inner = (typing.get_args(inner) or (str,))[0]
484 out["many"] = collection.__name__
485 if _coerce.fields_of(inner) is not None:
486 out["shape"] = _shape_spec(inner, seen)
487 else:
488 if (choices := _coerce.all_choices(inner)) is not None:
489 out["choices"] = choices
490 if tags := _coerce.element_tags(inner):
491 out["types"] = tags
492 if field.required:
493 out["required"] = True
494 return out
497def _marker_keys(
498 spec: dict[str, Any],
499 peeled: _coerce.Peeled,
500 param: inspect.Parameter,
501 has_default: bool,
502) -> None:
503 """Additive manifest keys for the `Annotated` markers (path/bounds/env/doc).
505 `check(fn)` deliberately never lands in the manifest — functions don't
506 serialize (the same reason `_finish` strips `_completer`); it runs at
507 binding time instead.
508 """
509 if peeled.hidden:
510 # Marked, never missing — the same rule a hidden *task* follows. A
511 # machine reading the manifest is exactly who is meant to find it;
512 # only the listing a human reads leaves it out.
513 spec["hidden"] = True
514 if peeled.ask is not None and peeled.ask.secret:
515 # A secret parameter never publishes values: no baked choices, no
516 # completer run — the flag completes, its value stays yours.
517 spec["secret"] = True
518 if peeled.doc is not None:
519 spec["doc"] = peeled.doc
520 if peeled.path_req is not None:
521 spec["path"] = peeled.path_req
522 if peeled.glob is not None:
523 # Completion only: the pattern the shell's own file completion filters
524 # by. Never a validation rule — see `matching`.
525 spec["glob"] = peeled.glob
526 if peeled.bounds is not None:
527 lo, hi = peeled.bounds
528 if lo is not None: 528 ↛ 530line 528 didn't jump to line 530 because the condition on line 528 was always true
529 spec["min"] = lo
530 if hi is not None:
531 spec["max"] = hi
532 if peeled.default_fn is not None:
533 if not has_default:
534 raise SpecError(
535 f"<{param.name}>: default(…) needs a declared default to sit "
536 f"on — a plain Python call of the task, with no run around it, "
537 f"has nothing else to fall back to"
538 )
539 # Called here, on the execution path, so `--help` prints what this run
540 # would actually use rather than a value baked whenever a cache was
541 # last built. A raising computer is left to raise at bind time instead
542 # of taking the whole tree's help down with it.
543 spec["computed"] = True # help says so: a bare number reads arbitrary
544 if peeled.default_fn.reads_siblings:
545 # It reads values only an invocation has, so there is nothing to
546 # show — and the *declared* default must go too: it exists so a
547 # plain Python call still binds, and printing that sentinel would
548 # advertise the very thing this marker replaces.
549 spec.pop("default", None)
550 else:
551 ok_computed, computed = False, None
552 with contextlib.suppress(Exception): # help degrades; the run teaches
553 ok_computed, computed = _describe.jsonable(peeled.default_fn.fn())
554 if ok_computed: 554 ↛ 556line 554 didn't jump to line 556 because the condition on line 554 was always true
555 spec["default"] = computed
556 if peeled.env is not None:
557 if spec.get("mapping"):
558 raise SpecError(
559 f"<{param.name}>: env() is not supported on dict parameters"
560 )
561 if not has_default:
562 raise SpecError(
563 f"<{param.name}>: env({peeled.env!r}) needs a default — an "
564 f"env fallback makes the parameter optional, so it needs "
565 f"somewhere to fall"
566 )
567 spec["env"] = peeled.env
568 if peeled.stdin is not None:
569 marker = peeled.stdin
570 if spec.get("mapping") and (marker.field is not None or marker.lines):
571 raise SpecError(
572 f"<{param.name}>: a dict parameter reads stdin whole (a JSON "
573 f"object) — stdin(field)/stdin(lines=True) do not apply"
574 )
575 if marker.lines and not peeled.multiple:
576 raise SpecError(
577 f"<{param.name}>: stdin(lines=True) needs a list parameter — "
578 f"each line binds as one element"
579 )
580 if marker.field is not None and peeled.multiple:
581 raise SpecError(
582 f"<{param.name}>: stdin({marker.field!r}) binds a single "
583 f"value — a list parameter reads lines (stdin(lines=True)) "
584 f"or a JSON array (bare stdin)"
585 )
586 if isinstance(peeled.element, str): 586 ↛ 587line 586 didn't jump to line 587 because the condition on line 586 was never true
587 raise SpecError(
588 f"<{param.name}>: annotation {peeled.element!r} did not "
589 f"resolve — a stdin payload type and everything it names "
590 f"must be module-level, where `eval_str` can see them"
591 )
592 if marker.field is not None:
593 spec["stdin"] = f"field:{marker.field}"
594 elif marker.lines:
595 spec["stdin"] = "lines"
596 elif peeled.element is bytes:
597 spec["stdin"] = "bytes"
598 elif (
599 spec.get("mapping")
600 or peeled.multiple
601 or spec.get("kind") == "stdin"
602 or _binder.is_document_target(peeled.element)
603 ):
604 spec["stdin"] = "json"
605 else:
606 spec["stdin"] = "text"
609class _Undescribable(Exception):
610 """Internal: a return annotation outside the describable set. Never a user
611 error — "describable" ⊆ "returnable", so the caller degrades to no spec."""
614def returned_spec(ann: Any) -> dict[str, Any] | None:
615 """The output spec a return annotation declares, or None for no claims.
617 The mirror of `param_spec` for the *output* side: the annotation is the
618 declaration, and the describable set is exactly what
619 `_describe.json_default` serialises — dataclasses (nested), TypedDict,
620 NamedTuple, `list`/`tuple[T, ...]`/`set`/`dict[str, …]`, the scalar
621 bridge types, `Literal`/`Enum` as choices, `T | None` as nullable. The
622 spec is footman's own compact shape (one vocabulary with the param
623 specs); `_describe.returns_json_schema` renders it as JSON Schema at the
624 describe door.
626 An annotation outside the set — a broken string, a wider union, an
627 exotic generic — yields `None`, never an error: the task still runs and
628 its value still serialises (or refuses) at runtime exactly as before, it
629 just makes no claims. Bare `int` is the exit-code channel, not data, so
630 it too declares nothing; `Stdout[T]` describes `T` — the document and
631 the returned value are one declaration.
632 """
633 if ann is inspect.Parameter.empty or ann is None or ann is type(None):
634 return None
635 # `emitted` fully normalises (Annotated *and* Optional stripped) — the
636 # right view for the bare-int test, where `int | None` is still the
637 # exit-code channel. The walk below gets the *original* annotation:
638 # it strips Annotated itself, so `Report | None` keeps its nullability.
639 _, flat = _coerce.emitted(ann)
640 if flat is int or flat is Any or flat is object:
641 return None # exit-code channel / no claims — a schema would lie
642 try:
643 return _returned_of(ann, ())
644 except _Undescribable:
645 return None
648_RETURN_SCALARS: list[tuple[type, str]] = [
649 # Order matters where the types nest: datetime before date (a datetime
650 # *is* a date), bool before int (checked via `is`, so moot, but kept
651 # explicit in `_returned_of` below).
652 (datetime.datetime, "datetime"),
653 (datetime.date, "date"),
654 (datetime.time, "time"),
655 (uuid.UUID, "uuid"),
656 (decimal.Decimal, "decimal"),
657 (PurePath, "path"),
658]
661def _returned_of(ann: Any, seen: tuple[Any, ...]) -> dict[str, Any]:
662 """One node of the output spec — recursive, raising on the undescribable."""
663 while typing.get_origin(ann) is typing.Annotated:
664 # Markers (`Stdout`, a doc) never change the shape; unions survive.
665 ann = typing.get_args(ann)[0]
666 if ann is Any or ann is object: 666 ↛ 667line 666 didn't jump to line 667 because the condition on line 666 was never true
667 return {"kind": "any"} # a real "no claims here" inside a container
668 if ann is type(None): 668 ↛ 669line 668 didn't jump to line 669 because the condition on line 668 was never true
669 return {"kind": "none"}
670 if _coerce._is_union(ann):
671 # T | None is nullable; a wider union has no json_default story to
672 # mirror, so it makes no claims at all rather than partial ones.
673 members = _coerce.union_members(ann)
674 if len(members) != 1:
675 raise _Undescribable
676 spec = _returned_of(members[0], seen)
677 spec["nullable"] = True
678 return spec
679 if typing.get_origin(ann) is typing.Literal:
680 values: list[Any] = []
681 nullable = False
682 for value in typing.get_args(ann):
683 if value is None: 683 ↛ 684line 683 didn't jump to line 684 because the condition on line 683 was never true
684 nullable = True
685 continue
686 ok, encoded = _describe.jsonable(value)
687 if not ok: 687 ↛ 688line 687 didn't jump to line 688 because the condition on line 687 was never true
688 raise _Undescribable
689 values.append(encoded)
690 if not values: 690 ↛ 691line 690 didn't jump to line 691 because the condition on line 690 was never true
691 raise _Undescribable
692 literal: dict[str, Any] = {"kind": "enum", "values": values}
693 if nullable: 693 ↛ 694line 693 didn't jump to line 694 because the condition on line 693 was never true
694 literal["nullable"] = True
695 return literal
696 if isinstance(ann, type) and issubclass(ann, enum.Enum):
697 if not len(ann): 697 ↛ 698line 697 didn't jump to line 698 because the condition on line 697 was never true
698 raise _Undescribable # an empty Enum has no members to claim
699 # The returns side moves in lockstep with choices: `values` speaks
700 # tokens (what documents carry, what the JSON Schema projection
701 # lists natively), and `members` carries the pairs first-class —
702 # token beside declared value, aliases as input synonyms — so a
703 # foreign frontend binds the same faces by contract.
704 pairs: list[dict[str, Any]] = []
705 for face in _coerce.enum_faces(ann):
706 ok, encoded = _describe.jsonable(face.member.value)
707 if not ok: 707 ↛ 708line 707 didn't jump to line 708 because the condition on line 707 was never true
708 raise _Undescribable
709 entry: dict[str, Any] = {"token": face.token, "value": encoded}
710 if face.aliases:
711 entry["aliases"] = list(face.aliases)
712 pairs.append(entry)
713 return {
714 "kind": "enum",
715 "name": ann.__name__,
716 "values": [m["token"] for m in pairs],
717 "members": pairs,
718 }
719 if ann is bool:
720 return {"kind": "bool"}
721 if ann is int:
722 return {"kind": "int"}
723 if ann is float:
724 return {"kind": "float"}
725 if ann is str:
726 return {"kind": "str"}
727 if isinstance(ann, type):
728 for base, kind in _RETURN_SCALARS:
729 if issubclass(ann, base):
730 return {"kind": kind}
731 origin = typing.get_origin(ann)
732 if ann in (list, set, frozenset) or origin in (list, set, frozenset):
733 args = typing.get_args(ann)
734 element = args[0] if args else Any
735 return {"kind": "list", "items": _returned_of(element, seen)}
736 if origin is tuple:
737 args = typing.get_args(ann)
738 if len(args) == 2 and args[1] is Ellipsis:
739 return {"kind": "list", "items": _returned_of(args[0], seen)}
740 raise _Undescribable # a heterogeneous tuple has no named positions
741 if ann is dict or origin is dict:
742 args = typing.get_args(ann)
743 if not args:
744 return {"kind": "object"} # no field claims
745 key, value = args
746 if key is not str:
747 raise _Undescribable # JSON object keys are strings
748 if value is Any or value is object:
749 return {"kind": "object"}
750 return {"kind": "map", "values": _returned_of(value, seen)}
751 if typing.is_typeddict(ann):
752 return _returned_fields(ann, seen, kind="object", typeddict=True)
753 if dataclasses.is_dataclass(ann) and isinstance(ann, type):
754 # Every dataclass field serialises (`asdict` has no optionality).
755 return _returned_fields(ann, seen, kind="object")
756 if isinstance(ann, type) and issubclass(ann, tuple) and hasattr(ann, "_fields"):
757 # A NamedTuple serialises as a JSON *array* (it is a tuple); "row"
758 # keeps that honest — named positions, not an object.
759 return _returned_fields(ann, seen, kind="row")
760 raise _Undescribable
763def _returned_fields(
764 cls: Any,
765 seen: tuple[Any, ...],
766 *,
767 kind: str,
768 typeddict: bool = False,
769) -> dict[str, Any]:
770 """An `object`/`row` spec for a dataclass, TypedDict, or NamedTuple.
772 Fields ride as a name-keyed mapping in declaration order (for a "row"
773 the order *is* the positions). A recursive shape has no finite spec to
774 bake, so re-entering a class under description bails the whole
775 annotation out to "no claims"."""
776 if any(cls is s for s in seen):
777 raise _Undescribable
778 # Read the class-level facts before any `is_dataclass` narrowing below
779 # can convince a type-checker `cls` no longer has them.
780 class_name = str(getattr(cls, "__name__", cls))
781 total = bool(getattr(cls, "__total__", True))
782 try:
783 # `include_extras` keeps `Required`/`NotRequired` visible: under
784 # `from __future__ import annotations` a TypedDict's own
785 # `__required_keys__` cannot see through the string forms, but the
786 # resolved wrappers can't lie.
787 hints = typing.get_type_hints(cls, include_extras=True)
788 except Exception as exc:
789 raise _Undescribable from exc # a name in a field didn't resolve
790 if dataclasses.is_dataclass(cls):
791 names = [f.name for f in dataclasses.fields(cls)]
792 elif kind == "row":
793 names = list(cls._fields)
794 else:
795 names = list(hints)
796 fields: dict[str, Any] = {}
797 for name in names:
798 hint = hints[name]
799 required = True
800 if typeddict:
801 origin = typing.get_origin(hint)
802 if origin is typing.Required: 802 ↛ 803line 802 didn't jump to line 803 because the condition on line 802 was never true
803 hint = typing.get_args(hint)[0]
804 elif origin is typing.NotRequired:
805 hint, required = typing.get_args(hint)[0], False
806 else:
807 required = total
808 spec = _returned_of(hint, (*seen, cls))
809 if not required:
810 spec["required"] = False
811 fields[name] = spec
812 if not fields: 812 ↛ 813line 812 didn't jump to line 813 because the condition on line 812 was never true
813 raise _Undescribable # a fieldless shape claims nothing worth baking
814 return {"kind": kind, "name": class_name, "fields": fields}
817def _run_completer(completer: suggest, memo: dict[int, list[str]]) -> list[str]:
818 """Call a completer at most once per build (deduped by function identity).
820 A raising *strict* completer aborts the build with `CompleterError` — its
821 whole point is validation, so failing silent would validate nothing. A
822 best-effort completer (`strict=False`) degrades to no candidates.
823 """
824 key = id(completer.fn)
825 if key not in memo:
826 try:
827 memo[key] = [str(v) for v in completer.fn()]
828 except Exception as exc:
829 if completer.strict:
830 name = getattr(completer.fn, "__qualname__", repr(completer.fn))
831 raise CompleterError(
832 f"dynamic choices from {name}() failed: "
833 f"{type(exc).__name__}: {exc} — fix the completer, or pass "
834 f"suggest(fn, strict=False) if this data is best-effort"
835 ) from exc
836 memo[key] = []
837 return memo[key]
840def _finish(
841 spec: dict[str, Any], memo: dict[int, list[str]], *, bake: bool = False
842) -> dict[str, Any]:
843 """Drop the transient completer, baking its choices only when asked.
845 *bake* is for a surface that renders every parameter it can find and has
846 nowhere to resolve later — the docs exporter. Everywhere else the
847 completer stays unrun: the command line resolves the one parameter it is
848 validating, help resolves the ones it prints, and a <kbd>Tab</kbd>
849 recomputes fresh anyway. Running them all on every invocation is what
850 made an unrelated `fm build` shell out to a git-branch completer.
851 """
852 completer = spec.pop("_completer", None)
853 if spec.get("secret"):
854 spec.pop("choices", None) # never bake a secret's values
855 completer = None
856 if completer is not None and bake:
857 spec["choices"] = _run_completer(completer, memo)
858 return spec
861def _cli_params(fn: Any) -> list[inspect.Parameter]:
862 """The parameters that form a task's CLI (the injected ctx is not one)."""
863 sig = resolved_signature(fn)
864 ctx_name = context_param_name(sig)
865 return [p for p in sig.parameters.values() if p.name != ctx_name]
868def _source_of(fn: Any) -> str:
869 """Where *fn* is written, for the shadow chain `--help` prints.
871 The one answer, shared with the refusals that name a declaration and the
872 failure line that places an exception: three spellings of
873 `co_filename:co_firstlineno` is how they drift apart.
874 """
875 return _describe.source_of(fn)
878def _task_node(
879 fn: Any,
880 memo: dict[int, list[str]],
881 *,
882 bake: bool = False,
883 user_dir: str | None = None,
884) -> dict[str, Any]:
885 sig = resolved_signature(fn)
886 infinite = registry.is_infinite(fn)
887 interactive = registry.is_interactive(fn)
888 confirm = registry.task_confirm(fn)
889 lane = registry.task_lane(fn)
890 ctx_name = context_param_name(sig) # the injected ctx param is not a CLI arg
891 # Imported here, not at module scope: the module compiles eleven regexes
892 # on the way in (~0.24 ms of its 0.65 ms) and only manifest *building*
893 # reads a docstring.
894 #
895 # Worth being honest about what that buys today: `sync_manifest` builds
896 # unconditionally, so every run reaches this line anyway and only
897 # `--version` is spared. It pays properly once building is skipped when
898 # nothing changed — see notes/20260816-startup-perf.md, which measures
899 # that rebuild at ~70 us per task on every run.
900 from footman import docstrings
902 parsed = docstrings.parse(inspect.getdoc(fn))
903 params = [
904 _finish(param_spec(p), memo, bake=bake)
905 for p in sig.parameters.values()
906 if p.name != ctx_name
907 ]
908 # An optional positional must trail everything positional: a required
909 # argument or a rest-consumer after it would make "which token is
910 # whose" ambiguous — exactly what the grammar refuses to be.
911 seen_optional: str | None = None
912 for spec in params:
913 if seen_optional is not None and (
914 spec["kind"] == "variadic"
915 or (spec["kind"] == "positional" and not spec.get("optional"))
916 or (spec["kind"] == "positional" and spec.get("multiple"))
917 ):
918 raise SpecError(
919 f"<{seen_optional}>: an Arg[…] optional positional must come "
920 f"last — <{spec['name']}> follows it, so which token belongs "
921 f"to whom would be a guess. Reorder, or make "
922 f"<{spec['name']}> an option."
923 )
924 if spec["kind"] == "positional" and spec.get("optional"):
925 seen_optional = spec["name"]
926 for spec in params:
927 if spec["name"] == "help" and spec["kind"] in ("flag", "option"):
928 raise SpecError(
929 "<help>: 'help' is a reserved parameter name — it maps to "
930 "--help, which footman intercepts anywhere on the line to show "
931 "help and never run a task, so the option could never be "
932 "reached. Rename it (e.g. show_help). It is the only reserved "
933 "name: every other global must come before the first task, so a "
934 "task parameter may reuse it (fm deploy --json binds --json to "
935 "deploy)."
936 )
937 known: set[str] = set()
938 for spec in params:
939 python_name = str(spec["name"]).replace("-", "_")
940 known.add(python_name)
941 # The docstring fills in; an explicit doc() marker already won.
942 if "doc" not in spec and (text := parsed.params.get(python_name)):
943 spec["doc"] = text
944 if ctx_name:
945 known.add(ctx_name) # documenting the injected ctx param is fine
946 if unknown := sorted(set(parsed.params) - known):
947 _warn(
948 f"footman: {getattr(fn, '__name__', fn)!s}: docstring documents "
949 f"unknown parameter(s): {', '.join(unknown)}"
950 )
951 node: dict[str, Any] = {"help": parsed.summary, "params": params}
952 # Where this task comes from, as the listings group it: a built-in
953 # carries no defining directory (the cascade never tagged it), and the
954 # personal rung's directory is the user tasks file's own. Everything
955 # else is the project's cascade — including a plugin the project chose
956 # to mount, which is the project's decision and reads as its own task.
957 # Marked only when global: the project's tasks are the common case and
958 # a per-row field for them would be bytes on the completion hot path.
959 defined_in = _discover.defining_dir(fn)
960 if defined_in is None or (user_dir is not None and defined_in == user_dir):
961 node["global"] = True
962 if (provider := registry.mounted_from(fn)) is not None:
963 # Additive: which provider this task came from — the entry-point
964 # identity for `plugin()`, the module name for `include()`. Omitted
965 # for a task the tasks file defines itself, which is the honest
966 # answer rather than a hole: the project owns it.
967 #
968 # A provider, not a file. A source path would answer only for a
969 # Python function (`__code__`), and would make a reader infer
970 # ownership by matching path prefixes; this names the owner
971 # outright, and keeps naming it if a task is one day implemented
972 # over something that is not Python. It also keeps absolute home
973 # paths out of `--json --list` output people paste and commit.
974 node["mounted_from"] = provider
975 used = registry.task_uses(fn)
976 if used:
977 # The globals this task declares it reads — help, the catalog and
978 # agents see the dependency without running anything.
979 node["uses"] = [opt.name for opt in used]
980 if (previous := _discover.shadowed(fn)) is not None:
981 # Additive, and only for the rare overridden task: the options of
982 # the task this one shadows, so `--help` can show the call
983 # `inherited()` will make.
984 node["shadows"] = {
985 "params": [
986 _finish(param_spec(p), memo, bake=bake) for p in _cli_params(previous)
987 ],
988 "where": _source_of(previous),
989 }
990 declares, _inner = _coerce.emitted(sig.return_annotation)
991 if declares:
992 if interactive:
993 raise SpecError(
994 f"{getattr(fn, '__name__', fn)!s}: Stdout[…] and "
995 f"interactive=True cannot both hold — an interactive task "
996 f"owns the real terminal, uncaptured, and a declaring task's "
997 f"stdout belongs to its return value. Drop one."
998 )
999 node["emits"] = True # additive: this task's stdout is its return value
1000 if (returned := returned_spec(sig.return_annotation)) is not None:
1001 # Additive: the output contract the return annotation declares —
1002 # static, so it bakes beside the param specs and every surface
1003 # (envelope, --describe, help, docs) reads the same shape.
1004 node["returned"] = returned
1005 if parsed.returns:
1006 node["returned_doc"] = parsed.returns # additive: what the value means
1007 if infinite:
1008 node["infinite"] = True # additive: listings and help say how it ends
1009 if interactive:
1010 node["interactive"] = True # additive: this task owns the terminal
1011 if lane is not None:
1012 node["lane"] = lane # additive: "serial" or "exclusive" scheduling
1013 if confirm:
1014 node["confirm"] = confirm # additive: the yes/no gate before it runs
1015 # Additive, and deliberately visible: a caller that can see a task already
1016 # retries will not wrap it in a retry of its own, and every layer that
1017 # cannot see it is one more that can stack — curl's `--retry` multiplying
1018 # with `@task(retries=)` is the case the design note names. Baking the
1019 # field in while the schema is being touched is free; adding it later is a
1020 # version bump (notes/20260807-timeout-and-retry.md).
1021 if retries := registry.task_retries(fn):
1022 node["retries"] = retries
1023 if timeout := registry.task_timeout(fn):
1024 node["timeout"] = timeout
1025 if parsed.long:
1026 node["long"] = parsed.long
1027 # Additive availability annotation (`@requires`): the name stays listed and
1028 # completable either way — execution re-checks the predicate live.
1029 if (reason := registry.availability(fn)) is not None:
1030 node["disabled"] = reason
1031 return node
1034def _scoped(
1035 node: dict[str, Any], fn: Any, inherited: str | None, project: bool
1036) -> dict[str, Any]:
1037 """Stamp `unexposed` on a task that does not belong here.
1039 Resolved at build time, where the mode is known, so every reader
1040 downstream — listing, help, completion, the address resolver — asks one
1041 flat question instead of each re-deriving "am I in a project". The three
1042 `expose` answers collapse to that one flag against the mode, and the
1043 flag alone is enough to phrase the refusal: a task kept out *of* a
1044 project can only be `global_only`, one kept out *outside* one can only
1045 be `project_only`.
1046 """
1047 # Unsealed means unclaimed: a person's own tasks file and a project's
1048 # own cascade ride everywhere. `project_only` is imposed by `seal_expose`
1049 # on the rung that promises it, never assumed here.
1050 where = registry.declared_expose(fn) or inherited or "always"
1051 if where == "always":
1052 return node
1053 if (where == "project_only") is not project:
1054 node["unexposed"] = True
1055 return node
1058def _hide(node: dict[str, Any], own: bool | None, inherited: bool) -> dict[str, Any]:
1059 """Stamp the resolved `hidden` onto a task node: its own answer if it gave
1060 one, otherwise the group's. Additive — absent means listed."""
1061 if inherited if own is None else own:
1062 node["hidden"] = True
1063 return node
1066def _node(
1067 g: Group,
1068 memo: dict[int, list[str]],
1069 hidden: bool = False,
1070 *,
1071 bake: bool = False,
1072 expose: str | None = None,
1073 project: bool = True,
1074 user_dir: str | None = None,
1075) -> dict[str, Any]:
1076 # `hidden` is resolved here, where the tree structure is: a node that never
1077 # declared one inherits its group's answer, so hiding a subtree is said once
1078 # at the root and `hidden=False` on a child is a real way back. The group's
1079 # own answer may come from its default action — hiding what a bare
1080 # `fm <group>` runs hides the group it speaks for.
1081 own = g.hidden
1082 if own is None and g.default_task is not None:
1083 own = registry.declared_hidden(g.default_task)
1084 mine = hidden if own is None else own
1085 # The same inheritance for scope, one line down: a group's answer covers
1086 # its subtree, a child overrides it, and silence asks whoever encloses it.
1087 here = g.expose if g.expose is not None else expose
1088 node: dict[str, Any] = {
1089 "help": g.help,
1090 "tasks": {
1091 name: _scoped(
1092 _hide(
1093 _task_node(fn, memo, bake=bake, user_dir=user_dir),
1094 registry.declared_hidden(fn),
1095 mine,
1096 ),
1097 fn,
1098 here,
1099 project,
1100 )
1101 for name, fn in g.tasks.items()
1102 },
1103 "groups": {
1104 name: _node(
1105 sub,
1106 memo,
1107 mine,
1108 bake=bake,
1109 expose=here,
1110 project=project,
1111 user_dir=user_dir,
1112 )
1113 for name, sub in g.groups.items()
1114 },
1115 }
1116 if mine:
1117 node["hidden"] = True # additive: listings skip it, the address still runs
1118 if g.mounted_from is not None:
1119 # The same field, answering a *stronger* question here: this whole
1120 # group is one provider's, which is true only when the mount landed
1121 # on a free name and grafted the stamped group whole. Compose into a
1122 # group that already exists and the destination group survives
1123 # unstamped — shared, so no single owner — while every task under it
1124 # still answers exactly. Read ownership off the task rows; this is
1125 # for "is this entire subtree theirs".
1126 node["mounted_from"] = g.mounted_from
1127 # A runnable group (one with `@group.default`) carries the default's option
1128 # surface — the same `{help, params}` shape a task node has — so the splitter
1129 # parses a bare `fm <group> [flags]` against it and completion/help render it.
1130 # The fan-out flag rides beside it (not inside: `_task_node` memoises task
1131 # specs, and fan-out is a property of this default, not of the function)
1132 # so listings can *say* what an undocumented default does.
1133 if g.default_task is not None:
1134 node["default"] = _scoped(
1135 _hide(
1136 _task_node(g.default_task, memo, bake=bake, user_dir=user_dir),
1137 registry.declared_hidden(g.default_task),
1138 mine,
1139 ),
1140 g.default_task,
1141 here,
1142 project,
1143 )
1144 node["default_fanout"] = registry.fans_out(g.default_task)
1145 # The bare group name IS this action's other spelling, so the group
1146 # answers whatever its default answers — stamped here, once, rather
1147 # than taught to each of the listing, completion and dispatch paths.
1148 # `hidden` already resolves this way a few lines up; this is the same
1149 # rule, and without it `fm lint` ran while `fm lint.default` refused.
1150 if node["default"].get("unexposed"):
1151 node["unexposed"] = True
1152 return node
1155def tree_hash(tree: dict[str, Any]) -> str:
1156 """Stable hash of the tree's structure (names, params, help)."""
1157 blob = json.dumps(tree, sort_keys=True, ensure_ascii=False)
1158 return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:16]
1161def build_manifest(
1162 root: Group,
1163 *,
1164 completion_max_age: int | None = None,
1165 bake_completers: bool = False,
1166 project: bool = True,
1167 user_dir: str | None = None,
1168) -> dict[str, Any]:
1169 """Introspect *root* into a serialisable manifest dict.
1171 *project* says whether this manifest describes a directory that has one.
1172 It is the only place the question is asked: tasks that need a project are
1173 stamped `unexposed` here, so listing, help, completion and the address
1174 resolver each read one flat flag rather than re-deriving the mode.
1176 *completion_max_age* (seconds, or `None` to disable) is baked in so the
1177 stdlib-only completion hot path can decide whether to trigger a background
1178 refresh without reading config.
1180 Dynamic completers do **not** run here. They used to — every invocation
1181 executed every `suggest()` function in the tree, so a git-branch completer
1182 on one task shelled out on every `fm anything` — and nothing needed it:
1183 a <kbd>Tab</kbd> recomputes fresh, the command line resolves the one
1184 parameter whose value it validates, and help resolves the ones it prints.
1185 *bake_completers* is for the surface with nowhere to resolve later: the
1186 docs exporter, which renders every parameter it can find into a page.
1187 """
1188 memo: dict[int, list[str]] = {}
1189 tree = _node(root, memo, bake=bake_completers, project=project, user_dir=user_dir)
1190 tree["globals"] = [
1191 _global_spec(opt, memo, bake=bake_completers) for opt in _unique_globals(root)
1192 ]
1193 tree["global_help"] = _core_global_help()
1194 # Which side this tree was built for. Every `unexposed` stamp under it
1195 # was resolved against this one answer, so a reader that has to *phrase*
1196 # the refusal can say which way the task was withheld without the stamp
1197 # having to carry a reason of its own.
1198 tree["project"] = project
1199 return {
1200 "schema": SCHEMA_VERSION,
1201 "hash": tree_hash(tree),
1202 "completion_max_age": completion_max_age,
1203 "tree": tree,
1204 }
1207def _core_global_help() -> dict[str, str]:
1208 """`--flag -> one line`, for footman's own globals.
1210 The completion hot path knows the core flags by name (its own frozenset
1211 mirror — it may not import `_split`), but a name with no words beside it
1212 is the difference between a Tab that teaches and a Tab that lists. Rather
1213 than mirror thirty-five help strings into `_complete` as well — prose is
1214 exactly what rots — the words ride in the manifest the hot path already
1215 reads and parses. `CORE_OPTIONS` stays the one place they are written.
1217 Aliases carry their long form's line: `-j` and `--jobs` are one option,
1218 and a reader scanning a column wants to know what it does, not that it
1219 has two spellings.
1221 `{prog}` is substituted here, the way `--help` substitutes it when it
1222 renders the same table — a manifest belongs to one brand (its path is
1223 brand-scoped and it already bakes `tasks_file`), so the brand's own
1224 command name is the honest thing to store. Left raw, a Tab would offer
1225 "help for {prog}" in braces.
1227 The keys are the *offerable spellings*, exactly as the splitter reads
1228 them — what plugin options always got from `_opt_rows`, the core table
1229 now gets too. An option's value goes attached, so `--opt=` is a key of
1230 its own; its bare mention stands for its default, so only a defaulted
1231 option gets a bare key (bare `--where` is a taught refusal, and a menu
1232 must not offer what the grammar refuses). The bare key's line carries
1233 the resolved default — a manifest belongs to this machine, so `--jobs`
1234 naming its actual width is the honest thing to store.
1235 """
1236 from footman import _app, _describe, _split
1238 rows: dict[str, str] = {}
1239 for name, alias, kind, _hint, default, help_text in _split.GLOBALS:
1240 if not help_text: 1240 ↛ 1241line 1240 didn't jump to line 1241 because the condition on line 1240 was never true
1241 continue
1242 line = help_text.replace("{prog}", _app._brand.prog)
1243 for flag in (name, alias):
1244 if not flag:
1245 continue
1246 if kind != "option":
1247 rows[flag] = line
1248 continue
1249 rows[flag + "="] = line
1250 if default is not None:
1251 rows[flag] = line + _describe.global_default_suffix(name)
1252 return rows
1255_REPLACE_ATTEMPTS = 25
1256_REPLACE_PAUSE = 0.04 # ≈1s of retries in all
1259def write_manifest(manifest: dict[str, Any], path: Path) -> None:
1260 """Write *manifest* to *path* atomically (never leave a half file).
1262 Windows refuses to replace a destination another process holds open, and
1263 a reader holding it open is the *design* here: a completion poll reads
1264 this file every few milliseconds while a detached refresh rewrites it.
1265 Losing that race silently means the rebuild never lands — the caller is
1266 a background child that swallows its errors — so the file stays stale
1267 until some later write happens to find a quiet moment.
1269 Hence a bounded retry: about a second of attempts, then give up without
1270 leaving the temp file behind. POSIX never takes this path (rename over
1271 an open file is fine), and no caller waits longer than it already did.
1272 """
1273 import time
1275 path.parent.mkdir(parents=True, exist_ok=True)
1276 tmp = path.with_name(f"{path.name}.{os.getpid()}.tmp")
1277 tmp.write_text(json.dumps(manifest, indent=1, ensure_ascii=False), "utf-8")
1278 for attempt in range(_REPLACE_ATTEMPTS): 1278 ↛ exitline 1278 didn't return from function 'write_manifest' because the loop on line 1278 didn't complete
1279 try:
1280 os.replace(tmp, path)
1281 return
1282 except PermissionError:
1283 if attempt == _REPLACE_ATTEMPTS - 1:
1284 # Out of patience: clean up rather than litter the cache
1285 # directory with `<name>.<pid>.tmp` files nobody reads.
1286 tmp.unlink(missing_ok=True)
1287 raise
1288 time.sleep(_REPLACE_PAUSE)
1291def load_manifest(path: Path) -> dict[str, Any] | None:
1292 """Read a cached manifest, or `None` if missing/unreadable/corrupt."""
1293 try:
1294 data = json.loads(path.read_text("utf-8"))
1295 except (OSError, ValueError):
1296 return None
1297 return data if isinstance(data, dict) else None
1300def sync_manifest(
1301 root: Group,
1302 key_dir: Path,
1303 *,
1304 completion_max_age: int | None = None,
1305 tasks_file: str | None = None,
1306 path: Path | None = None,
1307 bake_cwd: bool = True,
1308 builtin: tuple[str, ...] = (),
1309 project: bool = True,
1310 user_dir: str | None = None,
1311) -> dict[str, Any]:
1312 """Build the fresh manifest and rewrite the cache only on a hash change.
1314 Called on the execution path, which has already paid to import the tree.
1315 The cache is keyed by *key_dir* (the cwd), since the effective task set is
1316 the cascade from the repo root down. The hash guard avoids needless disk
1317 writes (and mtime churn) when nothing about the command surface changed — a
1318 changed *completion_max_age* also forces a rewrite so a config edit takes
1319 effect.
1321 The write itself is best-effort. Everything the caller needs is the
1322 returned tree, which is already in hand; the file is only there to make
1323 the next TAB fast. A cache nobody can write to — a read-only HOME in a
1324 locked-down image, a stray `chmod` — must cost a cold completion, not
1325 every command, so an `OSError` here is swallowed the way `_progress`
1326 swallows the timing history's.
1327 """
1328 fresh = build_manifest(
1329 root,
1330 completion_max_age=completion_max_age,
1331 project=project,
1332 user_dir=user_dir,
1333 )
1334 if bake_cwd:
1335 # The directory this manifest describes, baked in (additive) so the
1336 # cache collector can tell a deleted project's leftovers from a
1337 # living one's without guessing from hashes. The global manifest
1338 # bakes none — it describes no directory, so the collector's idle
1339 # sweep owns it rather than the deleted-project rule.
1340 fresh["cwd"] = str(key_dir)
1341 if builtin:
1342 # What the refresh child rebuilds the global tree from: the entry-
1343 # point names, which travel where live objects cannot.
1344 fresh["builtin"] = sorted(builtin)
1345 if tasks_file:
1346 # Additive, like `cwd`: the background refresh reads it back, so a
1347 # branded CLI's custom filename survives a refresh it can't attend.
1348 fresh["tasks_file"] = tasks_file
1349 # What the user-level rungs looked like when this was built. The
1350 # resolver compares it to spot a config or user-tasks edit the age
1351 # clock would otherwise hide for a whole `max_age` — and it joins the
1352 # write guard below, or a rebuild triggered by a stamp change would
1353 # leave the old stamp in place and every later TAB would spawn again.
1354 fresh["user_stamp"] = _paths.user_stamp()
1355 # `path` lets a caller key the cache file separately from the baked
1356 # `key_dir` — a `-f` run caches by (cwd, file) yet still bakes the cwd, so
1357 # the collector prunes it with the project like any other.
1358 path = path or _paths.manifest_path(key_dir)
1359 cached = load_manifest(path)
1360 if (
1361 cached is None
1362 # An upgrade bumps the schema with the tree unchanged, so the hash
1363 # alone says "nothing to do" — and the old-schema file then lives
1364 # forever, every TAB refusing it, spawning a rebuild that "succeeds"
1365 # without writing, and paying the full cold bound. The schema is
1366 # part of what the file IS, not part of what it describes.
1367 or cached.get("schema") != SCHEMA_VERSION
1368 or cached.get("hash") != fresh["hash"]
1369 or cached.get("completion_max_age") != completion_max_age
1370 or cached.get("cwd") != fresh.get("cwd")
1371 or cached.get("tasks_file") != fresh.get("tasks_file")
1372 or cached.get("user_stamp") != fresh.get("user_stamp")
1373 ):
1374 # Guarded here rather than inside `write_manifest`, whose contract is
1375 # "the write landed, or you hear about it" — that is what its retry
1376 # loop is for. Only a caller can say the file was optional.
1377 with contextlib.suppress(OSError):
1378 write_manifest(fresh, path)
1379 return fresh