@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.

One entry per SCANNER, not per leaf

A source feeding several leaves — one crawl of one site whose rows land in two cells — is ONE unit of scheduled work. scan_jobs/1 lists it per-cell, which is right for a control panel (a human triggers a leaf, and each leaf declares its own bound), and wrong for a scheduler: two leaves declaring the same module would emit two entries and crawl the same site twice an hour, each poll marking both leaves and the second achieving nothing.

So entries are deduplicated by scanner. The "cell" carried is the first of its leaves alphabetically — any of them reaches the same scanner, and refresh/3 marks every leaf the poll reports regardless of which one was named.

Two leaves declaring DIFFERENT cadences for one scanner get one entry each, on the assumption that a host writing two different cadences meant them; the duplicate crawl is then declared rather than accidental.

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.