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([]); + }); + }); });