Track session status via SSE updates
This commit is contained in:
@@ -13,6 +13,7 @@ import type {
|
||||
EventSessionError,
|
||||
EventSessionIdle,
|
||||
EventSessionUpdated,
|
||||
EventSessionStatus,
|
||||
} from "@opencode-ai/sdk"
|
||||
import { serverEvents } from "./server-events"
|
||||
import type {
|
||||
@@ -134,6 +135,9 @@ class SSEManager {
|
||||
case "session.idle":
|
||||
this.onSessionIdle?.(instanceId, event as EventSessionIdle)
|
||||
break
|
||||
case "session.status":
|
||||
this.onSessionStatus?.(instanceId, event as EventSessionStatus)
|
||||
break
|
||||
case "permission.updated":
|
||||
this.onPermissionUpdated?.(instanceId, event as EventPermissionUpdated)
|
||||
break
|
||||
@@ -171,6 +175,7 @@ class SSEManager {
|
||||
onSessionError?: (instanceId: string, event: EventSessionError) => void
|
||||
onTuiToast?: (instanceId: string, event: TuiToastEvent) => void
|
||||
onSessionIdle?: (instanceId: string, event: EventSessionIdle) => void
|
||||
onSessionStatus?: (instanceId: string, event: EventSessionStatus) => void
|
||||
onPermissionUpdated?: (instanceId: string, event: EventPermissionUpdated) => void
|
||||
onPermissionReplied?: (instanceId: string, event: EventPermissionReplied) => void
|
||||
onLspUpdated?: (instanceId: string, event: EventLspUpdated) => void
|
||||
|
||||
@@ -19,6 +19,7 @@ import {
|
||||
setProviders,
|
||||
setSessionInfoByInstance,
|
||||
setSessions,
|
||||
setSessionStatus,
|
||||
sessions,
|
||||
loading,
|
||||
setLoading,
|
||||
@@ -27,6 +28,7 @@ import {
|
||||
import { DEFAULT_MODEL_OUTPUT_LIMIT, getDefaultModel, isModelValid } from "./session-models"
|
||||
import { normalizeMessagePart } from "./message-v2/normalizers"
|
||||
import { updateSessionInfo } from "./message-v2/session-info"
|
||||
import { deriveSessionStatusFromMessages } from "./session-status"
|
||||
import { seedSessionMessagesV2 } from "./message-v2/bridge"
|
||||
import { messageStoreBus } from "./message-v2/bus"
|
||||
import { clearCacheForSession } from "../lib/global-cache"
|
||||
@@ -82,6 +84,10 @@ async function fetchSessions(instanceId: string): Promise<void> {
|
||||
for (const apiSession of response.data) {
|
||||
const existingSession = existingSessions?.get(apiSession.id)
|
||||
|
||||
const compactingFlag = (apiSession.time as (Session["time"] & { compacting?: number | boolean }) | undefined)?.compacting
|
||||
const isCompacting = typeof compactingFlag === "number" ? compactingFlag > 0 : Boolean(compactingFlag)
|
||||
const existingStatus = existingSession?.status
|
||||
|
||||
sessionMap.set(apiSession.id, {
|
||||
id: apiSession.id,
|
||||
instanceId,
|
||||
@@ -89,6 +95,7 @@ async function fetchSessions(instanceId: string): Promise<void> {
|
||||
parentId: apiSession.parentID || null,
|
||||
agent: existingSession?.agent ?? "",
|
||||
model: existingSession?.model ?? { providerId: "", modelId: "" },
|
||||
status: isCompacting ? "compacting" : (existingStatus ?? "idle"),
|
||||
version: apiSession.version,
|
||||
time: {
|
||||
...apiSession.time,
|
||||
@@ -183,6 +190,7 @@ async function createSession(instanceId: string, agent?: string): Promise<Sessio
|
||||
parentId: null,
|
||||
agent: selectedAgent,
|
||||
model: defaultModel,
|
||||
status: "idle",
|
||||
version: response.data.version,
|
||||
time: {
|
||||
...response.data.time,
|
||||
@@ -290,6 +298,7 @@ async function forkSession(
|
||||
providerId: info.model?.providerID || "",
|
||||
modelId: info.model?.modelID || "",
|
||||
},
|
||||
status: "idle",
|
||||
version: "0",
|
||||
time: info.time ? { ...info.time } : { created: Date.now(), updated: Date.now() },
|
||||
revert: info.revert
|
||||
@@ -606,6 +615,11 @@ async function loadMessages(instanceId: string, sessionId: string, force = false
|
||||
}
|
||||
seedSessionMessagesV2(instanceId, sessionForV2, messages, messagesInfo)
|
||||
|
||||
if (!alreadyLoaded) {
|
||||
const nextStatus = deriveSessionStatusFromMessages(instanceId, sessionId)
|
||||
setSessionStatus(instanceId, sessionId, nextStatus)
|
||||
}
|
||||
|
||||
} catch (error) {
|
||||
log.error("Failed to load messages:", error)
|
||||
throw error
|
||||
|
||||
@@ -12,6 +12,7 @@ import type {
|
||||
EventSessionError,
|
||||
EventSessionIdle,
|
||||
EventSessionUpdated,
|
||||
EventSessionStatus,
|
||||
} from "@opencode-ai/sdk"
|
||||
import type { MessageStatus } from "./message-v2/types"
|
||||
|
||||
@@ -19,11 +20,11 @@ import { getLogger } from "../lib/logger"
|
||||
import { showToastNotification, ToastVariant } from "../lib/notifications"
|
||||
import { instances, addPermissionToQueue, removePermissionFromQueue } from "./instances"
|
||||
import { showAlertDialog } from "./alerts"
|
||||
import { createClientSession, Session, SessionStatus } from "../types/session"
|
||||
import { sessions, setSessions, withSession } from "./session-state"
|
||||
import { normalizeMessagePart } from "./message-v2/normalizers"
|
||||
import { updateSessionInfo } from "./message-v2/session-info"
|
||||
|
||||
const log = getLogger("sse")
|
||||
import { loadMessages } from "./session-api"
|
||||
import { setSessionCompactionState } from "./session-compaction"
|
||||
import {
|
||||
@@ -39,6 +40,9 @@ import {
|
||||
import { messageStoreBus } from "./message-v2/bus"
|
||||
import type { InstanceMessageStore } from "./message-v2/instance-store"
|
||||
|
||||
const log = getLogger("sse")
|
||||
const pendingSessionFetches = new Map<string, Promise<void>>()
|
||||
|
||||
interface TuiToastEvent {
|
||||
type: "tui.toast.show"
|
||||
properties: {
|
||||
@@ -51,8 +55,83 @@ interface TuiToastEvent {
|
||||
|
||||
const ALLOWED_TOAST_VARIANTS = new Set<ToastVariant>(["info", "success", "warning", "error"])
|
||||
|
||||
const mapSdkSessionStatus = (status: EventSessionStatus["properties"]["status"]): SessionStatus => {
|
||||
if (!status || status.type === "idle") {
|
||||
return "idle"
|
||||
}
|
||||
if (status.type === "retry") {
|
||||
return "working"
|
||||
}
|
||||
return "working"
|
||||
}
|
||||
|
||||
function applySessionStatus(instanceId: string, sessionId: string, status: SessionStatus, bumpUpdated = false) {
|
||||
withSession(instanceId, sessionId, (session) => {
|
||||
session.status = status
|
||||
if (bumpUpdated) {
|
||||
session.time = { ...(session.time ?? {}), updated: Date.now() }
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
async function fetchSessionInfo(instanceId: string, sessionId: string): Promise<Session | null> {
|
||||
const instance = instances().get(instanceId)
|
||||
if (!instance?.client) return null
|
||||
|
||||
try {
|
||||
const response = await instance.client.session.get({ path: { id: sessionId } })
|
||||
if (!response.data) return null
|
||||
|
||||
const fetched = createClientSession(response.data, instanceId)
|
||||
|
||||
setSessions((prev) => {
|
||||
const next = new Map(prev)
|
||||
const instanceSessions = new Map(next.get(instanceId) ?? [])
|
||||
const existing = instanceSessions.get(sessionId)
|
||||
instanceSessions.set(sessionId, {
|
||||
...fetched,
|
||||
agent: existing?.agent ?? fetched.agent,
|
||||
model: existing?.model ?? fetched.model,
|
||||
status: existing?.status ?? fetched.status,
|
||||
pendingPermission: existing?.pendingPermission ?? fetched.pendingPermission,
|
||||
})
|
||||
next.set(instanceId, instanceSessions)
|
||||
return next
|
||||
})
|
||||
|
||||
return fetched
|
||||
} catch (error) {
|
||||
log.error("Failed to fetch session info", error)
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
function ensureSessionStatus(instanceId: string, sessionId: string, status: SessionStatus, bumpUpdated = false) {
|
||||
const instanceSessions = sessions().get(instanceId)
|
||||
const existing = instanceSessions?.get(sessionId)
|
||||
if (existing) {
|
||||
applySessionStatus(instanceId, sessionId, status, bumpUpdated)
|
||||
return
|
||||
}
|
||||
|
||||
const key = `${instanceId}:${sessionId}`
|
||||
if (pendingSessionFetches.has(key)) {
|
||||
return
|
||||
}
|
||||
|
||||
const pending = (async () => {
|
||||
const fetched = await fetchSessionInfo(instanceId, sessionId)
|
||||
if (!fetched) return
|
||||
applySessionStatus(instanceId, sessionId, status, bumpUpdated)
|
||||
})()
|
||||
|
||||
pendingSessionFetches.set(key, pending)
|
||||
void pending.finally(() => pendingSessionFetches.delete(key))
|
||||
}
|
||||
|
||||
type MessageRole = "user" | "assistant"
|
||||
|
||||
|
||||
function resolveMessageRole(info?: MessageInfo | null): MessageRole {
|
||||
return info?.role === "user" ? "user" : "assistant"
|
||||
}
|
||||
@@ -74,7 +153,6 @@ function findPendingMessageId(
|
||||
|
||||
function handleMessageUpdate(instanceId: string, event: MessageUpdateEvent | MessagePartUpdatedEvent): void {
|
||||
const instanceSessions = sessions().get(instanceId)
|
||||
if (!instanceSessions) return
|
||||
|
||||
if (event.type === "message.part.updated") {
|
||||
const rawPart = event.properties?.part
|
||||
@@ -90,9 +168,14 @@ function handleMessageUpdate(instanceId: string, event: MessageUpdateEvent | Mes
|
||||
const messageId = typeof part.messageID === "string" ? part.messageID : fallbackMessageId
|
||||
if (!sessionId || !messageId) return
|
||||
|
||||
const session = instanceSessions.get(sessionId)
|
||||
if (!session) return
|
||||
|
||||
const session = instanceSessions?.get(sessionId)
|
||||
if (!session) {
|
||||
ensureSessionStatus(instanceId, sessionId, "working", true)
|
||||
return
|
||||
}
|
||||
|
||||
applySessionStatus(instanceId, sessionId, "working", true)
|
||||
|
||||
const store = messageStoreBus.getOrCreate(instanceId)
|
||||
const role: MessageRole = resolveMessageRole(messageInfo)
|
||||
const createdAt = typeof messageInfo?.time?.created === "number" ? messageInfo.time.created : Date.now()
|
||||
@@ -135,10 +218,16 @@ function handleMessageUpdate(instanceId: string, event: MessageUpdateEvent | Mes
|
||||
const messageId = typeof info.id === "string" ? info.id : undefined
|
||||
if (!sessionId || !messageId) return
|
||||
|
||||
const session = instanceSessions.get(sessionId)
|
||||
if (!session) return
|
||||
const session = instanceSessions?.get(sessionId)
|
||||
if (!session) {
|
||||
ensureSessionStatus(instanceId, sessionId, "working", true)
|
||||
return
|
||||
}
|
||||
|
||||
applySessionStatus(instanceId, sessionId, "working", true)
|
||||
|
||||
const store = messageStoreBus.getOrCreate(instanceId)
|
||||
|
||||
const role: MessageRole = info.role === "user" ? "user" : "assistant"
|
||||
const hasError = Boolean((info as any).error)
|
||||
const status: MessageStatus = hasError ? "error" : "complete"
|
||||
@@ -180,8 +269,7 @@ function handleSessionUpdate(instanceId: string, event: EventSessionUpdated): vo
|
||||
const isCompacting = typeof compactingFlag === "number" ? compactingFlag > 0 : Boolean(compactingFlag)
|
||||
setSessionCompactionState(instanceId, info.id, isCompacting)
|
||||
|
||||
const instanceSessions = sessions().get(instanceId)
|
||||
if (!instanceSessions) return
|
||||
const instanceSessions = sessions().get(instanceId) ?? new Map<string, Session>()
|
||||
|
||||
const existingSession = instanceSessions.get(info.id)
|
||||
|
||||
@@ -196,6 +284,7 @@ function handleSessionUpdate(instanceId: string, event: EventSessionUpdated): vo
|
||||
providerId: "",
|
||||
modelId: "",
|
||||
},
|
||||
status: isCompacting ? "compacting" : "idle",
|
||||
version: info.version || "0",
|
||||
time: info.time
|
||||
? { ...info.time }
|
||||
@@ -203,7 +292,7 @@ function handleSessionUpdate(instanceId: string, event: EventSessionUpdated): vo
|
||||
created: Date.now(),
|
||||
updated: Date.now(),
|
||||
},
|
||||
} as any
|
||||
} as Session
|
||||
|
||||
setSessions((prev) => {
|
||||
const next = new Map(prev)
|
||||
@@ -227,6 +316,7 @@ function handleSessionUpdate(instanceId: string, event: EventSessionUpdated): vo
|
||||
const updatedSession = {
|
||||
...existingSession,
|
||||
title: info.title || existingSession.title,
|
||||
status: isCompacting ? "compacting" : (existingSession.status ?? "idle"),
|
||||
time: mergedTime,
|
||||
revert: info.revert
|
||||
? {
|
||||
@@ -249,13 +339,23 @@ function handleSessionUpdate(instanceId: string, event: EventSessionUpdated): vo
|
||||
}
|
||||
}
|
||||
|
||||
function handleSessionIdle(_instanceId: string, event: EventSessionIdle): void {
|
||||
function handleSessionIdle(instanceId: string, event: EventSessionIdle): void {
|
||||
const sessionId = event.properties?.sessionID
|
||||
if (!sessionId) return
|
||||
|
||||
ensureSessionStatus(instanceId, sessionId, "idle")
|
||||
log.info(`[SSE] Session idle: ${sessionId}`)
|
||||
}
|
||||
|
||||
function handleSessionStatus(instanceId: string, event: EventSessionStatus): void {
|
||||
const sessionId = event.properties?.sessionID
|
||||
if (!sessionId) return
|
||||
|
||||
const status = mapSdkSessionStatus(event.properties.status)
|
||||
ensureSessionStatus(instanceId, sessionId, status, status === "working")
|
||||
log.info(`[SSE] Session status updated: ${sessionId}`, { status })
|
||||
}
|
||||
|
||||
function handleSessionCompacted(instanceId: string, event: EventSessionCompacted): void {
|
||||
const sessionID = event.properties?.sessionID
|
||||
if (!sessionID) return
|
||||
@@ -263,6 +363,7 @@ function handleSessionCompacted(instanceId: string, event: EventSessionCompacted
|
||||
log.info(`[SSE] Session compacted: ${sessionID}`)
|
||||
|
||||
setSessionCompactionState(instanceId, sessionID, false)
|
||||
ensureSessionStatus(instanceId, sessionID, "idle")
|
||||
|
||||
withSession(instanceId, sessionID, (session) => {
|
||||
const time = { ...(session.time ?? {}) }
|
||||
@@ -368,6 +469,7 @@ export {
|
||||
handleSessionCompacted,
|
||||
handleSessionError,
|
||||
handleSessionIdle,
|
||||
handleSessionStatus,
|
||||
handleSessionUpdate,
|
||||
handleTuiToast,
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { createSignal } from "solid-js"
|
||||
|
||||
import type { Session, Agent, Provider } from "../types/session"
|
||||
import type { Session, SessionStatus, Agent, Provider } from "../types/session"
|
||||
import { deleteSession, loadMessages } from "./session-api"
|
||||
import { showToastNotification } from "../lib/notifications"
|
||||
import { messageStoreBus } from "./message-v2/bus"
|
||||
@@ -157,6 +157,11 @@ function setSessionCompactionState(instanceId: string, sessionId: string, isComp
|
||||
const time = { ...(session.time ?? {}) }
|
||||
time.compacting = isCompacting ? Date.now() : 0
|
||||
session.time = time
|
||||
if (isCompacting) {
|
||||
session.status = "compacting"
|
||||
} else if (session.status === "compacting") {
|
||||
session.status = "idle"
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -199,6 +204,12 @@ function clearActiveParentSession(instanceId: string): void {
|
||||
})
|
||||
}
|
||||
|
||||
function setSessionStatus(instanceId: string, sessionId: string, status: SessionStatus): void {
|
||||
withSession(instanceId, sessionId, (session) => {
|
||||
session.status = status
|
||||
})
|
||||
}
|
||||
|
||||
function getActiveParentSession(instanceId: string): Session | null {
|
||||
const parentId = activeParentSessionId().get(instanceId)
|
||||
if (!parentId) return null
|
||||
@@ -380,6 +391,7 @@ export {
|
||||
withSession,
|
||||
setSessionCompactionState,
|
||||
setSessionPendingPermission,
|
||||
setSessionStatus,
|
||||
setActiveSession,
|
||||
|
||||
setActiveParentSession,
|
||||
|
||||
@@ -112,7 +112,7 @@ function isAssistantStillGeneratingRecord(record: MessageRecord, info?: MessageI
|
||||
}
|
||||
|
||||
|
||||
export function getSessionStatus(instanceId: string, sessionId: string): SessionStatus {
|
||||
export function deriveSessionStatusFromMessages(instanceId: string, sessionId: string): SessionStatus {
|
||||
const session = getSession(instanceId, sessionId)
|
||||
if (!session) {
|
||||
return "idle"
|
||||
@@ -162,6 +162,14 @@ export function getSessionStatus(instanceId: string, sessionId: string): Session
|
||||
return "idle"
|
||||
}
|
||||
|
||||
export function getSessionStatus(instanceId: string, sessionId: string): SessionStatus {
|
||||
const session = getSession(instanceId, sessionId)
|
||||
if (!session) {
|
||||
return "idle"
|
||||
}
|
||||
return session.status ?? deriveSessionStatusFromMessages(instanceId, sessionId)
|
||||
}
|
||||
|
||||
export function isSessionBusy(instanceId: string, sessionId: string): boolean {
|
||||
const status = getSessionStatus(instanceId, sessionId)
|
||||
return status === "working" || status === "compacting"
|
||||
|
||||
@@ -26,6 +26,7 @@ import {
|
||||
setActiveParentSession,
|
||||
setActiveSession,
|
||||
setSessionDraftPrompt,
|
||||
setSessionStatus,
|
||||
} from "./session-state"
|
||||
|
||||
import { getDefaultModel } from "./session-models"
|
||||
@@ -56,6 +57,7 @@ import {
|
||||
handleSessionCompacted,
|
||||
handleSessionError,
|
||||
handleSessionIdle,
|
||||
handleSessionStatus,
|
||||
handleSessionUpdate,
|
||||
handleTuiToast,
|
||||
} from "./session-events"
|
||||
@@ -68,6 +70,7 @@ sseManager.onSessionUpdate = handleSessionUpdate
|
||||
sseManager.onSessionCompacted = handleSessionCompacted
|
||||
sseManager.onSessionError = handleSessionError
|
||||
sseManager.onSessionIdle = handleSessionIdle
|
||||
sseManager.onSessionStatus = handleSessionStatus
|
||||
sseManager.onTuiToast = handleTuiToast
|
||||
sseManager.onPermissionUpdated = handlePermissionUpdated
|
||||
sseManager.onPermissionReplied = handlePermissionReplied
|
||||
@@ -109,6 +112,7 @@ export {
|
||||
setActiveParentSession,
|
||||
setActiveSession,
|
||||
setSessionDraftPrompt,
|
||||
setSessionStatus,
|
||||
updateSessionAgent,
|
||||
updateSessionModel,
|
||||
}
|
||||
|
||||
@@ -27,6 +27,7 @@ export interface Session
|
||||
}
|
||||
version: string // Include version from SDK Session
|
||||
pendingPermission?: boolean // Indicates if session is waiting on user permission
|
||||
status: SessionStatus // Single source of truth for session status
|
||||
}
|
||||
|
||||
// Adapter function to convert SDK Session to client Session
|
||||
@@ -42,6 +43,7 @@ export function createClientSession(
|
||||
parentId: sdkSession.parentID || null,
|
||||
agent,
|
||||
model,
|
||||
status: "idle",
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user