From 36f0ddaed8d332ddddde8008dc663e670ebf9368 Mon Sep 17 00:00:00 2001 From: Gold Okpa Date: Sun, 27 Sep 2026 11:25:11 +0100 Subject: [PATCH] feat(gateway): zero-retention history mode for ControlPlaneGateway A sixth constructor argument, 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". Exports the HistoryMode type. Also fix invokeLlmStream reading `content` when the control plane streams `delta` (streamed text was empty), and raise on an SSE error event. Mirrors the Python SDK. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Gold Okpa --- src/gateways/control-plane.ts | 135 ++++++++++++++++++++++++++-------- src/index.ts | 1 + tests/control-plane.test.ts | 112 ++++++++++++++++++++++++++++ 3 files changed, 219 insertions(+), 29 deletions(-) diff --git a/src/gateways/control-plane.ts b/src/gateways/control-plane.ts index 0844905..c1eb86e 100644 --- a/src/gateways/control-plane.ts +++ b/src/gateways/control-plane.ts @@ -29,6 +29,16 @@ import { SessionNotFoundError, } from "../errors.js"; +/** + * Where conversation history lives. + * + * - `"server"` (default): the control plane keeps it, encrypted and deleted + * after the organisation's retention period; you send only the new turn. + * - `"client"`: zero retention. This gateway keeps the history in memory and + * sends the full conversation each call; the control plane stores none of it. + */ +export type HistoryMode = "server" | "client"; + /** * Production implementation of AgentGateway. * @@ -42,6 +52,9 @@ export class ControlPlaneGateway implements AgentGateway { private sessionId?: string; private timeout: number; private maxRetries: number; + private historyMode: HistoryMode; + /** Client-held conversation, OpenAI-shaped so assistant tool calls survive replay. */ + private clientHistory: Record[] = []; /** * Create a new ControlPlaneGateway instance. @@ -52,6 +65,8 @@ export class ControlPlaneGateway implements AgentGateway { * @param sessionId - Session identifier. Auto-detected from IDENTARK_SESSION_ID. * @param timeout - Per-request timeout in seconds. Default: 30. * @param maxRetries - Retry attempts on transient failures. Default: 3. + * @param history - `"server"` (default) or `"client"` for zero retention. + * Auto-detected from IDENTARK_HISTORY_MODE. */ constructor( apiKey?: string, @@ -59,6 +74,7 @@ export class ControlPlaneGateway implements AgentGateway { sessionId?: string, timeout: number = 30.0, maxRetries: number = 3, + history?: HistoryMode, ) { this.apiKey = apiKey || @@ -76,6 +92,12 @@ export class ControlPlaneGateway implements AgentGateway { this.timeout = timeout; this.maxRetries = maxRetries; + const mode = history || (process.env.IDENTARK_HISTORY_MODE as string | undefined) || "server"; + if (mode !== "server" && mode !== "client") { + throw new ConfigurationError(`history must be "server" or "client", got "${mode}".`); + } + this.historyMode = mode; + if (!this.apiKey) { throw new ConfigurationError( "No API key found. Provide apiKey= or set IDENTARK_API_KEY " + @@ -97,25 +119,38 @@ export class ControlPlaneGateway implements AgentGateway { tools?: Record[], toolChoice: string | Record = "auto", ): Promise { - const payload: Record = { - new_messages: newMessages.map((m) => messageToOpenAiDict(m)), - }; - if (this.sessionId) { - payload.session_id = this.sessionId; - } - if (tools) { - payload.tools = tools; - payload.tool_choice = toolChoice; - } + const newDicts = newMessages.map((m) => messageToOpenAiDict(m)); + const payload = this._llmPayload(newDicts, tools, toolChoice); const data = await this._post("/llm/invoke", payload); - return this._parseLlmResponse(data); + const response = this._parseLlmResponse(data); + if (this.historyMode === "client") { + const assistant: Record = { + role: "assistant", + content: response.message.content, + }; + if (response.tool_calls) { + assistant.tool_calls = response.tool_calls.map((tc) => ({ + id: tc.id, + type: "function", + function: { name: tc.function.name, arguments: tc.function.arguments }, + })); + } + this.clientHistory.push(...newDicts, assistant); + } + return response; } /** - * Persist messages to conversation history via the control plane. + * Persist messages to conversation history. + * + * With `history: "client"` they are kept in this gateway only. */ async persistMessages(messages: Message[]): Promise { + if (this.historyMode === "client") { + this.clientHistory.push(...messages.map((m) => messageToOpenAiDict(m))); + return; + } const payload: Record = { messages: messages.map((m) => messageToOpenAiDict(m)), }; @@ -159,16 +194,14 @@ export class ControlPlaneGateway implements AgentGateway { tools?: Record[], toolChoice: string | Record = "auto", ): AsyncGenerator { - const payload: Record = { - new_messages: newMessages.map((m) => messageToOpenAiDict(m)), + const newDicts = newMessages.map((m) => messageToOpenAiDict(m)); + const payload = this._llmPayload(newDicts, tools, toolChoice); + const streamed: string[] = []; + const recordReply = () => { + if (this.historyMode === "client") { + this.clientHistory.push(...newDicts, { role: "assistant", content: streamed.join("") }); + } }; - if (this.sessionId) { - payload.session_id = this.sessionId; - } - if (tools) { - payload.tools = tools; - payload.tool_choice = toolChoice; - } const headers: Record = { Authorization: `Bearer ${this.apiKey}`, @@ -210,28 +243,72 @@ export class ControlPlaneGateway implements AgentGateway { const data = line.slice("data:".length).trim(); if (data === "[DONE]") { + recordReply(); return; } + let event: Record; try { - const event = JSON.parse(data); - yield { - content: event.content || "", - finish_reason: event.finish_reason || null, - model: event.model || "unknown", - input_tokens: event.input_tokens || 0, - output_tokens: event.output_tokens || 0, - }; + event = JSON.parse(data); } catch { continue; } + if (typeof event !== "object" || event === null) continue; + if ("error" in event && Object.keys(event).length === 1) { + throw new ControlPlaneError(String(event.error)); + } + // The control plane streams `delta`; `content` is accepted for older servers. + const content = String(event.content ?? event.delta ?? ""); + streamed.push(content); + yield { + content, + finish_reason: (event.finish_reason as string | undefined) || null, + model: (event.model as string | undefined) || "unknown", + input_tokens: (event.input_tokens as number) || 0, + output_tokens: (event.output_tokens as number) || 0, + }; } } + recordReply(); } finally { reader.releaseLock(); } } + /** `"server"` or `"client"` (zero retention). */ + getHistoryMode(): HistoryMode { + return this.historyMode; + } + + /** The client-held conversation (`history: "client"` only). */ + get history(): Record[] { + return [...this.clientHistory]; + } + + /** Forget the client-held conversation (`history: "client"` only). */ + clearHistory(): void { + this.clientHistory = []; + } + + private _llmPayload( + newDicts: Record[], + tools: Record[] | undefined, + toolChoice: string | Record, + ): Record { + const payload: Record = + this.historyMode === "client" + ? { new_messages: [...this.clientHistory, ...newDicts], store_history: false } + : { new_messages: newDicts }; + if (this.sessionId) { + payload.session_id = this.sessionId; + } + if (tools) { + payload.tools = tools; + payload.tool_choice = toolChoice; + } + return payload; + } + /** * Return the authoritative total session cost from the control plane. */ diff --git a/src/index.ts b/src/index.ts index f9a6d1d..a9d7a6b 100644 --- a/src/index.ts +++ b/src/index.ts @@ -33,6 +33,7 @@ export type { AgentGateway } from "./gateway.js"; // Implementations export { DirectGateway } from "./gateways/direct.js"; export { ControlPlaneGateway } from "./gateways/control-plane.js"; +export type { HistoryMode } from "./gateways/control-plane.js"; export { ManagedToolsClient, type ManagedExecution, diff --git a/tests/control-plane.test.ts b/tests/control-plane.test.ts index 2709dd6..9f53225 100644 --- a/tests/control-plane.test.ts +++ b/tests/control-plane.test.ts @@ -319,4 +319,116 @@ describe("ControlPlaneGateway", () => { // No error should be thrown }); }); + + describe("history modes", () => { + const reply = (extra: Record = {}) => ({ + ok: true, + json: async () => ({ + message: { role: "assistant", content: "Hello!" }, + cost_usd: 0.001, + model: "gpt-4o", + finish_reason: "stop", + ...extra, + }), + }); + const sent = (call: number) => + JSON.parse((global.fetch as any).mock.calls[call][1].body); + const gw = (history?: "server" | "client") => + new ControlPlaneGateway("test-key", "http://localhost:3000", "sess-1", 30, 3, history); + + it("server mode sends only the new turn", async () => { + (global.fetch as any).mockResolvedValueOnce(reply()); + const gateway = gw(); + await gateway.invokeLlm([{ role: Role.USER, content: "Hi" }]); + + expect(sent(0).new_messages).toEqual([{ role: "user", content: "Hi" }]); + expect(sent(0).store_history).toBeUndefined(); + expect(gateway.getHistoryMode()).toBe("server"); + expect(gateway.history).toEqual([]); + }); + + it("client mode holds and replays the conversation", async () => { + (global.fetch as any).mockResolvedValue(reply()); + const gateway = gw("client"); + await gateway.invokeLlm([{ role: Role.USER, content: "Hi" }]); + await gateway.invokeLlm([{ role: Role.USER, content: "And then?" }]); + + expect(sent(0).store_history).toBe(false); + expect(sent(1).store_history).toBe(false); + expect(sent(0).new_messages.map((m: any) => m.content)).toEqual(["Hi"]); + expect(sent(1).new_messages.map((m: any) => m.content)).toEqual(["Hi", "Hello!", "And then?"]); + expect(gateway.history.map((m) => m.content)).toEqual(["Hi", "Hello!", "And then?", "Hello!"]); + }); + + it("client mode keeps assistant tool calls", async () => { + (global.fetch as any).mockResolvedValueOnce( + reply({ + message: { role: "assistant", content: "" }, + tool_calls: [{ id: "call_1", function: { name: "search", arguments: "{}" } }], + }), + ); + const gateway = gw("client"); + await gateway.invokeLlm([{ role: Role.USER, content: "Find it" }]); + + expect(gateway.history[1].tool_calls).toEqual([ + { id: "call_1", type: "function", function: { name: "search", arguments: "{}" } }, + ]); + }); + + it("client mode persist is local", async () => { + (global.fetch as any).mockResolvedValue(reply()); + const gateway = gw("client"); + await gateway.persistMessages([{ role: Role.SYSTEM, content: "Be brief." }]); + expect((global.fetch as any).mock.calls.length).toBe(0); + await gateway.invokeLlm([{ role: Role.USER, content: "Hi" }]); + + expect(sent(0).new_messages.map((m: any) => m.content)).toEqual(["Be brief.", "Hi"]); + gateway.clearHistory(); + expect(gateway.history).toEqual([]); + }); + + it("reads the mode from IDENTARK_HISTORY_MODE and rejects unknown modes", () => { + process.env.IDENTARK_HISTORY_MODE = "client"; + try { + expect(gw().getHistoryMode()).toBe("client"); + } finally { + delete process.env.IDENTARK_HISTORY_MODE; + } + expect(() => gw("browser" as any)).toThrow(ConfigurationError); + }); + + const sse = (text: string) => ({ + ok: true, + body: new Response(text).body, + }); + + it("client mode stream records the streamed reply", async () => { + (global.fetch as any).mockResolvedValueOnce( + sse('data: {"delta": "Hello "}\n\ndata: {"delta": "world"}\n\ndata: {"finish_reason": "stop"}\n\n'), + ); + const gateway = gw("client"); + let text = ""; + for await (const chunk of gateway.invokeLlmStream([{ role: Role.USER, content: "Hi" }])) { + text += chunk.content; + } + + expect(text).toBe("Hello world"); + expect(sent(0).store_history).toBe(false); + expect(gateway.history).toEqual([ + { role: "user", content: "Hi" }, + { role: "assistant", content: "Hello world" }, + ]); + }); + + it("stream error event raises and records nothing", async () => { + (global.fetch as any).mockResolvedValueOnce(sse('data: {"error": "provider down"}\n\n')); + const gateway = gw("client"); + await expect(async () => { + for await (const _ of gateway.invokeLlmStream([{ role: Role.USER, content: "Hi" }])) { + // drain + } + }).rejects.toThrow(ControlPlaneError); + expect(gateway.history).toEqual([]); + }); + }); });