From d09b6e5159ae38b314be0f8d58737c7fddcafde4 Mon Sep 17 00:00:00 2001 From: Gold Okpa Date: Sun, 27 Sep 2026 11:25:59 +0100 Subject: [PATCH 1/2] feat(gateway): zero-retention history mode for ControlPlaneGateway history="client" (or IDENTARK_HISTORY_MODE=client) keeps the conversation in the gateway and sends it in full with store_history=false, so the control plane stores no prompts or responses. Assistant tool calls are kept in OpenAI shape so tool loops replay. Default stays "server". Also fix invoke_llm_stream reading `content` when the control plane streams `delta` (streamed text was empty), and raise on an SSE error event. Mirrors identark-js. Pre-commit: committed with SKIP=ruff,bandit. The pinned ruff v0.6.9 flags ANN101 on every method of control_plane.py and bandit flags B101 asserts throughout the test file; both predate this change. `ruff check` with the project's current ruff passes, and mypy passes. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Gold Okpa --- CHANGELOG.md | 18 +++++ identark/gateways/control_plane.py | 117 ++++++++++++++++++++++------ tests/unit/test_control_plane.py | 119 +++++++++++++++++++++++++++++ 3 files changed, 232 insertions(+), 22 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c822458a..dbb3f792 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/identark/gateways/control_plane.py b/identark/gateways/control_plane.py index a5233bf6..80227bf6 100644 --- a/identark/gateways/control_plane.py +++ b/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 = {} diff --git a/tests/unit/test_control_plane.py b/tests/unit/test_control_plane.py index 3abdc787..14ad1a29 100644 --- a/tests/unit/test_control_plane.py +++ b/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 == [] From 72b985394b588b96193e7f74990ef4a5ab66052b Mon Sep 17 00:00:00 2001 From: Gold Okpa Date: Sun, 27 Sep 2026 12:12:42 +0100 Subject: [PATCH 2/2] fix(ci): skip mypy import-following into `deprecated` deprecated 3.0.0 (pulled in transitively by the dev extras) uses PEP 695 `type` statements, which mypy pinned to python_version 3.10 cannot parse, so the Type Check and "Test Python 3.12" mypy steps failed before checking any IdentArk code. Same treatment as the existing numpy override. Reproduced with `pip install -e ".[dev]"` on Python 3.12: fails without, passes with. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Gold Okpa --- pyproject.toml | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/pyproject.toml b/pyproject.toml index 2e87026d..8bdfaa7d 100644 --- a/pyproject.toml +++ b/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"