diff --git a/src/main/database/agent-permissions.ts b/src/main/database/agent-permissions.ts new file mode 100644 index 00000000..294b1eef --- /dev/null +++ b/src/main/database/agent-permissions.ts @@ -0,0 +1,151 @@ +import type { DatabaseSync } from "node:sqlite"; +import { type ReadAll, type Transaction, nowIso, parseJson, toBoolean } from "./helpers"; + +/** + * Agent permission, runtime-state, and heartbeat-policy helpers extracted + * from agents-hierarchy.ts. Every method is spread into the final + * `buildAgentMethods` result so AppDatabase call-sites remain unchanged. + */ +export function buildAgentPermissionMethods( + getDb: () => DatabaseSync, + _readAll: ReadAll, + _transaction: Transaction, +) { + return { + getAgentPermissions(agentId: string): Record { + const row = getDb() + .prepare("SELECT permissions_json FROM agents WHERE id = ?") + .get(agentId) as { permissions_json: string } | undefined; + return parseJson>(row?.permissions_json, {}); + }, + + setAgentPermission( + agentId: string, + companyId: string, + permissionKey: string, + enabled: boolean, + ): void { + const row = getDb() + .prepare("SELECT permissions_json FROM agents WHERE id = ?") + .get(agentId) as { permissions_json: string } | undefined; + const current = parseJson>(row?.permissions_json, {}); + if (enabled) { + current[permissionKey] = true; + } else { + delete current[permissionKey]; + } + const ts = nowIso(); + getDb() + .prepare( + "UPDATE agents SET permissions_json = ?, updated_at = ? WHERE id = ? AND company_id = ?", + ) + .run(JSON.stringify(current), ts, agentId, companyId); + }, + + hasAgentPermission(agentId: string, permissionKey: string): boolean { + const row = getDb() + .prepare("SELECT permissions_json FROM agents WHERE id = ?") + .get(agentId) as { permissions_json: string } | undefined; + const perms = parseJson>(row?.permissions_json, {}); + return perms[permissionKey] === true; + }, + + getRuntimeState(agentId: string): Record { + const row = getDb() + .prepare("SELECT runtime_state_json FROM agents WHERE id = ?") + .get(agentId) as { runtime_state_json: string | null } | undefined; + return parseJson>(row?.runtime_state_json, {}); + }, + + setRuntimeState( + agentId: string, + companyId: string, + state: Record, + ): void { + const ts = nowIso(); + getDb() + .prepare( + "UPDATE agents SET runtime_state_json = ?, updated_at = ? WHERE id = ? AND company_id = ?", + ) + .run(JSON.stringify(state), ts, agentId, companyId); + }, + + getHeartbeatPolicy(agentId: string): { + enabled: boolean; + intervalSec: number; + cooldownSec: number; + maxConcurrentRuns: number; + wakeOnDemand: boolean; + } { + const row = getDb() + .prepare( + "SELECT heartbeat_enabled, heartbeat_interval_sec, heartbeat_policy_json FROM agents WHERE id = ?", + ) + .get(agentId) as + | { + heartbeat_enabled: number; + heartbeat_interval_sec: number; + heartbeat_policy_json: string | null; + } + | undefined; + if (!row) throw new Error(`Agent ${agentId} not found.`); + const policy = parseJson>(row.heartbeat_policy_json, {}); + return { + enabled: toBoolean(row.heartbeat_enabled), + intervalSec: row.heartbeat_interval_sec ?? 120, + cooldownSec: typeof policy.cooldownSec === "number" ? policy.cooldownSec : 10, + maxConcurrentRuns: + typeof policy.maxConcurrentRuns === "number" ? policy.maxConcurrentRuns : 1, + wakeOnDemand: policy.wakeOnDemand !== false, + }; + }, + + setHeartbeatPolicy( + agentId: string, + companyId: string, + policy: { + enabled?: boolean; + intervalSec?: number; + cooldownSec?: number; + maxConcurrentRuns?: number; + wakeOnDemand?: boolean; + }, + ): void { + const row = getDb() + .prepare( + "SELECT heartbeat_enabled, heartbeat_interval_sec, heartbeat_policy_json FROM agents WHERE id = ?", + ) + .get(agentId) as + | { + heartbeat_enabled: number; + heartbeat_interval_sec: number; + heartbeat_policy_json: string | null; + } + | undefined; + if (!row) throw new Error(`Agent ${agentId} not found.`); + const current = parseJson>(row.heartbeat_policy_json, {}); + const merged = { + enabled: policy.enabled ?? toBoolean(row.heartbeat_enabled), + intervalSec: policy.intervalSec ?? row.heartbeat_interval_sec ?? 120, + cooldownSec: + policy.cooldownSec ?? + (typeof current.cooldownSec === "number" ? current.cooldownSec : 10), + maxConcurrentRuns: + policy.maxConcurrentRuns ?? + (typeof current.maxConcurrentRuns === "number" ? current.maxConcurrentRuns : 1), + wakeOnDemand: policy.wakeOnDemand ?? (current.wakeOnDemand !== false), + }; + const policyJson = JSON.stringify({ + cooldownSec: merged.cooldownSec, + maxConcurrentRuns: merged.maxConcurrentRuns, + wakeOnDemand: merged.wakeOnDemand, + }); + const ts = nowIso(); + getDb() + .prepare( + "UPDATE agents SET heartbeat_enabled = ?, heartbeat_interval_sec = ?, heartbeat_policy_json = ?, updated_at = ? WHERE id = ? AND company_id = ?", + ) + .run(merged.enabled ? 1 : 0, merged.intervalSec, policyJson, ts, agentId, companyId); + }, + }; +} diff --git a/src/main/database/agents-hierarchy.ts b/src/main/database/agents-hierarchy.ts index fea27214..fe6ccda7 100644 --- a/src/main/database/agents-hierarchy.ts +++ b/src/main/database/agents-hierarchy.ts @@ -4,7 +4,6 @@ import { type ReadAll, type Transaction, nowIso, - parseJson, toBoolean, } from "./helpers"; @@ -273,142 +272,6 @@ export function buildAgentHierarchyMethods( .run(nowIso(), companyId); }, - getAgentPermissions(agentId: string): Record { - const row = getDb() - .prepare("SELECT permissions_json FROM agents WHERE id = ?") - .get(agentId) as { permissions_json: string } | undefined; - return parseJson>(row?.permissions_json, {}); - }, - - setAgentPermission( - agentId: string, - companyId: string, - permissionKey: string, - enabled: boolean, - ): void { - const row = getDb() - .prepare("SELECT permissions_json FROM agents WHERE id = ?") - .get(agentId) as { permissions_json: string } | undefined; - const current = parseJson>(row?.permissions_json, {}); - if (enabled) { - current[permissionKey] = true; - } else { - delete current[permissionKey]; - } - const ts = nowIso(); - getDb() - .prepare( - "UPDATE agents SET permissions_json = ?, updated_at = ? WHERE id = ? AND company_id = ?", - ) - .run(JSON.stringify(current), ts, agentId, companyId); - }, - - hasAgentPermission(agentId: string, permissionKey: string): boolean { - const row = getDb() - .prepare("SELECT permissions_json FROM agents WHERE id = ?") - .get(agentId) as { permissions_json: string } | undefined; - const perms = parseJson>(row?.permissions_json, {}); - return perms[permissionKey] === true; - }, - - getRuntimeState(agentId: string): Record { - const row = getDb() - .prepare("SELECT runtime_state_json FROM agents WHERE id = ?") - .get(agentId) as { runtime_state_json: string | null } | undefined; - return parseJson>(row?.runtime_state_json, {}); - }, - - setRuntimeState( - agentId: string, - companyId: string, - state: Record, - ): void { - const ts = nowIso(); - getDb() - .prepare( - "UPDATE agents SET runtime_state_json = ?, updated_at = ? WHERE id = ? AND company_id = ?", - ) - .run(JSON.stringify(state), ts, agentId, companyId); - }, - - getHeartbeatPolicy(agentId: string): { - enabled: boolean; - intervalSec: number; - cooldownSec: number; - maxConcurrentRuns: number; - wakeOnDemand: boolean; - } { - const row = getDb() - .prepare( - "SELECT heartbeat_enabled, heartbeat_interval_sec, heartbeat_policy_json FROM agents WHERE id = ?", - ) - .get(agentId) as - | { - heartbeat_enabled: number; - heartbeat_interval_sec: number; - heartbeat_policy_json: string | null; - } - | undefined; - if (!row) throw new Error(`Agent ${agentId} not found.`); - const policy = parseJson>(row.heartbeat_policy_json, {}); - return { - enabled: toBoolean(row.heartbeat_enabled), - intervalSec: row.heartbeat_interval_sec ?? 120, - cooldownSec: typeof policy.cooldownSec === "number" ? policy.cooldownSec : 10, - maxConcurrentRuns: - typeof policy.maxConcurrentRuns === "number" ? policy.maxConcurrentRuns : 1, - wakeOnDemand: policy.wakeOnDemand !== false, - }; - }, - - setHeartbeatPolicy( - agentId: string, - companyId: string, - policy: { - enabled?: boolean; - intervalSec?: number; - cooldownSec?: number; - maxConcurrentRuns?: number; - wakeOnDemand?: boolean; - }, - ): void { - const row = getDb() - .prepare( - "SELECT heartbeat_enabled, heartbeat_interval_sec, heartbeat_policy_json FROM agents WHERE id = ?", - ) - .get(agentId) as - | { - heartbeat_enabled: number; - heartbeat_interval_sec: number; - heartbeat_policy_json: string | null; - } - | undefined; - if (!row) throw new Error(`Agent ${agentId} not found.`); - const current = parseJson>(row.heartbeat_policy_json, {}); - const merged = { - enabled: policy.enabled ?? toBoolean(row.heartbeat_enabled), - intervalSec: policy.intervalSec ?? row.heartbeat_interval_sec ?? 120, - cooldownSec: - policy.cooldownSec ?? - (typeof current.cooldownSec === "number" ? current.cooldownSec : 10), - maxConcurrentRuns: - policy.maxConcurrentRuns ?? - (typeof current.maxConcurrentRuns === "number" ? current.maxConcurrentRuns : 1), - wakeOnDemand: policy.wakeOnDemand ?? (current.wakeOnDemand !== false), - }; - const policyJson = JSON.stringify({ - cooldownSec: merged.cooldownSec, - maxConcurrentRuns: merged.maxConcurrentRuns, - wakeOnDemand: merged.wakeOnDemand, - }); - const ts = nowIso(); - getDb() - .prepare( - "UPDATE agents SET heartbeat_enabled = ?, heartbeat_interval_sec = ?, heartbeat_policy_json = ?, updated_at = ? WHERE id = ? AND company_id = ?", - ) - .run(merged.enabled ? 1 : 0, merged.intervalSec, policyJson, ts, agentId, companyId); - }, - /** * Returns ids of non-terminated agents in a department. Used by * `resolveMessageWakeTargets` to fan out "department" channel messages. diff --git a/src/main/database/agents.ts b/src/main/database/agents.ts index d886bf6a..456d077a 100644 --- a/src/main/database/agents.ts +++ b/src/main/database/agents.ts @@ -3,6 +3,7 @@ import type { AgentRecord } from "@shared/types"; import type { ReadAll, Transaction } from "./helpers"; import { buildAgentCrudMethods } from "./agents-crud"; import { buildAgentHierarchyMethods } from "./agents-hierarchy"; +import { buildAgentPermissionMethods } from "./agent-permissions"; export function buildAgentMethods( getDb: () => DatabaseSync, @@ -29,5 +30,6 @@ export function buildAgentMethods( return { ...buildAgentCrudMethods(getDb, readAll, transaction, deps), ...buildAgentHierarchyMethods(getDb, readAll, transaction, { getAgent: deps.getAgent }), + ...buildAgentPermissionMethods(getDb, readAll, transaction), }; } diff --git a/src/main/database/company-metrics.ts b/src/main/database/company-metrics.ts new file mode 100644 index 00000000..6cd57670 --- /dev/null +++ b/src/main/database/company-metrics.ts @@ -0,0 +1,370 @@ +import type { DatabaseSync } from "node:sqlite"; +import type { + CompanyMetrics, + AgentMetrics, + RunStatus, + StandupReport, +} from "@shared/types"; +import { type ReadAll, type Transaction } from "./helpers"; + +/** + * Dashboard-aggregate read queries extracted from company-settings.ts. + * Covers getCompanyMetrics and getStandupReport. Every method is spread + * into the final `buildCompanyMethods` result so AppDatabase call-sites + * remain unchanged. + */ +export function buildCompanyMetricsMethods( + getDb: () => DatabaseSync, + readAll: ReadAll, + _transaction: Transaction, +) { + return { + getCompanyMetrics(companyId: string, days: number = 30): CompanyMetrics { + const cutoff = new Date(Date.now() - days * 24 * 60 * 60 * 1000).toISOString(); + const db = getDb(); + + const agentRows = readAll<{ + id: string; name: string; status: string; + }>( + `select id, name, status from agents where company_id = ? order by created_at asc`, + companyId, + ); + const activeAgentRows = agentRows.filter( + (a) => a.status !== "pending_approval" && a.status !== "terminated", + ); + + const totalAgents = activeAgentRows.length; + const activeAgents = agentRows.filter( + (a) => a.status === "active" || a.status === "running", + ).length; + + const totalTasks = (db.prepare( + "select count(*) as c from tasks where company_id = ?", + ).get(companyId) as { c: number }).c; + const completedTasks = (db.prepare( + "select count(*) as c from tasks where company_id = ? and status = 'done'", + ).get(companyId) as { c: number }).c; + const blockedTasks = (db.prepare( + "select count(*) as c from tasks where company_id = ? and status = 'blocked'", + ).get(companyId) as { c: number }).c; + + const recentRunStats = db.prepare( + `select + count(*) as totalRuns, + sum(case when status = 'succeeded' then 1 else 0 end) as successful, + sum(case when status = 'failed' or status = 'timed_out' then 1 else 0 end) as failed + from runs where company_id = ? and created_at > ?`, + ).get(companyId, cutoff) as { totalRuns: number; successful: number | null; failed: number | null }; + const totalRuns = recentRunStats.totalRuns ?? 0; + const successfulRuns = recentRunStats.successful ?? 0; + const failedRuns = recentRunStats.failed ?? 0; + + const totalCost = (db.prepare( + "select coalesce(sum(amount_usd), 0) as s from cost_entries where company_id = ?", + ).get(companyId) as { s: number }).s; + + const recentTasksDone = (db.prepare( + "select count(*) as c from tasks where company_id = ? and status = 'done' and updated_at > ?", + ).get(companyId, cutoff) as { c: number }).c; + + const taskAggRows = readAll<{ + agentId: string; + done: number; + assigned: number; + cancelled: number; + }>( + `select assignee_agent_id as agentId, + sum(case when status = 'done' then 1 else 0 end) as done, + sum(case when status != 'done' and status != 'cancelled' then 1 else 0 end) as assigned, + sum(case when status = 'cancelled' then 1 else 0 end) as cancelled + from tasks + where company_id = ? and assignee_agent_id is not null + group by assignee_agent_id`, + companyId, + ); + const taskAggMap = new Map(taskAggRows.map((row) => [row.agentId, row])); + + const runRecentAggRows = readAll<{ + agentId: string; + total: number; + succeeded: number; + failed: number; + }>( + `select agent_id as agentId, + count(*) as total, + sum(case when status = 'succeeded' then 1 else 0 end) as succeeded, + sum(case when status = 'failed' or status = 'timed_out' then 1 else 0 end) as failed + from runs + where company_id = ? and created_at > ? + group by agent_id`, + companyId, + cutoff, + ); + const runRecentAggMap = new Map(runRecentAggRows.map((row) => [row.agentId, row])); + + const runDurationRows = readAll<{ + agentId: string; + avgDurationSec: number | null; + lastActiveAt: string | null; + }>( + `select agent_id as agentId, + avg((julianday(finished_at) - julianday(started_at)) * 86400.0) as avgDurationSec, + max(created_at) as lastActiveAt + from runs + where company_id = ? and started_at is not null and finished_at is not null + group by agent_id`, + companyId, + ); + const runDurationMap = new Map(runDurationRows.map((row) => [row.agentId, row])); + + const lastActiveRows = readAll<{ agentId: string; lastActiveAt: string }>( + `select agent_id as agentId, max(created_at) as lastActiveAt + from runs where company_id = ? group by agent_id`, + companyId, + ); + const lastActiveMap = new Map(lastActiveRows.map((row) => [row.agentId, row.lastActiveAt])); + + const costAggRows = readAll<{ agentId: string; totalCost: number }>( + `select agent_id as agentId, coalesce(sum(amount_usd), 0) as totalCost + from cost_entries where company_id = ? and agent_id is not null + group by agent_id`, + companyId, + ); + const costAggMap = new Map(costAggRows.map((row) => [row.agentId, row.totalCost])); + + const reviewCycleRows = readAll<{ actor: string; cycles: number }>( + `select actor, count(*) as cycles + from activity + where company_id = ? + and action in ('task.status_changed', 'task.updated', 'task.auto_assigned') + group by actor`, + companyId, + ); + const reviewCycleMap = new Map(reviewCycleRows.map((row) => [row.actor, row.cycles])); + + const agentMetrics: AgentMetrics[] = activeAgentRows.map((agent) => { + const taskAgg = taskAggMap.get(agent.id); + const runRecent = runRecentAggMap.get(agent.id); + const runDur = runDurationMap.get(agent.id); + const succeeded = runRecent?.succeeded ?? 0; + const total = runRecent?.total ?? 0; + return { + agentId: agent.id, + agentName: agent.name, + tasksCompleted: taskAgg?.done ?? 0, + tasksAssigned: taskAgg?.assigned ?? 0, + tasksFailed: taskAgg?.cancelled ?? 0, + successRate: total > 0 ? Math.round((succeeded / total) * 100) : 0, + avgRunDurationSec: Math.round(runDur?.avgDurationSec ?? 0), + totalCostUsd: costAggMap.get(agent.id) ?? 0, + runsTotal: total, + runsSucceeded: succeeded, + runsFailed: runRecent?.failed ?? 0, + reviewCycles: reviewCycleMap.get(agent.id) ?? 0, + lastActiveAt: lastActiveMap.get(agent.id) ?? null, + }; + }); + + return { + totalAgents, + activeAgents, + totalTasks, + completedTasks, + blockedTasks, + totalRuns, + successfulRuns, + failedRuns, + overallSuccessRate: totalRuns > 0 ? Math.round((successfulRuns / totalRuns) * 100) : 0, + totalCostUsd: totalCost, + avgCostPerTask: completedTasks > 0 ? totalCost / completedTasks : 0, + throughputTasksPerDay: days > 0 ? recentTasksDone / days : 0, + agentMetrics, + }; + }, + + getStandupReport(companyId: string): StandupReport { + const db = getDb(); + const oneDayAgo = new Date(Date.now() - 24 * 60 * 60 * 1000).toISOString(); + + type AgentRow = { + id: string; name: string; role: string; status: StandupReport["agentReports"][number]["status"]; + budgetMonthlyUsd: number; spentMonthlyUsd: number; + }; + const agents = readAll( + `select id, name, role, status, + budget_monthly_usd as budgetMonthlyUsd, + spent_monthly_usd as spentMonthlyUsd + from agents + where company_id = ? and status != 'terminated' and status != 'pending_approval' + order by created_at asc`, + companyId, + ); + + const recentCompletedTasksCount = (db.prepare( + "select count(*) as c from tasks where company_id = ? and status = 'done' and updated_at > ?", + ).get(companyId, oneDayAgo) as { c: number }).c; + + const recentRunStats = db.prepare( + `select + count(*) as total, + sum(case when status = 'succeeded' then 1 else 0 end) as succeeded, + sum(case when status = 'failed' or status = 'timed_out' then 1 else 0 end) as failed + from runs where company_id = ? and created_at > ?`, + ).get(companyId, oneDayAgo) as { total: number; succeeded: number | null; failed: number | null }; + const recentRunsTotal = recentRunStats.total ?? 0; + const successfulRuns = recentRunStats.succeeded ?? 0; + const failedRuns = recentRunStats.failed ?? 0; + + const blockedTaskRows = readAll<{ id: string; title: string; assigneeAgentId: string | null }>( + `select id, title, assignee_agent_id as assigneeAgentId + from tasks where company_id = ? and status = 'blocked'`, + companyId, + ); + + const pendingApprovals = (db.prepare( + "select count(*) as c from approvals where company_id = ? and state = 'pending'", + ).get(companyId) as { c: number }).c; + + const activeSprint = db.prepare( + `select name, velocity_points as velocityPoints, completed_points as completedPoints + from sprints where company_id = ? and status = 'active' order by start_date desc limit 1`, + ).get(companyId) as { name: string; velocityPoints: number; completedPoints: number } | undefined; + + const urgentUnread = (db.prepare( + `select count(*) as c + from agent_messages messages + left join agent_message_reads reads + on reads.message_id = messages.id and reads.agent_id = '__board__' + where messages.company_id = ? + and reads.read_at is null + and (messages.priority = 'urgent' or messages.channel = 'incident')`, + ).get(companyId) as { c: number }).c; + + const tasksByAgentRows = readAll<{ + agentId: string; inProgress: number; blocked: number; completed24h: number; + }>( + `select assignee_agent_id as agentId, + sum(case when status = 'in_progress' then 1 else 0 end) as inProgress, + sum(case when status = 'blocked' then 1 else 0 end) as blocked, + sum(case when status = 'done' and updated_at > ? then 1 else 0 end) as completed24h + from tasks + where company_id = ? and assignee_agent_id is not null + group by assignee_agent_id`, + oneDayAgo, + companyId, + ); + const tasksByAgent = new Map(tasksByAgentRows.map((row) => [row.agentId, row])); + + const recentRunsByAgent = readAll<{ + agentId: string; runCount: number; lastStatus: string | null; lastAt: string | null; + }>( + `select agent_id as agentId, + count(*) as runCount, + (select status from runs r2 where r2.agent_id = runs.agent_id and r2.company_id = runs.company_id and r2.created_at > ? order by created_at desc limit 1) as lastStatus, + max(created_at) as lastAt + from runs + where company_id = ? and created_at > ? + group by agent_id`, + oneDayAgo, + companyId, + oneDayAgo, + ); + const runsByAgent = new Map(recentRunsByAgent.map((row) => [row.agentId, row])); + + const costByAgentRows = readAll<{ agentId: string; cost24h: number }>( + `select agent_id as agentId, coalesce(sum(amount_usd), 0) as cost24h + from cost_entries + where company_id = ? and agent_id is not null and created_at > ? + group by agent_id`, + companyId, + oneDayAgo, + ); + const costByAgent = new Map(costByAgentRows.map((row) => [row.agentId, row.cost24h])); + + const reviewByAgentRows = readAll<{ actor: string; cycles: number }>( + `select actor, count(*) as cycles + from activity + where company_id = ? and created_at > ? + and action in ('task.status_changed', 'task.updated', 'task.auto_assigned') + group by actor`, + companyId, + oneDayAgo, + ); + const reviewByAgent = new Map(reviewByAgentRows.map((row) => [row.actor, row.cycles])); + + const agentNameMap = new Map(agents.map((a) => [a.id, a.name])); + + const companyHighlights: string[] = []; + const blockers: string[] = []; + if (recentCompletedTasksCount > 0) { + companyHighlights.push(`${recentCompletedTasksCount} task(s) completed in last 24h`); + } + if (recentRunsTotal > 0) { + companyHighlights.push( + `${recentRunsTotal} runs (${successfulRuns} succeeded, ${failedRuns} failed)`, + ); + } + for (const task of blockedTaskRows) { + const assigneeName = task.assigneeAgentId ? agentNameMap.get(task.assigneeAgentId) ?? null : null; + blockers.push(`${task.title} — blocked (${assigneeName ?? "unassigned"})`); + } + if (activeSprint) { + const pct = activeSprint.velocityPoints > 0 + ? Math.round((activeSprint.completedPoints / activeSprint.velocityPoints) * 100) + : 0; + companyHighlights.push( + `Sprint "${activeSprint.name}": ${activeSprint.completedPoints}/${activeSprint.velocityPoints} pts (${pct}%)`, + ); + } + if (urgentUnread > 0) { + blockers.push(`${urgentUnread} urgent unread message(s) need attention`); + } + + const totalBudget = agents.reduce((sum, a) => sum + a.budgetMonthlyUsd, 0); + const totalSpent = agents.reduce((sum, a) => sum + a.spentMonthlyUsd, 0); + const budgetUtilization = totalBudget > 0 ? Math.round((totalSpent / totalBudget) * 100) : 0; + + const agentReports = agents.map((agent) => { + const taskAgg = tasksByAgent.get(agent.id); + const runAgg = runsByAgent.get(agent.id); + const completed24h = taskAgg?.completed24h ?? 0; + const runs24h = runAgg?.runCount ?? 0; + const reviewedTasks = reviewByAgent.get(agent.id) ?? 0; + const highlights: string[] = []; + if (completed24h > 0) highlights.push(`Completed ${completed24h} task(s)`); + if (runs24h > 0) highlights.push(`${runs24h} run(s) in last 24h`); + if (reviewedTasks > 0) highlights.push(`${reviewedTasks} status update(s)`); + return { + agentId: agent.id, + agentName: agent.name, + role: agent.role, + status: agent.status, + tasksInProgress: taskAgg?.inProgress ?? 0, + tasksCompleted24h: completed24h, + tasksBlocked: taskAgg?.blocked ?? 0, + lastRunStatus: (runAgg?.lastStatus ?? null) as RunStatus | null, + lastRunAt: runAgg?.lastAt ?? null, + costLast24h: costByAgent.get(agent.id) ?? 0, + highlights, + }; + }); + + const summaryParts: string[] = []; + if (agents.length > 0) summaryParts.push(`${agents.length} active agents`); + summaryParts.push(`${recentCompletedTasksCount} tasks done`); + if (blockedTaskRows.length > 0) summaryParts.push(`${blockedTaskRows.length} blocked`); + if (pendingApprovals > 0) summaryParts.push(`${pendingApprovals} pending approvals`); + + return { + companyId, + generatedAt: new Date().toISOString(), + summary: summaryParts.join(" · "), + agentReports, + companyHighlights, + blockers, + pendingApprovals, + budgetUtilization, + }; + }, + }; +} diff --git a/src/main/database/company-settings.ts b/src/main/database/company-settings.ts index 3bd14198..f5cd420c 100644 --- a/src/main/database/company-settings.ts +++ b/src/main/database/company-settings.ts @@ -1,11 +1,5 @@ import type { DatabaseSync } from "node:sqlite"; -import type { - CompanyMetrics, - AgentMetrics, - ProfileSnapshot, - RunStatus, - StandupReport, -} from "@shared/types"; +import type { ProfileSnapshot } from "@shared/types"; import { BOARD_READER_ID, type ReadAll, @@ -59,354 +53,6 @@ export function buildCompanySettingsMethods( }; }, - getCompanyMetrics(companyId: string, days: number = 30): CompanyMetrics { - const cutoff = new Date(Date.now() - days * 24 * 60 * 60 * 1000).toISOString(); - const db = getDb(); - - const agentRows = readAll<{ - id: string; name: string; status: string; - }>( - `select id, name, status from agents where company_id = ? order by created_at asc`, - companyId, - ); - const activeAgentRows = agentRows.filter( - (a) => a.status !== "pending_approval" && a.status !== "terminated", - ); - - const totalAgents = activeAgentRows.length; - const activeAgents = agentRows.filter( - (a) => a.status === "active" || a.status === "running", - ).length; - - const totalTasks = (db.prepare( - "select count(*) as c from tasks where company_id = ?", - ).get(companyId) as { c: number }).c; - const completedTasks = (db.prepare( - "select count(*) as c from tasks where company_id = ? and status = 'done'", - ).get(companyId) as { c: number }).c; - const blockedTasks = (db.prepare( - "select count(*) as c from tasks where company_id = ? and status = 'blocked'", - ).get(companyId) as { c: number }).c; - - const recentRunStats = db.prepare( - `select - count(*) as totalRuns, - sum(case when status = 'succeeded' then 1 else 0 end) as successful, - sum(case when status = 'failed' or status = 'timed_out' then 1 else 0 end) as failed - from runs where company_id = ? and created_at > ?`, - ).get(companyId, cutoff) as { totalRuns: number; successful: number | null; failed: number | null }; - const totalRuns = recentRunStats.totalRuns ?? 0; - const successfulRuns = recentRunStats.successful ?? 0; - const failedRuns = recentRunStats.failed ?? 0; - - const totalCost = (db.prepare( - "select coalesce(sum(amount_usd), 0) as s from cost_entries where company_id = ?", - ).get(companyId) as { s: number }).s; - - const recentTasksDone = (db.prepare( - "select count(*) as c from tasks where company_id = ? and status = 'done' and updated_at > ?", - ).get(companyId, cutoff) as { c: number }).c; - - const taskAggRows = readAll<{ - agentId: string; - done: number; - assigned: number; - cancelled: number; - }>( - `select assignee_agent_id as agentId, - sum(case when status = 'done' then 1 else 0 end) as done, - sum(case when status != 'done' and status != 'cancelled' then 1 else 0 end) as assigned, - sum(case when status = 'cancelled' then 1 else 0 end) as cancelled - from tasks - where company_id = ? and assignee_agent_id is not null - group by assignee_agent_id`, - companyId, - ); - const taskAggMap = new Map(taskAggRows.map((row) => [row.agentId, row])); - - const runRecentAggRows = readAll<{ - agentId: string; - total: number; - succeeded: number; - failed: number; - }>( - `select agent_id as agentId, - count(*) as total, - sum(case when status = 'succeeded' then 1 else 0 end) as succeeded, - sum(case when status = 'failed' or status = 'timed_out' then 1 else 0 end) as failed - from runs - where company_id = ? and created_at > ? - group by agent_id`, - companyId, - cutoff, - ); - const runRecentAggMap = new Map(runRecentAggRows.map((row) => [row.agentId, row])); - - const runDurationRows = readAll<{ - agentId: string; - avgDurationSec: number | null; - lastActiveAt: string | null; - }>( - `select agent_id as agentId, - avg((julianday(finished_at) - julianday(started_at)) * 86400.0) as avgDurationSec, - max(created_at) as lastActiveAt - from runs - where company_id = ? and started_at is not null and finished_at is not null - group by agent_id`, - companyId, - ); - const runDurationMap = new Map(runDurationRows.map((row) => [row.agentId, row])); - - const lastActiveRows = readAll<{ agentId: string; lastActiveAt: string }>( - `select agent_id as agentId, max(created_at) as lastActiveAt - from runs where company_id = ? group by agent_id`, - companyId, - ); - const lastActiveMap = new Map(lastActiveRows.map((row) => [row.agentId, row.lastActiveAt])); - - const costAggRows = readAll<{ agentId: string; totalCost: number }>( - `select agent_id as agentId, coalesce(sum(amount_usd), 0) as totalCost - from cost_entries where company_id = ? and agent_id is not null - group by agent_id`, - companyId, - ); - const costAggMap = new Map(costAggRows.map((row) => [row.agentId, row.totalCost])); - - const reviewCycleRows = readAll<{ actor: string; cycles: number }>( - `select actor, count(*) as cycles - from activity - where company_id = ? - and action in ('task.status_changed', 'task.updated', 'task.auto_assigned') - group by actor`, - companyId, - ); - const reviewCycleMap = new Map(reviewCycleRows.map((row) => [row.actor, row.cycles])); - - const agentMetrics: AgentMetrics[] = activeAgentRows.map((agent) => { - const taskAgg = taskAggMap.get(agent.id); - const runRecent = runRecentAggMap.get(agent.id); - const runDur = runDurationMap.get(agent.id); - const succeeded = runRecent?.succeeded ?? 0; - const total = runRecent?.total ?? 0; - return { - agentId: agent.id, - agentName: agent.name, - tasksCompleted: taskAgg?.done ?? 0, - tasksAssigned: taskAgg?.assigned ?? 0, - tasksFailed: taskAgg?.cancelled ?? 0, - successRate: total > 0 ? Math.round((succeeded / total) * 100) : 0, - avgRunDurationSec: Math.round(runDur?.avgDurationSec ?? 0), - totalCostUsd: costAggMap.get(agent.id) ?? 0, - runsTotal: total, - runsSucceeded: succeeded, - runsFailed: runRecent?.failed ?? 0, - reviewCycles: reviewCycleMap.get(agent.id) ?? 0, - lastActiveAt: lastActiveMap.get(agent.id) ?? null, - }; - }); - - return { - totalAgents, - activeAgents, - totalTasks, - completedTasks, - blockedTasks, - totalRuns, - successfulRuns, - failedRuns, - overallSuccessRate: totalRuns > 0 ? Math.round((successfulRuns / totalRuns) * 100) : 0, - totalCostUsd: totalCost, - avgCostPerTask: completedTasks > 0 ? totalCost / completedTasks : 0, - throughputTasksPerDay: days > 0 ? recentTasksDone / days : 0, - agentMetrics, - }; - }, - - getStandupReport(companyId: string): StandupReport { - const db = getDb(); - const oneDayAgo = new Date(Date.now() - 24 * 60 * 60 * 1000).toISOString(); - - type AgentRow = { - id: string; name: string; role: string; status: StandupReport["agentReports"][number]["status"]; - budgetMonthlyUsd: number; spentMonthlyUsd: number; - }; - const agents = readAll( - `select id, name, role, status, - budget_monthly_usd as budgetMonthlyUsd, - spent_monthly_usd as spentMonthlyUsd - from agents - where company_id = ? and status != 'terminated' and status != 'pending_approval' - order by created_at asc`, - companyId, - ); - - const recentCompletedTasksCount = (db.prepare( - "select count(*) as c from tasks where company_id = ? and status = 'done' and updated_at > ?", - ).get(companyId, oneDayAgo) as { c: number }).c; - - const recentRunStats = db.prepare( - `select - count(*) as total, - sum(case when status = 'succeeded' then 1 else 0 end) as succeeded, - sum(case when status = 'failed' or status = 'timed_out' then 1 else 0 end) as failed - from runs where company_id = ? and created_at > ?`, - ).get(companyId, oneDayAgo) as { total: number; succeeded: number | null; failed: number | null }; - const recentRunsTotal = recentRunStats.total ?? 0; - const successfulRuns = recentRunStats.succeeded ?? 0; - const failedRuns = recentRunStats.failed ?? 0; - - const blockedTaskRows = readAll<{ id: string; title: string; assigneeAgentId: string | null }>( - `select id, title, assignee_agent_id as assigneeAgentId - from tasks where company_id = ? and status = 'blocked'`, - companyId, - ); - - const pendingApprovals = (db.prepare( - "select count(*) as c from approvals where company_id = ? and state = 'pending'", - ).get(companyId) as { c: number }).c; - - const activeSprint = db.prepare( - `select name, velocity_points as velocityPoints, completed_points as completedPoints - from sprints where company_id = ? and status = 'active' order by start_date desc limit 1`, - ).get(companyId) as { name: string; velocityPoints: number; completedPoints: number } | undefined; - - const urgentUnread = (db.prepare( - `select count(*) as c - from agent_messages messages - left join agent_message_reads reads - on reads.message_id = messages.id and reads.agent_id = '__board__' - where messages.company_id = ? - and reads.read_at is null - and (messages.priority = 'urgent' or messages.channel = 'incident')`, - ).get(companyId) as { c: number }).c; - - const tasksByAgentRows = readAll<{ - agentId: string; inProgress: number; blocked: number; completed24h: number; - }>( - `select assignee_agent_id as agentId, - sum(case when status = 'in_progress' then 1 else 0 end) as inProgress, - sum(case when status = 'blocked' then 1 else 0 end) as blocked, - sum(case when status = 'done' and updated_at > ? then 1 else 0 end) as completed24h - from tasks - where company_id = ? and assignee_agent_id is not null - group by assignee_agent_id`, - oneDayAgo, - companyId, - ); - const tasksByAgent = new Map(tasksByAgentRows.map((row) => [row.agentId, row])); - - const recentRunsByAgent = readAll<{ - agentId: string; runCount: number; lastStatus: string | null; lastAt: string | null; - }>( - `select agent_id as agentId, - count(*) as runCount, - (select status from runs r2 where r2.agent_id = runs.agent_id and r2.company_id = runs.company_id and r2.created_at > ? order by created_at desc limit 1) as lastStatus, - max(created_at) as lastAt - from runs - where company_id = ? and created_at > ? - group by agent_id`, - oneDayAgo, - companyId, - oneDayAgo, - ); - const runsByAgent = new Map(recentRunsByAgent.map((row) => [row.agentId, row])); - - const costByAgentRows = readAll<{ agentId: string; cost24h: number }>( - `select agent_id as agentId, coalesce(sum(amount_usd), 0) as cost24h - from cost_entries - where company_id = ? and agent_id is not null and created_at > ? - group by agent_id`, - companyId, - oneDayAgo, - ); - const costByAgent = new Map(costByAgentRows.map((row) => [row.agentId, row.cost24h])); - - const reviewByAgentRows = readAll<{ actor: string; cycles: number }>( - `select actor, count(*) as cycles - from activity - where company_id = ? and created_at > ? - and action in ('task.status_changed', 'task.updated', 'task.auto_assigned') - group by actor`, - companyId, - oneDayAgo, - ); - const reviewByAgent = new Map(reviewByAgentRows.map((row) => [row.actor, row.cycles])); - - const agentNameMap = new Map(agents.map((a) => [a.id, a.name])); - - const companyHighlights: string[] = []; - const blockers: string[] = []; - if (recentCompletedTasksCount > 0) { - companyHighlights.push(`${recentCompletedTasksCount} task(s) completed in last 24h`); - } - if (recentRunsTotal > 0) { - companyHighlights.push( - `${recentRunsTotal} runs (${successfulRuns} succeeded, ${failedRuns} failed)`, - ); - } - for (const task of blockedTaskRows) { - const assigneeName = task.assigneeAgentId ? agentNameMap.get(task.assigneeAgentId) ?? null : null; - blockers.push(`${task.title} — blocked (${assigneeName ?? "unassigned"})`); - } - if (activeSprint) { - const pct = activeSprint.velocityPoints > 0 - ? Math.round((activeSprint.completedPoints / activeSprint.velocityPoints) * 100) - : 0; - companyHighlights.push( - `Sprint "${activeSprint.name}": ${activeSprint.completedPoints}/${activeSprint.velocityPoints} pts (${pct}%)`, - ); - } - if (urgentUnread > 0) { - blockers.push(`${urgentUnread} urgent unread message(s) need attention`); - } - - const totalBudget = agents.reduce((sum, a) => sum + a.budgetMonthlyUsd, 0); - const totalSpent = agents.reduce((sum, a) => sum + a.spentMonthlyUsd, 0); - const budgetUtilization = totalBudget > 0 ? Math.round((totalSpent / totalBudget) * 100) : 0; - - const agentReports = agents.map((agent) => { - const taskAgg = tasksByAgent.get(agent.id); - const runAgg = runsByAgent.get(agent.id); - const completed24h = taskAgg?.completed24h ?? 0; - const runs24h = runAgg?.runCount ?? 0; - const reviewedTasks = reviewByAgent.get(agent.id) ?? 0; - const highlights: string[] = []; - if (completed24h > 0) highlights.push(`Completed ${completed24h} task(s)`); - if (runs24h > 0) highlights.push(`${runs24h} run(s) in last 24h`); - if (reviewedTasks > 0) highlights.push(`${reviewedTasks} status update(s)`); - return { - agentId: agent.id, - agentName: agent.name, - role: agent.role, - status: agent.status, - tasksInProgress: taskAgg?.inProgress ?? 0, - tasksCompleted24h: completed24h, - tasksBlocked: taskAgg?.blocked ?? 0, - lastRunStatus: (runAgg?.lastStatus ?? null) as RunStatus | null, - lastRunAt: runAgg?.lastAt ?? null, - costLast24h: costByAgent.get(agent.id) ?? 0, - highlights, - }; - }); - - const summaryParts: string[] = []; - if (agents.length > 0) summaryParts.push(`${agents.length} active agents`); - summaryParts.push(`${recentCompletedTasksCount} tasks done`); - if (blockedTaskRows.length > 0) summaryParts.push(`${blockedTaskRows.length} blocked`); - if (pendingApprovals > 0) summaryParts.push(`${pendingApprovals} pending approvals`); - - return { - companyId, - generatedAt: new Date().toISOString(), - summary: summaryParts.join(" · "), - agentReports, - companyHighlights, - blockers, - pendingApprovals, - budgetUtilization, - }; - }, - getSidebarBadges(companyId: string): { pendingApprovals: number; activeRuns: number; diff --git a/src/main/database/company.ts b/src/main/database/company.ts index 5e2a6b39..9c165a8c 100644 --- a/src/main/database/company.ts +++ b/src/main/database/company.ts @@ -2,6 +2,7 @@ import type { DatabaseSync } from "node:sqlite"; import type { ProfileSnapshot } from "@shared/types"; import type { ReadAll, Transaction } from "./helpers"; import { buildCompanyCrudMethods } from "./company-crud"; +import { buildCompanyMetricsMethods } from "./company-metrics"; import { buildCompanySettingsMethods } from "./company-settings"; export function buildCompanyMethods( @@ -18,5 +19,6 @@ export function buildCompanyMethods( return { ...buildCompanyCrudMethods(getDb, readAll, transaction, deps), ...buildCompanySettingsMethods(getDb, readAll, transaction, { listSnapshot: deps.listSnapshot }), + ...buildCompanyMetricsMethods(getDb, readAll, transaction), }; } diff --git a/src/main/database/config-revisions.ts b/src/main/database/config-revisions.ts new file mode 100644 index 00000000..487f3666 --- /dev/null +++ b/src/main/database/config-revisions.ts @@ -0,0 +1,126 @@ +import { randomUUID } from "node:crypto"; +import type { DatabaseSync } from "node:sqlite"; +import type { AgentRecord } from "@shared/types"; +import type { ReadAll, SqliteRow, Transaction } from "./helpers"; +import { nowIso, parseJson } from "./helpers"; + +/** + * Agent config-revision helpers extracted from documents.ts. Spread into + * the final `buildDocumentMethods` result so AppDatabase call-sites remain + * unchanged. + */ +export function buildConfigRevisionMethods( + getDb: () => DatabaseSync, + readAll: ReadAll, + transaction: Transaction, +) { + return { + recordConfigRevision(input: { + companyId: string; agentId: string; changedKeys: string[]; + configBefore: Record; configAfter: Record; + source?: string; rolledBackFromRevisionId?: string | null; + createdByAgentId?: string | null; createdByUserId?: string | null; + }): string { + const id = randomUUID(); + const ts = nowIso(); + const lastRev = getDb().prepare( + "SELECT MAX(revision_number) as maxRev FROM agent_config_revisions WHERE agent_id = ?" + ).get(input.agentId) as { maxRev: number | null } | undefined; + const revisionNumber = (lastRev?.maxRev ?? 0) + 1; + getDb().prepare( + "INSERT INTO agent_config_revisions (id, company_id, agent_id, revision_number, changed_keys, config_before, config_after, source, rolled_back_from_revision_id, created_by_agent_id, created_by_user_id, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" + ).run( + id, input.companyId, input.agentId, revisionNumber, + JSON.stringify(input.changedKeys), JSON.stringify(input.configBefore), + JSON.stringify(input.configAfter), input.source ?? "patch", + input.rolledBackFromRevisionId ?? null, input.createdByAgentId ?? null, + input.createdByUserId ?? null, ts + ); + return id; + }, + + listConfigRevisions(agentId: string, companyId: string): Array<{ + id: string; agentId: string; revisionNumber: number; changedKeys: string[]; + source: string; rolledBackFromRevisionId: string | null; + createdByAgentId: string | null; createdByUserId: string | null; createdAt: string; + }> { + return readAll( + "SELECT id, agent_id, revision_number, changed_keys, source, rolled_back_from_revision_id, created_by_agent_id, created_by_user_id, created_at FROM agent_config_revisions WHERE agent_id = ? AND company_id = ? ORDER BY revision_number DESC", + agentId, companyId + ).map((row) => ({ + id: row.id, agentId: row.agent_id, revisionNumber: row.revision_number, + changedKeys: parseJson(row.changed_keys, []), + source: row.source, rolledBackFromRevisionId: row.rolled_back_from_revision_id ?? null, + createdByAgentId: row.created_by_agent_id ?? null, + createdByUserId: row.created_by_user_id ?? null, createdAt: row.created_at, + })); + }, + + getConfigRevision(revisionId: string, companyId: string): { + id: string; agentId: string; revisionNumber: number; changedKeys: string[]; + configBefore: Record; configAfter: Record; + source: string; rolledBackFromRevisionId: string | null; + createdByAgentId: string | null; createdByUserId: string | null; createdAt: string; + } { + const row = getDb().prepare( + "SELECT * FROM agent_config_revisions WHERE id = ? AND company_id = ?" + ).get(revisionId, companyId) as { + id: string; agent_id: string; revision_number: number; changed_keys: string; + config_before: string; config_after: string; source: string; + rolled_back_from_revision_id: string | null; + created_by_agent_id: string | null; created_by_user_id: string | null; created_at: string; + } | undefined; + if (!row) throw new Error(`Config revision ${revisionId} not found.`); + return { + id: row.id, agentId: row.agent_id, revisionNumber: row.revision_number, + changedKeys: parseJson(row.changed_keys, []), + configBefore: parseJson>(row.config_before, {}), + configAfter: parseJson>(row.config_after, {}), + source: row.source, rolledBackFromRevisionId: row.rolled_back_from_revision_id ?? null, + createdByAgentId: row.created_by_agent_id ?? null, + createdByUserId: row.created_by_user_id ?? null, createdAt: row.created_at, + }; + }, + + rollbackConfigRevision(agentId: string, revisionId: string, companyId: string, selfDeps: { + getConfigRevision: (revisionId: string, companyId: string) => { + id: string; agentId: string; revisionNumber: number; changedKeys: string[]; + configBefore: Record; configAfter: Record; + source: string; rolledBackFromRevisionId: string | null; + createdByAgentId: string | null; createdByUserId: string | null; createdAt: string; + }; + getAgent: (id: string) => AgentRecord; + recordConfigRevision: (input: { + companyId: string; agentId: string; changedKeys: string[]; + configBefore: Record; configAfter: Record; + source?: string; rolledBackFromRevisionId?: string | null; + createdByAgentId?: string | null; createdByUserId?: string | null; + }) => string; + }): string { + return transaction(() => { + const revision = selfDeps.getConfigRevision(revisionId, companyId); + if (revision.agentId !== agentId) throw new Error("Revision does not belong to this agent."); + const before = revision.configBefore as Record; + const ts = nowIso(); + getDb().prepare( + "UPDATE agents SET name = ?, role = ?, title = ?, department = ?, capabilities = ?, connector_id = ?, model = ?, budget_monthly_usd = ?, metadata_json = ?, updated_at = ? WHERE id = ? AND company_id = ?" + ).run( + before.name ?? "", before.role ?? "", before.title ?? "", before.department ?? null, + before.capabilities ?? "", before.connectorId ?? before.connector_id ?? "", + before.model ?? null, before.budgetMonthlyUsd ?? before.budget_monthly_usd ?? 0, + before.metadataJson ?? before.metadata_json ?? "{}", ts, agentId, companyId + ); + const agent = selfDeps.getAgent(agentId); + const currentConfig = { name: agent.name, role: agent.role, title: agent.title, department: agent.department, capabilities: agent.capabilities, connectorId: agent.connectorId, model: agent.model, budgetMonthlyUsd: agent.budgetMonthlyUsd, metadataJson: agent.metadataJson }; + return selfDeps.recordConfigRevision({ + companyId, agentId, + changedKeys: revision.changedKeys, + configBefore: currentConfig, + configAfter: before, + source: "rollback", + rolledBackFromRevisionId: revisionId, + }); + }); + }, + }; +} diff --git a/src/main/database/cost-entries.ts b/src/main/database/cost-entries.ts new file mode 100644 index 00000000..934b1e94 --- /dev/null +++ b/src/main/database/cost-entries.ts @@ -0,0 +1,52 @@ +import { randomUUID } from "node:crypto"; +import type { DatabaseSync } from "node:sqlite"; +import { type ReadAll, type Transaction, nowIso } from "./helpers"; + +/** + * cost_entries writer extracted from runs.ts. Spread into the final + * `buildRunMethods` result so AppDatabase call-sites remain unchanged. + * + * Note: finishRun still reads SUM(cost_entries.amount_usd) inline. The + * SUM query stays in runs.ts to keep the run-aggregate read path local to + * the rest of finishRun. + */ +export function buildCostEntryMethods( + getDb: () => DatabaseSync, + _readAll: ReadAll, + _transaction: Transaction, +) { + return { + /** + * Single writer for `cost_entries`. CostWriter calls this per + * `message_end`; tests also call it to simulate per-turn cost. + * finishRun reads SUM(cost_entries) for the run-level aggregate, + * so this is the authoritative path for cost attribution. + */ + recordCostEntry(input: { + runId: string; + companyId: string; + agentId: string; + taskId: string | null; + connectorId: string; + amountUsd: number | null; + unattributed: boolean; + }): string { + const id = randomUUID(); + const ts = nowIso(); + const projectRow = input.taskId + ? getDb().prepare("SELECT project_id FROM tasks WHERE id = ?").get(input.taskId) as { project_id: string | null } | undefined + : undefined; + const projectId = projectRow?.project_id ?? null; + getDb() + .prepare( + "insert into cost_entries (id, company_id, run_id, task_id, agent_id, connector_id, amount_usd, unattributed, project_id, created_at) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ) + .run( + id, input.companyId, input.runId, input.taskId, input.agentId, + input.connectorId, input.amountUsd, + input.unattributed ? 1 : 0, projectId, ts, + ); + return id; + }, + }; +} diff --git a/src/main/database/documents.ts b/src/main/database/documents.ts index e0e53c2e..9f737f4a 100644 --- a/src/main/database/documents.ts +++ b/src/main/database/documents.ts @@ -1,8 +1,8 @@ import { randomUUID } from "node:crypto"; import type { DatabaseSync, SQLInputValue } from "node:sqlite"; -import type { AgentRecord } from "@shared/types"; -import type { ReadAll, SqliteRow, Transaction } from "./helpers"; -import { ensureId, nowIso, parseJson } from "./helpers"; +import type { ReadAll, Transaction } from "./helpers"; +import { ensureId, nowIso } from "./helpers"; +import { buildConfigRevisionMethods } from "./config-revisions"; /** * Document row shape returned by listDocuments — used by @@ -53,6 +53,7 @@ export function buildDocumentMethods( transaction: Transaction, ) { return { + ...buildConfigRevisionMethods(getDb, readAll, transaction), saveDocument(input: { id?: string | null; companyId: string; type: string; title: string; content?: string; authorAgentId?: string | null; reviewerAgentId?: string | null; projectId?: string | null; @@ -313,114 +314,5 @@ export function buildDocumentMethods( })); }, - // ── Config Revisions ── - - recordConfigRevision(input: { - companyId: string; agentId: string; changedKeys: string[]; - configBefore: Record; configAfter: Record; - source?: string; rolledBackFromRevisionId?: string | null; - createdByAgentId?: string | null; createdByUserId?: string | null; - }): string { - const id = randomUUID(); - const ts = nowIso(); - const lastRev = getDb().prepare( - "SELECT MAX(revision_number) as maxRev FROM agent_config_revisions WHERE agent_id = ?" - ).get(input.agentId) as { maxRev: number | null } | undefined; - const revisionNumber = (lastRev?.maxRev ?? 0) + 1; - getDb().prepare( - "INSERT INTO agent_config_revisions (id, company_id, agent_id, revision_number, changed_keys, config_before, config_after, source, rolled_back_from_revision_id, created_by_agent_id, created_by_user_id, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" - ).run( - id, input.companyId, input.agentId, revisionNumber, - JSON.stringify(input.changedKeys), JSON.stringify(input.configBefore), - JSON.stringify(input.configAfter), input.source ?? "patch", - input.rolledBackFromRevisionId ?? null, input.createdByAgentId ?? null, - input.createdByUserId ?? null, ts - ); - return id; - }, - - listConfigRevisions(agentId: string, companyId: string): Array<{ - id: string; agentId: string; revisionNumber: number; changedKeys: string[]; - source: string; rolledBackFromRevisionId: string | null; - createdByAgentId: string | null; createdByUserId: string | null; createdAt: string; - }> { - return readAll( - "SELECT id, agent_id, revision_number, changed_keys, source, rolled_back_from_revision_id, created_by_agent_id, created_by_user_id, created_at FROM agent_config_revisions WHERE agent_id = ? AND company_id = ? ORDER BY revision_number DESC", - agentId, companyId - ).map((row) => ({ - id: row.id, agentId: row.agent_id, revisionNumber: row.revision_number, - changedKeys: parseJson(row.changed_keys, []), - source: row.source, rolledBackFromRevisionId: row.rolled_back_from_revision_id ?? null, - createdByAgentId: row.created_by_agent_id ?? null, - createdByUserId: row.created_by_user_id ?? null, createdAt: row.created_at, - })); - }, - - getConfigRevision(revisionId: string, companyId: string): { - id: string; agentId: string; revisionNumber: number; changedKeys: string[]; - configBefore: Record; configAfter: Record; - source: string; rolledBackFromRevisionId: string | null; - createdByAgentId: string | null; createdByUserId: string | null; createdAt: string; - } { - const row = getDb().prepare( - "SELECT * FROM agent_config_revisions WHERE id = ? AND company_id = ?" - ).get(revisionId, companyId) as { - id: string; agent_id: string; revision_number: number; changed_keys: string; - config_before: string; config_after: string; source: string; - rolled_back_from_revision_id: string | null; - created_by_agent_id: string | null; created_by_user_id: string | null; created_at: string; - } | undefined; - if (!row) throw new Error(`Config revision ${revisionId} not found.`); - return { - id: row.id, agentId: row.agent_id, revisionNumber: row.revision_number, - changedKeys: parseJson(row.changed_keys, []), - configBefore: parseJson>(row.config_before, {}), - configAfter: parseJson>(row.config_after, {}), - source: row.source, rolledBackFromRevisionId: row.rolled_back_from_revision_id ?? null, - createdByAgentId: row.created_by_agent_id ?? null, - createdByUserId: row.created_by_user_id ?? null, createdAt: row.created_at, - }; - }, - - rollbackConfigRevision(agentId: string, revisionId: string, companyId: string, selfDeps: { - getConfigRevision: (revisionId: string, companyId: string) => { - id: string; agentId: string; revisionNumber: number; changedKeys: string[]; - configBefore: Record; configAfter: Record; - source: string; rolledBackFromRevisionId: string | null; - createdByAgentId: string | null; createdByUserId: string | null; createdAt: string; - }; - getAgent: (id: string) => AgentRecord; - recordConfigRevision: (input: { - companyId: string; agentId: string; changedKeys: string[]; - configBefore: Record; configAfter: Record; - source?: string; rolledBackFromRevisionId?: string | null; - createdByAgentId?: string | null; createdByUserId?: string | null; - }) => string; - }): string { - return transaction(() => { - const revision = selfDeps.getConfigRevision(revisionId, companyId); - if (revision.agentId !== agentId) throw new Error("Revision does not belong to this agent."); - const before = revision.configBefore as Record; - const ts = nowIso(); - getDb().prepare( - "UPDATE agents SET name = ?, role = ?, title = ?, department = ?, capabilities = ?, connector_id = ?, model = ?, budget_monthly_usd = ?, metadata_json = ?, updated_at = ? WHERE id = ? AND company_id = ?" - ).run( - before.name ?? "", before.role ?? "", before.title ?? "", before.department ?? null, - before.capabilities ?? "", before.connectorId ?? before.connector_id ?? "", - before.model ?? null, before.budgetMonthlyUsd ?? before.budget_monthly_usd ?? 0, - before.metadataJson ?? before.metadata_json ?? "{}", ts, agentId, companyId - ); - const agent = selfDeps.getAgent(agentId); - const currentConfig = { name: agent.name, role: agent.role, title: agent.title, department: agent.department, capabilities: agent.capabilities, connectorId: agent.connectorId, model: agent.model, budgetMonthlyUsd: agent.budgetMonthlyUsd, metadataJson: agent.metadataJson }; - return selfDeps.recordConfigRevision({ - companyId, agentId, - changedKeys: revision.changedKeys, - configBefore: currentConfig, - configAfter: before, - source: "rollback", - rolledBackFromRevisionId: revisionId, - }); - }); - }, }; } diff --git a/src/main/database/run-events.ts b/src/main/database/run-events.ts new file mode 100644 index 00000000..6e2d6b82 --- /dev/null +++ b/src/main/database/run-events.ts @@ -0,0 +1,82 @@ +import type { DatabaseSync } from "node:sqlite"; +import { type ReadAll, type SqliteRow, type Transaction, nowIso } from "./helpers"; + +/** + * run_events helpers extracted from runs.ts. Every method is spread into + * the final `buildRunMethods` result so AppDatabase call-sites remain + * unchanged. + */ +export function buildRunEventMethods( + getDb: () => DatabaseSync, + readAll: ReadAll, + transaction: Transaction, +) { + return { + appendRunEvent(input: { + companyId: string; runId: string; agentId: string; + eventType: string; stream?: string | null; level?: string | null; + message?: string | null; payload?: Record | null; + }): void { + const ts = nowIso(); + transaction(() => { + const lastSeq = getDb().prepare( + "SELECT MAX(seq) as maxSeq FROM run_events WHERE run_id = ?" + ).get(input.runId) as { maxSeq: number | null } | undefined; + const seq = (lastSeq?.maxSeq ?? 0) + 1; + getDb().prepare( + "INSERT INTO run_events (company_id, run_id, agent_id, seq, event_type, stream, level, message, payload, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" + ).run( + input.companyId, input.runId, input.agentId, seq, + input.eventType, input.stream ?? null, input.level ?? null, + input.message ?? null, input.payload ? JSON.stringify(input.payload) : null, ts + ); + }); + }, + + listRunEvents(runId: string, companyId: string, opts?: { offset?: number; limit?: number }): Array<{ + id: number; runId: string; agentId: string; seq: number; eventType: string; + stream: string | null; level: string | null; message: string | null; + payload: Record | null; createdAt: string; + }> { + const offset = opts?.offset ?? 0; + const limit = opts?.limit ?? 500; + return readAll( + "SELECT id, run_id, agent_id, seq, event_type, stream, level, message, payload, created_at FROM run_events WHERE run_id = ? AND company_id = ? ORDER BY seq ASC LIMIT ? OFFSET ?", + runId, companyId, limit, offset + ).map((row: SqliteRow) => ({ + id: row.id, + runId: row.run_id, + agentId: row.agent_id, + seq: row.seq, + eventType: row.event_type, + stream: row.stream, + level: row.level, + message: row.message, + payload: row.payload ? JSON.parse(row.payload) : null, + createdAt: row.created_at, + })); + }, + + listRunEventsAfter(runId: string, companyId: string, afterSeq: number, limit: number): Array<{ + id: number; runId: string; agentId: string; seq: number; eventType: string; + stream: string | null; level: string | null; message: string | null; + payload: Record | null; createdAt: string; + }> { + return readAll( + "SELECT id, run_id, agent_id, seq, event_type, stream, level, message, payload, created_at FROM run_events WHERE run_id = ? AND company_id = ? AND seq > ? ORDER BY seq ASC LIMIT ?", + runId, companyId, afterSeq, limit + ).map((row: SqliteRow) => ({ + id: row.id, + runId: row.run_id, + agentId: row.agent_id, + seq: row.seq, + eventType: row.event_type, + stream: row.stream, + level: row.level, + message: row.message, + payload: row.payload ? JSON.parse(row.payload) : null, + createdAt: row.created_at, + })); + }, + }; +} diff --git a/src/main/database/run-logs.ts b/src/main/database/run-logs.ts new file mode 100644 index 00000000..f081ece1 --- /dev/null +++ b/src/main/database/run-logs.ts @@ -0,0 +1,96 @@ +import { writeFileSync, readFileSync, existsSync, statSync, openSync, readSync, closeSync } from "node:fs"; +import { appendFile } from "node:fs/promises"; +import type { DatabaseSync } from "node:sqlite"; +import log from "electron-log/main.js"; +import type { RunLogChunk } from "@shared/types"; +import type { ReadAll, Transaction } from "./helpers"; + +/** + * Find the last complete UTF-8 character boundary in a buffer. + * Returns the number of bytes up to (and including) the last complete character. + * This prevents corrupting multi-byte CJK characters when slicing at chunk edges. + */ +export function findUtf8Boundary(buffer: Buffer, bytesRead: number): number { + if (bytesRead === 0) return 0; + let pos = bytesRead - 1; + while (pos >= 0 && (buffer[pos] & 0xc0) === 0x80) { + pos--; + } + if (pos < 0) return bytesRead; + const lead = buffer[pos]; + let charLen: number; + if ((lead & 0x80) === 0) charLen = 1; + else if ((lead & 0xe0) === 0xc0) charLen = 2; + else if ((lead & 0xf0) === 0xe0) charLen = 3; + else if ((lead & 0xf8) === 0xf0) charLen = 4; + else return bytesRead; + if (pos + charLen <= bytesRead) return bytesRead; + return pos; +} + +/** + * Run log file-IO helpers extracted from runs.ts. Every method is spread + * into the final `buildRunMethods` result so AppDatabase call-sites remain + * unchanged. `writeFileSync` is re-exported indirectly through createRun + * (still in runs.ts) via the shared logs directory. + */ +export function buildRunLogMethods( + getDb: () => DatabaseSync, + _readAll: ReadAll, + _transaction: Transaction, +) { + return { + appendRunLog(runId: string, stream: "stdout" | "stderr", chunk: string) { + const run = getDb().prepare("select log_path as logPath from runs where id = ?").get(runId) as { logPath: string | null } | undefined; + if (!run?.logPath) return; + return appendFile(run.logPath, `[${new Date().toISOString()}][${stream}] ${chunk}`, "utf8"); + }, + + readRunLog(runId: string) { + const run = getDb().prepare("select log_path as logPath from runs where id = ?").get(runId) as { logPath: string | null } | undefined; + if (!run?.logPath) return ""; + try { + return readFileSync(run.logPath, "utf8"); + } catch (err) { + log.warn(`[database/runs] readFileSync failed for run ${runId} log ${run.logPath}:`, err instanceof Error ? err.message : String(err)); + return ""; + } + }, + + readRunLogChunk(runId: string, offset = 0, limit = 65_536): RunLogChunk { + const run = getDb().prepare("select log_path as logPath from runs where id = ?").get(runId) as { logPath: string | null } | undefined; + if (!run?.logPath || !existsSync(run.logPath)) { + return { content: "", offset: 0, nextOffset: 0, totalBytes: 0, eof: true }; + } + + const totalBytes = statSync(run.logPath).size; + const safeOffset = Math.max(0, Math.min(offset, totalBytes)); + const safeLimit = Math.max(1_024, Math.min(limit, 262_144)); + const bytesToRead = Math.max(0, Math.min(safeLimit, totalBytes - safeOffset)); + + if (bytesToRead === 0) { + return { content: "", offset: safeOffset, nextOffset: safeOffset, totalBytes, eof: true }; + } + + const fileDescriptor = openSync(run.logPath, "r"); + try { + const buffer = Buffer.allocUnsafe(bytesToRead); + const bytesRead = readSync(fileDescriptor, buffer, 0, bytesToRead, safeOffset); + const safeBytesRead = findUtf8Boundary(buffer, bytesRead); + const nextOffset = safeOffset + safeBytesRead; + return { + content: buffer.subarray(0, safeBytesRead).toString("utf8"), + offset: safeOffset, + nextOffset, + totalBytes, + eof: nextOffset >= totalBytes, + }; + } finally { + closeSync(fileDescriptor); + } + }, + }; +} + +// Re-exported so runs.ts can still use writeFileSync on run creation. +export { writeFileSync }; diff --git a/src/main/database/runs.ts b/src/main/database/runs.ts index 9bf2024f..9ecb8a7a 100644 --- a/src/main/database/runs.ts +++ b/src/main/database/runs.ts @@ -1,34 +1,12 @@ import { randomUUID } from "node:crypto"; -import { writeFileSync, readFileSync, existsSync, statSync, openSync, readSync, closeSync } from "node:fs"; -import { appendFile } from "node:fs/promises"; +import { writeFileSync } from "node:fs"; import { join } from "node:path"; import type { DatabaseSync, SQLInputValue } from "node:sqlite"; -import log from "electron-log/main.js"; -import type { ConnectorId, RunLogChunk, RunRecord } from "@shared/types"; -import { type ReadAll, type SqliteRow, type Transaction, nowIso, toBoolean } from "./helpers"; - -/** - * Find the last complete UTF-8 character boundary in a buffer. - * Returns the number of bytes up to (and including) the last complete character. - * This prevents corrupting multi-byte CJK characters when slicing at chunk edges. - */ -function findUtf8Boundary(buffer: Buffer, bytesRead: number): number { - if (bytesRead === 0) return 0; - let pos = bytesRead - 1; - while (pos >= 0 && (buffer[pos] & 0xc0) === 0x80) { - pos--; - } - if (pos < 0) return bytesRead; - const lead = buffer[pos]; - let charLen: number; - if ((lead & 0x80) === 0) charLen = 1; - else if ((lead & 0xe0) === 0xc0) charLen = 2; - else if ((lead & 0xf0) === 0xe0) charLen = 3; - else if ((lead & 0xf8) === 0xf0) charLen = 4; - else return bytesRead; - if (pos + charLen <= bytesRead) return bytesRead; - return pos; -} +import type { ConnectorId, RunRecord } from "@shared/types"; +import { type ReadAll, type Transaction, nowIso, toBoolean } from "./helpers"; +import { buildRunLogMethods } from "./run-logs"; +import { buildRunEventMethods } from "./run-events"; +import { buildCostEntryMethods } from "./cost-entries"; export function buildRunMethods( getDb: () => DatabaseSync, @@ -39,6 +17,10 @@ export function buildRunMethods( }, ) { return { + ...buildRunLogMethods(getDb, readAll, transaction), + ...buildRunEventMethods(getDb, readAll, transaction), + ...buildCostEntryMethods(getDb, readAll, transaction), + getRun(id: string): RunRecord { const row = getDb() .prepare( @@ -82,56 +64,6 @@ export function buildRunMethods( return { id, logPath }; }, - appendRunLog(runId: string, stream: "stdout" | "stderr", chunk: string) { - const run = getDb().prepare("select log_path as logPath from runs where id = ?").get(runId) as { logPath: string | null } | undefined; - if (!run?.logPath) return; - return appendFile(run.logPath, `[${new Date().toISOString()}][${stream}] ${chunk}`, "utf8"); - }, - - readRunLog(runId: string) { - const run = getDb().prepare("select log_path as logPath from runs where id = ?").get(runId) as { logPath: string | null } | undefined; - if (!run?.logPath) return ""; - try { - return readFileSync(run.logPath, "utf8"); - } catch (err) { - log.warn(`[database/runs] readFileSync failed for run ${runId} log ${run.logPath}:`, err instanceof Error ? err.message : String(err)); - return ""; - } - }, - - readRunLogChunk(runId: string, offset = 0, limit = 65_536): RunLogChunk { - const run = getDb().prepare("select log_path as logPath from runs where id = ?").get(runId) as { logPath: string | null } | undefined; - if (!run?.logPath || !existsSync(run.logPath)) { - return { content: "", offset: 0, nextOffset: 0, totalBytes: 0, eof: true }; - } - - const totalBytes = statSync(run.logPath).size; - const safeOffset = Math.max(0, Math.min(offset, totalBytes)); - const safeLimit = Math.max(1_024, Math.min(limit, 262_144)); - const bytesToRead = Math.max(0, Math.min(safeLimit, totalBytes - safeOffset)); - - if (bytesToRead === 0) { - return { content: "", offset: safeOffset, nextOffset: safeOffset, totalBytes, eof: true }; - } - - const fileDescriptor = openSync(run.logPath, "r"); - try { - const buffer = Buffer.allocUnsafe(bytesToRead); - const bytesRead = readSync(fileDescriptor, buffer, 0, bytesToRead, safeOffset); - const safeBytesRead = findUtf8Boundary(buffer, bytesRead); - const nextOffset = safeOffset + safeBytesRead; - return { - content: buffer.subarray(0, safeBytesRead).toString("utf8"), - offset: safeOffset, - nextOffset, - totalBytes, - eof: nextOffset >= totalBytes, - }; - } finally { - closeSync(fileDescriptor); - } - }, - markRunStatus(runId: string, status: RunRecord["status"], message: string | null) { const ts = nowIso(); const startedAt = status === "running" ? ts : null; @@ -270,39 +202,6 @@ export function buildRunMethods( return Boolean(row?.present); }, - /** - * Single writer for `cost_entries`. CostWriter calls this per - * `message_end`; tests also call it to simulate per-turn cost. - * finishRun reads SUM(cost_entries) for the run-level aggregate, - * so this is the authoritative path for cost attribution. - */ - recordCostEntry(input: { - runId: string; - companyId: string; - agentId: string; - taskId: string | null; - connectorId: string; - amountUsd: number | null; - unattributed: boolean; - }): string { - const id = randomUUID(); - const ts = nowIso(); - const projectRow = input.taskId - ? getDb().prepare("SELECT project_id FROM tasks WHERE id = ?").get(input.taskId) as { project_id: string | null } | undefined - : undefined; - const projectId = projectRow?.project_id ?? null; - getDb() - .prepare( - "insert into cost_entries (id, company_id, run_id, task_id, agent_id, connector_id, amount_usd, unattributed, project_id, created_at) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", - ) - .run( - id, input.companyId, input.runId, input.taskId, input.agentId, - input.connectorId, input.amountUsd, - input.unattributed ? 1 : 0, projectId, ts, - ); - return id; - }, - listQueuedRuns() { return readAll( `select id, company_id as companyId, task_id as taskId, agent_id as agentId, workspace_id as workspaceId, @@ -337,73 +236,6 @@ export function buildRunMethods( ); }, - appendRunEvent(input: { - companyId: string; runId: string; agentId: string; - eventType: string; stream?: string | null; level?: string | null; - message?: string | null; payload?: Record | null; - }): void { - const ts = nowIso(); - transaction(() => { - const lastSeq = getDb().prepare( - "SELECT MAX(seq) as maxSeq FROM run_events WHERE run_id = ?" - ).get(input.runId) as { maxSeq: number | null } | undefined; - const seq = (lastSeq?.maxSeq ?? 0) + 1; - getDb().prepare( - "INSERT INTO run_events (company_id, run_id, agent_id, seq, event_type, stream, level, message, payload, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" - ).run( - input.companyId, input.runId, input.agentId, seq, - input.eventType, input.stream ?? null, input.level ?? null, - input.message ?? null, input.payload ? JSON.stringify(input.payload) : null, ts - ); - }); - }, - - listRunEvents(runId: string, companyId: string, opts?: { offset?: number; limit?: number }): Array<{ - id: number; runId: string; agentId: string; seq: number; eventType: string; - stream: string | null; level: string | null; message: string | null; - payload: Record | null; createdAt: string; - }> { - const offset = opts?.offset ?? 0; - const limit = opts?.limit ?? 500; - return readAll( - "SELECT id, run_id, agent_id, seq, event_type, stream, level, message, payload, created_at FROM run_events WHERE run_id = ? AND company_id = ? ORDER BY seq ASC LIMIT ? OFFSET ?", - runId, companyId, limit, offset - ).map((row: SqliteRow) => ({ - id: row.id, - runId: row.run_id, - agentId: row.agent_id, - seq: row.seq, - eventType: row.event_type, - stream: row.stream, - level: row.level, - message: row.message, - payload: row.payload ? JSON.parse(row.payload) : null, - createdAt: row.created_at, - })); - }, - - listRunEventsAfter(runId: string, companyId: string, afterSeq: number, limit: number): Array<{ - id: number; runId: string; agentId: string; seq: number; eventType: string; - stream: string | null; level: string | null; message: string | null; - payload: Record | null; createdAt: string; - }> { - return readAll( - "SELECT id, run_id, agent_id, seq, event_type, stream, level, message, payload, created_at FROM run_events WHERE run_id = ? AND company_id = ? AND seq > ? ORDER BY seq ASC LIMIT ?", - runId, companyId, afterSeq, limit - ).map((row: SqliteRow) => ({ - id: row.id, - runId: row.run_id, - agentId: row.agent_id, - seq: row.seq, - eventType: row.event_type, - stream: row.stream, - level: row.level, - message: row.message, - payload: row.payload ? JSON.parse(row.payload) : null, - createdAt: row.created_at, - })); - }, - updateRunActor(runId: string, actorType: string, actorId: string | null): void { getDb().prepare( "UPDATE runs SET actor_type = ?, actor_id = ?, updated_at = ? WHERE id = ?" diff --git a/src/main/database/task-queries.ts b/src/main/database/task-queries.ts new file mode 100644 index 00000000..4a57edb1 --- /dev/null +++ b/src/main/database/task-queries.ts @@ -0,0 +1,177 @@ +import type { DatabaseSync } from "node:sqlite"; +import type { TaskRecord } from "@shared/types"; +import { type ReadAll, type Transaction, taskPriorityOrderSql } from "./helpers"; +import type { TaskListByStatusesRow } from "./tasks-crud"; + +/** + * Task list/aggregate read queries extracted from tasks-crud.ts. + * + * Every method is spread into the final `buildTaskMethods` result so + * AppDatabase call-sites remain unchanged. + */ +export function buildTaskQueryMethods( + getDb: () => DatabaseSync, + readAll: ReadAll, + _transaction: Transaction, +) { + return { + /** + * Returns actionable real work for an agent. + */ + listAgentTasks( + agentId: string, + companyId: string, + statuses: string[], + ): TaskRecord[] { + const placeholders = statuses.map(() => "?").join(","); + return readAll( + `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 + from tasks + where company_id = ? and assignee_agent_id = ? and status in (${placeholders}) + order by + case status + when 'in_progress' then 0 + when 'todo' then 1 + when 'blocked' then 2 + when 'backlog' then 3 + else 4 + end asc, + ${taskPriorityOrderSql("priority")} desc, + created_at asc`, + companyId, + agentId, + ...statuses, + ); + }, + + /** + * Returns the slim task projection (id, title, status, priority) for a + * single assignee, ordered by created_at desc. + */ + listAssigneeTasksForStandup( + companyId: string, + agentId: string, + ): Array<{ id: string; title: string; status: string; priority: string }> { + return readAll( + "select id, title, status, priority from tasks where company_id = ? and assignee_agent_id = ? order by created_at desc", + companyId, + agentId, + ); + }, + + /** + * Returns the slim task projection for a set of assignees. Safe on + * an empty assignee list (returns []). + */ + listTasksForAssignees( + companyId: string, + assigneeIds: string[], + ): Array<{ + id: string; + title: string; + status: string; + priority: string; + assigneeAgentId: string; + }> { + if (assigneeIds.length === 0) return []; + const placeholders = assigneeIds.map(() => "?").join(","); + return readAll( + `select id, title, status, priority, assignee_agent_id as assigneeAgentId from tasks where company_id = ? and assignee_agent_id in (${placeholders}) order by created_at desc`, + companyId, + ...assigneeIds, + ); + }, + + /** + * Returns the slim task projection used by the me/standup aggregation. + * Ordered by created_at desc. + */ + listTasksForStandup(companyId: string): Array<{ + id: string; + companyId: string; + title: string; + assigneeAgentId: string | null; + status: string; + priority: string; + }> { + return readAll( + "select id, company_id as companyId, title, assignee_agent_id as assigneeAgentId, status, priority from tasks where company_id = ? order by created_at desc", + companyId, + ); + }, + + /** + * Returns tasks filtered by status. + */ + listTasksByStatuses( + companyId: string, + statuses: string[], + ): TaskListByStatusesRow[] { + const placeholders = statuses.map(() => "?").join(", "); + return readAll( + `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`, + companyId, + ...statuses, + ).map((t) => ({ + ...t, + requiresUserReview: Boolean(t.requiresUserReview), + startedAt: t.startedAt ?? null, + completedAt: t.completedAt ?? null, + cancelledAt: t.cancelledAt ?? null, + })) as TaskListByStatusesRow[]; + }, + + countActiveTasksForCompany(companyId: string): number { + const row = getDb() + .prepare( + "select count(*) as cnt from tasks where company_id = ? and status not in ('done', 'cancelled')", + ) + .get(companyId) as { cnt: number }; + return row.cnt; + }, + + listChildTasks( + parentId: string, + companyId: string, + ): Array<{ id: string; status: string; assigneeAgentId: string | null }> { + return readAll<{ id: string; status: string; assigneeAgentId: string | null }>( + `select id, status, assignee_agent_id as assigneeAgentId from tasks where parent_id = ? and company_id = ?`, + parentId, + companyId, + ); + }, + + /** + * Returns the set of distinct assignee agent ids on tasks in a project, + * excluding unassigned tasks. Used by message-dispatch to fan out wake + * targets for a project channel. + */ + listProjectTaskAssigneeIds(projectId: string, companyId: string): string[] { + const rows = readAll<{ assigneeAgentId: string | null }>( + "select distinct assignee_agent_id as assigneeAgentId from tasks where project_id = ? and company_id = ? and assignee_agent_id is not null", + projectId, + companyId, + ); + return rows.map((r) => r.assigneeAgentId).filter((id): id is string => id !== null); + }, + + /** + * Returns assignee agent ids for a set of task ids scoped to a company. + * Used by task-quality-gates sweep to find reviewers of in-review tasks + * that have exceeded the review timeout. + */ + listReviewTasksForSweep(companyId: string): Array<{ + id: string; + assigneeAgentId: string | null; + updatedAt: string; + title: string; + }> { + return readAll( + `SELECT id, assignee_agent_id as assigneeAgentId, updated_at as updatedAt, title + FROM tasks + WHERE company_id = ? AND status = 'in_review'`, + companyId, + ); + }, + }; +} diff --git a/src/main/database/task-sessions-crud.ts b/src/main/database/task-sessions-crud.ts new file mode 100644 index 00000000..c963a1b7 --- /dev/null +++ b/src/main/database/task-sessions-crud.ts @@ -0,0 +1,252 @@ +import { randomUUID } from "node:crypto"; +import type { DatabaseSync } from "node:sqlite"; +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. + */ +export function buildTaskSessionMethods( + getDb: () => DatabaseSync, + _readAll: ReadAll, + transaction: Transaction, +) { + return { + updateTaskSession( + taskId: string, + sessionJson: Record | 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(); + 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 (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .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 | 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 | null = null; + if (row?.sessionParamsJson) { + sessionParams = JSON.parse(row.sessionParamsJson) as Record; + } + 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 + ); + }, + + incrementSessionUsage( + sessionId: string, + inputTokens: number, + outputTokens: number, + runId: string, + ): void { + const ts = nowIso(); + getDb() + .prepare( + `UPDATE task_sessions + SET run_count = run_count + 1, + total_input_tokens = total_input_tokens + ?, + total_output_tokens = total_output_tokens + ?, + last_run_id = ?, + updated_at = ? + WHERE id = ?`, + ) + .run(inputTokens, outputTokens, runId, ts, sessionId); + }, + + archiveSession(sessionId: string, reason: string, handoffMarkdown: string): void { + const ts = nowIso(); + getDb() + .prepare( + `UPDATE task_sessions SET rotated_at = ?, rotation_reason = ?, handoff_markdown = ?, updated_at = ? WHERE id = ?`, + ) + .run(ts, reason, handoffMarkdown, ts, sessionId); + }, + + /** + * 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. + */ + ensureAgentSession(input: { + agentId: string; + companyId: string; + adapterType: string; + }): string { + const agentRow = getDb() + .prepare("SELECT session_id FROM agents WHERE id = ?") + .get(input.agentId) as { session_id: string | null } | undefined; + if (agentRow?.session_id) return agentRow.session_id; + + 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, + 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(sessionId, input.companyId, input.agentId, input.adapterType, ts, ts); + getDb() + .prepare("UPDATE agents SET session_id = ?, updated_at = ? WHERE id = ?") + .run(sessionId, ts, input.agentId); + }); + return sessionId; + }, + + setAgentSessionId(agentId: string, sessionId: string): void { + const ts = nowIso(); + getDb() + .prepare("UPDATE agents SET session_id = ?, updated_at = ? WHERE id = ?") + .run(sessionId, ts, agentId); + }, + + /** + * 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. + */ + 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; + sessionParamsJson: string | null; + sessionDisplayId: string | null; + }): string { + const sessionId = randomUUID(); + const ts = nowIso(); + 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 (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .run( + sessionId, + input.companyId, + input.taskId, + input.agentId, + input.adapterType, + input.sessionParamsJson, + input.sessionDisplayId, + ts, + ts, + ); + getDb() + .prepare("UPDATE tasks SET active_session_id = ?, updated_at = ? WHERE id = ?") + .run(sessionId, ts, input.taskId); + return sessionId; + }, + }; +} diff --git a/src/main/database/tasks-crud.ts b/src/main/database/tasks-crud.ts index 80ceede9..1bffddb4 100644 --- a/src/main/database/tasks-crud.ts +++ b/src/main/database/tasks-crud.ts @@ -1,13 +1,10 @@ -import { randomUUID } from "node:crypto"; import type { DatabaseSync } from "node:sqlite"; -import type { TaskRecord, TaskSessionRecord } from "@shared/types"; import { type ReadAll, type Transaction, ensureId, normalizeOptionalId, nowIso, - taskPriorityOrderSql, } from "./helpers"; /** @@ -168,360 +165,6 @@ export function buildTaskCrudMethods( }); }, - /** - * Returns actionable real work for an agent. - */ - listAgentTasks( - agentId: string, - companyId: string, - statuses: string[], - ): TaskRecord[] { - const placeholders = statuses.map(() => "?").join(","); - return readAll( - `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 - from tasks - where company_id = ? and assignee_agent_id = ? and status in (${placeholders}) - order by - case status - when 'in_progress' then 0 - when 'todo' then 1 - when 'blocked' then 2 - when 'backlog' then 3 - else 4 - end asc, - ${taskPriorityOrderSql("priority")} desc, - created_at asc`, - companyId, - agentId, - ...statuses, - ); - }, - - /** - * Returns the slim task projection (id, title, status, priority) for a - * single assignee, ordered by created_at desc. - */ - listAssigneeTasksForStandup( - companyId: string, - agentId: string, - ): Array<{ id: string; title: string; status: string; priority: string }> { - return readAll( - "select id, title, status, priority from tasks where company_id = ? and assignee_agent_id = ? order by created_at desc", - companyId, - agentId, - ); - }, - - /** - * Returns the slim task projection for a set of assignees. Safe on - * an empty assignee list (returns []). - */ - listTasksForAssignees( - companyId: string, - assigneeIds: string[], - ): Array<{ - id: string; - title: string; - status: string; - priority: string; - assigneeAgentId: string; - }> { - if (assigneeIds.length === 0) return []; - const placeholders = assigneeIds.map(() => "?").join(","); - return readAll( - `select id, title, status, priority, assignee_agent_id as assigneeAgentId from tasks where company_id = ? and assignee_agent_id in (${placeholders}) order by created_at desc`, - companyId, - ...assigneeIds, - ); - }, - - /** - * Returns the slim task projection used by the me/standup aggregation. - * Ordered by created_at desc. - */ - listTasksForStandup(companyId: string): Array<{ - id: string; - companyId: string; - title: string; - assigneeAgentId: string | null; - status: string; - priority: string; - }> { - return readAll( - "select id, company_id as companyId, title, assignee_agent_id as assigneeAgentId, status, priority from tasks where company_id = ? order by created_at desc", - companyId, - ); - }, - - /** - * Returns tasks filtered by status. - */ - listTasksByStatuses( - companyId: string, - statuses: string[], - ): TaskListByStatusesRow[] { - const placeholders = statuses.map(() => "?").join(", "); - return readAll( - `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`, - companyId, - ...statuses, - ).map((t) => ({ - ...t, - requiresUserReview: Boolean(t.requiresUserReview), - startedAt: t.startedAt ?? null, - completedAt: t.completedAt ?? null, - cancelledAt: t.cancelledAt ?? null, - })) as TaskListByStatusesRow[]; - }, - - countActiveTasksForCompany(companyId: string): number { - const row = getDb() - .prepare( - "select count(*) as cnt from tasks where company_id = ? and status not in ('done', 'cancelled')", - ) - .get(companyId) as { cnt: number }; - return row.cnt; - }, - - listChildTasks( - parentId: string, - companyId: string, - ): Array<{ id: string; status: string; assigneeAgentId: string | null }> { - return readAll<{ id: string; status: string; assigneeAgentId: string | null }>( - `select id, status, assignee_agent_id as assigneeAgentId from tasks where parent_id = ? and company_id = ?`, - parentId, - companyId, - ); - }, - - updateTaskSession( - taskId: string, - sessionJson: Record | 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(); - 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 (?, ?, ?, ?, ?, ?, ?, ?, ?)`, - ) - .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 | 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 | null = null; - if (row?.sessionParamsJson) { - sessionParams = JSON.parse(row.sessionParamsJson) as Record; - } - 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 - ); - }, - - incrementSessionUsage( - sessionId: string, - inputTokens: number, - outputTokens: number, - runId: string, - ): void { - const ts = nowIso(); - getDb() - .prepare( - `UPDATE task_sessions - SET run_count = run_count + 1, - total_input_tokens = total_input_tokens + ?, - total_output_tokens = total_output_tokens + ?, - last_run_id = ?, - updated_at = ? - WHERE id = ?`, - ) - .run(inputTokens, outputTokens, runId, ts, sessionId); - }, - - archiveSession(sessionId: string, reason: string, handoffMarkdown: string): void { - const ts = nowIso(); - getDb() - .prepare( - `UPDATE task_sessions SET rotated_at = ?, rotation_reason = ?, handoff_markdown = ?, updated_at = ? WHERE id = ?`, - ) - .run(ts, reason, handoffMarkdown, ts, sessionId); - }, - - /** - * 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. - */ - ensureAgentSession(input: { - agentId: string; - companyId: string; - adapterType: string; - }): string { - const agentRow = getDb() - .prepare("SELECT session_id FROM agents WHERE id = ?") - .get(input.agentId) as { session_id: string | null } | undefined; - if (agentRow?.session_id) return agentRow.session_id; - - 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, - 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(sessionId, input.companyId, input.agentId, input.adapterType, ts, ts); - getDb() - .prepare("UPDATE agents SET session_id = ?, updated_at = ? WHERE id = ?") - .run(sessionId, ts, input.agentId); - }); - return sessionId; - }, - - setAgentSessionId(agentId: string, sessionId: string): void { - const ts = nowIso(); - getDb() - .prepare("UPDATE agents SET session_id = ?, updated_at = ? WHERE id = ?") - .run(sessionId, ts, agentId); - }, - - /** - * 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. - */ - 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; - sessionParamsJson: string | null; - sessionDisplayId: string | null; - }): string { - const sessionId = randomUUID(); - const ts = nowIso(); - 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 (?, ?, ?, ?, ?, ?, ?, ?, ?)`, - ) - .run( - sessionId, - input.companyId, - input.taskId, - input.agentId, - input.adapterType, - input.sessionParamsJson, - input.sessionDisplayId, - ts, - ts, - ); - getDb() - .prepare("UPDATE tasks SET active_session_id = ?, updated_at = ? WHERE id = ?") - .run(sessionId, ts, input.taskId); - return sessionId; - }, - getRecentRunsForTask( taskId: string, limit: number, @@ -591,39 +234,6 @@ export function buildTaskCrudMethods( .run(ts, taskId); }, - /** - * Returns the set of distinct assignee agent ids on tasks in a project, - * excluding unassigned tasks. Used by message-dispatch to fan out wake - * targets for a project channel. - */ - listProjectTaskAssigneeIds(projectId: string, companyId: string): string[] { - const rows = readAll<{ assigneeAgentId: string | null }>( - "select distinct assignee_agent_id as assigneeAgentId from tasks where project_id = ? and company_id = ? and assignee_agent_id is not null", - projectId, - companyId, - ); - return rows.map((r) => r.assigneeAgentId).filter((id): id is string => id !== null); - }, - - /** - * Returns assignee agent ids for a set of task ids scoped to a company. - * Used by task-quality-gates sweep to find reviewers of in-review tasks - * that have exceeded the review timeout. - */ - listReviewTasksForSweep(companyId: string): Array<{ - id: string; - assigneeAgentId: string | null; - updatedAt: string; - title: string; - }> { - return readAll( - `SELECT id, assignee_agent_id as assigneeAgentId, updated_at as updatedAt, title - FROM tasks - WHERE company_id = ? AND status = 'in_review'`, - companyId, - ); - }, - /** * Returns `task_type` for a single task id, or null when the task is * missing. Used by checkpoint-engine to branch on task type without diff --git a/src/main/database/tasks.ts b/src/main/database/tasks.ts index 9b96954d..4f0b2b45 100644 --- a/src/main/database/tasks.ts +++ b/src/main/database/tasks.ts @@ -1,6 +1,8 @@ import type { DatabaseSync } from "node:sqlite"; import type { ReadAll, Transaction } from "./helpers"; import { buildTaskCrudMethods } from "./tasks-crud"; +import { buildTaskQueryMethods } from "./task-queries"; +import { buildTaskSessionMethods } from "./task-sessions-crud"; import { buildTaskStatusMethods } from "./tasks-status"; export function buildTaskMethods( @@ -14,6 +16,8 @@ export function buildTaskMethods( ) { return { ...buildTaskCrudMethods(getDb, readAll, transaction, deps), + ...buildTaskQueryMethods(getDb, readAll, transaction), + ...buildTaskSessionMethods(getDb, readAll, transaction), ...buildTaskStatusMethods(getDb, readAll, transaction, deps), }; }