router facade¶

Angelo › 8. stream-ingestion › Phase 1: Core package (protocols + portable drivers + executor + templates + config/feeds/router + facade)

Type: Decision · Status: active · Created: 2026-06-26

Tags: stream,executor,drivers,apidriver,claude,gpt,extraction-template,stream-yaml,daemon,portable

Implemented the core stream/ package (new top-level package) on branch feature/stream-ingestion.

Modules: - protocols.py: Document, Tool, AgentResult, Score, TaskSpec, GraphSpec + Protocols (FeedAdapter, SchemaSelector, Router, WorkflowTemplate, Driver). Light imports only -- importing stream pulls in none of anthropic/openai/cursor_sdk/coordinator/zettelkasten. - tools.py: in-process zettelkasten tool registry wrapping zettelkasten.server functions (search_notes/find_by_title/get_note/get_fulltext/suggest_connections read; add_note/update_note/link_notes/add_to_project write). tools_for_role() strips write tools for read-only roles. - drivers/: ApiDriver base (provider-agnostic tool-use loop); ClaudeDriver (Anthropic, no logprobs); GptDriver (OpenAI-compatible, base_url for vLLM, score() via logprobs); CursorSdkDriver optional lazy stub. make_driver() dispatch. - executor.py: headless wave loop calling coordinator.server in-process (get_ready_tasks->claim_task->get_persona/get_task_context->driver.run_agent->submit_result), honoring path-scoped write leases (claims what it can, retries blocked next wave), with a conservative P0-only default_extension_policy via extend_graph. - templates.py: WorkflowTemplate registry + built-in ExtractionTemplate. Multi-schema: one prep_sources, one spine per schema (prep_structure), namespaced trios per schema so each claim attaches to its schema's spine. (Single-pass union extraction remains a follow-up on this shape.) - config.py: stream.yaml loader (defaults/archive/feeds/schemas/router). - feeds.py: DropDirFeed + build_feed factory. router.py: DeclarativeSchemaSelector (union match rules + optional classifier) + DeclarativeRouter (metadata axes -> projects). contrib.py: entry-point discovery (angelo.stream.feeds/routers/selectors/templates). - api.py: prepare()/run()/apply() facade + Run handle; process_once() (poll feeds -> select -> route -> batch by (project,schema-set) -> apply -> archive originals per none/local/dvc policy + delete_after) + StreamDaemon.serve_forever(). - main.py: angelo-stream console entry (--once / daemon).

pyproject.toml: added stream* to packages.find include, angelo-stream console script, and an optional [stream] extra (anthropic+openai). Ran a fast editable reinstall (pip install -e . --no-deps) to register the package/script -- non-destructive to running MCP servers (stream is not an MCP server).

Verified: import resolves to workspace from a neutral dir, extraction template registered, role gating correct, stream.yaml parse + schema selection + routing, driver dispatch, feed poll, console script --help. No edits touched other agents' uncommitted WIP -- all new files plus 3 additive pyproject lines.

Deferred to Phase 2: analysis (matrix CSV + agent narrative synthesis), confidence wiring (Note.grounding/tables), grounding-ui dashboard surfacing, distribution (angelo init scaffolding + stream rule + starter stream.yaml/recipe docs).

Referenced files¶

Source: memory entry deci-bdfde33c.