@@ -83,8 +83,8 @@ def extract_event_fields(sdk_event: Any) -> dict[str, Any]:
 
         # reasoning_opaque for extended thinking signature preservation
         # Contract: streaming-contract:ThinkingBlock:MUST:1
-        # Anthropic models send encrypted extended thinking data that must be
-        # returned verbatim in subsequent turns for multi-turn extended thinking.
+        # Preserve Copilot's opaque bytes verbatim; the accumulator tags the
+        # persisted block as Copilot state, not a native upstream signature.
         reasoning_opaque = getattr(sdk_data, "reasoning_opaque", None)
         if reasoning_opaque and "reasoning_opaque" not in event_dict:
             event_dict["reasoning_opaque"] = reasoning_opaque
 
@@ -325,14 +325,20 @@ def to_chat_response(self) -> "StreamingChatResponse":
                     content_blocks.append(TextContent(text=block_text))
             elif btype == "thinking":
                 thinking_text: str = block["text"]
-                if thinking_text:
+                signature = block.get("opaque") or self.reasoning_opaque
+                # Only block-local opaque state can retain an empty block.
+                if thinking_text or block.get("opaque"):
                     # Contract: streaming-contract:ThinkingBlock:MUST:1
+                    # Copilot opaque state is not a native Anthropic/Gemini signature.
+                    # Keep its producer namespace through context persistence.
                     content.append(
                         ThinkingBlock(
                             thinking=thinking_text,
-                            signature=block.get("opaque") or self.reasoning_opaque,
+                            signature=signature,
+                            thinking_provenance={"version": 1, "provider": PROVIDER_ID},
                         )
                     )
+                if thinking_text:
                     content_blocks.append(ThinkingContent(text=thinking_text))
             elif btype == "tool_call":
                 tc = block["data"]
 
@@ -63,9 +63,9 @@ class ThinkingBlock(BaseModel):
     signature: str | None = None  # Opaque extended thinking signature
 ```
 
-**streaming-contract:ThinkingBlock:MUST:1**: Provider MUST preserve `reasoning_opaque` from SDK events as `ThinkingBlock.signature`.
+**streaming-contract:ThinkingBlock:MUST:1**: Provider MUST preserve `reasoning_opaque` from SDK events as `ThinkingBlock.signature`, including when visible thinking is empty. Persisted thinking blocks MUST carry the per-block extra `thinking_provenance: {"version": 1, "provider": "github-copilot"}`. Opaque state MUST NOT be rendered as text or streaming UI content.
 
-**Rationale:** Anthropic models send encrypted extended thinking data in `reasoning_opaque` which must be returned verbatim in subsequent turns for multi-turn extended thinking to work. The kernel's `ThinkingBlock.signature` field maps directly to the SDK's `reasoning_opaque`. GPT models do not use this field.
+**Rationale:** The SDK's `reasoning_opaque` is Copilot-owned state, not a native Anthropic or Gemini signature, even when Copilot hosts one of those models. The producer namespace lets native consumers distinguish foreign state after serialization or a provider switch; it does not bind state to a selected account or model. Copilot itself creates a fresh SDK session per request and renders only visible historical thinking through its text history adapter. Preserving and tagging opaque state does not change that SDK request behavior.
 
 ### ToolCallBlock (Pydantic) — for ChatResponse.content
 ```python
 
@@ -9,12 +9,22 @@
 from __future__ import annotations
 
 import logging
+from types import SimpleNamespace
 from typing import Any
 
+import pytest
+from amplifier_core import ChatRequest, ChatResponse, ThinkingBlock, ThinkingContent
+from amplifier_core.message_models import Message
+from pydantic import BaseModel
+
+from amplifier_module_provider_github_copilot.request_adapter import convert_chat_request
+from amplifier_module_provider_github_copilot.sdk_adapter import extract_event_fields
 from amplifier_module_provider_github_copilot.streaming import (
     DomainEvent,
     DomainEventType,
     StreamingAccumulator,
+    load_event_config,
+    translate_event,
 )
 
 
@@ -364,6 +374,134 @@ def test_thinking_block_handles_no_signature(self) -> None:
         # signature should be None (not missing, not error)
         assert thinking_block.signature is None
 
+    @pytest.mark.parametrize("empty_first", [False, True], ids=["empty-after", "empty-before"])
+    @pytest.mark.parametrize("thinking", ["Visible reasoning", ""], ids=["visible", "opaque-only"])
+    def test_empty_unsigned_block_does_not_inherit_retention_from_separate_block(
+        self, empty_first: bool, thinking: str
+    ) -> None:
+        """Only locally captured opaque state can retain an empty thinking block."""
+        unsigned = DomainEvent(
+            type=DomainEventType.CONTENT_DELTA, data={"text": ""}, block_type="THINKING"
+        )
+        signed = DomainEvent(
+            type=DomainEventType.CONTENT_DELTA,
+            data={"text": thinking, "reasoning_opaque": "copilot-opaque-fixture"},
+            block_type="THINKING",
+        )
+        separator = DomainEvent(type=DomainEventType.CONTENT_DELTA, data={"text": "Answer"})
+        accumulator = StreamingAccumulator()
+        events = [unsigned, separator, signed] if empty_first else [signed, separator, unsigned]
+        for event in events:
+            accumulator.add(event)
+
+        response = accumulator.to_chat_response()
+        assert [block.type for block in response.content] == (
+            ["text", "thinking"] if empty_first else ["thinking", "text"]
+        )
+        blocks = [block for block in response.content if isinstance(block, ThinkingBlock)]
+        assert len(blocks) == 1
+        assert blocks[0].thinking == thinking
+        assert blocks[0].signature == "copilot-opaque-fixture"
+        assert response.content_blocks is not None
+        ui_thinking = [
+            block for block in response.content_blocks if isinstance(block, ThinkingContent)
+        ]
+        assert [block.text for block in ui_thinking] == ([thinking] if thinking else [])
+
+    @pytest.mark.parametrize("object_event", [False, True], ids=["dict", "sdk-object"])
+    @pytest.mark.parametrize("split", [False, True], ids=["single", "split"])
+    @pytest.mark.parametrize(
+        ("thinking", "opaque"),
+        [
+            ("Visible reasoning", "copilot-opaque-fixture"),
+            ("", "copilot-opaque-fixture"),
+            ("Visible reasoning", None),
+            ("", None),
+        ],
+        ids=["signed-visible", "opaque-only", "unsigned-visible", "empty"],
+    )
+    def test_provenance_survives_capture_and_context_round_trip(
+        self,
+        object_event: bool,
+        split: bool,
+        thinking: str,
+        opaque: str | None,
+        caplog: pytest.LogCaptureFixture,
+    ) -> None:
+        """Tag producer state without rendering opaque bytes or changing SDK history."""
+        caplog.set_level(logging.DEBUG)
+        accumulator = StreamingAccumulator()
+        # A signature-only trailing delta must consolidate with visible reasoning.
+        chunks = [(thinking, None), ("", opaque)] if split else [(thinking, opaque)]
+        config = load_event_config()
+        for text, signature in chunks:
+            if object_event:
+                sdk_event = SimpleNamespace(
+                    type="assistant.reasoning_delta",
+                    data=SimpleNamespace(delta_content=text, reasoning_opaque=signature),
+                )
+                event = extract_event_fields(sdk_event)
+            else:
+                event = {
+                    "type": "assistant.reasoning_delta",
+                    "data": {"text": text, "reasoning_opaque": signature},
+                }
+            translated = translate_event(event, config)
+            assert translated is not None
+            accumulator.add(translated)
+
+        response = accumulator.to_chat_response()
+        blocks = [block for block in response.content if isinstance(block, ThinkingBlock)]
+        assert len(blocks) == int(bool(thinking or opaque))
+        if blocks:
+            assert blocks[0].thinking == thinking
+            assert blocks[0].signature == opaque
+            assert isinstance(blocks[0], BaseModel)
+            assert blocks[0].model_dump()["thinking_provenance"] == {
+                "version": 1,
+                "provider": "github-copilot",
+            }
+        assert response.text is None
+        if thinking:
+            assert response.content_blocks is not None
+            assert len(response.content_blocks) == 1
+            ui_block = response.content_blocks[0]
+            assert isinstance(ui_block, ThinkingContent)
+            assert ui_block.text == thinking
+        else:
+            assert response.content_blocks is None  # Opaque-only state is persistence-only.
+        assert "copilot-opaque-fixture" not in caplog.text
+
+        # The loop persists response content via model_dump, then rebuilds Messages
+        # and ChatRequests. Exercise both Python and JSON reloads without Core edits.
+        # Local Core stubs omit Pydantic APIs; retain real runtime validation here.
+        response_model: Any = ChatResponse
+        request_model: Any = ChatRequest
+        assert isinstance(response, BaseModel)
+        restored = response_model.model_validate(response.model_dump())
+        restored = response_model.model_validate_json(restored.model_dump_json())
+        assistant = Message(role="assistant", content=restored.content)
+        assert isinstance(assistant, BaseModel)
+        request = request_model(
+            messages=[assistant, Message(role="user", content="Continue")],
+        )
+        for _ in range(2):
+            request = request_model.model_validate(request.model_dump())
+            request = request_model.model_validate_json(request.model_dump_json())
+        assert request.messages[0].model_dump()["content"] == assistant.model_dump()["content"]
+
+        # Copilot history deliberately serializes visible thinking only. Stripping
+        # the new provenance must produce the exact same SDK prompt.
+        untagged = request.model_dump()
+        for block in untagged["messages"][0]["content"]:
+            block.pop("thinking_provenance", None)
+        prompt = convert_chat_request(request).prompt
+        assert prompt == convert_chat_request(request_model.model_validate(untagged)).prompt
+        assert "copilot-opaque-fixture" not in prompt
+        assert "thinking_provenance" not in prompt
+        if thinking:
+            assert f"[Thinking: {thinking}]" in prompt
+
 
 # ---------------------------------------------------------------------------
 # T-4 Contract Test Gaps — EventRouter integration tests
 
@@ -22,8 +22,10 @@
 
 from __future__ import annotations
 
+import logging
+from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -98,6 +100,65 @@ def _create_mock_request(model: str = "gpt-4o") -> MagicMock:
 # =============================================================================
 
 
+@pytest.mark.parametrize("use_streaming", [False, True], ids=["complete", "streaming"])
+@pytest.mark.parametrize("thinking", ["", "Visible reasoning"], ids=["opaque-only", "visible"])
+async def test_complete_tags_copilot_opaque_state(
+    use_streaming: bool,
+    thinking: str,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Both complete() modes persist producer provenance without leaking opaque state."""
+    from amplifier_core import ChatRequest, ThinkingBlock
+    from amplifier_core.message_models import Message
+    from pydantic import BaseModel
+
+    from amplifier_module_provider_github_copilot.provider import GitHubCopilotProvider
+
+    @dataclass
+    class ReasoningData(SessionEventData):
+        reasoning_opaque: str | None = None
+
+    client = MockCopilotClientWrapper(
+        events=[
+            SessionEvent(
+                type=SessionEventType.ASSISTANT_REASONING_DELTA,
+                data=ReasoningData(delta_content=thinking, reasoning_opaque="opaque-fixture"),
+            ),
+            text_delta_event("Answer"),
+        ],
+    )
+    coordinator = MagicMock()
+    coordinator.hooks.emit = AsyncMock()
+    provider = GitHubCopilotProvider(
+        client=client,  # type: ignore[arg-type]
+        coordinator=coordinator,
+        config={"use_streaming": use_streaming},
+    )
+    caplog.set_level(logging.DEBUG)
+    try:
+        response = await provider.complete(
+            ChatRequest(model="gpt-4o", messages=[Message(role="user", content="Test")]),
+        )
+        block = response.content[0]
+        assert isinstance(block, ThinkingBlock)
+        assert block.thinking == thinking
+        assert block.signature == "opaque-fixture"
+        assert isinstance(block, BaseModel)
+        assert block.model_dump()["thinking_provenance"] == {
+            "version": 1,
+            "provider": "github-copilot",
+        }
+        assert response.content[1].type == "text"
+        assert "opaque-fixture" not in caplog.text
+        assert "opaque-fixture" not in str(coordinator.hooks.emit.call_args_list)
+        emitted_names = [call.args[0] for call in coordinator.hooks.emit.call_args_list]
+        assert ("llm:stream_block_delta" in emitted_names) is use_streaming
+        assert client.session_instance is not None
+        assert client.session_instance.disconnected
+    finally:
+        await provider.close()
+
+
 class TestRetryWithEventualSuccess:
     """Test retry paths that succeed after initial failures."""