diff --git a/src/main/services/snapshot-service.ts b/src/main/services/snapshot-service.ts index a4fa041c..628641c7 100644 --- a/src/main/services/snapshot-service.ts +++ b/src/main/services/snapshot-service.ts @@ -1,6 +1,6 @@ import { loadViewState } from "./view-state"; import { loadOnboardingDraft } from "./onboarding-progress"; -import type { DatabaseSync, StatementSync } from "node:sqlite"; +import type { DatabaseSync, StatementSync, SQLInputValue } from "node:sqlite"; import type { AppSnapshot } from "../../shared/domain"; import { PROVIDER_CATALOG } from "../../shared/provider-catalog"; import { IntegrationKeyStore } from "./integration-key-store"; @@ -17,8 +17,8 @@ type ProviderUsageWindowRow = Pick< | "costCents" | "inputTokens" | "outputTokens" | "requestsWithoutUsage" | "unpricedRequests" >; -function all(statement: StatementSync): T[] { - return statement.all() as T[]; +function all(statement: StatementSync, ...params: SQLInputValue[]): T[] { + return statement.all(...params) as T[]; } function toBool(value: unknown): boolean { @@ -37,8 +37,14 @@ function list(db: DatabaseSync, table: string, orderBy: string): T[] { return all(db.prepare(`select * from ${table} order by ${orderBy}`)).map((row) => camel(row) as T); } -function recentList(db: DatabaseSync, table: string, orderBy: string, limit: number): T[] { - return all(db.prepare(`select * from ${table} order by ${orderBy} limit ${limit}`)).map((row) => camel(row) as T); +/** The newest rows of the shown company only, so another company's activity never pushes its history out of the window. */ +function recentInOrg(db: DatabaseSync, table: string, orderBy: string, limit: number, orgId: string | null): T[] { + return all(db.prepare(`select * from ${table} where org_id = ? order by ${orderBy} limit ${limit}`), orgId ?? "").map((row) => camel(row) as T); +} + +/** Exactly the rows of the loaded messages, so every shown message has its attachments, mentions and commands. */ +function forMessages(db: DatabaseSync, table: string, orderBy: string, messageIds: string[]): T[] { + return all(db.prepare(`select * from ${table} where message_id in (select value from json_each(?)) order by ${orderBy}`), JSON.stringify(messageIds)).map((row) => camel(row) as T); } export class SnapshotService { @@ -55,7 +61,8 @@ export class SnapshotService { const agents = list(this.db, "agents", "created_at asc") .map((agent) => ({ ...agent, heartbeatEnabled: toBool(agent.heartbeatEnabled) })); const holds = taskHolds(this.db, new Date()); - const chatMessages = recentList(this.db, "chat_messages", "created_at desc, rowid desc", 500).reverse(); + const chatMessages = recentInOrg(this.db, "chat_messages", "created_at desc, rowid desc", 500, activeOrgId).reverse(); + const loadedMessageIds = chatMessages.map((message) => message.id); const tasks = list(this.db, "tasks", "created_at asc") .map((task) => ({ ...task, requiresFilesystem: toBool(task.requiresFilesystem), hold: holds.get(task.id) ?? null })); const workspaces = list(this.db, "workspaces", "created_at asc") @@ -99,7 +106,7 @@ export class SnapshotService { workspaces, goals: list(this.db, "goals", "created_at asc"), projects: list(this.db, "projects", "created_at asc"), - taskRuns: recentList(this.db, "task_runs", "created_at desc, rowid desc", 200).reverse(), + taskRuns: recentInOrg(this.db, "task_runs", "created_at desc, rowid desc", 200, activeOrgId).reverse(), approvals: list(this.db, "approvals", "created_at desc"), chatSpaces: list(this.db, "chat_spaces", "created_at asc"), chatMemberships: list(this.db, "chat_memberships", "space_id asc, principal_type asc, principal_id asc"), @@ -113,13 +120,13 @@ export class SnapshotService { // Every thread a loaded message belongs to, plus the most recently active ones, so no shown message loses its thread. chatThreads: all(this.db.prepare(` select * from chat_threads - where id in (select thread_id from (select thread_id from chat_messages order by created_at desc, rowid desc limit 500)) - or id in (select id from chat_threads order by updated_at desc limit 200) + where id in (select thread_id from chat_messages where id in (select value from json_each(?))) + or id in (select id from chat_threads where org_id = ? order by updated_at desc limit 200) order by updated_at desc - `)).map((row) => camel(row) as unknown as AppSnapshot["chatThreads"][number]), - chatAttachments: recentList(this.db, "chat_attachments", "message_id desc", 500), - chatMentions: recentList(this.db, "chat_mentions", "message_id desc, offset_start asc", 500), - chatSlashInvocations: recentList(this.db, "chat_slash_invocations", "created_at desc", 200), + `), JSON.stringify(loadedMessageIds), activeOrgId ?? "").map((row) => camel(row) as unknown as AppSnapshot["chatThreads"][number]), + chatAttachments: forMessages(this.db, "chat_attachments", "message_id desc", loadedMessageIds), + chatMentions: forMessages(this.db, "chat_mentions", "message_id desc, offset_start asc", loadedMessageIds), + chatSlashInvocations: forMessages(this.db, "chat_slash_invocations", "created_at desc", loadedMessageIds), boardInterventions: list(this.db, "board_interventions", "created_at asc"), chatPresence: list(this.db, "chat_presence", "last_seen_at desc"), chatReadStates: list(this.db, "chat_read_state", "last_read_at desc"), @@ -139,8 +146,8 @@ export class SnapshotService { execution_mode, policy_decision, external_response_id, stream_event_count, cancelled_at, created_at, completed_at, agent_id, task_run_id, input_tokens, output_tokens, cached_input_tokens, reasoning_tokens, cost_cents - from provider_requests order by created_at desc limit 100 - `)).map((row) => camel(row) as unknown as AppSnapshot["providerRequests"][number]), + from provider_requests where org_id = ? order by created_at desc limit 100 + `), activeOrgId ?? "").map((row) => camel(row) as unknown as AppSnapshot["providerRequests"][number]), providerRateLimits, usageStats: buildUsageStats(this.db, new Date()), providerRateLimitObservations, @@ -162,7 +169,7 @@ export class SnapshotService { employeeIdentities, serviceIdentities: list(this.db, "service_identities", "created_at desc"), roleGrants: list(this.db, "role_grants", "created_at desc"), - accessDecisions: recentList(this.db, "access_decisions", "created_at desc", 200), + accessDecisions: recentInOrg(this.db, "access_decisions", "created_at desc", 200, activeOrgId), orgDepartments: list(this.db, "org_departments", "name asc"), hiringPolicies: list(this.db, "hiring_policies", "department_name asc"), approvalRoutingRules: list(this.db, "approval_routing_rules", "created_at asc"), @@ -170,7 +177,7 @@ export class SnapshotService { featureFlags, sloTargets: list(this.db, "slo_targets", "signal_key asc"), operationalEvents, - auditEvents: recentList(this.db, "audit_events", "created_at desc", 200), + auditEvents: recentInOrg(this.db, "audit_events", "created_at desc", 200, activeOrgId), documents: list(this.db, "documents", "updated_at desc"), knowledgeEntries: list(this.db, "knowledge_entries", "updated_at desc"), automationRules: list(this.db, "automation_rules", "priority asc") diff --git a/tests/unit/snapshot-org-windows.test.ts b/tests/unit/snapshot-org-windows.test.ts new file mode 100644 index 00000000..8d4a4bb8 --- /dev/null +++ b/tests/unit/snapshot-org-windows.test.ts @@ -0,0 +1,57 @@ +import { randomUUID } from "node:crypto"; +import type { DatabaseSync } from "node:sqlite"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { describe, expect, test } from "vitest"; +import { DatabaseConnection } from "../../src/main/database/connection"; +import { OrganizationService } from "../../src/main/services/organization-service"; +import { ChatService } from "../../src/main/services/chat-service"; +import { SnapshotService } from "../../src/main/services/snapshot-service"; +import { RISK_LEVEL, TASK_PRIORITY, TASK_STATUS } from "../../src/shared/domain"; + +/** Copies a row many times with new ids and later times, as a busy company would produce. */ +function copies(db: DatabaseSync, table: string, id: string, count: number): void { + const columns = (db.prepare(`pragma table_info(${table})`).all() as Array<{ name: string }>).map((column) => column.name); + const insert = db.prepare(`insert into ${table}(${columns.join(", ")}) select ${columns.map((column) => (column === "id" ? "?" : column === "created_at" ? "?" : column)).join(", ")} from ${table} where id = ?`); + for (let index = 0; index < count; index += 1) { + insert.run(`${id}_copy_${index}`, new Date(Date.now() + 60_000 + index).toISOString(), id); + } +} + +describe("snapshot history windows", () => { + test("are cut within the shown company, so another company's activity never pushes its history out", () => { + const connection = new DatabaseConnection(join(tmpdir(), `ac-org-windows-${randomUUID()}.db`)); + const db = connection.db; + const organization = new OrganizationService(db); + const chat = new ChatService(db); + const quiet = { orgId: "", employeeId: "", spaceId: "", messageId: "", taskId: "" }; + const busy = { orgId: "", employeeId: "", spaceId: "", messageId: "", taskId: "" }; + for (const company of [quiet, busy]) { + connection.transaction(() => { + company.orgId = organization.createOrganization({ organizationName: `Co ${randomUUID().slice(0, 4)}`, employeeName: "Dana", mission: "m", goal: "g" }); + company.employeeId = (db.prepare("select id from employees where org_id = ?").get(company.orgId) as { id: string }).id; + company.spaceId = (db.prepare("select id from chat_spaces where org_id = ? and name = 'general'").get(company.orgId) as { id: string }).id; + company.messageId = chat.sendMessage({ orgId: company.orgId, spaceId: company.spaceId, threadId: null, senderType: "employee", senderId: company.employeeId, bodyMd: "status update", attachmentFileIds: [] }); + company.taskId = organization.createTask({ + orgId: company.orgId, title: "Work", description: "d", status: TASK_STATUS.BACKLOG, assigneeAgentId: null, risk: RISK_LEVEL.LOW, + projectId: null, goalId: null, workspaceId: null, priority: TASK_PRIORITY.NORMAL, requiresFilesystem: false, + }); + }); + } + db.prepare("insert into task_runs(id, org_id, task_id, trigger, state, created_at) values ('run_quiet', ?, ?, 'manual', 'completed', ?)").run(quiet.orgId, quiet.taskId, new Date().toISOString()); + db.prepare("insert into task_runs(id, org_id, task_id, trigger, state, created_at) values ('run_busy', ?, ?, 'manual', 'completed', ?)").run(busy.orgId, busy.taskId, new Date().toISOString()); + db.prepare("insert into chat_mentions(message_id, target_type, target_id, offset_start, offset_end) values (?, 'employee', ?, 0, 6)").run(quiet.messageId, quiet.employeeId); + // The busy company is newer in everything than the quiet one. + copies(db, "chat_messages", busy.messageId, 600); + copies(db, "task_runs", "run_busy", 250); + organization.setActiveOrganization(quiet.orgId); + + const snapshot = new SnapshotService(db).load(); + expect(snapshot.chatMessages.map((message) => message.id)).toContain(quiet.messageId); + expect(snapshot.chatMessages.every((message) => message.orgId === quiet.orgId)).toBe(true); + expect(snapshot.taskRuns.map((run) => run.id)).toEqual(["run_quiet"]); + expect(snapshot.chatMentions.map((mention) => mention.messageId)).toEqual([quiet.messageId]); + expect(snapshot.chatThreads.map((thread) => thread.orgId)).not.toContain(busy.orgId); + connection.close(); + }); +});