fix: authenticate event-stream WebSocket before connecting
The global event-stream WebSocket opened before a valid oc_url_token was
minted, so the upgrade failed auth ("no valid credentials available") in
packaged builds with a UI password. The resulting reconnect storm churned
the sync store and made session status flicker busy<->idle. Await the URL
auth token before connecting (a WS upgrade can't send a bearer header like
SSE does) and drop a rejected token on pre-ready close so the next attempt
re-mints a fresh one.
Also harden /session/status reconciliation: the watchdog poll is now
monotonic (only confirms/raises active status, never blindly lowers a
busy/retry session to idle on a transient or misscoped snapshot). Idle is
applied only by the authoritative reconnect/escalation resync, which trusts
the live server snapshot as the source of truth. Add a Help -> Toggle
Developer Tools menu item so production builds can open the console.
This commit is contained in:
@@ -42,7 +42,7 @@ const buildAuthUrl = (apiBaseUrl: string | null | undefined, path: string): stri
|
||||
}
|
||||
};
|
||||
|
||||
const clearRuntimeUrlAuthToken = (): void => {
|
||||
export const clearRuntimeUrlAuthToken = (): void => {
|
||||
runtimeUrlAuthToken = '';
|
||||
runtimeUrlAuthTokenExpiresAt = 0;
|
||||
};
|
||||
|
||||
@@ -0,0 +1,117 @@
|
||||
import { describe, expect, test } from "bun:test"
|
||||
import { create, type StoreApi } from "zustand"
|
||||
import type { SessionStatus } from "@opencode-ai/sdk/v2/client"
|
||||
|
||||
import { INITIAL_STATE, type State } from "../types"
|
||||
import type { DirectoryStore } from "../child-store"
|
||||
import {
|
||||
applySessionStatusSnapshot,
|
||||
needsSnapshotAfterStatusPoll,
|
||||
} from "../sync-context"
|
||||
|
||||
type StatusSnapshot = Record<string, { type: "idle" | "busy" | "retry"; attempt?: number; message?: string; next?: number }>
|
||||
|
||||
function createDirectoryStore(initial: Partial<State>): StoreApi<DirectoryStore> {
|
||||
return create<DirectoryStore>()((set) => ({
|
||||
...INITIAL_STATE,
|
||||
...initial,
|
||||
session: initial.session ?? [],
|
||||
patch: (partial) => set(partial),
|
||||
replace: (next) => set(next),
|
||||
}))
|
||||
}
|
||||
|
||||
function streamingMessage() {
|
||||
// Trailing assistant message with no `time.completed` → actively streaming.
|
||||
return [{ id: "msg_1", role: "assistant", time: { created: 1 } }] as unknown as State["message"][string]
|
||||
}
|
||||
|
||||
function completedMessage() {
|
||||
return [{ id: "msg_1", role: "assistant", time: { created: 1, completed: 2 } }] as unknown as State["message"][string]
|
||||
}
|
||||
|
||||
const BUSY: SessionStatus = { type: "busy" }
|
||||
|
||||
describe("applySessionStatusSnapshot", () => {
|
||||
describe("monotonic mode (periodic poll)", () => {
|
||||
test("does NOT lower a busy session to idle when the snapshot omits it", () => {
|
||||
const store = createDirectoryStore({ session_status: { ses_a: BUSY } })
|
||||
const changed = applySessionStatusSnapshot(store, {} as StatusSnapshot, ["ses_a"], "monotonic")
|
||||
expect(changed).toBe(false)
|
||||
expect(store.getState().session_status.ses_a).toEqual(BUSY)
|
||||
})
|
||||
|
||||
test("does NOT lower a busy session even when the snapshot reports it idle", () => {
|
||||
const store = createDirectoryStore({ session_status: { ses_a: BUSY } })
|
||||
applySessionStatusSnapshot(store, { ses_a: { type: "idle" } }, ["ses_a"], "monotonic")
|
||||
expect(store.getState().session_status.ses_a).toEqual(BUSY)
|
||||
})
|
||||
|
||||
test("raises an idle/unknown session to busy when the snapshot reports it active (missed event)", () => {
|
||||
const store = createDirectoryStore({ session_status: {} })
|
||||
const changed = applySessionStatusSnapshot(store, { ses_a: { type: "busy" } }, ["ses_a"], "monotonic")
|
||||
expect(changed).toBe(true)
|
||||
expect(store.getState().session_status.ses_a).toEqual(BUSY)
|
||||
})
|
||||
|
||||
test("updates busy → retry from the snapshot", () => {
|
||||
const store = createDirectoryStore({ session_status: { ses_a: BUSY } })
|
||||
const retry: SessionStatus = { type: "retry", attempt: 2, message: "x", next: 30 }
|
||||
applySessionStatusSnapshot(store, { ses_a: { type: "retry", attempt: 2, message: "x", next: 30 } }, ["ses_a"], "monotonic")
|
||||
expect(store.getState().session_status.ses_a).toEqual(retry)
|
||||
})
|
||||
})
|
||||
|
||||
describe("authoritative mode (reconnect / escalated resync)", () => {
|
||||
test("lowers a busy session to idle when the snapshot omits it", () => {
|
||||
const store = createDirectoryStore({
|
||||
session_status: { ses_a: BUSY },
|
||||
message: { ses_a: completedMessage() },
|
||||
})
|
||||
const changed = applySessionStatusSnapshot(store, {} as StatusSnapshot, ["ses_a"], "authoritative")
|
||||
expect(changed).toBe(true)
|
||||
expect(store.getState().session_status.ses_a).toEqual({ type: "idle" })
|
||||
})
|
||||
|
||||
test("snapshot is the source of truth: lowers to idle even if the trailing message looks unfinished", () => {
|
||||
// The live /session/status snapshot wins over derived message state — a
|
||||
// stale/lost message.updated must never pin a session busy after the
|
||||
// server says idle. (Recovery from a missed idle event.)
|
||||
const store = createDirectoryStore({
|
||||
session_status: { ses_a: BUSY },
|
||||
message: { ses_a: streamingMessage() },
|
||||
})
|
||||
const changed = applySessionStatusSnapshot(store, {} as StatusSnapshot, ["ses_a"], "authoritative")
|
||||
expect(changed).toBe(true)
|
||||
expect(store.getState().session_status.ses_a).toEqual({ type: "idle" })
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
describe("needsSnapshotAfterStatusPoll", () => {
|
||||
test("escalates when the store says busy but the snapshot omits it", () => {
|
||||
const store = createDirectoryStore({
|
||||
session_status: { ses_a: BUSY },
|
||||
message: { ses_a: completedMessage() },
|
||||
})
|
||||
expect(needsSnapshotAfterStatusPoll(store.getState(), "ses_a", undefined)).toBe(true)
|
||||
})
|
||||
|
||||
test("escalates regardless of a still-streaming trailing message (snapshot drives recovery)", () => {
|
||||
const store = createDirectoryStore({
|
||||
session_status: { ses_a: BUSY },
|
||||
message: { ses_a: streamingMessage() },
|
||||
})
|
||||
expect(needsSnapshotAfterStatusPoll(store.getState(), "ses_a", undefined)).toBe(true)
|
||||
})
|
||||
|
||||
test("does NOT escalate when the snapshot confirms the session is active", () => {
|
||||
const store = createDirectoryStore({ session_status: { ses_a: BUSY } })
|
||||
expect(needsSnapshotAfterStatusPoll(store.getState(), "ses_a", { type: "busy" })).toBe(false)
|
||||
})
|
||||
|
||||
test("does NOT escalate when the store already considers the session idle", () => {
|
||||
const store = createDirectoryStore({ session_status: {} })
|
||||
expect(needsSnapshotAfterStatusPoll(store.getState(), "ses_a", undefined)).toBe(false)
|
||||
})
|
||||
})
|
||||
@@ -15,6 +15,7 @@
|
||||
import type { Event, OpencodeClient, SessionStatus } from "@opencode-ai/sdk/v2/client"
|
||||
import { opencodeClient } from "@/lib/opencode/client"
|
||||
import { getRuntimeUrlResolver } from "@/lib/runtime-url"
|
||||
import { clearRuntimeUrlAuthToken, refreshRuntimeUrlAuthToken } from "@/lib/runtime-auth"
|
||||
import { syncDebug } from "./debug"
|
||||
|
||||
export type QueuedEvent = {
|
||||
@@ -529,6 +530,28 @@ export function createEventPipeline(input: EventPipelineInput): EventPipeline {
|
||||
}
|
||||
|
||||
const runWsAttempt = async (signal: AbortSignal) => {
|
||||
// A WebSocket upgrade can't carry an Authorization header, so it
|
||||
// authenticates purely via the oc_url_token query param. The sync token
|
||||
// getter returns "" while the token is unminted or inside its expiry skew
|
||||
// window, which would open the socket WITHOUT credentials — the server then
|
||||
// rejects it ("HTTP Authentication failed; no valid credentials available")
|
||||
// and the resulting reconnect storm churns the sync store (transient
|
||||
// status-missing → idle flicker). Mint/await a valid token BEFORE
|
||||
// connecting. (SSE avoids this: the SDK fetch sends the bearer header.)
|
||||
try {
|
||||
await refreshRuntimeUrlAuthToken()
|
||||
} catch (error) {
|
||||
const wrapped = error instanceof Error ? error : new Error("Message stream WebSocket auth token unavailable")
|
||||
if (transport === "auto") {
|
||||
wsFallbackUntil = Date.now() + WS_FALLBACK_WINDOW_MS
|
||||
;(wrapped as Error & { code?: string }).code = "WS_FALLBACK"
|
||||
}
|
||||
;(wrapped as Error & { reason?: string }).reason = "ws_auth_token_unavailable"
|
||||
throw wrapped
|
||||
}
|
||||
if (signal.aborted) {
|
||||
throw new DOMException("Aborted", "AbortError")
|
||||
}
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
let settled = false
|
||||
let opened = false
|
||||
@@ -677,6 +700,14 @@ export function createEventPipeline(input: EventPipelineInput): EventPipeline {
|
||||
? `ws_closed:code=${event?.code ?? "?"}`
|
||||
: "ws_closed_before_ready"
|
||||
|
||||
// Closed before the socket ever opened → the server rejected the
|
||||
// upgrade, typically an auth failure on the oc_url_token. Drop the
|
||||
// cached token so the next attempt mints a fresh one instead of
|
||||
// replaying a token the server won't accept (which would loop).
|
||||
if (!opened) {
|
||||
clearRuntimeUrlAuthToken()
|
||||
}
|
||||
|
||||
// If the WS stream connects (ready) but then drops quickly, prefer SSE for a while.
|
||||
// This avoids tight reconnect loops with repeated console spam.
|
||||
const livedMs = readyAt > 0 ? Date.now() - readyAt : 0
|
||||
|
||||
@@ -420,7 +420,7 @@ function getViewedSessionMaterializationTarget(directory: string) {
|
||||
}
|
||||
}
|
||||
|
||||
function toSessionStatus(status: Awaited<ReturnType<typeof opencodeClient.getSessionStatus>>[string]): SessionStatus | undefined {
|
||||
function toSessionStatus(status: Awaited<ReturnType<typeof opencodeClient.getSessionStatus>>[string] | undefined): SessionStatus | undefined {
|
||||
if (!status) return undefined
|
||||
if (status.type === "idle" || status.type === "busy") {
|
||||
return { type: status.type }
|
||||
@@ -453,40 +453,64 @@ function getActiveSessionCandidateIds(directory: string, state: DirectoryStore):
|
||||
})
|
||||
}
|
||||
|
||||
function buildRelevantSessionStatuses(
|
||||
nextStatuses: Awaited<ReturnType<typeof opencodeClient.getSessionStatusForDirectory>>,
|
||||
candidateSessionIds: string[],
|
||||
): Record<string, SessionStatus> | null {
|
||||
if (nextStatuses === null) return null
|
||||
const relevantStatuses: Record<string, SessionStatus> = {}
|
||||
for (const sessionId of candidateSessionIds) {
|
||||
relevantStatuses[sessionId] = toSessionStatus(nextStatuses[sessionId]) ?? { type: "idle" }
|
||||
}
|
||||
return relevantStatuses
|
||||
}
|
||||
type DirectorySessionStatusSnapshot = NonNullable<
|
||||
Awaited<ReturnType<typeof opencodeClient.getSessionStatusForDirectory>>
|
||||
>
|
||||
|
||||
function applySessionStatusSnapshot(
|
||||
// How a /session/status snapshot is reconciled into the store.
|
||||
//
|
||||
// The directory-scoped snapshot lists only active (busy/retry) sessions; an
|
||||
// absent candidate means "idle per this snapshot".
|
||||
//
|
||||
// - "monotonic": only confirm/raise active status. Never lowers a busy/retry
|
||||
// session to idle. Used by the periodic watchdog poll — real idle arrives via
|
||||
// SSE (session.status / session.idle) or via an authoritative resync that the
|
||||
// watchdog escalates to when it detects a stale busy entry. This keeps the
|
||||
// blind 5s poll from clobbering live state on a transient/misscoped snapshot.
|
||||
// - "authoritative": treat the snapshot as ground truth — absent/idle candidates
|
||||
// are lowered to idle. Used by reconnect/escalated resyncs, a deliberate edge
|
||||
// where the live server snapshot is the source of truth (mirrors the bootstrap
|
||||
// snapshot). The snapshot wins over any derived message state here.
|
||||
type StatusSnapshotMode = "monotonic" | "authoritative"
|
||||
|
||||
export function applySessionStatusSnapshot(
|
||||
store: StoreApi<DirectoryStore>,
|
||||
relevantStatuses: Record<string, SessionStatus>,
|
||||
snapshot: DirectorySessionStatusSnapshot,
|
||||
candidateSessionIds: string[],
|
||||
mode: StatusSnapshotMode,
|
||||
): boolean {
|
||||
if (Object.keys(relevantStatuses).length === 0) return false
|
||||
if (candidateSessionIds.length === 0) return false
|
||||
|
||||
let changed = false
|
||||
store.setState((state: DirectoryStore) => {
|
||||
for (const [sessionId, nextStatus] of Object.entries(relevantStatuses)) {
|
||||
if (!haveEquivalentSyncSnapshots(state.session_status?.[sessionId], nextStatus)) {
|
||||
const current = state.session_status ?? {}
|
||||
let next: Record<string, SessionStatus> | undefined
|
||||
const draft = () => (next ??= { ...current })
|
||||
|
||||
for (const sessionId of candidateSessionIds) {
|
||||
const incoming = toSessionStatus(snapshot[sessionId])
|
||||
|
||||
if (incoming && incoming.type !== "idle") {
|
||||
// Confirm or raise active status (catches a busy event the SSE missed).
|
||||
if (!haveEquivalentSyncSnapshots(current[sessionId], incoming)) {
|
||||
draft()[sessionId] = incoming
|
||||
changed = true
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
// Snapshot reports this candidate idle (absent, or explicit idle).
|
||||
// Monotonic never lowers; authoritative trusts the snapshot as truth.
|
||||
if (mode === "monotonic") continue
|
||||
|
||||
const existing = current[sessionId]
|
||||
if (existing && existing.type !== "idle") {
|
||||
draft()[sessionId] = { type: "idle" }
|
||||
changed = true
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if (!changed) {
|
||||
return state
|
||||
}
|
||||
|
||||
return {
|
||||
session_status: { ...state.session_status, ...relevantStatuses },
|
||||
}
|
||||
return next ? { session_status: next } : state
|
||||
})
|
||||
|
||||
return changed
|
||||
@@ -496,30 +520,29 @@ async function resyncDirectorySessionStatuses(
|
||||
directory: string,
|
||||
store: StoreApi<DirectoryStore>,
|
||||
candidateSessionIds: string[],
|
||||
): Promise<Record<string, SessionStatus> | null> {
|
||||
mode: StatusSnapshotMode,
|
||||
): Promise<DirectorySessionStatusSnapshot | null> {
|
||||
const nextStatuses = await opencodeClient.getSessionStatusForDirectory(directory)
|
||||
// null = fetch failed; preserve existing state. {} or populated = authoritative
|
||||
// snapshot of active sessions — candidates not listed are idle now.
|
||||
const relevantStatuses = buildRelevantSessionStatuses(nextStatuses, candidateSessionIds)
|
||||
if (relevantStatuses === null) return null
|
||||
applySessionStatusSnapshot(store, relevantStatuses)
|
||||
return relevantStatuses
|
||||
// null = fetch failed; preserve existing state. {} or populated = a snapshot
|
||||
// of active sessions — reconciled per `mode` (absence ≠ idle under monotonic).
|
||||
if (nextStatuses === null) return null
|
||||
applySessionStatusSnapshot(store, nextStatuses, candidateSessionIds, mode)
|
||||
return nextStatuses
|
||||
}
|
||||
|
||||
function needsSnapshotAfterStatusPoll(
|
||||
// After a monotonic poll, decide whether to escalate to a full authoritative
|
||||
// resync: the store believes the session is active but the snapshot reports it
|
||||
// idle/absent — a suspected missed idle that the monotonic poll deliberately
|
||||
// won't lower on its own. The authoritative resync is the recovery path.
|
||||
export function needsSnapshotAfterStatusPoll(
|
||||
state: DirectoryStore,
|
||||
sessionId: string,
|
||||
nextStatus: SessionStatus | undefined,
|
||||
snapshotEntry: DirectorySessionStatusSnapshot[string] | undefined,
|
||||
): boolean {
|
||||
if (nextStatus?.type !== "idle") return false
|
||||
const incoming = toSessionStatus(snapshotEntry)
|
||||
if (incoming && incoming.type !== "idle") return false
|
||||
const currentStatus = state.session_status?.[sessionId]
|
||||
if (currentStatus && currentStatus.type !== "idle") return true
|
||||
|
||||
const messages = state.message[sessionId]
|
||||
const lastMessage = messages?.[messages.length - 1]
|
||||
return !!lastMessage
|
||||
&& lastMessage.role === "assistant"
|
||||
&& typeof (lastMessage as { time?: { completed?: number } }).time?.completed !== "number"
|
||||
return Boolean(currentStatus && currentStatus.type !== "idle")
|
||||
}
|
||||
|
||||
type EventRoutingIndex = {
|
||||
@@ -1177,7 +1200,7 @@ async function resyncDirectoryAfterReconnect(
|
||||
const candidateSessionIds = getActiveSessionCandidateIds(directory, current)
|
||||
if (candidateSessionIds.length === 0) return
|
||||
|
||||
await resyncDirectorySessionStatuses(directory, store, candidateSessionIds)
|
||||
await resyncDirectorySessionStatuses(directory, store, candidateSessionIds, "authoritative")
|
||||
|
||||
const scopedClient = opencodeClient.getScopedSdkClient(directory)
|
||||
await Promise.all(candidateSessionIds.map(async (sessionId) => {
|
||||
@@ -1868,7 +1891,7 @@ export function SyncProvider(props: {
|
||||
polling.add(directory)
|
||||
try {
|
||||
const before = store.getState()
|
||||
const statuses = await resyncDirectorySessionStatuses(directory, store, candidateSessionIds)
|
||||
const statuses = await resyncDirectorySessionStatuses(directory, store, candidateSessionIds, "monotonic")
|
||||
if (!statuses) return
|
||||
const needsSnapshot = candidateSessionIds.some((sessionId) => (
|
||||
needsSnapshotAfterStatusPoll(before, sessionId, statuses[sessionId])
|
||||
|
||||
Reference in New Issue
Block a user