feat(gateway): zero-retention history mode for ControlPlaneGateway

CHANGELOG.md

@@ -6,6 +6,24 @@ This project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ---
 
+## [Unreleased]
+
+### Added
+- **Zero-retention history mode.** `ControlPlaneGateway(history="client")` (or
+  `IDENTARK_HISTORY_MODE=client`) keeps the conversation in the gateway and
+  sends it in full each call with `store_history: false`, so the control plane
+  never stores prompts or responses. Assistant tool calls are kept so tool
+  loops replay correctly. New `history_mode`, `history` and `clear_history()`.
+  Requires a control plane that supports `store_history` (backend PR "session
+  history retention").
+
+### Fixed
+- `invoke_llm_stream` read `content` from SSE events, but the control plane
+  streams `delta`, so streamed text came back empty. Both are now accepted, and
+  an `{"error": ...}` event raises `ControlPlaneError` instead of ending silently.
+
+---
+
 ## [1.2.1] — 2026-07-02
 
 ### Fixed

identark/gateways/control_plane.py

@@ -15,7 +15,7 @@
 import os
 from collections.abc import AsyncGenerator
 from types import TracebackType
-from typing import Any, NoReturn
+from typing import Any, Literal, NoReturn
 
 import httpx
 
@@ -67,6 +67,13 @@ class ControlPlaneGateway:
         session_id:  Session identifier. Auto-detected from ``IDENTARK_SESSION_ID``.
         timeout:     Per-request timeout in seconds. Default: 30.
         max_retries: Retry attempts on transient failures. Default: 3.
+        history:     Where conversation history lives. ``"server"`` (default):
+                     the control plane keeps it, encrypted and deleted after
+                     the organisation's retention period, and you send only
+                     the new turn. ``"client"``: zero retention — this gateway
+                     keeps the history in memory, sends the full conversation
+                     each call, and the control plane stores none of it.
+                     Auto-detected from ``IDENTARK_HISTORY_MODE``.
 
     Examples::
 
@@ -82,6 +89,9 @@ class ControlPlaneGateway:
         # As an async context manager
         async with ControlPlaneGateway() as gateway:
             response = await gateway.invoke_llm(...)
+
+        # Zero retention: prompts and responses are never stored server-side
+        gateway = ControlPlaneGateway(history="client")
     """
 
     def __init__(
@@ -91,6 +101,7 @@ def __init__(
         session_id: str | None = None,
         timeout: float = 30.0,
         max_retries: int = 3,
+        history: Literal["server", "client"] | None = None,
     ) -> None:
         # Resolve credentials — constructor args take precedence over env vars
         self._api_key = (
@@ -108,9 +119,17 @@ def __init__(
             )
         if not self._url:
             raise ConfigurationError(
-                "No control plane URL found. Provide url= or set " "IDENTARK_CONTROL_PLANE_URL."
+                "No control plane URL found. Provide url= or set IDENTARK_CONTROL_PLANE_URL."
             )
 
+        mode = history or os.environ.get("IDENTARK_HISTORY_MODE") or "server"
+        if mode not in ("server", "client"):
+            raise ConfigurationError(f"history must be 'server' or 'client', got {mode!r}.")
+        self._history_mode: Literal["server", "client"] = mode  # type: ignore[assignment]
+        # Client-held conversation (OpenAI-shaped dicts, so assistant tool
+        # calls survive replay). Used only when history="client".
+        self._client_history: list[dict[str, Any]] = []
+
         self._timeout = timeout
         self._max_retries = max_retries
         self._client = httpx.AsyncClient(
@@ -157,26 +176,40 @@ async def invoke_llm(
     ) -> LLMResponse:
         """Send new messages to the LLM via the control plane."""
         validate_tool_definitions(tools)
-        payload: dict[str, Any] = {
-            "new_messages": [m.to_openai_dict() for m in new_messages],
-        }
-        if self._session_id:
-            payload["session_id"] = self._session_id
-        if tools:
-            payload["tools"] = tools
-            payload["tool_choice"] = tool_choice
+        new_dicts = [m.to_openai_dict() for m in new_messages]
+        payload = self._llm_payload(new_dicts, tools, tool_choice)
 
         logger.debug(
-            "invoke_llm new_messages=%d tools=%s",
+            "invoke_llm new_messages=%d tools=%s history=%s",
             len(new_messages),
             bool(tools),
+            self._history_mode,
         )
 
         data = await self._post("/llm/invoke", payload)
-        return self._parse_llm_response(data)
+        response = self._parse_llm_response(data)
+        if self._history_mode == "client":
+            assistant: dict[str, Any] = {"role": "assistant", "content": response.message.content}
+            if response.tool_calls:
+                assistant["tool_calls"] = [
+                    {
+                        "id": tc.id,
+                        "type": "function",
+                        "function": {"name": tc.function.name, "arguments": tc.function.arguments},
+                    }
+                    for tc in response.tool_calls
+                ]
+            self._client_history.extend([*new_dicts, assistant])
+        return response
 
     async def persist_messages(self, messages: list[Message]) -> None:
-        """Persist messages to conversation history via the control plane."""
+        """Persist messages to conversation history.
+
+        With ``history="client"`` they are kept in this gateway only.
+        """
+        if self._history_mode == "client":
+            self._client_history.extend(m.to_openai_dict() for m in messages)
+            return
         payload: dict[str, Any] = {
             "messages": [m.to_openai_dict() for m in messages],
         }
@@ -219,14 +252,9 @@ async def invoke_llm_stream(
         Also supports `event:` and `id:` fields (ignored).
         """
         validate_tool_definitions(tools)
-        payload: dict[str, Any] = {
-            "new_messages": [m.to_openai_dict() for m in new_messages],
-        }
-        if self._session_id:
-            payload["session_id"] = self._session_id
-        if tools:
-            payload["tools"] = tools
-            payload["tool_choice"] = tool_choice
+        new_dicts = [m.to_openai_dict() for m in new_messages]
+        payload = self._llm_payload(new_dicts, tools, tool_choice)
+        streamed: list[str] = []
 
         async with self._client.stream("POST", "/llm/stream", json=payload) as response:
             if response.status_code >= 400:
@@ -263,14 +291,59 @@ async def invoke_llm_stream(
                     logger.warning("SSE event is not a dict: %s", type(event))
                     continue
 
+                if "error" in event and len(event) == 1:
+                    raise ControlPlaneError(str(event["error"]))
+                # The control plane streams ``delta``; ``content`` is accepted
+                # for older servers.
+                content = event.get("content", event.get("delta", "")) or ""
+                streamed.append(content)
                 yield StreamChunk(
-                    content=event.get("content", ""),
+                    content=content,
                     finish_reason=event.get("finish_reason"),
                     model=event.get("model", "unknown"),
                     input_tokens=event.get("input_tokens", 0),
                     output_tokens=event.get("output_tokens", 0),
                 )
 
+        if self._history_mode == "client":
+            self._client_history.extend(
+                [*new_dicts, {"role": "assistant", "content": "".join(streamed)}]
+            )
+
+    @property
+    def history_mode(self) -> Literal["server", "client"]:
+        """``"server"`` or ``"client"`` (zero retention)."""
+        return self._history_mode
+
+    @property
+    def history(self) -> list[dict[str, Any]]:
+        """The client-held conversation (``history="client"`` only)."""
+        return list(self._client_history)
+
+    def clear_history(self) -> None:
+        """Forget the client-held conversation (``history="client"`` only)."""
+        self._client_history.clear()
+
+    def _llm_payload(
+        self,
+        new_dicts: list[dict[str, Any]],
+        tools: list[dict[str, Any]] | None,
+        tool_choice: str | dict[str, Any],
+    ) -> dict[str, Any]:
+        if self._history_mode == "client":
+            payload: dict[str, Any] = {
+                "new_messages": [*self._client_history, *new_dicts],
+                "store_history": False,
+            }
+        else:
+            payload = {"new_messages": new_dicts}
+        if self._session_id:
+            payload["session_id"] = self._session_id
+        if tools:
+            payload["tools"] = tools
+            payload["tool_choice"] = tool_choice
+        return payload
+
     async def get_session_cost(self) -> float:
         """Return the authoritative total session cost from the control plane."""
         params = {}

pyproject.toml

@@ -104,6 +104,13 @@ module = ["numpy", "numpy.*"]
 follow_imports = "skip"
 follow_imports_for_stubs = true
 
+# Same for `deprecated` (a transitive dependency): recent releases use PEP 695
+# `type` statements in classic.py, which break mypy's 3.10 parse in CI.
+[[tool.mypy.overrides]]
+module = ["deprecated", "deprecated.*"]
+follow_imports = "skip"
+follow_imports_for_stubs = true
+
 # ── Pytest ────────────────────────────────────────────────────────────────────
 [tool.pytest.ini_options]
 asyncio_mode    = "auto"

tests/unit/test_control_plane.py

@@ -345,3 +345,122 @@ async def test_network_exception_retries(self) -> None:
                 await gw.invoke_llm(new_messages=[Message(role=Role.USER, content="Hi")])
 
         assert gw._client.request.call_count == 2
+
+
+# ── History modes (server vs zero-retention client) ───────────────────────────
+
+
+def _sent_payload(gw: ControlPlaneGateway, call: int = -1) -> dict[str, Any]:
+    call_kwargs = gw._client.request.call_args_list[call]  # type: ignore[attr-defined]
+    payload: dict[str, Any] = call_kwargs.kwargs.get("json") or call_kwargs.args[2]
+    return payload
+
+
+class TestHistoryModes:
+    async def test_server_mode_sends_only_the_new_turn(self) -> None:
+        gw = _make_gateway(session_id="sess-1")
+        gw._client.request = AsyncMock(return_value=_mock_response(200, _llm_response_body()))
+
+        await gw.invoke_llm([Message(role=Role.USER, content="Hi")])
+
+        payload = _sent_payload(gw)
+        assert payload["new_messages"] == [{"role": Role.USER, "content": "Hi"}]
+        assert "store_history" not in payload
+        assert gw.history_mode == "server"
+        assert gw.history == []
+
+    async def test_client_mode_holds_and_replays_the_conversation(self) -> None:
+        gw = _make_gateway(session_id="sess-1", history="client")
+        gw._client.request = AsyncMock(return_value=_mock_response(200, _llm_response_body()))
+
+        await gw.invoke_llm([Message(role=Role.USER, content="Hi")])
+        await gw.invoke_llm([Message(role=Role.USER, content="And then?")])
+
+        first, second = _sent_payload(gw, 0), _sent_payload(gw, 1)
+        assert first["store_history"] is False and second["store_history"] is False
+        assert [m["content"] for m in first["new_messages"]] == ["Hi"]
+        assert [m["content"] for m in second["new_messages"]] == ["Hi", "Hello!", "And then?"]
+        assert [m["content"] for m in gw.history] == ["Hi", "Hello!", "And then?", "Hello!"]
+
+    async def test_client_mode_keeps_assistant_tool_calls(self) -> None:
+        gw = _make_gateway(history="client")
+        body = {
+            **_llm_response_body(),
+            "message": {"role": "assistant", "content": ""},
+            "tool_calls": [{"id": "call_1", "function": {"name": "search", "arguments": "{}"}}],
+        }
+        gw._client.request = AsyncMock(return_value=_mock_response(200, body))
+
+        await gw.invoke_llm([Message(role=Role.USER, content="Find it")])
+
+        assert gw.history[-1]["tool_calls"] == [
+            {"id": "call_1", "type": "function", "function": {"name": "search", "arguments": "{}"}}
+        ]
+
+    async def test_client_mode_persist_is_local(self) -> None:
+        gw = _make_gateway(history="client")
+        gw._client.request = AsyncMock(return_value=_mock_response(200, _llm_response_body()))
+
+        await gw.persist_messages([Message(role=Role.SYSTEM, content="Be brief.")])
+        assert gw._client.request.await_count == 0
+        await gw.invoke_llm([Message(role=Role.USER, content="Hi")])
+
+        assert [m["content"] for m in _sent_payload(gw)["new_messages"]] == ["Be brief.", "Hi"]
+        gw.clear_history()
+        assert gw.history == []
+
+    def test_history_mode_from_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("IDENTARK_HISTORY_MODE", "client")
+        assert _make_gateway().history_mode == "client"
+
+    def test_invalid_history_mode_is_rejected(self) -> None:
+        with pytest.raises(ConfigurationError):
+            _make_gateway(history="browser")  # type: ignore[arg-type]
+
+    async def test_client_mode_stream_records_the_streamed_reply(self) -> None:
+        import httpx
+
+        seen: list[dict[str, Any]] = []
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            seen.append(json.loads(request.content))
+            body = (
+                'data: {"delta": "Hello "}\n\n'
+                'data: {"delta": "world"}\n\n'
+                'data: {"delta": "", "finish_reason": "stop"}\n\n'
+            )
+            return httpx.Response(200, text=body, headers={"content-type": "text/event-stream"})
+
+        gw = _make_gateway(history="client")
+        gw._client = httpx.AsyncClient(
+            transport=httpx.MockTransport(handler), base_url="https://api.identark.io/v1"
+        )
+
+        chunks = [
+            c.content async for c in gw.invoke_llm_stream([Message(role=Role.USER, content="Hi")])
+        ]
+
+        assert "".join(chunks) == "Hello world"
+        assert seen[0]["store_history"] is False
+        assert gw.history == [
+            {"role": Role.USER, "content": "Hi"},
+            {"role": "assistant", "content": "Hello world"},
+        ]
+
+    async def test_stream_error_event_raises(self) -> None:
+        import httpx
+
+        from identark.exceptions import ControlPlaneError
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            return httpx.Response(200, text='data: {"error": "provider down"}\n\n')
+
+        gw = _make_gateway(history="client")
+        gw._client = httpx.AsyncClient(
+            transport=httpx.MockTransport(handler), base_url="https://api.identark.io/v1"
+        )
+
+        with pytest.raises(ControlPlaneError):
+            async for _ in gw.invoke_llm_stream([Message(role=Role.USER, content="Hi")]):
+                pass
+        assert gw.history == []