diff --git a/README.md b/README.md index 5bc0df0..f01e70f 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,10 @@ This plugin is intentionally narrow after the Claude Code Direction-2 lane updat For requests routed through the `cloudsigma` or `cloudsigma-staging` provider IDs, the plugin: -- passes OpenClaw's native `ctx.sessionId` through unchanged when available +- resolves identity in strict order: invocation `options.sessionId`, wrapper `ctx.sessionId`, exact trace bridge, explicit legacy environment fallback +- records authoritative `model_call_started` session identity against the exact W3C `traceId` + `spanId` exposed through public `ctx.trace` +- resolves generic/provider calls from the matching `StreamOptions.headers.traceparent` when direct session identity is absent +- keeps the trace bridge bounded (1,024 entries), short-lived (30-minute sliding TTL), and fail-closed for malformed or ambiguous correlation - generates a stable `oc:` only as a deprecated compatibility fallback from `OPENCLAW_SESSION_ID` - injects `metadata.session_id` when absent - injects `metadata.sticky_key` when absent @@ -18,6 +21,7 @@ For requests routed through the `cloudsigma` or `cloudsigma-staging` provider ID - injects `metadata.openclaw_correlation` for request/run tracing - captures TaaS autorouter + request/trace response headers - exposes the latest route capture via gateway method `taas.autorouter.lastRoute` +- exposes privacy-safe bridge outcome counters via `taas.affinity.stats` (hits, misses, expiries, ambiguous traces, and direct invocation IDs); no trace or session values are returned ## Startup compatibility @@ -34,6 +38,12 @@ The manifest explicitly asks OpenClaw to import the plugin at gateway startup: This is required because gateway RPC handlers must be attached during gateway startup. Provider/lazy activation is not enough for `taas.autorouter.lastRoute` to be present in the live gateway dispatch table. +## Trace bridge compatibility + +On OpenClaw versions that expose the public `model_call_started` lifecycle hook, the plugin feature-detects `api.on`, records an authoritative non-empty `ctx.sessionId` (while checking `event.sessionId` for consistency), and requires an exact valid W3C trace/span match in the later provider invocation. It never uses timing, agent ID, workspace, session key, or a process-global "current session" for correlation. Exact successful matches refresh a 30-minute sliding bridge TTL so delayed retries remain safe; TaaS retains the resulting session affinity independently for seven days. + +Older OpenClaw versions without this hook continue to work when `options.sessionId`, wrapper `ctx.sessionId`, or the explicit legacy `OPENCLAW_SESSION_ID` is available. Calls without one of those strong identities remain affinity-less. + ## Request metadata Example injected metadata: @@ -66,7 +76,7 @@ Example injected metadata: "openclaw_correlation": { "schema_version": "2026-06-05", "source": "openclaw-taas-affinity", - "plugin_version": "0.6.0", + "plugin_version": "0.11.0", "session_id": "oc:0123456789abcdef", "sticky_key": "oc:0123456789abcdef", "session_source_hint": "source:1a2b3c4d5e6f7890", @@ -78,7 +88,7 @@ Example injected metadata: } ``` -All metadata fields are no-overwrite. If the caller already supplied `metadata.session_id`, `metadata.sticky_key`, or `metadata.requester_runtime`, the plugin leaves them intact. +All metadata fields, including `openclaw_correlation`, are no-overwrite. If the caller already supplied `metadata.session_id`, `metadata.sticky_key`, `metadata.requester_runtime`, or `metadata.openclaw_correlation`, the plugin leaves them intact. The plugin does not include raw local paths (`workspace_dir`, `agent_dir`, `repo_root_hint`), environment variables, tokens, git remotes, full git status output, diffs, or arbitrary provider `extraParams`. @@ -178,6 +188,12 @@ npm run build Current tests cover: +- direct GPT invocation identity and strict precedence +- Kimi and generic CloudSigma trace-bridge identity +- simple-completion trace bridging +- concurrent traces, same-trace retries, and subagent isolation +- malformed, absent, expired, oversized, or ambiguous trace state failing closed +- graceful operation when the lifecycle hook is unavailable - manifest startup activation - provider hook registration for `cloudsigma` and `cloudsigma-staging` - metadata/header injection diff --git a/index.ts b/index.ts index 6c80595..e58c67b 100644 --- a/index.ts +++ b/index.ts @@ -29,11 +29,10 @@ 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.10.0" +const PLUGIN_VERSION = "0.11.0" +const TRACE_BRIDGE_TTL_MS = 30 * 60 * 1000 +const TRACE_BRIDGE_LIMIT = 1024 -// OpenClaw stores active registry state (including workspaceDir) on globalThis -// under this well-known symbol key. -const PLUGIN_REGISTRY_STATE = Symbol.for("openclaw.pluginRegistryState") const isDev = process.env.NODE_ENV === "development" || Boolean(process.env.OPENCLAW_DEBUG) type RequesterRuntime = Record @@ -62,6 +61,236 @@ function safeString(value: unknown): string | undefined { return typeof value === "string" && value.trim().length > 0 ? value.trim() : undefined } +type TraceBridgeDiagnostic = { + runId?: string + callId?: string +} + +type TraceBridgeEntry = TraceBridgeDiagnostic & { + sessionId: string | null + ambiguous: boolean + expiresAt: number +} + +type TraceBridgeStats = { + hit: number + miss: number + expired: number + ambiguous: number + directOptionsSessionId: number +} + +const traceBridgeStats: TraceBridgeStats = { + hit: 0, + miss: 0, + expired: 0, + ambiguous: 0, + directOptionsSessionId: 0, +} + +function resetTraceBridgeStats(): void { + for (const key of Object.keys(traceBridgeStats) as Array) { + traceBridgeStats[key] = 0 + } +} + +/** + * Short-lived, bounded correlation between OpenClaw's public model-call hook + * and the later provider invocation. Entries are deliberately non-consuming: + * a transport may retry or invoke multiple callbacks with the same exact span. + */ +class TraceSessionBridge { + private readonly entries = new Map() + private readonly limit: number + private readonly ttlMs: number + private readonly now: () => number + + constructor( + limit = TRACE_BRIDGE_LIMIT, + ttlMs = TRACE_BRIDGE_TTL_MS, + now: () => number = Date.now, + ) { + this.limit = limit + this.ttlMs = ttlMs + this.now = now + } + + private pruneExpired(now = this.now()): void { + for (const [key, entry] of this.entries) { + if (entry.expiresAt <= now) this.entries.delete(key) + } + } + + private enforceLimit(): void { + while (this.entries.size > this.limit) { + const oldest = this.entries.keys().next().value as string | undefined + if (!oldest) return + this.entries.delete(oldest) + } + } + + record(key: string, sessionId: string, diagnostic: TraceBridgeDiagnostic = {}): void { + const now = this.now() + this.pruneExpired(now) + const existing = this.entries.get(key) + const ambiguous = existing?.ambiguous === true || + (Boolean(existing?.sessionId) && existing?.sessionId !== sessionId) + const entry: TraceBridgeEntry = { + sessionId: ambiguous ? null : sessionId, + ambiguous, + expiresAt: now + this.ttlMs, + ...diagnostic, + } + this.entries.delete(key) + this.entries.set(key, entry) + this.enforceLimit() + } + + markAmbiguous(key: string, diagnostic: TraceBridgeDiagnostic = {}): void { + const now = this.now() + this.pruneExpired(now) + this.entries.delete(key) + this.entries.set(key, { + sessionId: null, + ambiguous: true, + expiresAt: now + this.ttlMs, + ...diagnostic, + }) + this.enforceLimit() + } + + resolve(key: string): string | undefined { + const now = this.now() + const entry = this.entries.get(key) + if (!entry) { + traceBridgeStats.miss += 1 + this.pruneExpired(now) + return undefined + } + if (entry.expiresAt <= now) { + this.entries.delete(key) + traceBridgeStats.expired += 1 + this.pruneExpired(now) + return undefined + } + if (entry.ambiguous || !entry.sessionId) { + traceBridgeStats.ambiguous += 1 + return undefined + } + + // Successful exact correlation refreshes the handoff window. Keep the + // entry non-consuming so delayed retries and concurrent provider calls do + // not race, while moving it to the newest position for bounded eviction. + entry.expiresAt = now + this.ttlMs + this.entries.delete(key) + this.entries.set(key, entry) + traceBridgeStats.hit += 1 + return entry.sessionId + } + + clear(resetStats = true): void { + this.entries.clear() + if (resetStats) resetTraceBridgeStats() + } + + get stats(): Readonly { + return { ...traceBridgeStats } + } + + get ttlMsValue(): number { + return this.ttlMs + } + + get limitValue(): number { + return this.limit + } + + get size(): number { + this.pruneExpired() + return this.entries.size + } +} + +const traceSessionBridge = new TraceSessionBridge() + +function isNonZeroHex(value: string): boolean { + return /[1-9a-f]/i.test(value) +} + +function traceKeyFromContext(value: unknown): string | undefined { + const trace = asRecord(value) + const traceId = safeString(trace?.traceId)?.toLowerCase() + const spanId = safeString(trace?.spanId)?.toLowerCase() + if (!traceId || !/^[0-9a-f]{32}$/.test(traceId) || !isNonZeroHex(traceId)) return undefined + if (!spanId || !/^[0-9a-f]{16}$/.test(spanId) || !isNonZeroHex(spanId)) return undefined + return `${traceId}:${spanId}` +} + +function traceKeyFromTraceparent(value: unknown): string | undefined { + const traceparent = safeString(value) + if (!traceparent) return undefined + // OpenClaw currently emits canonical W3C version 00. Reject extensions and + // malformed/future forms rather than risk a weak or ambiguous correlation. + const match = /^00-([0-9a-fA-F]{32})-([0-9a-fA-F]{16})-([0-9a-fA-F]{2})$/.exec(traceparent) + if (!match) return undefined + return traceKeyFromContext({ traceId: match[1], spanId: match[2] }) +} + +function traceKeyFromHeaders(value: unknown): string | undefined { + const headers = asRecord(value) + if (!headers) return undefined + let candidate: string | undefined + for (const [name, raw] of Object.entries(headers)) { + if (name.toLowerCase() !== "traceparent") continue + if (typeof raw !== "string") return undefined + const normalized = raw.trim().toLowerCase() + if (candidate !== undefined && candidate !== normalized) return undefined + candidate = normalized + } + return traceKeyFromTraceparent(candidate) +} + +function recordModelCallStarted(event: unknown, ctx: unknown): void { + const eventRecord = asRecord(event) + const ctxRecord = asRecord(ctx) + const traceKey = traceKeyFromContext(ctxRecord?.trace) + if (!traceKey) return + + // ctx.sessionId is the authoritative logical-session field in the public + // lifecycle contract. event.sessionId is checked only for consistency; it + // must never substitute for a missing authoritative context identity. + const contextSessionId = safeString(ctxRecord?.sessionId) + if (!contextSessionId) return + const eventSessionId = safeString(eventRecord?.sessionId) + + const diagnostic = { + runId: safeString(ctxRecord?.runId) ?? safeString(eventRecord?.runId), + callId: safeString(eventRecord?.callId), + } + if (eventSessionId && eventSessionId !== contextSessionId) { + traceSessionBridge.markAmbiguous(traceKey, diagnostic) + return + } + traceSessionBridge.record(traceKey, contextSessionId, diagnostic) +} + +function registerTraceSessionBridgeHook(api: OpenClawPluginApi): boolean { + const on = (api as unknown as { on?: unknown }).on + if (typeof on !== "function") return false + try { + ;(on as (name: string, handler: (event: unknown, ctx: unknown) => void) => void).call( + api, + "model_call_started", + recordModelCallStarted, + ) + return true + } catch { + // Older runtimes may expose api.on without this lifecycle hook. Direct + // options.sessionId/ctx.sessionId affinity remains fully functional. + return false + } +} + function stableHash(value: string, prefix: string, length = 16): string { const hex = createHash("sha256").update(value, "utf8").digest("hex") return `${prefix}:${hex.slice(0, length)}` @@ -77,35 +306,39 @@ type ResolvedSessionIdentity = { source: string sourceHint: string localSessionScoped: boolean - identityMode: "native" | "legacy_env" + identityMode: "native" | "trace_bridge" | "legacy_env" } /** - * Prefer OpenClaw's native conversation identity exactly as supplied by the - * current provider runtime contract. Generation is retained only for older - * runtimes that expose OPENCLAW_SESSION_ID but do not populate ctx.sessionId. - * - * We intentionally do not derive identity from workspace, agent, run, or state - * paths: those scopes can outlive a conversation and accidentally join two - * otherwise independent sessions in TaaS. + * Resolve only authoritative per-conversation identities. Provider invocation + * options win, followed by wrapper context, exact trace/span correlation, and + * finally the explicit legacy environment compatibility value. */ function resolveSessionIdentity( workspaceDirFromCtx?: string, + sessionIdFromOptions?: unknown, sessionIdFromCtx?: unknown, + sessionIdFromTrace?: unknown, agentIdFromCtx?: unknown, ): ResolvedSessionIdentity | null { - const nativeSessionId = safeString(sessionIdFromCtx) const sourceHint = workspaceDirFromCtx ? `workspaceDir:${workspaceDirFromCtx}` : `stateDir:${process.env.OPENCLAW_STATE_DIR ?? path.join(os.homedir(), ".openclaw")}` - - if (nativeSessionId) { - return { - sessionId: nativeSessionId, - source: "openclaw:ctx.sessionId", - sourceHint, - localSessionScoped: true, - identityMode: "native", + const candidates = [ + { value: sessionIdFromOptions, source: "openclaw:options.sessionId", identityMode: "native" as const }, + { value: sessionIdFromCtx, source: "openclaw:ctx.sessionId", identityMode: "native" as const }, + { value: sessionIdFromTrace, source: "openclaw:model_call_started.trace", identityMode: "trace_bridge" as const }, + ] + for (const candidate of candidates) { + const sessionId = safeString(candidate.value) + if (sessionId) { + return { + sessionId, + source: candidate.source, + sourceHint, + localSessionScoped: true, + identityMode: candidate.identityMode, + } } } @@ -120,13 +353,9 @@ function resolveSessionIdentity( } } - // 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. + // Agent and workspace scopes outlive conversations and are never identity + // fallbacks. Keep the parameter only for compatibility with existing callers. void agentIdFromCtx - return null } @@ -234,7 +463,7 @@ function buildCorrelationMetadata( session_id: sessionId, sticky_key: sessionId, session_source_hint: stableHash(sourceHint, "source"), - session_identity_scope: source === "openclaw:ctx.sessionId" ? "native_openclaw_session" : "legacy_generated_session", + session_identity_scope: source === "openclaw:env.OPENCLAW_SESSION_ID" ? "legacy_generated_session" : "native_openclaw_session", ...(agentId && { agent_id: agentId }), ...(provider && { provider }), ...(modelId && { model_id: modelId }), @@ -286,7 +515,7 @@ function buildRequesterRuntime( ...(provider && { provider }), ...(modelId && { model_id: modelId }), session_source_hint: stableHash(sourceHint, "source"), - session_identity_scope: source === "openclaw:ctx.sessionId" ? "native_openclaw_session" : "legacy_generated_session", + session_identity_scope: source === "openclaw:env.OPENCLAW_SESSION_ID" ? "legacy_generated_session" : "native_openclaw_session", tool_execution: "direction_2_gateway", metadata_classification: { identifiers: "hashed", @@ -380,7 +609,7 @@ function captureAutorouterFromHeaders( if (isDev) { console.debug( - `[taas-affinity] captured autorouter sessionId=${sessionId} ` + + `[taas-affinity] captured autorouter identityRef=${stableHash(sessionId, "session")} ` + `model=${capture.autorouterModel} algo=${capture.autorouterAlgo} ` + `source=${capture.autorouterAlgoSource} thinking=${capture.thinkingApplied} ` + `ctxWindow=${capture.routedContextWindow}` @@ -403,56 +632,88 @@ function buildWrapper(ctx: ProviderWrapStreamFnContext) { 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 ?? + const optionRecord = options as unknown as Record | undefined + const traceKey = traceKeyFromHeaders(optionRecord?.headers) + const directOptionsSessionId = safeString(optionRecord?.sessionId) + if (directOptionsSessionId) traceBridgeStats.directOptionsSessionId += 1 + let resolvedIdentity: ResolvedSessionIdentity | null | undefined + const getIdentity = (): ResolvedSessionIdentity | null => { + if (resolvedIdentity) return resolvedIdentity + const bridgeSessionId = traceKey ? traceSessionBridge.resolve(traceKey) : undefined + const identity = resolveSessionIdentity( + ctx.workspaceDir, + optionRecord?.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"}`, + bridgeSessionId, + (ctx as { agentId?: unknown }).agentId, ) + // Cache successful resolution in this invocation only. Leave misses + // uncached because model_call_started is intentionally fire-and-forget + // and may populate the exact trace bridge before payload construction. + if (identity) resolvedIdentity = identity + return identity } const prevOnPayload = options?.onPayload const onPayload: NonNullable["onPayload"] = async (payload, payloadModel) => { const payloadRecord = asRecord(payload) if (!payloadRecord) return prevOnPayload ? prevOnPayload(payload, payloadModel) : payload - const patched = identity - ? patchPayloadMetadata(payloadRecord, identity.sessionId, requesterRuntime, correlation, true) - : payloadRecord + if ( + traceKey && + !safeString(optionRecord?.sessionId) && + !safeString((ctx as { sessionId?: unknown }).sessionId) + ) { + // Core dispatches observational model-call hooks through a queued + // microtask before invoking the provider stream. Yield once so the + // synchronous hook handler can publish this exact trace mapping even + // when a transport constructs its payload immediately. This does not + // correlate by time: only the exact validated trace/span can resolve. + await Promise.resolve() + } + const identity = getIdentity() + if (!identity) return prevOnPayload ? prevOnPayload(payloadRecord, payloadModel) : payloadRecord + const agentIdForCapture = resolveAgentIdentity( + ctx as { agentDir?: string; workspaceDir?: string }, + identity.sessionId, + ) + const requesterRuntime = buildRequesterRuntime( + ctx, + identity.sessionId, + identity.source, + identity.sourceHint, + ) + const correlation = buildCorrelationMetadata( + identity.sessionId, + identity.source, + identity.sourceHint, + ctx, + agentIdForCapture, + ) + const patched = patchPayloadMetadata( + payloadRecord, + identity.sessionId, + requesterRuntime, + correlation, + true, + ) + if (isDev) { + console.debug( + `[taas-affinity] stream identityRef=${stableHash(identity.sessionId, "session")} mode=${identity.identityMode}`, + ) + } return prevOnPayload ? prevOnPayload(patched, payloadModel) : patched } const prevOnResponse = options?.onResponse const onResponse: NonNullable["onResponse"] = async (response, responseModel) => { try { - identity && captureAutorouterFromHeaders(identity.sessionId, response?.headers ?? {}, agentIdForCapture) + const identity = getIdentity() + if (identity) { + const agentIdForCapture = resolveAgentIdentity( + ctx as { agentDir?: string; workspaceDir?: string }, + identity.sessionId, + ) + captureAutorouterFromHeaders(identity.sessionId, response?.headers ?? {}, agentIdForCapture) + } } catch (err) { if (isDev) console.debug(`[taas-affinity] onResponse capture failed: ${(err as Error)?.message ?? err}`) } @@ -466,10 +727,12 @@ function buildTransportTurnState(ctx: ProviderResolveTransportTurnStateContext): const identity = resolveSessionIdentity( undefined, (ctx as { sessionId?: unknown }).sessionId, + undefined, + undefined, (ctx as { agentId?: unknown }).agentId, ) if (!identity) { - if (isDev) console.debug(`[taas-affinity] no native or legacy session identity; skipping affinity headers turnId=${ctx.turnId}`) + if (isDev) console.debug("[taas-affinity] no strong session identity; skipping affinity headers") return null } const agentId = resolveAgentIdentity( @@ -478,8 +741,8 @@ function buildTransportTurnState(ctx: ProviderResolveTransportTurnStateContext): ) if (isDev) { console.debug( - `[taas-affinity] resolveTransportTurnState sessionId=${identity.sessionId} ` + - `mode=${identity.identityMode} turnId=${ctx.turnId} attempt=${ctx.attempt}` + `[taas-affinity] resolveTransportTurnState identityRef=${stableHash(identity.sessionId, "session")} ` + + `mode=${identity.identityMode} attempt=${ctx.attempt}` ) } return { headers: buildCorrelationHeaders({ sessionId: identity.sessionId, turnId: ctx.turnId, attempt: ctx.attempt, agentId }) } @@ -494,6 +757,14 @@ export default { deriveAgentIdForCapture, buildCorrelationMetadata, buildCorrelationHeaders, + TraceSessionBridge, + traceKeyFromContext, + traceKeyFromTraceparent, + traceKeyFromHeaders, + recordModelCallStarted, + traceSessionBridge, + traceBridgeStats, + resetTraceBridgeStats, }, id: "openclaw-taas-affinity", name: "CloudSigma TaaS Token Cache Optimizer", @@ -502,6 +773,8 @@ export default { "pin sessions to the same upstream slot from turn 1, maximising prompt-cache hit rates.", register(api: OpenClawPluginApi) { + registerTraceSessionBridgeHook(api) + // The runtime supports wrapSimpleCompletionStreamFn, but the installed // plugin-sdk ProviderPlugin declaration lags that optional hook. Keep the // compatibility cast scoped to this registration object. @@ -548,6 +821,22 @@ export default { { scope: "operator.write" } ) + if (typeof api.registerGatewayMethod === "function") api.registerGatewayMethod( + "taas.affinity.stats", + async ({ respond }) => { + respond(true, { + pluginVersion: PLUGIN_VERSION, + bridge: { + ttlMs: traceSessionBridge.ttlMsValue, + limit: traceSessionBridge.limitValue, + size: traceSessionBridge.size, + }, + counters: traceSessionBridge.stats, + }) + }, + { scope: "operator.read" }, + ) + if (typeof api.registerGatewayMethod === "function") api.registerGatewayMethod( "taas.autorouter.lastRoute", async ({ params, respond }) => { @@ -568,7 +857,7 @@ export default { const resolvedIdentity = directSessionId ? null - : resolveSessionIdentity(workspaceDir, safeString(pp.localSessionId)) + : resolveSessionIdentity(workspaceDir, safeString(pp.localSessionId), undefined, undefined, undefined) const resolvedSessionId = directSessionId ?? resolvedIdentity?.sessionId ?? null respond(true, { sessionId: resolvedSessionId, @@ -589,5 +878,13 @@ export default { buildCorrelationMetadata, getLastRouteForAgent, getLastRouteForSession, + TraceSessionBridge, + traceKeyFromContext, + traceKeyFromTraceparent, + traceKeyFromHeaders, + recordModelCallStarted, + traceSessionBridge, + traceBridgeStats, + resetTraceBridgeStats, }, } diff --git a/package-lock.json b/package-lock.json index 7eaad22..f64c9d1 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "openclaw-taas-affinity", - "version": "0.10.0", + "version": "0.11.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "openclaw-taas-affinity", - "version": "0.10.0", + "version": "0.11.0", "license": "MIT", "devDependencies": { "@types/node": "^22.0.0", diff --git a/package.json b/package.json index 030e25f..1605145 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "openclaw-taas-affinity", - "version": "0.10.0", + "version": "0.11.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 a1a5d39..df13383 100644 --- a/test/agent-scoped-session.test.ts +++ b/test/agent-scoped-session.test.ts @@ -7,30 +7,32 @@ const internals = (plugin as unknown as { __test__?: Record }). type Resolve = ( workspaceDir?: string, - sessionId?: unknown, + optionsSessionId?: unknown, + contextSessionId?: unknown, + traceSessionId?: unknown, agentId?: unknown, ) => { sessionId: string; identityMode: string; source: string } | null test("agent id alone does not fabricate a conversation identity", () => { assert.ok(internals) const resolve = internals!.resolveSessionIdentity as Resolve - assert.equal(resolve("/home/x/.openclaw/workspace", undefined, "main"), null) + assert.equal(resolve("/home/x/.openclaw/workspace", undefined, undefined, undefined, "main"), null) }) test("workspace and agent scopes do not merge unrelated conversations", () => { const resolve = internals!.resolveSessionIdentity as Resolve - assert.equal(resolve("/ws/a", undefined, "main"), null) - assert.equal(resolve("/ws/b", undefined, "main"), null) + assert.equal(resolve("/ws/a", undefined, undefined, undefined, "main"), null) + assert.equal(resolve("/ws/b", undefined, undefined, undefined, "main"), null) }) test("a native session id always takes precedence", () => { const resolve = internals!.resolveSessionIdentity as Resolve - const r = resolve("/home/x/workspace", "agent:main:main", "main") + const r = resolve("/home/x/workspace", "agent:main:main", "stale-context", "trace-session", "main") assert.equal(r!.identityMode, "native") assert.equal(r!.sessionId, "agent:main:main") }) 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) + assert.equal(resolve("/home/x/workspace", undefined, undefined, undefined, undefined), null) }) diff --git a/test/requester-runtime.test.ts b/test/requester-runtime.test.ts index 6c07ce0..be7c207 100644 --- a/test/requester-runtime.test.ts +++ b/test/requester-runtime.test.ts @@ -153,3 +153,32 @@ test("no session identity means no affinity injection", async () => { restore() } }) + + +test("privacy-safe affinity stats gateway method exposes counters without identities", async () => { + const { plugin, restore } = await loadPlugin() + try { + const methods = new Map() + plugin.register({ + registerProvider() {}, + registerGatewayMethod(name: string, handler: any) { methods.set(name, handler) }, + runtime: { system: { enqueueSystemEvent: () => true, requestHeartbeat: () => {} } }, + }) + const handler = methods.get("taas.affinity.stats") + assert.equal(typeof handler, "function") + let response: any + await handler({ params: {}, respond(ok: boolean, payload: any) { response = { ok, payload } } }) + assert.equal(response.ok, true) + assert.equal(response.payload.pluginVersion, "0.11.0") + assert.equal(response.payload.bridge.ttlMs, 30 * 60 * 1000) + assert.equal(response.payload.bridge.limit, 1024) + assert.deepEqual(Object.keys(response.payload.counters).sort(), [ + "ambiguous", "directOptionsSessionId", "expired", "hit", "miss", + ]) + const encoded = JSON.stringify(response) + assert.equal(encoded.includes("traceId"), false) + assert.equal(encoded.includes("sessionId"), false) + } finally { + restore() + } +}) diff --git a/test/smoke.mjs b/test/smoke.mjs index acb4847..bf84636 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.10.0") +assert.equal(transportState.headers["X-OpenClaw-Plugin-Version"], "0.11.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.10.0") +assert.equal(capturedPayload.metadata.openclaw_correlation.plugin_version, "0.11.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") diff --git a/test/trace-session-bridge.test.ts b/test/trace-session-bridge.test.ts new file mode 100644 index 0000000..6732059 --- /dev/null +++ b/test/trace-session-bridge.test.ts @@ -0,0 +1,318 @@ +import assert from "node:assert/strict" +import test from "node:test" + +import plugin from "../index.ts" + +const internals = (plugin as any).__test__ as Record +const TRACE_A = { traceId: "11111111111111111111111111111111", spanId: "aaaaaaaaaaaaaaaa" } +const TRACE_B = { traceId: "22222222222222222222222222222222", spanId: "bbbbbbbbbbbbbbbb" } +const TRACE_C = { traceId: "33333333333333333333333333333333", spanId: "cccccccccccccccc" } + +function traceparent(trace: { traceId: string; spanId: string }): string { + return `00-${trace.traceId}-${trace.spanId}-01` +} + +function captureProvider(withHook = true) { + let provider: any + let modelCallStarted: ((event: unknown, ctx: unknown) => void) | undefined + const api: any = { + registerProvider(candidate: any) { provider = candidate }, + registerGatewayMethod() {}, + runtime: { system: { enqueueSystemEvent: () => true, requestHeartbeat: () => {} } }, + } + if (withHook) { + api.on = (name: string, handler: (event: unknown, ctx: unknown) => void) => { + assert.equal(name, "model_call_started") + modelCallStarted = handler + } + } + ;(plugin as any).register(api) + assert.ok(provider) + return { provider, modelCallStarted } +} + +function createModel(id: string) { + return { + provider: "cloudsigma", + id, + api: "openai-completions", + baseUrl: "https://api.cloudsigma.com/ai/v1", + } +} + +async function invoke( + provider: any, + modelId: string, + options: Record, + simple = false, + ctx: Record = {}, +): Promise { + let payload: any + const wrapper = (simple ? provider.wrapSimpleCompletionStreamFn : provider.wrapStreamFn)({ + provider: "cloudsigma", + modelId, + workspaceDir: "/workspace/ignored-for-identity", + agentId: "main", + streamFn: async (model: unknown, _context: unknown, streamOptions: any) => { + payload = await streamOptions.onPayload({ messages: [], metadata: {} }, model) + return payload + }, + ...ctx, + }) + await wrapper(createModel(modelId), { messages: [] }, options) + return payload +} + +function start( + hook: ((event: unknown, ctx: unknown) => void) | undefined, + trace: { traceId: string; spanId: string }, + sessionId?: string, + extraEvent: Record = {}, + extraCtx: Record = {}, +) { + assert.ok(hook) + hook( + { runId: "run-1", callId: "call-1", sessionId, provider: "cloudsigma", model: "model", ...extraEvent }, + { runId: "run-1", sessionId, trace, ...extraCtx }, + ) +} + +test.beforeEach(() => { + internals.traceSessionBridge.clear() + delete process.env.OPENCLAW_SESSION_ID +}) + +test("GPT direct options identity has precedence over context and trace bridge", async () => { + const { provider, modelCallStarted } = captureProvider() + start(modelCallStarted, TRACE_A, "trace-session") + const payload = await invoke(provider, "gpt-5.6-sol", { + sessionId: "options-session", + headers: { traceparent: traceparent(TRACE_A) }, + }, false, { sessionId: "context-session" }) + assert.equal(payload.metadata.session_id, "options-session") + assert.equal(internals.traceSessionBridge.stats.directOptionsSessionId, 1) +}) + +test("Kimi and another generic CloudSigma model resolve exact trace bridge identity", async () => { + const { provider, modelCallStarted } = captureProvider() + for (const [modelId, trace, sessionId] of [ + ["kimi-k2", TRACE_A, "kimi-session"], + ["glm-5", TRACE_B, "glm-session"], + ] as const) { + start(modelCallStarted, trace, sessionId) + const payload = await invoke(provider, modelId, { + headers: { traceparent: traceparent(trace) }, + }) + assert.equal(payload.metadata.session_id, sessionId, modelId) + assert.equal(payload.metadata.openclaw_correlation.model_id, modelId, modelId) + } +}) + +test("queued model_call_started hook wins the immediate payload-construction race", async () => { + const { provider, modelCallStarted } = captureProvider() + queueMicrotask(() => start(modelCallStarted, TRACE_A, "queued-hook-session")) + const payload = await invoke(provider, "kimi-k2", { + headers: { traceparent: traceparent(TRACE_A) }, + }) + assert.equal(payload.metadata.session_id, "queued-hook-session") +}) + +test("simple completion resolves exact trace bridge identity", async () => { + const { provider, modelCallStarted } = captureProvider() + start(modelCallStarted, TRACE_A, "simple-trace-session") + const payload = await invoke(provider, "kimi-k2", { + headers: { traceparent: traceparent(TRACE_A) }, + }, true) + assert.equal(payload.metadata.session_id, "simple-trace-session") +}) + +test("concurrent distinct traces remain isolated", async () => { + const { provider, modelCallStarted } = captureProvider() + start(modelCallStarted, TRACE_A, "session-alpha") + start(modelCallStarted, TRACE_B, "session-beta") + const [alpha, beta] = await Promise.all([ + invoke(provider, "kimi-k2", { headers: { traceparent: traceparent(TRACE_A) } }), + invoke(provider, "glm-5", { headers: { traceparent: traceparent(TRACE_B) } }), + ]) + assert.equal(alpha.metadata.session_id, "session-alpha") + assert.equal(beta.metadata.session_id, "session-beta") +}) + +test("same exact trace supports retries without consume races", async () => { + const { provider, modelCallStarted } = captureProvider() + start(modelCallStarted, TRACE_A, "retry-session") + const results = await Promise.all([ + invoke(provider, "kimi-k2", { headers: { traceparent: traceparent(TRACE_A) } }), + invoke(provider, "kimi-k2", { headers: { traceparent: traceparent(TRACE_A) } }), + invoke(provider, "kimi-k2", { headers: { traceparent: traceparent(TRACE_A) } }), + ]) + assert.deepEqual(results.map((payload) => payload.metadata.session_id), [ + "retry-session", + "retry-session", + "retry-session", + ]) +}) + +test("parent and subagent traces preserve distinct authoritative sessions", async () => { + const { provider, modelCallStarted } = captureProvider() + start(modelCallStarted, TRACE_A, "agent:main:main") + start(modelCallStarted, TRACE_B, "agent:main:subagent:child-1") + const parent = await invoke(provider, "kimi-k2", { headers: { traceparent: traceparent(TRACE_A) } }) + const child = await invoke(provider, "kimi-k2", { headers: { traceparent: traceparent(TRACE_B) } }) + assert.equal(parent.metadata.session_id, "agent:main:main") + assert.equal(child.metadata.session_id, "agent:main:subagent:child-1") +}) + +test("malformed, absent, partial, and mismatched trace data fail closed", async () => { + const { provider, modelCallStarted } = captureProvider() + start(modelCallStarted, TRACE_A, "must-not-leak") + const optionCases = [ + {}, + { headers: {} }, + { headers: { traceparent: "garbage" } }, + { headers: { traceparent: `00-${TRACE_A.traceId}-0000000000000000-01` } }, + { headers: { traceparent: traceparent(TRACE_C) } }, + { headers: { traceparent: traceparent(TRACE_A), TraceParent: traceparent(TRACE_B) } }, + ] + for (const options of optionCases) { + const payload = await invoke(provider, "kimi-k2", options) + assert.deepEqual(payload.metadata, {}) + } + + modelCallStarted?.( + { runId: "run", callId: "call", sessionId: "event-session", provider: "cloudsigma", model: "kimi-k2" }, + { runId: "run", sessionId: "context-session", trace: TRACE_B }, + ) + const ambiguous = await invoke(provider, "kimi-k2", { headers: { traceparent: traceparent(TRACE_B) } }) + assert.deepEqual(ambiguous.metadata, {}) +}) + +test("hook ignores missing authoritative session and malformed trace context", async () => { + const { provider, modelCallStarted } = captureProvider() + start(modelCallStarted, TRACE_A, undefined) + modelCallStarted?.( + { runId: "run", callId: "call", sessionId: "event-only-session", provider: "cloudsigma", model: "kimi-k2" }, + { runId: "run", trace: TRACE_A }, + ) + start(modelCallStarted, { traceId: TRACE_B.traceId, spanId: "bad" }, "session-b") + for (const trace of [TRACE_A, TRACE_B]) { + const payload = await invoke(provider, "kimi-k2", { headers: { traceparent: traceparent(trace) } }) + assert.deepEqual(payload.metadata, {}) + } +}) + +test("trace bridge prunes by TTL and bounded insertion order", () => { + let now = 1000 + const Bridge = internals.TraceSessionBridge + const bridge = new Bridge(2, 50, () => now) + bridge.record("trace-a", "session-a") + now += 1 + bridge.record("trace-b", "session-b") + now += 1 + bridge.record("trace-c", "session-c") + assert.equal(bridge.size, 2) + assert.equal(bridge.resolve("trace-a"), undefined) + assert.equal(bridge.resolve("trace-b"), "session-b") + assert.equal(bridge.resolve("trace-c"), "session-c") + // The successful resolutions above refresh the sliding TTL from `now`. + now += 50 + assert.equal(bridge.resolve("trace-b"), undefined) + assert.equal(bridge.size, 0) +}) + +test("successful exact matches refresh the sliding TTL for delayed retries", () => { + let now = 1000 + const Bridge = internals.TraceSessionBridge + const bridge = new Bridge(4, 100, () => now) + bridge.record("trace-a", "session-a") + now += 90 + assert.equal(bridge.resolve("trace-a"), "session-a") + now += 90 + assert.equal(bridge.resolve("trace-a"), "session-a") + now += 101 + assert.equal(bridge.resolve("trace-a"), undefined) +}) + +test("bridge outcome counters are privacy-safe and classify resolutions", () => { + let now = 1000 + const Bridge = internals.TraceSessionBridge + const bridge = new Bridge(4, 50, () => now) + bridge.clear() + bridge.record("hit-trace", "session-a") + assert.equal(bridge.resolve("hit-trace"), "session-a") + assert.equal(bridge.resolve("missing-trace"), undefined) + bridge.markAmbiguous("ambiguous-trace") + assert.equal(bridge.resolve("ambiguous-trace"), undefined) + bridge.record("expired-trace", "session-b") + now += 51 + assert.equal(bridge.resolve("expired-trace"), undefined) + assert.deepEqual(bridge.stats, { + hit: 1, + miss: 1, + expired: 1, + ambiguous: 1, + directOptionsSessionId: 0, + }) + assert.equal(JSON.stringify(bridge.stats).includes("session-a"), false) + assert.equal(JSON.stringify(bridge.stats).includes("trace"), false) +}) + +test("default bridge uses the documented 30-minute TTL and 1024-entry bound", () => { + assert.equal(internals.traceSessionBridge.ttlMsValue, 30 * 60 * 1000) + assert.equal(internals.traceSessionBridge.limitValue, 1024) +}) + +test("duplicate exact trace with conflicting sessions becomes ambiguous", () => { + const Bridge = internals.TraceSessionBridge + const bridge = new Bridge(4, 1000, () => 1) + bridge.record("same-trace", "session-a") + bridge.record("same-trace", "session-b") + bridge.record("same-trace", "session-a") + assert.equal(bridge.resolve("same-trace"), undefined) +}) + +test("existing metadata remains no-overwrite under trace bridge", async () => { + const { provider, modelCallStarted } = captureProvider() + start(modelCallStarted, TRACE_A, "bridge-session") + let payload: any + const wrapped = provider.wrapStreamFn({ + provider: "cloudsigma", + modelId: "kimi-k2", + streamFn: async (model: unknown, _context: unknown, options: any) => { + payload = await options.onPayload({ + messages: [], + metadata: { + session_id: "external-session", + sticky_key: "external-sticky", + requester_runtime: { source: "caller" }, + openclaw_correlation: { source: "caller" }, + }, + }, model) + return payload + }, + }) + await wrapped(createModel("kimi-k2"), {}, { headers: { traceparent: traceparent(TRACE_A) } }) + assert.deepEqual(payload.metadata, { + session_id: "external-session", + sticky_key: "external-sticky", + requester_runtime: { source: "caller" }, + openclaw_correlation: { source: "caller" }, + }) +}) + +test("hook unavailable is graceful and direct options.sessionId still works", async () => { + const { provider, modelCallStarted } = captureProvider(false) + assert.equal(modelCallStarted, undefined) + const direct = await invoke(provider, "gpt-5.6-sol", { sessionId: "direct-session" }) + assert.equal(direct.metadata.session_id, "direct-session") + const noIdentity = await invoke(provider, "kimi-k2", { headers: { traceparent: traceparent(TRACE_A) } }) + assert.deepEqual(noIdentity.metadata, {}) +}) + +test("trace parsers normalize only exact valid W3C correlation", () => { + const key = `${TRACE_A.traceId}:${TRACE_A.spanId}` + assert.equal(internals.traceKeyFromContext(TRACE_A), key) + assert.equal(internals.traceKeyFromTraceparent(traceparent(TRACE_A).toUpperCase()), key) + assert.equal(internals.traceKeyFromTraceparent(`01-${TRACE_A.traceId}-${TRACE_A.spanId}-01`), undefined) + assert.equal(internals.traceKeyFromContext({ traceId: TRACE_A.traceId }), undefined) +})