diff --git a/packages/agent/src/harness/agent-harness.ts b/packages/agent/src/harness/agent-harness.ts index cbc2d0a4a..c8004aa0d 100644 --- a/packages/agent/src/harness/agent-harness.ts +++ b/packages/agent/src/harness/agent-harness.ts @@ -529,7 +529,7 @@ export class AgentHarness< } else if (write.type === "session_info") { await this.session.appendSessionName(write.name ?? ""); } else if (write.type === "leaf") { - await this.session.getStorage().setLeafId(write.targetId); + await this.session.moveTo(write.targetId); } this.pendingSessionWrites.shift(); } diff --git a/packages/agent/src/harness/session/jsonl-repo.ts b/packages/agent/src/harness/session/jsonl-repo.ts index f1f165223..62fcec5b3 100644 --- a/packages/agent/src/harness/session/jsonl-repo.ts +++ b/packages/agent/src/harness/session/jsonl-repo.ts @@ -5,12 +5,15 @@ import type { JsonlSessionMetadata, JsonlSessionRepoApi, Session, + SessionSearchHit, + SessionSearchOptions, } from "../types.ts"; import { SessionError, toError } from "../types.ts"; import { JsonlSessionStorage, loadJsonlSessionMetadata } from "./jsonl-storage.ts"; import { createSessionId, createTimestamp, + findSessionEntryMatches, getEntriesToFork, getFileSystemResultOrThrow, toSession, @@ -124,6 +127,15 @@ export class JsonlSessionRepo implements JsonlSessionRepoApi { return sessions; } + async search(options: SessionSearchOptions): Promise[]> { + const hits: SessionSearchHit[] = []; + for (const metadata of await this.list({ cwd: options.cwd })) { + const session = await this.open(metadata); + hits.push(...findSessionEntryMatches(metadata, await session.getEntries(), options.text)); + } + return hits; + } + async delete(metadata: JsonlSessionMetadata): Promise { getFileSystemResultOrThrow( await this.fs.remove(metadata.path, { force: true }), diff --git a/packages/agent/src/harness/session/jsonl-storage.ts b/packages/agent/src/harness/session/jsonl-storage.ts index fea97fbdc..943ccfb0a 100644 --- a/packages/agent/src/harness/session/jsonl-storage.ts +++ b/packages/agent/src/harness/session/jsonl-storage.ts @@ -251,7 +251,7 @@ export class JsonlSessionStorage implements SessionStorage return this.currentLeafId; } - async setLeafId(leafId: string | null): Promise { + async setLeafId(leafId: string | null): Promise { if (leafId !== null && !this.byId.has(leafId)) { throw new SessionError("not_found", `Entry ${leafId} not found`); } @@ -269,6 +269,7 @@ export class JsonlSessionStorage implements SessionStorage this.entries.push(entry); this.byId.set(entry.id, entry); this.currentLeafId = leafId; + return entry; } async createEntryId(): Promise { diff --git a/packages/agent/src/harness/session/memory-repo.ts b/packages/agent/src/harness/session/memory-repo.ts index 145bc2b7c..34ab0ffe5 100644 --- a/packages/agent/src/harness/session/memory-repo.ts +++ b/packages/agent/src/harness/session/memory-repo.ts @@ -1,6 +1,19 @@ -import { type Session, SessionError, type SessionMetadata, type SessionRepo } from "../types.ts"; +import { + type Session, + SessionError, + type SessionMetadata, + type SessionRepo, + type SessionSearchHit, + type SessionSearchOptions, +} from "../types.ts"; import { InMemorySessionStorage } from "./memory-storage.ts"; -import { createSessionId, createTimestamp, getEntriesToFork, toSession } from "./repo-utils.ts"; +import { + createSessionId, + createTimestamp, + findSessionEntryMatches, + getEntriesToFork, + toSession, +} from "./repo-utils.ts"; export class InMemorySessionRepo implements SessionRepo { private sessions = new Map>(); @@ -28,6 +41,15 @@ export class InMemorySessionRepo implements SessionRepo session.getMetadata())); } + async search(options: SessionSearchOptions): Promise[]> { + const hits: SessionSearchHit[] = []; + for (const session of this.sessions.values()) { + const metadata = await session.getMetadata(); + hits.push(...findSessionEntryMatches(metadata, await session.getEntries(), options.text)); + } + return hits; + } + async delete(metadata: SessionMetadata): Promise { this.sessions.delete(metadata.id); } diff --git a/packages/agent/src/harness/session/memory-storage.ts b/packages/agent/src/harness/session/memory-storage.ts index 1bc24f04f..d8e53be3b 100644 --- a/packages/agent/src/harness/session/memory-storage.ts +++ b/packages/agent/src/harness/session/memory-storage.ts @@ -72,7 +72,7 @@ export class InMemorySessionStorage { + async setLeafId(leafId: string | null): Promise { if (leafId !== null && !this.byId.has(leafId)) { throw new SessionError("not_found", `Entry ${leafId} not found`); } @@ -86,6 +86,7 @@ export class InMemorySessionStorage { diff --git a/packages/agent/src/harness/session/repo-utils.ts b/packages/agent/src/harness/session/repo-utils.ts index fd6ee4f2a..e830e7678 100644 --- a/packages/agent/src/harness/session/repo-utils.ts +++ b/packages/agent/src/harness/session/repo-utils.ts @@ -4,6 +4,8 @@ import { type Result, SessionError, type SessionMetadata, + type SessionSearchHit, + type SessionSearchIndexSink, type SessionStorage, type SessionTreeEntry, } from "../types.ts"; @@ -17,8 +19,25 @@ export function createTimestamp(): string { return new Date().toISOString(); } -export function toSession(storage: SessionStorage): Session { - return new Session(storage); +export function toSession( + storage: SessionStorage, + searchIndexSink?: SessionSearchIndexSink, +): Session { + return new Session(storage, {}, searchIndexSink); +} + +export function findSessionEntryMatches( + metadata: TMetadata, + entries: SessionTreeEntry[], + text: string, +): SessionSearchHit[] { + const normalizedText = text.trim().toLowerCase(); + if (!normalizedText) return []; + return entries.flatMap((entry) => { + const payload = JSON.stringify(entry); + if (!payload.toLowerCase().includes(normalizedText)) return []; + return [{ metadata, entryId: entry.id, timestamp: entry.timestamp, snippet: payload }]; + }); } export function getFileSystemResultOrThrow(result: Result, message: string): TValue { diff --git a/packages/agent/src/harness/session/search-index-sink.ts b/packages/agent/src/harness/session/search-index-sink.ts new file mode 100644 index 000000000..7745e99cd --- /dev/null +++ b/packages/agent/src/harness/session/search-index-sink.ts @@ -0,0 +1,8 @@ +import type { SessionMetadata, SessionSearchIndexRecord, SessionTreeEntry } from "../types.ts"; + +export function toSessionSearchIndexRecord( + metadata: TMetadata, + entry: SessionTreeEntry, +): SessionSearchIndexRecord { + return { metadata, entry }; +} diff --git a/packages/agent/src/harness/session/session.ts b/packages/agent/src/harness/session/session.ts index 13007bc65..6e380f519 100644 --- a/packages/agent/src/harness/session/session.ts +++ b/packages/agent/src/harness/session/session.ts @@ -14,12 +14,14 @@ import type { SessionEntryCursorOptions, SessionInfoEntry, SessionMetadata, + SessionSearchIndexSink, SessionStats, SessionStorage, SessionTreeEntry, ThinkingLevelChangeEntry, } from "../types.ts"; import { SessionError } from "../types.ts"; +import { toSessionSearchIndexRecord } from "./search-index-sink.ts"; export type ContextEntryTransform = (entries: readonly SessionTreeEntry[]) => readonly SessionTreeEntry[]; @@ -150,10 +152,16 @@ export function buildSessionContext( export class Session { private storage: SessionStorage; private contextBuildOptions: SessionContextBuildOptions; + private searchIndexSink: SessionSearchIndexSink | undefined; - constructor(storage: SessionStorage, contextBuildOptions: SessionContextBuildOptions = {}) { + constructor( + storage: SessionStorage, + contextBuildOptions: SessionContextBuildOptions = {}, + searchIndexSink?: SessionSearchIndexSink, + ) { this.storage = storage; this.contextBuildOptions = contextBuildOptions; + this.searchIndexSink = searchIndexSink; } getMetadata(): Promise { @@ -211,8 +219,15 @@ export class Session { return this.storage.getSessionName(); } + private async writeSearchIndexRecord(entry: SessionTreeEntry): Promise { + if (this.searchIndexSink) { + await this.searchIndexSink.write(toSessionSearchIndexRecord(await this.storage.getMetadata(), entry)); + } + } + private async appendTypedEntry(entry: TEntry): Promise { await this.storage.appendEntry(entry); + await this.writeSearchIndexRecord(entry); return entry.id; } @@ -342,7 +357,8 @@ export class Session { if (entryId !== null && !(await this.storage.getEntry(entryId))) { throw new SessionError("not_found", `Entry ${entryId} not found`); } - await this.storage.setLeafId(entryId); + const leafEntry = await this.storage.setLeafId(entryId); + await this.writeSearchIndexRecord(leafEntry); if (!summary) return undefined; return this.appendTypedEntry({ type: "branch_summary", diff --git a/packages/agent/src/harness/types.ts b/packages/agent/src/harness/types.ts index 7a217ae34..76743791e 100644 --- a/packages/agent/src/harness/types.ts +++ b/packages/agent/src/harness/types.ts @@ -499,7 +499,7 @@ export interface SessionStorage; getLeafId(): Promise; /** Persist a leaf entry that records the active session-tree leaf. */ - setLeafId(leafId: string | null): Promise; + setLeafId(leafId: string | null): Promise; createEntryId(): Promise; appendEntry(entry: SessionTreeEntry): Promise; getEntry(id: string): Promise; @@ -519,6 +519,30 @@ export interface SessionCreateOptions { id?: string; } +export interface SessionSearchOptions { + text: string; + cwd?: string; +} + +export interface SessionSearchHit { + metadata: TMetadata; + entryId: string; + timestamp: string; + snippet?: string; + score?: number; +} + +/** A persisted session entry delivered to a search index. */ +export interface SessionSearchIndexRecord { + metadata: TMetadata; + entry: SessionTreeEntry; +} + +/** Receives entries after their session storage has persisted them for search indexing. */ +export interface SessionSearchIndexSink { + write(record: SessionSearchIndexRecord): Promise; +} + export interface SessionForkOptions { entryId?: string; position?: "before" | "at"; @@ -533,6 +557,7 @@ export interface SessionRepo< create(options: TCreateOptions): Promise>; open(metadata: TMetadata): Promise>; list(options?: TListOptions): Promise; + search(options: SessionSearchOptions): Promise[]>; delete(metadata: TMetadata): Promise; fork(source: TMetadata, options: SessionForkOptions & TCreateOptions): Promise>; } diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index e9bed1550..c0583af3d 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -34,6 +34,7 @@ export * from "./harness/session/jsonl-storage.ts"; export * from "./harness/session/memory-repo.ts"; export * from "./harness/session/memory-storage.ts"; export * from "./harness/session/repo-utils.ts"; +export * from "./harness/session/search-index-sink.ts"; export * from "./harness/session/session.ts"; export * from "./harness/skills.ts"; export * from "./harness/system-prompt.ts"; diff --git a/packages/agent/test/harness/repo.test.ts b/packages/agent/test/harness/repo.test.ts index 009fb0ce7..aeaa5b0f0 100644 --- a/packages/agent/test/harness/repo.test.ts +++ b/packages/agent/test/harness/repo.test.ts @@ -22,6 +22,16 @@ describe("InMemorySessionRepo", () => { await repo.delete(metadata); await expect(repo.open(metadata)).rejects.toThrow("Session not found: session-1"); }); + + it("searches canonical entry JSON", async () => { + const repo = new InMemorySessionRepo(); + const session = await repo.create({ id: "session-1" }); + const entryId = await session.appendMessage(createUserMessage("Find the auth defect")); + + await expect(repo.search({ text: "AUTH" })).resolves.toEqual([ + expect.objectContaining({ entryId, metadata: expect.objectContaining({ id: "session-1" }) }), + ]); + }); }); describe("JsonlSessionRepo", () => { @@ -44,6 +54,20 @@ describe("JsonlSessionRepo", () => { ); }); + it("searches session entries and filters by cwd", async () => { + const root = createTempDir(); + const env = new NodeExecutionEnv({ cwd: root }); + const repo = new JsonlSessionRepo({ fs: env, sessionsRoot: root }); + const included = await repo.create({ cwd: "/tmp/included", id: "included" }); + const excluded = await repo.create({ cwd: "/tmp/excluded", id: "excluded" }); + const entryId = await included.appendMessage(createUserMessage("Find the auth defect")); + await excluded.appendMessage(createUserMessage("Find the auth defect")); + + await expect(repo.search({ text: "AUTH", cwd: "/tmp/included" })).resolves.toEqual([ + expect.objectContaining({ entryId, metadata: expect.objectContaining({ id: "included" }) }), + ]); + }); + it("opens, deletes, and forks by metadata", async () => { const root = createTempDir(); const env = new NodeExecutionEnv({ cwd: root }); diff --git a/packages/agent/test/harness/sqlite-migrations.test.ts b/packages/agent/test/harness/sqlite-migrations.test.ts index 0cf74938d..6b91eb9d7 100644 --- a/packages/agent/test/harness/sqlite-migrations.test.ts +++ b/packages/agent/test/harness/sqlite-migrations.test.ts @@ -5,6 +5,7 @@ import { describe, expect, it } from "vitest"; import { applyMigrations, createNodeSqliteFactory, + loadMigrations, type SqliteDatabase, type SqliteDatabaseFactory, type SqliteRunResult, @@ -75,7 +76,7 @@ describe("SQLite migrations", () => { const db = await sqlite.open(databasePath); try { const rows = await db.prepare("SELECT id FROM migrations ORDER BY id").all<{ id: string }>(); - expect(rows.map((row) => row.id)).toEqual(["001_initial.sql"]); + expect(rows.map((row) => row.id)).toEqual(["001_initial.sql", "002_session_search.sql"]); const tables = await db .prepare("SELECT name, sql FROM sqlite_master WHERE type = 'table' ORDER BY name") .all<{ name: string; sql: string | null }>(); @@ -107,6 +108,53 @@ describe("SQLite migrations", () => { } }); + it("does not materialize existing entry data when the search migration runs", async () => { + const root = createTempDir(); + const db = await createNodeSqliteFactory().open(join(root, "sessions.sqlite")); + try { + const [initialMigration] = await loadMigrations(); + await db.exec(initialMigration!.sql); + await db.exec("CREATE TABLE migrations (id TEXT PRIMARY KEY, applied_at TEXT NOT NULL)"); + await db.prepare("INSERT INTO migrations (id, applied_at) VALUES (?, ?)").run("001_initial.sql", "now"); + await db + .prepare( + "INSERT INTO session_entries (session_id, id, entry_seq, parent_id, type, timestamp, payload) VALUES (?, ?, ?, ?, ?, ?, ?)", + ) + .run( + "session-1", + "entry-1", + 1, + null, + "custom", + "2026-01-01T00:00:00.000Z", + '{"customType":"note","data":{"text":"backfilled"}}', + ); + + await applyMigrations(db); + + await expect( + db + .prepare( + "SELECT session_entries.session_id, session_entries.id AS entry_id FROM session_entry_search_fts JOIN session_entries ON session_entries.rowid = session_entry_search_fts.rowid WHERE session_entry_search_fts MATCH ?", + ) + .get<{ session_id: string; entry_id: string }>("backfilled"), + ).resolves.toBeUndefined(); + + await expect( + db.prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'session_search_records'").get(), + ).resolves.toBeUndefined(); + await expect( + db + .prepare( + "SELECT name FROM sqlite_master WHERE type = 'trigger' AND name LIKE 'session_entries_search_%'", + ) + .get(), + ).resolves.toBeUndefined(); + } finally { + await db.close(); + } + }); + it("persists session metadata through create, list, open, and fork", async () => { const root = createTempDir(); const databasePath = join(root, "sessions.sqlite"); diff --git a/packages/agent/test/harness/sqlite-node.test.ts b/packages/agent/test/harness/sqlite-node.test.ts index a94d37858..a4248aa96 100644 --- a/packages/agent/test/harness/sqlite-node.test.ts +++ b/packages/agent/test/harness/sqlite-node.test.ts @@ -1,7 +1,8 @@ import { join } from "node:path"; import { describe, expect, it } from "vitest"; -import { createNodeSqliteFactory } from "../../../storage/sqlite-node/src/index.ts"; -import { createTempDir } from "./session-test-utils.ts"; +import { createNodeSqliteFactory, SqliteSessionRepo } from "../../../storage/sqlite-node/src/index.ts"; +import { NodeExecutionEnv } from "../../src/harness/env/nodejs.ts"; +import { createTempDir, createUserMessage } from "./session-test-utils.ts"; describe("sqlite-node adapter", () => { it("supports node:sqlite-style named parameters", async () => { @@ -18,4 +19,29 @@ describe("sqlite-node adapter", () => { await db.close(); } }); + + it("searches and deletes transactionally indexed canonical entries", async () => { + const root = createTempDir(); + const databasePath = join(root, "sessions.sqlite"); + const repo = new SqliteSessionRepo({ + env: new NodeExecutionEnv({ cwd: root }), + sqlite: createNodeSqliteFactory(), + databasePath, + }); + const included = await repo.create({ cwd: root, id: "included" }); + const excluded = await repo.create({ cwd: `${root}/other`, id: "excluded" }); + const metadata = await included.getMetadata(); + const entryId = await included.appendMessage(createUserMessage("Find the auth defect")); + await excluded.appendMessage(createUserMessage("Find the auth defect")); + + await expect(repo.search({ text: "auth", cwd: root })).resolves.toEqual([ + expect.objectContaining({ entryId, metadata: expect.objectContaining({ id: "included" }) }), + ]); + await expect(repo.search({ text: "uth", cwd: root })).resolves.toEqual([ + expect.objectContaining({ entryId, metadata: expect.objectContaining({ id: "included" }) }), + ]); + + await repo.delete(metadata); + await expect(repo.search({ text: "auth", cwd: root })).resolves.toEqual([]); + }); }); diff --git a/packages/agent/vitest.config.ts b/packages/agent/vitest.config.ts index 834c56a55..ed0ab064a 100644 --- a/packages/agent/vitest.config.ts +++ b/packages/agent/vitest.config.ts @@ -3,6 +3,7 @@ import { defineConfig } from "vitest/config"; const aiSrcIndex = fileURLToPath(new URL("../ai/src/index.ts", import.meta.url)); const aiSrcCompat = fileURLToPath(new URL("../ai/src/compat.ts", import.meta.url)); +const agentSrcIndex = fileURLToPath(new URL("./src/index.ts", import.meta.url)); export default defineConfig({ test: { @@ -14,6 +15,7 @@ export default defineConfig({ }, resolve: { alias: [ + { find: /^@earendil-works\/pi-agent-core$/, replacement: agentSrcIndex }, { find: /^@earendil-works\/pi-ai$/, replacement: aiSrcIndex }, { find: /^@earendil-works\/pi-ai\/compat$/, replacement: aiSrcCompat }, ], diff --git a/packages/ai/test/openai-responses-reasoning-replay-e2e.test.ts b/packages/ai/test/openai-responses-reasoning-replay-e2e.test.ts index 85753c039..26cd19d90 100644 --- a/packages/ai/test/openai-responses-reasoning-replay-e2e.test.ts +++ b/packages/ai/test/openai-responses-reasoning-replay-e2e.test.ts @@ -83,14 +83,14 @@ describe.skipIf(!process.env.OPENAI_API_KEY || !process.env.ANTHROPIC_API_KEY)( it("handles same-provider different-model handoff with tool calls", { retry: 2 }, async () => { // This tests the scenario where: // 1. Model A (gpt-5-mini) generates reasoning + function_call - // 2. User switches to Model B (gpt-5.2-codex) - same provider, different model + // 2. User switches to Model B (gpt-5.3-codex) - same provider, different model // 3. transform-messages: isSameModel=false, thinking converted to text // 4. But tool call ID still has OpenAI pairing history (fc_xxx paired with rs_xxx) // 5. Without fix: OpenAI returns 400 "function_call without required reasoning item" // 6. With fix: tool calls/results converted to text, conversation continues const modelA = getModel("openai", "gpt-5-mini"); - const modelB = getModel("openai", "gpt-5.2-codex"); + const modelB = getModel("openai", "gpt-5.3-codex"); const apiKey = getEnvApiKey("openai"); if (!apiKey) { @@ -189,7 +189,7 @@ describe.skipIf(!process.env.OPENAI_API_KEY || !process.env.ANTHROPIC_API_KEY)( // 5. Should work because foreign IDs have no pairing expectation const anthropicModel = getModel("anthropic", "claude-sonnet-4-5"); - const codexModel = getModel("openai", "gpt-5.2-codex"); + const codexModel = getModel("openai", "gpt-5.3-codex"); const anthropicApiKey = getEnvApiKey("anthropic"); const openaiApiKey = getEnvApiKey("openai"); diff --git a/packages/ai/test/xhigh.test.ts b/packages/ai/test/xhigh.test.ts index f1722b9a7..1e2d461ba 100644 --- a/packages/ai/test/xhigh.test.ts +++ b/packages/ai/test/xhigh.test.ts @@ -18,7 +18,7 @@ describe.skipIf(!process.env.OPENAI_API_KEY)("xhigh reasoning", () => { describe("codex-max (supports xhigh)", () => { // Note: codex models only support the responses API, not chat completions it("should work with openai-responses", async () => { - const model = getModel("openai", "gpt-5.1-codex-max"); + const model = getModel("openai", "gpt-5.3-codex"); const s = stream(model, makeContext(), { reasoningEffort: "xhigh" }); let hasThinking = false; diff --git a/packages/storage/sqlite-node/src/sqlite/migrations.ts b/packages/storage/sqlite-node/src/sqlite/migrations.ts index 2ccd70fc3..69b8a9709 100644 --- a/packages/storage/sqlite-node/src/sqlite/migrations.ts +++ b/packages/storage/sqlite-node/src/sqlite/migrations.ts @@ -19,6 +19,11 @@ export async function loadMigrations(): Promise { order: 1, sql: await loadMigrationSql("./migrations/001_initial.sql"), }, + { + id: "002_session_search.sql", + order: 2, + sql: await loadMigrationSql("./migrations/002_session_search.sql"), + }, ]; } diff --git a/packages/storage/sqlite-node/src/sqlite/migrations/002_session_search.sql b/packages/storage/sqlite-node/src/sqlite/migrations/002_session_search.sql new file mode 100644 index 000000000..c6681c8ad --- /dev/null +++ b/packages/storage/sqlite-node/src/sqlite/migrations/002_session_search.sql @@ -0,0 +1,5 @@ +CREATE VIRTUAL TABLE IF NOT EXISTS session_entry_search_fts USING fts5( + search_text, + content = '', + tokenize = 'trigram remove_diacritics 1' +); diff --git a/packages/storage/sqlite-node/src/sqlite/repo.ts b/packages/storage/sqlite-node/src/sqlite/repo.ts index 12a4f9d02..4efc7e6ff 100644 --- a/packages/storage/sqlite-node/src/sqlite/repo.ts +++ b/packages/storage/sqlite-node/src/sqlite/repo.ts @@ -1,13 +1,22 @@ -import type { Session, SessionStorage, SessionTreeEntry } from "@earendil-works/pi-agent-core"; +import type { + Session, + SessionSearchHit, + SessionSearchOptions, + SessionStorage, + SessionTreeEntry, +} from "@earendil-works/pi-agent-core"; import { createSessionId, getEntriesToFork, getFileSystemResultOrThrow, SessionError, toSession, + toSessionSearchIndexRecord, } from "@earendil-works/pi-agent-core"; import { applyMigrations } from "./migrations.ts"; +import { SqliteSessionSearchIndex } from "./search-index-sink.ts"; import { SqliteSessionStorage } from "./storage/index.ts"; +import { decodeEntry, type SessionEntryRow } from "./storage/session-entries.ts"; import { rowToMetadata, type SessionRow } from "./storage/sessions.ts"; import type { SqliteDatabase, @@ -94,7 +103,7 @@ export class SqliteSessionRepo implements SqliteSessionRepoApi { parentSessionId: options.parentSessionId, metadata: options.metadata, }); - return toSession(storage); + return toSession(storage, new SqliteSessionSearchIndex(db, await this.getDatabasePath())); } catch (error) { await db.close(); throw error; @@ -110,7 +119,7 @@ export class SqliteSessionRepo implements SqliteSessionRepoApi { const db = await this.openDatabase(); try { const storage = await SqliteSessionStorage.open(db, metadata); - return toSession(storage); + return toSession(storage, new SqliteSessionSearchIndex(db, await this.getDatabasePath())); } catch (error) { await db.close(); throw error; @@ -141,10 +150,30 @@ export class SqliteSessionRepo implements SqliteSessionRepoApi { } } + async search(options: SessionSearchOptions): Promise[]> { + const path = await this.getDatabasePath(); + if (!getFileSystemResultOrThrow(await this.env.exists(path), `Failed to check database ${path}`)) return []; + const db = await this.openDatabase(); + try { + return await new SqliteSessionSearchIndex(db, path).search(options); + } finally { + await db.close(); + } + } + async delete(metadata: SqliteSessionMetadata): Promise { const db = await this.openDatabase(); try { await db.transaction(async () => { + const searchIndexSink = new SqliteSessionSearchIndex(db, await this.getDatabasePath()); + const entries = await db + .prepare( + "SELECT rowid, session_id, id, entry_seq, parent_id, type, timestamp, payload FROM session_entries WHERE session_id = ?", + ) + .all(metadata.id); + for (const row of entries) { + await searchIndexSink.remove(toSessionSearchIndexRecord(metadata, decodeEntry(row))); + } await db.prepare("DELETE FROM branch_entries WHERE session_id = ?").run(metadata.id); await db.prepare("DELETE FROM session_entries WHERE session_id = ?").run(metadata.id); await db.prepare("DELETE FROM entry_materialized WHERE session_id = ?").run(metadata.id); @@ -180,10 +209,13 @@ export class SqliteSessionRepo implements SqliteSessionRepoApi { parentSessionId: options.parentSessionId ?? sourceMetadata.id, metadata: options.metadata ?? sourceMetadata.metadata, }); + const searchIndexSink = new SqliteSessionSearchIndex(db, await this.getDatabasePath()); + const metadata = await storage.getMetadata(); for (const entry of forkedEntries) { await storage.appendEntry(entry); + await searchIndexSink.write(toSessionSearchIndexRecord(metadata, entry)); } - return toSession(storage); + return toSession(storage, searchIndexSink); } catch (error) { await db.close(); throw error; diff --git a/packages/storage/sqlite-node/src/sqlite/search-index-sink.ts b/packages/storage/sqlite-node/src/sqlite/search-index-sink.ts new file mode 100644 index 000000000..0f271a51c --- /dev/null +++ b/packages/storage/sqlite-node/src/sqlite/search-index-sink.ts @@ -0,0 +1,65 @@ +import type { + SessionSearchHit, + SessionSearchIndexRecord, + SessionSearchIndexSink, + SessionSearchOptions, +} from "@earendil-works/pi-agent-core"; +import { SessionError } from "@earendil-works/pi-agent-core"; +import { rowToMetadata, type SessionRow } from "./storage/sessions.ts"; +import type { SqliteDatabase, SqliteSessionMetadata } from "./types.ts"; + +/** SQLite-backed search index and sink for persisted session entries. */ +export class SqliteSessionSearchIndex implements SessionSearchIndexSink { + private readonly db: SqliteDatabase; + private readonly databasePath: string; + + constructor(db: SqliteDatabase, databasePath: string) { + this.db = db; + this.databasePath = databasePath; + } + + async write(record: SessionSearchIndexRecord): Promise { + const row = await this.db + .prepare("SELECT rowid FROM session_entries WHERE session_id = ? AND id = ?") + .get<{ rowid: number }>(record.metadata.id, record.entry.id); + if (!row) { + throw new SessionError("not_found", `Entry ${record.entry.id} not found for search indexing`); + } + await this.db + .prepare("INSERT INTO session_entry_search_fts (rowid, search_text) VALUES (?, ?)") + .run(row.rowid, JSON.stringify(record.entry)); + } + + async remove(record: SessionSearchIndexRecord): Promise { + const row = await this.db + .prepare("SELECT rowid FROM session_entries WHERE session_id = ? AND id = ?") + .get<{ rowid: number }>(record.metadata.id, record.entry.id); + if (!row) return; + await this.db + .prepare( + "INSERT INTO session_entry_search_fts(session_entry_search_fts, rowid, search_text) VALUES ('delete', ?, ?)", + ) + .run(row.rowid, JSON.stringify(record.entry)); + } + + async search(options: SessionSearchOptions): Promise[]> { + const text = options.text.trim(); + if (!text) return []; + const query = `"${text.replaceAll('"', '""')}"`; + const rows = await this.db + .prepare( + "SELECT sessions.id, sessions.created_at, sessions.metadata, sessions.cwd, sessions.parent_session_id, sessions.active_leaf_id, session_entries.id AS entry_id, session_entries.timestamp, bm25(session_entry_search_fts) AS score FROM session_entry_search_fts JOIN session_entries ON session_entries.rowid = session_entry_search_fts.rowid JOIN sessions ON sessions.id = session_entries.session_id WHERE session_entry_search_fts MATCH ? AND (? IS NULL OR sessions.cwd = ?) ORDER BY score", + ) + .all( + query, + options.cwd ?? null, + options.cwd ?? null, + ); + return rows.map((row) => ({ + metadata: rowToMetadata(row, this.databasePath), + entryId: row.entry_id, + timestamp: row.timestamp, + score: row.score, + })); + } +} diff --git a/packages/storage/sqlite-node/src/sqlite/storage/index.ts b/packages/storage/sqlite-node/src/sqlite/storage/index.ts index 93376852e..833d5d7da 100644 --- a/packages/storage/sqlite-node/src/sqlite/storage/index.ts +++ b/packages/storage/sqlite-node/src/sqlite/storage/index.ts @@ -263,7 +263,7 @@ export class SqliteSessionStorage implements SessionStorage { + async setLeafId(leafId: string | null): Promise { if (leafId !== null && !(await this.getEntry(leafId))) { throw new SessionError("not_found", `Entry ${leafId} not found`); } @@ -275,6 +275,7 @@ export class SqliteSessionStorage implements SessionStorage {