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

The cells a source feeds, derived from the graph.

A node declares poll MyCrawler; everything reading that node is an ordinary edge, so a source's outputs are its children. One declaration, read in one place, and nothing that can go stale.

It used to be the reverse: the module declared leaf_cells/1, each fed leaf declared scan, and verify_scan!/3 caught the two disagreeing. Every implementation in this repo and its host apps ignored the graph argument and returned a literal list, restating what the leaves had already said.

Returns [] for a source no node polls — not an error: a plan built from a subset of resources legitimately excludes some.

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

One sweep, not N jobs

The default emits ONE entry per distinct cadence — a sweep. The job polls every source of that cadence sequentially, in graph order, then drains once.

That is the shape worth defaulting to, because N independent cron entries cannot order themselves: they fire concurrently whatever order the list is in, so a source that needs another to have run first has no way to say so. Inside one job it does — depends_on on a source is a sequencing edge, and scanners/2 sorts by depth.

# a source that must be polled after :observations
reactive do
  id :verdicts
  leaf? true
  poll MyApp.Sources.Verdicts, every: "0 * * * *"
  depends_on [:observations]
end

per_cell: true opts back into one entry per source, for a host that wants independent jobs — separate queues, separate retry policies — and accepts that they are unordered. order: then applies to the emitted list, which is a reading convenience rather than a happens-before. 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 detail_by(map() | [map()] | term(), atom()) :: %{
  optional(String.t() | atom()) => number()
}

One detail: key across a sweep, summed per bucket — the breakdown behind detail_total/2, mirroring ReactiveDag.Drain.Report.by/2.

Source.detail_by(results, :tokens_in)
#=> %{"claude-haiku-4-5" => 900, "openai/gpt-5.6-luna" => 300}

A source reporting the key as a bare number lands under :unattributed rather than being dropped, so the parts always sum to detail_total/2. A gap in attribution is worth seeing; a breakdown that silently disagrees with its own total is not.

@spec detail_total(map() | [map()] | term(), atom()) :: number()

Sum one detail: key across a sweep's results — the scan-side counterpart to ReactiveDag.Drain.Report.total/2.

{:ok, results} = Source.poll_all(plan)
Source.detail_total(results, :tokens_in)
#=> 41_200

Why a scan needs its own roll-up

A drain returns a %Report{} and the library totals across its steps. A poll returns one result per source and there is no report, because a scan is not a cascade — sources are independent and there is no propagation to trace.

That left a real gap. A crawler that classifies each new document with a model spends on every poll, and NONE of it reached the drain log: scans and drains are separate phases by design, so a scan's spend has no step to attach to. It was invisible not because nobody recorded it but because nothing aggregated it.

What a scanner reports

Whatever it likes, under detail: — the same "the library never interprets it" rule the drain's step meta follows:

{:ok, %{changed: keys, detail: %{tokens_in: 900, llm_calls: 12}}}

A count may be flat or broken down per bucket, exactly as on a drain step:

detail: %{tokens_in: %{"claude-haiku-4-5" => 900}}

Both total here; detail_by/2 returns the breakdown. A source reporting no detail:, or one lacking the key, contributes nothing rather than raising — a sweep mixing LLM and plain crawlers still totals.

Accepts what poll_all/2 returns (%{module => result}), a list of results, or a single result, so a host can total one poll_cell/3 the same way.

The arithmetic is ReactiveDag.Rollup, shared with ReactiveDag.Drain.Report.total/2 — the containers differ because the phases do, but "what did this cost" is one fold, not two that must agree.

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

Ordering

Sources are polled SEQUENTIALLY, in cell order. A host whose sweep order carries meaning — observers before the nodes that read them — declares it:

Source.poll_all(plan, order: [:observations, :verdicts])

Anything unlisted follows, in cell order. Unlike crontab/3's order:, this one is a real happens-before: these polls run one after another in this process, so a later source genuinely sees what an earlier one wrote.

Telemetry

[:reactive_dag, :scan, :source_stop] fires as each source finishes, with %{source: module, result: result} — including the failures, since a source that could not run is a thing a host wants recorded.

Sources are polled one at a time, so this doubles as PROGRESS ACROSS a sweep: a sweep over eight sources can run for minutes, and without it the only observable moments are the beginning and the end of the whole run.

Progress WITHIN one source is progress/3 — a crawl of 700 documents is a single :source_stop, and :source_stop fires when it is already over.

@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 progress(non_neg_integer(), non_neg_integer() | nil, keyword()) :: :ok

Report progress from inside a poll/1 — [:reactive_dag, :scan, :progress].

Source.progress(fetched, total, cell: "meeting_docs")

A scanner is opaque to the library: it is handed options and returns a result, and everything between is the host's. So a crawl of 700 documents emits ONE :source_stop, and it fires when the crawl is already over. For anything a person is waiting on, that is the wrong end.

Call this as each unit of work completes — per document fetched, per page walked. done and total are whatever the scanner counts; total may be nil when it is not yet known (a crawl still discovering pages), and a consumer then has a count without a denominator, which is still better than silence.

Emit per unit, not per batch

One event per document, not per twenty-five. Batching pushes an arbitrary N into every scanner, each host picks a different one, and a crawl smaller than N reports nothing at all. Coalescing is the CONSUMER's job and it is the one that can do it properly — ReactiveDagDashboard already flushes drain steps on a 150ms timer, so 700 events become a handful of renders.

The cost of an unhandled :telemetry.execute/3 is a lookup on an ETS table; a host with no handler attached pays approximately nothing.

Options

• :cell — which cell is being polled. A sweep polls several sources, so a consumer showing per-row progress needs to know whose progress this is.
• :label — what the number counts, for a UI that says "34/721 documents" rather than "34/721".

@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 resolve_args(keyword()) :: keyword()

Resolve deferred values in a standing args: list by calling them.

A zero-arity function as an arg VALUE is evaluated at poll time rather than read as data:

poll MyApp.Crawler, args: [recent: true, year: &MyApp.Clock.year/0]

This exists because args: is DSL data, frozen when the module compiles. A bound that depends on the clock — "the current year and the one before it" — cannot be written as a literal there: it would be correct on the day of the build and progressively wrong after, silently, until the next deploy.

Only VALUES are resolved, and only when they are functions of arity 0. A list with no functions in it comes back unchanged, so this is free for the ordinary case, and a keyword list remains a keyword list — the type every call site and both introspection paths already expect.

Applied to the MERGED opts, so a caller passing year: &other_clock/0 is resolved exactly like a declared default. Precedence is Keyword.merge's alone: an explicit poll_all(plan, year: 2019) replaced the standing pair before this ran, so there is no deferred value left to overwrite it.

Where this does NOT happen

controls/1 and scan_jobs/1 report args: verbatim, deferred values and all. Describing the graph must not run the host's code: a dashboard rendering a scan control would be calling a clock (or worse, whatever else a host deferred) on every page render, and a function value there is a truthful answer to "what did this leaf declare".

So a host rendering an arg value should expect a function and show the fact, not the result. poll_all/2 and poll_cell/3 resolve; nothing else does.

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

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_poll!(module(), String.t()) :: :ok

Verify a poll Mod declaration on cell_id: the module must be a loadable ReactiveDag.Source.

That is the whole check now. It used to also assert the module's own leaf_cells/1 claimed this cell — a check that existed only because the pairing was written twice, and whose error message asked you to work out which copy was stale. The cells a source feeds are plan.parents[id], so there is one declaration and nothing to disagree with.