From 8dc4ceb2a6bc4e92d855e55856e5547446ca3692 Mon Sep 17 00:00:00 2001 From: CloudSigma Date: Fri, 10 Jul 2026 08:22:48 +0000 Subject: [PATCH] fix: scope TaaS affinity to local sessions --- index.ts | 55 +++++++++++++++------ package.json | 2 +- test/requester-runtime.test.ts | 2 +- test/smoke.mjs | 88 ++++++++++++++++++++-------------- 4 files changed, 95 insertions(+), 52 deletions(-) diff --git a/index.ts b/index.ts index 378c05e..dcd01cb 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.5.2" +const PLUGIN_VERSION = "0.5.3" // OpenClaw stores active registry state (including workspaceDir) on globalThis // under this well-known symbol key. @@ -68,13 +68,18 @@ function stableHash(value: string, prefix: string, length = 16): string { } function deriveSessionId(source: string): string { - const normalised = source.startsWith("env:") || source.startsWith("agent:") + const normalised = source.startsWith("env:") || source.startsWith("agent:") || source.startsWith("session:") ? source : path.resolve(source) const hex = createHash("sha256").update(normalised, "utf8").digest("hex") return `${SESSION_ID_PREFIX}${hex.slice(0, 16)}` } +function resolveLocalConversationSource(ctx: { sessionId?: unknown }): string | undefined { + const sessionId = safeString(ctx.sessionId) + return sessionId ? `session:${sessionId}` : undefined +} + function getActiveSessionSource(): string | undefined { const envSessionId = process.env.OPENCLAW_SESSION_ID if (envSessionId) return `env:${envSessionId}` @@ -94,11 +99,24 @@ function fallbackSessionSource(): string { return `stateDir:${stateDir}` } -function resolveSessionId(workspaceDirFromCtx?: string): { sessionId: string; source: string } { +function resolveSessionId(workspaceDirFromCtx?: string, sessionIdFromCtx?: unknown): { sessionId: string; source: string; sourceHint: string; localSessionScoped: boolean } { + const conversationSource = resolveLocalConversationSource({ sessionId: sessionIdFromCtx }) + if (conversationSource) { + const diagnosticSource = workspaceDirFromCtx ? `workspaceDir:${workspaceDirFromCtx}` : (getActiveSessionSource() ?? fallbackSessionSource()) + return { + sessionId: deriveSessionId(conversationSource), + source: conversationSource, + sourceHint: diagnosticSource, + localSessionScoped: true, + } + } + if (workspaceDirFromCtx) { return { sessionId: deriveSessionId(workspaceDirFromCtx), source: `workspaceDir:${workspaceDirFromCtx}`, + sourceHint: `workspaceDir:${workspaceDirFromCtx}`, + localSessionScoped: false, } } @@ -106,6 +124,8 @@ function resolveSessionId(workspaceDirFromCtx?: string): { sessionId: string; so return { sessionId: deriveSessionId(activeSource), source: activeSource, + sourceHint: activeSource, + localSessionScoped: false, } } @@ -170,6 +190,7 @@ function deriveAgentIdForCapture(ctx: { agentDir?: string; workspaceDir?: string function buildCorrelationMetadata( sessionId: string, source: string, + sourceHint: string, ctx: ProviderWrapStreamFnContext, agentId: string | null ): Record { @@ -183,7 +204,8 @@ function buildCorrelationMetadata( plugin_version: PLUGIN_VERSION, session_id: sessionId, sticky_key: sessionId, - session_source_hint: stableHash(source, "source"), + session_source_hint: stableHash(sourceHint, "source"), + session_identity_scope: source.startsWith("session:") ? "local_session" : "legacy_source", ...(agentId && { agent_id: agentId }), ...(provider && { provider }), ...(modelId && { model_id: modelId }), @@ -213,7 +235,8 @@ function buildCorrelationHeaders(args: { function buildRequesterRuntime( ctx: ProviderWrapStreamFnContext, sessionId: string, - source: string + source: string, + sourceHint: string ): RequesterRuntime { const ctxRecord = ctx as unknown as Record const modelRecord = asRecord(ctxRecord.model) @@ -233,7 +256,8 @@ function buildRequesterRuntime( ...(repoRoot && { git_dirty_hint: readGitDirtyHint(repoRoot) }), ...(provider && { provider }), ...(modelId && { model_id: modelId }), - session_source_hint: stableHash(source, "source"), + session_source_hint: stableHash(sourceHint, "source"), + session_identity_scope: source.startsWith("session:") ? "local_session" : "legacy_source", tool_execution: "direction_2_gateway", metadata_classification: { identifiers: "hashed", @@ -248,11 +272,12 @@ function patchPayloadMetadata( payload: Record, sessionId: string, requesterRuntime?: RequesterRuntime, - correlation?: Record + correlation?: Record, + injectAffinityMetadata = true ): Record { const existingMeta = asRecord(payload.metadata) ?? {} - const needsSessionId = !existingMeta.session_id - const needsStickyKey = !existingMeta.sticky_key + const needsSessionId = injectAffinityMetadata && !existingMeta.session_id + const needsStickyKey = injectAffinityMetadata && !existingMeta.sticky_key const needsRequesterRuntime = requesterRuntime && !existingMeta.requester_runtime const needsCorrelation = correlation && !existingMeta.openclaw_correlation if (!needsSessionId && !needsStickyKey && !needsRequesterRuntime && !needsCorrelation) return payload @@ -346,10 +371,10 @@ function buildWrapper(ctx: ProviderWrapStreamFnContext) { const { streamFn } = ctx if (!streamFn) return undefined - const { sessionId, source } = resolveSessionId(ctx.workspaceDir) + const { sessionId, source, sourceHint, localSessionScoped } = resolveSessionId(ctx.workspaceDir, (ctx as { sessionId?: unknown }).sessionId) const agentIdForCapture = deriveAgentIdForCapture(ctx as { agentDir?: string; workspaceDir?: string }) - const requesterRuntime = buildRequesterRuntime(ctx, sessionId, source) - const correlation = buildCorrelationMetadata(sessionId, source, ctx, agentIdForCapture) + const requesterRuntime = localSessionScoped ? buildRequesterRuntime(ctx, sessionId, source, sourceHint) : undefined + const correlation = localSessionScoped ? buildCorrelationMetadata(sessionId, source, sourceHint, ctx, agentIdForCapture) : undefined if (isDev) console.debug(`[taas-affinity] wrapStreamFn sessionId=${sessionId} source=${source}`) @@ -360,7 +385,7 @@ function buildWrapper(ctx: ProviderWrapStreamFnContext) { const onPayload: NonNullable["onPayload"] = async (payload, payloadModel) => { const payloadRecord = asRecord(payload) if (!payloadRecord) return prevOnPayload ? prevOnPayload(payload, payloadModel) : payload - const patched = patchPayloadMetadata(payloadRecord, sessionId, requesterRuntime, correlation) + const patched = patchPayloadMetadata(payloadRecord, sessionId, requesterRuntime, correlation, localSessionScoped) return prevOnPayload ? prevOnPayload(patched, payloadModel) : patched } const prevOnResponse = options?.onResponse @@ -377,7 +402,7 @@ function buildWrapper(ctx: ProviderWrapStreamFnContext) { } function buildTransportTurnState(ctx: ProviderResolveTransportTurnStateContext): ProviderTransportTurnState | null { - const activeSource = getActiveSessionSource() ?? fallbackSessionSource() + const activeSource = resolveLocalConversationSource(ctx as { sessionId?: unknown }) ?? getActiveSessionSource() ?? fallbackSessionSource() const sessionId = deriveSessionId(activeSource) const agentId = deriveAgentIdForCapture(ctx as unknown as { agentDir?: string; workspaceDir?: string }) if (isDev) { @@ -424,7 +449,7 @@ export default { return } - const resolvedSessionId = directSessionId ?? resolveSessionId(workspaceDir).sessionId + const resolvedSessionId = directSessionId ?? resolveSessionId(workspaceDir, safeString(pp.localSessionId)).sessionId respond(true, { sessionId: resolvedSessionId, capture: getLastRouteForSession(resolvedSessionId), diff --git a/package.json b/package.json index c9f42f1..adf2b53 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "openclaw-taas-affinity", - "version": "0.5.2", + "version": "0.5.3", "description": "OpenClaw provider plugin — 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/requester-runtime.test.ts b/test/requester-runtime.test.ts index e1614b1..28eac29 100644 --- a/test/requester-runtime.test.ts +++ b/test/requester-runtime.test.ts @@ -50,7 +50,7 @@ test("Direction-2 metadata has no requester bridge descriptors or raw local path TAAS_REQUESTER_BRIDGE_LEASE_URL: "http://127.0.0.1:9/internal/requester-bridges/leases", }) try { - const payload = await runPayload(captureProvider(plugin), { messages: [], metadata: {} }) + const payload = await runPayload(captureProvider(plugin), { messages: [], metadata: {} }, { sessionId: "local-runtime-test" }) const runtime = payload.metadata.requester_runtime assert.equal("available_bridges" in runtime, false) assert.equal("capture_mode" in runtime, false) diff --git a/test/smoke.mjs b/test/smoke.mjs index f67d0ce..73cce2f 100644 --- a/test/smoke.mjs +++ b/test/smoke.mjs @@ -43,40 +43,10 @@ assert.equal(typeof wrapped, "function") await wrapped("model", { messages: [] }, {}) assert.equal(capturedPayload.metadata.existing, "keep") -assert.match(capturedPayload.metadata.session_id, /^oc:[a-f0-9]{16}$/) -assert.equal(capturedPayload.metadata.sticky_key, capturedPayload.metadata.session_id) -assert.equal( - capturedPayload.metadata.requester_runtime.source, - "openclaw-taas-affinity" -) -assert.equal( - capturedPayload.metadata.requester_runtime.session_key, - capturedPayload.metadata.session_id -) -assert.equal( - capturedPayload.metadata.requester_runtime.provider, - "cloudsigma" -) -assert.equal( - capturedPayload.metadata.requester_runtime.model_id, - "cloudsigma/test-model" -) -assert.equal( - capturedPayload.metadata.requester_runtime.redaction_policy, - "no_secrets;no_raw_local_paths;no_env_values;no_git_remotes;no_status_or_diffs;no_extra_params" -) -assert.equal( - capturedPayload.metadata.requester_runtime.tool_execution, - "direction_2_gateway" -) -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.5.2") -assert.equal(capturedPayload.metadata.openclaw_correlation.session_id, capturedPayload.metadata.session_id) -assert.equal(capturedPayload.metadata.openclaw_correlation.sticky_key, capturedPayload.metadata.session_id) -assert.equal(capturedPayload.metadata.openclaw_correlation.provider, "cloudsigma") -assert.equal(capturedPayload.metadata.openclaw_correlation.model_id, "cloudsigma/test-model") +assert.equal("session_id" in capturedPayload.metadata, false) +assert.equal("sticky_key" in capturedPayload.metadata, false) +assert.equal("requester_runtime" in capturedPayload.metadata, false) +assert.equal("openclaw_correlation" in capturedPayload.metadata, false) const transportState = provider.resolveTransportTurnState({ provider: "cloudsigma", @@ -88,10 +58,58 @@ const transportState = provider.resolveTransportTurnState({ assert.match(transportState.headers["X-Session-Id"], /^oc:[a-f0-9]{16}$/) assert.equal(transportState.headers["X-OpenClaw-Session-Id"], transportState.headers["X-Session-Id"]) -assert.equal(transportState.headers["X-OpenClaw-Plugin-Version"], "0.5.2") +assert.equal(transportState.headers["X-OpenClaw-Plugin-Version"], "0.5.3") assert.equal(transportState.headers["X-OpenClaw-Turn-Id"], "turn-smoke") assert.equal(transportState.headers["X-OpenClaw-Attempt"], "1") +const localSessionWrapped = provider.wrapStreamFn({ + streamFn, + sessionId: "local-session-a", + workspaceDir: "/tmp/openclaw-token-cache-optimizer-smoke", + provider: "cloudsigma", + modelId: "cloudsigma/test-model", + model: { id: "cloudsigma/test-model" }, +}) +await localSessionWrapped("model", { messages: [] }, {}) +const localSessionA = capturedPayload.metadata.session_id +assert.match(localSessionA, /^oc:[a-f0-9]{16}$/) +assert.equal(capturedPayload.metadata.sticky_key, localSessionA) +assert.equal(capturedPayload.metadata.requester_runtime.source, "openclaw-taas-affinity") +assert.equal(capturedPayload.metadata.requester_runtime.session_key, localSessionA) +assert.equal(capturedPayload.metadata.requester_runtime.provider, "cloudsigma") +assert.equal(capturedPayload.metadata.requester_runtime.model_id, "cloudsigma/test-model") +assert.equal(capturedPayload.metadata.requester_runtime.redaction_policy, "no_secrets;no_raw_local_paths;no_env_values;no_git_remotes;no_status_or_diffs;no_extra_params") +assert.equal(capturedPayload.metadata.requester_runtime.tool_execution, "direction_2_gateway") +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.5.3") +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") +assert.equal(capturedPayload.metadata.openclaw_correlation.model_id, "cloudsigma/test-model") +await provider.wrapStreamFn({ + streamFn, + sessionId: "local-session-b", + workspaceDir: "/tmp/openclaw-token-cache-optimizer-smoke", + provider: "cloudsigma", + modelId: "cloudsigma/test-model", + model: { id: "cloudsigma/test-model" }, +})("model", { messages: [] }, {}) +assert.notEqual(capturedPayload.metadata.session_id, localSessionA) +assert.equal(capturedPayload.metadata.openclaw_correlation.session_identity_scope, "local_session") +assert.equal(capturedPayload.metadata.requester_runtime.session_identity_scope, "local_session") + +const localTransportState = provider.resolveTransportTurnState({ + provider: "cloudsigma", + modelId: "cloudsigma/test-model", + sessionId: "local-session-a", + turnId: "turn-local", + attempt: 1, + transport: "stream", +}) +assert.equal(localTransportState.headers["X-Session-Id"], localSessionA) + console.log("smoke ok") // === autorouter capture (R7.1/R7.2 from Studio PRD) ===