From d3a06590e9d990f706fa69527310d8abcfd261bc Mon Sep 17 00:00:00 2001 From: joshavant <830519+joshavant@users.noreply.github.com> Date: Thu, 23 Jul 2026 01:29:28 -0500 Subject: [PATCH] refactor(worker): split live session binding --- .../live-event-session-binding.ts | 75 ++++++++++++++++ .../worker-environments/live-events.ts | 86 ++----------------- .../worker-turn-launcher.test.ts | 2 +- 3 files changed, 85 insertions(+), 78 deletions(-) create mode 100644 src/gateway/worker-environments/live-event-session-binding.ts diff --git a/src/gateway/worker-environments/live-event-session-binding.ts b/src/gateway/worker-environments/live-event-session-binding.ts new file mode 100644 index 000000000000..1d7e656d0deb --- /dev/null +++ b/src/gateway/worker-environments/live-event-session-binding.ts @@ -0,0 +1,75 @@ +import type { SessionIdentityMutation } from "../../config/sessions/session-accessor.js"; +import type { OpenClawConfig } from "../../config/types.openclaw.js"; +import type { WorkerLiveTrajectoryTarget } from "./live-event-projection.js"; +import { resolveWorkerSessionTarget } from "./session-target.js"; + +export type LiveEventTarget = WorkerLiveTrajectoryTarget; + +export type WorkerLiveSessionBinding = Readonly<{ + environmentId: string; + runEpoch: number; + sessionId: string; +}>; + +export type BoundLiveSession = WorkerLiveSessionBinding & { target: LiveEventTarget }; + +export function isValidLiveSessionBinding(binding: WorkerLiveSessionBinding): boolean { + return ( + binding.environmentId.length > 0 && + binding.sessionId.length > 0 && + Number.isSafeInteger(binding.runEpoch) && + binding.runEpoch >= 0 + ); +} + +function resolveLiveEventTarget( + config: OpenClawConfig, + sessionId: string, +): LiveEventTarget | undefined { + const target = resolveWorkerSessionTarget(config, sessionId); + if (!target) { + return undefined; + } + return { + ...(target.agentId ? { agentId: target.agentId } : {}), + sessionId: target.sessionId, + sessionKey: target.sessionKey, + storePath: target.storePath, + }; +} + +function prepareBoundLiveSession( + config: OpenClawConfig, + binding: WorkerLiveSessionBinding, +): BoundLiveSession | undefined { + if (!isValidLiveSessionBinding(binding)) { + return undefined; + } + const target = resolveLiveEventTarget(config, binding.sessionId); + return target ? { ...binding, target } : undefined; +} + +export function prepareBoundLiveSessionSafely( + config: OpenClawConfig, + binding: WorkerLiveSessionBinding, +): BoundLiveSession | undefined { + try { + return prepareBoundLiveSession(config, binding); + } catch { + return undefined; + } +} + +export function matchesSessionIdentityMutation( + binding: WorkerLiveSessionBinding, + prepared: BoundLiveSession | undefined, + mutation: SessionIdentityMutation, +): boolean { + const targets = + "current" in mutation ? [mutation.previous, mutation.current] : [mutation.previous]; + return targets.some( + (target) => + target.sessionId === binding.sessionId || + (prepared ? target.sessionKeys.includes(prepared.target.sessionKey) : false), + ); +} diff --git a/src/gateway/worker-environments/live-events.ts b/src/gateway/worker-environments/live-events.ts index bfcdb2aaabe7..70c308d7e3c1 100644 --- a/src/gateway/worker-environments/live-events.ts +++ b/src/gateway/worker-environments/live-events.ts @@ -4,10 +4,7 @@ import type { WorkerLiveEventParams, WorkerLiveEventResult, } from "../../../packages/gateway-protocol/src/schema/worker-admission.js"; -import { - onSessionIdentityMutation, - type SessionIdentityMutation, -} from "../../config/sessions/session-accessor.js"; +import { onSessionIdentityMutation } from "../../config/sessions/session-accessor.js"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; import { claimAgentRunContext, @@ -26,9 +23,15 @@ import { prepareWorkerLiveEventData, recordWorkerLiveTrajectoryEvent, type WorkerLiveTrajectoryRecorder, - type WorkerLiveTrajectoryTarget, } from "./live-event-projection.js"; -import { resolveWorkerSessionTarget } from "./session-target.js"; +import { + isValidLiveSessionBinding, + matchesSessionIdentityMutation, + prepareBoundLiveSessionSafely, + type BoundLiveSession, + type LiveEventTarget, + type WorkerLiveSessionBinding, +} from "./live-event-session-binding.js"; const DEFAULT_WINDOW_SIZE = 128; const DEFAULT_MAX_PENDING_BYTES = 512 * 1024; @@ -49,16 +52,6 @@ type OwnedLiveRun = { trajectoryRecorder: WorkerLiveTrajectoryRecorder; }; -type LiveEventTarget = WorkerLiveTrajectoryTarget; - -type WorkerLiveSessionBinding = Readonly<{ - environmentId: string; - runEpoch: number; - sessionId: string; -}>; - -type BoundLiveSession = WorkerLiveSessionBinding & { target: LiveEventTarget }; - type WorkerLiveCredentialRotation = Readonly<{ credentialHash: string; environmentId: string; @@ -105,67 +98,6 @@ function capacityExceeded(): WorkerLiveEventFailure { return { ok: false, details: { reason: "capacity-exceeded" } }; } -function resolveLiveEventTarget( - config: OpenClawConfig, - sessionId: string, -): LiveEventTarget | undefined { - const target = resolveWorkerSessionTarget(config, sessionId); - if (!target) { - return undefined; - } - return { - ...(target.agentId ? { agentId: target.agentId } : {}), - sessionId: target.sessionId, - sessionKey: target.sessionKey, - storePath: target.storePath, - }; -} - -function prepareBoundLiveSession( - config: OpenClawConfig, - binding: WorkerLiveSessionBinding, -): BoundLiveSession | undefined { - if (!isValidLiveSessionBinding(binding)) { - return undefined; - } - const target = resolveLiveEventTarget(config, binding.sessionId); - return target ? { ...binding, target } : undefined; -} - -function isValidLiveSessionBinding(binding: WorkerLiveSessionBinding): boolean { - return ( - binding.environmentId.length > 0 && - binding.sessionId.length > 0 && - Number.isSafeInteger(binding.runEpoch) && - binding.runEpoch >= 0 - ); -} - -function prepareBoundLiveSessionSafely( - config: OpenClawConfig, - binding: WorkerLiveSessionBinding, -): BoundLiveSession | undefined { - try { - return prepareBoundLiveSession(config, binding); - } catch { - return undefined; - } -} - -function matchesSessionIdentityMutation( - binding: WorkerLiveSessionBinding, - prepared: BoundLiveSession | undefined, - mutation: SessionIdentityMutation, -): boolean { - const targets = - "current" in mutation ? [mutation.previous, mutation.current] : [mutation.previous]; - return targets.some( - (target) => - target.sessionId === binding.sessionId || - (prepared ? target.sessionKeys.includes(prepared.target.sessionKey) : false), - ); -} - export function createWorkerLiveEventReceiver(options: WorkerLiveEventReceiverOptions) { const boundSessions = new Map(); const sessionBindings = new Map(); diff --git a/src/gateway/worker-environments/worker-turn-launcher.test.ts b/src/gateway/worker-environments/worker-turn-launcher.test.ts index e669061d1a6e..c3a47d035759 100644 --- a/src/gateway/worker-environments/worker-turn-launcher.test.ts +++ b/src/gateway/worker-environments/worker-turn-launcher.test.ts @@ -637,7 +637,7 @@ describe("worker turn launcher", () => { sessionKey: SESSION_KEY, agentId: "main", }); - expect(tunnel.reconcileWorkspace).toHaveBeenCalledWith( + expect(vi.mocked(tunnel.reconcileWorkspace)).toHaveBeenCalledWith( expect.objectContaining({ localPath: root }), ); const conflictSummary =