sscoble/pi-mono
1
0
Fork 0
mirror of https://github.com/badlogic/pi-mono.git synced 2026-11-05 23:37:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

rename: may not be index

Browse source
This commit is contained in:
Cristina Poncela Cubeiro 2026-07-27 16:12:43 +02:00
parent bd8dc676bf
commit 0ee1ec0030
11 changed files with 86 additions and 93 deletions
Download patch file Download diff file Expand all files Collapse all files

6
packages/agent/src/harness/session/repo-utils.ts
View file

@ -5,7 +5,7 @@ import {
SessionError,
type SessionMetadata,
type SessionSearchHit,
type SessionSearchIndexWriter,
type SessionSearchWriter,
type SessionStorage,
type SessionTreeEntry,
} from "../types.ts";
@ -21,9 +21,9 @@ export function createTimestamp(): string {
export function toSession<TMetadata extends SessionMetadata>(
storage: SessionStorage<TMetadata>,
searchIndexWriter?: SessionSearchIndexWriter<TMetadata>,
searchWriter?: SessionSearchWriter<TMetadata>,
): Session<TMetadata> {
return new Session(storage, {}, searchIndexWriter);
return new Session(storage, {}, searchWriter);
}
export function findSessionEntryMatches<TMetadata extends SessionMetadata>(

25
packages/agent/src/harness/session/search-backend.ts Normal file
View file

@ -0,0 +1,25 @@
import type { SessionMetadata, SessionSearchRecord, SessionSearchWriter, SessionTreeEntry } from "../types.ts";
export function toSessionSearchRecord<TMetadata extends SessionMetadata>(
metadata: TMetadata,
entry: SessionTreeEntry,
): SessionSearchRecord<TMetadata> {
return { metadata, entry };
}
/** Delivers each lifecycle event to its targets, like Go's io.MultiWriter. */
export class FanoutSessionSearchWriter<TMetadata extends SessionMetadata> implements SessionSearchWriter<TMetadata> {
private readonly writers: readonly SessionSearchWriter<TMetadata>[];
constructor(...writers: SessionSearchWriter<TMetadata>[]) {
this.writers = writers;
}
async upsert(record: SessionSearchRecord<TMetadata>): Promise<void> {
for (const writer of this.writers) await writer.upsert(record);
}
async remove(record: SessionSearchRecord<TMetadata>): Promise<void> {
for (const writer of this.writers) await writer.remove(record);
}
}

32
packages/agent/src/harness/session/search-index.ts
View file

@ -1,32 +0,0 @@
import type {
SessionMetadata,
SessionSearchIndexRecord,
SessionSearchIndexWriter,
SessionTreeEntry,
} from "../types.ts";
export function toSessionSearchIndexRecord<TMetadata extends SessionMetadata>(
metadata: TMetadata,
entry: SessionTreeEntry,
): SessionSearchIndexRecord<TMetadata> {
return { metadata, entry };
}
/** Delivers each lifecycle event to its targets, like Go's io.MultiWriter. */
export class MultiSessionSearchIndexWriter<TMetadata extends SessionMetadata>
implements SessionSearchIndexWriter<TMetadata>
{
private readonly writers: readonly SessionSearchIndexWriter<TMetadata>[];
constructor(...writers: SessionSearchIndexWriter<TMetadata>[]) {
this.writers = writers;
}
async write(record: SessionSearchIndexRecord<TMetadata>): Promise<void> {
for (const writer of this.writers) await writer.write(record);
}
async remove(record: SessionSearchIndexRecord<TMetadata>): Promise<void> {
for (const writer of this.writers) await writer.remove(record);
}
}

20
packages/agent/src/harness/session/session.ts
View file

@ -14,14 +14,14 @@ import type {
SessionEntryCursorOptions,
SessionInfoEntry,
SessionMetadata,
SessionSearchIndexWriter,
SessionSearchWriter,
SessionStats,
SessionStorage,
SessionTreeEntry,
ThinkingLevelChangeEntry,
} from "../types.ts";
import { SessionError } from "../types.ts";
import { toSessionSearchIndexRecord } from "./search-index.ts";
import { toSessionSearchRecord } from "./search-backend.ts";
export type ContextEntryTransform = (entries: readonly SessionTreeEntry[]) => readonly SessionTreeEntry[];
@ -152,16 +152,16 @@ export function buildSessionContext(
export class Session<TMetadata extends SessionMetadata = SessionMetadata> {
private storage: SessionStorage<TMetadata>;
private contextBuildOptions: SessionContextBuildOptions;
private searchIndexWriter: SessionSearchIndexWriter<TMetadata> | undefined;
private searchWriter: SessionSearchWriter<TMetadata> | undefined;
constructor(
storage: SessionStorage<TMetadata>,
contextBuildOptions: SessionContextBuildOptions = {},
searchIndexWriter?: SessionSearchIndexWriter<TMetadata>,
searchWriter?: SessionSearchWriter<TMetadata>,
) {
this.storage = storage;
this.contextBuildOptions = contextBuildOptions;
this.searchIndexWriter = searchIndexWriter;
this.searchWriter = searchWriter;
}
getMetadata(): Promise<TMetadata> {
@ -219,15 +219,15 @@ export class Session<TMetadata extends SessionMetadata = SessionMetadata> {
return this.storage.getSessionName();
}
private async writeSearchIndexRecord(entry: SessionTreeEntry): Promise<void> {
if (this.searchIndexWriter) {
await this.searchIndexWriter.write(toSessionSearchIndexRecord(await this.storage.getMetadata(), entry));
private async upsertSearchRecord(entry: SessionTreeEntry): Promise<void> {
if (this.searchWriter) {
await this.searchWriter.upsert(toSessionSearchRecord(await this.storage.getMetadata(), entry));
}
}
private async appendTypedEntry<TEntry extends SessionTreeEntry>(entry: TEntry): Promise<string> {
await this.storage.appendEntry(entry);
await this.writeSearchIndexRecord(entry);
await this.upsertSearchRecord(entry);
return entry.id;
}
@ -358,7 +358,7 @@ export class Session<TMetadata extends SessionMetadata = SessionMetadata> {
throw new SessionError("not_found", `Entry ${entryId} not found`);
}
const leafEntry = await this.storage.setLeafId(entryId);
await this.writeSearchIndexRecord(leafEntry);
await this.upsertSearchRecord(leafEntry);
if (!summary) return undefined;
return this.appendTypedEntry({
type: "branch_summary",

18
packages/agent/src/harness/types.ts
View file

@ -532,21 +532,21 @@ export interface SessionSearchHit<TMetadata extends SessionMetadata = SessionMet
score?: number;
}
/** A persisted session entry delivered to a search index. */
export interface SessionSearchIndexRecord<TMetadata extends SessionMetadata = SessionMetadata> {
/** A persisted session entry delivered to a search backend. */
export interface SessionSearchRecord<TMetadata extends SessionMetadata = SessionMetadata> {
metadata: TMetadata;
entry: SessionTreeEntry;
}
/** Receives persisted-entry lifecycle events for search indexing. */
export interface SessionSearchIndexWriter<TMetadata extends SessionMetadata = SessionMetadata> {
write(record: SessionSearchIndexRecord<TMetadata>): Promise<void>;
remove(record: SessionSearchIndexRecord<TMetadata>): Promise<void>;
/** Receives persisted-entry lifecycle events for a search backend. */
export interface SessionSearchWriter<TMetadata extends SessionMetadata = SessionMetadata> {
upsert(record: SessionSearchRecord<TMetadata>): Promise<void>;
remove(record: SessionSearchRecord<TMetadata>): Promise<void>;
}
/** A search index that can both receive session entries and query them. */
export interface SessionSearchIndex<TMetadata extends SessionMetadata = SessionMetadata>
extends SessionSearchIndexWriter<TMetadata> {
/** A search backend that can receive session entries and query them. */
export interface SessionSearchBackend<TMetadata extends SessionMetadata = SessionMetadata>
extends SessionSearchWriter<TMetadata> {
search(options: SessionSearchOptions): Promise<SessionSearchHit<TMetadata>[]>;
}

2
packages/agent/src/index.ts
View file

@ -34,7 +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.ts";
export * from "./harness/session/search-backend.ts";
export * from "./harness/session/session.ts";
export * from "./harness/skills.ts";
export * from "./harness/system-prompt.ts";

20
packages/agent/test/harness/sqlite-node.test.ts
View file

@ -7,10 +7,10 @@ import {
} from "../../../storage/sqlite-node/src/index.ts";
import { NodeExecutionEnv } from "../../src/harness/env/nodejs.ts";
import type {
SessionSearchBackend,
SessionSearchHit,
SessionSearchIndex,
SessionSearchIndexRecord,
SessionSearchOptions,
SessionSearchRecord,
} from "../../src/harness/types.ts";
import { createTempDir, createUserMessage } from "./session-test-utils.ts";
@ -30,14 +30,14 @@ describe("sqlite-node adapter", () => {
}
});
it("delegates lifecycle events and queries to an injected search index", async () => {
it("delegates lifecycle events and queries to an injected search backend", async () => {
const root = createTempDir();
const writes: SessionSearchIndexRecord<SqliteSessionMetadata>[] = [];
const removals: SessionSearchIndexRecord<SqliteSessionMetadata>[] = [];
const upserts: SessionSearchRecord<SqliteSessionMetadata>[] = [];
const removals: SessionSearchRecord<SqliteSessionMetadata>[] = [];
const searches: SessionSearchOptions[] = [];
const index: SessionSearchIndex<SqliteSessionMetadata> = {
async write(record) {
writes.push(record);
const index: SessionSearchBackend<SqliteSessionMetadata> = {
async upsert(record) {
upserts.push(record);
},
async remove(record) {
removals.push(record);
@ -51,13 +51,13 @@ describe("sqlite-node adapter", () => {
env: new NodeExecutionEnv({ cwd: root }),
sqlite: createNodeSqliteFactory(),
databasePath: join(root, "sessions.sqlite"),
searchIndexFactory: () => index,
searchBackendFactory: () => index,
});
const session = await repo.create({ cwd: root, id: "session-1" });
const metadata = await session.getMetadata();
const entryId = await session.appendMessage(createUserMessage("indexed remotely"));
expect(writes).toEqual([expect.objectContaining({ metadata, entry: expect.objectContaining({ id: entryId }) })]);
expect(upserts).toEqual([expect.objectContaining({ metadata, entry: expect.objectContaining({ id: entryId }) })]);
await expect(repo.search({ text: "indexed remotely" })).resolves.toEqual([]);
expect(searches).toEqual([{ text: "indexed remotely" }]);
await repo.delete(metadata);

2
packages/storage/sqlite-node/src/sqlite/index.ts
View file

@ -1,5 +1,5 @@
export * from "./migrations.ts";
export * from "./repo.ts";
export * from "./search-index.ts";
export * from "./search-backend.ts";
export * from "./storage/index.ts";
export * from "./types.ts";

34
packages/storage/sqlite-node/src/sqlite/repo.ts
View file

@ -11,10 +11,10 @@ import {
getFileSystemResultOrThrow,
SessionError,
toSession,
toSessionSearchIndexRecord,
toSessionSearchRecord,
} from "@earendil-works/pi-agent-core";
import { applyMigrations } from "./migrations.ts";
import { SqliteSessionSearchIndex } from "./search-index.ts";
import { SqliteSessionSearchBackend } from "./search-backend.ts";
import { SqliteSessionStorage } from "./storage/index.ts";
import { decodeEntry, type SessionEntryRow } from "./storage/session-entries.ts";
import { rowToMetadata, type SessionRow } from "./storage/sessions.ts";
@ -26,7 +26,7 @@ import type {
SqliteSessionMetadata,
SqliteSessionRepoApi,
SqliteSessionRepoEnv,
SqliteSessionSearchIndexFactory,
SqliteSessionSearchBackendFactory,
} from "./types.ts";
function getParentPath(path: string): string {
@ -54,20 +54,20 @@ export class SqliteSessionRepo implements SqliteSessionRepoApi {
private readonly env: SqliteSessionRepoEnv;
private readonly sqlite: SqliteDatabaseFactory;
private readonly databasePathInput: string;
private readonly searchIndexFactory: SqliteSessionSearchIndexFactory;
private readonly searchBackendFactory: SqliteSessionSearchBackendFactory;
private databasePath: string | undefined;
constructor(options: {
env: SqliteSessionRepoEnv;
sqlite: SqliteDatabaseFactory;
databasePath: string;
searchIndexFactory?: SqliteSessionSearchIndexFactory;
searchBackendFactory?: SqliteSessionSearchBackendFactory;
}) {
this.env = options.env;
this.sqlite = options.sqlite;
this.databasePathInput = options.databasePath;
this.searchIndexFactory =
options.searchIndexFactory ?? (({ db, databasePath }) => new SqliteSessionSearchIndex(db, databasePath));
this.searchBackendFactory =
options.searchBackendFactory ?? (({ db, databasePath }) => new SqliteSessionSearchBackend(db, databasePath));
}
private async getDatabasePath(): Promise<string> {
@ -80,8 +80,8 @@ export class SqliteSessionRepo implements SqliteSessionRepoApi {
return this.databasePath;
}
private async createSearchIndex(db: SqliteDatabase) {
return this.searchIndexFactory({ db, databasePath: await this.getDatabasePath() });
private async createSearchBackend(db: SqliteDatabase) {
return this.searchBackendFactory({ db, databasePath: await this.getDatabasePath() });
}
private async ensureDatabaseDir(): Promise<void> {
@ -116,7 +116,7 @@ export class SqliteSessionRepo implements SqliteSessionRepoApi {
parentSessionId: options.parentSessionId,
metadata: options.metadata,
});
return toSession(storage, await this.createSearchIndex(db));
return toSession(storage, await this.createSearchBackend(db));
} catch (error) {
await db.close();
throw error;
@ -132,7 +132,7 @@ export class SqliteSessionRepo implements SqliteSessionRepoApi {
const db = await this.openDatabase();
try {
const storage = await SqliteSessionStorage.open(db, metadata);
return toSession(storage, await this.createSearchIndex(db));
return toSession(storage, await this.createSearchBackend(db));
} catch (error) {
await db.close();
throw error;
@ -168,7 +168,7 @@ export class SqliteSessionRepo implements SqliteSessionRepoApi {
if (!getFileSystemResultOrThrow(await this.env.exists(path), `Failed to check database ${path}`)) return [];
const db = await this.openDatabase();
try {
return await (await this.createSearchIndex(db)).search(options);
return await (await this.createSearchBackend(db)).search(options);
} finally {
await db.close();
}
@ -178,14 +178,14 @@ export class SqliteSessionRepo implements SqliteSessionRepoApi {
const db = await this.openDatabase();
try {
await db.transaction(async () => {
const searchIndex = await this.createSearchIndex(db);
const searchBackend = await this.createSearchBackend(db);
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 searchIndex.remove(toSessionSearchIndexRecord(metadata, decodeEntry(row)));
await searchBackend.remove(toSessionSearchRecord(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);
@ -222,13 +222,13 @@ export class SqliteSessionRepo implements SqliteSessionRepoApi {
parentSessionId: options.parentSessionId ?? sourceMetadata.id,
metadata: options.metadata ?? sourceMetadata.metadata,
});
const searchIndex = await this.createSearchIndex(db);
const searchBackend = await this.createSearchBackend(db);
const metadata = await storage.getMetadata();
for (const entry of forkedEntries) {
await storage.appendEntry(entry);
await searchIndex.write(toSessionSearchIndexRecord(metadata, entry));
await searchBackend.upsert(toSessionSearchRecord(metadata, entry));
}
return toSession(storage, searchIndex);
return toSession(storage, searchBackend);
} catch (error) {
await db.close();
throw error;

14
packages/storage/sqlite-node/src/sqlite/search-index.ts → packages/storage/sqlite-node/src/sqlite/search-backend.ts
View file

@ -1,15 +1,15 @@
import type {
SessionSearchBackend,
SessionSearchHit,
SessionSearchIndex,
SessionSearchIndexRecord,
SessionSearchOptions,
SessionSearchRecord,
} 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 for persisted session entries. */
export class SqliteSessionSearchIndex implements SessionSearchIndex<SqliteSessionMetadata> {
/** SQLite-backed search backend for persisted session entries. */
export class SqliteSessionSearchBackend implements SessionSearchBackend<SqliteSessionMetadata> {
private readonly db: SqliteDatabase;
private readonly databasePath: string;
@ -18,19 +18,19 @@ export class SqliteSessionSearchIndex implements SessionSearchIndex<SqliteSessio
this.databasePath = databasePath;
}
async write(record: SessionSearchIndexRecord<SqliteSessionMetadata>): Promise<void> {
async upsert(record: SessionSearchRecord<SqliteSessionMetadata>): 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`);
throw new SessionError("not_found", `Entry ${record.entry.id} not found for search persistence`);
}
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<SqliteSessionMetadata>): Promise<void> {
async remove(record: SessionSearchRecord<SqliteSessionMetadata>): 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);

6
packages/storage/sqlite-node/src/sqlite/types.ts
View file

@ -3,7 +3,7 @@ import type {
SessionCreateOptions,
SessionMetadata,
SessionRepo,
SessionSearchIndex,
SessionSearchBackend,
} from "@earendil-works/pi-agent-core";
/** Result of a prepared SQLite statement execution. */
@ -60,7 +60,7 @@ export interface SqliteSessionRepoApi
export type SqliteSessionRepoEnv = Pick<FileSystem, "absolutePath" | "createDir" | "exists">;
export type SqliteSessionSearchIndexFactory = (options: {
export type SqliteSessionSearchBackendFactory = (options: {
db: SqliteDatabase;
databasePath: string;
}) => SessionSearchIndex<SqliteSessionMetadata>;
}) => SessionSearchBackend<SqliteSessionMetadata>;