fix: resolve affinity from authoritative invocation session

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<typeof inner>) {
 		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<typeof options>["onPayload"] = async (payload, payloadModel) => {
 			const payloadRecord = asRecord(payload)

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",

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<string, unknown> }).__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)
 })

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<string, unknown>) => (...args: any[]) => unknown
+	wrapSimpleCompletionStreamFn: (ctx: Record<string, unknown>) => (...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<string, () => void>()
+	const payloadReady = new Map<string, Promise<void>>()
+	for (const id of ["session-alpha", "session-beta"]) {
+		payloadReady.set(id, new Promise<void>((resolve) => payloadBarrier.set(id, resolve)))
+	}
+	const results = new Map<string, any>()
+	const responseCallbacks = new Map<string, string>()
+	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")
+})

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")
 })