fix(session-manager): fall back to file storage on SDK outages
This commit is contained in:
@@ -0,0 +1,203 @@
|
||||
import { existsSync } from "node:fs"
|
||||
import { readdir, readFile } from "node:fs/promises"
|
||||
import { join } from "node:path"
|
||||
import { MESSAGE_STORAGE, PART_STORAGE, SESSION_STORAGE, TODO_DIR, TRANSCRIPT_DIR } from "./constants"
|
||||
import { getMessageDir } from "../../shared/opencode-message-dir"
|
||||
import type { SessionInfo, SessionMessage, SessionMetadata, TodoItem } from "./types"
|
||||
|
||||
export async function getFileMainSessions(directory?: string): Promise<SessionMetadata[]> {
|
||||
if (!existsSync(SESSION_STORAGE)) return []
|
||||
|
||||
const sessions: SessionMetadata[] = []
|
||||
|
||||
try {
|
||||
const projectDirs = await readdir(SESSION_STORAGE, { withFileTypes: true })
|
||||
for (const projectDir of projectDirs) {
|
||||
if (!projectDir.isDirectory()) continue
|
||||
const projectPath = join(SESSION_STORAGE, projectDir.name)
|
||||
const sessionFiles = await readdir(projectPath)
|
||||
|
||||
for (const file of sessionFiles) {
|
||||
if (!file.endsWith(".json")) continue
|
||||
|
||||
try {
|
||||
const content = await readFile(join(projectPath, file), "utf-8")
|
||||
const meta = JSON.parse(content) as SessionMetadata
|
||||
if (meta.parentID) continue
|
||||
if (directory && meta.directory !== directory) continue
|
||||
sessions.push(meta)
|
||||
} catch {
|
||||
continue
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
return []
|
||||
}
|
||||
|
||||
return sessions.sort((a, b) => b.time.updated - a.time.updated)
|
||||
}
|
||||
|
||||
export async function getFileAllSessions(): Promise<string[]> {
|
||||
if (!existsSync(MESSAGE_STORAGE)) return []
|
||||
|
||||
const sessions: string[] = []
|
||||
|
||||
async function scanDirectory(dir: string): Promise<void> {
|
||||
try {
|
||||
const entries = await readdir(dir, { withFileTypes: true })
|
||||
for (const entry of entries) {
|
||||
if (!entry.isDirectory()) continue
|
||||
const sessionPath = join(dir, entry.name)
|
||||
const files = await readdir(sessionPath)
|
||||
if (files.some((file) => file.endsWith(".json"))) {
|
||||
sessions.push(entry.name)
|
||||
continue
|
||||
}
|
||||
await scanDirectory(sessionPath)
|
||||
}
|
||||
} catch {
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
await scanDirectory(MESSAGE_STORAGE)
|
||||
return [...new Set(sessions)]
|
||||
}
|
||||
|
||||
export async function fileSessionExists(sessionID: string): Promise<boolean> {
|
||||
return getMessageDir(sessionID) !== null
|
||||
}
|
||||
|
||||
export async function getFileSessionMessages(sessionID: string): Promise<SessionMessage[]> {
|
||||
const messageDir = getMessageDir(sessionID)
|
||||
if (!messageDir || !existsSync(messageDir)) return []
|
||||
|
||||
const messages: SessionMessage[] = []
|
||||
try {
|
||||
const files = await readdir(messageDir)
|
||||
for (const file of files) {
|
||||
if (!file.endsWith(".json")) continue
|
||||
try {
|
||||
const content = await readFile(join(messageDir, file), "utf-8")
|
||||
const meta = JSON.parse(content)
|
||||
const parts = await readParts(meta.id)
|
||||
messages.push({
|
||||
id: meta.id,
|
||||
role: meta.role,
|
||||
agent: meta.agent,
|
||||
time: meta.time,
|
||||
parts,
|
||||
})
|
||||
} catch {
|
||||
continue
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
return []
|
||||
}
|
||||
|
||||
return messages.sort((a, b) => {
|
||||
const aTime = a.time?.created ?? 0
|
||||
const bTime = b.time?.created ?? 0
|
||||
if (aTime !== bTime) return aTime - bTime
|
||||
return a.id.localeCompare(b.id)
|
||||
})
|
||||
}
|
||||
|
||||
async function readParts(messageID: string): Promise<Array<{ id: string; type: string; [key: string]: unknown }>> {
|
||||
const partDir = join(PART_STORAGE, messageID)
|
||||
if (!existsSync(partDir)) return []
|
||||
|
||||
const parts: Array<{ id: string; type: string; [key: string]: unknown }> = []
|
||||
try {
|
||||
const files = await readdir(partDir)
|
||||
for (const file of files) {
|
||||
if (!file.endsWith(".json")) continue
|
||||
try {
|
||||
const content = await readFile(join(partDir, file), "utf-8")
|
||||
parts.push(JSON.parse(content))
|
||||
} catch {
|
||||
continue
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
return []
|
||||
}
|
||||
|
||||
return parts.sort((a, b) => a.id.localeCompare(b.id))
|
||||
}
|
||||
|
||||
export async function getFileSessionTodos(sessionID: string): Promise<TodoItem[]> {
|
||||
if (!existsSync(TODO_DIR)) return []
|
||||
|
||||
try {
|
||||
const allFiles = await readdir(TODO_DIR)
|
||||
const todoFiles = allFiles.filter((file) => file === `${sessionID}.json`)
|
||||
|
||||
for (const file of todoFiles) {
|
||||
try {
|
||||
const content = await readFile(join(TODO_DIR, file), "utf-8")
|
||||
const data = JSON.parse(content)
|
||||
if (!Array.isArray(data)) continue
|
||||
return data.map((item) => ({
|
||||
id: item.id || "",
|
||||
content: item.content || "",
|
||||
status: item.status || "pending",
|
||||
priority: item.priority,
|
||||
}))
|
||||
} catch {
|
||||
continue
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
return []
|
||||
}
|
||||
|
||||
return []
|
||||
}
|
||||
|
||||
export async function getFileSessionTranscript(sessionID: string): Promise<number> {
|
||||
if (!existsSync(TRANSCRIPT_DIR)) return 0
|
||||
const transcriptFile = join(TRANSCRIPT_DIR, `${sessionID}.jsonl`)
|
||||
if (!existsSync(transcriptFile)) return 0
|
||||
|
||||
try {
|
||||
const content = await readFile(transcriptFile, "utf-8")
|
||||
return content.trim().split("\n").filter(Boolean).length
|
||||
} catch {
|
||||
return 0
|
||||
}
|
||||
}
|
||||
|
||||
export async function getFileSessionInfo(sessionID: string): Promise<SessionInfo | null> {
|
||||
const messages = await getFileSessionMessages(sessionID)
|
||||
if (messages.length === 0) return null
|
||||
|
||||
const agentsUsed = new Set<string>()
|
||||
let firstMessage: Date | undefined
|
||||
let lastMessage: Date | undefined
|
||||
|
||||
for (const msg of messages) {
|
||||
if (msg.agent) agentsUsed.add(msg.agent)
|
||||
if (!msg.time?.created) continue
|
||||
const date = new Date(msg.time.created)
|
||||
if (!firstMessage || date < firstMessage) firstMessage = date
|
||||
if (!lastMessage || date > lastMessage) lastMessage = date
|
||||
}
|
||||
|
||||
const todos = await getFileSessionTodos(sessionID)
|
||||
const transcriptEntries = await getFileSessionTranscript(sessionID)
|
||||
|
||||
return {
|
||||
id: sessionID,
|
||||
message_count: messages.length,
|
||||
first_message: firstMessage,
|
||||
last_message: lastMessage,
|
||||
agents_used: Array.from(agentsUsed),
|
||||
has_todos: todos.length > 0,
|
||||
has_transcript: transcriptEntries > 0,
|
||||
todos,
|
||||
transcript_entries: transcriptEntries,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,135 @@
|
||||
import type { PluginInput } from "@opencode-ai/plugin"
|
||||
import { normalizeSDKResponse } from "../../shared"
|
||||
import type { SessionMessage, SessionMetadata, TodoItem } from "./types"
|
||||
import { isSessionSdkUnavailableError } from "./sdk-unavailable"
|
||||
|
||||
function unwrapSdkResponseError(response: unknown): unknown {
|
||||
if (!response || typeof response !== "object" || !("error" in response)) {
|
||||
return null
|
||||
}
|
||||
|
||||
return (response as { error?: unknown }).error ?? null
|
||||
}
|
||||
|
||||
function throwOnNonFallbackableSdkError(response: unknown): void {
|
||||
const error = unwrapSdkResponseError(response)
|
||||
if (!error) return
|
||||
throw error
|
||||
}
|
||||
|
||||
export async function getSdkMainSessions(
|
||||
client: PluginInput["client"],
|
||||
directory?: string,
|
||||
): Promise<SessionMetadata[]> {
|
||||
const response = await client.session.list()
|
||||
const error = unwrapSdkResponseError(response)
|
||||
if (error) throw error
|
||||
|
||||
const sessions = normalizeSDKResponse(response, [] as SessionMetadata[])
|
||||
const mainSessions = sessions.filter((session) => !session.parentID)
|
||||
if (directory) {
|
||||
return mainSessions
|
||||
.filter((session) => session.directory === directory)
|
||||
.sort((a, b) => b.time.updated - a.time.updated)
|
||||
}
|
||||
|
||||
return mainSessions.sort((a, b) => b.time.updated - a.time.updated)
|
||||
}
|
||||
|
||||
export async function getSdkAllSessions(client: PluginInput["client"]): Promise<string[]> {
|
||||
const response = await client.session.list()
|
||||
throwOnNonFallbackableSdkError(response)
|
||||
const sessions = normalizeSDKResponse(response, [] as SessionMetadata[])
|
||||
return sessions.map((session) => session.id)
|
||||
}
|
||||
|
||||
export async function sdkSessionExists(client: PluginInput["client"], sessionID: string): Promise<boolean> {
|
||||
const response = await client.session.list()
|
||||
throwOnNonFallbackableSdkError(response)
|
||||
const sessions = normalizeSDKResponse(response, [] as Array<{ id?: string }>)
|
||||
return sessions.some((session) => session.id === sessionID)
|
||||
}
|
||||
|
||||
export async function getSdkSessionMessages(
|
||||
client: PluginInput["client"],
|
||||
sessionID: string,
|
||||
): Promise<SessionMessage[]> {
|
||||
const response = await client.session.messages({ path: { id: sessionID } })
|
||||
throwOnNonFallbackableSdkError(response)
|
||||
|
||||
const rawMessages = normalizeSDKResponse(response, [] as Array<{
|
||||
info?: {
|
||||
id?: string
|
||||
role?: string
|
||||
agent?: string
|
||||
time?: { created?: number; updated?: number }
|
||||
}
|
||||
parts?: Array<{
|
||||
id?: string
|
||||
type?: string
|
||||
text?: string
|
||||
thinking?: string
|
||||
tool?: string
|
||||
callID?: string
|
||||
input?: Record<string, unknown>
|
||||
output?: string
|
||||
error?: string
|
||||
}>
|
||||
}>)
|
||||
|
||||
const messages: SessionMessage[] = rawMessages
|
||||
.filter((message) => message.info?.id)
|
||||
.map((message) => ({
|
||||
id: message.info!.id!,
|
||||
role: (message.info!.role as "user" | "assistant") || "user",
|
||||
agent: message.info!.agent,
|
||||
time: message.info!.time?.created
|
||||
? {
|
||||
created: message.info!.time.created,
|
||||
updated: message.info!.time.updated,
|
||||
}
|
||||
: undefined,
|
||||
parts:
|
||||
message.parts?.map((part) => ({
|
||||
id: part.id || "",
|
||||
type: part.type || "text",
|
||||
text: part.text,
|
||||
thinking: part.thinking,
|
||||
tool: part.tool,
|
||||
callID: part.callID,
|
||||
input: part.input,
|
||||
output: part.output,
|
||||
error: part.error,
|
||||
})) || [],
|
||||
}))
|
||||
|
||||
return messages.sort((a, b) => {
|
||||
const aTime = a.time?.created ?? 0
|
||||
const bTime = b.time?.created ?? 0
|
||||
if (aTime !== bTime) return aTime - bTime
|
||||
return a.id.localeCompare(b.id)
|
||||
})
|
||||
}
|
||||
|
||||
export async function getSdkSessionTodos(client: PluginInput["client"], sessionID: string): Promise<TodoItem[]> {
|
||||
const response = await client.session.todo({ path: { id: sessionID } })
|
||||
throwOnNonFallbackableSdkError(response)
|
||||
|
||||
const data = normalizeSDKResponse(response, [] as Array<{
|
||||
id?: string
|
||||
content?: string
|
||||
status?: string
|
||||
priority?: string
|
||||
}>)
|
||||
|
||||
return data.map((item) => ({
|
||||
id: item.id || "",
|
||||
content: item.content || "",
|
||||
status: (item.status as TodoItem["status"]) || "pending",
|
||||
priority: item.priority,
|
||||
}))
|
||||
}
|
||||
|
||||
export function shouldFallbackFromSdkError(error: unknown): boolean {
|
||||
return isSessionSdkUnavailableError(error)
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
const SDK_UNAVAILABLE_PATTERNS = [
|
||||
"unable to connect",
|
||||
"econnrefused",
|
||||
"fetch failed",
|
||||
"network error",
|
||||
"network request failed",
|
||||
"server unreachable",
|
||||
"etimedout",
|
||||
"timed out",
|
||||
"timeout",
|
||||
"socket hang up",
|
||||
] as const
|
||||
|
||||
function collectErrorTexts(value: unknown): string[] {
|
||||
if (value instanceof Error) {
|
||||
return [value.message, value.name, ...collectErrorTexts(value.cause)]
|
||||
}
|
||||
|
||||
if (typeof value === "string") {
|
||||
return [value]
|
||||
}
|
||||
|
||||
if (!value || typeof value !== "object") {
|
||||
return []
|
||||
}
|
||||
|
||||
const record = value as Record<string, unknown>
|
||||
return [
|
||||
typeof record.message === "string" ? record.message : "",
|
||||
typeof record.code === "string" ? record.code : "",
|
||||
typeof record.name === "string" ? record.name : "",
|
||||
...collectErrorTexts(record.cause),
|
||||
...collectErrorTexts(record.error),
|
||||
].filter(Boolean)
|
||||
}
|
||||
|
||||
export function isSessionSdkUnavailableError(value: unknown): boolean {
|
||||
const haystack = collectErrorTexts(value)
|
||||
.join(" ")
|
||||
.toLowerCase()
|
||||
|
||||
return SDK_UNAVAILABLE_PATTERNS.some((pattern) => haystack.includes(pattern))
|
||||
}
|
||||
@@ -0,0 +1,165 @@
|
||||
import { afterAll, afterEach, beforeEach, describe, expect, mock, test } from "bun:test"
|
||||
import { existsSync, mkdirSync, readdirSync, rmSync, writeFileSync } from "node:fs"
|
||||
import { tmpdir } from "node:os"
|
||||
import { join } from "node:path"
|
||||
import { randomUUID } from "node:crypto"
|
||||
|
||||
const TEST_DIR = join(tmpdir(), `omo-test-session-manager-fallback-${randomUUID()}`)
|
||||
const TEST_MESSAGE_STORAGE = join(TEST_DIR, "message")
|
||||
const TEST_PART_STORAGE = join(TEST_DIR, "part")
|
||||
const TEST_SESSION_STORAGE = join(TEST_DIR, "session")
|
||||
const TEST_TODO_DIR = join(TEST_DIR, "todos")
|
||||
const TEST_TRANSCRIPT_DIR = join(TEST_DIR, "transcripts")
|
||||
|
||||
let sqliteBackend = false
|
||||
|
||||
mock.module("./constants", () => ({
|
||||
OPENCODE_STORAGE: TEST_DIR,
|
||||
MESSAGE_STORAGE: TEST_MESSAGE_STORAGE,
|
||||
PART_STORAGE: TEST_PART_STORAGE,
|
||||
SESSION_STORAGE: TEST_SESSION_STORAGE,
|
||||
TODO_DIR: TEST_TODO_DIR,
|
||||
TRANSCRIPT_DIR: TEST_TRANSCRIPT_DIR,
|
||||
SESSION_LIST_DESCRIPTION: "test",
|
||||
SESSION_READ_DESCRIPTION: "test",
|
||||
SESSION_SEARCH_DESCRIPTION: "test",
|
||||
SESSION_INFO_DESCRIPTION: "test",
|
||||
SESSION_DELETE_DESCRIPTION: "test",
|
||||
TOOL_NAME_PREFIX: "session_",
|
||||
}))
|
||||
|
||||
mock.module("../../shared/opencode-storage-detection", () => ({
|
||||
isSqliteBackend: () => sqliteBackend,
|
||||
resetSqliteBackendCache: () => {},
|
||||
}))
|
||||
|
||||
mock.module("../../shared/opencode-message-dir", () => ({
|
||||
getMessageDir: (sessionID: string) => {
|
||||
if (!sessionID.startsWith("ses_")) return null
|
||||
if (/[/\\]|\.\./.test(sessionID)) return null
|
||||
if (!existsSync(TEST_MESSAGE_STORAGE)) return null
|
||||
|
||||
const directPath = join(TEST_MESSAGE_STORAGE, sessionID)
|
||||
if (existsSync(directPath)) return directPath
|
||||
|
||||
for (const dir of readdirSync(TEST_MESSAGE_STORAGE)) {
|
||||
const nestedPath = join(TEST_MESSAGE_STORAGE, dir, sessionID)
|
||||
if (existsSync(nestedPath)) return nestedPath
|
||||
}
|
||||
|
||||
return null
|
||||
},
|
||||
}))
|
||||
|
||||
afterAll(() => {
|
||||
mock.restore()
|
||||
})
|
||||
|
||||
const storage = await import("./storage")
|
||||
|
||||
function createSdkUnavailableError(message: string): Error {
|
||||
return new Error(message)
|
||||
}
|
||||
|
||||
function createSessionMetadata(projectID: string, sessionID: string, directory: string, updated: number): void {
|
||||
const projectDir = join(TEST_SESSION_STORAGE, projectID)
|
||||
mkdirSync(projectDir, { recursive: true })
|
||||
writeFileSync(
|
||||
join(projectDir, `${sessionID}.json`),
|
||||
JSON.stringify({
|
||||
id: sessionID,
|
||||
projectID,
|
||||
directory,
|
||||
time: { created: updated - 1_000, updated },
|
||||
}),
|
||||
)
|
||||
}
|
||||
|
||||
function createSessionMessage(sessionID: string, messageID: string, created: number, role = "user"): void {
|
||||
const sessionPath = join(TEST_MESSAGE_STORAGE, sessionID)
|
||||
mkdirSync(sessionPath, { recursive: true })
|
||||
writeFileSync(
|
||||
join(sessionPath, `${messageID}.json`),
|
||||
JSON.stringify({ id: messageID, role, time: { created } }),
|
||||
)
|
||||
}
|
||||
|
||||
function createSessionTodo(sessionID: string, items: Array<Record<string, unknown>>): void {
|
||||
mkdirSync(TEST_TODO_DIR, { recursive: true })
|
||||
writeFileSync(join(TEST_TODO_DIR, `${sessionID}.json`), JSON.stringify(items))
|
||||
}
|
||||
|
||||
describe("session-manager storage fallback", () => {
|
||||
const mockClient = {
|
||||
session: {
|
||||
list: mock((): Promise<unknown> => Promise.resolve({ data: [] })),
|
||||
messages: mock((): Promise<unknown> => Promise.resolve({ data: [] })),
|
||||
todo: mock((): Promise<unknown> => Promise.resolve({ data: [] })),
|
||||
},
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
sqliteBackend = true
|
||||
if (existsSync(TEST_DIR)) rmSync(TEST_DIR, { recursive: true, force: true })
|
||||
mkdirSync(TEST_DIR, { recursive: true })
|
||||
mkdirSync(TEST_MESSAGE_STORAGE, { recursive: true })
|
||||
mkdirSync(TEST_PART_STORAGE, { recursive: true })
|
||||
mkdirSync(TEST_SESSION_STORAGE, { recursive: true })
|
||||
mkdirSync(TEST_TODO_DIR, { recursive: true })
|
||||
mkdirSync(TEST_TRANSCRIPT_DIR, { recursive: true })
|
||||
mockClient.session.list.mockReset()
|
||||
mockClient.session.messages.mockReset()
|
||||
mockClient.session.todo.mockReset()
|
||||
storage.setStorageClient(mockClient as never)
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
storage.resetStorageClient()
|
||||
if (existsSync(TEST_DIR)) rmSync(TEST_DIR, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
test("#given unreachable SDK list response #when getMainSessions runs #then falls back to file sessions", async () => {
|
||||
createSessionMetadata("proj_test", "ses_file", "/workspace/project", 2_000)
|
||||
mockClient.session.list.mockImplementation(() => Promise.resolve({ error: createSdkUnavailableError("fetch failed ECONNREFUSED") }))
|
||||
|
||||
const sessions = await storage.getMainSessions({ directory: "/workspace/project" })
|
||||
|
||||
expect(sessions).toHaveLength(1)
|
||||
expect(sessions[0].id).toBe("ses_file")
|
||||
})
|
||||
|
||||
test("#given unreachable SDK messages error #when readSessionMessages runs #then falls back to file messages", async () => {
|
||||
createSessionMessage("ses_file", "msg_001", 1_000)
|
||||
mockClient.session.messages.mockImplementation(() => Promise.reject(createSdkUnavailableError("Unable to connect to http://localhost:4096")))
|
||||
|
||||
const messages = await storage.readSessionMessages("ses_file")
|
||||
|
||||
expect(messages).toHaveLength(1)
|
||||
expect(messages[0].id).toBe("msg_001")
|
||||
})
|
||||
|
||||
test("#given unreachable SDK todo response #when readSessionTodos runs #then falls back to file todos", async () => {
|
||||
createSessionTodo("ses_file", [{ id: "todo_1", content: "Fallback todo", status: "pending" }])
|
||||
mockClient.session.todo.mockImplementation(() => Promise.resolve({ error: createSdkUnavailableError("network error: server unreachable") }))
|
||||
|
||||
const todos = await storage.readSessionTodos("ses_file")
|
||||
|
||||
expect(todos).toHaveLength(1)
|
||||
expect(todos[0].content).toBe("Fallback todo")
|
||||
})
|
||||
|
||||
test("#given unreachable SDK list error #when sessionExists runs #then falls back to file existence", async () => {
|
||||
createSessionMessage("ses_file", "msg_001", 1_000)
|
||||
mockClient.session.list.mockImplementation(() => Promise.reject(createSdkUnavailableError("ETIMEDOUT while connecting")))
|
||||
|
||||
const exists = await storage.sessionExists("ses_file")
|
||||
|
||||
expect(exists).toBe(true)
|
||||
})
|
||||
|
||||
test("#given semantic SDK error #when readSessionMessages runs #then rethrows instead of hiding bug", async () => {
|
||||
mockClient.session.messages.mockImplementation(() => Promise.resolve({ error: new Error("session not found") }))
|
||||
|
||||
await expect(storage.readSessionMessages("ses_missing")).rejects.toThrow("session not found")
|
||||
})
|
||||
})
|
||||
@@ -374,9 +374,9 @@ describe("session-manager storage - getMainSessions", () => {
|
||||
describe("session-manager storage - SDK path (beta mode)", () => {
|
||||
const mockClient = {
|
||||
session: {
|
||||
list: mock(() => Promise.resolve({ data: [] })),
|
||||
messages: mock(() => Promise.resolve({ data: [] })),
|
||||
todo: mock(() => Promise.resolve({ data: [] })),
|
||||
list: mock((): Promise<unknown> => Promise.resolve({ data: [] })),
|
||||
messages: mock((): Promise<unknown> => Promise.resolve({ data: [] })),
|
||||
todo: mock((): Promise<unknown> => Promise.resolve({ data: [] })),
|
||||
},
|
||||
}
|
||||
|
||||
@@ -500,7 +500,7 @@ describe("session-manager storage - SDK path (beta mode)", () => {
|
||||
expect(todos[1].status).toBe("completed")
|
||||
})
|
||||
|
||||
test("SDK path returns empty array on error", async () => {
|
||||
test("SDK path rethrows non-transport errors", async () => {
|
||||
// given
|
||||
mockClient.session.messages.mockImplementation(() => Promise.reject(new Error("API error")))
|
||||
|
||||
@@ -512,11 +512,7 @@ describe("session-manager storage - SDK path (beta mode)", () => {
|
||||
const { setStorageClient, readSessionMessages } = await import("./storage")
|
||||
setStorageClient(mockClient as unknown as Parameters<typeof setStorageClient>[0])
|
||||
|
||||
// when
|
||||
const messages = await readSessionMessages("ses_test")
|
||||
|
||||
// then
|
||||
expect(messages).toEqual([])
|
||||
await expect(readSessionMessages("ses_test")).rejects.toThrow("API error")
|
||||
})
|
||||
|
||||
test("SDK path returns empty array when client is not set", async () => {
|
||||
|
||||
@@ -1,12 +1,9 @@
|
||||
import { existsSync } from "node:fs"
|
||||
import { readdir, readFile } from "node:fs/promises"
|
||||
import { join } from "node:path"
|
||||
import type { PluginInput } from "@opencode-ai/plugin"
|
||||
import { MESSAGE_STORAGE, PART_STORAGE, SESSION_STORAGE, TODO_DIR, TRANSCRIPT_DIR } from "./constants"
|
||||
import { isSqliteBackend } from "../../shared/opencode-storage-detection"
|
||||
import { getMessageDir } from "../../shared/opencode-message-dir"
|
||||
import type { SessionMessage, SessionInfo, TodoItem, SessionMetadata } from "./types"
|
||||
import { normalizeSDKResponse } from "../../shared"
|
||||
import { log } from "../../shared"
|
||||
import { getFileAllSessions, getFileMainSessions, fileSessionExists, getFileSessionInfo, getFileSessionMessages, getFileSessionTodos, getFileSessionTranscript } from "./file-storage"
|
||||
import { getSdkAllSessions, getSdkMainSessions, getSdkSessionMessages, getSdkSessionTodos, sdkSessionExists, shouldFallbackFromSdkError } from "./sdk-storage"
|
||||
import type { SessionInfo, SessionMessage, SessionMetadata, TodoItem } from "./types"
|
||||
|
||||
export interface GetMainSessionsOptions {
|
||||
directory?: string
|
||||
@@ -24,327 +21,75 @@ export function resetStorageClient(): void {
|
||||
}
|
||||
|
||||
export async function getMainSessions(options: GetMainSessionsOptions): Promise<SessionMetadata[]> {
|
||||
// Beta mode: use SDK
|
||||
if (isSqliteBackend() && sdkClient) {
|
||||
try {
|
||||
const response = await sdkClient.session.list()
|
||||
const sessions = normalizeSDKResponse(response, [] as SessionMetadata[])
|
||||
const mainSessions = sessions.filter((s) => !s.parentID)
|
||||
if (options.directory) {
|
||||
return mainSessions
|
||||
.filter((s) => s.directory === options.directory)
|
||||
.sort((a, b) => b.time.updated - a.time.updated)
|
||||
}
|
||||
return mainSessions.sort((a, b) => b.time.updated - a.time.updated)
|
||||
} catch {
|
||||
return []
|
||||
return await getSdkMainSessions(sdkClient, options.directory)
|
||||
} catch (error) {
|
||||
if (!shouldFallbackFromSdkError(error)) throw error
|
||||
log("[session-manager] falling back to file session list after SDK unavailable error", { error: String(error) })
|
||||
}
|
||||
}
|
||||
|
||||
// Stable mode: use JSON files
|
||||
if (!existsSync(SESSION_STORAGE)) return []
|
||||
|
||||
const sessions: SessionMetadata[] = []
|
||||
|
||||
try {
|
||||
const projectDirs = await readdir(SESSION_STORAGE, { withFileTypes: true })
|
||||
for (const projectDir of projectDirs) {
|
||||
if (!projectDir.isDirectory()) continue
|
||||
|
||||
const projectPath = join(SESSION_STORAGE, projectDir.name)
|
||||
const sessionFiles = await readdir(projectPath)
|
||||
|
||||
for (const file of sessionFiles) {
|
||||
if (!file.endsWith(".json")) continue
|
||||
|
||||
try {
|
||||
const content = await readFile(join(projectPath, file), "utf-8")
|
||||
const meta = JSON.parse(content) as SessionMetadata
|
||||
|
||||
if (meta.parentID) continue
|
||||
|
||||
if (options.directory && meta.directory !== options.directory) continue
|
||||
|
||||
sessions.push(meta)
|
||||
} catch {
|
||||
continue
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
return []
|
||||
}
|
||||
|
||||
return sessions.sort((a, b) => b.time.updated - a.time.updated)
|
||||
return getFileMainSessions(options.directory)
|
||||
}
|
||||
|
||||
export async function getAllSessions(): Promise<string[]> {
|
||||
// Beta mode: use SDK
|
||||
if (isSqliteBackend() && sdkClient) {
|
||||
try {
|
||||
const response = await sdkClient.session.list()
|
||||
const sessions = normalizeSDKResponse(response, [] as SessionMetadata[])
|
||||
return sessions.map((s) => s.id)
|
||||
} catch {
|
||||
return []
|
||||
return await getSdkAllSessions(sdkClient)
|
||||
} catch (error) {
|
||||
if (!shouldFallbackFromSdkError(error)) throw error
|
||||
log("[session-manager] falling back to file session ids after SDK unavailable error", { error: String(error) })
|
||||
}
|
||||
}
|
||||
|
||||
// Stable mode: use JSON files
|
||||
if (!existsSync(MESSAGE_STORAGE)) return []
|
||||
|
||||
const sessions: string[] = []
|
||||
|
||||
async function scanDirectory(dir: string): Promise<void> {
|
||||
try {
|
||||
const entries = await readdir(dir, { withFileTypes: true })
|
||||
for (const entry of entries) {
|
||||
if (entry.isDirectory()) {
|
||||
const sessionPath = join(dir, entry.name)
|
||||
const files = await readdir(sessionPath)
|
||||
if (files.some((f) => f.endsWith(".json"))) {
|
||||
sessions.push(entry.name)
|
||||
} else {
|
||||
await scanDirectory(sessionPath)
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
await scanDirectory(MESSAGE_STORAGE)
|
||||
return [...new Set(sessions)]
|
||||
return getFileAllSessions()
|
||||
}
|
||||
|
||||
export { getMessageDir } from "../../shared/opencode-message-dir"
|
||||
|
||||
export async function sessionExists(sessionID: string): Promise<boolean> {
|
||||
if (isSqliteBackend() && sdkClient) {
|
||||
const response = await sdkClient.session.list()
|
||||
const sessions = normalizeSDKResponse(response, [] as Array<{ id?: string }>)
|
||||
return sessions.some((s) => s.id === sessionID)
|
||||
try {
|
||||
return await sdkSessionExists(sdkClient, sessionID)
|
||||
} catch (error) {
|
||||
if (!shouldFallbackFromSdkError(error)) throw error
|
||||
log("[session-manager] falling back to file sessionExists after SDK unavailable error", { error: String(error), sessionID })
|
||||
}
|
||||
}
|
||||
return getMessageDir(sessionID) !== null
|
||||
return fileSessionExists(sessionID)
|
||||
}
|
||||
|
||||
export async function readSessionMessages(sessionID: string): Promise<SessionMessage[]> {
|
||||
// Beta mode: use SDK
|
||||
if (isSqliteBackend() && sdkClient) {
|
||||
try {
|
||||
const response = await sdkClient.session.messages({ path: { id: sessionID } })
|
||||
const rawMessages = normalizeSDKResponse(response, [] as Array<{
|
||||
info?: {
|
||||
id?: string
|
||||
role?: string
|
||||
agent?: string
|
||||
time?: { created?: number; updated?: number }
|
||||
}
|
||||
parts?: Array<{
|
||||
id?: string
|
||||
type?: string
|
||||
text?: string
|
||||
thinking?: string
|
||||
tool?: string
|
||||
callID?: string
|
||||
input?: Record<string, unknown>
|
||||
output?: string
|
||||
error?: string
|
||||
}>
|
||||
}>)
|
||||
const messages: SessionMessage[] = rawMessages
|
||||
.filter((m) => m.info?.id)
|
||||
.map((m) => ({
|
||||
id: m.info!.id!,
|
||||
role: (m.info!.role as "user" | "assistant") || "user",
|
||||
agent: m.info!.agent,
|
||||
time: m.info!.time?.created
|
||||
? {
|
||||
created: m.info!.time.created,
|
||||
updated: m.info!.time.updated,
|
||||
}
|
||||
: undefined,
|
||||
parts:
|
||||
m.parts?.map((p) => ({
|
||||
id: p.id || "",
|
||||
type: p.type || "text",
|
||||
text: p.text,
|
||||
thinking: p.thinking,
|
||||
tool: p.tool,
|
||||
callID: p.callID,
|
||||
input: p.input,
|
||||
output: p.output,
|
||||
error: p.error,
|
||||
})) || [],
|
||||
}))
|
||||
return messages.sort((a, b) => {
|
||||
const aTime = a.time?.created ?? 0
|
||||
const bTime = b.time?.created ?? 0
|
||||
if (aTime !== bTime) return aTime - bTime
|
||||
return a.id.localeCompare(b.id)
|
||||
})
|
||||
} catch {
|
||||
return []
|
||||
return await getSdkSessionMessages(sdkClient, sessionID)
|
||||
} catch (error) {
|
||||
if (!shouldFallbackFromSdkError(error)) throw error
|
||||
log("[session-manager] falling back to file session messages after SDK unavailable error", { error: String(error), sessionID })
|
||||
}
|
||||
}
|
||||
|
||||
// Stable mode: use JSON files
|
||||
const messageDir = getMessageDir(sessionID)
|
||||
if (!messageDir || !existsSync(messageDir)) return []
|
||||
|
||||
const messages: SessionMessage[] = []
|
||||
try {
|
||||
const files = await readdir(messageDir)
|
||||
for (const file of files) {
|
||||
if (!file.endsWith(".json")) continue
|
||||
try {
|
||||
const content = await readFile(join(messageDir, file), "utf-8")
|
||||
const meta = JSON.parse(content)
|
||||
|
||||
const parts = await readParts(meta.id)
|
||||
|
||||
messages.push({
|
||||
id: meta.id,
|
||||
role: meta.role,
|
||||
agent: meta.agent,
|
||||
time: meta.time,
|
||||
parts,
|
||||
})
|
||||
} catch {
|
||||
continue
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
return []
|
||||
}
|
||||
|
||||
return messages.sort((a, b) => {
|
||||
const aTime = a.time?.created ?? 0
|
||||
const bTime = b.time?.created ?? 0
|
||||
if (aTime !== bTime) return aTime - bTime
|
||||
return a.id.localeCompare(b.id)
|
||||
})
|
||||
}
|
||||
|
||||
async function readParts(messageID: string): Promise<Array<{ id: string; type: string; [key: string]: unknown }>> {
|
||||
const partDir = join(PART_STORAGE, messageID)
|
||||
if (!existsSync(partDir)) return []
|
||||
|
||||
const parts: Array<{ id: string; type: string; [key: string]: unknown }> = []
|
||||
try {
|
||||
const files = await readdir(partDir)
|
||||
for (const file of files) {
|
||||
if (!file.endsWith(".json")) continue
|
||||
try {
|
||||
const content = await readFile(join(partDir, file), "utf-8")
|
||||
parts.push(JSON.parse(content))
|
||||
} catch {
|
||||
continue
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
return []
|
||||
}
|
||||
|
||||
return parts.sort((a, b) => a.id.localeCompare(b.id))
|
||||
return getFileSessionMessages(sessionID)
|
||||
}
|
||||
|
||||
export async function readSessionTodos(sessionID: string): Promise<TodoItem[]> {
|
||||
// Beta mode: use SDK
|
||||
if (isSqliteBackend() && sdkClient) {
|
||||
try {
|
||||
const response = await sdkClient.session.todo({ path: { id: sessionID } })
|
||||
const data = normalizeSDKResponse(response, [] as Array<{
|
||||
id?: string
|
||||
content?: string
|
||||
status?: string
|
||||
priority?: string
|
||||
}>)
|
||||
return data.map((item) => ({
|
||||
id: item.id || "",
|
||||
content: item.content || "",
|
||||
status: (item.status as TodoItem["status"]) || "pending",
|
||||
priority: item.priority,
|
||||
}))
|
||||
} catch {
|
||||
return []
|
||||
return await getSdkSessionTodos(sdkClient, sessionID)
|
||||
} catch (error) {
|
||||
if (!shouldFallbackFromSdkError(error)) throw error
|
||||
log("[session-manager] falling back to file session todos after SDK unavailable error", { error: String(error), sessionID })
|
||||
}
|
||||
}
|
||||
|
||||
// Stable mode: use JSON files
|
||||
if (!existsSync(TODO_DIR)) return []
|
||||
|
||||
try {
|
||||
const allFiles = await readdir(TODO_DIR)
|
||||
const todoFiles = allFiles.filter((f) => f === `${sessionID}.json`)
|
||||
|
||||
for (const file of todoFiles) {
|
||||
try {
|
||||
const content = await readFile(join(TODO_DIR, file), "utf-8")
|
||||
const data = JSON.parse(content)
|
||||
if (Array.isArray(data)) {
|
||||
return data.map((item) => ({
|
||||
id: item.id || "",
|
||||
content: item.content || "",
|
||||
status: item.status || "pending",
|
||||
priority: item.priority,
|
||||
}))
|
||||
}
|
||||
} catch {
|
||||
continue
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
return []
|
||||
}
|
||||
|
||||
return []
|
||||
return getFileSessionTodos(sessionID)
|
||||
}
|
||||
|
||||
export async function readSessionTranscript(sessionID: string): Promise<number> {
|
||||
if (!existsSync(TRANSCRIPT_DIR)) return 0
|
||||
|
||||
const transcriptFile = join(TRANSCRIPT_DIR, `${sessionID}.jsonl`)
|
||||
if (!existsSync(transcriptFile)) return 0
|
||||
|
||||
try {
|
||||
const content = await readFile(transcriptFile, "utf-8")
|
||||
return content.trim().split("\n").filter(Boolean).length
|
||||
} catch {
|
||||
return 0
|
||||
}
|
||||
return getFileSessionTranscript(sessionID)
|
||||
}
|
||||
|
||||
export async function getSessionInfo(sessionID: string): Promise<SessionInfo | null> {
|
||||
const messages = await readSessionMessages(sessionID)
|
||||
if (messages.length === 0) return null
|
||||
|
||||
const agentsUsed = new Set<string>()
|
||||
let firstMessage: Date | undefined
|
||||
let lastMessage: Date | undefined
|
||||
|
||||
for (const msg of messages) {
|
||||
if (msg.agent) agentsUsed.add(msg.agent)
|
||||
if (msg.time?.created) {
|
||||
const date = new Date(msg.time.created)
|
||||
if (!firstMessage || date < firstMessage) firstMessage = date
|
||||
if (!lastMessage || date > lastMessage) lastMessage = date
|
||||
}
|
||||
}
|
||||
|
||||
const todos = await readSessionTodos(sessionID)
|
||||
const transcriptEntries = await readSessionTranscript(sessionID)
|
||||
|
||||
return {
|
||||
id: sessionID,
|
||||
message_count: messages.length,
|
||||
first_message: firstMessage,
|
||||
last_message: lastMessage,
|
||||
agents_used: Array.from(agentsUsed),
|
||||
has_todos: todos.length > 0,
|
||||
has_transcript: transcriptEntries > 0,
|
||||
todos,
|
||||
transcript_entries: transcriptEntries,
|
||||
}
|
||||
return getFileSessionInfo(sessionID)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user