Coverage for src/footman/_complete.py: 93%
286 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-24 13:38 +0000
1"""Completion hot path — the code every TAB press runs.
3Hard constraints: standard library only, no framework import, no user-code
4import. The whole budget is ~50 ms including interpreter startup, so the work
5here is one file read + JSON parse + tree walk.
7Two ways in:
9* `footman --complete [--] WORD [WORD ...]` — the portable path. The console
10 script dispatches here *before importing anything else* and derives the cache
11 location from the current directory.
12* `python _complete.py --manifest PATH -- WORD [WORD ...]` — the baked-in
13 path. A generated completion script invokes the interpreter directly on this
14 file with the manifest location hard-coded, skipping the console-script shim
15 and the `footman` package import entirely.
17WORDs are the command line after the program name; the last word is the partial
18being completed ("" when the cursor follows a space).
19"""
21from __future__ import annotations
23import json
24import os
25import subprocess
26import sys
27import time
29# Hardcoded mirror of split.GLOBALS arity — the hot path can't import split (it
30# would pull the whole package). `test_completion_globals_mirror_split` rebuilds
31# these FROM split.GLOBALS, so renaming or re-typing a global fails CI.
32_GLOBAL_FLAG = frozenset(
33 {
34 "--help", "-h", "--version", "-V", "--list", "-l", "--tree",
35 "--dry-run", "-n", "--keep-going", "-k", "--fail-fast", "--sequential", "-s",
36 "--yes", "-y", "--no-input",
37 "--quiet", "-q", "--verbose", "-v", "--no-color", "--no-progress",
38 "--json", "--timings",
39 }
40) # fmt: skip
41_GLOBAL_VALUE = frozenset(
42 {
43 "--where",
44 "--directory",
45 "-C",
46 "--tasks-file",
47 "-f",
48 "--config",
49 "--jobs",
50 "-j",
51 "--color",
52 }
53) # consume the next word as the value
54_GLOBAL_MAYBE = frozenset(
55 {"--install-completion", "--setup-completion", "--uninstall-completion"}
56) # value optional
57# Value positions that are file paths. footman can't know the filesystem from a
58# cached manifest (and shouldn't try), so the resolver signals these and the
59# shell hooks defer to native file completion.
60_GLOBAL_FILES = frozenset({"--directory", "-C", "--tasks-file", "-f", "--config"})
61_FILES = "\x00files" # internal sentinel: complete() -> complete_cli()
62_EXIT_FILES = 100 # complete_cli exit code the hooks read as "complete files"
63_DYNAMIC = "\x00dynamic" # internal sentinel: a dynamic completer, recompute fresh
64_DYNAMIC_TIMEOUT = 2.0 # seconds to wait for a fresh dynamic completer subprocess
65_COLD_TIMEOUT = 3.0 # seconds to wait for a first-time cwd manifest build
66_SHELLS = ("bash", "zsh", "fish", "pwsh", "nushell")
67_GLOBAL_CHOICES = {
68 "--install-completion": _SHELLS,
69 "--setup-completion": _SHELLS,
70 "--uninstall-completion": _SHELLS,
71}
74def _consume_globals(prior: list[str]) -> tuple[list[str], str | None]:
75 """Strip leading global options (mirroring `split._parse_globals`).
77 Returns the remaining words (the task chain) and, when the partial itself is
78 a value-bearing global's value, that global's name — so `fm -C docs <TAB>`
79 treats `docs` as `-C`'s value instead of descending into a `docs` group.
80 """
81 i = 0
82 while i < len(prior):
83 word = prior[i]
84 name = word.split("=", 1)[0]
85 if name in _GLOBAL_FLAG:
86 i += 1
87 elif name in _GLOBAL_VALUE:
88 i += 1
89 if "=" in word: 89 ↛ 90line 89 didn't jump to line 90 because the condition on line 89 was never true
90 continue
91 if i >= len(prior):
92 return prior[i:], name # the value is the partial (no choices)
93 i += 1 # consume the value word
94 elif name in _GLOBAL_MAYBE:
95 i += 1
96 if "=" in word: 96 ↛ 97line 96 didn't jump to line 97 because the condition on line 96 was never true
97 continue
98 if i >= len(prior): 98 ↛ 100line 98 didn't jump to line 100 because the condition on line 98 was always true
99 return prior[i:], name # the partial completes its choices
100 if not prior[i].startswith("-"):
101 i += 1 # optional value present
102 else:
103 break # first non-global word: the task chain starts here
104 return prior[i:], None
107class _Segment:
108 """Walk state for one chain segment (mirrors the splitter's rules)."""
110 def __init__(self, task: dict | None = None) -> None:
111 self.task = task
112 self.opts: dict = {}
113 self.fixed: list[dict] = []
114 self.rest: dict | None = None
115 self.filled = 0
116 self.used: set[str] = set() # options already given in this segment
117 if task is not None:
118 params = task["params"]
119 self.opts = {
120 "--" + p["name"]: p for p in params if p["kind"] in ("flag", "option")
121 }
122 self.fixed = [
123 p for p in params if p["kind"] == "argument" and not p.get("multiple")
124 ]
125 self.rest = next(
126 (
127 p
128 for p in params
129 if (p["kind"] == "argument" and p.get("multiple"))
130 or p["kind"] == "variadic"
131 ),
132 None,
133 )
136def _describe(name: str, node: dict) -> str:
137 """`name\\tdescription` when *name* is a task/group in *node* with a help
138 line, else the bare name.
140 The tab is the backward-safe wire format: shells that render descriptions
141 (zsh, fish) split on it; bash (and others) keep the first field. Options and
142 choice values carry no help, so they pass through bare.
143 """
144 item = node["tasks"].get(name) or node["groups"].get(name)
145 summary = item.get("help") if isinstance(item, dict) else ""
146 return f"{name}\t{summary}" if summary else name
149def complete(tree: dict, words: list[str]) -> list[str]:
150 """Resolve completion candidates for *words* against a manifest *tree*.
152 Chain-aware: the walk tracks segments the way the splitter would — exact
153 positional arity first, then a trailing multiple/variadic consumer, then
154 the next bare word starts a new segment from the root. So in
155 `fm format lint --fi<TAB>` the options offered are *lint's*, and once a
156 task's arity is satisfied a bare TAB offers the next task names too.
157 `+` resets a segment explicitly; after `--` everything belongs to the
158 passthrough, so there is nothing to offer.
159 """
160 *prior, partial = words or [""]
162 # Leading global options bind before the task walk, exactly as the splitter
163 # consumes them — so `-C docs` reads `docs` as the value, not a group.
164 prior, value_global = _consume_globals(prior)
166 node, seg = tree, _Segment()
167 path: list[str] = [] # the group/task names of the current segment
168 value_opt: dict | None = None # the option whose value comes next
170 for word in prior:
171 if word == "--":
172 return [] # passthrough: the words after this aren't ours
173 if value_opt is not None:
174 if word == "=": # bash splits `--opt=val` into `--opt`, `=`, `val`;
175 continue # the `=` is a separator — stay armed for the value
176 value_opt = None
177 continue
178 if word == "+": # explicit segment boundary
179 node, seg, path = tree, _Segment(), []
180 continue
181 if seg.task is None:
182 if word in node["groups"]:
183 node = node["groups"][word]
184 path.append(word)
185 elif word in node["tasks"]: 185 ↛ 188line 185 didn't jump to line 188 because the condition on line 185 was always true
186 seg = _Segment(node["tasks"][word])
187 path.append(word)
188 continue
189 # Inside a task's tail: options and their values first.
190 name = word.split("=", 1)[0]
191 if name in seg.opts:
192 seg.used.add(name)
193 if seg.opts[name]["kind"] == "option" and "=" not in word:
194 value_opt = seg.opts[name]
195 continue
196 if name.startswith("--no-") and "--" + name[len("--no-") :] in seg.opts:
197 seg.used.add("--" + name[len("--no-") :])
198 continue
199 if word.startswith("-"): 199 ↛ 200line 199 didn't jump to line 200 because the condition on line 199 was never true
200 continue
201 # A bare word: a required positional, then the trailing consumer,
202 # then — arity satisfied — the start of the next segment (chains
203 # always resolve from the root).
204 if seg.filled < len(seg.fixed):
205 seg.filled += 1
206 continue
207 if seg.rest is not None: 207 ↛ 208line 207 didn't jump to line 208 because the condition on line 207 was never true
208 continue
209 node, seg, path = tree, _Segment(), []
210 if word in tree["groups"]:
211 node = tree["groups"][word]
212 path.append(word)
213 elif word in tree["tasks"]: 213 ↛ 170line 213 didn't jump to line 170 because the condition on line 213 was always true
214 seg = _Segment(tree["tasks"][word])
215 path.append(word)
217 # A leading global expecting a value (`fm --install-completion <TAB>`):
218 # offer its choices, if any (a PATH-valued global has none — the shell's
219 # default file completion covers it).
220 if value_global is not None:
221 if value_global in _GLOBAL_FILES:
222 return [_FILES] # a path value — hand off to the shell's file completion
223 return [
224 c for c in _GLOBAL_CHOICES.get(value_global, ()) if c.startswith(partial)
225 ]
227 # Value position: the previous word was an option expecting a value. A bash
228 # `--opt=<TAB>` can leave the `=` as the partial — strip it.
229 if value_opt is not None:
230 if partial.startswith("="): # bash `--opt=<TAB>` leaves `=` as the partial
231 partial = partial[1:]
232 if "path" in value_opt.get("types", []):
233 return [_FILES] # a Path-typed option value — files
234 if value_opt.get("dynamic"): # recompute fresh, never the baked snapshot
235 return [_DYNAMIC, partial, value_opt["name"], *path]
236 return [c for c in value_opt.get("choices", []) if c.startswith(partial)]
238 if seg.task is None:
239 names = list(node["groups"]) + list(node["tasks"])
240 out = [_describe(n, node) for n in names if n.startswith(partial)]
241 # A runnable group also offers its default action's flags/options, so
242 # `fm lint <TAB>` proposes `--fix` alongside the child names.
243 if "default" in node:
244 out += [
245 "--" + p["name"]
246 for p in node["default"]["params"]
247 if p["kind"] in ("flag", "option")
248 and ("--" + p["name"]).startswith(partial)
249 ]
250 # fm's own global options bind before the first task, so offer them when
251 # a flag is being typed at the root (`not prior` ⇒ nothing but globals
252 # preceded). A bare `<TAB>` still lists only tasks — globals would be
253 # noise there.
254 if not prior and partial.startswith("-"):
255 globals_ = _GLOBAL_FLAG | _GLOBAL_VALUE | _GLOBAL_MAYBE
256 out += [g for g in sorted(globals_) if g.startswith(partial)]
257 return out
259 # An attached `--opt=value` partial (zsh/fish don't split on `=`): offer the
260 # option's choices as full `--opt=choice` tokens.
261 if partial.startswith("-") and "=" in partial:
262 optname, _, valpart = partial.partition("=")
263 opt = seg.opts.get(optname)
264 if opt is not None and opt["kind"] == "option": 264 ↛ 272line 264 didn't jump to line 272 because the condition on line 264 was always true
265 choices = opt.get("choices", [])
266 return [f"{optname}={c}" for c in choices if c.startswith(valpart)]
268 # A path-typed positional (or trailing consumer): once the partial is a
269 # value being typed rather than an option, hand it to native file
270 # completion — the same handoff a Path-typed option value gets above.
271 # `-` still reaches the options below, so they stay one keystroke away.
272 if not partial.startswith("-"):
273 pending = seg.fixed[seg.filled] if seg.filled < len(seg.fixed) else seg.rest
274 if pending is not None:
275 if "path" in pending.get("types", []):
276 return [_FILES]
277 if pending.get("dynamic"): # recompute fresh, never the baked snapshot
278 return [_DYNAMIC, partial, pending["name"], *path]
280 # Option position: this task's flags/options — minus the ones already
281 # given, unless the param legitimately repeats — plus what the next bare
282 # word could be: the pending positional's choices, the trailing
283 # consumer's choices, or (arity satisfied) the next segment's names.
284 candidates = [
285 name
286 for name, p in seg.opts.items()
287 if name not in seg.used or p.get("multiple") or p.get("mapping")
288 ]
289 if seg.filled < len(seg.fixed):
290 candidates += seg.fixed[seg.filled].get("choices", [])
291 elif seg.rest is not None: 291 ↛ 292line 291 didn't jump to line 292 because the condition on line 291 was never true
292 candidates += seg.rest.get("choices", [])
293 elif not partial.startswith("-"):
294 candidates += list(tree["groups"]) + list(tree["tasks"])
295 seen: dict[str, None] = {}
296 for c in candidates:
297 if c.startswith(partial):
298 seen.setdefault(c)
299 # Next-segment task/group names carry their help line; an option carries
300 # its doc("...") text when the task author wrote one; choice values stay
301 # bare. Same tab-separated wire format either way.
302 out = []
303 for c in seen:
304 p = seg.opts.get(c)
305 if p is not None and p.get("doc"):
306 out.append(f"{c}\t{p['doc']}")
307 else:
308 out.append(_describe(c, tree))
309 return out
312def _load_manifest(path: str) -> dict | None:
313 try:
314 with open(path, "rb") as fh:
315 data = json.load(fh)
316 except (OSError, ValueError):
317 return None
318 return data if isinstance(data, dict) else None
321def _maybe_refresh(path: str, data: dict) -> None:
322 """Stale-while-revalidate: if the manifest is older than its baked
323 `completion_max_age`, bump its mtime and spawn a detached rebuild for *next*
324 time, then return. Never blocks the TAB (the rebuild imports the package and
325 shells completers) and never surfaces an error.
326 """
327 max_age = data.get("completion_max_age")
328 if not isinstance(max_age, int) or isinstance(max_age, bool) or max_age <= 0:
329 return # disabled (off, or an in-memory/`-f` manifest with no age baked)
330 try:
331 if time.time() - os.stat(path).st_mtime <= max_age:
332 return
333 # Bump the mtime *before* spawning: resets the clock even if the rebuild
334 # is a no-op (sync_manifest only writes on change), and storm-guards
335 # concurrent TABs so only the first in an aged window spawns.
336 os.utime(path)
337 except OSError:
338 return
339 _spawn_refresh()
342def _spawn_refresh(override: str | None = None) -> None:
343 # override set → rebuild that one -f file's (cwd, file) manifest; else the
344 # cwd cascade. The path rides as an argv word (not baked into the -c script),
345 # so a path with spaces or quotes needs no escaping.
346 if override:
347 script = (
348 "import sys; from footman import _refresh; "
349 "_refresh.refresh_source(sys.argv[1])"
350 )
351 cmd = [sys.executable, "-c", script, override]
352 else:
353 script = "from footman import _refresh; _refresh.refresh_cwd()"
354 cmd = [sys.executable, "-c", script]
355 null = subprocess.DEVNULL
356 try:
357 if os.name == "nt":
358 flags = getattr(subprocess, "DETACHED_PROCESS", 0) | getattr(
359 subprocess, "CREATE_NEW_PROCESS_GROUP", 0
360 )
361 subprocess.Popen(
362 cmd, stdin=null, stdout=null, stderr=null, creationflags=flags
363 )
364 else:
365 subprocess.Popen(
366 cmd, stdin=null, stdout=null, stderr=null, start_new_session=True
367 )
368 except OSError:
369 return # a background refresh must never break completion
372def _cold_build(manifest: str, override: str | None) -> dict | None:
373 """Build a cold-cache manifest once, then load it.
375 The first <kbd>Tab</kbd> in a fresh directory has nothing cached. Rather than
376 answer empty, spawn the same builder a real run uses and wait — bounded — for
377 it to land, then serve it (now cached for next time). *override* picks the
378 tree: a finished `-f <file>` builds that file's (cwd, file) manifest, else the
379 cwd cascade. Import-free on the hot path: it spawns rather than imports. A
380 slow `tasks.py` degrades to empty, and because the build was detached it still
381 finishes for the next TAB, so no keystroke ever hangs on it.
382 """
383 if override is not None:
384 from pathlib import Path
386 if not Path(override).expanduser().is_file():
387 return None # a still-being-typed or missing -f value: nothing to build
388 _spawn_refresh(override)
389 deadline = time.monotonic() + _COLD_TIMEOUT
390 while time.monotonic() < deadline:
391 time.sleep(0.03)
392 data = _load_manifest(manifest)
393 if isinstance(data, dict) and isinstance(data.get("tree"), dict):
394 return data
395 return None
398def _leading_global_value(args: list[str], names: tuple[str, ...]) -> str | None:
399 """The value of the first of *names* among the leading globals, or None.
401 Walks only the leading globals — stopping at the first task name — skipping
402 other flags and value-options by the same arity mirror the resolver uses.
403 """
404 i = 0
405 while i < len(args):
406 tok = args[i]
407 if tok in names:
408 return args[i + 1] if i + 1 < len(args) else None
409 if any(tok.startswith(n + "=") for n in names):
410 return tok.split("=", 1)[1]
411 name = tok.split("=", 1)[0]
412 if "=" not in tok and name in _GLOBAL_VALUE:
413 i += 2 # a value-option consumes the next word
414 elif name in _GLOBAL_FLAG or name in _GLOBAL_MAYBE or "=" in tok:
415 i += 1 # a flag, an option?, or --opt=value
416 else:
417 break # the first non-global — a task name (or its partial)
418 return None
421def _tasks_file_from(args: list[str]) -> str | None:
422 """The `-f`/`--tasks-file` value among the leading globals, or None."""
423 return _leading_global_value(args, ("-f", "--tasks-file"))
426def _emit(lines: list[str]) -> None:
427 """Write completion candidates, one per line, LF-terminated.
429 LF, always. The completion protocol is footman's own, and on Windows
430 text-mode stdout translates every "\\n" to "\\r\\n": a shell that reads lines
431 literally (git-bash's `read`) keeps the carriage return and completes
432 `--fix\\r`, planting a stray CR at the cursor. Writing bytes to the
433 underlying buffer skips the translation and pins UTF-8; captured stdout
434 (tests, some wrappers) has no buffer, so fall back.
435 """
436 if not lines: 436 ↛ 437line 436 didn't jump to line 437 because the condition on line 436 was never true
437 return
438 payload = "\n".join(lines) + "\n"
439 buffer = getattr(sys.stdout, "buffer", None)
440 if buffer is None: 440 ↛ 441line 440 didn't jump to line 441 because the condition on line 440 was never true
441 sys.stdout.write(payload)
442 else:
443 buffer.write(payload.encode("utf-8"))
444 buffer.flush()
447def _fresh_dynamic(param: str, path: list[str], args: list[str]) -> list[str] | None:
448 """Run *param*'s completer fresh in a subprocess; None on timeout/failure.
450 Isolated on purpose: the subprocess imports the framework and the user's
451 code, which the hot path must never do. A timeout or non-zero exit returns
452 None, and the caller shows nothing rather than a stale snapshot.
453 """
454 cmd = [sys.executable, "-m", "footman._suggest", "--param", param]
455 for name in path:
456 cmd += ["--path", name]
457 prior = args[:-1]
458 if (tf := _leading_global_value(prior, ("-f", "--tasks-file"))) is not None:
459 cmd += ["--tasks-file", tf]
460 if (cf := _leading_global_value(prior, ("--config",))) is not None:
461 cmd += ["--config", cf]
462 try:
463 proc = subprocess.run(
464 cmd, capture_output=True, text=True, timeout=_DYNAMIC_TIMEOUT
465 )
466 except (OSError, subprocess.TimeoutExpired):
467 return None
468 if proc.returncode != 0:
469 return None
470 return [v for v in proc.stdout.splitlines() if v]
473def complete_cli(args: list[str]) -> int:
474 """Entry for `footman --complete` and the standalone resolver."""
475 manifest = None
476 if args and args[0] == "--manifest":
477 if len(args) < 2: 477 ↛ 478line 477 didn't jump to line 478 because the condition on line 477 was never true
478 return 0
479 manifest, args = args[1], args[2:]
480 # WinPS 5.1 and pwsh 7.0-7.2 drop empty-string args to native commands, so
481 # the hook can't pass the trailing "" partial itself — it flags the empty
482 # position and we append the "" here instead.
483 empty_partial = False
484 if args and args[0] == "--empty-partial":
485 empty_partial, args = True, args[1:]
486 if args and args[0] == "--": 486 ↛ 488line 486 didn't jump to line 488 because the condition on line 486 was always true
487 args = args[1:]
488 if empty_partial:
489 args = [*args, ""]
491 derived = manifest is None
492 override: str | None = None
493 if manifest is None:
494 # Only the derive branch needs the package; keep the standalone
495 # --manifest path free of any `footman` import. The cache is keyed by
496 # cwd — the effective task set is the cascade from the repo root down —
497 # unless `-f <file>` names one file, which has its own (cwd, file) key.
498 from pathlib import Path
500 from footman import _paths
502 # The last word is the partial being completed: `fm -f <TAB>` is a file
503 # being typed, not a finished override — so read the override from the
504 # prior words only, leaving `-f`'s own value to native file completion
505 # (the resolver signals it below). A finished `-f file <TAB>` still keys
506 # by the pair.
507 override = _tasks_file_from(args[:-1])
508 manifest = str(
509 _paths.source_manifest_path(Path.cwd(), Path(override))
510 if override
511 else _paths.cwd_manifest_path()
512 )
514 data = _load_manifest(manifest)
515 if data is None or not isinstance(data.get("tree"), dict):
516 # Cold cache: rather than answer empty, build the manifest once (bounded)
517 # and serve it, so the first TAB in a fresh directory is accurate — for
518 # the cwd cascade and for a finished `-f <file>` alike.
519 data = _cold_build(manifest, override) if derived else None
520 if not isinstance(data, dict) or not isinstance(data.get("tree"), dict):
521 return 0 # cold and couldn't build in time — stay silent and fast
522 out = complete(data["tree"], args)
523 if out == [_FILES]:
524 # A path value: print nothing, and signal the hook to complete files.
525 _maybe_refresh(manifest, data)
526 return _EXIT_FILES
527 if out and out[0] == _DYNAMIC:
528 # A dynamic completer: recompute it fresh in a subprocess rather than
529 # serve the manifest's baked snapshot — a build-critical answer must not
530 # be stale. Empty on timeout or failure, never the old values.
531 partial, param, seg_path = out[1], out[2], out[3:]
532 fresh = _fresh_dynamic(param, seg_path, args)
533 if fresh is not None: 533 ↛ 535line 533 didn't jump to line 535 because the condition on line 533 was always true
534 _emit([c for c in fresh if c.startswith(partial)])
535 _maybe_refresh(manifest, data)
536 return 0
537 _emit(out)
538 _maybe_refresh(manifest, data) # SWR: refresh the baked fallback + structural set
539 return 0
542def main() -> int:
543 return complete_cli(sys.argv[1:])
546if __name__ == "__main__":
547 raise SystemExit(main())