Coverage for src/footman/_step.py: 94%
251 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-06 02:59 +0000
1"""Steps you make yourself: the lifter, the built item, and the pump.
3`step()` is one name in three grammatical positions. Decorating a local
4function (plain or generator) makes a **maker**: calling the maker BUILDS
5a bound, deferrable piece of work — the `range(10)` precedent, owned — it
6runs nothing. `with step("title"):` records a block of your own code
7where it stands. `step(fn, title=…)` is the expression form of the
8decorator, for functions you didn't write.
10A built item is callable — calling it executes it here, under the current
11task's management — which is exactly what `parallel()` does with one. The
12pump drives generator items: every bare `yield` is a checkpoint, the one
13kind of place a running item can be cancelled (fail-fast, Ctrl-C, or its
14own `timeout=` — all three arrive as "the loop never resumes"), and every
15yield evaluates to the item's own draft, so a title decided mid-work is a
16plain attribute write.
17"""
19from __future__ import annotations
21import inspect
22import io
23import time
24from collections.abc import Callable, Generator
25from contextvars import ContextVar
26from typing import Any, Generic, Literal, ParamSpec, TypeVar, cast, overload
28from footman import _describe, _signals
29from footman import context as _context
30from footman.context import (
31 AuditEntry,
32 Failed,
33 Result,
34 ResultView,
35 RunFailed,
36 RunTimeout,
37 _audit_entry,
38)
40P = ParamSpec("P")
41R = TypeVar("R")
42R_co = TypeVar("R_co", covariant=True)
44# Births inside a `with parallel()` block, so the block can refuse to run
45# while an item built in it was never handed over: building runs nothing,
46# and a dead item is almost always a forgotten `p(...)`. `None` outside a
47# block — the common case pays one contextvar read per build.
48_born: ContextVar[list[WorkItem[Any]] | None] = ContextVar(
49 "footman_step_births", default=None
50)
52# The closed policy vocabulary a step maker's `.opts()` accepts — execution
53# policy only: a step is anonymous, so boundary policy (confirm, gates,
54# sharing) has no request boundary to resolve at and is not spellable here.
55_STEP_OPTS = (
56 "title",
57 "capture",
58 "recorded",
59 "timeout",
60 "env",
61 "color",
62 "lanes",
63 "pre_record",
64)
67class WorkItem(Generic[R_co]):
68 """A bound, deferrable piece of work with a record to come.
70 Building one runs nothing. Calling it executes it under the current
71 task — captured, timed, recorded, reviewable — and returns the body's
72 value; a non-zero verdict raises `RunFailed`, exactly as `run()` does,
73 so a failing item fails whatever asked for it.
74 """
76 __slots__ = ("_args", "_claimed", "_kwargs", "_maker")
78 def __init__(
79 self, maker: StepFn[..., R_co], args: tuple[Any, ...], kwargs: dict[str, Any]
80 ) -> None:
81 self._maker = maker
82 self._args = args
83 self._kwargs = kwargs
84 self._claimed = False
85 born = _born.get()
86 if born is not None:
87 born.append(self)
89 @property
90 def __name__(self) -> str: # the label status lines and reports use
91 return self._maker.__name__
93 def __call__(self) -> R_co:
94 self._claimed = True
95 return cast(R_co, _pump(self))
97 def __repr__(self) -> str:
98 return f"<step {self.__name__} (built, not run)>"
101class StepFn(Generic[P, R_co]):
102 """What the lifter returns: calling it builds the bound item.
104 The maker carries the step's identity and policy: `.opts()` refines
105 execution policy per use; `.pre_record` attaches its reviewer and
106 `.post_step` its observer — permanently, wherever the maker travels.
107 """
109 __slots__ = ("_fn", "_is_gen", "_observers", "_opts", "_reviewers")
111 def __init__(
112 self,
113 fn: Callable[..., Any],
114 opts: dict[str, Any] | None = None,
115 reviewers: list[Callable[[ResultView], None]] | None = None,
116 observers: list[Callable[[Result], None]] | None = None,
117 ) -> None:
118 self._fn = fn
119 self._is_gen = inspect.isgeneratorfunction(fn)
120 self._opts: dict[str, Any] = opts or {}
121 # Shared by reference across `.opts()` copies: attachment is
122 # permanent — it changes what the step does for every user of the
123 # maker, exactly like a task's handle attachments.
124 self._reviewers = reviewers if reviewers is not None else []
125 self._observers = observers if observers is not None else []
127 @property
128 def __name__(self) -> str:
129 title = self._opts.get("title")
130 if isinstance(title, str) and title:
131 return title
132 return getattr(self._fn, "__name__", "step")
134 def __call__(self, *args: P.args, **kwargs: P.kwargs) -> WorkItem[R_co]:
135 return WorkItem(self, args, kwargs)
137 def opts(self, **overrides: Any) -> StepFn[P, R_co]:
138 """Per-use execution policy; the maker stays callable as itself."""
139 unknown = sorted(set(overrides) - set(_STEP_OPTS))
140 if unknown:
141 valid = ", ".join(_STEP_OPTS)
142 raise TypeError(
143 f"step .opts() got unknown option(s) {unknown}; valid options "
144 f"are {valid}. A step is anonymous, so boundary policy "
145 f"(confirm, sharing, gates) needs a declared task to live on."
146 )
147 if (color := overrides.get("color", "auto")) not in ("auto", "never", "always"):
148 # The same tri-state `run(color=)` takes, refused the same way:
149 # one vocabulary for colour wherever a caller decides it.
150 raise ValueError(
151 f"step .opts(color={color!r}) expects one of auto|never|always "
152 f"— auto follows the run's own decision"
153 )
154 if overrides.get("lanes") is not None:
155 from footman.registry import validate_lanes
157 overrides["lanes"] = validate_lanes(overrides["lanes"])
158 return StepFn(
159 self._fn,
160 {**self._opts, **overrides},
161 self._reviewers,
162 self._observers,
163 )
165 def pre_record(
166 self, hook: Callable[[ResultView], None], /
167 ) -> Callable[[ResultView], None]:
168 """Attach this step's reviewer — the draft, before the record seals."""
169 self._reviewers.append(hook)
170 return hook
172 def post_step(self, hook: Callable[[Result], None], /) -> Callable[[Result], None]:
173 """Watch this step's sealed record — read-only; veto via `fail()`."""
174 self._observers.append(hook)
175 return hook
178class _StepBlock:
179 """`with step("title") as s:` — record a block of your own code.
181 The record-only lift: no execution boundary is created (the block's
182 statements run exactly as they would bare, dry-run included), so what
183 the record keeps is the title, the verdict, and the honest duration.
184 The handle is the block's own draft — self-review is writing to it.
185 """
187 __slots__ = ("_start", "_view")
189 def __init__(self, title: str) -> None:
190 self._view = ResultView(
191 title=title,
192 code=0,
193 stdout="",
194 stderr="",
195 duration=0.0,
196 raw="",
197 command=title,
198 )
199 self._start = 0.0
201 def __enter__(self) -> ResultView:
202 self._start = time.perf_counter()
203 return self._view
205 def __exit__(self, exc_type: Any, exc: Any, tb: Any) -> Literal[False]:
206 duration = time.perf_counter() - self._start
207 view = self._view
208 code = view.code
209 if exc_type is not None and "code" not in view._touched:
210 code = 1 # the block failed; a self-review that already ruled wins
211 ctx = _context.current()
212 shown = str(_describe.redact(view.title))
213 result = Result(
214 code,
215 command=view.title,
216 shown=shown,
217 duration=duration,
218 address=_context._child_address(ctx, shown),
219 audit=(_audit_entry("body", shown, code),),
220 )
221 ctx.steps.append(result)
222 return False # an exception propagates; the record sealed first
225@overload
226def step(
227 target: Callable[P, Generator[None, ResultView, R]],
228 /,
229 *,
230 title: str | None = None,
231) -> StepFn[P, R]: ...
232@overload
233def step(target: Callable[P, R], /, *, title: str | None = None) -> StepFn[P, R]: ...
234@overload
235def step(target: str, /) -> _StepBlock: ...
236def step(target: Any = None, /, *, title: str | None = None) -> Any:
237 """One name, three positions: `@step` on a def, `with step("title"):`
238 over a block, `step(fn, title=…)` around a function you didn't write.
240 Decorator and expression positions are the same expression, so both
241 return the maker — calling it builds the item; nothing runs until
242 something executes it (`parallel()`, or calling the built item).
243 """
244 if isinstance(target, str):
245 if title is not None:
246 raise TypeError(
247 "step('title') is the block form — pass the title once, "
248 "positionally; title= belongs to the decorator/expression "
249 "forms (step(fn, title=…))"
250 )
251 return _StepBlock(target)
252 if target is None: 252 ↛ 253line 252 didn't jump to line 253 because the condition on line 252 was never true
253 raise TypeError(
254 "step() needs its subject: a function to lift (@step / "
255 "step(fn, title=…)) or a title string for the block form "
256 "(with step('…'):)"
257 )
258 if not callable(target): 258 ↛ 259line 258 didn't jump to line 259 because the condition on line 258 was never true
259 raise TypeError(
260 f"step() lifts a callable or opens a titled block; got "
261 f"{type(target).__name__}"
262 )
263 opts: dict[str, Any] = {}
264 if title is not None:
265 opts["title"] = title
266 return StepFn(target, opts)
269def _pump(item: WorkItem[Any]) -> Any:
270 """Execute a built item under the current task: capture, drive, review,
271 seal, observe — one record, committed once."""
272 maker = item._maker
273 ctx = _context.current()
274 label = maker.__name__
275 o = maker._opts
276 capture: bool = o.get("capture", True)
277 recorded: bool = o.get("recorded", True)
278 timeout: float | None = o.get("timeout")
279 env: dict[str, str] | None = o.get("env")
280 color: str = o.get("color", "auto")
281 lanes: tuple[Any, ...] = tuple(o.get("lanes", ()))
282 if any(getattr(ln, "name", "") == "console" for ln in lanes):
283 raise TypeError(
284 "the console follows interactive task bodies — claim it with "
285 "@task(interactive=True) (or lanes=(console_lane,) on the task); "
286 "a step-level console hold waits for a payload that needs one."
287 )
289 # A title is a plain string the author wrote, so it redacts by the same
290 # rule a command line does: the record keeps it, everything printed —
291 # the receipt, the address, the audit — reads the shown form.
292 shown = str(_describe.redact(label))
293 addr = _context._child_address(ctx, shown)
294 if ctx.dry_run and recorded:
295 # Dry-run fakes what footman owns and records: a deferred maker is
296 # exactly that. Declared, not executed — an empty audit says so.
297 result = Result(0, command=label, shown=shown, address=addr)
298 ctx.steps.append(result)
299 return None
301 view = ResultView(
302 title=label, code=0, stdout="", stderr="", duration=0.0, raw="", command=label
303 )
304 start = time.perf_counter()
305 deadline = start + timeout if timeout is not None else None
306 value: Any = None
307 error: BaseException | None = None
308 stack_text = ""
309 timed_out = False
310 out_buf, err_buf = io.StringIO(), io.StringIO()
312 def drive() -> Any:
313 nonlocal timed_out
314 if not maker._is_gen:
315 produced = maker._fn(*item._args, **item._kwargs)
316 if inspect.iscoroutine(produced):
317 # An `async def` is not a generator function, so it lands here
318 # and the call builds a coroutine that nothing drives: the step
319 # recorded a receipt for work that never happened. A generator
320 # body is the supported way to yield, and it is a cancellation
321 # point rather than a scheduling one.
322 produced.close() # or "was never awaited" trails the refusal
323 raise _context.coroutine_refusal("step", maker._fn)
324 if inspect.isasyncgen(produced):
325 # The coroutine refusal's missed sibling: `async def` plus
326 # `yield` builds an async generator — not a coroutine, so the
327 # check above never sees it, and not a generator function, so
328 # the pump never takes it. Sealed `ok` for zero work, same as
329 # the coroutine case, refused the same way.
330 raise _context.generator_refusal("step", maker._fn, is_async=True)
331 return produced
332 gen: Generator[Any, Any, Any] = maker._fn(*item._args, **item._kwargs)
333 payload: Any = None
334 while True:
335 try:
336 yielded = gen.send(payload)
337 except StopIteration as stop:
338 return stop.value
339 if yielded is not None:
340 # The vocabulary is closed: a bare yield is a checkpoint and
341 # evaluates to the draft — the value channel is reserved for
342 # a future that earns it, so using it is an error, not noise.
343 gen.close()
344 raise TypeError(
345 f"step {label!r} yielded {yielded!r} — a step's yields "
346 f"are checkpoints and carry nothing out. Write a bare "
347 f"`yield`; read the draft with `view = yield`; return "
348 f"the step's value with `return`."
349 )
350 payload = view # every yield evaluates to the item's own draft
351 # The checkpoint: the only place a running item is cancelled —
352 # fail-fast/Ctrl-C aborts, or the item's own timeout. All three
353 # arrive the same way: the loop never resumes, the generator's
354 # own try/finally runs, nothing is left half-held.
355 if deadline is not None and time.perf_counter() > deadline:
356 timed_out = True
357 gen.close()
358 return None
359 # The enclosing task's deadline, checked beside the step's own:
360 # `@task(timeout=…)` is the fail-fast event scoped to one task, so
361 # it stops a generator step at exactly the point a sibling's
362 # failure would. The step records `timed_out` either way — what
363 # ran out is a property of the record above it, not of this one.
364 # The enclosing task's deadline, checked beside the step's own:
365 # `@task(timeout=…)` is the fail-fast event scoped to one task, so
366 # it stops a generator step at exactly the point a sibling's
367 # failure would. The step records `timed_out` either way — what
368 # ran out is a property of the record above it, not of this one.
369 if ctx.overdue():
370 timed_out = True
371 gen.close()
372 return None
373 if _context._aborting.is_set() and not ctx.keep_going: 373 ↛ 374line 373 didn't jump to line 374 because the condition on line 373 was never true
374 gen.close()
375 return None
377 # Same env rule as every run(): `env=` is the whole environment, absent
378 # inherits the task's own — applied through the same overlay machinery.
379 # One honest exception keeps bare steps parallel: outside a routed run,
380 # with no env asked for, there is nothing to overlay — and the global
381 # patch would take a process-wide lock that serialises the pool.
382 from contextlib import nullcontext
384 from footman import _globals
386 overlay = dict(env) if env is not None else dict(ctx.env)
387 if color != "auto":
388 # The same door `run(color=)` opens, on the work item that wraps a
389 # whole body: the tri-state lands in the environment this step runs
390 # under, so anything it calls — a nested run(), an in-process tool
391 # reading the variables — sees one decision. Off means REMOVING the
392 # inherited force variables, never writing "0".
393 for key in _context._COLOR_VARS:
394 overlay.pop(key, None)
395 overlay.update(_context.color_env(color == "always"))
396 if _globals.active():
397 state: Any = _context._env_overlay(ctx, overlay)
398 elif env is not None or color != "auto": 398 ↛ 399line 398 didn't jump to line 399 because the condition on line 398 was never true
399 state = _context._process_state(overlay)
400 else:
401 state = nullcontext()
402 exit_code: int | None = None
403 try:
404 with _globals.named_lanes(lanes, name=label), state:
405 if capture:
406 with _context._captured_streams(out_buf, err_buf):
407 value = drive()
408 else:
409 value = drive()
410 except SystemExit as exc: # the classic "fail this step" idiom, honoured
411 if isinstance(exc.code, int):
412 exit_code = exc.code
413 elif exc.code is None:
414 exit_code = 0
415 else: # a string reason: print it, fail with 1 — sys.exit's contract
416 err_buf.write(f"{exc.code}\n")
417 exit_code = 1
418 except Failed as exc:
419 # A deliberate stop, not a crash: `footman.fail()` and the refusals
420 # built on it carry a reason written for the person reading it, and
421 # `Failed` promises that reason renders verbatim. A traceback here
422 # buried it under footman's own frames and made a considered stop look
423 # like an internal error. The reason reaches the report through the
424 # sealed record, so it is not written twice.
425 error = exc
426 except (KeyboardInterrupt, GeneratorExit, _signals.Stop):
427 # Control flow, not a failing step — the same three `_call` re-raises,
428 # for the same reasons. GeneratorExit especially: the pump below raises
429 # it into a step to cancel one, so sealing a record around it here would
430 # swallow footman's own cancellation. `Stop` is a supervisor's SIGTERM
431 # arriving in whichever step happened to be running; a step that
432 # answered for it would be reporting someone else's decision as its own.
433 raise
434 except BaseException as exc: # the step failed; the record still seals
435 # Everything else that leaves a step body is that step failing, which
436 # is the rule tasks already follow. `except Exception` let a
437 # BaseException past the record, so no step row sealed and the receipt
438 # was lost — the task still failed, correctly, but with nothing to say
439 # which step did it.
440 error = exc
441 # Trimmed of footman's own leading frames, so the first line is the one
442 # somebody wrote. It goes into the record unconditionally — `--json`
443 # and `recording()` read that, and a machine consumer is not reading
444 # for noise — while the *display* below follows the same rule the task
445 # summary does. Kept in hand as well as written, so the display can
446 # take it back out again without guessing where it starts.
447 stack_text = _describe.user_traceback(exc)
448 err_buf.write(stack_text)
450 duration = time.perf_counter() - start
451 if timed_out:
452 code = 124
453 elif exit_code is not None:
454 code = exit_code
455 elif error is not None:
456 code = 1
457 else:
458 code = 0
459 view._fill(out_buf.getvalue(), err_buf.getvalue(), duration)
460 view.code = code
461 view._touched.discard("code") # machinery write, not a review verdict
463 audit: tuple[AuditEntry, ...] = (_audit_entry("body", shown, code),)
464 reviewers = list(maker._reviewers)
465 if o.get("pre_record") is not None: 465 ↛ 466line 465 didn't jump to line 466 because the condition on line 465 was never true
466 reviewers.append(o["pre_record"]) # the use site keeps the final word
467 if recorded:
468 for reviewer in reviewers:
469 name = getattr(reviewer, "__name__", repr(reviewer))
470 try:
471 reviewer(view)
472 except Exception as exc:
473 ctx.steps.append(
474 Result(
475 code,
476 command=label,
477 shown=shown,
478 stdout=out_buf.getvalue(),
479 stderr=err_buf.getvalue(),
480 duration=duration,
481 timed_out=timed_out,
482 address=addr,
483 audit=(*audit, _audit_entry("review", name, None)),
484 )
485 )
486 raise RuntimeError(
487 f"pre_record hook {name!r} failed reviewing {shown!r}: "
488 f"{type(exc).__name__}: {exc}"
489 ) from exc
490 audit = (
491 *audit,
492 _audit_entry(
493 "review", name, view.code if "code" in view._touched else None
494 ),
495 )
497 # A reviewer may rename the record; the name it goes out under is shown
498 # by the same rule the original was.
499 if view.title != label:
500 shown = str(_describe.redact(view.title))
501 result = Result(
502 view.code,
503 command=view.title,
504 shown=shown,
505 stdout=out_buf.getvalue(),
506 stderr=err_buf.getvalue(),
507 duration=duration,
508 timed_out=timed_out,
509 address=addr,
510 audit=audit,
511 )
512 if recorded:
513 ctx.steps.append(result)
514 # Task grain at normal verbosity: the receipt shows under --verbose
515 # (and for uncaptured, live items) — and always when it failed.
516 # Under --json the item has a row of its own, so the receipt would
517 # only arrive twice: once as chrome inside the task's `output`, once
518 # as the fields a reader parses. Same call as run()'s step line.
519 if (
520 not ctx.quiet
521 and not ctx.machine_read
522 and (ctx.verbose or not capture or result.code != 0)
523 ):
524 import sys as _sys
526 out = _sys.stdout
527 out.write(
528 _context._step_line(ctx, result.code == 0, result.shown, duration)
529 )
530 combined = result.stdout + result.stderr
531 if stack_text:
532 # The stack stays in the record — `--json` and `recording()`
533 # read that — and out of the display, which the run summary
534 # owns. One place decides how a failure is placed, so a task
535 # and a step answer identically and neither says it twice.
536 # What the body printed before it fell over is untouched.
537 combined = combined.replace(stack_text, "")
538 if capture and combined and (result.code != 0 or ctx.verbose):
539 out.write(combined if combined.endswith("\n") else combined + "\n")
540 if result.code != 0 and len(result.audit) > 1: 540 ↛ 541line 540 didn't jump to line 541 because the condition on line 540 was never true
541 trail = " → ".join(
542 f"{e.moment} {e.actor}"
543 + (f" {e.code}" if e.code is not None else "")
544 for e in result.audit
545 )
546 out.write(
547 _context._dim(f" audit: {trail}", _context._colored(ctx)) + "\n"
548 )
549 out.flush()
550 for observer in maker._observers:
551 name = getattr(observer, "__name__", repr(observer))
552 try:
553 observer(result)
554 except _context.Failed as exc:
555 amended = Result(
556 exc.code,
557 command=result.command,
558 shown=result.shown,
559 stdout=result.stdout,
560 stderr=result.stderr,
561 duration=result.duration,
562 timed_out=result.timed_out,
563 address=result.address,
564 audit=(*result.audit, _audit_entry("observe", name, exc.code)),
565 )
566 ctx.steps[-1] = amended
567 raise
568 except Exception as exc:
569 amended = Result(
570 result.code if result.code != 0 else 1,
571 command=result.command,
572 shown=result.shown,
573 stdout=result.stdout,
574 stderr=result.stderr,
575 duration=result.duration,
576 timed_out=result.timed_out,
577 address=result.address,
578 audit=(*result.audit, _audit_entry("observe", name, None)),
579 )
580 ctx.steps[-1] = amended
581 raise RuntimeError(
582 f"post_step hook {name!r} failed observing {result.shown!r}: "
583 f"{type(exc).__name__}: {exc}"
584 ) from exc
586 if result.code != 0:
587 if timed_out:
588 raise RunTimeout(result, _context.effective_timeout(timeout, ctx))
589 if error is not None:
590 raise error
591 raise RunFailed(result)
592 return value