diff --git a/src/main/agent-runtime/sqlite-session-store.ts b/src/main/agent-runtime/sqlite-session-store.ts index 0202daea..7905532e 100644 --- a/src/main/agent-runtime/sqlite-session-store.ts +++ b/src/main/agent-runtime/sqlite-session-store.ts @@ -4,9 +4,12 @@ import type { AppDatabase } from "@main/database"; /** * SQLite-backed pi `AgentMessage[]` store. * - * One store instance per task_session. Owners must call `appendMessage` - * exactly once per pi `message_end` event, and `replaceMessages` after pi - * `compact` returns a compacted message list. + * One store instance per task_session. `appendMessage` runs exactly + * once per pi `message_end` event; `loadMessages` reads the live + * transcript back at `prepareRun`. Compaction happens at LLM-call + * time via pi's `transformContext` hook and does not rewrite the + * persisted transcript — the session shadow is the durable truth, + * the compacted payload is a per-call view. */ export class SqliteSessionStore { constructor( @@ -33,28 +36,4 @@ export class SqliteSessionStore { payloadJson: JSON.stringify(message), }); } - - appendCompaction(summary: string): void { - this.db.sessionEntries.append({ - taskSessionId: this.taskSessionId, - kind: "compaction", - role: null, - payloadJson: JSON.stringify({ summary }), - }); - } - - replaceMessages(messages: AgentMessage[], compactionSummary?: string): void { - const entries: Array<{ kind: "message" | "compaction"; role: string | null; payloadJson: string }> = []; - if (compactionSummary) { - entries.push({ kind: "compaction", role: null, payloadJson: JSON.stringify({ summary: compactionSummary }) }); - } - for (const message of messages) { - entries.push({ - kind: "message", - role: (message as { role?: string }).role ?? null, - payloadJson: JSON.stringify(message), - }); - } - this.db.sessionEntries.replace({ taskSessionId: this.taskSessionId, entries }); - } } diff --git a/src/main/database-base.ts b/src/main/database-base.ts index 7c540697..54f670a2 100644 --- a/src/main/database-base.ts +++ b/src/main/database-base.ts @@ -252,7 +252,6 @@ export class AppDatabaseBase { setHeartbeatPolicy(agentId: string, companyId: string, policy: { enabled?: boolean; intervalSec?: number; cooldownSec?: number; maxConcurrentRuns?: number; wakeOnDemand?: boolean }) { return this.agents.setHeartbeatPolicy(agentId, companyId, policy); } countActiveRunsForAgent(agentId: string, companyId: string) { return this.agents.countActiveRunsForAgent(agentId, companyId); } listDepartmentAgentIds(companyId: string, department: string) { return this.agents.listDepartmentAgentIds(companyId, department); } - listTopLevelAgentIds(companyId: string) { return this.agents.listTopLevelAgentIds(companyId); } listCompanyBroadcastAgentIds(companyId: string) { return this.agents.listCompanyBroadcastAgentIds(companyId); } listIncidentPingableAgentIds(companyId: string) { return this.agents.listIncidentPingableAgentIds(companyId); } findAgentName(agentId: string) { return this.agents.findAgentName(agentId); } diff --git a/src/main/database/agents-hierarchy.ts b/src/main/database/agents-hierarchy.ts index 2cf5dc13..2db09557 100644 --- a/src/main/database/agents-hierarchy.ts +++ b/src/main/database/agents-hierarchy.ts @@ -284,18 +284,6 @@ export function buildAgentHierarchyMethods( ).map((row) => row.id); }, - /** - * Returns ids of non-terminated agents that have no manager (top of the - * reporting tree). Used for board-only fan-outs where only top-level - * leadership should be notified. - */ - listTopLevelAgentIds(companyId: string): string[] { - return readAll<{ id: string }>( - "select id from agents where company_id = ? and reports_to is null and status != 'terminated'", - companyId, - ).map((row) => row.id); - }, - /** * Returns ids of every agent that is eligible to receive a routine * company-wide broadcast. Paused, pending-approval, and terminated diff --git a/src/main/database/session-entries.ts b/src/main/database/session-entries.ts index 40b5415d..85728bbe 100644 --- a/src/main/database/session-entries.ts +++ b/src/main/database/session-entries.ts @@ -3,7 +3,15 @@ import type { DatabaseSync } from "node:sqlite"; import type { ReadAll, Transaction } from "./helpers"; import { nowIso } from "./helpers"; -export type SessionEntryKind = "message" | "compaction" | "summary"; +/** + * The `session_entries` table only ever stores one kind — full pi + * `AgentMessage` envelopes emitted by `message_end`. Earlier builds + * reserved `"compaction"` and `"summary"` variants for a supersede + * pipeline that never shipped; their writers are gone and the column + * holds `"message"` for every live row. Narrowing the type prevents + * new code from writing a dead kind by accident. + */ +export type SessionEntryKind = "message"; export interface SessionEntryRow { id: string; @@ -12,7 +20,6 @@ export interface SessionEntryRow { kind: SessionEntryKind; role: string | null; payloadJson: string; - isSuperseded: boolean; createdAt: string; } @@ -23,15 +30,10 @@ export interface AppendInput { payloadJson: string; } -export interface ReplaceInput { - taskSessionId: string; - entries: Array<{ kind: SessionEntryKind; role?: string | null; payloadJson: string }>; -} - export function buildSessionEntryMethods( getDb: () => DatabaseSync, readAll: ReadAll, - transaction: Transaction, + _transaction: Transaction, ) { function nextSequence(taskSessionId: string): number { const rows = readAll<{ maxSeq: number | null }>( @@ -66,7 +68,6 @@ export function buildSessionEntryMethods( kind: input.kind, role: input.role ?? null, payloadJson: input.payloadJson, - isSuperseded: false, createdAt: created, }; } @@ -79,10 +80,9 @@ export function buildSessionEntryMethods( kind: SessionEntryKind; role: string | null; payload_json: string; - is_superseded: number; created_at: string; }>( - "select id, task_session_id, sequence, kind, role, payload_json, is_superseded, created_at from session_entries where task_session_id = ? and is_superseded = 0 order by created_at asc, sequence asc", + "select id, task_session_id, sequence, kind, role, payload_json, created_at from session_entries where task_session_id = ? and is_superseded = 0 order by created_at asc, sequence asc", taskSessionId, ); return rows.map((row) => ({ @@ -92,62 +92,9 @@ export function buildSessionEntryMethods( kind: row.kind, role: row.role, payloadJson: row.payload_json, - isSuperseded: false, createdAt: row.created_at, })); } - function listAll(taskSessionId: string): SessionEntryRow[] { - const rows = readAll<{ - id: string; - task_session_id: string; - sequence: number; - kind: SessionEntryKind; - role: string | null; - payload_json: string; - is_superseded: number; - created_at: string; - }>( - "select id, task_session_id, sequence, kind, role, payload_json, is_superseded, created_at from session_entries where task_session_id = ? order by sequence asc", - taskSessionId, - ); - return rows.map((row) => ({ - id: row.id, - taskSessionId: row.task_session_id, - sequence: row.sequence, - kind: row.kind, - role: row.role, - payloadJson: row.payload_json, - isSuperseded: row.is_superseded === 1, - createdAt: row.created_at, - })); - } - - function replace(input: ReplaceInput): SessionEntryRow[] { - return transaction(() => { - getDb() - .prepare("update session_entries set is_superseded = 1 where task_session_id = ? and is_superseded = 0") - .run(input.taskSessionId); - const inserted: SessionEntryRow[] = []; - for (const entry of input.entries) { - inserted.push( - append({ - taskSessionId: input.taskSessionId, - kind: entry.kind, - role: entry.role ?? null, - payloadJson: entry.payloadJson, - }), - ); - } - return inserted; - }); - } - - function deleteForSession(taskSessionId: string): void { - getDb() - .prepare("delete from session_entries where task_session_id = ?") - .run(taskSessionId); - } - - return { append, listActive, listAll, replace, deleteForSession }; + return { append, listActive }; } diff --git a/tests/integration/pi-mono-migrations.spec.ts b/tests/integration/pi-mono-migrations.spec.ts index bb40ae73..a49c2457 100644 --- a/tests/integration/pi-mono-migrations.spec.ts +++ b/tests/integration/pi-mono-migrations.spec.ts @@ -101,38 +101,6 @@ describe("pi-mono migrations 27 + 28", () => { dbB.close(); }); - it("session_entries respects the unique (task_session_id, sequence) index across replace cycles", async () => { - const dir = await freshProfile("agentcompany-migration-replace-"); - const db = new AppDatabase(dir); - db.init(); - - for (let i = 0; i < 3; i += 1) { - db.sessionEntries.append({ - taskSessionId: "replace-test", - kind: "message", - role: "user", - payloadJson: JSON.stringify({ role: "user", content: `t${i}` }), - }); - } - db.sessionEntries.replace({ - taskSessionId: "replace-test", - entries: [ - { kind: "compaction", role: null, payloadJson: JSON.stringify({ summary: "summarised" }) }, - { kind: "message", role: "user", payloadJson: JSON.stringify({ role: "user", content: "post" }) }, - ], - }); - - const active = db.sessionEntries.listActive("replace-test"); - expect(active).toHaveLength(2); - expect(active[0]?.kind).toBe("compaction"); - expect(active[1]?.kind).toBe("message"); - - const all = db.sessionEntries.listAll("replace-test"); - expect(all.length).toBe(5); - expect(all.filter((row) => row.isSuperseded).length).toBe(3); - db.close(); - }); - it("seedConnectors prunes legacy connector rows and inserts the full pi-mono + Fireworks catalog", async () => { const dir = await freshProfile("agentcompany-migration-seed-"); const db = new AppDatabase(dir); diff --git a/tests/unit/sqlite-session-store.test.ts b/tests/unit/sqlite-session-store.test.ts index 93f3b9b8..4ad7568c 100644 --- a/tests/unit/sqlite-session-store.test.ts +++ b/tests/unit/sqlite-session-store.test.ts @@ -8,7 +8,7 @@ import { SqliteSessionStore } from "../../src/main/agent-runtime/sqlite-session- /** * Verifies the SqliteSessionStore round-trips pi `AgentMessage` values - * correctly through the new `session_entries` table (migration 27). + * correctly through the `session_entries` table. */ describe("SqliteSessionStore", () => { let tempDir: string; @@ -49,32 +49,4 @@ describe("SqliteSessionStore", () => { expect((loaded[0] as { role: string }).role).toBe("user"); expect((loaded[1] as { role: string }).role).toBe("assistant"); }); - - it("appends a compaction marker that does not appear as a message", () => { - const store = new SqliteSessionStore(db, TASK_SESSION_ID); - store.appendMessage({ role: "user", content: [{ type: "text", text: "first" }], timestamp: Date.now() } as AgentMessage); - store.appendCompaction("Earlier conversation summarised: planning phase complete."); - const loaded = store.loadMessages(); - expect(loaded).toHaveLength(1); - }); - - it("replaceMessages supersedes prior rows and inserts new ones in order", () => { - const store = new SqliteSessionStore(db, TASK_SESSION_ID); - store.appendMessage({ role: "user", content: [{ type: "text", text: "old-1" }], timestamp: Date.now() } as AgentMessage); - store.appendMessage({ role: "assistant", content: [{ type: "text", text: "old-2" }], timestamp: Date.now() } as AgentMessage); - - const compacted: AgentMessage[] = [ - { role: "user", content: [{ type: "text", text: "context summary" }], timestamp: Date.now() } as AgentMessage, - ]; - store.replaceMessages(compacted, "Summary of prior turns."); - - const active = store.loadMessages(); - expect(active).toHaveLength(1); - expect((active[0] as { content: { text: string }[] }).content[0]?.text).toBe("context summary"); - - // Check: pre-replace messages must still exist as superseded audit rows. - const all = db.sessionEntries.listAll(TASK_SESSION_ID); - const supersededCount = all.filter((row) => row.isSuperseded).length; - expect(supersededCount).toBe(2); - }); });