ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(tui): extract event stream connection (#38872)

Browse source
This commit is contained in:
James Long 2026-07-27 16:31:21 -04:00 • committed by GitHub
commit 4e4cf9e25e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 698 additions and 45 deletions
Download patch file Download diff file Expand all files Collapse all files

109
packages/tui/src/context/client.tsx
View file

@ -49,6 +49,67 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext(
if (history.length > connectionHistoryLimit) history.shift()
}
async function connect(signal: AbortSignal, attempt: number) {
let connectedAt: number | undefined
// Bound the initial handshake and tie this request to the stream lifetime.
const request = new AbortController()
const cancel = () => request.abort(signal.reason)
const timeout = setTimeout(() => request.abort(new Error("Timed out connecting to server")), connectTimeout)
signal.addEventListener("abort", cancel, { once: true })
try {
// Open the event stream and validate its initial handshake.
record(attempt === 0 ? "connecting" : "reconnecting", attempt)
log.info("event stream connecting", { attempt })
const iterator = api.event.subscribe({ signal: request.signal })[Symbol.asyncIterator]()
const first = await iterator.next()
if (signal.aborted) return { error: undefined, connectedAt }
if (first.done) {
const error =
request.signal.reason instanceof Error ? request.signal.reason : new Error("Event stream disconnected")
return { error, connectedAt }
}
if (first.value.type !== "server.connected")
return { error: new Error("Event stream did not start with server.connected"), connectedAt }
// Publish the connected state before forwarding live events.
clearTimeout(timeout)
record("connected", attempt)
connectedAt = Date.now()
log.info("event stream connected")
events.emit(first.value.type, first.value)
setConnection({ status: "connected", attempt: 0, error: undefined })
// Forward events until the stream closes or this connection is cancelled.
while (!signal.aborted) {
const event = await iterator.next()
if (signal.aborted) return { error: undefined, connectedAt }
if (event.done) return { error: new Error("Event stream disconnected"), connectedAt }
if ("durable" in event.value)
log.debug("event", {
type: event.value.type,
aggregateID: event.value.durable.aggregateID,
seq: event.value.durable.seq,
})
events.emit(event.value.type, event.value)
}
return { error: undefined, connectedAt }
} catch (error) {
return { error, connectedAt }
} finally {
request.abort()
clearTimeout(timeout)
signal.removeEventListener("abort", cancel)
}
}
function start() {
stream?.abort()
const controller = new AbortController()
@ -56,53 +117,11 @@ export const { use: useClient, provider: ClientProvider } = createSimpleContext(
void (async () => {
let attempt = 0
while (!abort.signal.aborted && !controller.signal.aborted) {
let connectedAt: number | undefined
const request = new AbortController()
const cancel = () => request.abort(controller.signal.reason)
const timeout = setTimeout(() => request.abort(new Error("Timed out connecting to server")), connectTimeout)
controller.signal.addEventListener("abort", cancel, { once: true })
const error = await (async () => {
record(attempt === 0 ? "connecting" : "reconnecting", attempt)
log.info("event stream connecting", { attempt })
const iterator = api.event.subscribe({ signal: request.signal })[Symbol.asyncIterator]()
const first = await iterator.next()
if (abort.signal.aborted || controller.signal.aborted) return undefined
if (first.done)
return request.signal.reason instanceof Error
? request.signal.reason
: new Error("Event stream disconnected")
if (first.value.type !== "server.connected")
return new Error("Event stream did not start with server.connected")
clearTimeout(timeout)
record("connected", attempt)
connectedAt = Date.now()
log.info("event stream connected")
events.emit(first.value.type, first.value)
setConnection({ status: "connected", attempt: 0, error: undefined })
while (!abort.signal.aborted && !controller.signal.aborted) {
const event = await iterator.next()
if (abort.signal.aborted || controller.signal.aborted) return undefined
if (event.done) return new Error("Event stream disconnected")
if ("durable" in event.value)
log.debug("event", {
type: event.value.type,
aggregateID: event.value.durable.aggregateID,
seq: event.value.durable.seq,
})
events.emit(event.value.type, event.value)
}
return undefined
})()
.catch((error) => error)
.finally(() => {
request.abort()
clearTimeout(timeout)
controller.signal.removeEventListener("abort", cancel)
})
const result = await connect(controller.signal, attempt)
if (abort.signal.aborted || controller.signal.aborted) return
if (connectedAt !== undefined && Date.now() - connectedAt >= 1_000) attempt = 0
if (result.connectedAt !== undefined && Date.now() - result.connectedAt >= 1_000) attempt = 0
attempt += 1
const message = errorMessage(error)
const message = errorMessage(result.error)
record("disconnected", attempt, message)
log.info("event stream disconnected", {
attempt,