re-land: deliver a chat turn to every socket watching it, not the newest one
This is7726c9f, reverted a few hours ago as collateral with the tabs-and-panes work. It was never a panes feature — it is the fix for a bug that predates them, and tonight it was reproduced by hand. The symptom: on https://macbook.pastilhas.dev a chat connects, works briefly, and is dead after a refresh, never coming back. On http://localhost:9010 the same build is fine. The cause is ordering. A refresh means socket B attaches before socket A's close is delivered, and `detachWs(sessionId)` took no socket argument — it nulled the session's single `ws` field, so the dying socket silenced the live one that had already replaced it. Nothing re-attaches afterwards, which is why it never came back. Over loopback the close usually lands first and it survives; via NPM on alpha and back to this host the extra latency makes the late close the ordinary case. That is the whole of the localhost/domain asymmetry. `sockets: Set` plus `detachWs(sessionId, ws)` removes only the socket that actually closed, and delivery fans out to whatever is still attached. `hasSockets` then gates the idle GC, which used to arm on ANY close — a second pane closing could collect a conversation out from under the first. Ruled out on the way, so none of it is re-investigated: the reverse proxy relays upgrades correctly (a clean 101 through openresty, and a full turn streamed end to end over wss:// with deltas and a cost line); origin validation is off (ALLOW_ANY_ORIGIN defaults true and is unset here) and never runs on the upgrade, which is a literal Bun route and never reaches Hono; authenticated HTTP is 200 through both doors; the passkeys table is empty, so no origin-bound credential is involved; and the token-resolution fix52d5678— which I nearly re-landed first — was the WRONG diagnosis, because signin writes localStorage.BEARER_TOKEN, exactly where the socket url reads. That one is still worth having for embedded and ?officerToken= hosts, but it was never this. Not verified: a browser refresh against the domain, which is Andre's to confirm — it is the only step I cannot drive from here. Typecheck clean. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -24,7 +24,7 @@ class SessionManager {
|
|||||||
cwd,
|
cwd,
|
||||||
model,
|
model,
|
||||||
piProcess: null,
|
piProcess: null,
|
||||||
ws: null,
|
sockets: new Set(),
|
||||||
lastActivity: Date.now(),
|
lastActivity: Date.now(),
|
||||||
idleTimer: null,
|
idleTimer: null,
|
||||||
streamBuffer: '',
|
streamBuffer: '',
|
||||||
@@ -145,7 +145,7 @@ class SessionManager {
|
|||||||
attachWs(sessionId: string, ws: any): void {
|
attachWs(sessionId: string, ws: any): void {
|
||||||
const session = this.sessions.get(sessionId);
|
const session = this.sessions.get(sessionId);
|
||||||
if (session) {
|
if (session) {
|
||||||
session.ws = ws;
|
session.sockets.add(ws);
|
||||||
session.lastActivity = Date.now();
|
session.lastActivity = Date.now();
|
||||||
|
|
||||||
if (session.idleTimer) {
|
if (session.idleTimer) {
|
||||||
@@ -155,14 +155,24 @@ class SessionManager {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
detachWs(sessionId: string): void {
|
/**
|
||||||
|
* Removes one socket. The caller must say WHICH — a bare `detachWs(sessionId)` used to null the
|
||||||
|
* session's only socket field, so a stale client's close event silenced whichever client had attached
|
||||||
|
* after it. A close is only the end of the conversation when nothing else is still watching.
|
||||||
|
*/
|
||||||
|
detachWs(sessionId: string, ws: any): void {
|
||||||
const session = this.sessions.get(sessionId);
|
const session = this.sessions.get(sessionId);
|
||||||
if (session) {
|
if (session) {
|
||||||
session.ws = null;
|
session.sockets.delete(ws);
|
||||||
session.lastActivity = Date.now();
|
session.lastActivity = Date.now();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** Whether anything is still watching — the idle GC must not start while another client is attached. */
|
||||||
|
hasSockets(sessionId: string): boolean {
|
||||||
|
return (this.sessions.get(sessionId)?.sockets.size ?? 0) > 0;
|
||||||
|
}
|
||||||
|
|
||||||
setIdleTimeout(sessionId: string, timeoutMs: number): void {
|
setIdleTimeout(sessionId: string, timeoutMs: number): void {
|
||||||
const session = this.sessions.get(sessionId);
|
const session = this.sessions.get(sessionId);
|
||||||
if (!session) return;
|
if (!session) return;
|
||||||
|
|||||||
@@ -299,7 +299,13 @@ export type UserSession = {
|
|||||||
cwd: string;
|
cwd: string;
|
||||||
model: string;
|
model: string;
|
||||||
piProcess: any | null;
|
piProcess: any | null;
|
||||||
ws: any | null;
|
/**
|
||||||
|
* Every socket watching this conversation, not the most recent one. Two panes in one window, or a
|
||||||
|
* laptop and an iPad on the same chat, are both ordinary now that a tab holds several panes — and a
|
||||||
|
* single `ws` field meant the newest attach silently stole the turn from everyone else, while any one
|
||||||
|
* of them closing set it to null and killed delivery for the rest.
|
||||||
|
*/
|
||||||
|
sockets: Set<any>;
|
||||||
lastActivity: number;
|
lastActivity: number;
|
||||||
idleTimer: Timer | null;
|
idleTimer: Timer | null;
|
||||||
streamBuffer: string;
|
streamBuffer: string;
|
||||||
|
|||||||
@@ -158,8 +158,10 @@ export function close(ws: ServerWebSocket<WSData>): void {
|
|||||||
|
|
||||||
const sessionId = wsToSessionMap.get(ws);
|
const sessionId = wsToSessionMap.get(ws);
|
||||||
if (sessionId) {
|
if (sessionId) {
|
||||||
sessionManager.detachWs(sessionId);
|
sessionManager.detachWs(sessionId, ws);
|
||||||
sessionManager.setIdleTimeout(sessionId, IDLE_TIMEOUT_MS);
|
// Only once nothing is watching. Another pane or another device still attached means the
|
||||||
|
// conversation is live, and arming the idle GC here would collect it out from under them.
|
||||||
|
if (!sessionManager.hasSockets(sessionId)) sessionManager.setIdleTimeout(sessionId, IDLE_TIMEOUT_MS);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -244,7 +246,7 @@ function createMessageHandler(sessionId: string, model: string) {
|
|||||||
const session = sessionManager.getSession(sessionId);
|
const session = sessionManager.getSession(sessionId);
|
||||||
if (!session) return;
|
if (!session) return;
|
||||||
foldIntoSession(session, msg, model);
|
foldIntoSession(session, msg, model);
|
||||||
sendToClient(session.ws as ServerWebSocket<WSData> | null, msg, seq);
|
for (const socket of session.sockets) sendToClient(socket as ServerWebSocket<WSData>, msg, seq);
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -651,7 +653,7 @@ async function handleResumeCursor(
|
|||||||
// the Claude sidecar about an OpenCode session, hear "not generating", and write "the agent went
|
// the Claude sidecar about an OpenCode session, hear "not generating", and write "the agent went
|
||||||
// away" into a turn that was running perfectly well.
|
// away" into a turn that was running perfectly well.
|
||||||
if (msg.generating && decision.kind !== 'assume') {
|
if (msg.generating && decision.kind !== 'assume') {
|
||||||
await endTurnIfAgentIsGone(ws, sessionId, decision.model);
|
await endTurnIfAgentIsGone([ws], sessionId, decision.model);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -821,7 +823,7 @@ async function handleAttach(ws: ServerWebSocket<WSData>, msg: { claudeSessionId:
|
|||||||
* so its sessions are left alone rather than guessed at.
|
* so its sessions are left alone rather than guessed at.
|
||||||
*/
|
*/
|
||||||
async function endTurnIfAgentIsGone(
|
async function endTurnIfAgentIsGone(
|
||||||
ws: ServerWebSocket<WSData> | null,
|
targets: Iterable<ServerWebSocket<WSData> | null>,
|
||||||
sessionId: string,
|
sessionId: string,
|
||||||
model: string,
|
model: string,
|
||||||
): Promise<void> {
|
): Promise<void> {
|
||||||
@@ -834,11 +836,11 @@ async function endTurnIfAgentIsGone(
|
|||||||
const event: ServerMessage = { type: 'cut-off' };
|
const event: ServerMessage = { type: 'cut-off' };
|
||||||
try {
|
try {
|
||||||
const seq = await appendChatEvent(sessionId, event);
|
const seq = await appendChatEvent(sessionId, event);
|
||||||
sendToClient(ws, event, seq);
|
for (const target of targets) sendToClient(target, event, seq);
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
// Still tell this client — an un-replayable explanation beats a spinner that never stops.
|
// Still tell every client — an un-replayable explanation beats a spinner that never stops.
|
||||||
logger.error('Failed to persist cut-off notice', { sessionId, error: String(err) });
|
logger.error('Failed to persist cut-off notice', { sessionId, error: String(err) });
|
||||||
sendToClient(ws, event);
|
for (const target of targets) sendToClient(target, event);
|
||||||
}
|
}
|
||||||
logger.info('Ended a turn whose agent had gone', { sessionId });
|
logger.info('Ended a turn whose agent had gone', { sessionId });
|
||||||
}
|
}
|
||||||
@@ -850,7 +852,11 @@ async function endTurnIfAgentIsGone(
|
|||||||
sidecar.onClaudeSidecarStarted(() => {
|
sidecar.onClaudeSidecarStarted(() => {
|
||||||
for (const session of sessionManager.getAllSessions()) {
|
for (const session of sessionManager.getAllSessions()) {
|
||||||
if (!session.isGenerating) continue;
|
if (!session.isGenerating) continue;
|
||||||
void endTurnIfAgentIsGone(session.ws as ServerWebSocket<WSData> | null, session.sessionId, session.model);
|
void endTurnIfAgentIsGone(
|
||||||
|
session.sockets as Set<ServerWebSocket<WSData>>,
|
||||||
|
session.sessionId,
|
||||||
|
session.model,
|
||||||
|
);
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user