diff --git a/index.ts b/index.ts index 2d137d0..6c80595 100644 --- a/index.ts +++ b/index.ts @@ -29,7 +29,7 @@ const REQUESTER_RUNTIME_SCHEMA_VERSION = "2026-06-04" const REQUESTER_RUNTIME_SOURCE = "openclaw-taas-affinity" const GIT_PROBE_TIMEOUT_MS = 250 const LAST_ROUTE_LIMIT = 256 -const PLUGIN_VERSION = "0.9.0" +const PLUGIN_VERSION = "0.10.0" // OpenClaw stores active registry state (including workspaceDir) on globalThis // under this well-known symbol key. @@ -77,7 +77,7 @@ type ResolvedSessionIdentity = { source: string sourceHint: string localSessionScoped: boolean - identityMode: "native" | "legacy_env" | "agent_scoped" + identityMode: "native" | "legacy_env" } /** @@ -120,29 +120,12 @@ function resolveSessionIdentity( } } - // Agent-scoped fallback. - // - // OpenClaw does not include `sessionId` in the wrapStreamFn context for the - // openai-completions transport, so neither branch above fires and the plugin - // previously injected nothing at all. That lane carries all agent traffic, so - // TaaS minted a fresh session id per request: affinity never engaged and - // continuity read 0% while prompt-cache reads showed ~47% real continuation. - // - // `agentId` IS supplied and is stable for the agent conversation, which is the - // granularity affinity needs. Scope it per workspace so two agents sharing a - // name in different workspaces do not collide, and label the mode distinctly - // so it is never mistaken for a native session id. - const agentId = safeString(agentIdFromCtx) - if (agentId) { - const scope = workspaceDirFromCtx ? `:${workspaceDirFromCtx}` : "" - return { - sessionId: deriveFallbackSessionId(`agent-scope:${agentId}${scope}`), - source: "openclaw:ctx.agentId", - sourceHint, - localSessionScoped: true, - identityMode: "agent_scoped", - } - } + // Deliberately do not derive conversation identity from agentId or workspace. + // Those scopes outlive individual sessions and would merge unrelated chats, + // subprocesses, or workers onto one TaaS affinity key. Current OpenClaw runs + // provide the authoritative identity per invocation through options.sessionId; + // requests without native or explicit legacy identity remain identity-less. + void agentIdFromCtx return null } @@ -417,25 +400,46 @@ function buildWrapper(ctx: ProviderWrapStreamFnContext) { const { streamFn } = ctx if (!streamFn) return undefined - const identity = resolveSessionIdentity( - ctx.workspaceDir, - (ctx as { sessionId?: unknown }).sessionId, - (ctx as { agentId?: unknown }).agentId, - ) - // Resolve agent identity from the session key when no dir/env hint exists so - // TaaS never falls back to minting its own per-request session id. - const agentIdForCapture = resolveAgentIdentity( - ctx as { agentDir?: string; workspaceDir?: string }, - identity?.sessionId - ) - const requesterRuntime = identity ? buildRequesterRuntime(ctx, identity.sessionId, identity.source, identity.sourceHint) : undefined - const correlation = identity ? buildCorrelationMetadata(identity.sessionId, identity.source, identity.sourceHint, ctx, agentIdForCapture) : undefined - - if (isDev) console.debug(`[taas-affinity] wrapStreamFn sessionId=${identity?.sessionId ?? "none"} mode=${identity?.identityMode ?? "none"}`) - const inner = streamFn return function taasAffinityStreamFn(...args: Parameters) { const [model, context, options] = args + // The provider wrapper is created before an embedded run is bound, so the + // authoritative conversation identity lives on this invocation's stream + // options. Resolve it here rather than once at wrapper construction time. + // Keeping every derived value in this invocation closure also prevents two + // concurrent sessions sharing a provider wrapper from contaminating one + // another's payload metadata or response capture. + const identity = resolveSessionIdentity( + ctx.workspaceDir, + (options as { sessionId?: unknown } | undefined)?.sessionId ?? + (ctx as { sessionId?: unknown }).sessionId, + (ctx as { agentId?: unknown }).agentId, + ) + // Resolve agent identity from the session key when no dir/env hint exists so + // TaaS never falls back to minting its own per-request session id. + const agentIdForCapture = resolveAgentIdentity( + ctx as { agentDir?: string; workspaceDir?: string }, + identity?.sessionId, + ) + const requesterRuntime = identity + ? buildRequesterRuntime(ctx, identity.sessionId, identity.source, identity.sourceHint) + : undefined + const correlation = identity + ? buildCorrelationMetadata( + identity.sessionId, + identity.source, + identity.sourceHint, + ctx, + agentIdForCapture, + ) + : undefined + + if (isDev) { + console.debug( + `[taas-affinity] stream invocation sessionId=${identity?.sessionId ?? "none"} mode=${identity?.identityMode ?? "none"}`, + ) + } + const prevOnPayload = options?.onPayload const onPayload: NonNullable["onPayload"] = async (payload, payloadModel) => { const payloadRecord = asRecord(payload) diff --git a/package-lock.json b/package-lock.json index fc54d8d..7eaad22 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "openclaw-taas-affinity", - "version": "0.9.0", + "version": "0.10.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "openclaw-taas-affinity", - "version": "0.9.0", + "version": "0.10.0", "license": "MIT", "devDependencies": { "@types/node": "^22.0.0", diff --git a/package.json b/package.json index 7b7a1c2..030e25f 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "openclaw-taas-affinity", - "version": "0.9.0", + "version": "0.10.0", "description": "OpenClaw provider plugin \u2014 CloudSigma TaaS session affinity. Injects a stable X-Session-Id header per conversation so TaaS can pin the session to the same OAuth token / Bedrock region / Claude Code node, maximising prompt-cache hit rates.", "type": "module", "main": "dist/index.js", diff --git a/test/agent-scoped-session.test.ts b/test/agent-scoped-session.test.ts index be81c44..a1a5d39 100644 --- a/test/agent-scoped-session.test.ts +++ b/test/agent-scoped-session.test.ts @@ -3,54 +3,24 @@ import test from "node:test" import plugin from "../index.ts" -/** - * OpenClaw does not include `sessionId` in the wrapStreamFn context for the - * openai-completions transport (see extra-params: it passes config, agentDir, - * workspaceDir, agentId, provider, modelId, extraParams, thinkingLevel, model, - * streamFn). That lane carries all agent traffic, so the plugin previously - * injected nothing and TaaS minted a fresh session id per request: - * - * 2248 OpenAI session records -> 2247 minted, 1 caller-supplied - * 272 requests -> 272 distinct session ids -> continuity 0% - * - * `agentId` is present and stable for the conversation, so it is a valid - * affinity key when no native session id exists. - */ - const internals = (plugin as unknown as { __test__?: Record }).__test__ type Resolve = ( workspaceDir?: string, sessionId?: unknown, - agentId?: unknown + agentId?: unknown, ) => { sessionId: string; identityMode: string; source: string } | null -test("agent id yields a stable session identity when no native session id exists", () => { +test("agent id alone does not fabricate a conversation identity", () => { assert.ok(internals) const resolve = internals!.resolveSessionIdentity as Resolve - const ws = "/home/x/.openclaw/workspace" - - const a = resolve(ws, undefined, "main") - const b = resolve(ws, undefined, "main") - - assert.ok(a, "must resolve an identity from agentId alone") - assert.equal(a!.sessionId, b!.sessionId, "identity must be stable across requests") - assert.equal(a!.identityMode, "agent_scoped") - assert.equal(a!.source, "openclaw:ctx.agentId") -}) - -test("different agents get different identities", () => { - const resolve = internals!.resolveSessionIdentity as Resolve - const ws = "/home/x/.openclaw/workspace" - assert.notEqual(resolve(ws, undefined, "main")!.sessionId, resolve(ws, undefined, "other")!.sessionId) + assert.equal(resolve("/home/x/.openclaw/workspace", undefined, "main"), null) }) -test("identity is workspace scoped so same-named agents do not collide", () => { +test("workspace and agent scopes do not merge unrelated conversations", () => { const resolve = internals!.resolveSessionIdentity as Resolve - assert.notEqual( - resolve("/ws/a", undefined, "main")!.sessionId, - resolve("/ws/b", undefined, "main")!.sessionId - ) + assert.equal(resolve("/ws/a", undefined, "main"), null) + assert.equal(resolve("/ws/b", undefined, "main"), null) }) test("a native session id always takes precedence", () => { @@ -60,7 +30,7 @@ test("a native session id always takes precedence", () => { assert.equal(r!.sessionId, "agent:main:main") }) -test("no session and no agent still yields no fabricated identity", () => { +test("no session and no agent yields no fabricated identity", () => { const resolve = internals!.resolveSessionIdentity as Resolve assert.equal(resolve("/home/x/workspace", undefined, undefined), null) }) diff --git a/test/invocation-session-id.test.ts b/test/invocation-session-id.test.ts new file mode 100644 index 0000000..35b1ebf --- /dev/null +++ b/test/invocation-session-id.test.ts @@ -0,0 +1,158 @@ +import assert from "node:assert/strict" +import test from "node:test" + +import plugin from "../index.ts" + +type RegisteredProvider = { + wrapStreamFn: (ctx: Record) => (...args: any[]) => unknown + wrapSimpleCompletionStreamFn: (ctx: Record) => (...args: any[]) => unknown +} + +function captureProvider(): RegisteredProvider { + let provider: RegisteredProvider | undefined + ;(plugin as any).register({ + registerProvider(candidate: RegisteredProvider) { + provider = candidate + }, + registerGatewayMethod() {}, + runtime: { system: { enqueueSystemEvent: () => true, requestHeartbeat: () => {} } }, + }) + assert.ok(provider) + return provider +} + +function createModel(id: string) { + return { + provider: "cloudsigma", + id, + api: "openai-completions", + baseUrl: "https://api.cloudsigma.com/ai/v1", + } +} + +test("invocation options.sessionId overrides wrapper context and drives the whole envelope", async () => { + const provider = captureProvider() + let receivedOptions: any + const wrapped = provider.wrapStreamFn({ + provider: "cloudsigma", + modelId: "gpt-5.6-sol", + agentId: "main", + workspaceDir: "/home/cloudsigma/.openclaw/workspace", + sessionId: "stale-context-session", + streamFn: async (model: unknown, _context: unknown, options: any) => { + receivedOptions = options + return await options.onPayload({ messages: [], metadata: {} }, model) + }, + }) + + const result: any = await wrapped(createModel("gpt-5.6-sol"), { messages: [] }, { + sessionId: "native-invocation-session", + }) + + assert.equal(result.metadata.session_id, "native-invocation-session") + assert.equal(result.metadata.sticky_key, "native-invocation-session") + assert.equal(result.metadata.requester_runtime.session_key, "native-invocation-session") + assert.equal(result.metadata.openclaw_correlation.session_id, "native-invocation-session") + assert.equal(receivedOptions.sessionId, "native-invocation-session") +}) + +test("managed GPT-5.6 and Kimi openai-completions invocations use options.sessionId", async () => { + const provider = captureProvider() + for (const modelId of ["gpt-5.6-sol", "kimi-k2"]) { + let payload: any + const wrapped = provider.wrapStreamFn({ + provider: "cloudsigma", + modelId, + agentId: "main", + workspaceDir: "/home/cloudsigma/.openclaw/workspace", + streamFn: async (model: unknown, _context: unknown, options: any) => { + payload = await options.onPayload({ messages: [], metadata: {} }, model) + return payload + }, + }) + const sessionId = `native-${modelId}` + await wrapped(createModel(modelId), { messages: [] }, { sessionId }) + assert.equal(payload.metadata.session_id, sessionId, modelId) + assert.equal(payload.metadata.openclaw_correlation.model_id, modelId, modelId) + } +}) + +test("simple completion wrapper uses an invocation sessionId when supplied", async () => { + const provider = captureProvider() + assert.equal(provider.wrapSimpleCompletionStreamFn, provider.wrapStreamFn) + let payload: any + const wrapped = provider.wrapSimpleCompletionStreamFn({ + provider: "cloudsigma", + modelId: "gpt-5.6-sol", + agentId: "main", + workspaceDir: "/home/cloudsigma/.openclaw/workspace", + streamFn: async (model: unknown, _context: unknown, options: any) => { + payload = await options.onPayload({ messages: [], metadata: {} }, model) + return payload + }, + }) + await wrapped(createModel("gpt-5.6-sol"), { messages: [] }, { + sessionId: "simple-native-session", + }) + assert.equal(payload.metadata.session_id, "simple-native-session") + assert.equal(payload.metadata.requester_runtime.session_key, "simple-native-session") +}) + +test("concurrent calls on one wrapper keep payload and response identity isolated", async () => { + const provider = captureProvider() + const payloadBarrier = new Map void>() + const payloadReady = new Map>() + for (const id of ["session-alpha", "session-beta"]) { + payloadReady.set(id, new Promise((resolve) => payloadBarrier.set(id, resolve))) + } + const results = new Map() + const responseCallbacks = new Map() + const wrapped = provider.wrapStreamFn({ + provider: "cloudsigma", + modelId: "gpt-5.6-sol", + agentId: "main", + workspaceDir: "/home/cloudsigma/.openclaw/workspace", + streamFn: async (model: unknown, _context: unknown, options: any) => { + const id = options.sessionId as string + payloadBarrier.get(id)!() + await Promise.all(payloadReady.values()) + const payload = await options.onPayload({ messages: [], metadata: {} }, model) + // Cross the calls again before response capture to expose shared mutable state. + await Promise.resolve() + await options.onResponse( + { + status: 200, + headers: { + "x-taas-autorouted": "true", + "x-taas-autorouter-model": `route-${id}`, + }, + }, + model, + ) + results.set(id, payload) + return payload + }, + }) + + await Promise.all([ + wrapped(createModel("gpt-5.6-sol"), { messages: [] }, { + sessionId: "session-alpha", + onResponse: async (response: any) => { + responseCallbacks.set("session-alpha", response.headers["x-taas-autorouter-model"]) + }, + }), + wrapped(createModel("gpt-5.6-sol"), { messages: [] }, { + sessionId: "session-beta", + onResponse: async (response: any) => { + responseCallbacks.set("session-beta", response.headers["x-taas-autorouter-model"]) + }, + }), + ]) + + assert.equal(results.get("session-alpha").metadata.session_id, "session-alpha") + assert.equal(results.get("session-beta").metadata.session_id, "session-beta") + assert.equal(results.get("session-alpha").metadata.openclaw_correlation.session_id, "session-alpha") + assert.equal(results.get("session-beta").metadata.openclaw_correlation.session_id, "session-beta") + assert.equal(responseCallbacks.get("session-alpha"), "route-session-alpha") + assert.equal(responseCallbacks.get("session-beta"), "route-session-beta") +}) diff --git a/test/simple-completion-affinity.test.ts b/test/simple-completion-affinity.test.ts index b9748b8..4ee19b8 100644 --- a/test/simple-completion-affinity.test.ts +++ b/test/simple-completion-affinity.test.ts @@ -12,10 +12,13 @@ test("simple completion hook is registered and shares affinity injection", async provider: "cloudsigma", modelId: "gpt-5.6-sol", agentId: "main", workspaceDir: "/home/cloudsigma/.openclaw/workspace", streamFn: inner, }) - const gen = wrapped("gpt-5.6-sol", {}, { onPayload: async (p: any) => p }) + const gen = wrapped("gpt-5.6-sol", {}, { + sessionId: "simple-completion-session", + onPayload: async (p: any) => p, + }) await gen.next() const patched = await innerOptions.onPayload({ model: "gpt-5.6-sol", messages: [{ role: "user", content: "hi" }] }, "gpt-5.6-sol") - assert.ok(patched.metadata?.session_id?.startsWith("oc:")) + assert.equal(patched.metadata?.session_id, "simple-completion-session") assert.equal(patched.metadata?.sticky_key, patched.metadata?.session_id) assert.equal(patched.metadata?.openclaw_correlation?.agent_id, "main") }) diff --git a/test/smoke.mjs b/test/smoke.mjs index 3772534..acb4847 100644 --- a/test/smoke.mjs +++ b/test/smoke.mjs @@ -68,7 +68,7 @@ const transportState = provider.resolveTransportTurnState({ sessionId: "smoke-lo assert.equal(transportState.headers["X-Session-Id"], "smoke-local-session") assert.equal(transportState.headers["X-OpenClaw-Session-Id"], transportState.headers["X-Session-Id"]) -assert.equal(transportState.headers["X-OpenClaw-Plugin-Version"], "0.9.0") +assert.equal(transportState.headers["X-OpenClaw-Plugin-Version"], "0.10.0") assert.equal(transportState.headers["X-OpenClaw-Turn-Id"], "turn-smoke") assert.equal(transportState.headers["X-OpenClaw-Attempt"], "1") @@ -93,7 +93,7 @@ assert.equal(capturedPayload.metadata.requester_runtime.tool_execution, "directi assert.equal("available_bridges" in capturedPayload.metadata.requester_runtime, false) assert.equal(capturedPayload.metadata.openclaw_correlation.schema_version, "2026-06-05") assert.equal(capturedPayload.metadata.openclaw_correlation.source, "openclaw-taas-affinity") -assert.equal(capturedPayload.metadata.openclaw_correlation.plugin_version, "0.9.0") +assert.equal(capturedPayload.metadata.openclaw_correlation.plugin_version, "0.10.0") assert.equal(capturedPayload.metadata.openclaw_correlation.session_id, localSessionA) assert.equal(capturedPayload.metadata.openclaw_correlation.sticky_key, localSessionA) assert.equal(capturedPayload.metadata.openclaw_correlation.provider, "cloudsigma")