refactor(session): nuke dead sessionParams / sessionDisplayId pipeline (v43)
These columns and the messages that plumbed them existed for the CLI-
adapter era — Gemini's resume id and provider-specific session params
rode the run-finished envelope back onto the session row so the next
turn could resume. Pi Mono dispatches every provider through pi-ai and
emits neither value. The worker has been posting literal null on every
finish, finalizeRun has been writing null into task_sessions, and
rotateSession has been carrying null forward.
Remove the wire, the writer, the carry-forward, and the columns:
- v43 migration rebuilds task_sessions without session_params_json /
session_display_id, and runs without session_display_id. Follows
the idempotent "create new, copy, swap, reindex" pattern used by
v41. context_snapshot_json is preserved (also unreferenced today
but outside this cleanup's scope).
- RunJob / EnqueueRunPayload / ChildMessage / worker-types zod
schema no longer carry the fields. finishRun() / createNewSession()
signatures tighten in lockstep.
- rotateSession starts the new session with fresh counters and no
carry-forward.
- turn-context resolveSession returns just the session id.
- TaskSessionRecord / RunRecord lose the dead fields.
498/498 unit tests pass. typecheck + build clean.
This commit is contained in:
1 parent
88a1f8b334
commit
e0e43d6127
32 files changed
+167
-205
No files matched your search
@@ -0,0 +1,143 @@
|
||||
import log from "electron-log/main.js";
|
||||
import type { Migration, MigrationDeps } from "./types";
|
||||
import { columnExists, tableExists } from "./types";
|
||||
|
||||
/**
|
||||
* Drop the Pi Mono-dead session carry-forward columns.
|
||||
*
|
||||
* The `task_sessions.session_params_json` / `task_sessions.session_display_id`
|
||||
* columns and the matching `runs.session_display_id` column were meaningful
|
||||
* in the CLI-adapter era: Gemini's resume id and provider-specific session
|
||||
* params rode the run-finished message back into the session row so the next
|
||||
* turn could resume. Pi Mono dispatches every provider through `pi-ai` and
|
||||
* emits neither value — the worker has been writing literal `null` on every
|
||||
* finish and `rotateSession` has been carrying `null` forward.
|
||||
*
|
||||
* Three columns of unused storage plus a hot path that writes them every
|
||||
* run. Drop them.
|
||||
*
|
||||
* The migration mirrors v41's "copy to new table, swap, reindex" shape —
|
||||
* SQLite still has no multi-column `DROP COLUMN`, and the task_sessions
|
||||
* index topology plus the runs secondary indexes need to be restored
|
||||
* identically. `PRAGMA foreign_keys` is disabled across the swap window
|
||||
* so child rows are not cascade-deleted mid-rebuild.
|
||||
*/
|
||||
export function createMigrationsV43V43(deps: MigrationDeps): Migration[] {
|
||||
const { db } = deps;
|
||||
return [
|
||||
{
|
||||
version: 43,
|
||||
name: "drop_dead_pi_mono_session_columns",
|
||||
fn: () => {
|
||||
const sessionsHasDeadCols =
|
||||
tableExists(db, "task_sessions") &&
|
||||
(columnExists(db, "task_sessions", "session_params_json") ||
|
||||
columnExists(db, "task_sessions", "session_display_id"));
|
||||
const runsHasDeadCol =
|
||||
tableExists(db, "runs") && columnExists(db, "runs", "session_display_id");
|
||||
|
||||
if (!sessionsHasDeadCols && !runsHasDeadCol) {
|
||||
log.info("[migration v43] dead columns already absent; skipping");
|
||||
return;
|
||||
}
|
||||
|
||||
db.exec("PRAGMA foreign_keys = OFF");
|
||||
try {
|
||||
if (sessionsHasDeadCols) {
|
||||
db.exec(`
|
||||
CREATE TABLE task_sessions_new (
|
||||
id TEXT PRIMARY KEY,
|
||||
company_id TEXT NOT NULL,
|
||||
agent_id TEXT NOT NULL,
|
||||
adapter_type TEXT NOT NULL,
|
||||
last_run_id TEXT,
|
||||
last_error TEXT,
|
||||
run_count INTEGER NOT NULL DEFAULT 0,
|
||||
total_input_tokens INTEGER NOT NULL DEFAULT 0,
|
||||
total_output_tokens INTEGER NOT NULL DEFAULT 0,
|
||||
rotated_at TEXT,
|
||||
rotation_reason TEXT,
|
||||
handoff_markdown TEXT,
|
||||
created_at TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL
|
||||
);
|
||||
INSERT INTO task_sessions_new (
|
||||
id, company_id, agent_id, adapter_type, last_run_id, last_error,
|
||||
run_count, total_input_tokens, total_output_tokens, rotated_at,
|
||||
rotation_reason, handoff_markdown, created_at, updated_at
|
||||
)
|
||||
SELECT
|
||||
id, company_id, agent_id, adapter_type, last_run_id, last_error,
|
||||
run_count, total_input_tokens, total_output_tokens, rotated_at,
|
||||
rotation_reason, handoff_markdown, created_at, updated_at
|
||||
FROM task_sessions;
|
||||
DROP TABLE task_sessions;
|
||||
ALTER TABLE task_sessions_new RENAME TO task_sessions;
|
||||
CREATE INDEX IF NOT EXISTS idx_task_sessions_company_agent
|
||||
ON task_sessions(company_id, agent_id, updated_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_task_sessions_lookup
|
||||
ON task_sessions(company_id, agent_id, adapter_type);
|
||||
`);
|
||||
log.info("[migration v43] rebuilt task_sessions without session_params_json/session_display_id");
|
||||
}
|
||||
|
||||
if (runsHasDeadCol) {
|
||||
// v16-v25 added `context_snapshot_json`; it is unreferenced by
|
||||
// production code today but the column exists on real databases
|
||||
// and carries no policy against preservation. Keep it so this
|
||||
// migration's blast radius stays exactly one column.
|
||||
db.exec(`
|
||||
CREATE TABLE runs_new (
|
||||
id TEXT PRIMARY KEY,
|
||||
company_id TEXT NOT NULL,
|
||||
task_id TEXT,
|
||||
agent_id TEXT NOT NULL,
|
||||
workspace_id TEXT,
|
||||
connector_id TEXT NOT NULL,
|
||||
status TEXT NOT NULL,
|
||||
summary TEXT NOT NULL,
|
||||
error_message TEXT,
|
||||
exit_code INTEGER,
|
||||
signal TEXT,
|
||||
model TEXT,
|
||||
cost_usd REAL,
|
||||
unattributed_cost INTEGER NOT NULL DEFAULT 0,
|
||||
context_snapshot_json TEXT,
|
||||
actor_type TEXT,
|
||||
actor_id TEXT,
|
||||
started_at TEXT,
|
||||
finished_at TEXT,
|
||||
log_path TEXT,
|
||||
created_at TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL
|
||||
);
|
||||
INSERT INTO runs_new (
|
||||
id, company_id, task_id, agent_id, workspace_id, connector_id,
|
||||
status, summary, error_message, exit_code, signal, model,
|
||||
cost_usd, unattributed_cost, context_snapshot_json, actor_type,
|
||||
actor_id, started_at, finished_at, log_path, created_at, updated_at
|
||||
)
|
||||
SELECT
|
||||
id, company_id, task_id, agent_id, workspace_id, connector_id,
|
||||
status, summary, error_message, exit_code, signal, model,
|
||||
cost_usd, unattributed_cost, context_snapshot_json, actor_type,
|
||||
actor_id, started_at, finished_at, log_path, created_at, updated_at
|
||||
FROM runs;
|
||||
DROP TABLE runs;
|
||||
ALTER TABLE runs_new RENAME TO runs;
|
||||
CREATE INDEX IF NOT EXISTS idx_runs_company ON runs(company_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_runs_task ON runs(task_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_runs_agent ON runs(agent_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_runs_company_status ON runs(company_id, status);
|
||||
CREATE INDEX IF NOT EXISTS idx_runs_company_created_at ON runs(company_id, created_at DESC);
|
||||
CREATE INDEX IF NOT EXISTS idx_runs_agent_created_at ON runs(agent_id, created_at DESC);
|
||||
`);
|
||||
log.info("[migration v43] rebuilt runs without session_display_id");
|
||||
}
|
||||
} finally {
|
||||
db.exec("PRAGMA foreign_keys = ON");
|
||||
}
|
||||
},
|
||||
},
|
||||
];
|
||||
}
|
||||
@@ -24,7 +24,7 @@ export function buildRunMethods(
|
||||
getRun(id: string): RunRecord {
|
||||
const row = getDb()
|
||||
.prepare(
|
||||
"select id, company_id as companyId, task_id as taskId, agent_id as agentId, workspace_id as workspaceId, connector_id as connectorId, status, summary, error_message as errorMessage, exit_code as exitCode, signal, model, session_display_id as sessionDisplayId, cost_usd as costUsd, unattributed_cost as unattributedCost, started_at as startedAt, finished_at as finishedAt, log_path as logPath, created_at as createdAt, updated_at as updatedAt from runs where id = ?",
|
||||
"select id, company_id as companyId, task_id as taskId, agent_id as agentId, workspace_id as workspaceId, connector_id as connectorId, status, summary, error_message as errorMessage, exit_code as exitCode, signal, model, cost_usd as costUsd, unattributed_cost as unattributedCost, started_at as startedAt, finished_at as finishedAt, log_path as logPath, created_at as createdAt, updated_at as updatedAt from runs where id = ?",
|
||||
)
|
||||
.get(id) as RunRecord | undefined;
|
||||
if (!row) throw new Error(`Run not found: ${id}`);
|
||||
@@ -93,7 +93,6 @@ export function buildRunMethods(
|
||||
exitCode: number | null;
|
||||
signal: string | null;
|
||||
model: string | null;
|
||||
sessionDisplayId: string | null;
|
||||
unattributedCost: boolean;
|
||||
}) {
|
||||
const run = getDb().prepare("select task_id as taskId, agent_id as agentId, company_id as companyId, connector_id as connectorId, started_at as startedAt from runs where id = ?").get(input.runId) as
|
||||
@@ -134,11 +133,11 @@ export function buildRunMethods(
|
||||
transaction(() => {
|
||||
getDb()
|
||||
.prepare(
|
||||
"update runs set status = ?, summary = ?, error_message = ?, exit_code = ?, signal = ?, model = ?, session_display_id = ?, cost_usd = ?, unattributed_cost = ?, finished_at = ?, updated_at = ? where id = ?",
|
||||
"update runs set status = ?, summary = ?, error_message = ?, exit_code = ?, signal = ?, model = ?, cost_usd = ?, unattributed_cost = ?, finished_at = ?, updated_at = ? where id = ?",
|
||||
)
|
||||
.run(
|
||||
input.status, input.summary ?? "", input.errorMessage, input.exitCode,
|
||||
input.signal, input.model, input.sessionDisplayId, runCostUsd,
|
||||
input.signal, input.model, runCostUsd,
|
||||
input.unattributedCost ? 1 : 0, ts, ts, input.runId,
|
||||
);
|
||||
if (run.taskId !== null) {
|
||||
@@ -201,7 +200,7 @@ export function buildRunMethods(
|
||||
const ts = nowIso();
|
||||
getDb()
|
||||
.prepare(
|
||||
"update runs set status = 'cancelled', summary = ?, error_message = null, exit_code = null, signal = null, model = null, session_display_id = null, cost_usd = null, unattributed_cost = 1, finished_at = ?, updated_at = ? where id = ?",
|
||||
"update runs set status = 'cancelled', summary = ?, error_message = null, exit_code = null, signal = null, model = null, cost_usd = null, unattributed_cost = 1, finished_at = ?, updated_at = ? where id = ?",
|
||||
)
|
||||
.run(summary, ts, ts, runId);
|
||||
if (taskId !== null) {
|
||||
@@ -242,7 +241,7 @@ export function buildRunMethods(
|
||||
return readAll<RunRecord>(
|
||||
`select id, company_id as companyId, task_id as taskId, agent_id as agentId, workspace_id as workspaceId,
|
||||
connector_id as connectorId, status, summary, error_message as errorMessage, exit_code as exitCode,
|
||||
signal, model, session_display_id as sessionDisplayId, cost_usd as costUsd,
|
||||
signal, model, cost_usd as costUsd,
|
||||
unattributed_cost as unattributedCost, started_at as startedAt, finished_at as finishedAt,
|
||||
log_path as logPath, created_at as createdAt, updated_at as updatedAt
|
||||
from runs
|
||||
|
||||
@@ -7,6 +7,7 @@ import { createMigrationsV38V39 } from "./migrations/v38-v39";
|
||||
import { createMigrationsV40V40 } from "./migrations/v40-v40";
|
||||
import { createMigrationsV41V41 } from "./migrations/v41-v41";
|
||||
import { createMigrationsV42V42 } from "./migrations/v42-v42";
|
||||
import { createMigrationsV43V43 } from "./migrations/v43-v43";
|
||||
import {
|
||||
columnExists,
|
||||
nowIso,
|
||||
@@ -180,6 +181,7 @@ export function runMigrations(
|
||||
...createMigrationsV40V40(deps),
|
||||
...createMigrationsV41V41(deps),
|
||||
...createMigrationsV42V42(deps),
|
||||
...createMigrationsV43V43(deps),
|
||||
];
|
||||
|
||||
for (const migration of migrations) {
|
||||
|
||||
@@ -83,7 +83,7 @@ export function buildSnapshotMethods(
|
||||
...companyArgs,
|
||||
),
|
||||
runs: readAll<RunRecord>(
|
||||
`select id, company_id as companyId, task_id as taskId, agent_id as agentId, workspace_id as workspaceId, connector_id as connectorId, status, summary, error_message as errorMessage, exit_code as exitCode, signal, model, session_display_id as sessionDisplayId, cost_usd as costUsd, unattributed_cost as unattributedCost, started_at as startedAt, finished_at as finishedAt, log_path as logPath, created_at as createdAt, updated_at as updatedAt from runs${companyWhere} order by created_at desc limit ${companyId ? 2000 : 2000}`,
|
||||
`select id, company_id as companyId, task_id as taskId, agent_id as agentId, workspace_id as workspaceId, connector_id as connectorId, status, summary, error_message as errorMessage, exit_code as exitCode, signal, model, cost_usd as costUsd, unattributed_cost as unattributedCost, started_at as startedAt, finished_at as finishedAt, log_path as logPath, created_at as createdAt, updated_at as updatedAt from runs${companyWhere} order by created_at desc limit ${companyId ? 2000 : 2000}`,
|
||||
...companyArgs,
|
||||
).map((record) => ({ ...record, unattributedCost: Boolean(record.unattributedCost) })) as RunRecord[],
|
||||
costs: readAll<CostRecord>(
|
||||
|
||||
@@ -17,25 +17,6 @@ export function buildTaskSessionMethods(
|
||||
transaction: Transaction,
|
||||
) {
|
||||
return {
|
||||
/**
|
||||
* Stamp pi-ai's transient per-turn state (session resume params,
|
||||
* display id) onto the given agent session row. Called by the
|
||||
* worker's run-finished handler with `event.taskSessionId`.
|
||||
*/
|
||||
updateSessionParams(
|
||||
sessionId: string,
|
||||
sessionParams: Record<string, unknown> | null,
|
||||
sessionDisplayId: string | null,
|
||||
): void {
|
||||
const ts = nowIso();
|
||||
const sessionParamsJson = sessionParams ? JSON.stringify(sessionParams) : null;
|
||||
getDb()
|
||||
.prepare(
|
||||
"update task_sessions set session_params_json = ?, session_display_id = ?, updated_at = ? where id = ?",
|
||||
)
|
||||
.run(sessionParamsJson, sessionDisplayId, ts, sessionId);
|
||||
},
|
||||
|
||||
/**
|
||||
* Fetch a session by its primary key. This is the single lookup
|
||||
* path — the rotation evaluator, the worker's finalizer, and the
|
||||
@@ -48,8 +29,7 @@ export function buildTaskSessionMethods(
|
||||
(getDb()
|
||||
.prepare(
|
||||
`SELECT id, company_id as companyId, agent_id as agentId,
|
||||
adapter_type as adapterType, session_params_json as sessionParamsJson,
|
||||
session_display_id as sessionDisplayId, last_run_id as lastRunId,
|
||||
adapter_type as adapterType, last_run_id as lastRunId,
|
||||
last_error as lastError, run_count as runCount,
|
||||
total_input_tokens as totalInputTokens, total_output_tokens as totalOutputTokens,
|
||||
rotated_at as rotatedAt, rotation_reason as rotationReason,
|
||||
@@ -111,9 +91,8 @@ export function buildTaskSessionMethods(
|
||||
getDb()
|
||||
.prepare(
|
||||
`INSERT INTO task_sessions (id, company_id, agent_id, adapter_type,
|
||||
session_params_json, session_display_id, run_count, total_input_tokens,
|
||||
total_output_tokens, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, NULL, NULL, 0, 0, 0, ?, ?)`,
|
||||
run_count, total_input_tokens, total_output_tokens, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, 0, 0, 0, ?, ?)`,
|
||||
)
|
||||
.run(sessionId, input.companyId, input.agentId, input.adapterType, ts, ts);
|
||||
getDb()
|
||||
@@ -132,34 +111,22 @@ export function buildTaskSessionMethods(
|
||||
|
||||
/**
|
||||
* Create a replacement session row after rotation. Inherits the
|
||||
* agent / company / adapter type from the caller; session params
|
||||
* and display id carry forward so the next pi-ai turn can resume.
|
||||
* agent / company / adapter type from the caller. Counters start
|
||||
* at zero — Pi Mono has no per-turn provider state to carry forward.
|
||||
*/
|
||||
createNewSession(input: {
|
||||
companyId: string;
|
||||
agentId: string;
|
||||
adapterType: string;
|
||||
sessionParamsJson: string | null;
|
||||
sessionDisplayId: string | null;
|
||||
}): string {
|
||||
const sessionId = randomUUID();
|
||||
const ts = nowIso();
|
||||
getDb()
|
||||
.prepare(
|
||||
`INSERT INTO task_sessions (id, company_id, agent_id, adapter_type,
|
||||
session_params_json, session_display_id, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
`INSERT INTO task_sessions (id, company_id, agent_id, adapter_type, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?)`,
|
||||
)
|
||||
.run(
|
||||
sessionId,
|
||||
input.companyId,
|
||||
input.agentId,
|
||||
input.adapterType,
|
||||
input.sessionParamsJson,
|
||||
input.sessionDisplayId,
|
||||
ts,
|
||||
ts,
|
||||
);
|
||||
.run(sessionId, input.companyId, input.agentId, input.adapterType, ts, ts);
|
||||
return sessionId;
|
||||
},
|
||||
};
|
||||
|
||||
Reference in new issue
Block a user