Coverage for src/footman/schedule.py: 94%
301 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-24 13:38 +0000
1"""Build the task DAG and run it — in parallel by default, or sequentially.
3The chain segments plus each task's `pre`/`post` form a dependency graph
4(deduped by task identity). Independent nodes run concurrently on a thread pool;
5a node runs once all its prerequisites have succeeded. footman tasks are almost
6always I/O-bound (they shell out through `footman.run`, releasing the GIL),
7so threads give real concurrency without process isolation.
9Output is buffered per task and flushed atomically on completion, so concurrent
10tasks never interleave.
11"""
13from __future__ import annotations
15import io
16import os
17import sys
18import threading
19from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
20from dataclasses import dataclass, field
21from itertools import count
22from typing import Any, TextIO
24from footman import _describe, _progress, context, executor
25from footman.registry import (
26 Group,
27 Task,
28 _Opted,
29 default_group,
30 fans_out,
31 is_infinite,
32 is_interactive,
33 keeps_going,
34 post_tasks,
35 pre_tasks,
36 task_confirm,
37 wants_progress,
38)
39from footman.split import ChainError, Segment
42@dataclass
43class _Node:
44 fn: Task
45 seg: Segment
46 key: int
47 deps: set[int] = field(default_factory=set)
48 state: str = "pending" # pending / running / done / skipped
49 result: executor.TaskResult | None = None
50 forwarded: dict[str, Any] = field(default_factory=dict) # `forward`ed values in
51 forward_targets: list[_Node] = field(default_factory=list) # …and out
52 keep_going: bool = False # resolved failure policy for THIS node (per-subtree)
55def _default_seg(fn: Task) -> Segment:
56 return Segment(task=fn.__name__, path=[fn.__name__])
59def _dep_key(fn: Task) -> tuple[int, frozenset[tuple[str, Any]]]:
60 """The deduplication identity of a DAG dependency: its base task plus its
61 frozen option overrides. A bare task is simply "base with no overrides", so
62 it shares one uniform key shape with an `.opts()` reference — and an empty
63 `.opts()` (no override at all) collapses onto the bare task by construction,
64 with no int-vs-tuple asymmetry. Identical policies share a node (a shared
65 prerequisite still runs once); a genuinely different policy is a distinct
66 node. The override identity itself lives in `_Opted._dedup_key`."""
67 return fn._dedup_key() if isinstance(fn, _Opted) else (id(fn), frozenset())
70def _as_task(dep: Task | Group | _Opted) -> Task:
71 """A `pre`/`post` dependency may name a runnable group; resolve it to the
72 group's default action so it runs (and fans out) like any other task. An
73 `.opts()`-wrapped group resolves the same way, carrying its overrides onto
74 the default task."""
75 if isinstance(dep, _Opted):
76 base = dep._opted_base
77 if isinstance(base, Group):
78 return _Opted(_as_task(base), dep._opted_overrides)
79 return dep # opts on a task is already a valid task reference
80 if isinstance(dep, Group):
81 if dep.default_task is None: 81 ↛ 82line 81 didn't jump to line 82 because the condition on line 81 was never true
82 raise ChainError(
83 f"group {dep.name!r} is a prerequisite but has no @group.default"
84 )
85 return dep.default_task
86 return dep
89def _build_dag(root: Group, segments: list[Segment]) -> list[_Node]:
90 """Nodes for the chain plus their transitive pre/post deps.
92 Explicit chain segments each get their own node — repeating a task in the
93 chain (`build web build api`) runs it once per mention. Only shared
94 pre/post prerequisites are deduped, by task identity, so a prerequisite
95 pulled in twice still runs once. Node keys are serial ints; `dep_nodes`
96 maps a task to the node its bare deps resolve to.
97 """
98 nodes: list[_Node] = []
99 dep_nodes: dict[object, _Node] = {}
100 counter = count()
101 seen_explicit: set[object] = set()
103 def new_node(fn: Task, seg: Segment) -> _Node:
104 node = _Node(fn, seg, next(counter))
105 nodes.append(node)
106 return node
108 def add_dep(fn: Task) -> _Node:
109 node = dep_nodes.get(_dep_key(fn))
110 if node is None:
111 node = new_node(fn, _default_seg(fn))
112 dep_nodes[_dep_key(fn)] = node
113 _link(node)
114 return node
116 def _thread(dep: _Node, fmap: dict[str, Any], source: str) -> None:
117 # A forwarded value reaches only a dispatched task that *declares* the
118 # parameter (partial reach); two dispatchers sending different values to
119 # a shared prerequisite is a taught error, not a silent last-wins.
120 if not fmap:
121 return
122 declared = {
123 p.name for p in executor.resolved_signature(dep.fn).parameters.values()
124 }
125 for name, value in fmap.items():
126 if name not in declared:
127 continue
128 if name in dep.forwarded and dep.forwarded[name] != value:
129 raise ChainError(
130 f"{dep.seg.task}: {name!r} is forwarded with conflicting values "
131 f"(one from {source!r}); run the forwarding tasks separately"
132 )
133 dep.forwarded[name] = value
135 def _link(node: _Node) -> None:
136 pre = list(pre_tasks(node.fn))
137 # An empty-body group default fans out the group's own tasks: they become
138 # implicit prerequisites, so the scheduler runs them (in parallel) and the
139 # default's forward-marked values thread into the ones that declare them.
140 group = default_group(node.fn)
141 if group is not None and fans_out(node.fn):
142 pre = [*group.tasks.values(), *pre]
143 for dep in pre:
144 d = add_dep(_as_task(dep))
145 node.deps.add(d.key)
146 node.forward_targets.append(d) # forwarding threaded in a later pass
147 for dep in post_tasks(node.fn):
148 d = add_dep(_as_task(dep))
149 d.deps.add(node.key)
150 node.forward_targets.append(d)
152 for seg in segments:
153 fn = executor.resolve(root, seg.path)
154 key = _dep_key(fn)
155 existing = dep_nodes.get(key)
156 if existing is not None and key not in seen_explicit: 156 ↛ 159line 156 didn't jump to line 159 because the condition on line 156 was never true
157 # First explicit mention of a task already pulled in as a bare dep:
158 # adopt this segment's args instead of creating a duplicate.
159 existing.seg = seg
160 seen_explicit.add(key)
161 continue
162 node = new_node(fn, seg)
163 if existing is None:
164 dep_nodes[key] = node
165 seen_explicit.add(key)
166 _link(node)
168 # Thread forwarded values in a second pass, dependents before their deps (the
169 # reverse of the run order), so a node's *received* values are complete before
170 # it forwards on — this is what makes forwarding chain through a group default
171 # into its surfaces. It runs after segment adoption above, so each node's seg
172 # (hence its forward map) is final.
173 for node in reversed(_toposort(nodes)):
174 fmap = executor.forward_map(node.fn, node.seg, node.forwarded)
175 for target in node.forward_targets:
176 _thread(target, fmap, node.seg.task)
177 return nodes
180def _check_cycles(nodes: list[_Node]) -> None:
181 """Reject a cyclic dependency graph with a taught error naming the cycle.
183 Without this check the run loop would find no ready node, run nothing, and
184 exit 0 — a silent success that lies.
185 """
186 by_key = {n.key: n for n in nodes}
187 state: dict[int, int] = {} # 1 = on the current path, 2 = fully explored
189 def visit(node: _Node, path: list[str]) -> None:
190 state[node.key] = 1
191 path.append(node.seg.task)
192 for dep in node.deps:
193 child = by_key.get(dep)
194 if child is None: 194 ↛ 195line 194 didn't jump to line 195 because the condition on line 194 was never true
195 continue
196 mark = state.get(child.key, 0)
197 if mark == 1:
198 cycle = [*path[path.index(child.seg.task) :], child.seg.task]
199 raise ChainError(
200 f"dependency cycle: {' -> '.join(cycle)} "
201 f"(check the pre/post declarations of these tasks)"
202 )
203 if mark == 0:
204 visit(child, path)
205 path.pop()
206 state[node.key] = 2
208 for node in nodes:
209 if state.get(node.key, 0) == 0:
210 visit(node, [])
213def _toposort(nodes: list[_Node]) -> list[_Node]:
214 """Deps before dependents, stable by appearance order."""
215 by_key = {n.key: n for n in nodes}
216 result: list[_Node] = []
217 seen: set[int] = set()
219 def visit(node: _Node) -> None:
220 if node.key in seen:
221 return
222 seen.add(node.key)
223 for dep in node.deps:
224 if dep in by_key: 224 ↛ 223line 224 didn't jump to line 223 because the condition on line 224 was always true
225 visit(by_key[dep])
226 result.append(node)
228 for node in nodes:
229 visit(node)
230 return result
233def _plain_output(no_color: bool) -> bool:
234 """No colour at all: the `--no-color` flag, `NO_COLOR`, or a dumb terminal.
236 Per D6 this means the live rewrite is *absent*, not rewritten without escape
237 codes — the same output a pipe gets.
238 """
239 return no_color or "NO_COLOR" in os.environ or os.environ.get("TERM") == "dumb"
242def _make_ctx(
243 seg: Segment,
244 ctx_config: dict[str, Any] | None,
245 *,
246 sequential: bool,
247 capture: bool,
248 real: TextIO,
249 name_width: int = 0,
250 keep_going: bool = False,
251) -> context.Context:
252 ctx = context.Context(**(ctx_config or {}), passthrough=list(seg.passthrough or []))
253 ctx.keep_going = keep_going # per-subtree policy; tags this task's subprocesses
254 # One buffer for both streams at task level: the atomic flush keeps this
255 # task's stdout/stderr in order, while a run() inside it still splits the
256 # step's streams via a temporary swap of the two.
257 ctx.sink = ctx.err_sink = None if (sequential and not capture) else io.StringIO()
258 # Step lines dress for their *destination*: a buffered block replays
259 # onto `real`, so its children style exactly as parallel() children
260 # style for their parent's terminal — both engines, one look. Only
261 # liveness (sink is None, judged in run()) gates in-place rewrites.
262 ctx.tty = not capture and real.isatty() and not _plain_output(ctx.no_color)
263 # `--color=always` forces colour even off a terminal, but not into a captured
264 # envelope: ANSI in `--json` stdout would corrupt it, so capture wins here
265 # exactly as it does for `tty` above.
266 ctx.force_color = ctx.force_color and not capture
267 ctx.task = seg.task
268 ctx.name_width = name_width
269 return ctx
272def resolve_keep_going(root: Group, segments: list[Segment], cli: bool | None) -> bool:
273 """A run-wide *summary* of the failure policy: does anything keep going?
275 An explicit command-line choice (`-k` / `--fail-fast`) wins; unspecified,
276 true if any invoked task — a chain task or a `pre`/`post` prerequisite —
277 declares (or `.opts()`-overrides) `keep_going=True`. The scheduler resolves
278 the actual *per-node* policy with `_scope_keep_going`; this summary is for
279 callers that just want the one-bit answer.
280 """
281 if cli is not None:
282 return cli
283 try:
284 nodes = _build_dag(root, segments)
285 except (KeyError, IndexError, ChainError):
286 return False # a malformed chain surfaces its real error in run_plan
287 return any(keeps_going(n.fn) is True for n in nodes)
290def _scope_keep_going(nodes: list[_Node], cli: bool | None) -> None:
291 """Assign each node its failure policy — the per-subtree scoping.
293 A command-line `-k`/`--fail-fast` wins run-wide. Otherwise each node takes
294 its own declared (or `.opts()`-overridden) `keep_going`, and a keep-going
295 node propagates that down its own subtree — its `pre`/`post` prerequisites
296 keep going with it — so a mixed chain honours each side: a keep-going gate
297 surfaces all of its own failures while an independent fail-fast task still
298 bails on the first. A node's own policy always wins over an inherited one, so
299 an explicit fail-fast prerequisite stays a fail-fast boundary. Unspecified
300 everywhere is the built-in fail-fast.
301 """
302 if cli is not None:
303 for node in nodes:
304 node.keep_going = cli
305 return
306 inherited: set[int] = set() # keys a keep-going dependent has reached
307 for node in reversed(_toposort(nodes)): # dependents resolve before their deps
308 own = keeps_going(node.fn) # True / False / None (reads declared + opted)
309 node.keep_going = own if own is not None else node.key in inherited
310 if node.keep_going:
311 inherited.update(node.deps) # keep this subtree's prerequisites going
314def dag_wants_progress(root: Group, segments: list[Segment]) -> bool:
315 """Whether every task in the expanded DAG — pre/post deps included —
316 consented to timing. One `@task(progress=False)` opts the run out of
317 recording and of a determinate bar (the pulse still shows)."""
318 return all(wants_progress(n.fn) for n in _build_dag(root, segments))
321class NotConfirmed(Exception):
322 """A `@task(confirm=…)` gate was declined (or unanswerable off a terminal)."""
324 def __init__(self, task: str) -> None:
325 super().__init__(f"{task}: not confirmed")
328def _ask_confirm(message: str, *, no_input: bool) -> bool:
329 """The `@task(confirm=)` gate. Off a terminal or under `--no-input` the
330 answer is no — like just and go-task, a confirm fails without `--yes`
331 rather than proceeding unasked. Asked on stderr before output routing."""
332 if no_input or not context._stdin_is_tty():
333 return False
334 reply = context._prompt_core(f"{message} [y/N] ", default="n")
335 return reply.strip().lower() in ("y", "yes")
338def _gate_confirms(
339 root: Group, segments: list[Segment], ctx_config: dict[str, Any] | None
340) -> tuple[list[Segment], list[executor.TaskResult]]:
341 """Resolve each invoked task's `@task(confirm=)` before the DAG is built —
342 asked in invocation order, before any prerequisite runs. A confirmed task
343 is kept; a denied one is dropped (so its exclusive pre-deps are pruned with
344 it) and reported as a failed 'not confirmed' result, so the run exits
345 non-zero. `--yes` auto-confirms every gate."""
346 cfg = ctx_config or {}
347 assume_yes = bool(cfg.get("assume_yes"))
348 no_input = bool(cfg.get("no_input"))
349 kept: list[Segment] = []
350 denied: list[executor.TaskResult] = []
351 for seg in segments:
352 message = task_confirm(executor.resolve(root, seg.path))
353 if not message or assume_yes or _ask_confirm(message, no_input=no_input):
354 kept.append(seg)
355 else:
356 denied.append(
357 executor.TaskResult(
358 task=seg.task, ok=False, code=1, error=NotConfirmed(seg.task)
359 )
360 )
361 return kept, denied
364def run_plan(
365 root: Group,
366 segments: list[Segment],
367 *,
368 sequential: bool = False,
369 keep_going: bool | None = None,
370 capture: bool = False,
371 ctx_config: dict[str, Any] | None = None,
372 estimate: _progress.Estimate | None = None,
373 progress: bool = True,
374 jobs: int = 0,
375) -> list[executor.TaskResult]:
376 """Build and run the DAG; return results in dependency order."""
377 context.reset_abort() # clear any latched fail-fast from a previous run
378 segments, denied = _gate_confirms(root, segments, ctx_config)
379 nodes = _build_dag(root, segments)
380 _check_cycles(nodes)
381 _scope_keep_going(nodes, keep_going) # per-node failure policy (tri-state + scope)
382 # One node has nothing to parallelise — run it on the sequential-live
383 # path instead: output streams as it happens, and run()'s TTY mode
384 # (colour, in-place step rewrite) applies. `fm check` is this shape. An
385 # interactive task also forces sequential: it owns the terminal, so it
386 # can't share with parallel siblings (a human-wait is the bottleneck).
387 # An interactive task owns the real terminal: it forces sequential (it can't
388 # share with parallel siblings) and suppresses the status line, whose
389 # clear-line repaints would otherwise erase its prompt.
390 interactive = any(is_interactive(n.fn) for n in nodes)
391 sequential = sequential or len(nodes) == 1 or interactive
392 # A run containing an infinite task has no progress to show — its
393 # duration isn't late, it's intentional. The status line yields to a
394 # one-time hint (printed at the node's start) saying how this ends.
395 endless = any(is_infinite(n.fn) for n in nodes)
396 # Resolve colour once for the whole run and publish it into os.environ, so
397 # every tool — subprocess (inherits) and in-process (reads it) — sees one
398 # answer set once, with no per-call environment patching. sys.stdout is the
399 # run's real stdout here (routing captures it next), so its tty-ness is the
400 # same input `_make_ctx` uses per node.
401 cfg = ctx_config or {}
402 try:
403 stdout_tty = sys.stdout.isatty()
404 except Exception:
405 stdout_tty = False
406 colour_on = context.run_colour_on(
407 no_color=bool(cfg.get("no_color")),
408 force_color=bool(cfg.get("force_color")),
409 capture=capture,
410 isatty=stdout_tty,
411 )
412 with context.routing() as (real, err):
413 # Decide the status line and the infinite-task hint from the *user's*
414 # environment — before `color_environment` publishes footman's own
415 # NO_COLOR/FORCE_COLOR for the children, which `_plain_output` would
416 # otherwise read back and mistake for a user's monochrome request (a
417 # `fm check > log` still wants its spinner on the stderr terminal).
418 status = _make_status(
419 err,
420 ctx_config,
421 capture,
422 estimate,
423 progress and not endless and not interactive,
424 )
425 hint_err = (
426 err
427 if sequential
428 and endless
429 and not capture
430 and not cfg.get("quiet")
431 and err.isatty()
432 and not _plain_output(bool(cfg.get("no_color")))
433 else None
434 )
435 if status is not None:
436 status.unit_added(len(nodes))
437 context.set_status(status) # parallel() and the routers find it
438 status.open()
439 try:
440 with context.color_environment(colour_on):
441 if sequential:
442 try:
443 _run_sequential(
444 nodes, real, capture, ctx_config, status, hint_err
445 )
446 except BaseException:
447 # Ctrl-C mid-task: the running child is group-isolated, so
448 # it missed the terminal's SIGINT — reap its tree by hand
449 # before the interrupt propagates. (Parallel does this.)
450 context.terminate_live_children()
451 raise
452 else:
453 _run_parallel(nodes, real, err, capture, ctx_config, status, jobs)
454 finally:
455 if status is not None:
456 context.set_status(None)
457 status.close()
458 return denied + [n.result for n in _toposort(nodes) if n.result is not None]
461def _run_sequential(nodes, real, capture, ctx_config, status, err=None) -> None:
462 done: dict[int, bool] = {}
463 failed = False
464 width = max((len(n.seg.task) for n in nodes), default=0)
465 for node in _toposort(nodes):
466 if any(not done.get(d) for d in node.deps) or (failed and not node.keep_going):
467 node.state = "skipped"
468 if status is not None: 468 ↛ 469line 468 didn't jump to line 469 because the condition on line 468 was never true
469 status.unit_skipped(node.seg.task)
470 continue
471 ctx = _make_ctx(
472 node.seg,
473 ctx_config,
474 sequential=True,
475 capture=capture,
476 real=real,
477 name_width=width,
478 keep_going=node.keep_going,
479 )
480 if status is not None:
481 status.unit_started(node.seg.task)
482 if err is not None and is_infinite(node.fn):
483 hint = f"{node.seg.task} runs until you stop it — Ctrl-C"
484 err.write(_describe.dim(hint, True) + "\n")
485 err.flush()
486 node.result = executor.run_task(node.fn, node.seg, ctx, node.forwarded)
487 node.state = "done"
488 if status is not None:
489 status.unit_finished(node.seg.task, node.result.ok)
490 done[node.key] = node.result.ok
491 failed = failed or not node.result.ok
494def _make_status(
495 err: TextIO,
496 ctx_config: dict[str, Any] | None,
497 capture: bool,
498 estimate: _progress.Estimate | None,
499 enabled: bool,
500) -> _progress.StatusLine | None:
501 """The run's live line — bar or pulse — or None when it can't show.
503 Status is commentary, so it lives on stderr: piping stdout
504 (`fm check > log`) keeps the line visible on the terminal. Applies to
505 every run shape, single node included — that's `fm check`.
506 """
507 cfg = ctx_config or {}
508 if (
509 not enabled
510 or capture
511 or cfg.get("quiet")
512 or not err.isatty() # the status stream's own tty-ness decides
513 or _plain_output(bool(cfg.get("no_color")))
514 ):
515 return None
516 # Past the guard the run is colourful by definition (no_color/NO_COLOR/dumb
517 # all bail above), so the live line always renders with escapes.
518 return _progress.StatusLine(err, estimate, color=True)
521def _run_parallel(nodes, real, err, capture, ctx_config, status, jobs) -> None:
522 by_key = {n.key: n for n in nodes}
523 lock = threading.Lock()
524 failed = False
525 width = max((len(n.seg.task) for n in nodes), default=0)
527 def dep_ok(n: _Node) -> bool:
528 return all(
529 by_key[d].state == "done" and by_key[d].result and by_key[d].result.ok
530 for d in n.deps
531 if d in by_key
532 )
534 def dep_lost(n: _Node) -> bool:
535 def lost(m: _Node) -> bool:
536 return m.state == "skipped" or (
537 m.state == "done" and bool(m.result) and not m.result.ok # type: ignore[union-attr]
538 )
540 return any(lost(by_key[d]) for d in n.deps if d in by_key)
542 def run_node(n: _Node) -> None:
543 ctx = _make_ctx(
544 n.seg,
545 ctx_config,
546 sequential=False,
547 capture=capture,
548 real=real,
549 name_width=width,
550 keep_going=n.keep_going,
551 )
552 n.result = executor.run_task(n.fn, n.seg, ctx, n.forwarded)
553 if not capture: # flush this task's buffered output as one block 553 ↛ exitline 553 didn't return from function 'run_node' because the condition on line 553 was always true
554 with lock:
555 blob = ctx.sink.getvalue() # type: ignore[union-attr]
556 if status is not None:
557 # A direct real-stream write (bypasses the routers): the
558 # status line clears itself and tracks the column.
559 status.notify(blob)
560 real.write(blob)
561 real.flush()
563 with ThreadPoolExecutor(max_workers=jobs if jobs > 0 else None) as pool:
564 futures: dict[Any, _Node] = {}
565 try:
566 while True:
567 for n in nodes:
568 if n.state == "pending" and (
569 dep_lost(n) or (failed and not n.keep_going)
570 ):
571 n.state = "skipped"
572 if status is not None: 572 ↛ 573line 572 didn't jump to line 573 because the condition on line 572 was never true
573 status.unit_skipped(n.seg.task)
574 for n in nodes:
575 if n.state == "pending" and dep_ok(n):
576 n.state = "running"
577 if status is not None:
578 status.unit_started(n.seg.task)
579 futures[pool.submit(run_node, n)] = n
580 if not futures:
581 break
582 completed, _ = wait(list(futures), return_when=FIRST_COMPLETED)
583 for fut in completed:
584 node = futures.pop(fut)
585 # `run_task` catches task exceptions itself; anything the
586 # future carries (KeyboardInterrupt in the worker, an
587 # internal error) must propagate, not read as success.
588 exc = fut.exception()
589 if exc is not None: 589 ↛ 590line 589 didn't jump to line 590 because the condition on line 589 was never true
590 raise exc
591 node.state = "done"
592 ok = bool(node.result and node.result.ok)
593 if status is not None:
594 status.unit_finished(node.seg.task, ok)
595 if not ok:
596 failed = True
597 # True fail-fast: stop launching new nodes (the skip pass
598 # above) *and* reap the FAIL-FAST siblings already in
599 # flight, so a doomed branch dies now instead of waiting
600 # out a five-minute test suite. `failfast_only` spares a
601 # keep-going task in a mixed run — it isn't doomed.
602 context.terminate_live_children(failfast_only=True)
603 except BaseException:
604 # Abort (Ctrl-C, or an internal error surfaced above): drop
605 # everything not yet started, then kill in-flight subprocess trees.
606 # This must happen *before* the pool's `with` exit joins the worker
607 # threads: each is blocked in communicate() on a group-isolated child
608 # that no longer receives the terminal's SIGINT, so without an
609 # explicit kill the join — and the whole Ctrl-C — would hang. The app
610 # layer reports "interrupted" and exits 130; run_plan's finally
611 # clears the status line.
612 context.terminate_live_children()
613 pool.shutdown(wait=False, cancel_futures=True)
614 raise