@spec guard(String.t(), (-> result), (uncaught() -> result)) :: result
when result: var

Run a per-item body closure under the always-on link-safe catch.

catch kind, reason, never rescue: an exit is not an exception, and Task.async_stream links its children, so an uncaught exit kills the caller. on_uncaught builds the caller's own fallback pair from the tagged failure, because the two consumers — reduce/5 here and Dsl.Runtime.guarded_item/7 — degrade to different shapes. This is the single home of the catch: writing it a second time is how the two paths silently diverge on one kind (package CLAUDE.md §7).

@spec item_parent(parent_mode(), term(), Ecto.UUID.t() | nil) :: Ecto.UUID.t() | nil

The lineage parent for one item of a fan-out.

Under :source_stage, source_parent — every item's steps hang off the source stage's step log. Under :per_item, the item's OWN producing step log, which requires the item to be an %Item{} wrapper (ok_items/1 keeps it), and raises otherwise.

@spec reduce(
  ALLM.Pipeline.Context.t(),
  Enumerable.t(),
  acc(),
  (ALLM.Pipeline.Context.t(), term(), acc() -> {item_result(), acc()}),
  keyword()
) :: {[ALLM.Pipeline.Dsl.Item.t()], acc()}

Fold a body over a list of items, sequentially, with per-item failure isolation.

This is the FUNCTION form of a body:-mode fan_out: an ordinary stage body calls it instead of declaring the mode. It owns exactly the properties §8.3 of the DSL review found genuinely need centralising — the always-on per-item catch (failure isolation, not link safety — reduce/5 is sequential; see the bullet below), the %Item{} wrapping with its producing-step-log capture, and per-item lineage — and pushes section:/delay:/over: back to the caller as ordinary code.

Runs fun.(ctx, item, acc) for each item, threading acc sequentially (a fold), and returns the list of %Item{} in input order plus the final accumulator.

• Each fun call runs under an always-on catch kind, reason wrapper whose job here is per-item failure isolation: an uncaught raise/exit/throw becomes %Item{result: {:error, {:uncaught, kind, reason}}} and leaves acc unchanged for that item, so the fold degrades one item and keeps folding rather than aborting the run. This is reduce/5's own reason — a sequential fold has no link hazard (the moduledoc's ProjectRefreshPipeline sibling makes the same point) — and it also keeps reduce/5 behaviourally equivalent to the concurrent fan-out's always-on catch. It is catch, never rescue: an exit is not an exception, and Playwright/GenServer teardowns arrive as exits. (Link safety proper applies only to the Task.async_stream path in Dsl.Runtime.run_concurrent/7.)
• fun's item_result is wrapped into %Item{}: a 3-tuple {:ok, value, %StepLog{}} carries the producing step log onto %Item{step_log: …}; every other shape leaves it nil.
• reduce/5 is sequential only — it folds a changing accumulator, and a concurrent fold has undefined order (Dsl.Runtime's assert_no_concurrent_fold!/2 exists to reject that very shape). A concurrency-capable non-folding sibling is deliberately not provided (Phase 4.5 Alternatives).

Options

• parent: — :source_stage (default) hangs every item's steps off Context.input_step_id(ctx); :per_item re-parents each item's context to its OWN producing step log, which requires the items to be %Item{} wrappers (filter a previous fan-out with ALLM.Pipeline.Dsl.Item.ok_items/1), and raises otherwise.

section: and delay: are not options here — the caller inlines them (Executor.log_section/3 and Process.sleep/1). That split is the whole point: it keeps this function from re-accepting the costume Phase 4.5 removed.

@spec tag_uncaught(String.t(), atom(), term(), Exception.stacktrace()) :: uncaught()

Log an uncaught per-item failure with its stacktrace and return a tagged tuple.

Call from inside a catch kind, reason -> clause, where __STACKTRACE__ is available. These handlers exist for failures nobody anticipated, so the stacktrace is the whole value — the fact of the failure is already implied by the degraded result.

label should identify the item (a URL, an index, a record id), not the module — the message reads "<label> aborted with <kind> <reason>".

@spec to_item(term(), item_result()) :: ALLM.Pipeline.Dsl.Item.t()

Wrap a per-item body's item_result into an %Item{}.

A 3-tuple {:ok, value, %StepLog{}} moves its log onto step_log and keeps {:ok, value} as the result; every other shape leaves step_log nil (the two-element {:ok, value} form nominates no new parent — Phase 4 D2).