refactor: prune session-store and agent-list dead code

The session_entries supersede pipeline was a design that never
shipped: `SqliteSessionStore.replaceMessages` / `appendCompaction`,
`sessionEntries.replace` / `listAll` / `deleteForSession`, and the
`"compaction" | "summary"` kind variants had zero live callers.
Compaction wires through pi's `transformContext` and leaves the
persisted transcript alone, so the shadow-row machinery was pure
cruft with two tests propping it up.

Rip it out: the store exposes only `loadMessages` and
`appendMessage`, `sessionEntries` exposes only `append` and
`listActive`, and `SessionEntryKind` narrows to `"message"` so new
writers can't re-introduce a dead kind by accident.
`is_superseded` stays on the table — the DB column is harmless
with no writer, and dropping it is a separate schema migration
concern.

`listTopLevelAgentIds` loses its only caller in the previous
commit (company channel now fans out through
`listCompanyBroadcastAgentIds`), so it goes too. Three tests that
validated the removed behaviour are deleted; the one remaining
session-store test still covers the live `append + load` path.
This commit is contained in:
ZacharyZhang-NY committed 2026-04-19 01:34:14 -04:00
1 parent 5097591725
commit 5e6e929779
6 files changed
+19 -166

No files matched your search

+6 -27
View File
@@ -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 });
}
}
-1
View File
@@ -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); }
-12
View File
@@ -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
+12 -65
View File
@@ -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 };
}
@@ -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);
+1 -29
View File
@@ -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);
});
});