Coverage for src/footman/_futures.py: 94%
350 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-06 02:59 +0000
1"""Tasks as run-scoped futures: what happens when a task body calls a task.
3`artifact = build()` inside a task body used to be a plain function call — it
4ran `build` again even if the DAG had just run it, on the caller's thread, with
5no context of its own and no `TaskResult` anywhere. That was the one execution
6path the framework couldn't see.
8Now a call routes through here, and per **(task, resolved arguments, starting
9environment)** the run keeps a once-cell:
11* already run → the memoised value comes back;
12* running on another thread → the caller blocks on its future;
13* neither → the caller claims it and runs it **inline, on its own thread**.
15So `pre=[build]` plus `build()` in the body is a cache hit, not a second build,
16and the callee gets full task treatment: its own context, its own lane
17decision, its own reported result.
19Sharing is a property of the *request*, not of this layer: an unshared request
20(`@task(shared=False)`, `.opts(shared=False)`, or inherited from the task that
21asked) neither reads a cell nor becomes one — its execution is its own, and a
22later shared request runs fresh rather than reusing it. `schedule` resolves the
23same ladder for DAG nodes, so a declared request and a called one behave the
24same.
26Two things are deliberately *not* here. A call outside a run is a plain call —
27importing a tasks file and calling a function must keep working. And lane
28acquisition stays at the task boundary, so a body call to a `serial=`/
29`exclusive=` task is refused rather than deadlocking mid-body.
30"""
32from __future__ import annotations
34import enum
35import io
36import itertools
37import threading
38from contextlib import contextmanager
39from contextvars import ContextVar
40from typing import TYPE_CHECKING, Any
42from footman import _coerce, registry
43from footman._split import ChainError
45if TYPE_CHECKING:
46 from collections.abc import Generator
48 from footman._executor import TaskResult
50# The memo key's argument half when the arguments cannot be frozen (an
51# unhashable value with no obvious frozen form — a live object, a generator).
52# Such a call is honest work every time rather than a wrong cache hit.
53_UNKEYABLE = object()
56class _Settled:
57 """The five-method slice of `Future` this module actually uses.
59 `concurrent.futures` costs ~4.9 ms to import — it drags `logging` through
60 `traceback`, and `logging` drags `_colorize` — and every run that executes
61 a task claims a cell, so every run paid it. Nothing here needs the
62 executor half of that module: a cell is written once by its owner and read
63 by whoever waits, which is a `threading.Event` and two slots.
64 """
66 __slots__ = ("_done", "_exc", "_value")
68 def __init__(self) -> None:
69 self._done = threading.Event()
70 self._value: Any = None
71 self._exc: BaseException | None = None
73 def done(self) -> bool:
74 return self._done.is_set()
76 def set_result(self, value: Any) -> None:
77 self._value = value
78 self._done.set()
80 def set_exception(self, exc: BaseException) -> None:
81 self._exc = exc
82 self._done.set()
84 def result(self) -> Any:
85 # No timeout: the wait graph above this layer is what refuses a claim
86 # that could not be satisfied, so a wait here is always one that ends.
87 self._done.wait()
88 if self._exc is not None:
89 raise self._exc
90 return self._value
93class _Cell:
94 """One (task, arguments) execution: its future, its owner, its label."""
96 __slots__ = ("future", "label", "owner", "record")
98 def __init__(self, owner: int, label: str) -> None:
99 self.future = _Settled()
100 self.owner = owner # the thread that claimed it (for the wait graph)
101 self.label = label
102 self.record: Any = None # the execution's sealed row, for reuse
105class _Session:
106 """The run's cells, plus the wait-for graph that keeps them honest."""
108 def __init__(self) -> None:
109 self.lock = threading.Lock()
110 self.cells: dict[Any, _Cell] = {}
111 self.waits: dict[int, Any] = {} # thread -> the key it is blocked on
112 self.results: list[TaskResult] = []
113 # Tasks defined *while this run was in flight* — `(group, name, what
114 # was there before)` — put back the way they were when it ends.
115 self.ephemeral: list[tuple[Any, str, Any]] = []
118_active: _Session | None = None
120# The run-wide request counter: every request for task-shaped work takes a
121# number at the moment it is *made* — plan order for scheduled segments, the
122# written line for a `parallel()` block's queued calls, the call moment for
123# body calls. The report is ordered by it outright: the clock cannot do the
124# job, because two independent tasks start in whatever order the pool hands
125# them workers, and that reshuffles between runs of the same command.
126_seq = itertools.count()
127# A queued call re-enters `call()` on a pool worker; the queue moment (the
128# written line) already took its number, carried here around the invocation.
129_pending_seq: ContextVar[int | None] = ContextVar("footman_request_seq", default=None)
132@contextmanager
133def session() -> Generator[_Session]:
134 """Install a fresh cell registry for the duration of one run.
136 Run-scoped by construction: memoisation means "this work happened in *this*
137 run", never a cache that outlives it. Nested installs (a `Runner` inside a
138 task, tests) keep the outer session — one run, one memo.
139 """
140 global _active
141 if _active is not None: 141 ↛ 142line 141 didn't jump to line 142 because the condition on line 141 was never true
142 yield _active
143 return
144 _active = _Session()
145 try:
146 yield _active
147 finally:
148 run, _active = _active, None
149 _sweep_ephemeral(run)
152def active_session() -> _Session | None:
153 """The run's cell registry, or `None` outside a run."""
154 return _active
157def _sweep_ephemeral(run: _Session) -> None:
158 """Undo the registrations a run made in its own course.
160 A `@task` written inside a task body is ordinary Python and makes a real,
161 callable task — but the manifest was written before the run started, so
162 such a task is invisible to every listing and would go on shadowing the
163 tree for the rest of the process. It lives for the run that made it, and
164 the tree it was grafted onto goes back to what it was.
165 """
166 for group, name, previous in reversed(run.ephemeral):
167 if previous is None: 167 ↛ 170line 167 didn't jump to line 170 because the condition on line 167 was always true
168 group.tasks.pop(name, None)
169 else:
170 group.tasks[name] = previous
173def collected() -> list[TaskResult]:
174 """Results of tasks run by body calls this run, in completion order."""
175 return list(_active.results) if _active is not None else []
178def _freeze(value: Any) -> Any:
179 """A hashable normal form for one argument value, or `_UNKEYABLE`.
181 Collections get frozen shapes (a list and a tuple of the same items are
182 the same work), and anything already hashable is its own key. What has no
183 frozen form is not forced into one: it reads as unkeyable, and its call
184 runs.
185 """
186 if isinstance(value, enum.Enum):
187 # Before the scalar line on purpose: IntEnum and str-valued enums
188 # would otherwise freeze as their value face. The normal form
189 # canonicalises an enum argument as its token — the member is the
190 # meaning, the value is representation — so renumbering a member
191 # does not churn work identity, and every spelling that reached
192 # binding froze to the same key anyway (they resolve to one member
193 # before anything downstream looks).
194 return ("enum", type(value).__name__, _coerce.token_of(value))
195 if isinstance(value, (str, bytes, int, float, bool, type(None))):
196 return value
197 if isinstance(value, (list, tuple)):
198 parts = tuple(_freeze(v) for v in value)
199 return _UNKEYABLE if _UNKEYABLE in parts else parts
200 if isinstance(value, (set, frozenset)):
201 parts = tuple(sorted((_freeze(v) for v in value), key=repr))
202 return _UNKEYABLE if _UNKEYABLE in parts else frozenset(parts)
203 if isinstance(value, dict):
204 items = tuple(
205 sorted(((_freeze(k), _freeze(v)) for k, v in value.items()), key=repr)
206 )
207 return _UNKEYABLE if any(_UNKEYABLE in p for p in items) else items
208 try:
209 hash(value)
210 except TypeError:
211 return _UNKEYABLE
212 return value
215def unshared(task: Any) -> bool:
216 """Whether *this request* must run rather than reuse an execution.
218 The sharing ladder for a call: the reference's own `.opts(shared=…)` or the
219 task's declaration, then what the calling task inherited (`ctx.shared` — an
220 unshared request asks unshared for everything it needs), then shared. The
221 scheduler resolves the same ladder for a node.
222 """
223 from footman import context
225 own = registry.sharing(task)
226 if own is not None:
227 return not own
228 ctx = context._current.get()
229 return ctx is not None and not ctx.shared
232def _env_digest(env: dict[str, str]) -> str:
233 """The environment half of a cell key.
235 Order-blind (a dict's insertion order must not split cells), case-blind
236 exactly where the OS is (`_norm`: Windows), and a digest rather than the
237 pairs themselves so cell keys stay small — equality is all a key needs.
238 """
239 import hashlib
241 from footman import _globals
243 h = hashlib.sha1(usedforsecurity=False)
244 for k, v in sorted((_globals._norm(k), v) for k, v in env.items()):
245 h.update(k.encode("utf-8", "surrogateescape"))
246 h.update(b"=")
247 h.update(v.encode("utf-8", "surrogateescape"))
248 h.update(b"\0")
249 return h.hexdigest()
252def _key(
253 task: Any,
254 args: Any,
255 kwargs: dict[str, Any],
256 given: frozenset[str],
257 env: dict[str, str],
258) -> Any | None:
259 """The cell key for this work: the task's dedup identity, its arguments in
260 the same normal form binding produces, what its caller actually asked
261 for, and the environment it would start from — so a DAG node and a body
262 call that resolve to the same arguments the same way name one piece of
263 work. `None` when the arguments have no frozen form.
265 *given* joins the key because `apply_defaults()` below erases the very
266 distinction `given()` reads: `build()` and `build(profile=<the default>)`
267 freeze to identical arguments, and a task that branches on presence does
268 different work for each. Keyed on values alone they would silently share a
269 cell, and the second request would be answered by the first — no error, and
270 nothing in the report to say the two were ever different.
272 *env* joins it for the same shape of bug with the other input a body can
273 observe: a caller that set `ctx.env` before calling would otherwise be
274 answered by an execution that never saw its variable — a wrong cache hit
275 with no error. Every DAG node starts from the pinned snapshot and an
276 untouched caller holds an equal copy, so digests diverge — splitting the
277 cell, honestly — exactly when someone wrote the environment on purpose.
278 """
279 from footman import _manifest
281 # The caller's signature: `ctx` is injected at the task boundary, so it is
282 # stripped before binding — binding against the declared signature would
283 # put the first positional value in the `ctx` slot and key every call on
284 # the wrong arguments.
285 try:
286 bound = _manifest.call_signature(task).bind(*args, **kwargs)
287 except TypeError:
288 return None # a call that won't bind: let it raise where it is made
289 bound.apply_defaults()
290 frozen = tuple((name, _freeze(value)) for name, value in bound.arguments.items())
291 if any(value is _UNKEYABLE for _, value in frozen):
292 return None
293 return (registry.work_key(task), frozen, given, _env_digest(env))
296def _deadlock(run: _Session, key: Any, me: int) -> list[str] | None:
297 """The wait-for chain proving this wait can never resolve, or `None`.
299 The static DAG has `_check_cycles`; a call graph only exists at runtime, so
300 it is walked here: follow the key to its owning thread, then that thread's
301 own wait, until either the chain ends (fine) or it arrives back at this
302 thread (a cycle — nobody would ever finish).
303 """
304 chain: list[str] = []
305 at: Any = key
306 seen: set[Any] = set()
307 while at is not None and at not in seen:
308 seen.add(at)
309 cell = run.cells.get(at)
310 if cell is None or cell.future.done(): 310 ↛ 311line 310 didn't jump to line 311 because the condition on line 310 was never true
311 return None # finished work blocks nobody
312 chain.append(cell.label)
313 if cell.owner == me:
314 return chain
315 at = run.waits.get(cell.owner)
316 return None
319def _label(task: Any) -> str:
320 return registry.cli_name(getattr(task, "__name__", "task"))
323def call(task: Any, args: tuple[Any, ...], kwargs: dict[str, Any]) -> Any:
324 """A task called from Python — the body-call path.
326 Outside a run this is the plain function call it always was. Inside one it
327 is the once-cell: hit the memo, wait for the thread that is running it, or
328 claim it and run it here.
329 """
330 from footman import context as _context
332 # The request takes its number now, in the caller's thread — for a
333 # queued call this IS the written line; the thunk carries the number
334 # back in through `_pending_seq` when the pool later runs it.
335 carried = _pending_seq.get()
336 seq = carried if carried is not None else next(_seq)
338 # A `with parallel()` block collects instead of running: the call is
339 # queued and answers with a `Pending`. Checked first, because the block
340 # is explicit intent — it holds whether or not a run is in flight.
341 if (queued := _context._queue_call(task, args, kwargs, seq)) is not None:
342 return queued
344 run = _active
345 if run is None:
346 _refuse_wide_moment(task)
347 if isinstance(task, registry._Opted):
348 return task._plain_call(args, kwargs)
349 return registry.task_body(task)(*args, **kwargs)
350 _refuse_unrunnable(task)
351 from footman import _executor, _schedule, context
353 label = _label(task)
354 seg = _schedule._default_seg(task)
355 # Availability is a per-request gate on the declared path (`run_task`
356 # checks it before the window opens), so a call checks it in the same
357 # place: before any hook fires.
358 if (refusal := _executor.unavailable(task, seg)) is not None:
359 refusal.seq = seq
360 _record(refusal)
361 raise refusal.error or ChainError(f"{label} is unavailable")
363 # A call binds like a segment: omitted parameters consult the same sources
364 # binding would (stdin, env, a required `ask()`), explicit values are
365 # validated against their annotation, and resolution happens before the
366 # key is computed so identity reads the values the body will receive.
367 # With a lifecycle armed, the callee's context is born first and made
368 # current around binding, so a `pre_bind` hook's `task.env` writes reach
369 # `env()` fallbacks here exactly as they do on the declared path.
370 life = _executor._lifecycle
371 child: Any = None
372 handle: Any = None
373 if life is not None or registry.has_own_hooks(task):
374 parent = context.current()
375 buf = io.StringIO()
376 child = parent.child(
377 label,
378 fn=task,
379 cwd=None, # let the callee's own cwd policy resolve
380 sink=buf,
381 err_sink=buf,
382 )
383 handle = _executor.TaskHandle(task, seg, child)
384 if (err := _executor._enter_bind_hooks(life, handle)) is not None:
385 # The attempt concluded before binding: the posts fire, the row
386 # is recorded, and the failure raises at the call site.
387 result = _executor._result(seg, 1, None, err, 0.0)
388 result.seq = seq
389 _executor._exit_task_hooks(life, handle, result)
390 _record(result)
391 raise err
392 token = context._current.set(child)
393 child.in_task = True
394 try:
395 args, kwargs, supplied = _executor.bind_call(task, args, kwargs)
396 except Exception as exc:
397 result = _executor._result(seg, _executor.EX_USAGE, None, exc, 0.0)
398 result.seq = seq
399 _executor._exit_task_hooks(life, handle, result)
400 _record(result)
401 raise
402 finally:
403 context._current.reset(token)
404 child.given = supplied
405 else:
406 args, kwargs, supplied = _executor.bind_call(task, args, kwargs)
407 # The environment the callee would start from, as known right here: the
408 # already-born child's (post `pre_bind`, exactly what the declared path
409 # keys on), else a copy of the caller's — which is what `child()` would
410 # copy at birth. An untouched caller holds the run snapshot, so this
411 # digests equal to a DAG node's and sharing is preserved.
412 birth_env = (child if child is not None else context.current()).env
413 key = _key(task, args, kwargs, supplied, birth_env)
414 if key is None: # arguments with no frozen form: honest work every time
415 return _run_now(
416 task,
417 args,
418 kwargs,
419 seg=seg,
420 child=child,
421 handle=handle,
422 seq=seq,
423 given=supplied,
424 )
425 if unshared(task):
426 # Asked for unshared: it neither reads a cell nor becomes one. Its
427 # result is its own — the run's answer is only ever an execution that
428 # was itself shareable, so how much work a run does cannot depend on
429 # which of two nodes the scheduler happened to start first.
430 return _run_now(
431 task,
432 args,
433 kwargs,
434 shared=False,
435 seg=seg,
436 child=child,
437 handle=handle,
438 seq=seq,
439 given=supplied,
440 )
442 me = threading.get_ident()
443 claimed, cell = _claim(run, key, me, label)
444 if claimed: # nobody had run it: this thread owns it, inline, right here
445 try:
446 # The cell rides along so the sealed row lands on it *before* the
447 # future resolves — a later sharer copies what this execution
448 # reported (title, audit, reported value) and seats after it.
449 value = _run_now(
450 task,
451 args,
452 kwargs,
453 seg=seg,
454 child=child,
455 handle=handle,
456 cell=cell,
457 seq=seq,
458 given=supplied,
459 )
460 except BaseException as exc:
461 cell.future.set_exception(exc)
462 raise
463 cell.future.set_result(value)
464 return value
465 # The pair is per request — only the body is shared. The pre fires before
466 # the wait, so a span honestly covers it; the post closes the request
467 # with its `shared` row. A crashing hook must not pass silently here
468 # either: it fails this request, at the call site.
469 if handle is not None:
470 handle._bind(args, kwargs)
471 if (hook_error := _executor._enter_task_hooks(life, handle)) is not None:
472 row = _executor._result(seg, 1, None, hook_error, 0.0)
473 row.seq = seq
474 _executor._exit_task_hooks(life, handle, row)
475 _record(row)
476 with run.lock:
477 run.waits.pop(me, None)
478 raise hook_error
479 # The waiting request is a unit like any other — a scheduler node
480 # satisfied by sharing counts, so a body call satisfied by sharing does
481 # too, and the caller's block on the share is visible on the line.
482 status = _claim_unit(label)
483 try:
484 # A finished cell answers instantly (the memo); a live one blocks until
485 # the thread that claimed it is done, and both hand back one value.
486 value = cell.future.result()
487 except BaseException:
488 if status is not None: 488 ↛ 489line 488 didn't jump to line 489 because the condition on line 488 was never true
489 status.unit_finished(label, False)
490 raise
491 finally:
492 with run.lock:
493 run.waits.pop(me, None)
494 if status is not None:
495 status.unit_finished(label, True)
496 row = _shared_result(cell.label, value, cell.record, seq=seq)
497 row.address = (
498 child.address
499 if child is not None
500 else context._child_address(context.current(), label)
501 )
502 if handle is not None:
503 post_error = _executor._exit_task_hooks(life, handle, row)
504 folded = _executor.fold_post_error(row, post_error, clear_state=True)
505 if folded is not None:
506 _record(row)
507 raise folded
508 _record(row)
509 return value
512def _claim(run: _Session, key: Any, me: int, label: str) -> tuple[bool, _Cell]:
513 """Take ownership of *key*'s cell, or hand back the cell already there.
515 Returns `(claimed, cell)` — `claimed` means this thread must run the work.
516 Waiting on a cell is registered in the wait-for graph while the lock is
517 held, so a cycle is caught before anyone blocks on it.
518 """
519 with run.lock:
520 cell = run.cells.get(key)
521 if cell is None:
522 cell = _Cell(me, label)
523 run.cells[key] = cell
524 return True, cell
525 if cell.future.done():
526 return False, cell # already run this run: the memo answers
527 if (chain := _deadlock(run, key, me)) is not None:
528 raise ChainError(
529 f"{label}: this call can never return — "
530 f"{' → '.join([*chain, chain[0]])} waits on itself. "
531 f"A task cannot call (directly or through another task) "
532 f"a task that is waiting for it; declare the order with "
533 f"pre= instead."
534 )
535 run.waits[me] = key
536 return False, cell
539def work_of(
540 task: Any,
541 args: Any,
542 kwargs: dict[str, Any],
543 given: frozenset[str],
544 env: dict[str, str],
545) -> Any | None:
546 """This request's cell key, or `None` when it cannot have one."""
547 return _key(task, tuple(args), kwargs, given, env)
550def retire(key: Any) -> None:
551 """Drop a finished cell so the next request for it runs fresh.
553 The one caller is the retry loop: an attempt that failed with attempts
554 left must not be the answer a later request receives. Left in place, the
555 memo would hand every subsequent attempt the first attempt's failure —
556 the body would run once and be reported N times, which is the opposite
557 of "each attempt IS a record".
559 Only ever called between attempts of work this thread owns, and only on
560 a resolved cell. A requester that joined *during* the failed attempt has
561 already been handed it: sharing binds to the terminal attempt for
562 everyone who asks from here on, not retroactively.
563 """
564 run = _active
565 if run is None or key is None: 565 ↛ 566line 565 didn't jump to line 566 because the condition on line 565 was never true
566 return
567 with run.lock:
568 cell = run.cells.get(key)
569 if cell is not None and cell.future.done(): 569 ↛ exitline 569 didn't jump to the function exit
570 del run.cells[key]
573def claim(key: Any, label: str) -> tuple[bool, Any]:
574 """Take this work, or hand back the cell that already holds it.
576 `(claimed, cell)` — `claimed` means the caller must perform the work and
577 then `resolve` the cell. Otherwise the cell is someone else's: `join` it.
578 `(True, None)` when there is no run to share within, so the caller simply
579 proceeds.
581 A node comes through here exactly as a body call does. Peeking would not do:
582 it can only see a *finished* cell, so two independent requests racing each
583 other would both run — the duplication the cell exists to prevent, and
584 unpredictable because it depends on which one started first.
585 """
586 run = _active
587 if run is None or key is None:
588 return True, None
589 return _claim(run, key, threading.get_ident(), label)
592def join(cell: Any) -> Any:
593 """Wait for the execution that holds this cell, and return its value.
595 Raises whatever the work raised, so a request answered by a failed
596 execution fails too. Occupies this thread while it waits, bounded by
597 `jobs`, exactly as a waiting body call does.
598 """
599 me = threading.get_ident()
600 try:
601 return cell.future.result()
602 finally:
603 if (run := _active) is not None:
604 with run.lock:
605 run.waits.pop(me, None)
608def resolve(
609 cell: Any, value: Any, error: BaseException | None, record: Any = None
610) -> None:
611 """Hand this claimed cell its outcome, so anyone waiting is answered.
613 Always called for a claimed cell, on every path out — an unresolved cell
614 would leave a waiter blocked for the rest of the run. *record* is the
615 execution's sealed row, when there is one: a shared answer is the record
616 reused, so a later request's row copies what this one *reported* —
617 reviewed title, reported value, audit — not just the body's value.
618 """
619 if cell is None or cell.future.done():
620 return
621 cell.record = record
622 if error is not None:
623 cell.future.set_exception(error)
624 else:
625 cell.future.set_result(value)
628def shared_result(label: str, value: Any, record: Any = None) -> TaskResult:
629 """The report entry for a request an earlier execution satisfied."""
630 return _shared_result(label, value, record)
633def _record(result: TaskResult) -> None:
634 """Add a called task's outcome to the run's report — ran, refused, denied,
635 or satisfied by an execution the run had already performed."""
636 if (run := _active) is not None: 636 ↛ exitline 636 didn't return from function '_record' because the condition on line 636 was always true
637 run.results.append(result)
640def _shared_result(
641 label: str, value: Any, record: Any = None, seq: int | None = None
642) -> TaskResult:
643 """The report entry for a request the run had already satisfied.
645 Recorded rather than left invisible: the work happened, and a reader (or a
646 journal) should see that a second request for it was answered instead of
647 silently vanishing. Nothing blocked it — it was answered, instantly or
648 after a wait — so it carries no blame; `blocked_by` belongs to rows that
649 are holes. The *request* has a real moment, the instant it concluded, and
650 that is its `started`: the report seats it where it actually happened,
651 exactly as an executed body-callee seats after its caller. `ok` is true
652 because the work did succeed, just earlier, so the exit code is untouched.
653 """
654 import time
656 from footman._executor import TaskResult
658 row = TaskResult(
659 task=label,
660 ok=True,
661 returned=value,
662 state="shared",
663 started=time.perf_counter(),
664 )
665 if record is not None: 665 ↛ 675line 665 didn't jump to line 675 because the condition on line 665 was always true
666 # A shared answer is the record reused: the row a reader sees carries
667 # what the execution *reported* — a reviewed title, a rewritten
668 # reported value, the audit — never a fresher, less honest copy.
669 row.returned = record.returned
670 row.title = record.title
671 row.audit = list(record.audit)
672 # The share concluded after the execution it joined, and the tie-break
673 # must never say otherwise: the row's stamp is floored just above its
674 # record's, whatever number the request itself took.
675 floor = record.seq + 1 if record is not None and record.seq is not None else None
676 row.seq = (
677 max(seq, floor)
678 if seq is not None and floor is not None
679 else (seq if floor is None else floor)
680 )
681 return row
684def _claim_unit(label: str) -> Any:
685 """Start this request's unit on the live status line, or claim the one
686 already counted for it; `None` when no line is running.
688 A request is a unit — a scheduler node, a `parallel()` child, a body
689 call. The one thing that must never happen is counting a single piece of
690 work twice, and `parallel()` cannot tell from the outside whether the
691 thunk it was handed is a task call in disguise: a closure is opaque. So
692 it counts every child and hands the unit down (`Context.unit_pending`),
693 and the first request inside takes it over rather than adding its own.
694 Cleared on claim, so a second call in the same thunk — and everything
695 the callee goes on to ask for — counts honestly.
696 """
697 from footman import context
699 status = context.active_status()
700 if status is None:
701 return None
702 ctx = context.current()
703 if ctx.unit_pending:
704 ctx.unit_pending = False
705 return None # counted and displayed by whoever handed it down
706 status.unit_added(1)
707 status.unit_started(label)
708 return status
711def _refuse_wide_moment(task: Any) -> None:
712 """Refuse a task call from `pre_tasks` / `post_tasks`.
714 Both moments sit outside the run — `pre_tasks` at discovery, `post_tasks`
715 once the plan is finished — so there is no boundary to give a call: no
716 result row, no sharing, no availability gate, and a task declaring `ctx`
717 would take the first argument into that slot. Worse for `pre_tasks`, which
718 also runs in the child that rebuilds the completion manifest: a call there
719 executes the task on a <kbd>Tab</kbd> press.
721 A call from *outside footman entirely* (a REPL, an import of the tasks
722 module) is untouched — it is the plain function call it looks like.
723 """
724 from footman._executor import _wide_moment
726 moment = _wide_moment.get()
727 if moment is None:
728 return
729 name = _label(task)
730 when = (
731 "runs at discovery — including in the child that rebuilds the "
732 "completion manifest, where it would run on a Tab press"
733 if moment == "pre_tasks"
734 else "runs after the plan is finished, when the report is already built"
735 )
736 raise ChainError(
737 f"{name} cannot be called from @{moment}, which {when}. Edit the tree "
738 f"instead (inv.tasks — say pre=[{name}] on the tasks that need it), or "
739 f"move the call to a per-task moment (@pre_task / @post_task), which "
740 f"runs inside the run and gives the call a real task boundary."
741 )
744def _refuse_unrunnable(task: Any) -> None:
745 """Refuse the two body calls that cannot be given a task boundary.
747 A `serial=`/`exclusive=` task's lane is acquired at the boundary and never
748 mid-body — the invariant that keeps the arbiter deadlock-free — and an
749 `infinite` task is a future that never resolves.
750 """
751 name = _label(task)
752 if registry.task_lane(task) is not None:
753 raise ChainError(
754 f"{name} declares a serial/exclusive lane, so it cannot be called "
755 f"from a task body — its lane is taken at the task boundary, never "
756 f"mid-body. Declare it as pre=[{name}] instead."
757 )
758 if registry.is_infinite(task):
759 raise ChainError(
760 f"{name} is an infinite task, so calling it from a task body would "
761 f"never return. Run it as its own segment, or declare it as "
762 f"pre=[{name}]."
763 )
766def _run_now(
767 task: Any,
768 args: tuple[Any, ...],
769 kwargs: dict[str, Any],
770 *,
771 shared: bool = True,
772 seg: Any = None,
773 child: Any = None,
774 handle: Any = None,
775 cell: Any = None,
776 seq: int | None = None,
777 given: frozenset[str] = frozenset(),
778) -> Any:
779 """Run *task* here and now with full task semantics, and return its value.
781 A failure raises in the caller, exactly as a direct call always did — and
782 the callee's own `TaskResult` still joins the run's report, so the work is
783 visible even though the value came back through Python.
785 Availability was checked in `call()`, before any hook fired; `confirm=`
786 is the one gate that cannot be resolved up front the way the scheduler
787 resolves a segment's — a call is not knowable before the run — so it is
788 asked here, at the moment of execution (a request the run has already
789 answered never re-asks).
790 """
791 from footman import _executor, _globals, _schedule, context
793 parent = context.current()
794 label = _label(task)
795 if seg is None: 795 ↛ 796line 795 didn't jump to line 796 because the condition on line 795 was never true
796 seg = _schedule._default_seg(task)
797 if (denial := _schedule.confirm_gate(task, seg, parent)) is not None:
798 denial.seq = seq
799 _record(denial)
800 raise denial.error or ChainError(f"{label} was not confirmed")
801 if child is None:
802 buf = io.StringIO()
803 child = parent.child(
804 label,
805 fn=task,
806 cwd=None, # let the callee's own cwd policy resolve
807 sink=buf,
808 err_sink=buf,
809 )
810 else:
811 buf = child.sink
812 # Presence is the callee's own, never the caller's: the birth inherits
813 # every unlisted field, so a child born from a parent that was given
814 # `--agent` would otherwise inherit the claim that *it* was given one.
815 child.given = given
816 # Unsharedness propagates: what this callee asks for is asked the same
817 # way, unless that task declares its own answer.
818 child.shared = parent.shared and shared
819 # A body call is a request like a scheduler node or a parallel() child:
820 # a unit on the live status line, counted the moment it becomes real
821 # work (a confirm= denial above never was one).
822 status = _claim_unit(label)
823 try:
824 result = _executor.run_bound(
825 task, seg, child, list(args), dict(kwargs), as_call=True, handle=handle
826 )
827 except BaseException:
828 if status is not None:
829 status.unit_finished(label, False)
830 raise
831 if status is not None:
832 status.unit_finished(label, result.ok)
833 result.seq = seq
834 _record(result)
835 if cell is not None:
836 cell.record = result # the sealed row, for later requests to reuse
837 if text := buf.getvalue():
838 # The callee ran with its own buffer so its output stays one block.
839 # Where that block goes depends on the caller: a capturing parent
840 # (`--json`, a document run, a `parallel()` child) owns it and takes
841 # it; an uncaptured parent is streaming to the terminal, so it goes
842 # there — the same handoff `parallel()` makes for its children,
843 # status line warned first because this write bypasses the routers.
844 if parent.sink is not None: 844 ↛ 845line 844 didn't jump to line 845 because the condition on line 844 was never true
845 parent.sink.write(text)
846 else:
847 dest = context.real_stdout()
848 with _globals.console_gate():
849 if status is not None: 849 ↛ 850line 849 didn't jump to line 850 because the condition on line 849 was never true
850 status.notify(text)
851 dest.write(text)
852 dest.flush()
853 if result.error is not None:
854 # A failure raises in the caller, exactly as a direct call always did.
855 raise result.error
856 if not result.ok: # a bare `sys.exit(N)`, which carries no error object 856 ↛ 857line 856 didn't jump to line 857 because the condition on line 856 was never true
857 raise ChainError(f"{label} exited with code {result.code}")
858 # The body's own return, never the reported one: what a body caller
859 # receives was snapshotted at the body's exit, whatever a reviewer
860 # later did to the report.
861 return result.body_returned