Runner API

The runner drives the crawl loop: it discovers a plugin's roots, expands refs through the expander chain, and passes each leaf to the sink.

Use run_plugin() or async_run_plugin() for a complete plugin run. They call Source.discover() once and return a PluginRunResult containing the individual RunResult values and their aggregate counts. Use run_crawl() or async_run_crawl() only when the caller deliberately owns root discovery or needs to process one known root.

All runners use the same RunConfig. Its leaf_limit is a per-root cap for whole-plugin runs. If a later root raises a globally fatal error, earlier roots may already have invoked on_leaf; persistence callbacks must therefore be idempotent when retrying run_plugin() or async_run_plugin().

run_plugin

Discover and run every top-level ref exposed by a sync plugin.

This is the whole-plugin counterpart to :func:run_crawl. It calls plugin.source.discover(client) exactly once, then processes discovered refs in source order by delegating to :func:run_crawl.

Processing roots sequentially deliberately avoids introducing a second, undocumented concurrency layer. Use :func:async_run_plugin for async adapters; it preserves this root ordering while each root's leaves retain the configured async_concurrency.

Parameters:

Name	Type	Description	Default
`plugin`	`CrawlPlugin[Any, Any, LeafRecordT]`	Crawl plugin providing a source, expanders, and sink.	required
`client`	`SyncHttpClientProtocol`	Configured synchronous HTTP client protocol implementation.	required
`config`	`RunConfig`	Run configuration forwarded to each discovered root.	required
`on_leaf`	`OnLeafCallback[LeafRecordT, object] \| None`	Optional callback forwarded to each :func:`run_crawl` call.	`None`

Returns:

Type	Description
`PluginRunResult`	A PluginRunResult containing source-order per-root outcomes and totals.

Raises:

Type	Description
`ExpansionNotReadyError`	Propagates unchanged from a per-root :func:`run_crawl` invocation.
`PartialExpansionError`	Propagates unchanged from a per-root :func:`run_crawl` invocation.
`ChildListUnavailableError`	Propagates unchanged from a per-root :func:`run_crawl` invocation. Per that function's own `Raises:` contract, globally fatal expansion errors are never converted into a partial aggregate. Roots completed before a later fatal error may already have invoked `on_leaf`; callbacks must therefore be idempotent when the caller retries the whole plugin.
`AssetDownloadError`	A Sink or Expander raised this explicitly fatal plugin error while processing a discovered root.
`Exception`	Propagates ordinary exceptions from `source.discover` and :func:`run_crawl`.
`BaseException`	KeyboardInterrupt and other fatal errors from a per-root :func:`run_crawl` invocation propagate unchanged.

Source code in src/ladon/runner.py

def run_plugin(
    plugin: CrawlPlugin[Any, Any, LeafRecordT],
    client: SyncHttpClientProtocol,
    config: RunConfig,
    on_leaf: OnLeafCallback[LeafRecordT, object] | None = None,
) -> PluginRunResult:
    """Discover and run every top-level ref exposed by a sync plugin.

    This is the whole-plugin counterpart to :func:`run_crawl`. It calls
    ``plugin.source.discover(client)`` exactly once, then processes discovered
    refs in source order by delegating to :func:`run_crawl`.

    Processing roots sequentially deliberately avoids introducing a second,
    undocumented concurrency layer. Use :func:`async_run_plugin` for async
    adapters; it preserves this root ordering while each root's leaves retain
    the configured ``async_concurrency``.

    Args:
        plugin:  Crawl plugin providing a source, expanders, and sink.
        client:  Configured synchronous HTTP client protocol implementation.
        config:  Run configuration forwarded to each discovered root.
        on_leaf: Optional callback forwarded to each :func:`run_crawl` call.

    Returns:
        A PluginRunResult containing source-order per-root outcomes and totals.

    Raises:
        ExpansionNotReadyError: Propagates unchanged from a per-root
            :func:`run_crawl` invocation.
        PartialExpansionError: Propagates unchanged from a per-root
            :func:`run_crawl` invocation.
        ChildListUnavailableError: Propagates unchanged from a per-root
            :func:`run_crawl` invocation. Per that function's own ``Raises:``
            contract, globally fatal expansion errors are never converted into
            a partial aggregate. Roots completed before a later fatal error may
            already have invoked ``on_leaf``; callbacks must therefore be
            idempotent when the caller retries the whole plugin.
        AssetDownloadError: A Sink or Expander raised this explicitly fatal
            plugin error while processing a discovered root.
        Exception: Propagates ordinary exceptions from ``source.discover`` and
            :func:`run_crawl`.
        BaseException: KeyboardInterrupt and other fatal errors from a per-root
            :func:`run_crawl` invocation propagate unchanged.
    """

    top_refs = tuple(plugin.source.discover(client))
    results = tuple(
        run_crawl(top_ref, plugin, client, config, on_leaf=on_leaf)
        for top_ref in top_refs
    )
    return PluginRunResult.from_runs(top_refs, results)

async_run_plugin

Discover and run every top-level ref exposed by an async plugin.

This is the whole-plugin counterpart to :func:async_run_crawl. It awaits plugin.source.discover(client) exactly once, then processes roots in source order. Each root retains the existing bounded concurrent leaf processing from :func:async_run_crawl; roots are intentionally not run concurrently so the public API has one clear concurrency boundary.

Discovery and globally-fatal per-root errors propagate unchanged rather than being converted into a partial aggregate. Earlier roots may already have invoked on_leaf when a later root aborts, so callbacks must be idempotent if the caller retries the plugin.

Raises:

Type	Description
`ExpansionNotReadyError`	Propagates unchanged from a per-root :func:`async_run_crawl` invocation.
`PartialExpansionError`	Propagates unchanged from a per-root :func:`async_run_crawl` invocation.
`ChildListUnavailableError`	Propagates unchanged from a per-root :func:`async_run_crawl` invocation, per that function's own `Raises:` contract.
`AssetDownloadError`	A Sink or Expander raised this explicitly fatal plugin error while processing a discovered root.
`BaseException`	Cancellation and other fatal errors from a per-root :func:`async_run_crawl` invocation propagate unchanged.

Source code in src/ladon/async_runner.py

async def async_run_plugin(
    plugin: AsyncCrawlPlugin[Any, Any, AsyncLeafRecordT],
    client: AsyncHttpClientProtocol,
    config: RunConfig,
    on_leaf: AsyncOnLeafCallback[AsyncLeafRecordT, object] | None = None,
) -> PluginRunResult:
    """Discover and run every top-level ref exposed by an async plugin.

    This is the whole-plugin counterpart to :func:`async_run_crawl`. It
    awaits ``plugin.source.discover(client)`` exactly once, then processes
    roots in source order. Each root retains the existing bounded concurrent
    leaf processing from :func:`async_run_crawl`; roots are intentionally not
    run concurrently so the public API has one clear concurrency boundary.

    Discovery and globally-fatal per-root errors propagate unchanged rather
    than being converted into a partial aggregate. Earlier roots may already
    have invoked ``on_leaf`` when a later root aborts, so callbacks must be
    idempotent if the caller retries the plugin.

    Raises:
        ExpansionNotReadyError: Propagates unchanged from a per-root
            :func:`async_run_crawl` invocation.
        PartialExpansionError: Propagates unchanged from a per-root
            :func:`async_run_crawl` invocation.
        ChildListUnavailableError: Propagates unchanged from a per-root
            :func:`async_run_crawl` invocation, per that function's own
            ``Raises:`` contract.
        AssetDownloadError: A Sink or Expander raised this explicitly fatal
            plugin error while processing a discovered root.
        BaseException: Cancellation and other fatal errors from a per-root
            :func:`async_run_crawl` invocation propagate unchanged.
    """

    top_refs = tuple(await plugin.source.discover(client))
    results: list[RunResult] = []
    for top_ref in top_refs:
        results.append(
            await async_run_crawl(
                top_ref, plugin, client, config, on_leaf=on_leaf
            )
        )
    return PluginRunResult.from_runs(top_refs, tuple(results))

run_crawl

Run a single top-level ref through the plugin adapter stack.

Parameters:

Name	Type	Description	Default
`top_ref`	`object`	Reference to the resource to expand.	required
`plugin`	`CrawlPlugin[Any, Any, LeafRecordT]`	Crawl plugin providing source, expanders, and sink.	required
`client`	`SyncHttpClientProtocol`	Configured synchronous HTTP client protocol implementation.	required
`config`	`RunConfig`	Run-level configuration (limits, flags).	required
`on_leaf`	`OnLeafCallback[LeafRecordT, object] \| None`	Optional callback invoked after each successful leaf consume. Use this hook for DB writes, serialization, etc. Receives (leaf_record, parent_record).	`None`

Returns:

Type	Description
`RunResult`	RunResult with counts and any per-leaf error messages.

Raises:

Type	Description
`ExpansionNotReadyError`	Raised from any expander or the Sink. The ref (or an intermediate ref) is not ready. Caller should record the event and move on; retry on the next scheduled run.
`PartialExpansionError`	Raised from the first expander or Sink. From non-first expanders the failing branch is isolated and recorded in RunResult.errors instead.
`ChildListUnavailableError`	Raised from the first expander or Sink. Same isolation rule applies to non-first expanders as for PartialExpansionError.
`AssetDownloadError`	Raised from the Sink or an Expander. This explicitly fatal plugin error propagates.
`BaseException`	KeyboardInterrupt and other fatal errors propagate unchanged.
`ValueError`	Plugin has no expanders configured.

Source code in src/ladon/runner.py

def run_crawl(
    top_ref: object,
    plugin: CrawlPlugin[Any, Any, LeafRecordT],
    client: SyncHttpClientProtocol,
    config: RunConfig,
    on_leaf: OnLeafCallback[LeafRecordT, object] | None = None,
) -> RunResult:
    """Run a single top-level ref through the plugin adapter stack.

    Args:
        top_ref:  Reference to the resource to expand.
        plugin:   Crawl plugin providing source, expanders, and sink.
        client:   Configured synchronous HTTP client protocol implementation.
        config:   Run-level configuration (limits, flags).
        on_leaf:  Optional callback invoked after each successful leaf
                  consume. Use this hook for DB writes, serialization,
                  etc. Receives (leaf_record, parent_record).

    Returns:
        RunResult with counts and any per-leaf error messages.

    Raises:
        ExpansionNotReadyError:     Raised from any expander or the Sink. The
                                    ref (or an intermediate ref) is not ready.
                                    Caller should record the event and
                                    move on; retry on the next scheduled run.
        PartialExpansionError:      Raised from the first expander or Sink.
                                    From non-first expanders the failing
                                    branch is isolated and recorded in
                                    RunResult.errors instead.
        ChildListUnavailableError:  Raised from the first expander or Sink.
                                    Same isolation rule applies to non-first
                                    expanders as for PartialExpansionError.
        AssetDownloadError:         Raised from the Sink or an Expander. This
                                    explicitly fatal plugin error propagates.
        BaseException:              KeyboardInterrupt and other fatal errors
                                    propagate unchanged.
        ValueError:                 Plugin has no expanders configured.
    """
    if not plugin.expanders:
        raise ValueError(
            f"CrawlPlugin '{plugin.name}' has no expanders configured"
        )

    logger.info(
        "run_crawl started",
        extra={"plugin": plugin.name, "ref": str(top_ref)},
    )

    errors: list[str] = []

    # Phase 1 — traverse all expanders in order.
    #
    # The first expander handles top_ref and yields the top-level record
    # (e.g. AuctionRecord) stored in RunResult.record. Remaining expanders
    # chain through the refs produced by the previous level, carrying
    # (child_ref, parent_record) pairs so each leaf knows its direct parent.
    #
    # Single-expander behaviour is identical to the previous implementation.
    #
    # For non-first expanders, exceptions are isolated per branch:
    #   - ExpansionNotReadyError  → re-raised (run is globally premature)
    #   - PartialExpansionError   → branch skipped, error accumulated
    #   - ChildListUnavailableError → branch skipped, error accumulated
    first_expansion = plugin.expanders[0].expand(top_ref, client)
    top_record: object = first_expansion.record
    pairs: list[tuple[object, object]] = [
        (child_ref, first_expansion.record)
        for child_ref in first_expansion.child_refs
    ]

    for expander in plugin.expanders[1:]:
        next_pairs: list[tuple[object, object]] = []
        for ref, _ in pairs:
            try:
                expansion = expander.expand(ref, client)
            except ExpansionNotReadyError:
                raise  # run is globally premature — abort
            except (PartialExpansionError, ChildListUnavailableError) as exc:
                errors.append(f"expander branch '{ref}': {exc}")
                logger.warning(
                    "expander branch failed",
                    extra={
                        "plugin": plugin.name,
                        "ref": str(ref),
                        "error": str(exc),
                        "error_type": type(exc).__name__,
                    },
                )
                continue
            for child_ref in expansion.child_refs:
                next_pairs.append((child_ref, expansion.record))
        pairs = next_pairs

    # Phase 2 — apply leaf limit at the leaf level.
    if config.leaf_limit > 0:
        pairs = pairs[: config.leaf_limit]

    # Phase 3 — sink consumes each leaf ref.
    leaves_consumed = 0
    leaves_persisted = 0
    leaves_failed = 0

    for i, (leaf_ref, parent_record) in enumerate(pairs):
        # Bounded repr: large records (e.g. stories with many comment IDs)
        # can produce kilobyte-long repr strings; truncate for log readability.
        _parent_repr = repr(parent_record)
        if len(_parent_repr) > 120:
            _parent_repr = _parent_repr[:117] + "..."

        try:
            leaf_record = plugin.sink.consume(leaf_ref, client)
        except LeafUnavailableError as exc:
            leaves_failed += 1
            errors.append(f"ref[{i}] consume failed: {exc}")
            logger.warning(
                "leaf unavailable — ref[%d] parent=%s error=%s",
                i,
                _parent_repr,
                exc,
                extra={
                    "plugin": plugin.name,
                    "ref_index": i,
                    "error": str(exc),
                },
            )
            continue
        except FATAL_PLUGIN_ERRORS as exc:
            log_leaf_exception(exc, i, plugin.name)
            raise
        except Exception as exc:
            leaves_failed += 1
            errors.append(f"ref[{i}] consume failed: {exc}")
            log_leaf_exception(exc, i, plugin.name)
            continue
        except BaseException as exc:
            log_leaf_exception(exc, i, plugin.name)
            raise

        leaves_consumed += 1

        if on_leaf is not None:
            try:
                on_leaf(leaf_record, parent_record)
                leaves_persisted += 1
            except Exception as exc:
                errors.append(f"ref[{i}] callback failed: {exc}")
                logger.warning(
                    "on_leaf callback failed — ref[%d] parent=%s error=%s",
                    i,
                    _parent_repr,
                    exc,
                    extra={
                        "plugin": plugin.name,
                        "ref_index": i,
                        "error": str(exc),
                    },
                )
            except BaseException as exc:
                log_leaf_exception(exc, i, plugin.name)
                raise
        else:
            leaves_persisted += 1

    logger.info(
        "run_crawl finished",
        extra={
            "plugin": plugin.name,
            "leaves_consumed": leaves_consumed,
            "leaves_persisted": leaves_persisted,
            "leaves_failed": leaves_failed,
        },
    )

    return RunResult(
        record=top_record,
        leaves_consumed=leaves_consumed,
        leaves_persisted=leaves_persisted,
        leaves_failed=leaves_failed,
        errors=tuple(errors),
    )

async_run_crawl

Run a single top-level ref through an async plugin adapter stack.

Parameters:

Name	Type	Description	Default
`top_ref`	`object`	Reference to the resource to expand.	required
`plugin`	`AsyncCrawlPlugin[Any, Any, AsyncLeafRecordT]`	Async crawl plugin providing expanders and sink.	required
`client`	`AsyncHttpClientProtocol`	Configured asynchronous HTTP client protocol implementation.	required
`config`	`RunConfig`	Run-level configuration (limits, concurrency).	required
`on_leaf`	`AsyncOnLeafCallback[AsyncLeafRecordT, object] \| None`	Optional async callback invoked after each successful leaf consume. Receives (leaf_record, parent_record).	`None`

Returns:

Type	Description
`RunResult`	RunResult with counts and any per-leaf error messages.

Raises:

Type	Description
`ExpansionNotReadyError`	Any expander or the Sink raised this. The ref is not yet ready; retry on the next run.
`PartialExpansionError`	Raised from the first expander or Sink.
`ChildListUnavailableError`	Raised from the first expander or Sink.
`AssetDownloadError`	Raised from the Sink or an Expander. This explicitly fatal plugin error propagates.
`CancelledError`	A leaf consume or `on_leaf` callback was cancelled.
`BaseException`	Other fatal BaseException subclasses raised by a leaf task propagate unchanged.
`ValueError`	Plugin has no expanders configured.

Source code in src/ladon/async_runner.py

async def async_run_crawl(
    top_ref: object,
    plugin: AsyncCrawlPlugin[Any, Any, AsyncLeafRecordT],
    client: AsyncHttpClientProtocol,
    config: RunConfig,
    on_leaf: AsyncOnLeafCallback[AsyncLeafRecordT, object] | None = None,
) -> RunResult:
    """Run a single top-level ref through an async plugin adapter stack.

    Args:
        top_ref:  Reference to the resource to expand.
        plugin:   Async crawl plugin providing expanders and sink.
        client:   Configured asynchronous HTTP client protocol implementation.
        config:   Run-level configuration (limits, concurrency).
        on_leaf:  Optional async callback invoked after each successful leaf
                  consume. Receives (leaf_record, parent_record).

    Returns:
        RunResult with counts and any per-leaf error messages.

    Raises:
        ExpansionNotReadyError:     Any expander or the Sink raised this. The
                                    ref is not yet ready; retry on the next run.
        PartialExpansionError:      Raised from the first expander or Sink.
        ChildListUnavailableError:  Raised from the first expander or Sink.
        AssetDownloadError:         Raised from the Sink or an Expander. This
                                    explicitly fatal plugin error propagates.
        asyncio.CancelledError:     A leaf consume or ``on_leaf`` callback was
                                    cancelled.
        BaseException:              Other fatal BaseException subclasses raised
                                    by a leaf task propagate unchanged.
        ValueError:                 Plugin has no expanders configured.
    """
    if not plugin.expanders:
        raise ValueError(
            f"AsyncCrawlPlugin '{plugin.name}' has no expanders configured"
        )

    logger.info(
        "async_run_crawl started",
        extra={"plugin": plugin.name, "ref": str(top_ref)},
    )

    errors: list[str] = []

    # Phase 1 — sequential await through all expanders.
    #
    # Identical isolation rules to the sync runner:
    #   - ExpansionNotReadyError  → re-raised (run is globally premature)
    #   - PartialExpansionError   → branch skipped, error accumulated (non-first only)
    #   - ChildListUnavailableError → branch skipped, error accumulated (non-first only)
    first_expansion = await plugin.expanders[0].expand(top_ref, client)
    top_record: object = first_expansion.record
    pairs: list[tuple[object, object]] = [
        (child_ref, first_expansion.record)
        for child_ref in first_expansion.child_refs
    ]

    for expander in plugin.expanders[1:]:
        next_pairs: list[tuple[object, object]] = []
        for ref, _ in pairs:
            try:
                expansion = await expander.expand(ref, client)
            except ExpansionNotReadyError:
                raise
            except (PartialExpansionError, ChildListUnavailableError) as exc:
                errors.append(f"expander branch '{ref}': {exc}")
                logger.warning(
                    "expander branch failed",
                    extra={
                        "plugin": plugin.name,
                        "ref": str(ref),
                        "error": str(exc),
                        "error_type": type(exc).__name__,
                    },
                )
                continue
            for child_ref in expansion.child_refs:
                next_pairs.append((child_ref, expansion.record))
        pairs = next_pairs

    # Phase 2 — apply leaf limit.
    if config.leaf_limit > 0:
        pairs = pairs[: config.leaf_limit]

    # Phase 3 — concurrent sink calls bounded by Semaphore.
    semaphore = asyncio.Semaphore(config.async_concurrency)

    async def _process_leaf(
        i: int, leaf_ref: object, parent_record: object
    ) -> tuple[bool, bool, list[str]] | BaseException:
        """Returns (consumed, persisted, leaf_errors).

        consumed=True  when sink.consume() succeeded.
        persisted=True when consumed AND on_leaf succeeded (or no callback).
        leaf_errors    holds at most one error string.
        """
        async with semaphore:
            _parent_repr = repr(parent_record)
            if len(_parent_repr) > 120:
                _parent_repr = _parent_repr[:117] + "..."

            try:
                leaf_record = await plugin.sink.consume(leaf_ref, client)
            except LeafUnavailableError as exc:
                logger.warning(
                    "leaf unavailable — ref[%d] parent=%s error=%s",
                    i,
                    _parent_repr,
                    exc,
                    extra={
                        "plugin": plugin.name,
                        "ref_index": i,
                        "error": str(exc),
                    },
                )
                return (False, False, [f"ref[{i}] consume failed: {exc}"])
            except FATAL_PLUGIN_ERRORS:
                raise
            except Exception as exc:
                log_leaf_exception(exc, i, plugin.name)
                return (False, False, [f"ref[{i}] consume failed: {exc}"])
            except BaseException as exc:
                # Any non-Exception BaseException that escapes a Task can be
                # mangled or crash the event loop through CPython's Task
                # machinery. Return it as a value so the caller receives the
                # original exception unchanged.
                return exc

            if on_leaf is not None:
                try:
                    await on_leaf(leaf_record, parent_record)
                    return (True, True, [])
                except Exception as exc:
                    logger.warning(
                        "on_leaf callback failed — ref[%d] parent=%s error=%s",
                        i,
                        _parent_repr,
                        exc,
                        extra={
                            "plugin": plugin.name,
                            "ref_index": i,
                            "error": str(exc),
                        },
                    )
                    return (True, False, [f"ref[{i}] callback failed: {exc}"])
                except BaseException as exc:
                    # Any non-Exception BaseException that escapes a Task can be
                    # mangled or crash the event loop through CPython's Task
                    # machinery. Return it as a value so the caller receives the
                    # original exception unchanged.
                    return exc

            return (True, True, [])

    outcomes = await asyncio.gather(
        *[
            _process_leaf(i, leaf_ref, parent)
            for i, (leaf_ref, parent) in enumerate(pairs)
        ],
        return_exceptions=True,
    )

    leaves_consumed = 0
    leaves_persisted = 0
    leaves_failed = 0

    leaf_outcomes = _raise_first_fatal_outcome(
        outcomes, plugin_name=plugin.name
    )

    for consumed, persisted, leaf_errors in leaf_outcomes:
        if consumed:
            leaves_consumed += 1
        else:
            leaves_failed += 1
        if persisted:
            leaves_persisted += 1
        errors.extend(leaf_errors)

    logger.info(
        "async_run_crawl finished",
        extra={
            "plugin": plugin.name,
            "leaves_consumed": leaves_consumed,
            "leaves_persisted": leaves_persisted,
            "leaves_failed": leaves_failed,
        },
    )

    return RunResult(
        record=top_record,
        leaves_consumed=leaves_consumed,
        leaves_persisted=leaves_persisted,
        leaves_failed=leaves_failed,
        errors=tuple(errors),
    )

RunConfig

Configuration for one single-root runner invocation.

leaf_limit caps the number of leaves processed by one root; 0 means no limit. run_plugin() applies the same cap independently to every root discovered by its source. async_concurrency bounds the number of concurrent leaf-processing slots in async_run_crawl() and execute_plan() — each slot covers the full sink.consume() + callback pair, so slow on_leaf or on_planned_leaf callbacks reduce effective fetch concurrency. Ignored by the synchronous runners.

Source code in src/ladon/runner.py

@dataclass(frozen=True)
class RunConfig:
    """Configuration for one single-root runner invocation.

    ``leaf_limit`` caps the number of leaves processed by one root; 0 means
    no limit. ``run_plugin()`` applies the same cap independently to every
    root discovered by its source.
    ``async_concurrency`` bounds the number of concurrent leaf-processing
    slots in ``async_run_crawl()`` and ``execute_plan()`` — each slot covers
    the full ``sink.consume()`` + callback pair, so slow ``on_leaf`` or
    ``on_planned_leaf`` callbacks reduce effective fetch concurrency. Ignored
    by the synchronous runners.
    """

    leaf_limit: int = 0
    async_concurrency: int = 10

    def __post_init__(self) -> None:
        if self.async_concurrency < 1:
            raise ValueError(
                f"async_concurrency must be >= 1, got {self.async_concurrency}"
            )

RunResult

Outcome of a crawl run — returned by run_crawl(), execute_plan_sync(), and execute_plan().

leaves_consumed counts leaves for which sink.consume() succeeded, regardless of whether the applicable on_leaf or on_planned_leaf callback also succeeded.

leaves_persisted counts leaves for which the full pipeline succeeded: sink.consume() completed and the applicable callback completed without raising. When no callback is supplied, leaves_persisted equals leaves_consumed (the pipeline trivially succeeds after consume).

leaves_failed counts leaves for which sink.consume() raised a non-fatal exception. Callback failures are NOT included here — derive them from leaves_consumed - leaves_persisted.

The following invariant always holds::

leaves_consumed + leaves_failed == total leaves passed to Phase 3
                                  (after leaf_limit is applied)

errors accumulates expander branch failures (Phase 1, format "expander branch '...': ...") and leaf-level failures (Phase 3, formats "ref[N] consume failed: ..." and "ref[N] callback failed: ..."). A result with leaves_failed == 0 may still contain branch or callback errors — always inspect errors for a complete picture of what went wrong.

Source code in src/ladon/runner.py

@dataclass(frozen=True)
class RunResult:
    """Outcome of a crawl run — returned by run_crawl(), execute_plan_sync(), and execute_plan().

    ``leaves_consumed`` counts leaves for which ``sink.consume()`` succeeded,
    regardless of whether the applicable ``on_leaf`` or ``on_planned_leaf``
    callback also succeeded.

    ``leaves_persisted`` counts leaves for which the full pipeline succeeded:
    ``sink.consume()`` completed *and* the applicable callback completed
    without raising. When no callback is supplied, ``leaves_persisted`` equals
    ``leaves_consumed`` (the pipeline trivially succeeds after consume).

    ``leaves_failed`` counts leaves for which ``sink.consume()`` raised a
    non-fatal exception. Callback failures are NOT included here — derive them
    from ``leaves_consumed - leaves_persisted``.

    The following invariant always holds::

        leaves_consumed + leaves_failed == total leaves passed to Phase 3
                                          (after leaf_limit is applied)

    ``errors`` accumulates expander branch failures (Phase 1, format
    ``"expander branch '...': ..."``) and leaf-level failures (Phase 3,
    formats ``"ref[N] consume failed: ..."`` and
    ``"ref[N] callback failed: ..."``). A result with ``leaves_failed == 0``
    may still contain branch or callback errors — always inspect ``errors``
    for a complete picture of what went wrong.
    """

    record: object
    leaves_consumed: int
    leaves_persisted: int
    leaves_failed: int
    errors: tuple[str, ...]

PluginRunResult

Aggregate outcome of running every top-level ref from a plugin Source.

top_refs and results have the same order and length: result i is the outcome for discovered ref i. errors flattens each per-root result's errors with its stable source-order index so callers can consume a single summary without losing the individual RunResult values.

RunConfig.leaf_limit applies independently to each top-level run, because :func:run_plugin delegates to :func:run_crawl once per discovered ref. A discovery failure or an exception that is globally fatal for a root keeps the existing runner semantics and propagates instead of returning a partial aggregate. Earlier roots may already have invoked on_leaf when a later root aborts, so callbacks passed to :func:run_plugin must be idempotent if the caller retries the plugin.

Source code in src/ladon/runner.py

@dataclass(frozen=True)
class PluginRunResult:
    """Aggregate outcome of running every top-level ref from a plugin Source.

    ``top_refs`` and ``results`` have the same order and length: result ``i``
    is the outcome for discovered ref ``i``. ``errors`` flattens each
    per-root result's errors with its stable source-order index so callers can
    consume a single summary without losing the individual ``RunResult``
    values.

    ``RunConfig.leaf_limit`` applies independently to each top-level run,
    because :func:`run_plugin` delegates to :func:`run_crawl` once per
    discovered ref. A discovery failure or an exception that is globally fatal
    for a root keeps the existing runner semantics and propagates instead of
    returning a partial aggregate. Earlier roots may already have invoked
    ``on_leaf`` when a later root aborts, so callbacks passed to
    :func:`run_plugin` must be idempotent if the caller retries the plugin.
    """

    top_refs: tuple[object, ...]
    results: tuple[RunResult, ...]
    leaves_consumed: int
    leaves_persisted: int
    leaves_failed: int
    errors: tuple[str, ...]

    @classmethod
    def from_runs(
        cls, top_refs: tuple[object, ...], results: tuple[RunResult, ...]
    ) -> PluginRunResult:
        """Build a stable aggregate from source-order per-root outcomes."""

        return cls(
            top_refs=top_refs,
            results=results,
            leaves_consumed=sum(result.leaves_consumed for result in results),
            leaves_persisted=sum(result.leaves_persisted for result in results),
            leaves_failed=sum(result.leaves_failed for result in results),
            errors=tuple(
                f"top_ref[{index}]: {error}"
                for index, result in enumerate(results)
                for error in result.errors
            ),
        )

`from_runs(top_refs, results)` `classmethod`

Build a stable aggregate from source-order per-root outcomes.

Source code in src/ladon/runner.py

@classmethod
def from_runs(
    cls, top_refs: tuple[object, ...], results: tuple[RunResult, ...]
) -> PluginRunResult:
    """Build a stable aggregate from source-order per-root outcomes."""

    return cls(
        top_refs=top_refs,
        results=results,
        leaves_consumed=sum(result.leaves_consumed for result in results),
        leaves_persisted=sum(result.leaves_persisted for result in results),
        leaves_failed=sum(result.leaves_failed for result in results),
        errors=tuple(
            f"top_ref[{index}]: {error}"
            for index, result in enumerate(results)
            for error in result.errors
        ),
    )

Runner API

See also

run_plugin

async_run_plugin

run_crawl

async_run_crawl

RunConfig

RunResult

PluginRunResult

from_runs(top_refs, results) classmethod

`from_runs(top_refs, results)` `classmethod`