stream.live

stream.live

Generic segmenting live-feed base for streaming text sources.

Turns an append-only text stream (a live transcript, a websocket, a log tail) into a single, growing source that is extracted incrementally -- one delta slice per poll -- so claims accumulate into ONE box per logical source (e.g. one box per earnings CALL, not one box per transcript clipping). The base owns the mechanics every live text source needs -- per-stream accumulation, a monotonic extraction cursor, delta slicing with a small overlap, and stable per-stream box identity -- so a concrete feed only implements one hook: :meth:SegmentingLiveFeed.read_deltas, which returns whatever text newly arrived.

How one box per call works with the extraction engine's static-document model: each poll appends the new transcript text to the call's ONE accumulating file and emits a Document that (a) shares the call's box name, (b) is marked refresh=True so the box's content pointer is refreshed rather than duplicated, and (c) carries a slices window covering only the NEW characters (plus a small overlap) so extraction mines just the delta while grounding verifies quotes against the whole accumulated transcript. See the create_source(update=True) growable-source path in zettelkasten/server.py and refresh handling in coordinator/extraction.py.

This module is deliberately domain-agnostic: it knows nothing about any vendor, schema, or document format. A vendor-specific live feed (e.g. an LSEG earnings-call feed) lives in a separate package that pip installs angelo and subclasses :class:SegmentingLiveFeed. The dependency-free :class:ReplayFeed built-in replays a saved transcript in timed chunks -- a worked example and a harness for developing/testing a live pipeline without a live source.

Statefulness: a live feed buffers across polls, so it must be a PERSISTENT instance. The daemon caches feeds across passes (process_once(..., feeds=...)), so this works under StreamDaemon; a bare one-shot process_once rebuilds feeds per call and is intended for stateless feeds (drop_dir / EDGAR).

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)

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

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