feat: proper token streaming via standard llm:stream_* contract (#70)

* feat(streaming): implement proper token streaming per provider-streaming-contract

Implements the five-event streaming contract defined in docs/provider-streaming-contract.md.

BREAKING CHANGE: llm:content_block is removed from the production streaming path.
It is replaced by:
  - llm:stream_block_start  {request_id, block_index, block_type}
  - llm:stream_block_delta  {request_id, block_index, sequence, text}
  - llm:stream_thinking_delta {request_id, block_index, sequence, text}
  - llm:stream_block_end    {request_id, block_index, block_type}
  - llm:stream_aborted      {request_id, error:{type, msg}}  (partial stream only)

## Ordering Approach

The core challenge: the SDK fires synchronous callbacks but emit() is async.
Fire-and-forget create_task() would allow tasks to run out of order.

Chosen solution: ordered asyncio.Queue consumer
- Per-call _StreamingContext holds an asyncio.Queue
- EventRouter calls stream_ctx.handle_delta() synchronously (FIFO enqueue)
- A single _run_stream_consumer coroutine drains the queue, awaiting each emit
- FIFO queue + single consumer = strict start→deltas→end ordering
- No locks needed (asyncio is single-threaded)

## Changes

### provider.py
- Add _StreamingContext dataclass: per-call state (NOT on self, concurrent-safe)
  - handle_delta(): emits block_start on block-type transition, delta event
  - close_current_block(): emits block_end for open block
  - signal_done(): puts None sentinel to stop consumer
- Add _run_stream_consumer(): ordered async drain, awaits each hooks.emit
- complete(): generate request_id once; compute _use_streaming with
  request.metadata['stream'] is False override (identity check per contract)
- _execute_sdk_completion(): create _StreamingContext + consumer task when
  use_streaming=True; pass stream_ctx to EventRouter; after idle_event:
  error path → emit stream_aborted if partial_emitted, signal_done, await;
  success path → close_current_block, signal_done, await consumer

### event_router.py
- Add optional stream_ctx parameter; when present, _emit_progressive_content
  routes to stream_ctx.handle_delta() instead of legacy emit_streaming_content
- _extract_delta_text(): also handles dict-style SDK events (tests + robustness)

### config/data/events.yaml
- Enable thinking_content_types: add assistant.reasoning_delta
  Required so EventRouter routes reasoning deltas to stream_ctx.handle_delta(text, 'thinking')

## Tests (TDD — written before implementation)
- tests/test_provider_streaming_contract.py (new, 32 tests):
  - _StreamingContext state machine (initial state, block tracking, transitions,
    sequence numbering, empty-text guard, partial_emitted flag)
  - _run_stream_consumer (ordering, graceful degradation, error handling)
  - EventRouter→_StreamingContext integration (text deltas, thinking deltas,
    transitions, ordering, per-block sequence, single request_id)
  - stream_aborted (after partial emit; NOT before any emit)
  - Non-streaming path (stream_ctx=None → no stream events)
  - use_streaming config + metadata override logic

### Updated tests (intentional behavior change)
- tests/test_contract_streaming.py: Updated 2 tests that tested old behavior:
  - test_thinking_content_types_empty_in_events_yaml → now asserts
    assistant.reasoning_delta IS in thinking_content_types (enabled for contract)
  - test_thinking_delta_does_not_trigger_emit_callback → now verifies that
    with stream_ctx provided, thinking deltas route to stream_ctx NOT legacy callback

## Test results
- 32 new contract tests: all pass
- 1305 existing tests: all pass (0 regressions)
- 10 pre-existing failures (SDK not installed, pyrightconfig.json missing): unchanged

🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier)

Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com>

* refactor(streaming): collapse thinking_delta into block_delta carrying block_type

Per provider-streaming-contract.md revision: ONE delta event (llm:stream_block_delta)
for ALL content; llm:stream_thinking_delta is REMOVED. block_type ("text"|"thinking")
is carried on EVERY delta payload so consumers route on block_type, not event name.

Changes:
- provider.py (_StreamingContext.handle_delta): remove conditional event_name;
  always emit llm:stream_block_delta with block_type captured by value from the
  function argument (NOT self.current_block_type — mutable shared field that could
  change before the FIFO consumer task processes the queue item).
- events.yaml: update comments referencing old llm:stream_thinking_delta name.
- tests/test_provider_streaming_contract.py: update all assertions for new contract:
    * test_delta_payload: expect block_type in payload
    * test_sequence_resets_on_block_transition: filter deltas by block_type
    * test_thinking_delta_event_name: check block_delta + block_type=thinking
    * test_transition_emits_block_end_then_new_block_start: names[1] is block_delta
    * test_thinking_delta_queues_thinking_events: assert block_delta + block_type
    * test_thinking_to_text_transition_full_sequence: both deltas are block_delta;
      payload assertions verify thinking vs text by block_type field
    * TestEventsYamlThinkingTypes: update docstrings
- tests/test_contract_streaming.py: update thinking_delta filter in
  test_thinking_delta_routes_to_stream_ctx_not_legacy_callback to match
  block_delta with block_type=thinking instead of the removed thinking_delta name.

Result: grep for llm:stream_thinking_delta is clean across all source and test
files. Non-streaming path and block_start/block_end/aborted events unchanged.

🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier)

Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com>

---------

Co-authored-by: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com>

Author: bkrabach

amplifier_module_provider_github_copilot/config/data/events.yaml

@@ -18,11 +18,13 @@ streaming_emission:
   text_content_types:
     - assistant.message_delta
     - assistant.streaming_delta
-  # thinking_content_types is intentionally empty: per-token ThinkingContent emission suppressed.
-  # Per-delta emission renders one 🧠 box per streaming chunk in the CLI.
-  # Full consolidated ThinkingContent is present in StreamingChatResponse.content_blocks.
-  # Contract: streaming-contract:ProgressiveStreaming:SHOULD:5
-  thinking_content_types: []
+  # thinking_content_types: enabled for provider-streaming-contract compliance.
+  # EventRouter uses these types to call stream_ctx.handle_delta(text, "thinking"),
+  # emitting llm:stream_block_delta with block_type="thinking" per token.
+  # The old per-token ThinkingContent (llm:content_block) is retired.
+  # Contract: provider-streaming-contract:llm:stream_block_delta
+  thinking_content_types:
+    - assistant.reasoning_delta
 
 event_classifications:
   bridge:

amplifier_module_provider_github_copilot/event_router.py

@@ -31,6 +31,7 @@
 )
 
 if TYPE_CHECKING:
+    from .provider import _StreamingContext
     from .streaming import EventConfig
 
 logger = logging.getLogger(__name__)
@@ -42,24 +43,34 @@ def _extract_delta_text(sdk_event: Any) -> str | None:
     Contract: streaming-contract:ProgressiveStreaming:SHOULD:1
 
     Handles SDK v0.3.0 nested data structure:
-    - event.data.delta_content for text deltas
+    - event.data.delta_content for text deltas (object-style SDK events)
+    - event["data"]["delta_content"] for dict-style events (tests and some SDK variants)
     - Direct string content as fallback
 
     Args:
-        sdk_event: SDK SessionEvent object
+        sdk_event: SDK SessionEvent object or dict
 
     Returns:
         Extracted text delta or None if not found
     """
-    # Try nested data structure first (SDK v0.3.0)
+    # Try nested data structure first (SDK v0.3.0 object-style)
     sdk_data = getattr(sdk_event, "data", None)
+    if sdk_data is None and isinstance(sdk_event, dict):
+        # Also handle dict-style events (used in tests and some SDK variants)
+        sdk_data = sdk_event.get("data")  # type: ignore[union-attr]
     if sdk_data is not None:
+        # Try attribute access (SDK objects)
         delta_content = getattr(sdk_data, "delta_content", None)
+        if delta_content is None and isinstance(sdk_data, dict):
+            # Also handle dict-style nested data
+            delta_content = sdk_data.get("delta_content")  # type: ignore[union-attr]
         if delta_content and isinstance(delta_content, str):
             return delta_content
 
     # Fallback: check direct delta_content attribute
     delta_content = getattr(sdk_event, "delta_content", None)
+    if delta_content is None and isinstance(sdk_event, dict):
+        delta_content = sdk_event.get("delta_content")  # type: ignore[union-attr]
     if delta_content and isinstance(delta_content, str):
         return delta_content
 
@@ -88,7 +99,8 @@ def __init__(
         ttft_state: dict[str, Any],
         ttft_threshold_ms: int,
         event_config: EventConfig,
-        emit_streaming_content: Callable[[Any], None],
+        emit_streaming_content: Callable[[Any], None] | None = None,
+        stream_ctx: _StreamingContext | None = None,
     ) -> None:
         """Initialize event router with all required dependencies.
 
@@ -101,7 +113,10 @@ def __init__(
             ttft_state: Mutable dict for TTFT tracking
             ttft_threshold_ms: TTFT warning threshold
             event_config: Event classification config
-            emit_streaming_content: Callback for progressive streaming
+            emit_streaming_content: Deprecated legacy callback. Prefer stream_ctx.
+            stream_ctx: Per-call streaming context for the five-event contract.
+                        If provided, stream events are routed here. If None, falls
+                        back to emit_streaming_content (legacy path).
         """
         self._queue = queue
         self._idle = idle_event
@@ -112,6 +127,7 @@ def __init__(
         self._ttft_threshold_ms = ttft_threshold_ms
         self._config = event_config
         self._emit_streaming = emit_streaming_content
+        self._stream_ctx = stream_ctx
 
     def __call__(self, sdk_event: Any) -> None:
         """Handle incoming SDK event.
@@ -269,17 +285,35 @@ def _emit_progressive_content(self, event_type: str, sdk_event: Any) -> None:
         """Emit content deltas for real-time UI updates.
 
         Contract: streaming-contract:ProgressiveStreaming:SHOULD:1
+        Contract: provider-streaming-contract.md (five-event contract)
+
+        When stream_ctx is set, routes through the per-call _StreamingContext queue
+        which guarantees block_start -> deltas -> block_end ordering via a single
+        async consumer. When stream_ctx is None (non-streaming path or legacy tests),
+        falls back to the emit_streaming_content callback if available.
         """
-        if event_type in self._config.text_content_types:
-            delta_text = _extract_delta_text(sdk_event)
-            if delta_text:
-                from amplifier_core import TextContent
-
-                self._emit_streaming(TextContent(text=delta_text))
-        elif event_type in self._config.thinking_content_types:  # pragma: no cover
-            # Thinking content only emitted by models with extended thinking
-            delta_text = _extract_delta_text(sdk_event)
-            if delta_text:
-                from amplifier_core import ThinkingContent
-
-                self._emit_streaming(ThinkingContent(text=delta_text))
+        if self._stream_ctx is not None:
+            # New five-event contract path: put events into the ordered queue
+            if event_type in self._config.text_content_types:
+                delta_text = _extract_delta_text(sdk_event)
+                if delta_text:
+                    self._stream_ctx.handle_delta(delta_text, "text")
+            elif event_type in self._config.thinking_content_types:
+                delta_text = _extract_delta_text(sdk_event)
+                if delta_text:
+                    self._stream_ctx.handle_delta(delta_text, "thinking")
+        elif self._emit_streaming is not None:
+            # Legacy path: kept for backward compat with existing tests
+            if event_type in self._config.text_content_types:
+                delta_text = _extract_delta_text(sdk_event)
+                if delta_text:
+                    from amplifier_core import TextContent
+
+                    self._emit_streaming(TextContent(text=delta_text))
+            elif event_type in self._config.thinking_content_types:  # pragma: no cover
+                # Thinking content only emitted by models with extended thinking
+                delta_text = _extract_delta_text(sdk_event)
+                if delta_text:
+                    from amplifier_core import ThinkingContent
+
+                    self._emit_streaming(ThinkingContent(text=delta_text))