refactor(session): drop dead task-scoped session columns (v41)
Sessions have been agent-scoped since v34, but the schema kept four
dead task-scoped columns that every session code path had to pretend
were meaningful:
- tasks.active_session_id (mirrored agents.session_id)
- tasks.session_json (superseded by task_sessions.session_params_json)
- tasks.session_display_id (superseded by task_sessions.session_display_id)
- task_sessions.task_id (always empty-string on new rows)
v41 drops all four via SQLite table rebuild — the only safe path when
multiple columns need to go in one step and indexes / FK directions
need to be preserved. `PRAGMA foreign_keys = OFF` spans the swap
window so cascade-delete does not fire mid-rebuild. The migration is
idempotent and also backfills the optional task columns
(task_type / requires_user_review / retry_count / last_failure_at /
started_at / completed_at / cancelled_at / checkout_run_id /
execution_locked_at) before rebuilding so it works whether or not
`ensureOptionalColumns` has run yet on this install.
Along with the schema change, every code path that read or wrote
those columns has been rewired to the agent-scoped equivalents:
- task-sessions-crud.ts: deleted updateTaskSession,
getTaskSession, getActiveTaskSession, updateTasksActiveSessionForAgent;
added updateSessionParams(sessionId, ...) for the worker's
finalizer. ensureAgentSession and createNewSession no longer touch
task_id.
- database-base.ts: facade drops the now-unused pass-throughs.
- snapshot.ts / task-queries.ts / tasks-status.ts / entity-crud.ts:
SELECT projections no longer pull session_json / session_display_id
/ active_session_id from tasks.
- tasks-crud.ts (TaskRow) and shared/types/records-core.ts
(TaskRecord): sessionJson / sessionDisplayId fields removed.
- TaskSessionRecord: taskId field removed.
- turn-context.ts::resolveSession: no more
updateTasksActiveSessionForAgent mirror; sessionParams read directly
from the current session via getSessionById.
- run-finished-helpers.ts::finalizeRun: stamps sessionParams via
updateSessionParams on the agent's current session, resolved through
agents.session_id.
- session-management-service.ts: rotateSession stops calling
updateTasksActiveSessionForAgent.
Test suite updated to use the new session-id API. Existing tests used
`getActiveTaskSession(taskId)` and `createNewSession({ taskId, ...})`
as probes; they now use `getSessionById(sessionId)` and the taskId-less
createNewSession input.
Tests: 603/603 green. Typecheck clean. Migration tested on fresh DB
(runs v1→v41 cleanly) and idempotent re-apply (detects absent dead
columns, skips).
This commit is contained in:
1 parent
6abc1a823f
commit
2ace9183b0
17 files changed
+270
-243
No files matched your search
@@ -49,7 +49,7 @@ export function buildEntityCrudMethods(
|
||||
getTask(id: string): TaskRecord {
|
||||
const row = getDb()
|
||||
.prepare(
|
||||
"select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, active_run_id as activeRunId, session_json as sessionJson, session_display_id as sessionDisplayId, cost_usd as costUsd, task_type as taskType, requires_user_review as requiresUserReview, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks where id = ?",
|
||||
"select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, active_run_id as activeRunId, cost_usd as costUsd, task_type as taskType, requires_user_review as requiresUserReview, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks where id = ?",
|
||||
)
|
||||
.get(id) as TaskRecord | undefined;
|
||||
if (!row) throw new Error(`Task not found: ${id}`);
|
||||
|
||||
@@ -0,0 +1,173 @@
|
||||
import log from "electron-log/main.js";
|
||||
import type { Migration, MigrationDeps } from "./types";
|
||||
import { columnExists, tableExists } from "./types";
|
||||
|
||||
/**
|
||||
* Drop dead columns left behind by the agent-scoped session refactor.
|
||||
*
|
||||
* Pre-v34, a session belonged to a task; the schema reflected that
|
||||
* with `tasks.active_session_id`, `tasks.session_json`,
|
||||
* `tasks.session_display_id`, and a `task_sessions.task_id` FK back to
|
||||
* tasks. Since v34, sessions are agent-scoped — `agents.session_id`
|
||||
* is the canonical pointer, and every runtime read goes through that
|
||||
* or through the session's primary key. The four columns have been
|
||||
* dead storage ever since, and every touch of session code had to
|
||||
* pretend they were still meaningful.
|
||||
*
|
||||
* This migration rebuilds both tables without those columns. SQLite
|
||||
* has no `ALTER TABLE DROP COLUMN` that drops multiple columns
|
||||
* atomically and preserves every constraint; the "copy to new table,
|
||||
* swap, reindex" pattern is the only safe route. `PRAGMA foreign_keys`
|
||||
* is disabled for the swap window and restored afterward so child
|
||||
* tables are not cascade-deleted mid-rebuild.
|
||||
*
|
||||
* The migration is idempotent: each table rebuild checks whether the
|
||||
* dead columns are still present and skips the work otherwise.
|
||||
*/
|
||||
export function createMigrationsV41V41(deps: MigrationDeps): Migration[] {
|
||||
const { db } = deps;
|
||||
return [
|
||||
{
|
||||
version: 41,
|
||||
name: "drop_dead_session_columns",
|
||||
fn: () => {
|
||||
if (!tableExists(db, "tasks") || !tableExists(db, "task_sessions")) {
|
||||
log.info("[migration v41] tasks or task_sessions table missing; skipping");
|
||||
return;
|
||||
}
|
||||
|
||||
const tasksHasDeadCols =
|
||||
columnExists(db, "tasks", "session_json") ||
|
||||
columnExists(db, "tasks", "session_display_id") ||
|
||||
columnExists(db, "tasks", "active_session_id");
|
||||
const sessionsHasDeadCol = columnExists(db, "task_sessions", "task_id");
|
||||
|
||||
if (!tasksHasDeadCols && !sessionsHasDeadCol) {
|
||||
log.info("[migration v41] dead columns already absent; skipping");
|
||||
return;
|
||||
}
|
||||
|
||||
// The `ensureOptionalColumns` helper in schema.ts normally
|
||||
// backfills `task_type`, `requires_user_review`, `retry_count`,
|
||||
// `last_failure_at` on existing installs — but that runs AFTER
|
||||
// migrations, not before. Rebuild the table here with the same
|
||||
// shape as the post-ensureOptionalColumns schema so the old data
|
||||
// is preserved and the new table stays consistent whether
|
||||
// initializeDatabase has reached ensureOptionalColumns yet or not.
|
||||
const ensureTaskCol = (column: string, typeSql: string) => {
|
||||
if (!columnExists(db, "tasks", column)) {
|
||||
db.prepare(`ALTER TABLE tasks ADD COLUMN ${column} ${typeSql}`).run();
|
||||
}
|
||||
};
|
||||
ensureTaskCol("task_type", "TEXT NOT NULL DEFAULT 'general'");
|
||||
ensureTaskCol("requires_user_review", "INTEGER NOT NULL DEFAULT 0");
|
||||
ensureTaskCol("retry_count", "INTEGER NOT NULL DEFAULT 0");
|
||||
ensureTaskCol("last_failure_at", "TEXT");
|
||||
ensureTaskCol("checkout_run_id", "TEXT");
|
||||
ensureTaskCol("execution_locked_at", "TEXT");
|
||||
ensureTaskCol("started_at", "TEXT");
|
||||
ensureTaskCol("completed_at", "TEXT");
|
||||
ensureTaskCol("cancelled_at", "TEXT");
|
||||
|
||||
db.exec("PRAGMA foreign_keys = OFF");
|
||||
try {
|
||||
if (tasksHasDeadCols) {
|
||||
db.exec(`
|
||||
CREATE TABLE tasks_new (
|
||||
id TEXT PRIMARY KEY,
|
||||
company_id TEXT NOT NULL,
|
||||
project_id TEXT,
|
||||
goal_id TEXT,
|
||||
parent_id TEXT,
|
||||
title TEXT NOT NULL,
|
||||
description TEXT NOT NULL,
|
||||
assignee_agent_id TEXT,
|
||||
workspace_id TEXT,
|
||||
status TEXT NOT NULL,
|
||||
priority TEXT NOT NULL,
|
||||
active_run_id TEXT,
|
||||
cost_usd REAL NOT NULL DEFAULT 0,
|
||||
task_type TEXT NOT NULL DEFAULT 'general',
|
||||
requires_user_review INTEGER NOT NULL DEFAULT 0,
|
||||
retry_count INTEGER NOT NULL DEFAULT 0,
|
||||
last_failure_at TEXT,
|
||||
checkout_run_id TEXT,
|
||||
execution_locked_at TEXT,
|
||||
started_at TEXT,
|
||||
completed_at TEXT,
|
||||
cancelled_at TEXT,
|
||||
created_at TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL
|
||||
);
|
||||
INSERT INTO tasks_new (
|
||||
id, company_id, project_id, goal_id, parent_id, title, description,
|
||||
assignee_agent_id, workspace_id, status, priority, active_run_id,
|
||||
cost_usd, task_type, requires_user_review, retry_count, last_failure_at,
|
||||
checkout_run_id, execution_locked_at, started_at, completed_at,
|
||||
cancelled_at, created_at, updated_at
|
||||
)
|
||||
SELECT
|
||||
id, company_id, project_id, goal_id, parent_id, title, description,
|
||||
assignee_agent_id, workspace_id, status, priority, active_run_id,
|
||||
cost_usd, task_type, requires_user_review, retry_count, last_failure_at,
|
||||
checkout_run_id, execution_locked_at, started_at, completed_at,
|
||||
cancelled_at, created_at, updated_at
|
||||
FROM tasks;
|
||||
DROP TABLE tasks;
|
||||
ALTER TABLE tasks_new RENAME TO tasks;
|
||||
CREATE INDEX IF NOT EXISTS idx_tasks_company ON tasks(company_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_tasks_assignee ON tasks(assignee_agent_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_tasks_project ON tasks(project_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_tasks_company_status ON tasks(company_id, status);
|
||||
`);
|
||||
log.info("[migration v41] rebuilt tasks without session_json/session_display_id/active_session_id");
|
||||
}
|
||||
|
||||
if (sessionsHasDeadCol) {
|
||||
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,
|
||||
session_params_json TEXT,
|
||||
session_display_id TEXT,
|
||||
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, session_params_json,
|
||||
session_display_id, 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, session_params_json,
|
||||
session_display_id, 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 v41] rebuilt task_sessions without task_id");
|
||||
}
|
||||
} finally {
|
||||
db.exec("PRAGMA foreign_keys = ON");
|
||||
}
|
||||
},
|
||||
},
|
||||
];
|
||||
}
|
||||
@@ -5,6 +5,7 @@ import { createMigrationsV26V34 } from "./migrations/v26-v34";
|
||||
import { createMigrationsV35V37 } from "./migrations/v35-v37";
|
||||
import { createMigrationsV38V39 } from "./migrations/v38-v39";
|
||||
import { createMigrationsV40V40 } from "./migrations/v40-v40";
|
||||
import { createMigrationsV41V41 } from "./migrations/v41-v41";
|
||||
import {
|
||||
columnExists,
|
||||
nowIso,
|
||||
@@ -176,6 +177,7 @@ export function runMigrations(
|
||||
...createMigrationsV35V37(deps),
|
||||
...createMigrationsV38V39(deps),
|
||||
...createMigrationsV40V40(deps),
|
||||
...createMigrationsV41V41(deps),
|
||||
];
|
||||
|
||||
for (const migration of migrations) {
|
||||
|
||||
@@ -75,7 +75,7 @@ export function buildSnapshotMethods(
|
||||
...companyArgs,
|
||||
),
|
||||
tasks: readAll<TaskRecord>(
|
||||
`select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, active_run_id as activeRunId, session_json as sessionJson, session_display_id as sessionDisplayId, cost_usd as costUsd, task_type as taskType, requires_user_review as requiresUserReview, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks${companyWhere} order by created_at desc limit 2000`,
|
||||
`select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, active_run_id as activeRunId, cost_usd as costUsd, task_type as taskType, requires_user_review as requiresUserReview, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks${companyWhere} order by created_at desc limit 2000`,
|
||||
...companyArgs,
|
||||
).map((record) => ({ ...record, requiresUserReview: toBoolean(record.requiresUserReview), startedAt: record.startedAt ?? null, completedAt: record.completedAt ?? null, cancelledAt: record.cancelledAt ?? null })) as TaskRecord[],
|
||||
approvals: readAll<ApprovalRecord>(
|
||||
|
||||
@@ -25,7 +25,7 @@ export function buildTaskQueryMethods(
|
||||
): TaskRecord[] {
|
||||
const placeholders = statuses.map(() => "?").join(",");
|
||||
return readAll<TaskRecord>(
|
||||
`select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, active_run_id as activeRunId, session_json as sessionJson, session_display_id as sessionDisplayId, cost_usd as costUsd, task_type as taskType, created_at as createdAt, updated_at as updatedAt
|
||||
`select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, active_run_id as activeRunId, cost_usd as costUsd, task_type as taskType, created_at as createdAt, updated_at as updatedAt
|
||||
from tasks
|
||||
where company_id = ? and assignee_agent_id = ? and status in (${placeholders})
|
||||
order by
|
||||
@@ -109,7 +109,7 @@ export function buildTaskQueryMethods(
|
||||
): TaskListByStatusesRow[] {
|
||||
const placeholders = statuses.map(() => "?").join(", ");
|
||||
return readAll<TaskListByStatusesRow & { requiresUserReview: number | boolean }>(
|
||||
`select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, active_run_id as activeRunId, session_json as sessionJson, session_display_id as sessionDisplayId, cost_usd as costUsd, requires_user_review as requiresUserReview, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks where company_id = ? and status in (${placeholders}) order by created_at desc`,
|
||||
`select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, active_run_id as activeRunId, cost_usd as costUsd, requires_user_review as requiresUserReview, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks where company_id = ? and status in (${placeholders}) order by created_at desc`,
|
||||
companyId,
|
||||
...statuses,
|
||||
).map((t) => ({
|
||||
|
||||
@@ -4,16 +4,12 @@ import type { TaskSessionRecord } from "@shared/types";
|
||||
import { type ReadAll, type Transaction, nowIso } from "./helpers";
|
||||
|
||||
/**
|
||||
* Task-session lifecycle helpers extracted from tasks-crud.ts.
|
||||
*
|
||||
* Covers the agent-scoped session lifecycle:
|
||||
* - updateTaskSession / getTaskSession / getActiveTaskSession
|
||||
* - incrementSessionUsage / archiveSession
|
||||
* - ensureAgentSession / setAgentSessionId / updateTasksActiveSessionForAgent
|
||||
* - createNewSession
|
||||
*
|
||||
* Every method is spread into the final `buildTaskMethods` result, so
|
||||
* call-sites on AppDatabase remain unchanged.
|
||||
* Agent-scoped session lifecycle. Sessions belong to agents, not
|
||||
* tasks — `agents.session_id` is the canonical pointer, and
|
||||
* `task_sessions` rows are looked up by their own primary key only.
|
||||
* The pre-migration `tasks.active_session_id` / `task_sessions.task_id`
|
||||
* columns were dropped by v41; any read that still needs a session
|
||||
* given a task id resolves it through the task's assignee agent.
|
||||
*/
|
||||
export function buildTaskSessionMethods(
|
||||
getDb: () => DatabaseSync,
|
||||
@@ -21,118 +17,37 @@ export function buildTaskSessionMethods(
|
||||
transaction: Transaction,
|
||||
) {
|
||||
return {
|
||||
updateTaskSession(
|
||||
taskId: string,
|
||||
sessionJson: Record<string, unknown> | null,
|
||||
/**
|
||||
* 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 = sessionJson ? JSON.stringify(sessionJson) : null;
|
||||
const taskRow = getDb()
|
||||
.prepare(
|
||||
"select company_id as companyId, assignee_agent_id as assigneeAgentId, active_session_id as activeSessionId from tasks where id = ?",
|
||||
)
|
||||
.get(taskId) as
|
||||
| { companyId: string; assigneeAgentId: string | null; activeSessionId: string | null }
|
||||
| undefined;
|
||||
if (!taskRow) throw new Error(`Task not found: ${taskId}`);
|
||||
if (taskRow.activeSessionId) {
|
||||
getDb()
|
||||
.prepare(
|
||||
"update task_sessions set session_params_json = ?, session_display_id = ?, updated_at = ? where id = ?",
|
||||
)
|
||||
.run(sessionParamsJson, sessionDisplayId, ts, taskRow.activeSessionId);
|
||||
return;
|
||||
}
|
||||
if (sessionParamsJson === null && sessionDisplayId === null) return;
|
||||
const agentId = taskRow.assigneeAgentId ?? "unassigned";
|
||||
const agent = taskRow.assigneeAgentId
|
||||
? (getDb()
|
||||
.prepare("select connector_id as connectorId from agents where id = ?")
|
||||
.get(taskRow.assigneeAgentId) as { connectorId: string } | undefined)
|
||||
: undefined;
|
||||
const adapterType = agent?.connectorId ?? "unknown";
|
||||
const sessionId = randomUUID();
|
||||
const sessionParamsJson = sessionParams ? JSON.stringify(sessionParams) : null;
|
||||
getDb()
|
||||
.prepare(
|
||||
`insert into task_sessions (id, company_id, task_id, agent_id, adapter_type, session_params_json, session_display_id, created_at, updated_at)
|
||||
values (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
"update task_sessions set session_params_json = ?, session_display_id = ?, updated_at = ? where id = ?",
|
||||
)
|
||||
.run(
|
||||
sessionId,
|
||||
taskRow.companyId,
|
||||
taskId,
|
||||
agentId,
|
||||
adapterType,
|
||||
sessionParamsJson,
|
||||
sessionDisplayId,
|
||||
ts,
|
||||
ts,
|
||||
);
|
||||
getDb()
|
||||
.prepare("update tasks set active_session_id = ?, updated_at = ? where id = ?")
|
||||
.run(sessionId, ts, taskId);
|
||||
},
|
||||
|
||||
getTaskSession(taskId: string): {
|
||||
sessionParams: Record<string, unknown> | null;
|
||||
sessionDisplayId: string | null;
|
||||
} {
|
||||
const taskRow = getDb()
|
||||
.prepare("select active_session_id as activeSessionId from tasks where id = ?")
|
||||
.get(taskId) as { activeSessionId: string | null } | undefined;
|
||||
if (!taskRow?.activeSessionId) {
|
||||
return { sessionParams: null, sessionDisplayId: null };
|
||||
}
|
||||
const row = getDb()
|
||||
.prepare(
|
||||
"select session_params_json as sessionParamsJson, session_display_id as sessionDisplayId from task_sessions where id = ?",
|
||||
)
|
||||
.get(taskRow.activeSessionId) as
|
||||
| { sessionParamsJson: string | null; sessionDisplayId: string | null }
|
||||
| undefined;
|
||||
let sessionParams: Record<string, unknown> | null = null;
|
||||
if (row?.sessionParamsJson) {
|
||||
sessionParams = JSON.parse(row.sessionParamsJson) as Record<string, unknown>;
|
||||
}
|
||||
return {
|
||||
sessionParams,
|
||||
sessionDisplayId: row?.sessionDisplayId ?? null,
|
||||
};
|
||||
},
|
||||
|
||||
getActiveTaskSession(taskId: string): TaskSessionRecord | null {
|
||||
const task = getDb()
|
||||
.prepare("SELECT active_session_id FROM tasks WHERE id = ?")
|
||||
.get(taskId) as { active_session_id: string | null } | undefined;
|
||||
if (!task?.active_session_id) return null;
|
||||
return (
|
||||
(getDb()
|
||||
.prepare(
|
||||
`SELECT id, company_id as companyId, task_id as taskId, agent_id as agentId,
|
||||
adapter_type as adapterType, session_params_json as sessionParamsJson,
|
||||
session_display_id as sessionDisplayId, 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,
|
||||
handoff_markdown as handoffMarkdown, created_at as createdAt, updated_at as updatedAt
|
||||
FROM task_sessions WHERE id = ?`,
|
||||
)
|
||||
.get(task.active_session_id) as TaskSessionRecord | undefined) ?? null
|
||||
);
|
||||
.run(sessionParamsJson, sessionDisplayId, ts, sessionId);
|
||||
},
|
||||
|
||||
/**
|
||||
* Fetch a session by its primary key. Sessions are agent-scoped
|
||||
* (post-v34 migration): a heartbeat run has no task to dereference,
|
||||
* so the session lookup must take the session id directly. Use with
|
||||
* `agents.session_id` as the entry point.
|
||||
* Fetch a session by its primary key. This is the single lookup
|
||||
* path — the rotation evaluator, the worker's finalizer, and the
|
||||
* turn-context builder all know which session id they want because
|
||||
* they either receive it from the agent record or from the worker
|
||||
* event envelope.
|
||||
*/
|
||||
getSessionById(sessionId: string): TaskSessionRecord | null {
|
||||
return (
|
||||
(getDb()
|
||||
.prepare(
|
||||
`SELECT id, company_id as companyId, task_id as taskId, agent_id as agentId,
|
||||
`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,
|
||||
last_error as lastError, run_count as runCount,
|
||||
@@ -175,17 +90,10 @@ export function buildTaskSessionMethods(
|
||||
},
|
||||
|
||||
/**
|
||||
* Returns the agent's active session id, creating one lazily on first
|
||||
* call. Session is scoped to the agent (not to any particular task);
|
||||
* switching tasks emits a user turn on the ongoing transcript rather
|
||||
* than spawning a new session.
|
||||
*
|
||||
* Existing agents with pre-migration per-task sessions start fresh
|
||||
* after this upgrade: their first run creates a new agent-scoped
|
||||
* session. The old task_sessions + session_entries rows remain in the
|
||||
* DB as historical records but are not loaded for new runs. This is a
|
||||
* one-time transcript reset, accepted over a risky merge on a live
|
||||
* production DB.
|
||||
* Return the agent's active session id, creating one lazily on
|
||||
* first call. One session per agent — switching tasks emits a
|
||||
* "now work on X" user turn on the ongoing transcript rather than
|
||||
* spawning a new session.
|
||||
*/
|
||||
ensureAgentSession(input: {
|
||||
agentId: string;
|
||||
@@ -200,15 +108,12 @@ export function buildTaskSessionMethods(
|
||||
const sessionId = randomUUID();
|
||||
const ts = nowIso();
|
||||
transaction(() => {
|
||||
// task_id is left as empty string — the session is agent-scoped;
|
||||
// the column persists for backwards-compatible schema but no
|
||||
// longer keys the session to a particular task.
|
||||
getDb()
|
||||
.prepare(
|
||||
`INSERT INTO task_sessions (id, company_id, task_id, agent_id, adapter_type,
|
||||
`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, ?, ?)`,
|
||||
VALUES (?, ?, ?, ?, NULL, NULL, 0, 0, 0, ?, ?)`,
|
||||
)
|
||||
.run(sessionId, input.companyId, input.agentId, input.adapterType, ts, ts);
|
||||
getDb()
|
||||
@@ -226,21 +131,11 @@ export function buildTaskSessionMethods(
|
||||
},
|
||||
|
||||
/**
|
||||
* Point every task currently assigned to an agent at the agent's new
|
||||
* active session. Called after session rotation so UI paths reading
|
||||
* per-task active_session_id still see the agent's current session.
|
||||
* 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.
|
||||
*/
|
||||
updateTasksActiveSessionForAgent(agentId: string, sessionId: string): void {
|
||||
const ts = nowIso();
|
||||
getDb()
|
||||
.prepare(
|
||||
"UPDATE tasks SET active_session_id = ?, updated_at = ? WHERE assignee_agent_id = ?",
|
||||
)
|
||||
.run(sessionId, ts, agentId);
|
||||
},
|
||||
|
||||
createNewSession(input: {
|
||||
taskId: string;
|
||||
companyId: string;
|
||||
agentId: string;
|
||||
adapterType: string;
|
||||
@@ -251,14 +146,13 @@ export function buildTaskSessionMethods(
|
||||
const ts = nowIso();
|
||||
getDb()
|
||||
.prepare(
|
||||
`INSERT INTO task_sessions (id, company_id, task_id, agent_id, adapter_type,
|
||||
`INSERT INTO task_sessions (id, company_id, agent_id, adapter_type,
|
||||
session_params_json, session_display_id, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
)
|
||||
.run(
|
||||
sessionId,
|
||||
input.companyId,
|
||||
input.taskId,
|
||||
input.agentId,
|
||||
input.adapterType,
|
||||
input.sessionParamsJson,
|
||||
@@ -266,9 +160,6 @@ export function buildTaskSessionMethods(
|
||||
ts,
|
||||
ts,
|
||||
);
|
||||
getDb()
|
||||
.prepare("UPDATE tasks SET active_session_id = ?, updated_at = ? WHERE id = ?")
|
||||
.run(sessionId, ts, input.taskId);
|
||||
return sessionId;
|
||||
},
|
||||
};
|
||||
|
||||
@@ -26,8 +26,6 @@ export type TaskListByStatusesRow = {
|
||||
status: string;
|
||||
priority: string;
|
||||
activeRunId: string | null;
|
||||
sessionJson: string | null;
|
||||
sessionDisplayId: string | null;
|
||||
costUsd: number | null;
|
||||
requiresUserReview: boolean;
|
||||
startedAt: string | null;
|
||||
|
||||
@@ -107,7 +107,6 @@ export function buildTaskStatusMethods(
|
||||
t.parent_id as parentId, t.title, t.description,
|
||||
t.assignee_agent_id as assigneeAgentId, t.workspace_id as workspaceId,
|
||||
t.status, t.priority, t.active_run_id as activeRunId,
|
||||
t.session_json as sessionJson, t.session_display_id as sessionDisplayId,
|
||||
t.cost_usd as costUsd,
|
||||
t.created_at as createdAt, t.updated_at as updatedAt
|
||||
from tasks t
|
||||
|
||||
Reference in new issue
Block a user