refactor(gateway): share talk relay session lifecycle

vincentkoc · vincentkoc · commit 548785581580 · 2026-06-02T11:57:08.000+02:00
diff --git a/src/gateway/talk-realtime-relay.ts b/src/gateway/talk-realtime-relay.ts
@@ -1,8 +1,5 @@
 import { randomUUID } from "node:crypto";
-import {
-  asDateTimestampMs,
-  resolveExpiresAtMsFromDurationMs,
-} from "@openclaw/normalization-core/number-coercion";
+import { resolveExpiresAtMsFromDurationMs } from "@openclaw/normalization-core/number-coercion";
 import type { OpenClawConfig } from "../config/types.js";
 import type { RealtimeVoiceProviderPlugin } from "../plugins/types.js";
 import {
@@ -45,6 +42,10 @@ import {
 } from "../talk/talk-session-controller.js";
 import { abortChatRunById } from "./chat-abort.js";
 import type { GatewayRequestContext } from "./server-methods/shared-types.js";
+import {
+  closeExpiredTalkRelaySessions,
+  requireActiveTalkRelaySession,
+} from "./talk-relay-session-lifecycle.js";
 import { forgetUnifiedTalkSession } from "./talk-session-registry.js";
 
 const RELAY_SESSION_TTL_MS = 30 * 60 * 1000;
@@ -231,9 +232,7 @@ function broadcastToolResultToOwner(
   },
 ): void {
   const payload =
-    params.forced === true
-      ? { result: params.result, forced: true }
-      : { result: params.result };
+    params.forced === true ? { result: params.result, forced: true } : { result: params.result };
   broadcastToOwner(session.context, session.connId, {
     relaySessionId: session.id,
     type: "toolResult",
@@ -301,16 +300,11 @@ function closeRelaySession(session: RelaySession, reason: "completed" | "error")
 }
 
 function pruneExpiredRelaySessions(nowMs = Date.now()): void {
-  const validNowMs = asDateTimestampMs(nowMs);
-  if (validNowMs === undefined) {
-    return;
-  }
-  for (const session of relaySessions.values()) {
-    const expiresAtMs = asDateTimestampMs(session.expiresAtMs);
-    if (expiresAtMs === undefined || validNowMs > expiresAtMs) {
-      closeRelaySession(session, "completed");
-    }
-  }
+  closeExpiredTalkRelaySessions({
+    sessions: relaySessions.values(),
+    closeSession: (session) => closeRelaySession(session, "completed"),
+    nowMs,
+  });
 }
 
 function countRelaySessionsForConn(connId: string): number {
@@ -724,22 +718,13 @@ function ensureRelayTurn(session: RelaySession): string {
 }
 
 function getRelaySession(relaySessionId: string, connId: string): RelaySession {
-  const session = relaySessions.get(relaySessionId);
-  const nowMs = asDateTimestampMs(Date.now());
-  const expiresAtMs = session ? asDateTimestampMs(session.expiresAtMs) : undefined;
-  if (
-    !session ||
-    session.connId !== connId ||
-    nowMs === undefined ||
-    expiresAtMs === undefined ||
-    nowMs > expiresAtMs
-  ) {
-    if (session) {
-      closeRelaySession(session, "completed");
-    }
-    throw new Error("Unknown realtime relay session");
-  }
-  return session;
+  return requireActiveTalkRelaySession({
+    sessions: relaySessions,
+    sessionId: relaySessionId,
+    connId,
+    closeSession: (session) => closeRelaySession(session, "completed"),
+    unknownSessionMessage: "Unknown realtime relay session",
+  });
 }
 
 export function sendTalkRealtimeRelayAudio(params: {
diff --git a/src/gateway/talk-relay-session-lifecycle.ts b/src/gateway/talk-relay-session-lifecycle.ts
@@ -0,0 +1,57 @@
+import { asDateTimestampMs } from "@openclaw/normalization-core/number-coercion";
+
+type TalkRelayLifecycleSession = {
+  connId: string;
+  expiresAtMs: number;
+};
+
+type CloseTalkRelaySession<TSession extends TalkRelayLifecycleSession> = (
+  session: TSession,
+) => void;
+
+function isExpiredTalkRelaySession(
+  session: TalkRelayLifecycleSession,
+  validNowMs: number,
+): boolean {
+  const expiresAtMs = asDateTimestampMs(session.expiresAtMs);
+  return expiresAtMs === undefined || validNowMs > expiresAtMs;
+}
+
+export function closeExpiredTalkRelaySessions<TSession extends TalkRelayLifecycleSession>(params: {
+  sessions: Iterable<TSession>;
+  closeSession: CloseTalkRelaySession<TSession>;
+  nowMs?: number;
+}): void {
+  const validNowMs = asDateTimestampMs(params.nowMs ?? Date.now());
+  if (validNowMs === undefined) {
+    return;
+  }
+  for (const session of params.sessions) {
+    if (isExpiredTalkRelaySession(session, validNowMs)) {
+      params.closeSession(session);
+    }
+  }
+}
+
+export function requireActiveTalkRelaySession<TSession extends TalkRelayLifecycleSession>(params: {
+  sessions: ReadonlyMap<string, TSession>;
+  sessionId: string;
+  connId: string;
+  closeSession: CloseTalkRelaySession<TSession>;
+  unknownSessionMessage: string;
+}): TSession {
+  const session = params.sessions.get(params.sessionId);
+  const nowMs = asDateTimestampMs(Date.now());
+  if (
+    !session ||
+    session.connId !== params.connId ||
+    nowMs === undefined ||
+    isExpiredTalkRelaySession(session, nowMs)
+  ) {
+    if (session) {
+      params.closeSession(session);
+    }
+    throw new Error(params.unknownSessionMessage);
+  }
+  return session;
+}
diff --git a/src/gateway/talk-transcription-relay.ts b/src/gateway/talk-transcription-relay.ts
@@ -1,6 +1,5 @@
 import { randomUUID } from "node:crypto";
 import {
-  asDateTimestampMs,
   parseFiniteNumber as readFiniteNumber,
   resolveExpiresAtMsFromDurationMs,
 } from "@openclaw/normalization-core/number-coercion";
@@ -14,6 +13,10 @@ import {
   createTalkSessionController,
 } from "../talk/talk-session-controller.js";
 import type { GatewayRequestContext } from "./server-methods/shared-types.js";
+import {
+  closeExpiredTalkRelaySessions,
+  requireActiveTalkRelaySession,
+} from "./talk-relay-session-lifecycle.js";
 
 const TRANSCRIPTION_SESSION_TTL_MS = 30 * 60 * 1000;
 const MAX_AUDIO_BASE64_BYTES = 512 * 1024;
@@ -181,16 +184,11 @@ function closeTranscriptionSession(
 }
 
 function pruneExpiredTranscriptionSessions(nowMs = Date.now()): void {
-  const validNowMs = asDateTimestampMs(nowMs);
-  if (validNowMs === undefined) {
-    return;
-  }
-  for (const session of transcriptionSessions.values()) {
-    const expiresAtMs = asDateTimestampMs(session.expiresAtMs);
-    if (expiresAtMs === undefined || validNowMs > expiresAtMs) {
-      closeTranscriptionSession(session, "completed");
-    }
-  }
+  closeExpiredTalkRelaySessions({
+    sessions: transcriptionSessions.values(),
+    closeSession: (session) => closeTranscriptionSession(session, "completed"),
+    nowMs,
+  });
 }
 
 function countTranscriptionSessionsForConn(connId: string): number {
@@ -361,22 +359,13 @@ function getTranscriptionSession(
   transcriptionSessionId: string,
   connId: string,
 ): TranscriptionRelaySession {
-  const session = transcriptionSessions.get(transcriptionSessionId);
-  const nowMs = asDateTimestampMs(Date.now());
-  const expiresAtMs = session ? asDateTimestampMs(session.expiresAtMs) : undefined;
-  if (
-    !session ||
-    session.connId !== connId ||
-    nowMs === undefined ||
-    expiresAtMs === undefined ||
-    nowMs > expiresAtMs
-  ) {
-    if (session) {
-      closeTranscriptionSession(session, "completed");
-    }
-    throw new Error("Unknown transcription Talk session");
-  }
-  return session;
+  return requireActiveTalkRelaySession({
+    sessions: transcriptionSessions,
+    sessionId: transcriptionSessionId,
+    connId,
+    closeSession: (session) => closeTranscriptionSession(session, "completed"),
+    unknownSessionMessage: "Unknown transcription Talk session",
+  });
 }
 
 export function sendTalkTranscriptionRelayAudio(params: {
