@spec cells_of(module(), graph()) :: [String.t()]

The cells source feeds in graph — the resolver behind verify!/2. Uses the module's leaf_cells/1 when exported, else the single-leaf leaf_cell/0 fallback (as [to_string(leaf_cell())]). Raises ArgumentError when the module exports neither.

@spec controls(graph()) :: %{required(String.t()) => map()}

What a host needs to render a scan control for each cell that has one: %{cell_id => %{source:, args:, every:, origin:}}.

The library describes; the host renders. A cell with no scanner is absent, and a scanner declaring no args:/every: reports them empty — so a leaf cheap enough to run whole gets a plain "refresh" and no misleading range picker, without the dashboard having to know which scanners are expensive.

origin: is the source's own origin/0 when it implements it, so a control can say where it is about to fetch from.

@spec crontab(graph(), module(), keyword()) :: [{String.t(), module(), keyword()}]

The Oban-style crontab entries a plan's leaves declare, as {cron, worker, args: %{"cell" => id}}.

The library never schedules anything. A leaf declaring every: states how often a routine poll should run; this collects those declarations into data the host hands to its own scheduler:

plugins: [
  {Oban.Plugins.Cron, crontab: ReactiveDag.Source.crontab(plan, MyApp.ScanWorker)}
]

Emitting data rather than inserting jobs keeps the library out of the host's supervision tree and out of its deploy story — and lets a host filter, rewrite or ignore the entries, which it could not do if they were already scheduled.

The worker receives %{"cell" => "agenda_docs"} and is expected to poll that one leaf. A leaf declaring no every: contributes nothing, which is the correct outcome for a scanner cheap enough to run on any cadence the host likes.

Adding your own arguments

A crontab entry is built once at config time, so anything it carries is fixed for every firing — a per-run id has to be minted when the job fires, inside the worker. What a host CAN do here is add its own standing arguments, with args::

Source.crontab(plan, MyApp.ScanWorker, args: %{"queue" => "crawls"})

merged UNDER the "cell" this computes, so a typo cannot silently retarget the job at a different leaf. A host that wants per-firing values wraps the worker rather than the crontab: mint the id in perform/1, then call refresh/3 with reason: "scan:#{run_id}" — :reason is a free string and rides through to the frontier rows, so the trace says which run dirtied a cell.

@spec drivers(graph(), [module()]) :: [module()]

The scanner drivers feeding a lowered graph: the inline ones declared with driver MyApp.Sources.FleetScan on a leaf's reactive block (read from each leaf cell's meta.driver), unioned with any extra fan-out drivers a host passes (drivers that name their cells via leaf_cells/1 because no single leaf owns them). This is the full scanner set — feed it to verify!/2, poll it in phase 1.

@spec poll_all(
  graph(),
  keyword()
) :: {:ok, map()} | {:error, [{module(), term()}]}

Poll every scanner the PLAN declares (via scan Mod on its leaves), in the poll phase before a drain.

Scanners are found from the graph rather than a list the host maintains alongside it — the list is the thing that drifts. A source feeding many leaves appears once, however many leaves declare it.

Returns {:ok, %{module => result}}, or {:error, failures} where failures are {module, reason}: one scanner failing must not silently cancel the others, and must not look like success.

@spec poll_cell(graph(), String.t(), keyword()) ::
  {:ok, map()} | {:error, term()} | {:error, :no_scanner}

Poll the scanner feeding ONE cell — the "re-run this scanner" affordance.

poll_all/2 is the routine sweep. This is what a host wires a button to: a dashboard has a cell in hand, not a source module, and a human asking to refresh is asking about this leaf, not about every scanner in the graph.

# routine, on the declared cadence
Source.poll_all(plan)

# a human pressed "refresh", accepting the cheap default
Source.poll_cell(plan, "agenda_docs")

# ...or asked for the deep pass
Source.poll_cell(plan, "agenda_docs", recent: false)

The leaf's declared args: apply exactly as they do in poll_all/2, with the caller's opts winning — so a button that passes nothing gets the cheap pass, and one that passes recent: false gets the expensive one.

Returns {:ok, result}, {:error, reason} if the poll failed, or {:error, :no_scanner} when the cell declares none — which a host should render as "no refresh available" rather than as a failure.

Note a source feeding several leaves is polled whole: poll/1 takes options, not a cell, so asking for one leaf runs whatever that scanner does. The scanner narrows itself through args: if that matters.

@spec refresh(graph(), String.t(), keyword()) ::
  {:ok, %{changed: [String.t()], marked: map(), unreachable: list()}}
  | {:error, term()}

Poll a cell's scanner AND mark what changed — the whole poll half of the two-phase loop, in one call.

poll_cell/3 returns what the scanner said and stops, leaving every host to hand-write the same three steps: normalise the return shape, mark the frontier, propagate to parents. That loop is documented in the guide and was provided nowhere, which is why every host grew a worker to hold it.

{:ok, result} = Source.refresh(plan, "agenda_docs")
#=> %{changed: ["a", "b"], marked: %{"agenda_docs" => ["a", "b"]},
#     unreachable: []}

Then drain — separately, and deliberately so. The poll/drain split is a design invariant (external I/O must not sit inside a depth-ordered recompute), and a host polling several sources usually wants ONE drain after all of them rather than one each.

Return shapes

A single-leaf source returns a flat key list; a fan-out source returns %{leaf_id => keys}. Both are in the Source contract, so both are normalised here rather than in each host.

:reason labels the frontier rows (default "scan"), which is what makes a drain's trace say why a cell was dirty.

Nothing is marked for an unreachable upstream, because the scanner reported no keys for it — an outage propagates nothing, which is the honest gap holding by construction rather than by rule.

@spec scan_jobs(graph()) :: [
  %{
    cell: String.t(),
    source: module(),
    args: keyword(),
    every: String.t() | nil
  }
]

Every scannable unit of work in the plan: %{cell:, source:, args:, every:}.

One entry per cell that declares a scan, whether or not it declares a cadence. This is the list a host schedules from, renders controls from, and triggers ad-hoc runs from — crontab/2 is a projection of the subset that declared every:.

Source.scan_jobs(plan)
#=> [%{cell: "agenda_docs", source: MyApp.Crawler,
#      args: [recent: true], every: "0 * * * *"}]

Keyed by CELL rather than by source, because a source feeding several leaves is several units of work — each with its own declared bound.

@spec scanners(graph()) :: [module()]

The distinct scanner modules a plan's leaves declare via scan.

@spec standing_args(graph()) :: %{required(module()) => keyword()}

The standing args: each scanner's leaf declared, as %{module => keyword}.

poll_all/2 merges these under the caller's opts. Exposed because a host driving one scanner directly wants the same default rather than a second copy of it.

@spec verify!([module()], graph()) :: :ok

Verify every source's declared leaves resolve to real cells in graph — the authoritative scanner↔leaf check. Each driver's leaves are resolved via cells_of/2 (leaf_cells/1, or the single-leaf leaf_cell/0 fallback); this confirms every one is a real cell in the built plan. Needs the lowered graph (a host may expand generator leaves from live data), so it runs at assembly/boot time, not compile time.

Returns :ok, or raises ArgumentError naming every {source, dangling_leaf}.

@spec verify_cells!([{module(), [String.t()]}], graph()) :: :ok

The same dangling-leaf check as verify!/2, but over already-resolved {source, [cell_id]} pairs instead of resolving each module via cells_of/2. Use this when a host resolves fed cells itself (its own conventions beyond leaf_cells/1 / leaf_cell/0). Returns :ok, or raises ArgumentError naming every {source, dangling_leaf}.

@spec verify_scan!(module(), String.t(), graph()) :: :ok

Verify a scan Mod declaration on the leaf cell_id: the module must be a loadable ReactiveDag.Source, and its own leaf_cells/1 must claim this leaf.

The second half matters more than it looks. A scanner already knows which cells it feeds; scan states the same fact from the other side. Two statements of one fact can disagree, so this is the check that they don't — a scanner refactored to feed "agenda_docs_v2" while a resource still declares scan fails at assembly rather than polling into a cell nobody reads.

Called by ReactiveDag.Node.graph/2 for every cell carrying a scan, so a host declaring scanners in the DSL needs no verify!/2 call of its own.