mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
feat: search index sqlite
This commit is contained in:
parent
7df73a00c6
commit
0f5a6e422b
21 changed files with 335 additions and 22 deletions
|
|
@ -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();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<SessionSearchHit<JsonlSessionMetadata>[]> {
|
||||
const hits: SessionSearchHit<JsonlSessionMetadata>[] = [];
|
||||
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<void> {
|
||||
getFileSystemResultOrThrow(
|
||||
await this.fs.remove(metadata.path, { force: true }),
|
||||
|
|
|
|||
|
|
@ -251,7 +251,7 @@ export class JsonlSessionStorage implements SessionStorage<JsonlSessionMetadata>
|
|||
return this.currentLeafId;
|
||||
}
|
||||
|
||||
async setLeafId(leafId: string | null): Promise<void> {
|
||||
async setLeafId(leafId: string | null): Promise<LeafEntry> {
|
||||
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<JsonlSessionMetadata>
|
|||
this.entries.push(entry);
|
||||
this.byId.set(entry.id, entry);
|
||||
this.currentLeafId = leafId;
|
||||
return entry;
|
||||
}
|
||||
|
||||
async createEntryId(): Promise<string> {
|
||||
|
|
|
|||
|
|
@ -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<SessionMetadata, { id?: string }, void> {
|
||||
private sessions = new Map<string, Session<SessionMetadata>>();
|
||||
|
|
@ -28,6 +41,15 @@ export class InMemorySessionRepo implements SessionRepo<SessionMetadata, { id?:
|
|||
return Promise.all([...this.sessions.values()].map((session) => session.getMetadata()));
|
||||
}
|
||||
|
||||
async search(options: SessionSearchOptions): Promise<SessionSearchHit<SessionMetadata>[]> {
|
||||
const hits: SessionSearchHit<SessionMetadata>[] = [];
|
||||
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<void> {
|
||||
this.sessions.delete(metadata.id);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -72,7 +72,7 @@ export class InMemorySessionStorage<TMetadata extends SessionMetadata = SessionM
|
|||
return this.leafId;
|
||||
}
|
||||
|
||||
async setLeafId(leafId: string | null): Promise<void> {
|
||||
async setLeafId(leafId: string | null): Promise<LeafEntry> {
|
||||
if (leafId !== null && !this.byId.has(leafId)) {
|
||||
throw new SessionError("not_found", `Entry ${leafId} not found`);
|
||||
}
|
||||
|
|
@ -86,6 +86,7 @@ export class InMemorySessionStorage<TMetadata extends SessionMetadata = SessionM
|
|||
this.entries.push(entry);
|
||||
this.byId.set(entry.id, entry);
|
||||
this.leafId = leafId;
|
||||
return entry;
|
||||
}
|
||||
|
||||
async createEntryId(): Promise<string> {
|
||||
|
|
|
|||
|
|
@ -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<TMetadata extends SessionMetadata>(storage: SessionStorage<TMetadata>): Session<TMetadata> {
|
||||
return new Session(storage);
|
||||
export function toSession<TMetadata extends SessionMetadata>(
|
||||
storage: SessionStorage<TMetadata>,
|
||||
searchIndexSink?: SessionSearchIndexSink,
|
||||
): Session<TMetadata> {
|
||||
return new Session(storage, {}, searchIndexSink);
|
||||
}
|
||||
|
||||
export function findSessionEntryMatches<TMetadata extends SessionMetadata>(
|
||||
metadata: TMetadata,
|
||||
entries: SessionTreeEntry[],
|
||||
text: string,
|
||||
): SessionSearchHit<TMetadata>[] {
|
||||
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<TValue>(result: Result<TValue, FileError>, message: string): TValue {
|
||||
|
|
|
|||
8
packages/agent/src/harness/session/search-index-sink.ts
Normal file
8
packages/agent/src/harness/session/search-index-sink.ts
Normal file
|
|
@ -0,0 +1,8 @@
|
|||
import type { SessionMetadata, SessionSearchIndexRecord, SessionTreeEntry } from "../types.ts";
|
||||
|
||||
export function toSessionSearchIndexRecord<TMetadata extends SessionMetadata>(
|
||||
metadata: TMetadata,
|
||||
entry: SessionTreeEntry,
|
||||
): SessionSearchIndexRecord<TMetadata> {
|
||||
return { metadata, entry };
|
||||
}
|
||||
|
|
@ -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<TMetadata extends SessionMetadata = SessionMetadata> {
|
||||
private storage: SessionStorage<TMetadata>;
|
||||
private contextBuildOptions: SessionContextBuildOptions;
|
||||
private searchIndexSink: SessionSearchIndexSink | undefined;
|
||||
|
||||
constructor(storage: SessionStorage<TMetadata>, contextBuildOptions: SessionContextBuildOptions = {}) {
|
||||
constructor(
|
||||
storage: SessionStorage<TMetadata>,
|
||||
contextBuildOptions: SessionContextBuildOptions = {},
|
||||
searchIndexSink?: SessionSearchIndexSink,
|
||||
) {
|
||||
this.storage = storage;
|
||||
this.contextBuildOptions = contextBuildOptions;
|
||||
this.searchIndexSink = searchIndexSink;
|
||||
}
|
||||
|
||||
getMetadata(): Promise<TMetadata> {
|
||||
|
|
@ -211,8 +219,15 @@ export class Session<TMetadata extends SessionMetadata = SessionMetadata> {
|
|||
return this.storage.getSessionName();
|
||||
}
|
||||
|
||||
private async writeSearchIndexRecord(entry: SessionTreeEntry): Promise<void> {
|
||||
if (this.searchIndexSink) {
|
||||
await this.searchIndexSink.write(toSessionSearchIndexRecord(await this.storage.getMetadata(), entry));
|
||||
}
|
||||
}
|
||||
|
||||
private async appendTypedEntry<TEntry extends SessionTreeEntry>(entry: TEntry): Promise<string> {
|
||||
await this.storage.appendEntry(entry);
|
||||
await this.writeSearchIndexRecord(entry);
|
||||
return entry.id;
|
||||
}
|
||||
|
||||
|
|
@ -342,7 +357,8 @@ export class Session<TMetadata extends SessionMetadata = SessionMetadata> {
|
|||
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",
|
||||
|
|
|
|||
|
|
@ -499,7 +499,7 @@ export interface SessionStorage<TMetadata extends SessionMetadata = SessionMetad
|
|||
getMetadata(): Promise<TMetadata>;
|
||||
getLeafId(): Promise<string | null>;
|
||||
/** Persist a leaf entry that records the active session-tree leaf. */
|
||||
setLeafId(leafId: string | null): Promise<void>;
|
||||
setLeafId(leafId: string | null): Promise<LeafEntry>;
|
||||
createEntryId(): Promise<string>;
|
||||
appendEntry(entry: SessionTreeEntry): Promise<void>;
|
||||
getEntry(id: string): Promise<SessionTreeEntry | undefined>;
|
||||
|
|
@ -519,6 +519,30 @@ export interface SessionCreateOptions {
|
|||
id?: string;
|
||||
}
|
||||
|
||||
export interface SessionSearchOptions {
|
||||
text: string;
|
||||
cwd?: string;
|
||||
}
|
||||
|
||||
export interface SessionSearchHit<TMetadata extends SessionMetadata = SessionMetadata> {
|
||||
metadata: TMetadata;
|
||||
entryId: string;
|
||||
timestamp: string;
|
||||
snippet?: string;
|
||||
score?: number;
|
||||
}
|
||||
|
||||
/** A persisted session entry delivered to a search index. */
|
||||
export interface SessionSearchIndexRecord<TMetadata extends SessionMetadata = SessionMetadata> {
|
||||
metadata: TMetadata;
|
||||
entry: SessionTreeEntry;
|
||||
}
|
||||
|
||||
/** Receives entries after their session storage has persisted them for search indexing. */
|
||||
export interface SessionSearchIndexSink {
|
||||
write(record: SessionSearchIndexRecord): Promise<void>;
|
||||
}
|
||||
|
||||
export interface SessionForkOptions {
|
||||
entryId?: string;
|
||||
position?: "before" | "at";
|
||||
|
|
@ -533,6 +557,7 @@ export interface SessionRepo<
|
|||
create(options: TCreateOptions): Promise<Session<TMetadata>>;
|
||||
open(metadata: TMetadata): Promise<Session<TMetadata>>;
|
||||
list(options?: TListOptions): Promise<TMetadata[]>;
|
||||
search(options: SessionSearchOptions): Promise<SessionSearchHit<TMetadata>[]>;
|
||||
delete(metadata: TMetadata): Promise<void>;
|
||||
fork(source: TMetadata, options: SessionForkOptions & TCreateOptions): Promise<Session<TMetadata>>;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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";
|
||||
|
|
|
|||
|
|
@ -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 });
|
||||
|
|
|
|||
|
|
@ -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");
|
||||
|
|
|
|||
|
|
@ -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([]);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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 },
|
||||
],
|
||||
|
|
|
|||
|
|
@ -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");
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
||||
|
|
|
|||
|
|
@ -19,6 +19,11 @@ export async function loadMigrations(): Promise<SqliteMigration[]> {
|
|||
order: 1,
|
||||
sql: await loadMigrationSql("./migrations/001_initial.sql"),
|
||||
},
|
||||
{
|
||||
id: "002_session_search.sql",
|
||||
order: 2,
|
||||
sql: await loadMigrationSql("./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'
|
||||
);
|
||||
|
|
@ -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<SessionSearchHit<SqliteSessionMetadata>[]> {
|
||||
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<void> {
|
||||
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<SessionEntryRow & { rowid: number }>(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;
|
||||
|
|
|
|||
65
packages/storage/sqlite-node/src/sqlite/search-index-sink.ts
Normal file
65
packages/storage/sqlite-node/src/sqlite/search-index-sink.ts
Normal file
|
|
@ -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<void> {
|
||||
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<void> {
|
||||
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<SessionSearchHit<SqliteSessionMetadata>[]> {
|
||||
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<SessionRow & { entry_id: string; timestamp: string; score: number }>(
|
||||
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,
|
||||
}));
|
||||
}
|
||||
}
|
||||
|
|
@ -263,7 +263,7 @@ export class SqliteSessionStorage implements SessionStorage<SqliteSessionMetadat
|
|||
return this.currentLeafId;
|
||||
}
|
||||
|
||||
async setLeafId(leafId: string | null): Promise<void> {
|
||||
async setLeafId(leafId: string | null): Promise<LeafEntry> {
|
||||
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<SqliteSessionMetadat
|
|||
targetId: leafId,
|
||||
};
|
||||
await this.appendEntry(entry);
|
||||
return entry;
|
||||
}
|
||||
|
||||
async createEntryId(): Promise<string> {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue