oussamadouhou/oh-my-opencode-free-fork
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(session-manager): convert blocking sync I/O to async for improved concurrency

Browse Source 
Convert session-manager storage layer from synchronous blocking I/O (readdirSync, readFileSync) to non-blocking async I/O (readdir, readFile from fs/promises). This fixes hanging issues in session_search and other tools caused by blocking filesystem operations.

Changes:
- storage.ts: getAllSessions, readSessionMessages, getSessionInfo now async
- utils.ts: Updated utility functions to be async-compatible
- tools.ts: Added await calls for async storage functions
- storage.test.ts, utils.test.ts: Updated tests with async/await patterns

This resolves the session_search tool hang issue and improves overall responsiveness.

🤖 Generated with assistance of OhMyOpenCode (https://github.com/code-yeongyu/oh-my-opencode)
This commit is contained in:
YeonGyu-Kim
2025-12-28 14:33:47 +09:00
parent 6d6102f1ff
commit 49f3be5a1f
5 changed files with 269 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
src/tools/session-manager/storage.test.ts
View File

@@ -23,7 +23,8 @@ mock.module("./constants", () => ({
TOOL_NAME_PREFIX: "session_",
}))
const { getAllSessions, getMessageDir, sessionExists, readSessionMessages, readSessionTodos, getSessionInfo } = await import("./storage")
const { getAllSessions, getMessageDir, sessionExists, readSessionMessages, readSessionTodos, getSessionInfo } =
await import("./storage")
describe("session-manager storage", () => {
beforeEach(() => {
@@ -43,48 +44,61 @@ describe("session-manager storage", () => {
}
})
test("getAllSessions returns empty array when no sessions exist", () => {
const sessions = getAllSessions()
test("getAllSessions returns empty array when no sessions exist", async () => {
// #when
const sessions = await getAllSessions()
// #then
expect(Array.isArray(sessions)).toBe(true)
expect(sessions).toEqual([])
})
test("getMessageDir finds session in direct path", () => {
// #given
const sessionID = "ses_test123"
const sessionPath = join(TEST_MESSAGE_STORAGE, sessionID)
mkdirSync(sessionPath, { recursive: true })
writeFileSync(join(sessionPath, "msg_001.json"), JSON.stringify({ id: "msg_001", role: "user" }))
// #when
const result = getMessageDir(sessionID)
// #then
expect(result).toBe(sessionPath)
})
test("sessionExists returns false for non-existent session", () => {
// #when
const exists = sessionExists("ses_nonexistent")
// #then
expect(exists).toBe(false)
})
test("sessionExists returns true for existing session", () => {
// #given
const sessionID = "ses_exists"
const sessionPath = join(TEST_MESSAGE_STORAGE, sessionID)
mkdirSync(sessionPath, { recursive: true })
writeFileSync(join(sessionPath, "msg_001.json"), JSON.stringify({ id: "msg_001" }))
// #when
const exists = sessionExists(sessionID)
// #then
expect(exists).toBe(true)
})
test("readSessionMessages returns empty array for non-existent session", () => {
const messages = readSessionMessages("ses_nonexistent")
test("readSessionMessages returns empty array for non-existent session", async () => {
// #when
const messages = await readSessionMessages("ses_nonexistent")
// #then
expect(messages).toEqual([])
})
test("readSessionMessages sorts messages by timestamp", () => {
test("readSessionMessages sorts messages by timestamp", async () => {
// #given
const sessionID = "ses_test123"
const sessionPath = join(TEST_MESSAGE_STORAGE, sessionID)
mkdirSync(sessionPath, { recursive: true })
@@ -98,26 +112,33 @@ describe("session-manager storage", () => {
JSON.stringify({ id: "msg_001", role: "user", time: { created: 1000 } })
)
const messages = readSessionMessages(sessionID)
// #when
const messages = await readSessionMessages(sessionID)
// #then
expect(messages.length).toBe(2)
expect(messages[0].id).toBe("msg_001")
expect(messages[1].id).toBe("msg_002")
})
test("readSessionTodos returns empty array when no todos exist", () => {
const todos = readSessionTodos("ses_nonexistent")
test("readSessionTodos returns empty array when no todos exist", async () => {
// #when
const todos = await readSessionTodos("ses_nonexistent")
// #then
expect(todos).toEqual([])
})
test("getSessionInfo returns null for non-existent session", () => {
const info = getSessionInfo("ses_nonexistent")
test("getSessionInfo returns null for non-existent session", async () => {
// #when
const info = await getSessionInfo("ses_nonexistent")
// #then
expect(info).toBeNull()
})
test("getSessionInfo aggregates session metadata correctly", () => {
test("getSessionInfo aggregates session metadata correctly", async () => {
// #given
const sessionID = "ses_test123"
const sessionPath = join(TEST_MESSAGE_STORAGE, sessionID)
mkdirSync(sessionPath, { recursive: true })
@@ -142,8 +163,10 @@ describe("session-manager storage", () => {
})
)
const info = getSessionInfo(sessionID)
// #when
const info = await getSessionInfo(sessionID)
// #then
expect(info).not.toBeNull()
expect(info?.id).toBe(sessionID)
expect(info?.message_count).toBe(2)

133
src/tools/session-manager/storage.ts
View File

@@ -1,23 +1,25 @@
import { existsSync, readdirSync, readFileSync } from "node:fs"
import { existsSync, readdirSync } from "node:fs"
import { readdir, readFile } from "node:fs/promises"
import { join } from "node:path"
import { MESSAGE_STORAGE, PART_STORAGE, TODO_DIR, TRANSCRIPT_DIR } from "./constants"
import type { SessionMessage, SessionInfo, TodoItem } from "./types"
export function getAllSessions(): string[] {
export async function getAllSessions(): Promise<string[]> {
if (!existsSync(MESSAGE_STORAGE)) return []
const sessions: string[] = []
function scanDirectory(dir: string): void {
async function scanDirectory(dir: string): Promise<void> {
try {
for (const entry of readdirSync(dir, { withFileTypes: true })) {
const entries = await readdir(dir, { withFileTypes: true })
for (const entry of entries) {
if (entry.isDirectory()) {
const sessionPath = join(dir, entry.name)
const files = readdirSync(sessionPath)
const files = await readdir(sessionPath)
if (files.some((f) => f.endsWith(".json"))) {
sessions.push(entry.name)
} else {
scanDirectory(sessionPath)
await scanDirectory(sessionPath)
}
}
}
@@ -26,7 +28,7 @@ export function getAllSessions(): string[] {
}
}
scanDirectory(MESSAGE_STORAGE)
await scanDirectory(MESSAGE_STORAGE)
return [...new Set(sessions)]
}
@@ -38,11 +40,15 @@ export function getMessageDir(sessionID: string): string {
return directPath
}
for (const dir of readdirSync(MESSAGE_STORAGE)) {
const sessionPath = join(MESSAGE_STORAGE, dir, sessionID)
if (existsSync(sessionPath)) {
return sessionPath
try {
for (const dir of readdirSync(MESSAGE_STORAGE)) {
const sessionPath = join(MESSAGE_STORAGE, dir, sessionID)
if (existsSync(sessionPath)) {
return sessionPath
}
}
} catch {
return ""
}
return ""
@@ -52,29 +58,34 @@ export function sessionExists(sessionID: string): boolean {
return getMessageDir(sessionID) !== ""
}
export function readSessionMessages(sessionID: string): SessionMessage[] {
export async function readSessionMessages(sessionID: string): Promise<SessionMessage[]> {
const messageDir = getMessageDir(sessionID)
if (!messageDir || !existsSync(messageDir)) return []
const messages: SessionMessage[] = []
for (const file of readdirSync(messageDir)) {
if (!file.endsWith(".json")) continue
try {
const content = readFileSync(join(messageDir, file), "utf-8")
const meta = JSON.parse(content)
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 = readParts(meta.id)
const parts = await readParts(meta.id)
messages.push({
id: meta.id,
role: meta.role,
agent: meta.agent,
time: meta.time,
parts,
})
} catch {
continue
messages.push({
id: meta.id,
role: meta.role,
agent: meta.agent,
time: meta.time,
parts,
})
} catch {
continue
}
}
} catch {
return []
}
return messages.sort((a, b) => {
@@ -85,65 +96,75 @@ export function readSessionMessages(sessionID: string): SessionMessage[] {
})
}
function readParts(messageID: string): Array<{ id: string; type: string; [key: string]: unknown }> {
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 }> = []
for (const file of readdirSync(partDir)) {
if (!file.endsWith(".json")) continue
try {
const content = readFileSync(join(partDir, file), "utf-8")
parts.push(JSON.parse(content))
} catch {
continue
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 function readSessionTodos(sessionID: string): TodoItem[] {
export async function readSessionTodos(sessionID: string): Promise<TodoItem[]> {
if (!existsSync(TODO_DIR)) return []
const todoFiles = readdirSync(TODO_DIR).filter((f) => f.includes(sessionID) && f.endsWith(".json"))
try {
const allFiles = await readdir(TODO_DIR)
const todoFiles = allFiles.filter((f) => f.includes(sessionID) && f.endsWith(".json"))
for (const file of todoFiles) {
try {
const content = readFileSync(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,
}))
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 {
continue
}
} catch {
return []
}
return []
}
export function readSessionTranscript(sessionID: string): number {
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 = readFileSync(transcriptFile, "utf-8")
const content = await readFile(transcriptFile, "utf-8")
return content.trim().split("\n").filter(Boolean).length
} catch {
return 0
}
}
export function getSessionInfo(sessionID: string): SessionInfo | null {
const messages = readSessionMessages(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>()
@@ -159,8 +180,8 @@ export function getSessionInfo(sessionID: string): SessionInfo | null {
}
}
const todos = readSessionTodos(sessionID)
const transcriptEntries = readSessionTranscript(sessionID)
const todos = await readSessionTodos(sessionID)
const transcriptEntries = await readSessionTranscript(sessionID)
return {
id: sessionID,

59
src/tools/session-manager/tools.ts
View File

@@ -6,8 +6,25 @@ import {
SESSION_INFO_DESCRIPTION,
} from "./constants"
import { getAllSessions, getSessionInfo, readSessionMessages, readSessionTodos, sessionExists } from "./storage"
import { filterSessionsByDate, formatSessionInfo, formatSessionList, formatSessionMessages, formatSearchResults, searchInSession } from "./utils"
import type { SessionListArgs, SessionReadArgs, SessionSearchArgs, SessionInfoArgs } from "./types"
import {
filterSessionsByDate,
formatSessionInfo,
formatSessionList,
formatSessionMessages,
formatSearchResults,
searchInSession,
} from "./utils"
import type { SessionListArgs, SessionReadArgs, SessionSearchArgs, SessionInfoArgs, SearchResult } from "./types"
const SEARCH_TIMEOUT_MS = 60_000
const MAX_SESSIONS_TO_SCAN = 50
function withTimeout<T>(promise: Promise<T>, ms: number, operation: string): Promise<T> {
return Promise.race([
promise,
new Promise<T>((_, reject) => setTimeout(() => reject(new Error(`${operation} timed out after ${ms}ms`)), ms)),
])
}
export const session_list: ToolDefinition = tool({
description: SESSION_LIST_DESCRIPTION,
@@ -18,17 +35,17 @@ export const session_list: ToolDefinition = tool({
},
execute: async (args: SessionListArgs, _context) => {
try {
let sessions = getAllSessions()
let sessions = await getAllSessions()
if (args.from_date || args.to_date) {
sessions = filterSessionsByDate(sessions, args.from_date, args.to_date)
sessions = await filterSessionsByDate(sessions, args.from_date, args.to_date)
}
if (args.limit && args.limit > 0) {
sessions = sessions.slice(0, args.limit)
}
return formatSessionList(sessions)
return await formatSessionList(sessions)
} catch (e) {
return `Error: ${e instanceof Error ? e.message : String(e)}`
}
@@ -49,13 +66,13 @@ export const session_read: ToolDefinition = tool({
return `Session not found: ${args.session_id}`
}
let messages = readSessionMessages(args.session_id)
let messages = await readSessionMessages(args.session_id)
if (args.limit && args.limit > 0) {
messages = messages.slice(0, args.limit)
}
const todos = args.include_todos ? readSessionTodos(args.session_id) : undefined
const todos = args.include_todos ? await readSessionTodos(args.session_id) : undefined
return formatSessionMessages(messages, args.include_todos, todos)
} catch (e) {
@@ -74,13 +91,31 @@ export const session_search: ToolDefinition = tool({
},
execute: async (args: SessionSearchArgs, _context) => {
try {
const sessions = args.session_id ? [args.session_id] : getAllSessions()
const resultLimit = args.limit && args.limit > 0 ? args.limit : 20
const allResults = sessions.flatMap((sid) => searchInSession(sid, args.query, args.case_sensitive))
const searchOperation = async (): Promise<SearchResult[]> => {
if (args.session_id) {
return searchInSession(args.session_id, args.query, args.case_sensitive, resultLimit)
}
const limited = args.limit && args.limit > 0 ? allResults.slice(0, args.limit) : allResults.slice(0, 20)
const allSessions = await getAllSessions()
const sessionsToScan = allSessions.slice(0, MAX_SESSIONS_TO_SCAN)
return formatSearchResults(limited)
const allResults: SearchResult[] = []
for (const sid of sessionsToScan) {
if (allResults.length >= resultLimit) break
const remaining = resultLimit - allResults.length
const sessionResults = await searchInSession(sid, args.query, args.case_sensitive, remaining)
allResults.push(...sessionResults)
}
return allResults.slice(0, resultLimit)
}
const results = await withTimeout(searchOperation(), SEARCH_TIMEOUT_MS, "Search")
return formatSearchResults(results)
} catch (e) {
return `Error: ${e instanceof Error ? e.message : String(e)}`
}
@@ -94,7 +129,7 @@ export const session_info: ToolDefinition = tool({
},
execute: async (args: SessionInfoArgs, _context) => {
try {
const info = getSessionInfo(args.session_id)
const info = await getSessionInfo(args.session_id)
if (!info) {
return `Session not found: ${args.session_id}`

90
src/tools/session-manager/utils.test.ts
View File

@@ -1,21 +1,39 @@
import { describe, test, expect } from "bun:test"
import { formatSessionList, formatSessionMessages, formatSessionInfo, formatSearchResults, filterSessionsByDate, searchInSession } from "./utils"
import {
formatSessionList,
formatSessionMessages,
formatSessionInfo,
formatSearchResults,
filterSessionsByDate,
searchInSession,
} from "./utils"
import type { SessionInfo, SessionMessage, SearchResult } from "./types"
describe("session-manager utils", () => {
test("formatSessionList handles empty array", () => {
const result = formatSessionList([])
test("formatSessionList handles empty array", async () => {
// #given
const sessions: string[] = []
// #when
const result = await formatSessionList(sessions)
// #then
expect(result).toContain("No sessions found")
})
test("formatSessionMessages handles empty array", () => {
const result = formatSessionMessages([])
// #given
const messages: SessionMessage[] = []
// #when
const result = formatSessionMessages(messages)
// #then
expect(result).toContain("No messages")
})
test("formatSessionMessages includes message content", () => {
// #given
const messages: SessionMessage[] = [
{
id: "msg_001",
@@ -24,14 +42,17 @@ describe("session-manager utils", () => {
parts: [{ id: "prt_001", type: "text", text: "Hello world" }],
},
]
// #when
const result = formatSessionMessages(messages)
// #then
expect(result).toContain("user")
expect(result).toContain("Hello world")
})
test("formatSessionMessages includes todos when requested", () => {
// #given
const messages: SessionMessage[] = [
{
id: "msg_001",
@@ -40,20 +61,22 @@ describe("session-manager utils", () => {
parts: [{ id: "prt_001", type: "text", text: "Test" }],
},
]
const todos = [
{ id: "1", content: "Task 1", status: "completed" as const },
{ id: "2", content: "Task 2", status: "pending" as const },
]
// #when
const result = formatSessionMessages(messages, true, todos)
// #then
expect(result).toContain("Todos")
expect(result).toContain("Task 1")
expect(result).toContain("Task 2")
})
test("formatSessionInfo includes all metadata", () => {
// #given
const info: SessionInfo = {
id: "ses_test123",
message_count: 42,
@@ -65,9 +88,11 @@ describe("session-manager utils", () => {
todos: [{ id: "1", content: "Test", status: "pending" }],
transcript_entries: 123,
}
// #when
const result = formatSessionInfo(info)
// #then
expect(result).toContain("ses_test123")
expect(result).toContain("42")
expect(result).toContain("build, oracle")
@@ -75,12 +100,18 @@ describe("session-manager utils", () => {
})
test("formatSearchResults handles empty array", () => {
const result = formatSearchResults([])
// #given
const results: SearchResult[] = []
// #when
const result = formatSearchResults(results)
// #then
expect(result).toContain("No matches")
})
test("formatSearchResults formats matches correctly", () => {
// #given
const results: SearchResult[] = [
{
session_id: "ses_test123",
@@ -91,9 +122,11 @@ describe("session-manager utils", () => {
timestamp: Date.now(),
},
]
// #when
const result = formatSearchResults(results)
// #then
expect(result).toContain("Found 1 matches")
expect(result).toContain("ses_test123")
expect(result).toContain("msg_001")
@@ -101,17 +134,26 @@ describe("session-manager utils", () => {
expect(result).toContain("Matches: 3")
})
test("filterSessionsByDate filters correctly", () => {
test("filterSessionsByDate filters correctly", async () => {
// #given
const sessionIDs = ["ses_001", "ses_002", "ses_003"]
const result = filterSessionsByDate(sessionIDs)
// #when
const result = await filterSessionsByDate(sessionIDs)
// #then
expect(Array.isArray(result)).toBe(true)
})
test("searchInSession finds matches case-insensitively", () => {
const results = searchInSession("ses_nonexistent", "test", false)
test("searchInSession finds matches case-insensitively", async () => {
// #given
const sessionID = "ses_nonexistent"
const query = "test"
// #when
const results = await searchInSession(sessionID, query, false)
// #then
expect(Array.isArray(results)).toBe(true)
expect(results.length).toBe(0)
})

48
src/tools/session-manager/utils.ts
View File

@@ -1,12 +1,14 @@
import type { SessionInfo, SessionMessage, SearchResult } from "./types"
import { getSessionInfo, readSessionMessages } from "./storage"
export function formatSessionList(sessionIDs: string[]): string {
export async function formatSessionList(sessionIDs: string[]): Promise<string> {
if (sessionIDs.length === 0) {
return "No sessions found."
}
const infos = sessionIDs.map((id) => getSessionInfo(id)).filter((info): info is SessionInfo => info !== null)
const infos = (await Promise.all(sessionIDs.map((id) => getSessionInfo(id)))).filter(
(info): info is SessionInfo => info !== null
)
if (infos.length === 0) {
return "No valid sessions found."
@@ -39,7 +41,11 @@ export function formatSessionList(sessionIDs: string[]): string {
return [formatRow(headers), separator, ...rows.map(formatRow)].join("\n")
}
export function formatSessionMessages(messages: SessionMessage[], includeTodos?: boolean, todos?: Array<{id: string; content: string; status: string}>): string {
export function formatSessionMessages(
messages: SessionMessage[],
includeTodos?: boolean,
todos?: Array<{ id: string; content: string; status: string }>
): string {
if (messages.length === 0) {
return "No messages found in this session."
}
@@ -116,32 +122,46 @@ export function formatSearchResults(results: SearchResult[]): string {
return lines.join("\n")
}
export function filterSessionsByDate(sessionIDs: string[], fromDate?: string, toDate?: string): string[] {
export async function filterSessionsByDate(
sessionIDs: string[],
fromDate?: string,
toDate?: string
): Promise<string[]> {
if (!fromDate && !toDate) return sessionIDs
const from = fromDate ? new Date(fromDate) : null
const to = toDate ? new Date(toDate) : null
return sessionIDs.filter((id) => {
const info = getSessionInfo(id)
if (!info || !info.last_message) return false
const results: string[] = []
for (const id of sessionIDs) {
const info = await getSessionInfo(id)
if (!info || !info.last_message) continue
if (from && info.last_message < from) return false
if (to && info.last_message > to) return false
if (from && info.last_message < from) continue
if (to && info.last_message > to) continue
return true
})
results.push(id)
}
return results
}
export function searchInSession(sessionID: string, query: string, caseSensitive = false): SearchResult[] {
const messages = readSessionMessages(sessionID)
export async function searchInSession(
sessionID: string,
query: string,
caseSensitive = false,
maxResults?: number
): Promise<SearchResult[]> {
const messages = await readSessionMessages(sessionID)
const results: SearchResult[] = []
const searchQuery = caseSensitive ? query : query.toLowerCase()
for (const msg of messages) {
if (maxResults && results.length >= maxResults) break
let matchCount = 0
let excerpts: string[] = []
const excerpts: string[] = []
for (const part of msg.parts) {
if (part.type === "text" && part.text) {