stream

stream

Stream -- live document ingestion into the zettelkasten.

Stream turns a feed of documents into grounded zettelkasten notes by driving a coordinator graph with a portable agent backend (Claude/GPT by default, Cursor optional). It is general-purpose: earnings reports are the first worked recipe, not the only use.

Quick start (programmatic)::

import stream
report = stream.apply(
    ["./inbox/AAPL-10Q.pdf"],
    project="aapl",
    schemas=["earnings"],
    driver="claude",
)

Config-driven (declarative stream.yaml)::

import stream
stream.process_once("stream.yaml")          # one pass
stream.StreamDaemon.from_file().serve_forever()  # always on

Importing this package is cheap: provider SDKs (anthropic/openai/cursor_sdk), the coordinator, and the zettelkasten store are all imported lazily by the concrete implementations.

StreamDaemon dataclass

Always-on ingestion loop over a stream config.

Source code in stream/api.py

@dataclass
class StreamDaemon:
    """Always-on ingestion loop over a stream config."""

    config: StreamConfig
    interval: float = 30.0
    _seen: set[str] = field(default_factory=set)
    _feeds: dict[str, Any] = field(default_factory=dict)
    _stop: bool = False

    @classmethod
    def from_file(cls, path: str | Path = "stream.yaml", interval: float = 30.0) -> "StreamDaemon":
        return cls(config=StreamConfig.load(path), interval=interval)

    def stop(self) -> None:
        self._stop = True

    def serve_forever(self) -> None:
        logger.info("angelo-stream daemon starting (interval=%ss)", self.interval)
        while not self._stop:
            try:
                results = process_once(
                    self.config, seen=self._seen, feeds=self._feeds
                )
                if results:
                    logger.info("processed %d run(s)", len(results))
            except Exception as e:  # keep the daemon alive across transient errors
                logger.exception("process_once failed: %s", e)
            time.sleep(self.interval)

ReplayFeed

Bases: SegmentingLiveFeed

Replay a saved transcript file as if it were arriving live.

A dependency-free, stdlib-only :class:SegmentingLiveFeed subclass: it reads path once, splits it into fixed-size chunks, and releases them over successive :meth:poll calls to simulate text trickling in -- accumulating into ONE box, exactly like a real live feed. This is the development and test harness for a live pipeline (drive the real daemon end to end without a live source), and a worked example of the base.

chunk_interval (seconds) paces the release: on each poll the chunks whose scheduled time has arrived are released. The default 0 releases everything on the first poll -- deterministic, so unit tests need no clock; the whole transcript is then extracted as one delta into one box. Use a positive interval (with a persistent instance across polls) to exercise the true incremental, delta-by-delta path.

Source code in stream/live.py

class ReplayFeed(SegmentingLiveFeed):
    """Replay a saved transcript file as if it were arriving live.

    A dependency-free, stdlib-only :class:`SegmentingLiveFeed` subclass: it reads
    ``path`` once, splits it into fixed-size chunks, and releases them over
    successive :meth:`poll` calls to simulate text trickling in -- accumulating
    into ONE box, exactly like a real live feed. This is the development and test
    harness for a live pipeline (drive the real daemon end to end without a live
    source), and a worked example of the base.

    ``chunk_interval`` (seconds) paces the release: on each poll the chunks whose
    scheduled time has arrived are released. The default ``0`` releases everything
    on the first poll -- deterministic, so unit tests need no clock; the whole
    transcript is then extracted as one delta into one box. Use a positive
    interval (with a persistent instance across polls) to exercise the true
    incremental, delta-by-delta path.
    """

    def __init__(
        self,
        *,
        path: str,
        name: str = "replay",
        stream_key: str = "",
        title: str = "",
        chunk_size: int = 800,
        chunk_interval: float = 0.0,
        metadata: dict[str, Any] | None = None,
        drop_dir: str = "./inbox/replay",
        flush_threshold: int = 1500,
        overlap: int = 400,
        doc_type: str = "txt",
    ) -> None:
        super().__init__(
            name=name,
            drop_dir=drop_dir,
            flush_threshold=flush_threshold,
            overlap=overlap,
            doc_type=doc_type,
        )
        if chunk_size <= 0:
            raise ValueError("chunk_size must be positive")
        self.path = Path(path)
        self.stream_key = (stream_key or _slug(self.path.stem)).strip()
        self.title = title or self.path.stem
        self.chunk_size = chunk_size
        self.chunk_interval = max(0.0, float(chunk_interval))
        self.extra_metadata = dict(metadata or {})
        self._chunks: list[str] | None = None
        self._released = 0
        self._start: float | None = None

    def _load_chunks(self) -> list[str]:
        if self._chunks is None:
            text = self.path.read_text(encoding="utf-8") if self.path.is_file() else ""
            self._chunks = _split_chunks(text, self.chunk_size)
        return self._chunks

    def read_deltas(self) -> Iterable[StreamDelta]:
        import time

        chunks = self._load_chunks()
        total = len(chunks)
        if total == 0 or self._released >= total:
            return []

        if self.chunk_interval <= 0:
            due = total
        else:
            if self._start is None:
                self._start = time.monotonic()
            elapsed = time.monotonic() - self._start
            # +1 so the first chunk is due immediately at elapsed 0.
            due = min(total, int(elapsed // self.chunk_interval) + 1)

        deltas: list[StreamDelta] = []
        while self._released < due:
            i = self._released
            self._released += 1
            deltas.append(
                StreamDelta(
                    stream_key=self.stream_key,
                    text=chunks[i],
                    final=(self._released >= total),
                    title=self.title,
                    doc_type=self.doc_type,
                    metadata=dict(self.extra_metadata),
                )
            )
        return deltas

SegmentingLiveFeed

Bases: ABC

Base FeedAdapter for append-only text streams (one box per stream).

Subclasses implement :meth:read_deltas; the base accumulates each stream's text into one file and, once at least flush_threshold new characters have arrived (or on a final delta), emits ONE Document that extracts just the new window. Each emitted slice starts overlap characters before the prior cursor so a sentence split across a poll boundary is still mined and grounded in one pass. All of a stream's Documents share the box name slug(stream_key) and are refresh=True, so claims land in a single box.

Identity is monotonic per stream: (stream_key, "seg-NNNN"). The daemon dedups on it across polls; each poll's delta is a fresh identity, so it is extracted, while the shared box name keeps everything in one place.

Source code in stream/live.py

class SegmentingLiveFeed(abc.ABC):
    """Base ``FeedAdapter`` for append-only text streams (one box per stream).

    Subclasses implement :meth:`read_deltas`; the base accumulates each stream's
    text into one file and, once at least ``flush_threshold`` new characters have
    arrived (or on a ``final`` delta), emits ONE ``Document`` that extracts just
    the new window. Each emitted slice starts ``overlap`` characters before the
    prior cursor so a sentence split across a poll boundary is still mined and
    grounded in one pass. All of a stream's Documents share the box ``name``
    ``slug(stream_key)`` and are ``refresh=True``, so claims land in a single box.

    Identity is monotonic per stream: ``(stream_key, "seg-NNNN")``. The daemon
    dedups on it across polls; each poll's delta is a fresh identity, so it is
    extracted, while the shared box name keeps everything in one place.
    """

    #: Feed name; also stamped onto each segment's ``metadata["source_feed"]``.
    name: str = "live"

    def __init__(
        self,
        *,
        name: str = "live",
        drop_dir: str = "./inbox/live",
        flush_threshold: int = 1500,
        overlap: int = 400,
        doc_type: str = "txt",
    ) -> None:
        if flush_threshold <= 0:
            raise ValueError("flush_threshold must be positive")
        if overlap < 0:
            raise ValueError("overlap must be non-negative")
        if overlap >= flush_threshold:
            raise ValueError(
                f"overlap ({overlap}) must be smaller than flush_threshold "
                f"({flush_threshold})"
            )
        self.name = name
        self.drop_dir = Path(drop_dir)
        self.flush_threshold = flush_threshold
        self.overlap = overlap
        self.doc_type = (doc_type or "txt").lstrip(".").lower()
        # Per-stream mutable state (persists across polls).
        self._paths: dict[str, Path] = {}         # the ONE accumulating file per stream
        self._length: dict[str, int] = {}         # chars accumulated so far
        self._emitted_end: dict[str, int] = {}    # cursor: chars already handed to extraction
        self._seg_counts: dict[str, int] = {}     # next segment index
        self._titles: dict[str, str] = {}         # human title base per stream
        self._doc_types: dict[str, str] = {}      # resolved doc_type per stream
        self._meta: dict[str, dict[str, Any]] = {}  # accumulated per-stream metadata
        self._closed: set[str] = set()            # streams that emitted a final segment

    # -- subclass hook ------------------------------------------------------

    @abc.abstractmethod
    def read_deltas(self) -> Iterable[StreamDelta]:
        """Return text that has newly arrived since the last call.

        Must be non-blocking and idempotent-friendly: return only *new* text
        (the base accumulates and tracks the extraction cursor). Return an empty
        iterable when nothing is available. Raising is tolerated by the daemon
        (it logs and continues), but returning ``[]`` on a transient outage is
        preferred.
        """
        raise NotImplementedError

    # -- FeedAdapter.poll ---------------------------------------------------

    def poll(self) -> Iterable[Document]:
        touched: list[str] = []
        finals: set[str] = set()
        for delta in self.read_deltas():
            key = (delta.stream_key or self.name).strip() or self.name
            if key in self._closed and (delta.text or "").strip():
                logger.warning(
                    "%s: text for already-finalized stream %r; ignoring.",
                    self.name, key,
                )
                continue
            self._append(key, delta)
            touched.append(key)
            if delta.final:
                finals.add(key)

        docs: list[Document] = []
        for key in dict.fromkeys(touched):  # order-preserving unique
            pending = self._length.get(key, 0) - self._emitted_end.get(key, 0)
            is_final = key in finals
            if pending <= 0:
                if is_final:
                    self._closed.add(key)
                continue
            if pending < self.flush_threshold and not is_final:
                continue  # not enough new text yet; wait for the next poll
            docs.append(self._emit(key, final=is_final))
            if is_final:
                self._closed.add(key)
        return docs

    # -- internals ----------------------------------------------------------

    def _append(self, key: str, delta: StreamDelta) -> None:
        if delta.title and key not in self._titles:
            self._titles[key] = delta.title
        if delta.metadata:
            self._meta.setdefault(key, {}).update(delta.metadata)

        path = self._paths.get(key)
        if path is None:
            doc_type = (delta.doc_type or self.doc_type).lstrip(".").lower() or "txt"
            self._doc_types[key] = doc_type
            self.drop_dir.mkdir(parents=True, exist_ok=True)
            path = self.drop_dir / f"{_slug(key)}.{doc_type}"
            path.write_text("", encoding="utf-8")  # start a fresh accumulating file
            self._paths[key] = path
            self._length[key] = 0
            self._emitted_end[key] = 0
            self._seg_counts[key] = 0

        text = delta.text or ""
        if text:
            with path.open("a", encoding="utf-8") as fh:
                fh.write(text)
            self._length[key] = self._length.get(key, 0) + len(text)

    def _emit(self, key: str, *, final: bool) -> Document:
        idx = self._seg_counts.get(key, 0)
        self._seg_counts[key] = idx + 1

        start = self._emitted_end.get(key, 0)
        end = self._length.get(key, 0)
        slice_start = max(0, start - self.overlap)
        self._emitted_end[key] = end

        path = self._paths[key]
        doc_type = self._doc_types.get(key, self.doc_type)
        title_base = self._titles.get(key) or key
        seg_label = f"seg-{idx:04d}"
        metadata: dict[str, Any] = {
            **self._meta.get(key, {}),
            "stream_key": key,
            "segment": idx,
            "char_start": slice_start,
            "char_end": end,
            "final": final,
            "source_feed": self.name,
        }
        return Document(
            # ONE box per stream: every segment of a stream shares this name.
            name=_slug(key),
            path=str(path),
            title=f"{title_base} (through segment {idx})",
            doc_type=doc_type,
            # Monotonic per-stream identity so each poll's delta is a distinct,
            # un-deduped unit even though the box name is stable.
            identity=(key, seg_label),
            metadata=metadata,
            # Refresh the box's content pointer instead of creating a new box,
            # and extract only the new window.
            refresh=True,
            slices=[{"char_start": slice_start, "char_end": end}],
        )

read_deltas abstractmethod

read_deltas() -> Iterable[StreamDelta]

Return text that has newly arrived since the last call.

Must be non-blocking and idempotent-friendly: return only new text (the base accumulates and tracks the extraction cursor). Return an empty iterable when nothing is available. Raising is tolerated by the daemon (it logs and continues), but returning [] on a transient outage is preferred.

Source code in stream/live.py

@abc.abstractmethod
def read_deltas(self) -> Iterable[StreamDelta]:
    """Return text that has newly arrived since the last call.

    Must be non-blocking and idempotent-friendly: return only *new* text
    (the base accumulates and tracks the extraction cursor). Return an empty
    iterable when nothing is available. Raising is tolerated by the daemon
    (it logs and continues), but returning ``[]`` on a transient outage is
    preferred.
    """
    raise NotImplementedError

StreamDelta dataclass

Newly-arrived text for one logical stream, returned by read_deltas.

stream_key identifies the logical stream -- and thus the ONE box its text accumulates into (e.g. an event id like "AAPL-FY25Q1-call") -- so several concurrent streams can be multiplexed through one feed. final signals the stream has ended, so the base flushes any remaining un-extracted tail as a last delta even if it is under the threshold. metadata is merged onto every segment Document the stream produces (e.g. a ticker for the router).

Source code in stream/live.py

@dataclass
class StreamDelta:
    """Newly-arrived text for one logical stream, returned by ``read_deltas``.

    ``stream_key`` identifies the logical stream -- and thus the ONE box its text
    accumulates into (e.g. an event id like ``"AAPL-FY25Q1-call"``) -- so several
    concurrent streams can be multiplexed through one feed. ``final`` signals the
    stream has ended, so the base flushes any remaining un-extracted tail as a last
    delta even if it is under the threshold. ``metadata`` is merged onto every
    segment ``Document`` the stream produces (e.g. a ticker for the router).
    """

    stream_key: str
    text: str
    final: bool = False
    title: str = ""
    doc_type: str = ""
    metadata: dict[str, Any] = field(default_factory=dict)

AgentResult dataclass

The outcome of a single agent run.

Source code in stream/protocols.py

@dataclass
class AgentResult:
    """The outcome of a single agent run."""

    text: str
    ok: bool = True
    tool_calls: int = 0
    raw: Any = None

Document dataclass

A unit of source material flowing through the pipeline.

identity is the period-keyed logical identity used for dedup and restatement handling (e.g. ("AAPL", "FY2024Q3", "10-Q")), kept distinct from content_hash so an amended filing supersedes rather than duplicates the original. name is the kebab-case graph id used as the source-graph name in the zettelkasten store.

Source code in stream/protocols.py

@dataclass
class Document:
    """A unit of source material flowing through the pipeline.

    ``identity`` is the period-keyed logical identity used for dedup and
    restatement handling (e.g. ``("AAPL", "FY2024Q3", "10-Q")``), kept distinct
    from ``content_hash`` so an amended filing supersedes rather than duplicates
    the original. ``name`` is the kebab-case graph id used as the source-graph
    name in the zettelkasten store.
    """

    name: str
    path: str | None = None
    content_hash: str = ""
    text: str | None = None
    doc_type: str = ""
    title: str = ""
    identity: tuple[str, ...] | None = None
    metadata: dict[str, Any] = field(default_factory=dict)
    # GROWABLE / SLICED SOURCE (live streaming). ``refresh`` marks a source whose
    # underlying document grows over time: several Documents share one ``name``
    # (the box), and each refreshes that box's content pointer rather than
    # creating a new box, so incremental claims accumulate into ONE box per
    # logical source (e.g. one box per earnings CALL, not one per transcript
    # clipping). ``slices`` restricts extraction to the NEW char window so a
    # re-ingested growing document is not re-mined end to end each pass.
    refresh: bool = False
    slices: list[dict[str, Any]] | None = None

    def as_source(self) -> dict[str, Any]:
        """Render as a ``create_extraction_graph`` source dict."""
        src: dict[str, Any] = {"name": self.name}
        if self.path:
            src["path"] = self.path
        if self.content_hash:
            src["content_hash"] = self.content_hash
        if self.title:
            src["title"] = self.title
        if self.doc_type:
            src["doc_type"] = self.doc_type
        if self.refresh:
            src["refresh"] = True
        if self.slices:
            src["slices"] = list(self.slices)
        for key in ("authors", "year", "venue", "doi", "source_path", "date"):
            if key in self.metadata:
                src[key] = self.metadata[key]
        return src

as_source

as_source() -> dict[str, Any]

Render as a create_extraction_graph source dict.

Source code in stream/protocols.py

def as_source(self) -> dict[str, Any]:
    """Render as a ``create_extraction_graph`` source dict."""
    src: dict[str, Any] = {"name": self.name}
    if self.path:
        src["path"] = self.path
    if self.content_hash:
        src["content_hash"] = self.content_hash
    if self.title:
        src["title"] = self.title
    if self.doc_type:
        src["doc_type"] = self.doc_type
    if self.refresh:
        src["refresh"] = True
    if self.slices:
        src["slices"] = list(self.slices)
    for key in ("authors", "year", "venue", "doi", "source_path", "date"):
        if key in self.metadata:
            src[key] = self.metadata[key]
    return src

Driver

Bases: Protocol

Executes a single agent turn (a tool-use loop) for a coordinator task.

The executor owns role->tool gating and prompt assembly; the driver only runs the loop against whatever tools it is handed and returns the final text. score is optional (advertised by supports_logprobs) and is a single-shot scoring primitive separate from run_agent.

Source code in stream/protocols.py

@runtime_checkable
class Driver(Protocol):
    """Executes a single agent turn (a tool-use loop) for a coordinator task.

    The executor owns role->tool gating and prompt assembly; the driver only
    runs the loop against whatever tools it is handed and returns the final
    text. ``score`` is optional (advertised by ``supports_logprobs``) and is a
    single-shot scoring primitive separate from ``run_agent``.
    """

    name: str
    supports_logprobs: bool

    def run_agent(
        self,
        *,
        role: str,
        persona: str,
        prompt: str,
        tools: list[Tool],
        model: str = "",
        ctx: dict[str, Any] | None = None,
    ) -> AgentResult:
        ...

    def score(
        self,
        prompt: str,
        *,
        choices: list[str] | None = None,
        model: str = "",
    ) -> Score:
        ...

FeedAdapter

Bases: Protocol

Source of documents. Implementations may watch a directory, poll an API, or query a warehouse. poll returns documents newly available since the last call; it must be idempotent-friendly (the daemon dedups by identity).

Source code in stream/protocols.py

@runtime_checkable
class FeedAdapter(Protocol):
    """Source of documents. Implementations may watch a directory, poll an API,
    or query a warehouse. ``poll`` returns documents newly available since the
    last call; it must be idempotent-friendly (the daemon dedups by identity).
    """

    name: str

    def poll(self) -> Iterable[Document]:
        ...

GraphSpec dataclass

A built coordinator graph plus the prep metadata a template produced.

A template may have already performed synchronous side effects (ingesting sources, seeding hubs, pre-creating a synthesis structure). tasks is the DAG to register; prep carries any structured records (sources, structure) the caller wants to surface in the run report.

Source code in stream/protocols.py

@dataclass
class GraphSpec:
    """A built coordinator graph plus the prep metadata a template produced.

    A template may have already performed synchronous side effects (ingesting
    sources, seeding hubs, pre-creating a synthesis structure). ``tasks`` is the
    DAG to register; ``prep`` carries any structured records (sources, structure)
    the caller wants to surface in the run report.
    """

    tasks: list[dict[str, Any]]
    goal: str = ""
    prep: dict[str, Any] = field(default_factory=dict)

Router

Bases: Protocol

Maps a document to the zettelkasten project(s) it belongs to.

Routing is orthogonal to schema selection: schemas are global organizing lenses, projects are read-time scopes. A document may join several projects.

Source code in stream/protocols.py

@runtime_checkable
class Router(Protocol):
    """Maps a document to the zettelkasten project(s) it belongs to.

    Routing is orthogonal to schema selection: schemas are global organizing
    lenses, projects are read-time scopes. A document may join several projects.
    """

    def route(self, document: Document) -> list[str]:
        ...

SchemaSelector

Bases: Protocol

Maps a document to the schema name(s) it should be extracted under.

Selection is a union: a document can be processed under several schemas in a single extraction pass. Returning an empty list means schema-free ingestion (semantic notes only, no spine).

Source code in stream/protocols.py

@runtime_checkable
class SchemaSelector(Protocol):
    """Maps a document to the schema name(s) it should be extracted under.

    Selection is a union: a document can be processed under several schemas in a
    single extraction pass. Returning an empty list means schema-free ingestion
    (semantic notes only, no spine).
    """

    def select(self, document: Document) -> list[str]:
        ...

Score dataclass

A single-shot scoring result, optionally backed by logprobs.

Source code in stream/protocols.py

@dataclass
class Score:
    """A single-shot scoring result, optionally backed by logprobs."""

    value: str
    confidence: float | None = None
    logprob: float | None = None
    method: str = ""
    raw: Any = None

Tool dataclass

A function tool exposed to a Driver's tool-use loop.

write flags a tool that mutates the store; the executor strips write tools for read-only roles (planner/checker/meta) so only implementers (engineer/scribe) can change the graph.

Source code in stream/protocols.py

@dataclass
class Tool:
    """A function tool exposed to a Driver's tool-use loop.

    ``write`` flags a tool that mutates the store; the executor strips write
    tools for read-only roles (planner/checker/meta) so only implementers
    (engineer/scribe) can change the graph.
    """

    name: str
    description: str
    input_schema: dict[str, Any]
    execute: Callable[..., Any]
    write: bool = False

WorkflowTemplate

Bases: Protocol

Builds a coordinator graph for a batch of documents under given schemas.

The built-in extraction template wraps the grounded-extraction trio (extractor -> scribe -> auditor). Custom templates may build arbitrary coordinator graphs.

projects (optional) is the FULL set of projects a batch is routed to -- the built-in template extracts the sources ONCE and attaches the resulting claims to every project's synthesis spine. The singular project is the back-compat single-target shim (and the primary); a template may ignore projects and use project alone.

Source code in stream/protocols.py

@runtime_checkable
class WorkflowTemplate(Protocol):
    """Builds a coordinator graph for a batch of documents under given schemas.

    The built-in ``extraction`` template wraps the grounded-extraction trio
    (extractor -> scribe -> auditor). Custom templates may build arbitrary
    coordinator graphs.

    ``projects`` (optional) is the FULL set of projects a batch is routed to --
    the built-in template extracts the sources ONCE and attaches the resulting
    claims to every project's synthesis spine. The singular ``project`` is the
    back-compat single-target shim (and the primary); a template may ignore
    ``projects`` and use ``project`` alone.
    """

    name: str

    def build(
        self,
        *,
        documents: list[Document],
        schemas: list[str],
        project: str = "",
        projects: list[str] | None = None,
        options: dict[str, Any] | None = None,
    ) -> GraphSpec:
        ...

build_matrix

build_matrix(project: str, *, schema: str = '', columns: list[dict[str, Any]] | None = None, row_axis: dict[str, Any] | None = None, ai: bool = False, driver: Driver | None = None, model: str = '', table_id: str = 'stream-matrix', title: str = '') -> dict[str, Any]

Build the synthesis matrix for a project.

By default (ai=False) this is a cost-free deterministic projection: dimension-tagged notes route into schema_tag columns, one row per source. Pass ai=True with a driver to also fill free-text prompt columns and per-cell summaries via the agent (this persists the grid).

Source code in stream/analysis.py

def build_matrix(
    project: str,
    *,
    schema: str = "",
    columns: list[dict[str, Any]] | None = None,
    row_axis: dict[str, Any] | None = None,
    ai: bool = False,
    driver: Driver | None = None,
    model: str = "",
    table_id: str = "stream-matrix",
    title: str = "",
) -> dict[str, Any]:
    """Build the synthesis matrix for a project.

    By default (``ai=False``) this is a cost-free deterministic projection:
    dimension-tagged notes route into ``schema_tag`` columns, one row per source.
    Pass ``ai=True`` with a ``driver`` to also fill free-text ``prompt`` columns
    and per-cell summaries via the agent (this persists the grid).
    """
    from zettelkasten import server as zk
    # ``zettelkasten.tables.build_matrix`` (formerly ``build_table``) is aliased to
    # ``_build_matrix`` on purpose: it collides by name with this module's own
    # ``build_matrix`` above. The alias is what keeps the call below delegating to
    # the tables engine rather than recursing into this function.
    from zettelkasten.tables import GRAPHS_DIR, build_matrix as _build_matrix

    if columns is None:
        if not schema:
            raise ValueError("build_matrix needs either explicit columns or a schema.")
        columns = _schema_columns(schema)

    extract_fn = summarize_fn = None
    if ai:
        if driver is None:
            raise ValueError("ai=True requires a driver.")
        extract_fn = _driver_text_fn(driver, model)
        summarize_fn = _driver_text_fn(driver, model)

    return _build_matrix(
        zk._get_graph,
        project=project,
        name=project,
        table_id=table_id,
        title=title or f"{project} ({schema})" if schema else project,
        columns=columns,
        row_axis=row_axis,
        preview=not ai,  # deterministic, no persist, unless AI fills are requested
        force=ai,
        extract_fn=extract_fn,
        summarize_fn=summarize_fn,
        graphs_dir=GRAPHS_DIR,
    )

export_csv

export_csv(project: str, path: str, *, schema: str = '', columns: list[dict[str, Any]] | None = None, **kwargs: Any) -> str

Build the matrix and write it to path as CSV. Returns the path.

Source code in stream/analysis.py

def export_csv(
    project: str,
    path: str,
    *,
    schema: str = "",
    columns: list[dict[str, Any]] | None = None,
    **kwargs: Any,
) -> str:
    """Build the matrix and write it to ``path`` as CSV. Returns the path."""
    table = build_matrix(project, schema=schema, columns=columns, **kwargs)
    with open(path, "w", encoding="utf-8", newline="") as fh:
        fh.write(matrix_to_csv(table))
    return path

matrix_to_csv

matrix_to_csv(table: dict[str, Any]) -> str

Render a built matrix as CSV text (one row per source, columns as headers).

Source code in stream/analysis.py

def matrix_to_csv(table: dict[str, Any]) -> str:
    """Render a built matrix as CSV text (one row per source, columns as headers)."""
    columns = table.get("columns", [])
    rows = table.get("rows", [])
    buf = io.StringIO()
    writer = csv.writer(buf)
    writer.writerow(["source", *[c.get("label", c.get("key", "")) for c in columns]])
    for row in rows:
        cells = row.get("cells", {})
        line = [row.get("label", row.get("id", ""))]
        for col in columns:
            cell = cells.get(col.get("key"), {})
            value = cell.get("value") or cell.get("summary") or ""
            if cell.get("gap"):
                value = ""
            line.append(value)
        writer.writerow(line)
    return buf.getvalue()

render_grounded_matrix

render_grounded_matrix(table: dict[str, Any], *, max_quote: int = 240) -> str

Render the matrix as grounded text: values plus verbatim evidence quotes.

Source code in stream/analysis.py

def render_grounded_matrix(table: dict[str, Any], *, max_quote: int = 240) -> str:
    """Render the matrix as grounded text: values plus verbatim evidence quotes."""
    columns = table.get("columns", [])
    rows = table.get("rows", [])
    lines: list[str] = []
    header = " | ".join(["source", *[c.get("label", "") for c in columns]])
    lines.append(header)
    lines.append("-" * len(header))
    evidence: list[str] = []
    for row in rows:
        cells = row.get("cells", {})
        label = row.get("label", row.get("id", ""))
        values = []
        for col in columns:
            cell = cells.get(col.get("key"), {})
            if cell.get("gap"):
                values.append("")
            else:
                val = cell.get("value") or ""
                # Surface advisory low-confidence so the synthesizing agent can
                # hedge on weakly-extracted cells. Confidence is 0..1; flag only.
                if cell.get("low_confidence") and val:
                    conf = cell.get("confidence")
                    pct = f"{round(float(conf) * 100)}%" if isinstance(conf, (int, float)) else "low"
                    val = f"{val} [⚠ low confidence: {pct}]"
                values.append(val)
            ev = cell.get("evidence") or {}
            quote = (ev.get("quote") or "").strip() if isinstance(ev, dict) else ""
            if quote:
                evidence.append(
                    f"[{label} / {col.get('label','')}] \"{quote[:max_quote]}\""
                )
        lines.append(" | ".join([label, *values]))
    if evidence:
        lines.append("\nEVIDENCE (verbatim):")
        lines.extend(evidence)
    return "\n".join(lines)

synthesize_narrative

synthesize_narrative(project: str, *, driver: Driver, schema: str = '', columns: list[dict[str, Any]] | None = None, question: str = '', model: str = '', write_back: bool = False, cross_graph: str = _CROSS_GRAPH, title: str = '') -> dict[str, Any]

Build the grounded matrix and ask a Driver to synthesize a narrative.

Returns {narrative, matrix, note_id}. When write_back is set the narrative is committed as a synthesis note in cross_graph, claimed by project and tagged origin:agent so agent-authored writeback is distinguishable from human notes (governance).

Source code in stream/analysis.py

def synthesize_narrative(
    project: str,
    *,
    driver: Driver,
    schema: str = "",
    columns: list[dict[str, Any]] | None = None,
    question: str = "",
    model: str = "",
    write_back: bool = False,
    cross_graph: str = _CROSS_GRAPH,
    title: str = "",
) -> dict[str, Any]:
    """Build the grounded matrix and ask a Driver to synthesize a narrative.

    Returns ``{narrative, matrix, note_id}``. When ``write_back`` is set the
    narrative is committed as a ``synthesis`` note in ``cross_graph``, claimed by
    ``project`` and tagged ``origin:agent`` so agent-authored writeback is
    distinguishable from human notes (governance).
    """
    table = build_matrix(project, schema=schema, columns=columns)
    grounded = render_grounded_matrix(table)

    ask = question or (
        f"Synthesize the key cross-source findings for project '{project}'"
        + (f" under the '{schema}' lens" if schema else "")
        + ". Compare and contrast the sources, call out agreements, divergences, "
        "and notable gaps. Ground every claim in the evidence provided."
    )
    persona = (
        "You are a research analyst. Write a tight, well-structured narrative "
        "synthesis grounded ONLY in the matrix and verbatim evidence given. Do "
        "not invent facts. Note where evidence is missing rather than guessing."
    )
    prompt = f"{ask}\n\nGROUNDED MATRIX:\n{grounded}"
    narrative = driver.run_agent(
        role="checker", persona=persona, prompt=prompt, tools=[], model=model
    ).text

    note_id = ""
    if write_back:
        from zettelkasten import server as zk

        res = zk.add_note(
            graph=cross_graph,
            title=title or f"Synthesis: {project}{(' / ' + schema) if schema else ''}",
            type="synthesis",
            body=narrative,
            tags=["stream-synthesis", "origin:agent"],
            project=project,
        )
        if isinstance(res, dict):
            note_id = res.get("id") or res.get("note_id") or ""

    return {"narrative": narrative, "matrix": table, "note_id": note_id}

apply

apply(documents: Iterable[Any], *, project: str = '', schemas: list[str], projects: list[str] | None = None, template: str = 'extraction', driver: Any = 'claude', model: str = '', base_url: str = '', concurrency: int = 1, rigor: str = '', target_entry_id: str = '', options: dict[str, Any] | None = None, score_driver: str = '', score_model: str = '', score_base_url: str = '') -> dict[str, Any]

Prepare and run in one call. Returns {run_id, report, prep, projects}.

projects (when given) attaches the extracted claims to every listed project's spine in a SINGLE extraction run (extract-once, attach-to-many). The singular project is a back-compat shim and the primary; the returned projects reflects the full normalized target set.

Source code in stream/api.py

def apply(
    documents: Iterable[Any],
    *,
    project: str = "",
    schemas: list[str],
    projects: list[str] | None = None,
    template: str = "extraction",
    driver: Any = "claude",
    model: str = "",
    base_url: str = "",
    concurrency: int = 1,
    rigor: str = "",
    target_entry_id: str = "",
    options: dict[str, Any] | None = None,
    score_driver: str = "",
    score_model: str = "",
    score_base_url: str = "",
) -> dict[str, Any]:
    """Prepare and run in one call. Returns ``{run_id, report, prep, projects}``.

    ``projects`` (when given) attaches the extracted claims to every listed
    project's spine in a SINGLE extraction run (extract-once, attach-to-many).
    The singular ``project`` is a back-compat shim and the primary; the returned
    ``projects`` reflects the full normalized target set.
    """
    # Dedup while preserving order so the returned ``projects`` matches the
    # single-project run a duplicated set actually produces.
    normalized = list(dict.fromkeys(
        s for p in (projects if projects is not None else [project])
        if (s := (p or "").strip())
    ))
    handle = prepare(
        documents,
        project=project,
        projects=projects,
        schemas=schemas,
        template=template,
        target_entry_id=target_entry_id,
        rigor=rigor,
        options=options,
    )
    report = run(
        handle,
        driver=driver,
        model=model,
        base_url=base_url,
        concurrency=concurrency,
        score_driver=score_driver,
        score_model=score_model,
        score_base_url=score_base_url,
    )
    return {
        "run_id": handle.run_id,
        "report": report,
        "prep": handle.prep,
        "projects": normalized,
    }

prepare

prepare(documents: Iterable[Any], *, project: str = '', schemas: list[str], projects: list[str] | None = None, template: str = 'extraction', target_entry_id: str = '', rigor: str = '', options: dict[str, Any] | None = None) -> Run

Build the workflow graph and register it as a coordinator run.

projects (when given) is the FULL set of projects to attach extracted claims to: the sources are extracted ONCE and their claims attach to each project's spine in a single run. The singular project is a back-compat shim and the primary; at least one of project/projects is required. A single project is byte-identical to the legacy single-project path.

Source code in stream/api.py

def prepare(
    documents: Iterable[Any],
    *,
    project: str = "",
    schemas: list[str],
    projects: list[str] | None = None,
    template: str = "extraction",
    target_entry_id: str = "",
    rigor: str = "",
    options: dict[str, Any] | None = None,
) -> Run:
    """Build the workflow graph and register it as a coordinator run.

    ``projects`` (when given) is the FULL set of projects to attach extracted
    claims to: the sources are extracted ONCE and their claims attach to each
    project's spine in a single run. The singular ``project`` is a back-compat
    shim and the primary; at least one of ``project``/``projects`` is required.
    A single project is byte-identical to the legacy single-project path.
    """
    from coordinator import server as coord

    docs = _coerce_documents(documents)
    # Dedup while preserving order so ``projects=["A", "A"]`` collapses to a
    # single-project run (byte-identical to ``projects=["A"]``).
    normalized = list(dict.fromkeys(
        s for p in (projects if projects is not None else [project])
        if (s := (p or "").strip())
    ))
    if not normalized:
        raise ValueError(
            "prepare() requires at least one project (pass project= or projects=)."
        )
    primary = normalized[0]
    # Thread ``projects`` ONLY for a genuine multi-project run so the single-project
    # build call (and any custom template that predates the kwarg) is unaffected.
    build_kwargs = {"projects": normalized} if len(normalized) > 1 else {}
    spec = get_template(template).build(
        documents=docs, schemas=schemas, project=primary, options=options,
        **build_kwargs,
    )
    # EMPTY-GRAPH NO-OP: the template's incremental-skip guard emits an EMPTY
    # ``tasks`` list when every source is already fully covered by the
    # completed-slice ledger (no extraction trios -> no trailing memory task).
    # ``coord.create_graph([])`` would raise ("Graph must have at least one
    # task."), so detect it HERE and return a clean no-op ``Run`` (no coordinator
    # run registered, ``run_id=""``). The ``noop`` payload uses the STREAM-NATIVE
    # shape: it shares ``status``/``reason``/``reextract`` and the ``sources`` list
    # of ``{name, title, hub_id}`` dicts with ``create_extraction_graph``'s
    # ``status="noop"`` payload, but carries multi-valued ``schemas``/``projects``
    # lists (not a singular ``schema`` string) and OMITS the coordinator-only
    # ``skipped_sources``/``graph_summary`` keys (see the ``Run.noop`` note). Any
    # non-empty build falls through to the unchanged create_graph path.
    if not spec.tasks:
        noop = {
            "status": "noop",
            "reason": (
                "all sources already fully extracted (incremental skip); "
                "pass reextract=true to force a full re-extraction"
            ),
            "schemas": list(schemas),
            "projects": normalized,
            "sources": spec.prep.get("sources", []),
            "reextract": bool((options or {}).get("reextract", False)),
        }
        return Run(
            run_id="", goal=spec.goal, tasks=[], prep=spec.prep, noop=noop,
        )
    res = json.loads(
        coord.create_graph(
            spec.tasks,
            goal=spec.goal,
            target_entry_id=target_entry_id,
            rigor=rigor,
        )
    )
    return Run(run_id=res["run_id"], goal=spec.goal, tasks=spec.tasks, prep=spec.prep)

process_once

process_once(config: StreamConfig | str | Path = 'stream.yaml', *, seen: set[str] | None = None, feeds: dict[str, Any] | None = None) -> list[dict[str, Any]]

One pass: poll feeds, select+route, apply, archive. Returns run results.

feeds is an optional per-feed instance cache keyed by feed name. When given, each feed adapter is constructed once and REUSED across passes, so a STATEFUL feed (e.g. a :class:~stream.live.SegmentingLiveFeed that buffers a partial segment between polls, or a timed :class:~stream.live.ReplayFeed) keeps its state instead of being rebuilt each pass. The daemon passes its own cache; a bare one-shot process_once omits it and rebuilds per call (which is correct for stateless feeds like drop_dir / EDGAR).

Source code in stream/api.py

def process_once(
    config: StreamConfig | str | Path = "stream.yaml",
    *,
    seen: set[str] | None = None,
    feeds: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    """One pass: poll feeds, select+route, apply, archive. Returns run results.

    ``feeds`` is an optional per-feed instance cache keyed by feed name. When
    given, each feed adapter is constructed once and REUSED across passes, so a
    STATEFUL feed (e.g. a :class:`~stream.live.SegmentingLiveFeed` that buffers a
    partial segment between polls, or a timed :class:`~stream.live.ReplayFeed`)
    keeps its state instead of being rebuilt each pass. The daemon passes its own
    cache; a bare one-shot ``process_once`` omits it and rebuilds per call (which
    is correct for stateless feeds like ``drop_dir`` / EDGAR).
    """
    from .feeds import build_feed
    from .router import build_router, build_schema_selector, build_spine_selector

    cfg = config if isinstance(config, StreamConfig) else StreamConfig.load(config)
    selector = build_schema_selector(cfg.schemas)
    router = build_router(cfg.router)
    spine_selector = build_spine_selector(cfg.spine_routing)
    seen = seen if seen is not None else set()

    def _feed_for(fc):
        if feeds is None:
            return build_feed(fc)
        cached = feeds.get(fc.name)
        if cached is None:
            cached = build_feed(fc)
            feeds[fc.name] = cached
        return cached

    # Gather and dedup documents from every feed.
    docs: list[Document] = []
    for fc in cfg.feeds:
        for doc in _feed_for(fc).poll():
            ident = "|".join(str(x) for x in (doc.identity or (doc.name, doc.path)))
            if ident in seen:
                continue
            seen.add(ident)
            docs.append(doc)

    if not docs:
        return []

    # Divert tabular documents to the dataset sink BEFORE schema selection: their
    # numbers are the data, not something to extract. A doc whose doc_type is in
    # the configured sink set becomes a ``dataset`` note directly; the rest flow
    # to grounded extraction below. An empty sink set is byte-identical to before.
    results: list[dict[str, Any]] = []
    sink_types = set(cfg.datasets.doc_types)
    if sink_types:
        sink_docs: list[Document] = []
        rest: list[Document] = []
        for d in docs:
            dt = (d.doc_type or "").strip().lower().lstrip(".")
            (sink_docs if dt in sink_types else rest).append(d)
        docs = rest
        if sink_docs:
            results.extend(_dataset_sink(sink_docs, cfg, router))
            # In ``ingest`` mode the bytes are copied into the source folder, so
            # the original may be archived/deleted like an extracted doc. In
            # ``reference`` mode the dataset POINTS at the original, so it must
            # survive — archive it (a durable copy) but never delete it.
            for doc in sink_docs:
                if not doc.path:
                    continue
                p = Path(doc.path)
                _archive_original(p, cfg.archive)
                if cfg.datasets.mode == "ingest":
                    feed_cfg = next(
                        (f for f in cfg.feeds if f.name == doc.metadata.get("source_feed")), None
                    )
                    if feed_cfg and feed_cfg.options.get("delete_after") and p.exists():
                        try:
                            p.unlink()
                        except OSError as e:
                            logger.warning("could not delete %s: %s", p, e)

    if not docs:
        return results

    # Group by (project-SET, schema-set, spine-set) so documents sharing a lens
    # AND routed to the SAME set of projects AND spines batch into ONE run -- the
    # source is then extracted ONCE and its claims attach to every project in the
    # set (extract-once, attach-to-many-projects). Grouping is by the EXACT project
    # set: a doc routed to {A,B} extracts once into both; docs routed to {A} and
    # {A,C} stay separate runs (no cross-doc sharing for partial overlaps). The
    # spine-set is the per-document routing result UNIONed with the run-global
    # ``cfg.spines`` baseline. With a single-project router and no ``spine_routing:``
    # block every document resolves to the same one-element project set and the
    # baseline alone, so groups collapse exactly as before (byte-identical).
    groups: dict[
        tuple[frozenset[str], tuple[str, ...], tuple[str, ...]], list[Document]
    ] = {}
    for doc in docs:
        schemas = selector.select(doc)
        if not schemas:
            logger.info("Skipping %s: no schema selected.", doc.name)
            continue
        spine_set: list[str] = []
        for s in [*spine_selector.select(doc), *cfg.spines]:
            if s and s not in spine_set:
                spine_set.append(s)
        doc_projects = router.route(doc)
        if not doc_projects:
            continue
        key = (frozenset(doc_projects), tuple(schemas), tuple(spine_set))
        groups.setdefault(key, []).append(doc)

    for (project_set, schemas, spine_set), group_docs in groups.items():
        projects = sorted(project_set)
        primary = projects[0]
        try:
            result = apply(
                group_docs,
                project=primary,
                projects=projects,
                schemas=list(schemas),
                template=cfg.defaults.template,
                driver=cfg.defaults.driver,
                model=cfg.defaults.model,
                base_url=cfg.defaults.base_url,
                concurrency=cfg.defaults.concurrency,
                rigor=cfg.defaults.rigor,
                # Spine-sourced extraction (phase 3b): pass the opt-out mode +
                # explicit extra spine org ids. The extraction template resolves
                # each routed project's default spine (auto-included) and sources
                # structure from it when one is promoted.
                options={
                    "spine": cfg.defaults.spine,
                    "spines": list(spine_set),
                    "multi_spine": cfg.defaults.multi_spine,
                },
                score_driver=cfg.defaults.score_driver,
                score_model=cfg.defaults.score_model,
                score_base_url=cfg.defaults.score_base_url,
            )
            results.append({"projects": projects, "schemas": list(schemas), **result})
        except Exception as e:
            logger.exception("apply failed for projects %s: %s", projects, e)
            results.append(
                {"projects": projects, "schemas": list(schemas), "error": str(e)}
            )
            continue

        # Archive + clean up originals once their group succeeded.
        for doc in group_docs:
            if not doc.path:
                continue
            p = Path(doc.path)
            _archive_original(p, cfg.archive)
            feed_cfg = next((f for f in cfg.feeds if f.name == doc.metadata.get("source_feed")), None)
            if feed_cfg and feed_cfg.options.get("delete_after") and p.exists():
                try:
                    p.unlink()
                except OSError as e:
                    logger.warning("could not delete %s: %s", p, e)

    return results

run

run(handle: Run | str, *, driver: Any = 'claude', model: str = '', base_url: str = '', concurrency: int = 1, score_driver: str = '', score_model: str = '', score_base_url: str = '') -> dict[str, Any]

Drive a prepared run to completion and return the coordinator report.

When score_driver is set and handle is a :class:Run (so its source graphs are known), an advisory confidence-scoring pass runs after the graph completes — scoring each extracted claim against its evidence and persisting the result onto the note (see :mod:stream.scoring). The report gains a confidence summary. Scoring failures are logged, never fatal.

Source code in stream/api.py

def run(
    handle: Run | str,
    *,
    driver: Any = "claude",
    model: str = "",
    base_url: str = "",
    concurrency: int = 1,
    score_driver: str = "",
    score_model: str = "",
    score_base_url: str = "",
) -> dict[str, Any]:
    """Drive a prepared run to completion and return the coordinator report.

    When ``score_driver`` is set and ``handle`` is a :class:`Run` (so its source
    graphs are known), an advisory confidence-scoring pass runs after the graph
    completes — scoring each extracted claim against its evidence and persisting
    the result onto the note (see :mod:`stream.scoring`). The report gains a
    ``confidence`` summary. Scoring failures are logged, never fatal.
    """
    # An all-already-extracted no-op prepared no coordinator run, so there is
    # nothing to drive: return its ``noop`` payload verbatim as the report rather
    # than handing an empty ``run_id`` to the Executor (which would fail to find a
    # run). Only a :class:`Run` can carry this; a bare run-id string never does.
    if isinstance(handle, Run) and handle.noop is not None:
        return dict(handle.noop)
    run_id = handle.run_id if isinstance(handle, Run) else handle
    drv = (
        make_driver(driver, model=model, base_url=base_url)
        if isinstance(driver, str)
        else driver
    )
    executor = Executor(driver=drv, concurrency=concurrency)
    report = executor.run(run_id)

    if score_driver and isinstance(handle, Run):
        from .scoring import score_run_prep

        try:
            scorer = make_driver(score_driver, model=score_model, base_url=score_base_url)
            report["confidence"] = score_run_prep(handle.prep, driver=scorer, fallback=drv)
        except Exception as e:  # scoring is advisory — never fail the run over it
            logger.warning("confidence scoring pass failed: %s", e)
            report["confidence"] = {"error": str(e)}
    return report

is_low_confidence

is_low_confidence(score: Score, threshold: float = _LOW_CONFIDENCE) -> bool

Flag a score as low-confidence (advisory only; never gates publish).

Source code in stream/confidence.py

def is_low_confidence(score: Score, threshold: float = _LOW_CONFIDENCE) -> bool:
    """Flag a score as low-confidence (advisory only; never gates publish)."""
    return score.confidence is not None and score.confidence < threshold

pick_scorer

pick_scorer(primary: Driver, fallback: Driver | None = None) -> Driver | None

Return a logprob-capable driver, preferring primary.

Falls back to fallback when the primary cannot produce logprobs. Returns None when neither can (caller should skip or use the sampling proxy).

Source code in stream/confidence.py

def pick_scorer(primary: Driver, fallback: Driver | None = None) -> Driver | None:
    """Return a logprob-capable driver, preferring ``primary``.

    Falls back to ``fallback`` when the primary cannot produce logprobs. Returns
    ``None`` when neither can (caller should skip or use the sampling proxy).
    """
    if getattr(primary, "supports_logprobs", False):
        return primary
    if fallback is not None and getattr(fallback, "supports_logprobs", False):
        return fallback
    return None

score_claim

score_claim(claim: str, *, driver: Driver, evidence: str = '', fallback: Driver | None = None, model: str = '') -> Score

Score how well evidence supports claim (advisory).

Uses a logprob-capable driver when available (method "logprob"). When none is available it degrades to a single sampled yes/no judgment (method "sampling", confidence=None) so callers always get a Score.

Source code in stream/confidence.py

def score_claim(
    claim: str,
    *,
    driver: Driver,
    evidence: str = "",
    fallback: Driver | None = None,
    model: str = "",
) -> Score:
    """Score how well ``evidence`` supports ``claim`` (advisory).

    Uses a logprob-capable driver when available (method ``"logprob"``). When
    none is available it degrades to a single sampled yes/no judgment
    (method ``"sampling"``, ``confidence=None``) so callers always get a Score.
    """
    prompt = (
        "Does the evidence support the claim? Answer with a single word: "
        "yes or no.\n\n"
        f"CLAIM: {claim}\n"
        f"EVIDENCE: {evidence or '(none provided)'}"
    )
    scorer = pick_scorer(driver, fallback)
    if scorer is not None:
        score = scorer.score(prompt, choices=["yes", "no"], model=model)
        # Normalize so confidence reflects support (probability of "yes").
        if score.value.strip().lower().startswith("no") and score.confidence is not None:
            score.confidence = 1.0 - score.confidence
        return score

    # Sampling proxy: no logprobs anywhere. A bare judgment, confidence unknown.
    text = driver.run_agent(
        role="checker",
        persona="You are a strict grounding judge. Reply yes or no only.",
        prompt=prompt,
        tools=[],
        model=model,
    ).text
    supported = text.strip().lower().startswith("y")
    return Score(value="yes" if supported else "no", confidence=None, method="sampling")

score_run_prep

score_run_prep(prep: dict[str, Any], *, driver: Driver, **kwargs: Any) -> dict[str, Any]

Convenience: score the source graphs recorded in a Run's prep.

Source code in stream/scoring.py

def score_run_prep(prep: dict[str, Any], *, driver: Driver, **kwargs: Any) -> dict[str, Any]:
    """Convenience: score the source graphs recorded in a Run's ``prep``."""
    sources = [s.get("name", "") for s in (prep.get("sources") or []) if s.get("name")]
    if not sources:
        return {"scored": 0, "skipped": 0, "graphs": 0}
    return score_sources(sources, driver=driver, **kwargs)

score_sources

score_sources(source_graphs: list[str], *, driver: Driver, model: str = '', fallback: Driver | None = None, rescore: bool = False) -> dict[str, Any]

Score + persist advisory confidence for claim/finding notes.

Iterates each source graph's claim/finding notes, resolves the supporting quote (its evidence), asks the driver to judge support, and writes the resulting probability onto the note via :func:zettelkasten.server.set_note_confidence.

Already-scored notes are skipped unless rescore is set. Notes the scorer could not produce a numeric confidence for (no logprobs anywhere) are left untouched. Returns a summary {scored, skipped, graphs, ...}.

Source code in stream/scoring.py

def score_sources(
    source_graphs: list[str],
    *,
    driver: Driver,
    model: str = "",
    fallback: Driver | None = None,
    rescore: bool = False,
) -> dict[str, Any]:
    """Score + persist advisory confidence for claim/finding notes.

    Iterates each source graph's claim/finding notes, resolves the supporting
    quote (its evidence), asks the driver to judge support, and writes the
    resulting probability onto the note via
    :func:`zettelkasten.server.set_note_confidence`.

    Already-scored notes are skipped unless ``rescore`` is set. Notes the scorer
    could not produce a numeric confidence for (no logprobs anywhere) are left
    untouched. Returns a summary ``{scored, skipped, graphs, ...}``.
    """
    from zettelkasten import server as zk
    from zettelkasten.tables import _find_quote_evidence

    scored = 0
    skipped = 0
    no_confidence = 0
    for graph in source_graphs:
        try:
            zg = zk._get_graph(graph)
        except Exception as e:  # pragma: no cover - defensive
            logger.warning("score_sources: cannot load graph %s: %s", graph, e)
            continue
        # Snapshot values() — set_note_confidence mutates+saves as we go.
        for note in list(zg.notes.values()):
            if note.type not in _CLAIM_TYPES:
                continue
            g = note.grounding if isinstance(note.grounding, dict) else None
            if not rescore and g and g.get("score") is not None:
                skipped += 1
                continue
            claim_text = (note.body or note.title or "").strip()
            if not claim_text:
                continue
            ev = _find_quote_evidence(zg, note.id)
            evidence = ev["quote"] if ev else ""
            try:
                sc = score_claim(
                    claim_text,
                    driver=driver,
                    evidence=evidence,
                    fallback=fallback,
                    model=model,
                )
            except Exception as e:  # pragma: no cover - driver/runtime errors
                logger.warning("score_claim failed for %s/%s: %s", graph, note.id, e)
                continue
            if sc.confidence is None:
                # Sampling proxy (no logprobs): no numeric confidence to persist.
                no_confidence += 1
                continue
            zk.set_note_confidence(graph, note.id, sc.confidence, method=sc.method)
            scored += 1

    result: dict[str, Any] = {
        "scored": scored,
        "skipped": skipped,
        "graphs": len(source_graphs),
    }
    if no_confidence:
        result["no_confidence"] = no_confidence
        if scored == 0:
            result["note"] = (
                "scoring driver produced no logprobs; no confidence persisted. "
                "Point score_driver at a logprob-capable backend (e.g. gpt/vLLM)."
            )
    return result