Core protocols and data types for the Stream ingestion capability.
Stream turns a live feed of documents into grounded zettelkasten notes. It is a
general-purpose capability: earnings reports are only the first worked recipe.
The pieces here are the stable seams every recipe plugs into:
feed -> document(s) (FeedAdapter)
document -> schema name(s) (SchemaSelector)
document -> project name(s) (Router)
documents + schemas -> graph (WorkflowTemplate)
graph + role -> agent output (Driver, via the Executor)
This module deliberately keeps imports light: importing stream must not pull
in anthropic, openai, cursor_sdk, the coordinator, or the
zettelkasten store. Those are resolved lazily by the concrete implementations so
the package stays portable and cheap to import.
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
|
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
|
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
|
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
|
TaskSpec
dataclass
One node in a coordinator graph, in template-neutral form.
Source code in stream/protocols.py
| @dataclass
class TaskSpec:
"""One node in a coordinator graph, in template-neutral form."""
alias: str
agent: str
description: str
depends_on: list[str] = field(default_factory=list)
writes: list[str] | None = None
model: str = ""
passes: int = 0
schema: str = ""
def as_dict(self) -> dict[str, Any]:
out: dict[str, Any] = {
"alias": self.alias,
"agent": self.agent,
"description": self.description,
"depends_on": list(self.depends_on),
}
if self.writes is not None:
out["writes"] = list(self.writes)
if self.model:
out["model"] = self.model
if self.passes:
out["passes"] = self.passes
if self.schema:
out["schema"] = self.schema
return out
|
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)
|
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]:
...
|
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]:
...
|
SpineSelector
Bases: Protocol
Maps a document to the spine org id(s) to source extraction structure from.
Selection is a union, mirroring :class:SchemaSelector: a document may route
to several spines. The resolved set is merged with the run-global spine
baseline; returning an empty list means "no per-document spine" (only the
baseline, if any, applies).
Source code in stream/protocols.py
| @runtime_checkable
class SpineSelector(Protocol):
"""Maps a document to the spine org id(s) to source extraction structure from.
Selection is a union, mirroring :class:`SchemaSelector`: a document may route
to several spines. The resolved set is merged with the run-global spine
baseline; returning an empty list means "no per-document spine" (only the
baseline, if any, applies).
"""
def select(self, document: Document) -> list[str]:
...
|
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]:
...
|
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:
...
|
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:
...
|