feat(streaming): stream member answers + synthesis (closes #7)

DOCUMENTATION_INDEX.md

@@ -35,20 +35,21 @@ Package root: `src/conclave/` (installed as the `conclave` package; console scri
 | Module | Path | Responsibility |
 |--------|------|----------------|
 | Package API | [`src/conclave/__init__.py`](src/conclave/__init__.py) | Public exports: `Council`, `CouncilResult`, `ModelAnswer`, `TokenUsage`, `DebateRound`, `AdversarialResult`, `ConclaveConfig`, `load_config`, `__version__`. |
-| Council | [`src/conclave/council.py`](src/conclave/council.py) | Primary importable entry point. Reusable primitives (`fan_out`, `synthesize_blocks`) + the public mode API (`ask`/`debate`/`adversarial`, async + sync). |
+| Council | [`src/conclave/council.py`](src/conclave/council.py) | Primary importable entry point. Reusable primitives (`fan_out`, `synthesize_blocks`) + the public mode API (`ask`/`debate`/`adversarial`, async + sync) + streaming (`ask_stream`/`stream_sync`, synthesize/raw). |
 | Modes | [`src/conclave/modes.py`](src/conclave/modes.py) | `debate` (multi-round, anonymized peers, drop-out) and `adversarial` (propose → refute → verdict) orchestration, built on `Council.fan_out`/`synthesize_blocks`. |
+| Streaming | [`src/conclave/streaming.py`](src/conclave/streaming.py) | `stream_ask` — council-level streaming engine behind `Council.ask_stream`: concurrent member interleaving via an `asyncio.Queue`, optional synthesizer streaming, terminal `done` event with the full `CouncilResult` (synthesize/raw only). |
 | Prompts | [`src/conclave/prompts.py`](src/conclave/prompts.py) | Role/template strings for debate + adversarial and the anonymized peer-block builder. |
-| Providers | [`src/conclave/providers.py`](src/conclave/providers.py) | Single async `call_model` path: resolves the adapter, reads the key by name at call time, calls transport, parses; latency/usage/redacted-error capture; never raises. |
-| Transport | [`src/conclave/transport.py`](src/conclave/transport.py) | `post_json` — the single async httpx network boundary for the whole provider highway. |
+| Providers | [`src/conclave/providers.py`](src/conclave/providers.py) | Async `call_model` (buffered) + `call_model_stream` (SSE) paths: resolve the adapter, read the key by name at call time, call transport, parse; latency/usage/redacted-error capture; never raises (partial text preserved on mid-stream failure). |
+| Transport | [`src/conclave/transport.py`](src/conclave/transport.py) | `post_json` + `stream_sse` — the single async httpx network boundary (buffered POST and `client.stream(...)` SSE) for the whole provider highway. |
 | Adapter registry | [`src/conclave/adapters/__init__.py`](src/conclave/adapters/__init__.py) | `resolve_adapter(model_id, config)` — provider registry + **extension seam** (one registration per family; config-only for OpenAI-compatible endpoints). |
-| Adapter base | [`src/conclave/adapters/base.py`](src/conclave/adapters/base.py) | `ProviderAdapter` protocol, `ProviderError`, and `redact()` (error-string secret scrubber). |
+| Adapter base | [`src/conclave/adapters/base.py`](src/conclave/adapters/base.py) | `ProviderAdapter` protocol (`build_request`/`parse_response` + `stream_request`/`parse_sse_event`), `SSEDelta`, `ProviderError`, and `redact()` (error-string secret scrubber). |
 | OpenAI-compat adapter | [`src/conclave/adapters/openai_compat.py`](src/conclave/adapters/openai_compat.py) | `OpenAICompatAdapter` — openai/xai/perplexity + custom OpenAI-compatible endpoints. |
 | Anthropic adapter | [`src/conclave/adapters/anthropic.py`](src/conclave/adapters/anthropic.py) | `AnthropicAdapter` — native `/v1/messages`, system-prompt hoist, required `max_tokens`. |
 | Gemini adapter | [`src/conclave/adapters/gemini.py`](src/conclave/adapters/gemini.py) | `GeminiAdapter` — native `generateContent`, OpenAI-role mapping, `usageMetadata`. |
 | Registry | [`src/conclave/registry.py`](src/conclave/registry.py) | Friendly-name → model-id defaults; provider → env-var mapping; key **presence** logic (never values). |
 | Config | [`src/conclave/config.py`](src/conclave/config.py) | Loads/merges `~/.conclave/config.yml` over defaults; resolves model ids and named/CSV councils; parses the `endpoints:` section (custom OpenAI-compatible providers). |
-| Models | [`src/conclave/models.py`](src/conclave/models.py) | Pydantic result contract: `TokenUsage`, `ModelAnswer`, `DebateRound`, `AdversarialResult`, `CouncilResult` (`mode`/`rounds`/`adversarial`). Stable downstream surface. |
-| CLI | [`src/conclave/cli.py`](src/conclave/cli.py) | `conclave ask` (synthesize/raw/debate/adversarial; `--rounds`/`--proposer`) + `conclave providers`; rich panels and `--json`; never prints key values. |
+| Models | [`src/conclave/models.py`](src/conclave/models.py) | Pydantic result contract: `TokenUsage`, `ModelAnswer`, `StreamEvent`, `DebateRound`, `AdversarialResult`, `CouncilResult` (`mode`/`rounds`/`adversarial`). Stable downstream surface. |
+| CLI | [`src/conclave/cli.py`](src/conclave/cli.py) | `conclave ask` (synthesize/raw/debate/adversarial; `--rounds`/`--proposer`/`--stream`) + `conclave providers`; rich panels, live `--stream` output, and `--json`; never prints key values. |
 | Logging | [`src/conclave/logging.py`](src/conclave/logging.py) | Logger factory; stderr; verbosity via `CONCLAVE_LOG_LEVEL` (default `WARNING`). |
 
 ## Tests
@@ -62,6 +63,7 @@ Package root: `src/conclave/` (installed as the `conclave` package; console scri
 | Registry/config tests | [`tests/test_registry_config.py`](tests/test_registry_config.py) | Name resolution, key-presence logic, config merge. |
 | CLI tests | [`tests/test_cli.py`](tests/test_cli.py) | Typer `CliRunner`: exit-code contract (0 success / 1 zero-usable-answers / 2 usage error), `--json` payload + exit code, human renderers per mode, `providers` table never prints secrets, aclose lifecycle. |
 | Transport tests | [`tests/test_transport.py`](tests/test_transport.py) | `post_json` via httpx `MockTransport`: success/error-status/non-JSON fallback, timeout & connect/HTTP errors → `TransportError` (key never leaks), client reuse/pooling, aclose idempotency. |
+| Streaming tests | [`tests/test_streaming.py`](tests/test_streaming.py) | Per-adapter SSE via `MockTransport` (openai-compat/anthropic/gemini): incremental chunks + assembled answer == concatenation == buffered result; mid-stream malformed-frame/connection-drop/non-2xx → error set with partial text preserved (never raises); key redaction in stream errors; buffered `ask()` never opens a stream; `Council.ask_stream` interleaving + terminal `done` shape; CLI `--stream` smoke + exit-code contract + debate rejection; `--stream` + cache one-shot replay. |
 | Logging tests | [`tests/test_logging.py`](tests/test_logging.py) | `CONCLAVE_LOG_LEVEL` resolution (default `WARNING`, case-insensitive, unknown → `WARNING`), factory contract, one-shot configuration. |
 | Fixtures | [`tests/conftest.py`](tests/conftest.py) | Shared fixtures; mocks the httpx transport so the suite needs no network and no API keys. |
 

README.md

@@ -112,6 +112,18 @@ the synthesizer *and* the adversarial judge.
 of a council defined in your config (see below). The built-in `default` council
 is all known providers.
 
+Add `--stream` to render member (and synthesizer) tokens live as they arrive
+(`synthesize`/`raw` modes only):
+
+```bash
+conclave ask "Explain CRDTs in two paragraphs." -c grok,gemini,claude --stream
+```
+
+Streaming and the non-streaming default produce the **same** final
+`CouncilResult`; `--stream` only changes how output is rendered. It is ignored
+with `--json` (which always emits the full structured payload), and on a cache
+hit the cached text is rendered in one shot rather than as a fake token stream.
+
 ## Quickstart (library)
 
 ```python
@@ -131,6 +143,28 @@ for answer in result.answers:
 print("SYNTHESIS:\n", result.synthesis)
 ```
 
+### Streaming (synthesize/raw)
+
+`Council.ask_stream` is an async generator that yields incremental `StreamEvent`s
+as member (and synthesizer) tokens arrive, then a terminal `done` event carrying
+the full `CouncilResult` — the same shape `ask()` returns, so downstream code is
+unaffected:
+
+```python
+async for event in council.ask_stream("What is the capital of France?"):
+    if event.type in ("member_delta", "synthesis_delta"):
+        print(event.text, end="", flush=True)        # live token
+    elif event.type == "done":
+        result = event.result                          # full CouncilResult
+```
+
+Event types: `member_delta` / `member_done` (per member, interleaved),
+`synthesis_delta` / `synthesis_done` (when synthesizing), and the final `done`.
+A member that cannot stream, or any mid-stream failure, degrades gracefully —
+partial text is preserved and the error lands on that member's `ModelAnswer`,
+never raising (the same never-raises contract as `ask`). Streaming applies to
+`synthesize`/`raw` only; `debate`/`adversarial` are not streamed.
+
 ### Debate and adversarial modes
 
 ```python

SYSTEM_CONTEXT_DIAGRAM.md

@@ -105,6 +105,15 @@ flowchart TB
   speaks native `generateContent` (OpenAI roles mapped, `systemInstruction` hoisted,
   `usageMetadata` parsed). Every adapter builds a request and hands it to the **single**
   network boundary — `transport.post_json` (`transport.py`), one async httpx call site.
+- **Streaming shares the same boundary (PDD §9 #5).** A `--stream` run (and the library
+  `Council.ask_stream` async generator) flows through a streaming sibling of the call path:
+  `call_model_stream` (`providers.py`) → `transport.stream_sse` (`transport.py`, the single
+  streaming httpx call site, `client.stream(...)`) → each adapter's `stream_request` +
+  `parse_sse_event` (OpenAI-compat `data:`/`[DONE]` deltas; Anthropic named SSE events;
+  Gemini `streamGenerateContent?alt=sse`). `streaming.py` interleaves members concurrently
+  and emits `StreamEvent`s, ending with a `done` event whose `CouncilResult` matches the
+  non-streaming shape. Streaming covers `synthesize`/`raw` only; the never-raises +
+  `redact()` invariants hold identically, with partial text preserved on mid-stream failure.
 - **`resolve_adapter` is the extension seam.** Adding a *new provider family* is one
   registration in `adapters/__init__.py`; adding an *OpenAI-compatible endpoint* is
   **config-only** — a `~/.conclave/config.yml` `endpoints:` entry, no code. That is why

docs/PRODUCT_DESIGN_DOCUMENT.md

@@ -246,7 +246,8 @@ CLI (cli.py, typer+rich)   Library (from conclave import Council)
 | `modes.py` | Deliberation orchestration: `run_debate` (multi-round, anonymized peers, drop-out) and `run_adversarial` (propose → refute → verdict). Built entirely on `Council.fan_out` + `synthesize_blocks` — no duplicated concurrency or synthesizer code. |
 | `prompts.py` | Role/template strings for debate and adversarial (member, critic, judge, debate-final system prompts) and the anonymized peer-block builder. Separates *what each role is told* from *when to call whom*. |
 | `providers.py` | `call_model` — the single async call path. Resolves the adapter for a model id, reads the key value *by name at call time*, calls the adapter+transport, parses the reply, and captures latency, token usage, and any (redacted) error into a `ModelAnswer`; never raises for provider-side failures. Signature and never-raises contract unchanged from v0.1/v0.2. |
-| `transport.py` | The single async network boundary: `post_json` — one httpx call site for the whole highway. Nothing else in conclave touches the network. |
+| `transport.py` | The single async network boundary: `post_json` (buffered) + `stream_sse` (SSE via `client.stream(...)`, issue #7) — the only two httpx call sites for the whole highway. Nothing else in conclave touches the network. |
+| `streaming.py` | Council-level streaming engine (issue #7): `stream_ask` fans members out concurrently (interleaved via an `asyncio.Queue`), optionally streams the synthesizer, and emits `StreamEvent`s ending with a `done` event whose `CouncilResult` matches the buffered shape. Behind `Council.ask_stream`/`stream_sync`. synthesize/raw only. |
 | `adapters/__init__.py` | `resolve_adapter(model_id, config)` — the provider registry and **extension seam**. Maps a model-id prefix (or a config `endpoints:` entry) to the adapter that serves it. Adding a provider family = one registration here; adding an OpenAI-compatible endpoint = config-only. |
 | `adapters/base.py` | The `ProviderAdapter` protocol, `ProviderError`, and `redact()` — the secret-scrubber applied to every error string before it reaches `ModelAnswer.error`. |
 | `adapters/openai_compat.py` | `OpenAICompatAdapter` — serves openai / xai / perplexity / groq / deepseek / mistral / together and any custom OpenAI-compatible endpoint. Per-provider full completions URL (note: Perplexity has no `/v1` segment; Groq nests its OpenAI surface under `/openai/v1`). |
@@ -341,7 +342,8 @@ LLM-SDK dependency.** Packaged with hatchling; console script `conclave = concla
 - **No hosted/proxied token path.** No conclave-operated endpoint that sees user prompts or
   takes a margin. BYO-keys, direct-to-provider, always. (Permanent.)
 - **No persistence/caching of results** in v0.1 (caching is Roadmap, §9).
-- **No streaming** in v0.1 (Roadmap, §9).
+- **No streaming** in v0.1 (Roadmap, §9 — **LANDED for synthesize/raw in v0.3**, issue #7;
+  `debate`/`adversarial` streaming still out of scope).
 - **No server mode** in v0.1 (possible Roadmap, §9).
 - **No `vote` mode** yet (Roadmap, §9 — flagged for build, not for removal). `debate` and
   `adversarial` shipped in v0.2.
@@ -378,7 +380,23 @@ Ordered roughly by strategic value to the origin use case and to mcp-warden.
    defaults stay open under §12 #5. Struck through for traceability.
 4. **Caching** — optional result cache keyed on (prompt, council, mode, model ids) to make
    repeated/eval runs cheap. Must remain off by default and never persist keys.
-5. **Streaming** — stream member answers and/or the synthesis to the terminal/library.
+5. ~~**Streaming** — stream member answers and/or the synthesis to the terminal/library.~~
+   **LANDED** (issue #7): streaming for the `synthesize`/`raw` path. Library API is the
+   `Council.ask_stream` async generator (plus `Council.stream_sync` for non-async callers),
+   yielding `StreamEvent`s (`member_delta`/`member_done`, `synthesis_delta`/`synthesis_done`,
+   terminal `done` carrying the full `CouncilResult`); CLI flag is `--stream`. The single
+   streaming network boundary is `transport.stream_sse` (`client.stream(...)`), reusing the
+   pooled client/timeout/`redact()` plumbing; each adapter parses its own SSE delta shape
+   (OpenAI-compat `choices[].delta.content` + `[DONE]` with `stream_options.include_usage`;
+   Anthropic `content_block_delta` text + `message_start`/`message_delta` usage +
+   `message_stop`; Gemini `streamGenerateContent?alt=sse` `parts[].text` + cumulative
+   `usageMetadata`). A non-streamable model or any mid-stream error degrades gracefully —
+   partial text preserved, error on `ModelAnswer`, never raises — and the assembled
+   `ModelAnswer` (text + usage) matches the buffered result, so synthesis/`CouncilResult`
+   are unaffected. Non-streaming remains the default and is byte-for-byte unchanged.
+   `--stream` + cache: a hit renders the cached final text in one shot (no fake token
+   stream). `debate`/`adversarial` streaming is intentionally **out of scope** (a later
+   issue may extend them). Struck through for traceability.
 6. **Local HTTP/server mode (under evaluation)** — a *local* server for convenience only;
    must not become a hosted token path or violate the no-middleman non-goal.
 7. ~~**Key-leak hardening** — scrub/limit provider-originated error strings before they land

src/conclave/__init__.py

@@ -9,6 +9,13 @@
     result = council.ask_sync("Your prompt")            # sync
     result = await council.ask("Your prompt")            # async
 
+    # streaming (synthesize/raw only): incremental StreamEvents
+    async for event in council.ask_stream("Your prompt"):
+        if event.type in ("member_delta", "synthesis_delta"):
+            print(event.text, end="", flush=True)
+        elif event.type == "done":
+            result = event.result   # full CouncilResult, same shape as ask()
+
     # multi-round debate
     result = await council.debate("Your prompt", rounds=3)
     result = council.debate_sync("Your prompt", rounds=3)
@@ -39,6 +46,7 @@
     CouncilResult,
     DebateRound,
     ModelAnswer,
+    StreamEvent,
     TokenUsage,
 )
 from .transport import aclose
@@ -52,6 +60,7 @@
     "TokenUsage",
     "DebateRound",
     "AdversarialResult",
+    "StreamEvent",
     "ConclaveConfig",
     "load_config",
     "aclose",