diff --git a/backend/src/main/java/com/babelflux/backend/websocket/HandoffSessionSubscriber.java b/backend/src/main/java/com/babelflux/backend/websocket/HandoffSessionSubscriber.java
new file mode 100644
index 0000000..aeb6602
--- /dev/null
+++ b/backend/src/main/java/com/babelflux/backend/websocket/HandoffSessionSubscriber.java
@@ -0,0 +1,69 @@
+package com.babelflux.backend.websocket;
+
+import jakarta.annotation.PreDestroy;
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.springframework.web.socket.WebSocketSession;
+
+/** Owns read-only handoff subscriptions and their lifecycle. */
+final class HandoffSessionSubscriber {
+    private final SessionEventHub eventHub;
+    private final WebSocketEventSender sender;
+    private final ConcurrentMap<String, SessionEventHub.Subscription> subscriptions = new ConcurrentHashMap<>();
+    private final ConcurrentMap<String, Future<?>> tasks = new ConcurrentHashMap<>();
+    private final ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
+
+    HandoffSessionSubscriber(SessionEventHub eventHub, WebSocketEventSender sender) {
+        this.eventHub = eventHub;
+        this.sender = sender;
+    }
+
+    boolean isHandoff(String socketId) { return subscriptions.containsKey(socketId); }
+
+    void start(WebSocketSession socket, String sessionId) throws IOException {
+        SessionEventHub.Subscription subscription = eventHub.subscribe(sessionId);
+        subscriptions.put(socket.getId(), subscription);
+        try {
+            sender.send(socket, Map.of("type", "session_started", "sessionId", sessionId));
+            for (Map<String, Object> event : subscription.replay()) sender.send(socket, event);
+            tasks.put(socket.getId(), executor.submit(() -> forward(socket, subscription)));
+        } catch (IOException error) {
+            close(socket.getId());
+            throw error;
+        }
+    }
+
+    boolean close(String socketId) {
+        SessionEventHub.Subscription subscription = subscriptions.remove(socketId);
+        if (subscription == null) return false;
+        eventHub.unsubscribe(subscription);
+        Future<?> task = tasks.remove(socketId);
+        if (task != null) task.cancel(true);
+        return true;
+    }
+
+    @PreDestroy
+    void shutdown() {
+        tasks.values().forEach(task -> task.cancel(true));
+        executor.shutdownNow();
+    }
+
+    private void forward(WebSocketSession socket, SessionEventHub.Subscription subscription) {
+        try {
+            while (!Thread.currentThread().isInterrupted()) {
+                Map<String, Object> event = subscription.await(1, TimeUnit.SECONDS);
+                if (event != null) sender.send(socket, event);
+            }
+        } catch (InterruptedException error) {
+            Thread.currentThread().interrupt();
+        } catch (Exception ignored) {
+            // A display-only socket can close while its primary session continues.
+        }
+    }
+}
diff --git a/backend/src/main/java/com/babelflux/backend/websocket/RealtimeSessionCoordinator.java b/backend/src/main/java/com/babelflux/backend/websocket/RealtimeSessionCoordinator.java
new file mode 100644
index 0000000..cc8947d
--- /dev/null
+++ b/backend/src/main/java/com/babelflux/backend/websocket/RealtimeSessionCoordinator.java
@@ -0,0 +1,175 @@
+package com.babelflux.backend.websocket;
+
+import com.babelflux.backend.domain.Session;
+import com.babelflux.backend.domain.SessionReport;
+import com.babelflux.backend.service.RealtimeSessionRunner;
+import com.babelflux.backend.service.SessionService;
+import jakarta.annotation.PreDestroy;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+/** Application coordinator for one realtime session runner; WebSocket parsing stays in the adapter. */
+final class RealtimeSessionCoordinator {
+    @FunctionalInterface
+    interface EventSink { void emit(Map<String, Object> event) throws Exception; }
+
+    record StartRequest(String sourceLanguage, String targetLanguage, String domain, String inputMode,
+                        String sourceUrl, String modelProfile) {}
+    record CommandResult(String error, SessionReport report) {
+        static CommandResult ok() { return new CommandResult(null, null); }
+        static CommandResult report(SessionReport report) { return new CommandResult(null, report); }
+        static CommandResult error(String error) { return new CommandResult(error, null); }
+        boolean failed() { return error != null; }
+    }
+
+    private final SessionService sessions;
+    private final RealtimeSessionRunner runner;
+    private final RunnerLeaseManager leases;
+    private final ConcurrentMap<String, RealtimeSessionRunner.RunHandle> runs = new ConcurrentHashMap<>();
+    private final ConcurrentMap<String, EventSink> eventSinks = new ConcurrentHashMap<>();
+
+    RealtimeSessionCoordinator(SessionService sessions, RealtimeSessionRunner runner, RunnerLeaseManager leases) {
+        this.sessions = sessions;
+        this.runner = runner;
+        this.leases = leases;
+    }
+
+    CommandResult start(String sessionId, String socketId, StartRequest request, EventSink sink) {
+        if (!SessionService.isSupportedInputMode(request.inputMode())) {
+            return CommandResult.error("暂不支持的输入模式：" + request.inputMode());
+        }
+        synchronized (runs) {
+            if (runs.containsKey(socketId)) return CommandResult.ok();
+            RunnerLeaseManager.AcquireResult acquired = leases.acquire(sessionId, socketId);
+            if (acquired != RunnerLeaseManager.AcquireResult.ACQUIRED) {
+                return CommandResult.error(acquireMessage(acquired));
+            }
+            Session session;
+            try {
+                session = sessions.get(sessionId);
+            } catch (RuntimeException | Error error) {
+                leases.release(sessionId, socketId);
+                throw error;
+            }
+            session.applyOverrides(request.sourceLanguage(), request.targetLanguage(), request.domain(),
+                    request.inputMode(), request.sourceUrl(), request.modelProfile());
+            session.start();
+            AtomicBoolean reportEmitted = new AtomicBoolean();
+            AtomicReference<RealtimeSessionRunner.RunHandle> handleRef = new AtomicReference<>();
+            try {
+                // Save the lifecycle fact before another instance can read stale session history.
+                sessions.saveProgress(session);
+                eventSinks.put(socketId, sink);
+                RealtimeSessionRunner.RunHandle handle = runner.start(session, event -> {
+                    if ("session_report".equals(event.get("type"))) {
+                        reportEmitted.set(true);
+                        RealtimeSessionRunner.RunHandle current = handleRef.get();
+                        if (current != null) runs.remove(socketId, current);
+                        leases.release(sessionId, socketId);
+                        eventSinks.remove(socketId, sink);
+                    }
+                    sink.emit(event);
+                });
+                handleRef.set(handle);
+                runs.put(socketId, handle);
+                if (reportEmitted.get()) {
+                    runs.remove(socketId, handle);
+                    leases.release(sessionId, socketId);
+                    eventSinks.remove(socketId, sink);
+                } else {
+                    leases.scheduleRenewal(sessionId, socketId, () -> leaseLost(sessionId, socketId));
+                }
+                return CommandResult.ok();
+            } catch (RuntimeException | Error error) {
+                eventSinks.remove(socketId, sink);
+                session.end();
+                try {
+                    sessions.saveProgress(session);
+                } catch (RuntimeException | Error ignored) {
+                    error.addSuppressed(ignored);
+                }
+                leases.release(sessionId, socketId);
+                throw error;
+            }
+        }
+    }
+
+    CommandResult stop(String sessionId, String socketId) {
+        RealtimeSessionRunner.RunHandle handle = runs.get(socketId);
+        if (handle != null) {
+            handle.stop();
+            return CommandResult.ok();
+        }
+        RunnerLeaseManager.FinishPermission permission = leases.canFinish(sessionId, socketId);
+        if (permission != RunnerLeaseManager.FinishPermission.ALLOWED) {
+            return CommandResult.error(finishMessage(permission));
+        }
+        return CommandResult.report(sessions.finish(sessionId));
+    }
+
+    void acceptAudio(String socketId, byte[] pcm) {
+        RealtimeSessionRunner.RunHandle handle = runs.get(socketId);
+        if (handle != null) handle.acceptAudio(pcm);
+    }
+
+    boolean hasRun(String socketId) { return runs.containsKey(socketId); }
+
+    void updateClock(String socketId, long playbackMs, long sentAudioMs) {
+        RealtimeSessionRunner.RunHandle handle = runs.get(socketId);
+        if (handle != null) handle.updateClientClock(playbackMs, sentAudioMs);
+    }
+
+    void pause(String socketId) {
+        RealtimeSessionRunner.RunHandle handle = runs.get(socketId);
+        if (handle != null) handle.pause();
+    }
+
+    void resume(String socketId) {
+        RealtimeSessionRunner.RunHandle handle = runs.get(socketId);
+        if (handle != null) handle.resume();
+    }
+
+    void close(String sessionId, String socketId) {
+        eventSinks.remove(socketId);
+        RealtimeSessionRunner.RunHandle handle = runs.remove(socketId);
+        if (handle != null) handle.stop();
+        leases.release(sessionId, socketId);
+    }
+
+    @PreDestroy
+    void shutdown() { eventSinks.clear(); }
+
+    private void leaseLost(String sessionId, String socketId) {
+        RealtimeSessionRunner.RunHandle handle = runs.get(socketId);
+        if (handle != null) handle.stop();
+        leases.release(sessionId, socketId);
+        EventSink sink = eventSinks.remove(socketId);
+        if (sink == null) return;
+        try {
+            sink.emit(Map.of("type", "error", "message", "会话租约续期失败，已停止实时会话"));
+        } catch (Exception ignored) {
+            // Socket ownership has already been released; sending the notice is best effort.
+        }
+    }
+
+    private static String acquireMessage(RunnerLeaseManager.AcquireResult result) {
+        return switch (result) {
+            case LOCAL_BUSY -> "会话已在其他连接中运行";
+            case REMOTE_BUSY -> "会话已在其他实例中运行";
+            case UNAVAILABLE -> "会话锁服务暂不可用";
+            case ACQUIRED -> throw new IllegalArgumentException("acquired has no failure message");
+        };
+    }
+
+    private static String finishMessage(RunnerLeaseManager.FinishPermission permission) {
+        return switch (permission) {
+            case LOCAL_BUSY -> "会话正在其他连接中运行";
+            case REMOTE_BUSY -> "会话正在其他实例中运行";
+            case UNAVAILABLE -> "会话锁服务暂不可用";
+            case ALLOWED -> throw new IllegalArgumentException("allowed has no failure message");
+        };
+    }
+}
diff --git a/backend/src/main/java/com/babelflux/backend/websocket/RunnerLeaseManager.java b/backend/src/main/java/com/babelflux/backend/websocket/RunnerLeaseManager.java
new file mode 100644
index 0000000..5e59682
--- /dev/null
+++ b/backend/src/main/java/com/babelflux/backend/websocket/RunnerLeaseManager.java
@@ -0,0 +1,91 @@
+package com.babelflux.backend.websocket;
+
+import com.babelflux.backend.infrastructure.RedisSessionRepository;
+import jakarta.annotation.PreDestroy;
+import java.time.Duration;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+/** Coordinates local ownership and the optional Redis runner lease without touching WebSocket protocol code. */
+final class RunnerLeaseManager {
+    private static final long TTL_SECONDS = 30;
+    private static final long RENEW_SECONDS = 10;
+
+    enum AcquireResult { ACQUIRED, LOCAL_BUSY, REMOTE_BUSY, UNAVAILABLE }
+    enum FinishPermission { ALLOWED, LOCAL_BUSY, REMOTE_BUSY, UNAVAILABLE }
+
+    private final RedisSessionRepository redis;
+    private final ConcurrentMap<String, String> owners = new ConcurrentHashMap<>();
+    private final ConcurrentMap<String, ScheduledFuture<?>> renewals = new ConcurrentHashMap<>();
+    private final ScheduledExecutorService executor;
+
+    RunnerLeaseManager(RedisSessionRepository redis) {
+        this.redis = redis;
+        this.executor = redis == null ? null
+                : Executors.newSingleThreadScheduledExecutor(Thread.ofVirtual().name("runner-lease-renew").factory());
+    }
+
+    AcquireResult acquire(String sessionId, String socketId) {
+        if (owners.putIfAbsent(sessionId, socketId) != null) return AcquireResult.LOCAL_BUSY;
+        if (redis == null) return AcquireResult.ACQUIRED;
+        try {
+            if (redis.tryAcquireRunnerLease(sessionId, socketId, Duration.ofSeconds(TTL_SECONDS))) {
+                return AcquireResult.ACQUIRED;
+            }
+            owners.remove(sessionId, socketId);
+            return AcquireResult.REMOTE_BUSY;
+        } catch (RuntimeException error) {
+            owners.remove(sessionId, socketId);
+            return AcquireResult.UNAVAILABLE;
+        }
+    }
+
+    FinishPermission canFinish(String sessionId, String socketId) {
+        String owner = owners.get(sessionId);
+        if (owner != null && !owner.equals(socketId)) return FinishPermission.LOCAL_BUSY;
+        if (owner != null || redis == null) return FinishPermission.ALLOWED;
+        try {
+            return redis.runnerLeaseHeld(sessionId) ? FinishPermission.REMOTE_BUSY : FinishPermission.ALLOWED;
+        } catch (RuntimeException error) {
+            return FinishPermission.UNAVAILABLE;
+        }
+    }
+
+    void scheduleRenewal(String sessionId, String socketId, Runnable onRenewalLost) {
+        if (redis == null || executor == null) return;
+        ScheduledFuture<?> task = executor.scheduleAtFixedRate(() -> {
+            if (!socketId.equals(owners.get(sessionId))) return;
+            boolean renewed;
+            try {
+                renewed = redis.renewRunnerLease(sessionId, socketId, Duration.ofSeconds(TTL_SECONDS));
+            } catch (RuntimeException error) {
+                renewed = false;
+            }
+            if (!renewed) onRenewalLost.run();
+        }, RENEW_SECONDS, RENEW_SECONDS, TimeUnit.SECONDS);
+        ScheduledFuture<?> previous = renewals.put(sessionId, task);
+        if (previous != null) previous.cancel(false);
+    }
+
+    void release(String sessionId, String socketId) {
+        if (!owners.remove(sessionId, socketId)) return;
+        ScheduledFuture<?> renewal = renewals.remove(sessionId);
+        if (renewal != null) renewal.cancel(false);
+        if (redis == null) return;
+        try {
+            redis.releaseRunnerLease(sessionId, socketId);
+        } catch (RuntimeException ignored) {
+            // Redis TTL expires a lease that cannot be released during cleanup.
+        }
+    }
+
+    @PreDestroy
+    void shutdown() {
+        renewals.values().forEach(task -> task.cancel(true));
+        if (executor != null) executor.shutdownNow();
+    }
+}
diff --git a/backend/src/main/java/com/babelflux/backend/websocket/SessionEventDispatcher.java b/backend/src/main/java/com/babelflux/backend/websocket/SessionEventDispatcher.java
new file mode 100644
index 0000000..dc41179
--- /dev/null
+++ b/backend/src/main/java/com/babelflux/backend/websocket/SessionEventDispatcher.java
@@ -0,0 +1,32 @@
+package com.babelflux.backend.websocket;
+
+import java.io.IOException;
+import java.util.Map;
+import org.springframework.web.socket.WebSocketSession;
+
+/** Publishes session events for handoff subscribers before sending to the primary socket. */
+final class SessionEventDispatcher {
+    private final SessionEventHub eventHub;
+    private final WebSocketEventSender sender;
+
+    SessionEventDispatcher(SessionEventHub eventHub, WebSocketEventSender sender) {
+        this.eventHub = eventHub;
+        this.sender = sender;
+    }
+
+    void sendDirect(WebSocketSession socket, Object body) throws IOException { sender.send(socket, body); }
+
+    void dispatchRunnerEvent(String sessionId, WebSocketSession socket, Map<String, Object> event) throws IOException {
+        eventHub.publish(sessionId, event);
+        if ("session_report".equals(event.get("type"))) eventHub.complete(sessionId);
+        sender.send(socket, event);
+    }
+
+    void dispatchSourceState(String sessionId, WebSocketSession socket, String status, String message)
+            throws IOException {
+        Map<String, Object> event = Map.of("type", "source_sync_state",
+                "state", Map.of("status", status, "lagMs", 0, "message", message));
+        eventHub.publish(sessionId, event);
+        sender.send(socket, event);
+    }
+}
diff --git a/backend/src/main/java/com/babelflux/backend/websocket/SessionWebSocketHandler.java b/backend/src/main/java/com/babelflux/backend/websocket/SessionWebSocketHandler.java
index 7322494..aaf9fb1 100644
--- a/backend/src/main/java/com/babelflux/backend/websocket/SessionWebSocketHandler.java
+++ b/backend/src/main/java/com/babelflux/backend/websocket/SessionWebSocketHandler.java
@@ -1,30 +1,17 @@
 package com.babelflux.backend.websocket;
 
+import com.babelflux.backend.infrastructure.RedisSessionRepository;
+import com.babelflux.backend.service.RealtimeSessionRunner;
 import com.babelflux.backend.service.SessionService;
 import com.babelflux.backend.service.SessionTokenService;
-import com.babelflux.backend.service.RealtimeSessionRunner;
-import com.babelflux.backend.infrastructure.RedisSessionRepository;
-import com.babelflux.backend.observability.OperationalMetrics;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import jakarta.annotation.PreDestroy;
 import java.io.IOException;
 import java.nio.ByteBuffer;
-import java.time.Duration;
 import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicReference;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import org.springframework.web.socket.BinaryMessage;
 import org.springframework.web.socket.CloseStatus;
@@ -33,103 +20,63 @@
 import org.springframework.web.socket.WebSocketSession;
 import org.springframework.web.socket.handler.TextWebSocketHandler;
 
+/** WebSocket protocol adapter. Session execution, leases and fan-out live in dedicated collaborators. */
 @Component
 public class SessionWebSocketHandler extends TextWebSocketHandler implements WebSocketHandler {
-    private static final long RUNNER_LEASE_TTL_SECONDS = 30;
-    private static final long RUNNER_LEASE_RENEW_SECONDS = 10;
     private final ObjectMapper mapper;
     private final SessionService sessions;
     private final SessionTokenService tokens;
-    private final RealtimeSessionRunner runner;
-    private final SessionEventHub eventHub;
-    private final OperationalMetrics metrics;
-    private final ConcurrentMap<String, RealtimeSessionRunner.RunHandle> runs = new ConcurrentHashMap<>();
-    /** One primary realtime runner is allowed per session; handoff sockets are read-only. */
-    private final ConcurrentMap<String, String> activeSessionSockets = new ConcurrentHashMap<>();
-    private final Set<String> observedSockets = ConcurrentHashMap.newKeySet();
-    private final RedisSessionRepository redis;
-    private final ConcurrentMap<String, ScheduledFuture<?>> leaseRenewals = new ConcurrentHashMap<>();
-    private final ConcurrentMap<String, SessionEventHub.Subscription> handoffSubscriptions = new ConcurrentHashMap<>();
-    private final ConcurrentMap<String, Future<?>> handoffTasks = new ConcurrentHashMap<>();
-    private final ExecutorService handoffExecutor = Executors.newVirtualThreadPerTaskExecutor();
-    private final ScheduledExecutorService leaseExecutor;
+    private final RealtimeSessionCoordinator coordinator;
+    private final SessionEventDispatcher dispatcher;
+    private final HandoffSessionSubscriber handoffs;
+    private final RunnerLeaseManager leases;
 
     public SessionWebSocketHandler(ObjectMapper mapper, SessionService sessions, SessionTokenService tokens,
                                    RealtimeSessionRunner runner) {
-        this(mapper, sessions, tokens, runner, new SessionEventHub(), (RedisSessionRepository) null,
-                OperationalMetrics.NOOP);
+        this(mapper, sessions, tokens, runner, new SessionEventHub(), (RedisSessionRepository) null);
     }
 
     public SessionWebSocketHandler(ObjectMapper mapper, SessionService sessions, SessionTokenService tokens,
                                    RealtimeSessionRunner runner, SessionEventHub eventHub) {
-        this(mapper, sessions, tokens, runner, eventHub, (RedisSessionRepository) null, OperationalMetrics.NOOP);
-    }
-
-    public SessionWebSocketHandler(ObjectMapper mapper, SessionService sessions, SessionTokenService tokens,
-                                   RealtimeSessionRunner runner, SessionEventHub eventHub,
-                                   ObjectProvider<RedisSessionRepository> redisProvider) {
-        this(mapper, sessions, tokens, runner, eventHub, (RedisSessionRepository) redisProvider.getIfAvailable(),
-                OperationalMetrics.NOOP);
+        this(mapper, sessions, tokens, runner, eventHub, (RedisSessionRepository) null);
     }
 
     @Autowired
     public SessionWebSocketHandler(ObjectMapper mapper, SessionService sessions, SessionTokenService tokens,
                                    RealtimeSessionRunner runner, SessionEventHub eventHub,
-                                   ObjectProvider<RedisSessionRepository> redisProvider,
-                                   ObjectProvider<OperationalMetrics> metricsProvider) {
-        this(mapper, sessions, tokens, runner, eventHub, (RedisSessionRepository) redisProvider.getIfAvailable(),
-                metricsProvider.getIfAvailable());
+                                   ObjectProvider<RedisSessionRepository> redisProvider) {
+        this(mapper, sessions, tokens, runner, eventHub, redisProvider.getIfAvailable());
     }
 
     SessionWebSocketHandler(ObjectMapper mapper, SessionService sessions, SessionTokenService tokens,
                             RealtimeSessionRunner runner, SessionEventHub eventHub,
                             RedisSessionRepository redis) {
-        this(mapper, sessions, tokens, runner, eventHub, redis, OperationalMetrics.NOOP);
-    }
-
-    SessionWebSocketHandler(ObjectMapper mapper, SessionService sessions, SessionTokenService tokens,
-                            RealtimeSessionRunner runner, SessionEventHub eventHub,
-                            RedisSessionRepository redis, OperationalMetrics metrics) {
         this.mapper = mapper;
         this.sessions = sessions;
         this.tokens = tokens;
-        this.runner = runner;
-        this.eventHub = eventHub;
-        this.redis = redis;
-        this.metrics = metrics == null ? OperationalMetrics.NOOP : metrics;
-        this.leaseExecutor = redis == null ? null
-                : Executors.newSingleThreadScheduledExecutor(Thread.ofVirtual().name("runner-lease-renew").factory());
+        WebSocketEventSender sender = new WebSocketEventSender(mapper);
+        this.dispatcher = new SessionEventDispatcher(eventHub, sender);
+        this.handoffs = new HandoffSessionSubscriber(eventHub, sender);
+        this.leases = new RunnerLeaseManager(redis);
+        this.coordinator = new RealtimeSessionCoordinator(sessions, runner, leases);
     }
 
     @Override
     public void afterConnectionEstablished(WebSocketSession socket) throws Exception {
-        String id = pathVariable(socket, "sessionId");
+        String sessionId = pathVariable(socket, "sessionId");
         String token = query(socket, "token");
         if (token == null || token.isBlank()) {
-            metrics.webSocketRejected("missing_token");
-            send(socket, Map.of("type", "error", "message", "Missing WebSocket token"));
-            socket.close(CloseStatus.POLICY_VIOLATION);
+            reject(socket, "Missing WebSocket token");
             return;
         }
-        boolean handoff;
-        try {
-            handoff = tokens.valid(id, token, "handoff");
-            if (!handoff && !tokens.valid(id, token, "session")) {
-                metrics.webSocketRejected("invalid_token");
-                send(socket, Map.of("type", "error", "message", "Invalid WebSocket token"));
-                socket.close(CloseStatus.POLICY_VIOLATION);
-                return;
-            }
-        } catch (SessionTokenService.TokenStateUnavailableException error) {
-            metrics.webSocketRejected("token_state_unavailable");
-            send(socket, Map.of("type", "error", "message", "WebSocket token state unavailable"));
-            socket.close(CloseStatus.SERVER_ERROR);
+        boolean handoff = tokens.valid(sessionId, token, "handoff");
+        if (!handoff && !tokens.valid(sessionId, token, "session")) {
+            reject(socket, "Invalid WebSocket token");
             return;
         }
-        sessions.get(id);
-        if (observedSockets.add(socket.getId())) metrics.webSocketOpened();
-        if (handoff) startHandoff(socket, id);
-        else send(socket, Map.of("type", "session_started", "sessionId", id));
+        sessions.get(sessionId);
+        if (handoff) handoffs.start(socket, sessionId);
+        else dispatcher.sendDirect(socket, Map.of("type", "session_started", "sessionId", sessionId));
     }
 
     @Override
@@ -138,143 +85,37 @@ protected void handleTextMessage(WebSocketSession socket, TextMessage message) t
         try {
             payload = mapper.readTree(message.getPayload());
         } catch (Exception error) {
-            send(socket, Map.of("type", "error", "message", "Invalid JSON message"));
+            dispatcher.sendDirect(socket, Map.of("type", "error", "message", "Invalid JSON message"));
             return;
         }
         String type = payload.path("type").asText();
-        String id = pathVariable(socket, "sessionId");
-        if (handoffSubscriptions.containsKey(socket.getId())) {
+        if (handoffs.isHandoff(socket.getId())) {
             if ("stop_session".equals(type)) socket.close(CloseStatus.NORMAL);
             return;
         }
-        var session = sessions.get(id);
-        if ("start_session".equals(type)) {
-            String requestedInputMode = text(payload, "inputMode");
-            if (!SessionService.isSupportedInputMode(requestedInputMode)) {
-                send(socket, Map.of("type", "error", "message",
-                        "暂不支持的输入模式：" + requestedInputMode));
-                return;
-            }
-            synchronized (runs) {
-                if (runs.containsKey(socket.getId())) return;
-                if (activeSessionSockets.putIfAbsent(id, socket.getId()) != null) {
-                    send(socket, Map.of("type", "error", "message", "会话已在其他连接中运行"));
-                    return;
-                }
-                if (redis != null) {
-                    boolean acquired;
-                    try {
-                        acquired = redis.tryAcquireRunnerLease(id, socket.getId(),
-                                Duration.ofSeconds(RUNNER_LEASE_TTL_SECONDS));
-                    } catch (RuntimeException error) {
-                        metrics.dependencyFailure("redis");
-                        activeSessionSockets.remove(id, socket.getId());
-                        send(socket, Map.of("type", "error", "message", "会话锁服务暂不可用"));
-                        return;
-                    }
-                    if (!acquired) {
-                        activeSessionSockets.remove(id, socket.getId());
-                        send(socket, Map.of("type", "error", "message", "会话已在其他实例中运行"));
-                        return;
-                    }
-                }
-                session.applyOverrides(text(payload, "sourceLanguage"), text(payload, "targetLanguage"),
-                        text(payload, "domain"), text(payload, "inputMode"), text(payload, "sourceUrl"),
-                        text(payload, "modelProfile"));
-                session.start();
-                AtomicBoolean reportEmitted = new AtomicBoolean();
-                AtomicReference<RealtimeSessionRunner.RunHandle> handleRef = new AtomicReference<>();
-                try {
-                    // Persist the startup snapshot before another instance can read stale history.
-                    sessions.saveProgress(session);
-                    RealtimeSessionRunner.RunHandle handle = runner.start(session, event -> {
-                        if ("session_report".equals(event.get("type"))) {
-                            reportEmitted.set(true);
-                            RealtimeSessionRunner.RunHandle current = handleRef.get();
-                            if (current != null) runs.remove(socket.getId(), current);
-                            releaseRunnerOwner(id, socket.getId());
-                        }
-                        eventHub.publish(id, event);
-                        if ("session_report".equals(event.get("type"))) eventHub.complete(id);
-                        send(socket, event);
-                    });
-                    handleRef.set(handle);
-                    runs.put(socket.getId(), handle);
-                    if (reportEmitted.get()) {
-                        runs.remove(socket.getId(), handle);
-                        releaseRunnerOwner(id, socket.getId());
-                    } else {
-                        scheduleLeaseRenewal(id, socket);
-                    }
-                } catch (RuntimeException | Error error) {
-                    session.end();
-                    try {
-                        sessions.saveProgress(session);
-                    } catch (RuntimeException | Error ignored) {
-                        error.addSuppressed(ignored);
-                    }
-                    releaseRunnerOwner(id, socket.getId());
-                    throw error;
-                }
-            }
-        } else if ("stop_session".equals(type) || "audio_end".equals(type)) {
-            RealtimeSessionRunner.RunHandle handle = runs.get(socket.getId());
-            if (handle == null) {
-                String owner = activeSessionSockets.get(id);
-                if (owner != null && !owner.equals(socket.getId())) {
-                    send(socket, Map.of("type", "error", "message", "会话正在其他连接中运行"));
-                    return;
-                }
-                if (owner == null && redis != null) {
-                    boolean leaseHeld;
-                    try {
-                        leaseHeld = redis.runnerLeaseHeld(id);
-                    } catch (RuntimeException error) {
-                        metrics.dependencyFailure("redis");
-                        send(socket, Map.of("type", "error", "message", "会话锁服务暂不可用"));
-                        return;
-                    }
-                    if (leaseHeld) {
-                        send(socket, Map.of("type", "error", "message", "会话正在其他实例中运行"));
-                        return;
-                    }
-                }
-                var report = sessions.finish(id);
-                send(socket, Map.of("type", "session_report", "reportId", report.reportId(),
-                        "correctionStatus", report.correctionStatus()));
-            } else {
-                handle.stop();
+        String sessionId = pathVariable(socket, "sessionId");
+        switch (type) {
+            case "start_session" -> start(socket, sessionId, payload);
+            case "stop_session", "audio_end" -> stop(socket, sessionId);
+            case "media_clock" -> updateClock(socket, payload);
+            case "pause_session" -> {
+                coordinator.pause(socket.getId());
+                dispatcher.dispatchSourceState(sessionId, socket, "missing", "会话已暂停");
             }
-        } else if ("media_clock".equals(type)) {
-            Long playbackMs = clockValue(payload.get("playbackMs"));
-            Long sentAudioMs = clockValue(payload.get("sentAudioMs"));
-            if (playbackMs == null || sentAudioMs == null) {
-                send(socket, Map.of("type", "error", "message",
-                        "media_clock 的 playbackMs 和 sentAudioMs 必须是非负整数"));
-                return;
+            case "resume_session" -> {
+                coordinator.resume(socket.getId());
+                dispatcher.dispatchSourceState(sessionId, socket, "listening", "会话已恢复");
             }
-            RealtimeSessionRunner.RunHandle handle = runs.get(socket.getId());
-            if (handle != null) handle.updateClientClock(playbackMs, sentAudioMs);
-        } else if ("pause_session".equals(type)) {
-            RealtimeSessionRunner.RunHandle handle = runs.get(socket.getId());
-            if (handle != null) handle.pause();
-            broadcastSourceState(id, socket, "missing", "会话已暂停");
-        } else if ("resume_session".equals(type)) {
-            RealtimeSessionRunner.RunHandle handle = runs.get(socket.getId());
-            if (handle != null) handle.resume();
-            broadcastSourceState(id, socket, "listening", "会话已恢复");
-        } else {
-            send(socket, Map.of("type", "error", "message", "Unsupported client event"));
+            default -> dispatcher.sendDirect(socket, Map.of("type", "error", "message", "Unsupported client event"));
         }
     }
 
     @Override
     protected void handleBinaryMessage(WebSocketSession socket, BinaryMessage message) {
-        if (handoffSubscriptions.containsKey(socket.getId())) return;
-        RealtimeSessionRunner.RunHandle handle = runs.get(socket.getId());
-        if (handle == null) {
+        if (handoffs.isHandoff(socket.getId())) return;
+        if (!coordinator.hasRun(socket.getId())) {
             try {
-                send(socket, Map.of("type", "error", "message", "Start the session before sending audio"));
+                dispatcher.sendDirect(socket, Map.of("type", "error", "message", "Start the session before sending audio"));
             } catch (IOException error) {
                 throw new IllegalStateException("failed to send audio state", error);
             }
@@ -283,125 +124,76 @@ protected void handleBinaryMessage(WebSocketSession socket, BinaryMessage messag
         ByteBuffer payload = message.getPayload().asReadOnlyBuffer();
         byte[] bytes = new byte[payload.remaining()];
         payload.get(bytes);
-        handle.acceptAudio(bytes);
+        coordinator.acceptAudio(socket.getId(), bytes);
     }
 
     @Override
     public void afterConnectionClosed(WebSocketSession socket, CloseStatus status) {
-        RealtimeSessionRunner.RunHandle handle = runs.remove(socket.getId());
-        if (handle != null) handle.stop();
-        releaseRunnerOwner(pathVariable(socket, "sessionId"), socket.getId());
-        SessionEventHub.Subscription subscription = handoffSubscriptions.remove(socket.getId());
-        if (subscription != null) eventHub.unsubscribe(subscription);
-        Future<?> task = handoffTasks.remove(socket.getId());
-        if (task != null) task.cancel(true);
-        if (observedSockets.remove(socket.getId())) metrics.webSocketClosed(closeOutcome(status));
+        if (handoffs.close(socket.getId())) return;
+        coordinator.close(pathVariable(socket, "sessionId"), socket.getId());
     }
 
-    private void startHandoff(WebSocketSession socket, String sessionId) throws IOException {
-        SessionEventHub.Subscription subscription = eventHub.subscribe(sessionId);
-        handoffSubscriptions.put(socket.getId(), subscription);
-        send(socket, Map.of("type", "session_started", "sessionId", sessionId));
-        for (Map<String, Object> event : subscription.replay()) send(socket, event);
-        Future<?> task = handoffExecutor.submit(() -> {
-            try {
-                while (!Thread.currentThread().isInterrupted()) {
-                    Map<String, Object> event = subscription.await(1, java.util.concurrent.TimeUnit.SECONDS);
-                    if (event != null) send(socket, event);
-                }
-            } catch (InterruptedException error) {
-                Thread.currentThread().interrupt();
-            } catch (Exception ignored) {
-                // The display-only socket may close while the primary session continues.
-            }
-        });
-        handoffTasks.put(socket.getId(), task);
+    @PreDestroy
+    void shutdown() {
+        handoffs.shutdown();
+        coordinator.shutdown();
+        leases.shutdown();
     }
 
-    private void scheduleLeaseRenewal(String sessionId, WebSocketSession socket) {
-        if (redis == null || leaseExecutor == null) return;
-        ScheduledFuture<?> task = leaseExecutor.scheduleAtFixedRate(
-                () -> renewLease(sessionId, socket), RUNNER_LEASE_RENEW_SECONDS,
-                RUNNER_LEASE_RENEW_SECONDS, TimeUnit.SECONDS);
-        ScheduledFuture<?> previous = leaseRenewals.put(sessionId, task);
-        if (previous != null) previous.cancel(false);
+    private void start(WebSocketSession socket, String sessionId, JsonNode payload) throws Exception {
+        RealtimeSessionCoordinator.CommandResult result = coordinator.start(sessionId, socket.getId(),
+                new RealtimeSessionCoordinator.StartRequest(text(payload, "sourceLanguage"),
+                        text(payload, "targetLanguage"), text(payload, "domain"), text(payload, "inputMode"),
+                        text(payload, "sourceUrl"), text(payload, "modelProfile")),
+                event -> dispatcher.dispatchRunnerEvent(sessionId, socket, event));
+        if (result.failed()) dispatcher.sendDirect(socket, Map.of("type", "error", "message", result.error()));
     }
 
-    private void renewLease(String sessionId, WebSocketSession socket) {
-        String socketId = socket.getId();
-        if (!socketId.equals(activeSessionSockets.get(sessionId))) return;
-        boolean renewed;
-        try {
-            renewed = redis.renewRunnerLease(sessionId, socketId,
-                    Duration.ofSeconds(RUNNER_LEASE_TTL_SECONDS));
-        } catch (RuntimeException error) {
-            metrics.dependencyFailure("redis");
-            renewed = false;
+    private void stop(WebSocketSession socket, String sessionId) throws IOException {
+        RealtimeSessionCoordinator.CommandResult result = coordinator.stop(sessionId, socket.getId());
+        if (result.failed()) {
+            dispatcher.sendDirect(socket, Map.of("type", "error", "message", result.error()));
+            return;
         }
-        if (renewed) return;
-        RealtimeSessionRunner.RunHandle handle = runs.get(socketId);
-        if (handle != null) handle.stop();
-        releaseRunnerOwner(sessionId, socketId);
-        try {
-            send(socket, Map.of("type", "error", "message", "会话租约续期失败，已停止实时会话"));
-        } catch (IOException ignored) {
-            // The socket may already be closed while the lease task is running.
+        if (result.report() != null) {
+            dispatcher.sendDirect(socket, Map.of("type", "session_report", "reportId", result.report().reportId(),
+                    "correctionStatus", result.report().correctionStatus()));
         }
     }
 
-    private void releaseRunnerOwner(String sessionId, String socketId) {
-        String current = activeSessionSockets.get(sessionId);
-        if (current != null && !current.equals(socketId)) return;
-        activeSessionSockets.remove(sessionId, socketId);
-        ScheduledFuture<?> renewal = leaseRenewals.remove(sessionId);
-        if (renewal != null) renewal.cancel(false);
-        if (redis != null) {
-            try {
-                redis.releaseRunnerLease(sessionId, socketId);
-            } catch (RuntimeException ignored) {
-                metrics.dependencyFailure("redis");
-                // The lease has a TTL and will expire if Redis is unavailable during cleanup.
-            }
+    private void updateClock(WebSocketSession socket, JsonNode payload) throws IOException {
+        Long playbackMs = clockValue(payload.get("playbackMs"));
+        Long sentAudioMs = clockValue(payload.get("sentAudioMs"));
+        if (playbackMs == null || sentAudioMs == null) {
+            dispatcher.sendDirect(socket, Map.of("type", "error", "message",
+                    "media_clock 的 playbackMs 和 sentAudioMs 必须是非负整数"));
+            return;
         }
+        coordinator.updateClock(socket.getId(), playbackMs, sentAudioMs);
     }
 
-    @PreDestroy
-    void shutdown() {
-        handoffTasks.values().forEach(task -> task.cancel(true));
-        leaseRenewals.values().forEach(task -> task.cancel(true));
-        if (leaseExecutor != null) leaseExecutor.shutdownNow();
-        handoffExecutor.shutdownNow();
-    }
-
-    private void send(WebSocketSession socket, Object body) throws IOException {
-        synchronized (socket) {
-            socket.sendMessage(new TextMessage(mapper.writeValueAsString(body)));
-        }
+    private void reject(WebSocketSession socket, String message) throws IOException {
+        dispatcher.sendDirect(socket, Map.of("type", "error", "message", message));
+        socket.close(CloseStatus.POLICY_VIOLATION);
     }
 
-    private void broadcastSourceState(String sessionId, WebSocketSession primary,
-                                       String status, String message) throws IOException {
-        Map<String, Object> event = Map.of("type", "source_sync_state",
-                "state", Map.of("status", status, "lagMs", 0, "message", message));
-        eventHub.publish(sessionId, event);
-        send(primary, event);
-    }
     private static String text(JsonNode payload, String field) {
         JsonNode value = payload == null ? null : payload.get(field);
         return value == null || value.isNull() || value.asText().isBlank() ? null : value.asText();
     }
+
     private static Long clockValue(JsonNode value) {
         if (value == null || value.isNull()) return 0L;
         if (!value.isIntegralNumber() || !value.canConvertToLong() || value.longValue() < 0) return null;
         return value.longValue();
     }
-    private static String query(WebSocketSession socket, String key) { return socket.getUri() == null ? null : org.springframework.web.util.UriComponentsBuilder.fromUri(socket.getUri()).build().getQueryParams().getFirst(key); }
-    private static String closeOutcome(CloseStatus status) {
-        if (status == null) return "unknown";
-        if (CloseStatus.NORMAL.equals(status)) return "normal";
-        if (CloseStatus.POLICY_VIOLATION.equals(status)) return "policy";
-        return "other";
+
+    private static String query(WebSocketSession socket, String key) {
+        return socket.getUri() == null ? null
+                : org.springframework.web.util.UriComponentsBuilder.fromUri(socket.getUri()).build()
+                        .getQueryParams().getFirst(key);
     }
+
     private static String pathVariable(WebSocketSession socket, String key) {
         Object attribute = socket.getAttributes().get(key);
         if (attribute != null) return attribute.toString();
diff --git a/backend/src/main/java/com/babelflux/backend/websocket/WebSocketEventSender.java b/backend/src/main/java/com/babelflux/backend/websocket/WebSocketEventSender.java
new file mode 100644
index 0000000..52bb1c3
--- /dev/null
+++ b/backend/src/main/java/com/babelflux/backend/websocket/WebSocketEventSender.java
@@ -0,0 +1,19 @@
+package com.babelflux.backend.websocket;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.IOException;
+import org.springframework.web.socket.TextMessage;
+import org.springframework.web.socket.WebSocketSession;
+
+/** Serializes outbound events and keeps each socket write ordered. */
+final class WebSocketEventSender {
+    private final ObjectMapper mapper;
+
+    WebSocketEventSender(ObjectMapper mapper) { this.mapper = mapper; }
+
+    void send(WebSocketSession socket, Object body) throws IOException {
+        synchronized (socket) {
+            socket.sendMessage(new TextMessage(mapper.writeValueAsString(body)));
+        }
+    }
+}
diff --git a/backend/src/test/java/com/babelflux/backend/websocket/SessionWebSocketHandlerTest.java b/backend/src/test/java/com/babelflux/backend/websocket/SessionWebSocketHandlerTest.java
index 57693f3..079aabb 100644
--- a/backend/src/test/java/com/babelflux/backend/websocket/SessionWebSocketHandlerTest.java
+++ b/backend/src/test/java/com/babelflux/backend/websocket/SessionWebSocketHandlerTest.java
@@ -33,6 +33,7 @@
 import java.util.Map;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
 import org.springframework.web.socket.BinaryMessage;
 import org.springframework.web.socket.CloseStatus;
 import org.springframework.web.socket.TextMessage;
@@ -240,6 +241,35 @@ void handoffSocketReplaysEventsWithoutStartingAnotherRunner() throws Exception {
         handler.afterConnectionClosed(socket, CloseStatus.NORMAL);
     }
 
+    @Test
+    void relaysRunnerEventsToHandoffSubscribersWithoutChangingPrimaryEventContract() throws Exception {
+        SessionEventHub hub = new SessionEventHub();
+        handler = new SessionWebSocketHandler(mapper, sessions, tokens, runner, hub);
+        WebSocketSession primary = socket("ws-1", tokens.issue("ws-1"));
+        var handoff = tokens.issueHandoff("ws-1", null, "en", "zh", "bilingual");
+        WebSocketSession handoffSocket = socket("ws-1", tokens.issueHandoffWebSocket("ws-1", handoff.expiresAt()));
+        when(primary.getId()).thenReturn("primary-relay");
+        when(handoffSocket.getId()).thenReturn("handoff-relay");
+        when(runner.start(any(Session.class), any())).thenReturn(run);
+
+        handler.afterConnectionEstablished(primary);
+        handler.afterConnectionEstablished(handoffSocket);
+        handler.handleTextMessage(primary, new TextMessage("{\"type\":\"start_session\"}"));
+        ArgumentCaptor<RealtimeSessionRunner.Sink> sink = ArgumentCaptor.forClass(RealtimeSessionRunner.Sink.class);
+        verify(runner).start(any(Session.class), sink.capture());
+
+        sink.getValue().emit(Map.of("type", "translation_segment", "segment", Map.of(
+                "segmentId", "relay-1", "text", "已转发", "language", "zh",
+                "startMs", 0, "endMs", 40, "status", "final")));
+
+        var messages = org.mockito.ArgumentCaptor.forClass(TextMessage.class);
+        verify(handoffSocket, timeout(1000).atLeast(2)).sendMessage(messages.capture());
+        assertTrue(messages.getAllValues().stream()
+                .anyMatch(message -> message.getPayload().contains("translation_segment")));
+        handler.afterConnectionClosed(primary, CloseStatus.NORMAL);
+        handler.afterConnectionClosed(handoffSocket, CloseStatus.NORMAL);
+    }
+
     @Test
     void broadcastsPauseAndResumeStateToHandoffSocket() throws Exception {
         SessionEventHub hub = new SessionEventHub();
diff --git a/docs/standards/architecture.md b/docs/standards/architecture.md
index 0f4ce14..3460826 100644
--- a/docs/standards/architecture.md
+++ b/docs/standards/architecture.md
@@ -22,6 +22,23 @@ Infrastructure / provider adapters (infrastructure, messaging, search, provider)
 - `provider`、`infrastructure`、`messaging`、`search` 只能通过稳定端口向应用层提供能力，不能把第三方 SDK 类型泄漏到 API 或前端。
 - `frontend` 和 `desktop` 共享协议类型语义，但不能直接依赖后端实现细节。
 
+## 实时会话边界
+
+Java 实时会话保持单一 WebSocket 契约，但把连接适配、运行编排、租约和投送拆开，避免协议处理类同时管理线程、Redis 和字幕事件。
+
+| 组件 | 唯一职责 | 不负责 |
+| --- | --- | --- |
+| `SessionWebSocketHandler` | 鉴权、JSON/二进制帧校验、命令路由和关闭通知 | session 状态机、Redis 租约、handoff 消费线程、事件序列化细节 |
+| `RealtimeSessionCoordinator` | 启动/停止 `RealtimeSessionRunner`、保存会话启动快照、转发控制命令和结束收尾 | WebSocket 路径、token 或 JSON 解析 |
+| `RunnerLeaseManager` | 单 JVM owner 与可选 Redis runner lease 的获取、续租、释放和失败分类 | 操作 runner、发送 WebSocket 消息或持久化 session |
+| `SessionEventDispatcher` | 先写入 `SessionEventHub`，再将同一事件发送给主 socket；终态完成 replay | 修改 session 或决定租约 |
+| `HandoffSessionSubscriber` | 只读 handoff 的 replay、订阅线程和资源释放 | 启动 runner、控制音频或抢占 lease |
+| `RealtimeSessionRunner` | PCM 有界队列、provider 驱动、段落状态、纠偏和报告事件产生 | HTTP/WebSocket 鉴权、Redis lease、跨 socket fan-out |
+
+- `session_started`、`source_sync_state`、`transcript_segment`、`translation_segment`、`audio_segment`、`revision_event`、`session_report` 和 `error` 的事件名、字段与错误文案保持兼容；拆分不能借机修改公开协议。
+- 租约续期失败先停止本 socket 的 runner、释放 owner，再尽力发送错误；Redis 不可用时只把锁服务错误暴露给当前会话，不得静默接管远端 runner。
+- handoff socket 始终只读：只能关闭自身连接，不能发送 PCM、暂停、恢复或停止主 runner。
+
 ## 依赖方向
 
 依赖只能向下流动：API -> services -> providers。禁止 provider 反向导入 API，禁止服务模块互相读取对方的私有状态。需要共享行为时，提取小型纯模块或明确的领域接口。
diff --git a/docs/verification/websocket-session-orchestration.md b/docs/verification/websocket-session-orchestration.md
new file mode 100644
index 0000000..e4e3cd1
--- /dev/null
+++ b/docs/verification/websocket-session-orchestration.md
@@ -0,0 +1,48 @@
+# WebSocket 会话编排拆分验证
+
+## 范围
+
+本次调整不改 REST 路由、WebSocket URL、token 类型、客户端命令或服务端事件契约。它将原先集中在 `SessionWebSocketHandler` 的运行编排拆为独立协作组件：
+
+```text
+WebSocket frame
+      |
+SessionWebSocketHandler (auth / parse / route)
+      |
+RealtimeSessionCoordinator ---- RunnerLeaseManager ---- Redis lease
+      |                                      |
+      v                                      v
+RealtimeSessionRunner                 renewal failure cleanup
+      |
+SessionEventDispatcher ---- SessionEventHub ---- HandoffSessionSubscriber
+      |
+primary socket
+```
+
+完整职责边界和禁止依赖见 `docs/standards/architecture.md`。
+
+## 行为对比
+
+| 场景 | 拆分前后不变的行为 | 回归测试 |
+| --- | --- | --- |
+| 主连接启动 | token 通过后启动一个 runner；启动快照先落库；同 session 的第二连接被拒绝 | `startsRunAndForwardsPcmFrames`、`persistsStartupSnapshotBeforeCreatingRunner`、`rejectsSecondPrimarySocketForSameSession` |
+| Redis lease | lease 已被远端占用时本机不启动或结束会话；续租失败停止本 runner 并释放 owner | `rejectsStopFromSocketWhenAnotherInstanceHoldsRunnerLease`；续租逻辑由 `RunnerLeaseManager` 单独封装 |
+| handoff | 只读 socket 可 replay 与订阅事件，不能再启动第二个 runner | `handoffSocketReplaysEventsWithoutStartingAnotherRunner` |
+| 控制命令 | `media_clock` 校验非负整数；暂停、恢复广播兼容的 `source_sync_state` | `forwardsMediaClockAndRejectsInvalidClockValues`、`broadcastsPauseAndResumeStateToHandoffSocket` |
+| 结束/断连 | 主 runner 收 stop；无活跃 runner 时返回同一持久化报告；断连释放订阅和 owner | `audioEndWithoutActiveRunEmitsPersistedReport`、`SessionServiceDistributedFinishTest` |
+
+## 验证
+
+```text
+cd backend && mvn -B -Dtest=SessionWebSocketHandlerTest test
+结果：12 tests, 0 failures, 0 errors
+
+cd backend && mvn -B test
+结果：完整后端质量门禁通过（外部集成按环境开关跳过）
+```
+
+## 风险与回滚
+
+- 风险：组件间的关闭顺序若被后续改动打破，可能导致终态事件或 lease 释放遗漏。`RealtimeSessionCoordinator.close` 与 `RunnerLeaseManager.release` 保持幂等，测试覆盖主连接与 handoff 的关闭路径。
+- 风险：Redis 续租失败时只能保证本机停止与 TTL 最终释放，不能宣称多节点 HA。
+- 回滚：回退本 PR 即恢复单个 handler 的原有组织方式；不包含 schema、配置、API、事件字段或数据迁移。