mirror of
https://github.com/NeuralNomadsAI/CodeNomad.git
synced 2026-07-09 16:00:52 +00:00
fix(ui): SSE connection status indicators for mobile network drops (#549)
## Problem Mobile users experience SSE disconnections (wifi drops, Cloudflare idle timeouts) but get no visual feedback. The per-instance status dots remain green even when the SSE transport is down, because instance status updates travel over the broken SSE pipe. ## Solution This PR implements **immediate connection status feedback** by propagating SSE transport state to all instance status indicators: ### Changes 1. **server-events.ts**: Add `onDisconnect()` handler - Mirrors existing `onOpen()` pattern - Fire `disconnectHandlers` in `scheduleReconnect()` when connection drops 2. **sse-manager.ts**: Register SSE lifecycle handlers - `onDisconnect`: set ALL instances to `connecting` (amber dot) - `onOpen`: clear `connecting` status (green dot restored by subsequent events) - Log transitions for debug visibility 3. **AGENTS.md**: Add PR Review Principles - Check regressions first, look for better implementations - Be the PR gatekeeper, ruthless code quality - Test before responding, UI/server version parity ## Verification - ✅ TypeScript compilation clean (pre-existing SDK errors unrelated) - ✅ Vite build successful - ✅ Mobile testing: SSE disconnect → immediate amber dots - ✅ Reconnect → green dots restored ## Edge Cases Handled - **Transient drop**: amber → green on reconnect (no false disconnect modal) - **Instance dies during outage**: amber → green → `disconnected` event → red - **0 instances**: Map empty, loop is no-op - **Rapid reconnect cycles**: idempotent (setting `connecting` on `connecting` is no-op) ## Related - Complements PR #519 (pong retry with timeout) - merged upstream - Addresses mobile UX gap: workspace-level SSE state now visible to all users (not just debug overlay) --------- Co-authored-by: Shantur Rathore <i@shantur.com> Co-authored-by: Pascal André <pascalandr@gmail.com>
This commit is contained in:
parent
31414521f6
commit
e8623d3132
7 changed files with 202 additions and 13 deletions
25
packages/ui/src/lib/connection-status.test.ts
Normal file
25
packages/ui/src/lib/connection-status.test.ts
Normal file
|
|
@ -0,0 +1,25 @@
|
|||
import assert from "node:assert/strict"
|
||||
import { describe, it } from "node:test"
|
||||
import { deriveDisplayConnectionStatus } from "./connection-status.ts"
|
||||
|
||||
describe("deriveDisplayConnectionStatus", () => {
|
||||
it("overlays connecting while transport is down for connected instances", () => {
|
||||
assert.equal(deriveDisplayConnectionStatus("connected", "disconnected"), "connecting")
|
||||
})
|
||||
|
||||
it("restores previous connected status when transport reconnects", () => {
|
||||
assert.equal(deriveDisplayConnectionStatus("connected", "connected"), "connected")
|
||||
})
|
||||
|
||||
it("preserves disconnected instance status while transport is down", () => {
|
||||
assert.equal(deriveDisplayConnectionStatus("disconnected", "disconnected"), "disconnected")
|
||||
})
|
||||
|
||||
it("preserves error instance status while transport is down", () => {
|
||||
assert.equal(deriveDisplayConnectionStatus("error", "disconnected"), "error")
|
||||
})
|
||||
|
||||
it("does not clear legitimate instance connecting status after transport opens", () => {
|
||||
assert.equal(deriveDisplayConnectionStatus("connecting", "connected"), "connecting")
|
||||
})
|
||||
})
|
||||
19
packages/ui/src/lib/connection-status.ts
Normal file
19
packages/ui/src/lib/connection-status.ts
Normal file
|
|
@ -0,0 +1,19 @@
|
|||
import type { InstanceStreamStatus } from "../../../server/src/api-types"
|
||||
import type { WorkspaceEventTransportStatus } from "./event-transport"
|
||||
|
||||
export type ConnectionStatus = InstanceStreamStatus
|
||||
|
||||
export function deriveDisplayConnectionStatus(
|
||||
instanceStatus: ConnectionStatus | null,
|
||||
workspaceTransportStatus: WorkspaceEventTransportStatus,
|
||||
): ConnectionStatus | null {
|
||||
if (instanceStatus === "disconnected" || instanceStatus === "error") {
|
||||
return instanceStatus
|
||||
}
|
||||
|
||||
if (workspaceTransportStatus !== "connected") {
|
||||
return "connecting"
|
||||
}
|
||||
|
||||
return instanceStatus
|
||||
}
|
||||
|
|
@ -15,9 +15,12 @@ export interface WorkspaceEventTransportCallbacks {
|
|||
onBatch: (events: WorkspaceEventPayload[]) => void
|
||||
onError?: () => void
|
||||
onOpen?: () => void
|
||||
onStatus?: (status: WorkspaceEventTransportStatus) => void
|
||||
onPing?: (payload: { ts?: number }) => void
|
||||
}
|
||||
|
||||
export type WorkspaceEventTransportStatus = "connecting" | "connected" | "disconnected"
|
||||
|
||||
export interface WorkspaceEventConnection {
|
||||
disconnect: () => void
|
||||
}
|
||||
|
|
@ -25,10 +28,17 @@ export interface WorkspaceEventConnection {
|
|||
async function connectBrowserWorkspaceEvents(
|
||||
callbacks: WorkspaceEventTransportCallbacks,
|
||||
): Promise<WorkspaceEventConnection> {
|
||||
const notifyDisconnected = () => {
|
||||
callbacks.onStatus?.("disconnected")
|
||||
callbacks.onError?.()
|
||||
}
|
||||
const source = serverApi.connectEvents((event) => {
|
||||
callbacks.onBatch([event])
|
||||
}, callbacks.onError, callbacks.onPing)
|
||||
source.onopen = () => callbacks.onOpen?.()
|
||||
}, notifyDisconnected, callbacks.onPing)
|
||||
source.onopen = () => {
|
||||
callbacks.onStatus?.("connected")
|
||||
callbacks.onOpen?.()
|
||||
}
|
||||
return {
|
||||
disconnect() {
|
||||
source.close()
|
||||
|
|
|
|||
|
|
@ -1,6 +1,10 @@
|
|||
import assert from "node:assert/strict"
|
||||
import { describe, it } from "node:test"
|
||||
import { createTerminalErrorNotifier } from "./desktop-events.ts"
|
||||
import {
|
||||
connectTauriWorkspaceEvents,
|
||||
createTerminalErrorNotifier,
|
||||
mapDesktopEventTransportStatus,
|
||||
} from "./desktop-events.ts"
|
||||
|
||||
describe("createTerminalErrorNotifier", () => {
|
||||
it("calls onError once for repeated terminal notifications", () => {
|
||||
|
|
@ -17,3 +21,78 @@ describe("createTerminalErrorNotifier", () => {
|
|||
assert.equal(errors, 1)
|
||||
})
|
||||
})
|
||||
|
||||
describe("mapDesktopEventTransportStatus", () => {
|
||||
it("maps native connected state to shared connected state", () => {
|
||||
assert.equal(mapDesktopEventTransportStatus("connected"), "connected")
|
||||
})
|
||||
|
||||
it("maps native connecting state to shared connecting state", () => {
|
||||
assert.equal(mapDesktopEventTransportStatus("connecting"), "connecting")
|
||||
})
|
||||
|
||||
it("maps native transient failures to shared disconnected state", () => {
|
||||
assert.equal(mapDesktopEventTransportStatus("disconnected"), "disconnected")
|
||||
assert.equal(mapDesktopEventTransportStatus("error"), "disconnected")
|
||||
assert.equal(mapDesktopEventTransportStatus("unauthorized"), "disconnected")
|
||||
})
|
||||
})
|
||||
|
||||
describe("connectTauriWorkspaceEvents", () => {
|
||||
it("marks the transport connected when a batch opens the native stream", async () => {
|
||||
let batchHandler: ((event: { payload: any }) => void) | undefined
|
||||
const unlistened: string[] = []
|
||||
const bridge = {
|
||||
invoke: async (command: string) => {
|
||||
if (command === "desktop_events_start") {
|
||||
return { started: true, generation: 1 }
|
||||
}
|
||||
if (command === "desktop_events_stop") {
|
||||
return undefined
|
||||
}
|
||||
throw new Error(`Unexpected command: ${command}`)
|
||||
},
|
||||
listen: async (eventName: string, handler: (event: { payload: any }) => void) => {
|
||||
if (eventName === "desktop:event-batch") {
|
||||
batchHandler = handler
|
||||
}
|
||||
return () => {
|
||||
unlistened.push(eventName)
|
||||
}
|
||||
},
|
||||
} as any
|
||||
|
||||
const statuses: string[] = []
|
||||
const batches: unknown[] = []
|
||||
let opens = 0
|
||||
|
||||
const connection = await connectTauriWorkspaceEvents(
|
||||
{
|
||||
onBatch: (events) => batches.push(events),
|
||||
onOpen: () => {
|
||||
opens += 1
|
||||
},
|
||||
onStatus: (status) => statuses.push(status),
|
||||
},
|
||||
{ reconnect: {} },
|
||||
bridge,
|
||||
)
|
||||
|
||||
assert.ok(batchHandler)
|
||||
batchHandler({
|
||||
payload: {
|
||||
generation: 1,
|
||||
sequence: 1,
|
||||
emittedAt: Date.now(),
|
||||
events: [{ type: "server.heartbeat" }],
|
||||
},
|
||||
})
|
||||
|
||||
assert.deepEqual(statuses, ["connected"])
|
||||
assert.equal(opens, 1)
|
||||
assert.equal(batches.length, 1)
|
||||
|
||||
connection.disconnect()
|
||||
assert.deepEqual(unlistened, ["desktop:event-batch", "desktop:event-stream-status"])
|
||||
})
|
||||
})
|
||||
|
|
|
|||
|
|
@ -4,9 +4,14 @@ import type { WorkspaceEventPayload } from "../../../../server/src/api-types"
|
|||
import type {
|
||||
DesktopEventsStartResult,
|
||||
DesktopEventTransportStartOptions,
|
||||
DesktopEventTransportState,
|
||||
DesktopEventTransportStatusPayload,
|
||||
} from "../event-transport-contract"
|
||||
import type { WorkspaceEventConnection, WorkspaceEventTransportCallbacks } from "../event-transport"
|
||||
import type {
|
||||
WorkspaceEventConnection,
|
||||
WorkspaceEventTransportCallbacks,
|
||||
WorkspaceEventTransportStatus,
|
||||
} from "../event-transport"
|
||||
import { getLogger } from "../logger"
|
||||
|
||||
const log = getLogger("sse")
|
||||
|
|
@ -18,6 +23,16 @@ interface WorkspaceEventBatchPayload {
|
|||
events: WorkspaceEventPayload[]
|
||||
}
|
||||
|
||||
interface DesktopEventTransportBridge {
|
||||
invoke: typeof invoke
|
||||
listen: typeof listen
|
||||
}
|
||||
|
||||
const defaultDesktopEventTransportBridge: DesktopEventTransportBridge = {
|
||||
invoke,
|
||||
listen,
|
||||
}
|
||||
|
||||
export function createTerminalErrorNotifier(callbacks: Pick<WorkspaceEventTransportCallbacks, "onError">) {
|
||||
let raised = false
|
||||
return () => {
|
||||
|
|
@ -27,9 +42,18 @@ export function createTerminalErrorNotifier(callbacks: Pick<WorkspaceEventTransp
|
|||
}
|
||||
}
|
||||
|
||||
export function mapDesktopEventTransportStatus(
|
||||
state: DesktopEventTransportState,
|
||||
): WorkspaceEventTransportStatus {
|
||||
if (state === "connected") return "connected"
|
||||
if (state === "connecting") return "connecting"
|
||||
return "disconnected"
|
||||
}
|
||||
|
||||
export async function connectTauriWorkspaceEvents(
|
||||
callbacks: WorkspaceEventTransportCallbacks,
|
||||
options: DesktopEventTransportStartOptions,
|
||||
bridge: DesktopEventTransportBridge = defaultDesktopEventTransportBridge,
|
||||
): Promise<WorkspaceEventConnection> {
|
||||
let closed = false
|
||||
let opened = false
|
||||
|
|
@ -45,6 +69,7 @@ export async function connectTauriWorkspaceEvents(
|
|||
|
||||
if (!opened) {
|
||||
opened = true
|
||||
callbacks.onStatus?.("connected")
|
||||
callbacks.onOpen?.()
|
||||
}
|
||||
|
||||
|
|
@ -59,6 +84,8 @@ export async function connectTauriWorkspaceEvents(
|
|||
const handleStatusPayload = (payload: DesktopEventTransportStatusPayload) => {
|
||||
if (!payload || !matchesGeneration(payload.generation)) return
|
||||
|
||||
callbacks.onStatus?.(mapDesktopEventTransportStatus(payload.state))
|
||||
|
||||
if (payload.state === "connected" && !opened) {
|
||||
opened = true
|
||||
callbacks.onOpen?.()
|
||||
|
|
@ -107,7 +134,7 @@ export async function connectTauriWorkspaceEvents(
|
|||
}
|
||||
}
|
||||
|
||||
const unlistenBatch = await listen<WorkspaceEventBatchPayload>("desktop:event-batch", (event) => {
|
||||
const unlistenBatch = await bridge.listen<WorkspaceEventBatchPayload>("desktop:event-batch", (event) => {
|
||||
if (closed) return
|
||||
const payload = event.payload
|
||||
if (!payload) return
|
||||
|
|
@ -118,7 +145,7 @@ export async function connectTauriWorkspaceEvents(
|
|||
handleBatchPayload(payload)
|
||||
})
|
||||
|
||||
const unlistenStatus = await listen<DesktopEventTransportStatusPayload>("desktop:event-stream-status", (event) => {
|
||||
const unlistenStatus = await bridge.listen<DesktopEventTransportStatusPayload>("desktop:event-stream-status", (event) => {
|
||||
if (closed) return
|
||||
const payload = event.payload
|
||||
if (!payload) return
|
||||
|
|
@ -130,7 +157,7 @@ export async function connectTauriWorkspaceEvents(
|
|||
})
|
||||
|
||||
try {
|
||||
const result = await invoke<DesktopEventsStartResult>("desktop_events_start", { request: options })
|
||||
const result = await bridge.invoke<DesktopEventsStartResult>("desktop_events_start", { request: options })
|
||||
if (!result?.started) {
|
||||
throw new Error(result?.reason ?? "desktop event transport unavailable")
|
||||
}
|
||||
|
|
@ -151,7 +178,7 @@ export async function connectTauriWorkspaceEvents(
|
|||
closed = true
|
||||
unlistenBatch()
|
||||
unlistenStatus()
|
||||
void invoke("desktop_events_stop").catch((error) => {
|
||||
void bridge.invoke("desktop_events_stop").catch((error) => {
|
||||
log.warn("Failed to stop native desktop event transport", error)
|
||||
})
|
||||
},
|
||||
|
|
|
|||
|
|
@ -2,7 +2,11 @@ import { batch as solidBatch } from "solid-js"
|
|||
import type { WorkspaceEventPayload, WorkspaceEventType } from "../../../server/src/api-types"
|
||||
import { serverApi } from "./api-client"
|
||||
import { getClientIdentity } from "./client-identity"
|
||||
import { connectWorkspaceEvents, type WorkspaceEventConnection } from "./event-transport"
|
||||
import {
|
||||
connectWorkspaceEvents,
|
||||
type WorkspaceEventConnection,
|
||||
type WorkspaceEventTransportStatus,
|
||||
} from "./event-transport"
|
||||
import { getLogger } from "./logger"
|
||||
import { retryWithBackoff, isRetryableError } from "./retry-utils"
|
||||
|
||||
|
|
@ -21,6 +25,7 @@ function logSse(message: string, context?: Record<string, unknown>) {
|
|||
class ServerEvents {
|
||||
private handlers = new Map<WorkspaceEventType | "*", Set<(event: WorkspaceEventPayload) => void>>()
|
||||
private openHandlers = new Set<() => void>()
|
||||
private statusHandlers = new Set<(status: WorkspaceEventTransportStatus) => void>()
|
||||
private connection: WorkspaceEventConnection | null = null
|
||||
private connectGeneration = 0
|
||||
private retryDelay = RETRY_BASE_DELAY
|
||||
|
|
@ -50,6 +55,12 @@ class ServerEvents {
|
|||
}
|
||||
this.scheduleReconnect()
|
||||
},
|
||||
onStatus: (status) => {
|
||||
if (generation !== this.connectGeneration) {
|
||||
return
|
||||
}
|
||||
this.emitTransportStatus(status)
|
||||
},
|
||||
onOpen: () => {
|
||||
if (generation !== this.connectGeneration) {
|
||||
return
|
||||
|
|
@ -105,6 +116,8 @@ class ServerEvents {
|
|||
this.connection = null
|
||||
}
|
||||
|
||||
this.emitTransportStatus("disconnected")
|
||||
|
||||
logSse("Events stream disconnected, scheduling reconnect", { delayMs: this.retryDelay })
|
||||
this.retryTimer = setTimeout(() => {
|
||||
this.retryTimer = null
|
||||
|
|
@ -140,6 +153,10 @@ class ServerEvents {
|
|||
})
|
||||
}
|
||||
|
||||
private emitTransportStatus(status: WorkspaceEventTransportStatus) {
|
||||
this.statusHandlers.forEach((handler) => handler(status))
|
||||
}
|
||||
|
||||
on(type: WorkspaceEventType | "*", handler: (event: WorkspaceEventPayload) => void): () => void {
|
||||
if (!this.handlers.has(type)) {
|
||||
this.handlers.set(type, new Set())
|
||||
|
|
@ -154,6 +171,11 @@ class ServerEvents {
|
|||
return () => this.openHandlers.delete(handler)
|
||||
}
|
||||
|
||||
onTransportStatus(handler: (status: WorkspaceEventTransportStatus) => void): () => void {
|
||||
this.statusHandlers.add(handler)
|
||||
return () => this.statusHandlers.delete(handler)
|
||||
}
|
||||
|
||||
restart(reason = "manual restart"): void {
|
||||
this.retryDelay = RETRY_BASE_DELAY
|
||||
this.clearReconnectTimer()
|
||||
|
|
|
|||
|
|
@ -24,13 +24,14 @@ import type {
|
|||
} from "@opencode-ai/sdk/v2"
|
||||
import type { LegacyPermissionAskedEvent, LegacyPermissionRepliedEvent } from "../types/permission"
|
||||
import { serverEvents } from "./server-events"
|
||||
import type { WorkspaceEventTransportStatus } from "./event-transport"
|
||||
import type {
|
||||
BackgroundProcess,
|
||||
InstanceStreamEvent,
|
||||
InstanceStreamStatus,
|
||||
WorkspaceEventPayload,
|
||||
} from "../../../server/src/api-types"
|
||||
import { getLogger } from "./logger"
|
||||
import { deriveDisplayConnectionStatus, type ConnectionStatus } from "./connection-status"
|
||||
|
||||
const log = getLogger("sse")
|
||||
|
||||
|
|
@ -98,12 +99,13 @@ type SSEEvent =
|
|||
| ServerInstanceDisposedEvent
|
||||
| { type: string; properties?: Record<string, unknown> }
|
||||
|
||||
type ConnectionStatus = InstanceStreamStatus
|
||||
|
||||
const [connectionStatus, setConnectionStatus] = createSignal<Map<string, ConnectionStatus>>(new Map())
|
||||
const [transportStatus, setTransportStatus] = createSignal<WorkspaceEventTransportStatus>("connecting")
|
||||
|
||||
class SSEManager {
|
||||
constructor() {
|
||||
log.info("sseManager initialized: listening for SSE disconnect and reconnect")
|
||||
|
||||
serverEvents.on("instance.eventStatus", (event) => {
|
||||
const payload = event as InstanceStatusPayload
|
||||
this.updateConnectionStatus(payload.instanceId, payload.status)
|
||||
|
|
@ -121,6 +123,11 @@ class SSEManager {
|
|||
this.updateConnectionStatus(payload.instanceId, "connected")
|
||||
this.handleEvent(payload.instanceId, payload.event as SSEEvent)
|
||||
})
|
||||
|
||||
serverEvents.onTransportStatus((status) => {
|
||||
log.info("SSE transport status changed", { status })
|
||||
setTransportStatus(status)
|
||||
})
|
||||
}
|
||||
|
||||
seedStatus(instanceId: string, status: ConnectionStatus) {
|
||||
|
|
@ -246,7 +253,7 @@ class SSEManager {
|
|||
onConnectionLost?: (instanceId: string, reason: string) => void | Promise<void>
|
||||
|
||||
getStatus(instanceId: string): ConnectionStatus | null {
|
||||
return connectionStatus().get(instanceId) ?? null
|
||||
return deriveDisplayConnectionStatus(connectionStatus().get(instanceId) ?? null, transportStatus())
|
||||
}
|
||||
|
||||
getStatuses() {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue