diff --git a/services/platform/backend/core/audit_logs/agent_run_ledger.ts b/services/platform/backend/core/audit_logs/agent_run_ledger.ts deleted file mode 100644 index f4801565bf..0000000000 --- a/services/platform/backend/core/audit_logs/agent_run_ledger.ts +++ /dev/null @@ -1,389 +0,0 @@ -/** - * The provenance ledger: one IMMUTABLE audit-chain entry per settled agent - * run — the customer-facing record binding a run's artefacts to the model - * identity, capability scope, knowledge read-set, and reviewer that produced - * them. Entries ride the per-org hash-chained `auditLogs` table (category - * `'agent'`, declared in the schema union since before anything emitted it), - * so chain integrity verification, export, and retention already cover them. - * - * Exactly-once by construction, not by bookkeeping of its own: each writer - * below is invoked INSIDE the one mutation that flips its run row from a - * live status to a terminal one — the settle election's once-only claim. - * That status guard admits exactly one terminal transition per run, the - * ledger write shares its transaction, and so a raced double-settle that - * degrades to a no-op also writes no second entry. - * - * Payloads are BOUNDED — an audit row is an ordinary Convex document, so - * every embedded array is capped (the named constants below) and every read - * this module performs is index-backed with an explicit scan bound. - */ - -import { isRecord } from '../../../lib/utils/type-utils'; -import type { MutationCtx } from '../lib/ctx'; -import type { Doc } from '../lib/rows'; -import { convexStorageId } from '../lib/storage/blob_ref'; -import { createAuditLog } from './helpers'; -import type { AuditLogActorType } from './types'; - -/** The deliverable rows a task carries, as much of one as the ledger reads. */ -interface TaskOutputRow { - runId?: string; - fileId: string; - fileName: string; - fileSize?: number; -} - -/** One action for both surfaces; `metadata.surface` tells them apart. */ -export const AGENT_RUN_LEDGER_ACTION = 'agent.run_settled'; -export const AGENT_RUN_LEDGER_RESOURCE_TYPE = 'agent_run'; - -/** Deliverables embedded per entry (`metadata.outputs`); `outputCount` keeps - * the true total when the cap clips. */ -export const AGENT_RUN_LEDGER_OUTPUTS_CAP = 50; -/** Distinct knowledge-source refs embedded per entry. */ -export const AGENT_RUN_LEDGER_KNOWLEDGE_READS_CAP = 200; -/** Approval ids embedded per automation entry. */ -export const AGENT_RUN_LEDGER_APPROVALS_CAP = 50; -/** Newest-first bound on the `sandboxToolCalls` read-set scan — clips the - * OLDEST calls of a pathological run, never the latest. */ -export const AGENT_RUN_LEDGER_TOOL_CALL_SCAN_CAP = 500; - -/** A standing project-agent session accumulates one gateway token per turn; - * this run's own token (matched by key id) is among the newest. */ -const TOKEN_SCAN_CAP = 50; -/** Reviews per task = runs per task; ours was minted moments before the - * settle, so it is among the newest. */ -const REVIEW_SCAN_CAP = 32; -/** Sandbox sessions per automation run (one per `sandbox`-scoped step). */ -const RUN_SESSION_SCAN_CAP = 8; -/** Turn ops per automation-run session (one per agent exec + handoffs). */ -const RUN_SESSION_OP_SCAN_CAP = 100; - -/** - * Write the ledger entry for a TASK agent run (`projectAgentRuns`). Must be - * called from the mutation that stamps the run's terminal status, with the - * PRE-PATCH row — see the module doctrine above. Every enrichment source is - * optional-tolerant: a start that died before its op row, a pre-`toolGrants` - * token, an `s3:` blob with no `_storage` system row, a refused review park - * all degrade to omitted fields, never to a failed settle. - */ -export async function recordTaskAgentRunLedgerEntry( - ctx: MutationCtx, - args: { - run: Doc<'projectAgentRuns'>; - finalStatus: 'settled' | 'failed' | 'cancelled'; - settledAt: number; - /** The failure reason for a `failed` stamp (mirrors the run row's patch). */ - error?: string; - }, -): Promise { - const { run, finalStatus, settledAt } = args; - const runKey = String(run._id); - - const task = await ctx.db.get(run.taskId); - const agent = await ctx.db.get(run.agentId); - - // The turn's op row records what actually SERVED the run: the gateway - // model ref, the vision polyfill's pick, the minted key id, live spend. - const op = await ctx.db - .query('sandboxSessionOps') - .withIndex('by_sessionId_and_execId', (q) => - q.eq('sessionId', run.sessionId).eq('execId', run.execId), - ) - .first(); - - // The run's gateway-token scope snapshot (capability bounds at mint), - // matched to THIS run via the op row's minted key id. - let token: Doc<'sandboxSessionTokens'> | undefined; - if (op?.mintedKeyId !== undefined) { - const mintedKeyId = op.mintedKeyId; - const recentTokens = await ctx.db - .query('sandboxSessionTokens') - .withIndex('by_sessionId', (q) => q.eq('sessionId', run.sessionId)) - .order('desc') - .take(TOKEN_SCAN_CAP); - token = recentTokens.find((row) => row.llmGatewayKeyId === mintedKeyId); - } - - // Deliverables THIS run stamped onto the task (recorded by the settle - // choreography before the terminal mark, so they are visible here). - // sha256/size ride Convex `_storage` system metadata; an `s3:` blob ref - // has no system row — size falls back to the task row's own snapshot and - // the hash is omitted. - const producedByRun: TaskOutputRow[] = (task?.outputs ?? []).filter( - (output: TaskOutputRow) => output.runId === run._id, - ); - const outputs: Record[] = []; - for (const output of producedByRun.slice(0, AGENT_RUN_LEDGER_OUTPUTS_CAP)) { - // `convexStorageId` blind-casts every non-`s3:` string, and - // `db.system.get` THROWS on an undecodable id — which would wedge the - // terminal mutation forever (module doctrine: degrade to omitted - // fields, never to a failed settle). Normalize first; a malformed ref - // degrades exactly like an `s3:` one — hash omitted, size from the row. - const rawStorageId = convexStorageId(output.fileId); - const storageId = - rawStorageId === null - ? null - : ctx.db.system.normalizeId('_storage', rawStorageId); - const sys = storageId !== null ? await ctx.db.system.get(storageId) : null; - outputs.push({ - fileName: output.fileName, - ...(sys?.sha256 !== undefined ? { sha256: sys.sha256 } : {}), - size: sys?.size ?? output.fileSize, - }); - } - - // The run's knowledge read-set: distinct refs from this session's tool - // calls. `knowledgeRefs` is optional — absent on rows written before the - // field shipped and on non-RAG calls. Attribution prefers the row's - // exec pin: a row pinned to THIS run's exec is definitively its read; a - // row pinned to a DIFFERENT exec is a sibling turn's read on this - // standing session and is excluded even inside the time window — false - // provenance is worse than omission, and the full trail stays queryable - // on `sandboxToolCalls`. (A steered run that rotated execs under-reports - // its pre-rotation reads here for the same reason.) Un-pinned rows keep - // the [startedAt, settledAt] window fallback. - const knowledgeReads = new Set(); - let toolCallsScanned = 0; - for await (const call of ctx.db - .query('sandboxToolCalls') - .withIndex('by_sessionId', (q) => q.eq('sessionId', run.sessionId)) - .order('desc')) { - if (++toolCallsScanned > AGENT_RUN_LEDGER_TOOL_CALL_SCAN_CAP) break; - if (call.calledAt > settledAt) continue; - // Newest-first scan: everything from here on predates the run. - if (call.calledAt < run.startedAt) break; - if (call.execId !== undefined && call.execId !== run.execId) continue; - for (const ref of call.knowledgeRefs ?? []) { - if (knowledgeReads.size >= AGENT_RUN_LEDGER_KNOWLEDGE_READS_CAP) break; - knowledgeReads.add(ref); - } - if (knowledgeReads.size >= AGENT_RUN_LEDGER_KNOWLEDGE_READS_CAP) break; - } - - // Reviewer linkage: the settle minted this run's workflow-free - // `task_review` earlier in the same choreography; its `requestedFor` - // names the human the work now waits on. Absent for failed/cancelled - // runs (nothing parked at in_review) and when the park was refused. - let reviewerUserId: string | undefined; - if (finalStatus === 'settled') { - const reviews = await ctx.db - .query('approvals') - .withIndex('by_resource', (q) => - q - .eq('resourceType', 'task_review') - .eq('resourceId', String(run.taskId)), - ) - .order('desc') - .take(REVIEW_SCAN_CAP); - for (const review of reviews) { - const metadata: unknown = review.metadata; - if (!isRecord(metadata) || metadata.runId !== runKey) continue; - if (typeof metadata.requestedFor === 'string') { - reviewerUserId = metadata.requestedFor; - } - break; - } - } - - const gateway: Record = { - ...(op?.mintedKeyId !== undefined ? { keyId: op.mintedKeyId } : {}), - ...(token !== undefined - ? { - allowedModels: token.scope.allowedModels, - budgetCents: token.scope.budgetCents, - } - : {}), - ...(op?.spentCents !== undefined ? { spentCents: op.spentCents } : {}), - }; - - await createAuditLog(ctx, { - organizationId: run.organizationId, - // The kick is a person's act on every task lane (board verb, comment - // @mention, review request-changes): `startedBy` holds their userId. An - // auto_retry run carries its failed predecessor's starter — the retry - // continues THAT person's kick; `metadata.trigger` tells the two apart. - actorId: run.startedBy, - actorType: 'user', - action: AGENT_RUN_LEDGER_ACTION, - category: 'agent', - resourceType: AGENT_RUN_LEDGER_RESOURCE_TYPE, - resourceId: runKey, - ...(task !== null ? { resourceName: task.title } : {}), - status: finalStatus === 'failed' ? 'failure' : 'success', - ...(finalStatus === 'failed' && args.error !== undefined - ? { errorMessage: args.error } - : {}), - metadata: { - surface: 'task', - runId: runKey, - execId: run.execId, - taskId: String(run.taskId), - projectId: String(run.projectId), - agentId: String(run.agentId), - ...(agent !== null ? { agentName: agent.name } : {}), - harness: run.harness, - ...(run.trigger !== undefined ? { trigger: run.trigger } : {}), - finalStatus, - startedAt: run.startedAt, - settledAt, - durationMs: Math.max(0, settledAt - run.startedAt), - model: { - requested: run.model, - ...(op?.modelRef !== undefined ? { servedRef: op.modelRef } : {}), - ...(op?.visionModelRef !== undefined - ? { visionRef: op.visionModelRef } - : {}), - }, - ...(Object.keys(gateway).length > 0 ? { gateway } : {}), - ...(token !== undefined - ? { - grants: { - connectors: token.scope.connectorGrants, - ...(token.scope.toolGrants !== undefined - ? { tools: token.scope.toolGrants } - : {}), - }, - } - : {}), - ...(outputs.length > 0 - ? { outputs, outputCount: producedByRun.length } - : {}), - ...(knowledgeReads.size > 0 - ? { knowledgeReads: [...knowledgeReads] } - : {}), - ...(reviewerUserId !== undefined ? { review: { reviewerUserId } } : {}), - }, - }); -} - -/** - * Write the ledger entry for an AUTOMATION run (`automationRuns`). Same - * calling contract as the task writer: inside the run's one terminal - * mutation, with the pre-patch row. LIVE runs only — a `mock` run is a test - * that touches no outside world, so it mints no provenance (enforced here so - * no call site can forget). - */ -export async function recordAutomationRunLedgerEntry( - ctx: MutationCtx, - args: { - run: Doc<'automationRuns'>; - finalStatus: 'success' | 'failed' | 'cancelled'; - finishedAt: number; - /** Length of the run's effect log, where the terminal writer holds it. */ - effectsCount?: number; - /** The failure detail for a `failed` finish (mirrors the row's patch). */ - detail?: string; - }, -): Promise { - const { run, finalStatus, finishedAt } = args; - if (run.mode !== 'live') return; - - const runKey = String(run._id); - - // Approvals raised for this run: the live-write gate keys every - // `connector_operation` approval on `:` (approvals/gate.ts), - // so the run's rows are one contiguous `by_resource` index range — - // `';'` = `':' + 1`, the established range-scan idiom (see - // `listAutomationRunSessionsForExecution`). The org filter is defensive: - // the runId prefix is globally unique by construction. - const approvalRows = await ctx.db - .query('approvals') - .withIndex('by_resource', (q) => - q - .eq('resourceType', 'connector_operation') - .gte('resourceId', `${runKey}:`) - .lt('resourceId', `${runKey};`), - ) - .take(AGENT_RUN_LEDGER_APPROVALS_CAP); - const approvals = approvalRows - .filter((row) => row.organizationId === run.organizationId) - .map((row) => String(row._id)); - - // In-run LLM spend: the run's sandbox sessions are owner-keyed on the same - // `${runId}:` prefix; each agent turn's op row carries its polled spend. - // Best-effort — ops are session-scoped and can be purged with the session. - let spentCents = 0; - let sawSpend = false; - const sessions = await ctx.db - .query('sandboxSessions') - .withIndex('by_owner', (q) => - q - .eq('ownerType', 'workflow_run') - .gte('ownerId', `${runKey}:`) - .lt('ownerId', `${runKey};`), - ) - .take(RUN_SESSION_SCAN_CAP); - for (const session of sessions) { - if (session.organizationId !== run.organizationId) continue; - const ops = await ctx.db - .query('sandboxSessionOps') - .withIndex('by_sessionId', (q) => q.eq('sessionId', session.sessionId)) - .take(RUN_SESSION_OP_SCAN_CAP); - for (const op of ops) { - if (op.spentCents !== undefined) { - sawSpend = true; - spentCents += op.spentCents; - } - } - } - - const actor = automationRunActor(run.startedBy); - - await createAuditLog(ctx, { - organizationId: run.organizationId, - actorId: actor.actorId, - actorType: actor.actorType, - action: AGENT_RUN_LEDGER_ACTION, - category: 'agent', - resourceType: AGENT_RUN_LEDGER_RESOURCE_TYPE, - resourceId: runKey, - resourceName: run.name, - status: finalStatus === 'failed' ? 'failure' : 'success', - ...(finalStatus === 'failed' && args.detail !== undefined - ? { errorMessage: args.detail } - : {}), - metadata: { - surface: 'automation', - runId: runKey, - automationName: run.name, - automationVersion: run.version, - ...(run.projectId !== undefined - ? { projectId: String(run.projectId) } - : {}), - startedBy: run.startedBy, - finalStatus, - startedAt: run.startedAt, - settledAt: finishedAt, - durationMs: Math.max(0, finishedAt - run.startedAt), - ...(approvals.length > 0 ? { approvals } : {}), - ...(args.effectsCount !== undefined - ? { effectsCount: args.effectsCount } - : {}), - ...(sawSpend ? { spentCents } : {}), - }, - }); -} - -/** - * The audit actor behind an automation run's `startedBy` marker: `user:` - * is a person, `api-key:` is a person acting through an org API key, and - * everything else (`trigger:`, store actors) keeps its origin marker as - * the actorId under the system actor pattern (`actorId: 'system'` writers - * elsewhere carry no origin; here the marker IS the origin). - */ -function automationRunActor(startedBy: string): { - actorId: string; - actorType: AuditLogActorType; -} { - if (startedBy.startsWith('user:')) { - return { actorId: startedBy.slice('user:'.length), actorType: 'user' }; - } - if (startedBy.startsWith('api-key:')) { - return { actorId: startedBy.slice('api-key:'.length), actorType: 'api' }; - } - return { - actorId: startedBy === '' ? 'system' : startedBy, - actorType: 'system', - }; -} diff --git a/services/platform/backend/core/automations/store.ts b/services/platform/backend/core/automations/store.ts deleted file mode 100644 index b1e6197e77..0000000000 --- a/services/platform/backend/core/automations/store.ts +++ /dev/null @@ -1,877 +0,0 @@ -/** - * The Convex host behind the engine's `DispatchStore` — the same interface the - * in-memory reference store implements (`lib/engine/store/memory.ts`), backed - * by the four automation tables. - * - * Two rules the memory store establishes and this one MUST mirror, because the - * builder selftest runs against one and production against the other: - * - * - **versions are immutable and contiguous** — `save` only ever appends, so - * version N of an automation is byte-identical forever and `latest` is the - * row count; - * - **deploy is a separate, explicit act** — it names an existing version and - * replaces the single deployment row; it never touches history. - * - * The one thing this store adds is the boundary the reference store has no - * concept of: EVERY read and write is scoped to one organization. The scope is - * bound at construction and applied through the `by_org…` indexes, so no method - * can be called in a way that reaches another organization's rows — including - * `deploy`, whose version lookup is org-scoped too. - * - * The store is transactional (it takes a Convex query/mutation ctx), which is - * what makes "the next version is latest + 1" safe under concurrency. An action - * — where `dispatch()` and the authoring loop live — has no database handle, so - * it uses {@link automationActionStore}: the same interface, forwarding to the - * registered functions that wrap THIS store. The versioning and deploy rules - * therefore exist in exactly one place whatever the caller. - */ - -import type { - DispatchStore, - TriggerSpec, -} from '../../../lib/engine/api/dispatch'; -import type { StoreAdapter } from '../../../lib/engine/core/slots'; -import type { Automation, RunResult } from '../../../lib/engine/core/types'; -import { AppError } from '../../../lib/shared/errors/app-error'; -import { recordAutomationRunLedgerEntry } from '../audit_logs/agent_run_ledger'; -import type { ActionCtx, MutationCtx, QueryCtx } from '../lib/ctx'; -import { internal } from '../lib/handler_names'; -import type { Doc, Id } from '../lib/rows'; -import { boundRunTrace, truncateRunDetail } from './bound_run_payload'; - -/** Who a write is attributed to — a user id, or a system marker for a write - * the platform performs on its own behalf. */ -export type Actor = string; - -export interface AutomationStoreScope { - organizationId: string; - actor: Actor; - /** - * Install target for a NEW automation this store creates: the first save - * also binds the name to this project (a row in - * `automationProjectBindings`), atomically with the version insert. Saves - * of an EXISTING name ignore it — project membership is managed explicitly - * (`setAutomationProjects`, the upload lane's install target), never moved - * as a side effect of saving a version. - */ - projectId?: Id<'projects'>; -} - -/** Extra facts a save carries that the engine's `DispatchStore.save` signature - * has no room for. `testsPassed` is the deploy gate's evidence: it records - * whether the version's own acceptance tests passed at save time, so promotion - * reads a fact instead of re-running them. */ -export interface SaveOptions { - testsPassed?: boolean; - /** The version's task-surface contract (already zod-validated by the - * caller); stored beside the document. */ - taskContract?: unknown; - /** The version's settings declaration (already zod-validated by the - * caller); stored beside the document. */ - settings?: unknown; - /** How the version names itself to people (already zod-validated by the - * caller) — the pack manifest's display half. */ - presentation?: unknown; - /** A create, not an append: refuse if the name already has versions. The - * blank-automation wizard sets this so "create" cannot silently append a - * version to — and rebind the trigger of — a live automation that happens to - * share the slug. Plain authoring saves (which are appends) leave it unset. */ - create?: boolean; -} - -/** What `setTrigger` may persist. Mirrors the engine's `TriggerSpec` plus the - * one field only the host can produce: the hash of a webhook token (the - * plaintext is shown once at creation and never stored). */ -export interface StoredTrigger extends TriggerSpec { - cron?: string; - timezone?: string; - event?: string; - tokenHash?: string; - enabled?: boolean; -} - -/** - * The authoring contract both Convex-backed stores satisfy. The engine's - * MANAGEMENT methods (`startRun`, `getRun`, `listTriggers`, …) stay optional as - * the engine declares them: only {@link automationActionStore} fills them in, - * because starting a durable run needs the scheduler and an authorization check - * that belongs to a registered function rather than to the transactional core. - */ -export interface ConvexAutomationStore extends DispatchStore { - save( - automation: Automation, - message?: string, - options?: SaveOptions, - ): Promise<{ name: string; version: number }>; - setTrigger(name: string, trigger: StoredTrigger): Promise; - recordRun( - name: string, - version: number, - result: RunResult, - mode: 'mock' | 'live', - ): Promise; -} - -/** - * The kinds a write may bind. `api-key` is absent on purpose: a programmatic - * start is what the REST and MCP surfaces are for, so the kind never had a - * delivery path of its own and is refused here. The stored union in the schema - * still ALLOWS the value, so a row written before it was retired stays readable. - */ -const TRIGGER_KINDS = ['schedule', 'webhook', 'event'] as const; -type TriggerKind = (typeof TRIGGER_KINDS)[number]; - -function isTriggerKind(value: unknown): value is TriggerKind { - return ( - typeof value === 'string' && - (TRIGGER_KINDS as readonly string[]).includes(value) - ); -} - -/** An automation name is a `/`-separated path, unique per organization. */ -const NAME_RE = - /^[a-z0-9]+(?:[-_][a-z0-9]+)*(?:\/[a-z0-9]+(?:[-_][a-z0-9]+)*)*$/; -const NAME_MAX = 200; - -export function assertAutomationName(name: string): string { - const value = name.trim(); - if (value.length === 0 || value.length > NAME_MAX || !NAME_RE.test(value)) { - throw new Error( - `"${name}" is not a valid automation name — use lowercase slug segments separated by "/" (e.g. "billing/dunning-reminder")`, - ); - } - return value; -} - -// ------------------------------------------------------------------- reads - -/** Every version of one automation, oldest first. */ -export async function versionsOf( - ctx: QueryCtx, - organizationId: string, - name: string, -): Promise>> { - const rows = await ctx.db - .query('automations') - .withIndex('by_org_name', (q) => - q.eq('organizationId', organizationId).eq('name', name), - ) - .collect(); - return rows.sort((a, b) => a.version - b.version); -} - -/** One version, or the latest when `version` is omitted. */ -export async function versionRow( - ctx: QueryCtx, - organizationId: string, - name: string, - version?: number, -): Promise | null> { - if (version === undefined) { - const rows = await versionsOf(ctx, organizationId, name); - return rows.at(-1) ?? null; - } - return await ctx.db - .query('automations') - .withIndex('by_org_name_version', (q) => - q - .eq('organizationId', organizationId) - .eq('name', name) - .eq('version', version), - ) - .unique(); -} - -export async function deploymentRow( - ctx: QueryCtx, - organizationId: string, - name: string, -): Promise | null> { - return await ctx.db - .query('automationDeployments') - .withIndex('by_org_name', (q) => - q.eq('organizationId', organizationId).eq('name', name), - ) - .unique(); -} - -export async function triggerRow( - ctx: QueryCtx, - organizationId: string, - name: string, -): Promise | null> { - return await ctx.db - .query('automationTriggers') - .withIndex('by_org_name', (q) => - q.eq('organizationId', organizationId).eq('name', name), - ) - .unique(); -} - -/** The deletion marker for a name, if one stands. The default-pack seeder - * consults it so a deliberately deleted builtin does not come back on the - * next deploy; `save` clears it (the name is alive again). */ -export async function tombstoneRow( - ctx: QueryCtx, - organizationId: string, - name: string, -): Promise | null> { - return await ctx.db - .query('automationTombstones') - .withIndex('by_org_name', (q) => - q.eq('organizationId', organizationId).eq('name', name), - ) - .unique(); -} - -/** One automation's project bindings. The binding set is the scope: empty - * means org-level, non-empty means exactly those projects. */ -export async function bindingsOf( - ctx: QueryCtx, - organizationId: string, - name: string, -): Promise>> { - return await ctx.db - .query('automationProjectBindings') - .withIndex('by_org_name_project', (q) => - q.eq('organizationId', organizationId).eq('automationName', name), - ) - .collect(); -} - -/** - * The single project an automation is bound to, when that is unambiguous — - * what a run started WITHOUT a project context (a trigger firing, a manual - * run) is attributed to. Multi-bound and org-level automations resolve to - * nothing: their runs belong to no one project unless the caller says so. - */ -export async function soleBindingProject( - ctx: QueryCtx, - organizationId: string, - name: string, -): Promise | undefined> { - const bindings = await bindingsOf(ctx, organizationId, name); - return bindings.length === 1 ? bindings[0]?.projectId : undefined; -} - -/** - * Guard a caller-supplied run project: the target must be a project in the org, - * and — when the automation is BOUND to projects — one of them (running a - * project-bound automation "for" an unbound project would escape its scope). An - * org-level automation (no bindings) may run for any project. Throws a coded - * `AppError` on violation; used by the run-control entry points that accept - * a caller `projectId` (manual UI, REST, MCP). The task-surface path trusts the - * task's own project and does not go through here. - */ -export async function assertRunProjectAllowed( - ctx: QueryCtx, - organizationId: string, - name: string, - projectId: Id<'projects'>, -): Promise { - const project = await ctx.db.get(projectId); - if (project === null || project.organizationId !== organizationId) { - throw new AppError({ - code: 'PROJECT_NOT_FOUND', - message: 'No such project in this organization.', - }); - } - const bindings = await bindingsOf(ctx, organizationId, name); - if ( - bindings.length > 0 && - !bindings.some((binding) => binding.projectId === projectId) - ) { - throw new AppError({ - code: 'PROJECT_NOT_BOUND', - message: - 'This automation is bound to specific projects; a run can only ' + - 'target one of them.', - }); - } -} - -/** The organization's automations with their latest version — `list()`'s data, - * shared with the read surface so the two can never disagree. */ -export async function listAutomationsFor( - ctx: QueryCtx, - organizationId: string, - /** - * Surface filter: an id lists ONE project's automations (the names bound to - * it), `null` lists the org-level ones (no bindings), and `undefined` — the - * engine's view — lists everything, so subautomation resolution and the - * chat capability registry see project automations too. - */ - projectId?: Id<'projects'> | null, -): Promise< - Array<{ name: string; latest: number; projectIds: Array> }> -> { - const rows = await ctx.db - .query('automations') - .withIndex('by_org', (q) => q.eq('organizationId', organizationId)) - .collect(); - const bindings = await ctx.db - .query('automationProjectBindings') - .withIndex('by_org_name_project', (q) => - q.eq('organizationId', organizationId), - ) - .collect(); - const bound = new Map>>(); - for (const binding of bindings) { - const list = bound.get(binding.automationName) ?? []; - list.push(binding.projectId); - bound.set(binding.automationName, list); - } - const latest = new Map(); - for (const row of rows) { - latest.set(row.name, Math.max(latest.get(row.name) ?? 0, row.version)); - } - return [...latest.entries()] - .map(([name, latestVersion]) => ({ - name, - latest: latestVersion, - projectIds: bound.get(name) ?? [], - })) - .filter((entry) => { - if (projectId === undefined) return true; - if (projectId === null) return entry.projectIds.length === 0; - return entry.projectIds.includes(projectId); - }) - .sort((a, b) => a.name.localeCompare(b.name)); -} - -/** - * The read half of the store — everything the executor needs to resolve a - * subautomation reference. Usable from a plain query, and org-scoped like the - * full store. - */ -export function automationReadStore( - ctx: QueryCtx, - organizationId: string, -): StoreAdapter { - return { - // The adapter's contract is name+latest only — project membership is a - // surface concern the engine has no business seeing. - list: async () => - (await listAutomationsFor(ctx, organizationId)).map( - ({ name, latest }) => ({ name, latest }), - ), - async get(name, version) { - const row = await versionRow(ctx, organizationId, name, version); - if (!row) return null; - return { meta: { version: row.version }, automation: row.document }; - }, - async deployedVersion(name) { - const row = await deploymentRow(ctx, organizationId, name); - return row?.version ?? null; - }, - }; -} - -// ------------------------------------------------------------------ writes - -/** - * Bind one automation name to one project — idempotent, and transactional - * with whatever mutation hosts it (the first save's install intent, the - * upload lane's target, the reconcile mutation). Refuses a project that does - * not exist or lives in another organization, so a binding row is only ever - * a true statement. - */ -export async function bindAutomationToProject( - ctx: MutationCtx, - args: { - organizationId: string; - automationName: string; - projectId: Id<'projects'>; - actor: Actor; - }, -): Promise<{ bound: boolean }> { - const project = await ctx.db.get(args.projectId); - if (!project || project.organizationId !== args.organizationId) { - throw new Error( - `cannot bind "${args.automationName}" — the project does not exist in this organization`, - ); - } - const existing = await ctx.db - .query('automationProjectBindings') - .withIndex('by_org_name_project', (q) => - q - .eq('organizationId', args.organizationId) - .eq('automationName', args.automationName) - .eq('projectId', args.projectId), - ) - .unique(); - if (existing) return { bound: false }; - await ctx.db.insert('automationProjectBindings', { - organizationId: args.organizationId, - automationName: args.automationName, - projectId: args.projectId, - boundAt: Date.now(), - boundBy: args.actor, - }); - return { bound: true }; -} - -/** - * Reconcile one automation's binding set to exactly `projectIds` — the - * Projects panel saves the whole selection, so add and remove land in one - * transaction and two concurrent saves converge on one of the two complete - * selections rather than an interleaving. Empty = org-level. Refuses an - * unknown name: a binding must always point at a real automation. - */ -export async function reconcileAutomationProjects( - ctx: MutationCtx, - args: { - organizationId: string; - actor: Actor; - name: string; - projectIds: ReadonlyArray>; - }, -): Promise<{ bound: number; unbound: number }> { - const name = assertAutomationName(args.name); - const versions = await versionsOf(ctx, args.organizationId, name); - if (versions.length === 0) { - throw new Error(`"${name}" has no versions in this organization`); - } - const desired = new Set(args.projectIds); - const existing = await bindingsOf(ctx, args.organizationId, name); - const current = new Set(existing.map((row) => row.projectId)); - let bound = 0; - let unbound = 0; - for (const row of existing) { - if (desired.has(row.projectId)) continue; - await ctx.db.delete(row._id); - unbound++; - } - for (const projectId of desired) { - if (current.has(projectId)) continue; - await bindAutomationToProject(ctx, { - organizationId: args.organizationId, - automationName: name, - projectId, - actor: args.actor, - }); - bound++; - } - return { bound, unbound }; -} - -/** - * Delete an automation: every version, its deployment, its triggers (a - * webhook URL dies here, a schedule stops firing) and its project bindings — - * one transaction, so no partial automation is ever observable. - * - * Refused while any run is still queued, running or waiting: a live run - * holds a sandbox session and may be parked on a human question, and failing - * it as a side effect of deletion would destroy in-flight work silently. - * Cancel the runs (or let them finish) first. - * - * Run history is deliberately KEPT — terminal runs are the audit record of - * what the automation did, and the `workflowLog` retention sweep already - * owns their lifecycle. A tombstone is recorded so the default-pack seeder - * does not resurrect a deleted builtin on the next deploy; saving the name - * again clears it. - */ -export async function deleteAutomationCascade( - ctx: MutationCtx, - args: { organizationId: string; name: string; actor: Actor }, -): Promise<{ name: string; versions: number }> { - const name = assertAutomationName(args.name); - const versions = await versionsOf(ctx, args.organizationId, name); - if (versions.length === 0) { - throw new AppError({ - code: 'AUTOMATION_NOT_FOUND', - message: 'No such automation for this organization.', - }); - } - // The index range is (org, name) — every run of this automation, live and - // terminal alike. The scan is bounded in practice by the workflowLog - // retention sweep, which expires and hard-deletes aged terminal rows. - const activeRun = await ctx.db - .query('automationRuns') - .withIndex('by_org_name', (q) => - q.eq('organizationId', args.organizationId).eq('name', name), - ) - .filter((q) => - q.or( - q.eq(q.field('status'), 'queued'), - q.eq(q.field('status'), 'running'), - q.eq(q.field('status'), 'waiting'), - ), - ) - .first(); - if (activeRun !== null) { - throw new AppError({ - code: 'AUTOMATION_HAS_ACTIVE_RUNS', - message: `A run of "${name}" is still ${activeRun.status} — cancel it (or let it finish) before deleting the automation.`, - }); - } - for (const version of versions) { - await ctx.db.delete(version._id); - } - const deployment = await deploymentRow(ctx, args.organizationId, name); - if (deployment !== null) await ctx.db.delete(deployment._id); - // One trigger per name is the store's rule, but the delete sweeps the - // whole index range so a historic duplicate cannot keep firing. - const triggers = await ctx.db - .query('automationTriggers') - .withIndex('by_org_name', (q) => - q.eq('organizationId', args.organizationId).eq('name', name), - ) - .collect(); - for (const trigger of triggers) { - await ctx.db.delete(trigger._id); - } - for (const binding of await bindingsOf(ctx, args.organizationId, name)) { - await ctx.db.delete(binding._id); - } - const tombstone = await tombstoneRow(ctx, args.organizationId, name); - if (tombstone === null) { - await ctx.db.insert('automationTombstones', { - organizationId: args.organizationId, - name, - deletedBy: args.actor, - deletedAt: Date.now(), - }); - } else { - await ctx.db.patch(tombstone._id, { - deletedBy: args.actor, - deletedAt: Date.now(), - }); - } - return { name, versions: versions.length }; -} - -/** - * The full store for one organization. `ctx` must be a mutation context: the - * write methods are what make this more than the read adapter above. - */ -export function automationStore( - ctx: MutationCtx, - scope: AutomationStoreScope, -): ConvexAutomationStore { - const { organizationId, actor } = scope; - const reads = automationReadStore(ctx, organizationId); - - return { - // Arrow wrappers keep the read helpers bound to their own closure rather - // than being passed as free references. - list: () => reads.list(), - get: (name: string, version?: number) => reads.get(name, version), - deployedVersion: (name: string) => reads.deployedVersion(name), - - /** - * Append a version. The next number is `latest + 1` computed inside this - * transaction, so two concurrent saves cannot mint the same version: Convex - * serializes conflicting transactions and the loser retries against the - * row it did not see. - */ - async save(automation, message, options) { - const name = assertAutomationName(automation.name ?? ''); - const rows = await versionsOf(ctx, organizationId, name); - // A create must not append to an existing automation. The check and the - // insert share one transaction, so a concurrent create loses the OCC race - // and retries into this same refusal rather than minting a second v1. - if (options?.create === true && rows.length > 0) { - throw new AppError({ - code: 'AUTOMATION_NAME_TAKEN', - message: `An automation named "${name}" already exists.`, - }); - } - const version = (rows.at(-1)?.version ?? 0) + 1; - // A save under a deleted name revives it: without this, a re-created - // builtin would stay invisible to the default-pack seeder forever. - const tombstone = await tombstoneRow(ctx, organizationId, name); - if (tombstone !== null) await ctx.db.delete(tombstone._id); - await ctx.db.insert('automations', { - organizationId, - name, - version, - document: automation, - ...(message !== undefined && message !== '' && { message }), - ...(options?.testsPassed !== undefined && { - testsPassed: options.testsPassed, - }), - ...(options?.taskContract !== undefined && { - taskContract: options.taskContract, - }), - ...(options?.settings !== undefined && { - settings: options.settings, - }), - ...(options?.presentation !== undefined && { - presentation: options.presentation, - }), - createdBy: actor, - createdAt: Date.now(), - }); - // A NEW automation saved from a project surface starts bound to it — - // the install intent lands atomically with the first version. Later - // saves never touch bindings: membership is managed explicitly, so - // saving a version cannot move an automation between surfaces. - if (rows.length === 0 && scope.projectId !== undefined) { - await bindAutomationToProject(ctx, { - organizationId, - automationName: name, - projectId: scope.projectId, - actor, - }); - } - return { name, version }; - }, - - /** - * Promote one version. Refuses a version that does not exist in THIS - * organization (same message as the reference store) and one whose own - * tests failed at save time — the deploy gate, applied here as well as in - * `dispatch()` so no caller can route around it. - */ - async deploy(name, version) { - const row = await versionRow(ctx, organizationId, name, version); - if (!row) { - throw new Error(`cannot deploy unknown version ${name}@${version}`); - } - if (row.testsPassed === false) { - throw new Error( - `deploy gate: ${name}@${version} was saved with failing tests — fix them and save a new version`, - ); - } - const existing = await deploymentRow(ctx, organizationId, name); - const patch = { - version, - deployedBy: actor, - deployedAt: Date.now(), - }; - if (existing) await ctx.db.patch(existing._id, patch); - else - await ctx.db.insert('automationDeployments', { - organizationId, - name, - ...patch, - }); - return { name, version }; - }, - - /** - * Bind what starts the automation. One trigger per automation name, so - * re-recording replaces the binding in place and a webhook URL survives a - * redeploy. A webhook token hash is never cleared by an update that does - * not carry one — the URL a vendor already holds keeps working. - */ - async setTrigger(name, trigger) { - const automation = assertAutomationName(name); - if (!isTriggerKind(trigger.kind)) { - throw new Error( - `unknown trigger kind "${String(trigger.kind)}" — one of ${TRIGGER_KINDS.join(', ')}`, - ); - } - const kind = trigger.kind; - if (kind === 'schedule' && !trigger.cron) { - throw new Error('a schedule trigger needs a cron expression'); - } - if (kind === 'event' && !trigger.event) { - throw new Error('an event trigger needs an event name'); - } - const existing = await triggerRow(ctx, organizationId, automation); - const now = Date.now(); - const fields = { - kind, - ...(trigger.cron !== undefined && { cron: trigger.cron }), - ...(trigger.timezone !== undefined && { timezone: trigger.timezone }), - ...(trigger.event !== undefined && { event: trigger.event }), - ...(trigger.tokenHash !== undefined && { - tokenHash: trigger.tokenHash, - }), - enabled: trigger.enabled ?? true, - updatedAt: now, - }; - if (existing) { - await ctx.db.patch(existing._id, fields); - return; - } - await ctx.db.insert('automationTriggers', { - organizationId, - name: automation, - ...fields, - createdBy: actor, - createdAt: now, - }); - }, - - /** - * Record a run the caller executed in one piece (the dispatch surface's - * `run_deployed`). Durable runs are written by the stepper instead, which - * needs the row to exist BEFORE the first node runs. - */ - async recordRun(name, version, result, mode) { - const now = Date.now(); - // A one-piece run has no caller project context, so it is attributed - // to the automation's sole bound project when that is unambiguous — - // the same rule `beginRun` applies to trigger-started runs. - const soleProject = await soleBindingProject(ctx, organizationId, name); - const runId = await ctx.db.insert('automationRuns', { - organizationId, - name, - version, - ...(soleProject !== undefined && { projectId: soleProject }), - status: result.status === 'success' ? 'success' : 'failed', - mode, - startedBy: actor, - input: null, - ...(result.output !== undefined && { output: result.output }), - // First (and only) write of this run's log — bound the diagnostics - // here. `output` and `effects` are left whole: one is returned to the - // caller, the other is the side-effect audit trail. - trace: boundRunTrace(result.trace), - effects: result.effects, - ...(result.error?.message !== undefined && { - detail: truncateRunDetail(result.error.message), - }), - startedAt: now, - finishedAt: now, - }); - // Provenance ledger for the one-piece lane too: a LIVE `run_deployed` - // run is born terminal, so this insert IS its exactly-once terminal - // transition. Mock runs are tests and the helper skips them. - const row = await ctx.db.get(runId); - if (row !== null) { - await recordAutomationRunLedgerEntry(ctx, { - run: row, - finalStatus: result.status === 'success' ? 'success' : 'failed', - finishedAt: now, - effectsCount: result.effects.length, - ...(result.error?.message !== undefined - ? { detail: result.error.message } - : {}), - }); - } - }, - }; -} - -// ------------------------------------------------------------ from an action - -/** - * The same store, for a caller that has no database handle. - * - * `dispatch()` — the one method table behind every authoring surface — runs in - * an action, because executing an automation needs the code sandbox. It is handed - * one of these: every method forwards to the registered function that wraps the - * transactional store above, so an agent editing an automation and a person - * clicking Save go through identical rules, and the organization scope travels - * with every call rather than being remembered somewhere. - * - * This is also the store that hosts DURABLE runs, which the transactional one - * cannot: `startRun` hands the run to the stepper through the scheduler, and its - * mutation authorizes the actor first — the MCP endpoint reaches this with an - * org API key, so "who may start a live run" is decided here rather than assumed. - */ -export function automationActionStore( - ctx: ActionCtx, - scope: AutomationStoreScope, -): ConvexAutomationStore { - const { organizationId, actor } = scope; - return { - list: () => - ctx.runQuery(internal.automations.queries.storeList, { organizationId }), - get: (name, version) => - ctx.runQuery(internal.automations.queries.storeGet, { - organizationId, - name, - ...(version !== undefined && { version }), - }), - deployedVersion: (name) => - ctx.runQuery(internal.automations.queries.storeDeployedVersion, { - organizationId, - name, - }), - save: (automation, message, options) => - ctx.runMutation(internal.automations.mutations.storeSave, { - organizationId, - actor, - automation, - // Ownership travels with the scope: an action-side save into a - // project surface pins the project exactly as a transactional one. - ...(scope.projectId !== undefined && { projectId: scope.projectId }), - ...(message !== undefined && message !== '' && { message }), - ...(options?.testsPassed !== undefined && { - testsPassed: options.testsPassed, - }), - }), - deploy: (name, version) => - ctx.runMutation(internal.automations.mutations.storeDeploy, { - organizationId, - actor, - name, - version, - }), - setTrigger: async (name, trigger) => { - await ctx.runMutation(internal.automations.mutations.storeSetTrigger, { - organizationId, - actor, - name, - trigger, - }); - }, - recordRun: async (name, version, result, mode) => { - await ctx.runMutation(internal.automations.mutations.storeRecordRun, { - organizationId, - actor, - name, - version, - result, - mode, - }); - }, - - // The management half. Starting and cancelling go through mutations that - // authorize the ACTOR — an org API key reaching this through the MCP - // endpoint has proved who it is but not what its role may do, and a live - // run may touch real systems. - startRun: (name, input, mode, version, projectId) => - ctx.runMutation(internal.automations.mutations.storeStartRun, { - organizationId, - actor, - name, - input, - mode, - ...(version !== undefined && { version }), - ...(projectId !== undefined && { projectId }), - }), - cancelRun: async (runId) => { - const result = await ctx.runMutation( - internal.automations.mutations.storeCancelRun, - { organizationId, actor, runId }, - ); - // An unusable handle is reported as a miss, the same way `getRun` does. - if (result === null) throw new Error(`no run "${runId}"`); - return result; - }, - deleteTrigger: async (name) => { - await ctx.runMutation(internal.automations.mutations.storeDeleteTrigger, { - organizationId, - actor, - name, - }); - }, - listRuns: (options) => - ctx.runQuery(internal.automations.queries.storeListRuns, { - organizationId, - ...(options.name !== undefined && { name: options.name }), - ...(options.limit !== undefined && { limit: options.limit }), - }), - getRun: (runId) => - ctx.runQuery(internal.automations.queries.storeGetRun, { - organizationId, - runId, - }), - listVersions: (name) => - ctx.runQuery(internal.automations.queries.storeListVersions, { - organizationId, - name, - }), - listTriggers: (name) => - ctx.runQuery(internal.automations.queries.storeListTriggers, { - organizationId, - ...(name !== undefined && { name }), - }), - }; -} diff --git a/services/platform/backend/core/collab/dismiss_review_notifications.ts b/services/platform/backend/core/collab/dismiss_review_notifications.ts deleted file mode 100644 index fe3226bed6..0000000000 --- a/services/platform/backend/core/collab/dismiss_review_notifications.ts +++ /dev/null @@ -1,88 +0,0 @@ -/** - * Marks stale review-request bell rows read when an approval resolves. - * Covers the initial transactional ping (resourceId = approvalId), automation - * reminders (resourceId = taskId, params.approvalId), and admin escalations. - */ - -import { isRecord } from '../../../lib/utils/type-utils'; -import type { MutationCtx } from '../lib/ctx'; -import type { Id } from '../lib/rows'; - -const MEMBER_SCAN_CAP = 500; -const UNREAD_SCAN_CAP = 100; - -function matchesResolvedReviewNotification( - row: { - type: string; - read: boolean; - resourceId: string; - params?: unknown; - }, - approvalId: string, -): boolean { - if (row.type !== 'task_review_requested' || row.read) return false; - - // Match only THIS approval. Every real review bell carries the approval id as - // `resourceId` (transactional ping) or `params.approvalId` (automation - // reminders). A task-level match would wrongly clear OTHER still-pending - // reviews on the same task (a task can hold >1 concurrent review approval). - if (row.resourceId === approvalId) return true; - if (isRecord(row.params) && row.params.approvalId === approvalId) return true; - - return false; -} - -/** Returns the number of notification rows marked read. */ -export async function dismissReviewRequestNotifications( - ctx: MutationCtx, - args: { - organizationId: string; - approvalId: Id<'approvals'>; - taskId: Id<'tasks'>; - }, -): Promise { - const approvalIdStr = args.approvalId; - const now = Date.now(); - let dismissed = 0; - - // NOTE (scale limitation): `userNotifications` is indexed by user first, so - // there is no way to find a review bell by approval/task without the userId — - // hence this per-member fan-out. It caps at MEMBER_SCAN_CAP, so in an org with - // more members a reviewer past the cap won't get their stale bell auto-cleared - // (the review is still actionable in the task sheet — no correctness loss, just - // a lingering bell). The clean fix is a `userNotifications` index on - // (organizationId, taskId) so we can target `args.taskId` directly; deferred to - // avoid a schema change in this PR. - const userIds: string[] = []; - for await (const member of ctx.db - .query('memberMirror') - .withIndex('by_organizationId', (q) => - q.eq('organizationId', args.organizationId), - )) { - userIds.push(member.userId); - if (userIds.length >= MEMBER_SCAN_CAP) break; - } - - for (const userId of userIds) { - const unread = await ctx.db - .query('userNotifications') - .withIndex('by_user_org_read', (q) => - q - .eq('userId', userId) - .eq('organizationId', args.organizationId) - .eq('read', false), - ) - .order('desc') - .take(UNREAD_SCAN_CAP); - - for (const row of unread) { - if (!matchesResolvedReviewNotification(row, approvalIdStr)) { - continue; - } - await ctx.db.patch(row._id, { read: true, readAt: now }); - dismissed += 1; - } - } - - return dismissed; -} diff --git a/services/platform/backend/core/collab/notify_task_reviews.ts b/services/platform/backend/core/collab/notify_task_reviews.ts deleted file mode 100644 index de54ceac69..0000000000 --- a/services/platform/backend/core/collab/notify_task_reviews.ts +++ /dev/null @@ -1,215 +0,0 @@ -/** - * Task-review inbox emitters — the human half of human-in-the-loop. - * Transactional with their source mutations, mirroring `notify.ts`. - * - * `task_review_requested` deliberately SKIPS the preference gate for the - * designated reviewer: the review gate is a safety signal and must never - * starve silently because someone muted a category. `task_review_resolved` - * also ignores the stored `taskReview` preference (including a stale - * `false` persisted before the settings UI locked the toggle always-on, - * #2651) — see `prefAllows` below. - */ - -import type { MutationCtx } from '../lib/ctx'; -import type { Doc, Id } from '../lib/rows'; -import { resolveUserDisplayName } from '../notifications/actor_name'; -import { writeCoalescedNotification } from './coalesce'; - -type TaskReviewNotificationType = - | 'task_review_requested' - | 'task_review_resolved' - | 'task_reviewer_assigned'; - -async function prefAllows( - ctx: MutationCtx, - userId: string, - organizationId: string, - field: 'taskReview' | 'escalation', -): Promise { - // Review requests are a safety signal — the settings UI locks the toggle - // always-on (#2651). Ignore any stored `taskReview` value, including a - // stale `false` persisted before that lock shipped: the server must not - // let an old row silently keep suppressing review notifications forever. - // No migration needed since the stored value is simply never read. - if (field === 'taskReview') return true; - const prefs = await ctx.db - .query('notificationPreferences') - .withIndex('by_userId_organizationId', (q) => - q.eq('userId', userId).eq('organizationId', organizationId), - ) - .first(); - const value = prefs?.[field]; - return value === undefined ? true : value; -} - -async function insertTaskReviewNotification( - ctx: MutationCtx, - args: { - userId: string; - organizationId: string; - type: TaskReviewNotificationType; - titleKey: string; - bodyKey: string; - params: Record; - resourceType: 'task_review' | 'task'; - resourceId: string; - taskId: Id<'tasks'>; - actorType: 'user' | 'agent' | 'system'; - actorId?: string; - }, -): Promise { - await writeCoalescedNotification(ctx, args); -} - -/** Inbox params stay PII-lean: ids + titles only (org content, not subject PII). */ -function reviewParams( - task: Doc<'tasks'>, - extra: Record = {}, -): Record { - return { - taskId: String(task._id), - projectId: String(task.projectId), - taskTitle: task.title, - ...extra, - }; -} - -/** - * Who submitted the work now waiting on review. The gate is a STATE, so the - * submitter can be an agent run's driver or the person who moved the card — - * the copy names whichever it was instead of asserting "agent work". - */ -export type TaskReviewSubmitter = - | { kind: 'agent'; name?: string } - | { kind: 'user'; userId: string }; - -/** Actionable review request to the designated reviewer (pref gate skipped). */ -export async function notifyTaskReviewRequested( - ctx: MutationCtx, - args: { - task: Doc<'tasks'>; - reviewerUserId: string; - approvalId: Id<'approvals'>; - submitter: TaskReviewSubmitter; - }, -): Promise { - // Nobody is asked to review their own submission: the approval is still - // minted (the board chip and the needs-my-review facet read it), only the - // ping is pointless. - if ( - args.submitter.kind === 'user' && - args.submitter.userId === args.reviewerUserId - ) { - return; - } - - // Agent submissions name the driver (`{agentSlug}`) and fall back to the - // impersonal body when no driver name resolves, so the bell never renders a - // raw token. Human submissions name the person (a proper noun, locale-safe) - // and fall back to a body that doesn't claim an agent did the work. - const actorName = - args.submitter.kind === 'user' - ? await resolveUserDisplayName(ctx, args.submitter.userId) - : null; - const agentName = - args.submitter.kind === 'agent' ? args.submitter.name : undefined; - const bodyKey = - args.submitter.kind === 'agent' - ? agentName - ? 'taskReviewRequestedBody' - : 'taskReviewRequestedBodyNoAgent' - : actorName - ? 'taskReviewRequestedByBody' - : 'taskReviewRequestedBodyHuman'; - - await insertTaskReviewNotification(ctx, { - userId: args.reviewerUserId, - organizationId: args.task.organizationId, - type: 'task_review_requested', - titleKey: 'taskReviewRequested', - bodyKey, - params: reviewParams(args.task, { - approvalId: args.approvalId, - ...(agentName ? { agentSlug: agentName } : {}), - ...(actorName ? { actor: actorName } : {}), - }), - resourceType: 'task_review', - resourceId: args.approvalId, - taskId: args.task._id, - ...(args.submitter.kind === 'user' - ? { actorType: 'user' as const, actorId: args.submitter.userId } - : { actorType: 'agent' as const, actorId: agentName }), - }); -} - -/** - * Heads-up to a freshly designated reviewer while the work is still in flight - * — "you're on the hook for this one". Bell only: the review is not due yet, so - * this deliberately stays out of `ACTIONABLE_NOTIFICATION_TYPES` (no email). - * The actionable request + email follows when the task reaches `in_review`. - */ -export async function notifyTaskReviewerAssigned( - ctx: MutationCtx, - args: { - task: Doc<'tasks'>; - reviewerUserId: string; - actorUserId: string; - }, -): Promise { - if (args.reviewerUserId === args.actorUserId) return; - const actorName = await resolveUserDisplayName(ctx, args.actorUserId); - await insertTaskReviewNotification(ctx, { - userId: args.reviewerUserId, - organizationId: args.task.organizationId, - type: 'task_reviewer_assigned', - titleKey: 'taskReviewerAssigned', - bodyKey: actorName - ? 'taskReviewerAssignedByBody' - : 'taskReviewerAssignedBody', - params: reviewParams(args.task, actorName ? { actor: actorName } : {}), - resourceType: 'task', - resourceId: String(args.task._id), - taskId: args.task._id, - actorType: 'user', - actorId: args.actorUserId, - }); -} - -/** Review outcome to watchers (minus the deciding actor), pref-gated. */ -export async function notifyTaskReviewResolved( - ctx: MutationCtx, - args: { - task: Doc<'tasks'>; - decision: 'approve' | 'request_changes'; - decidedByUserId: string; - recipientUserIds: string[]; - }, -): Promise { - for (const userId of args.recipientUserIds) { - if (userId === args.decidedByUserId) continue; - if ( - !(await prefAllows(ctx, userId, args.task.organizationId, 'taskReview')) - ) { - continue; - } - await insertTaskReviewNotification(ctx, { - userId, - organizationId: args.task.organizationId, - type: 'task_review_resolved', - titleKey: - args.decision === 'approve' - ? 'taskReviewApproved' - : 'taskReviewChangesRequested', - bodyKey: - args.decision === 'approve' - ? 'taskReviewApprovedBody' - : 'taskReviewChangesRequestedBody', - params: reviewParams(args.task), - resourceType: 'task', - resourceId: String(args.task._id), - taskId: args.task._id, - actorType: 'user', - actorId: args.decidedByUserId, - }); - } -} diff --git a/services/platform/backend/core/governance/competence.ts b/services/platform/backend/core/governance/competence.ts deleted file mode 100644 index 794b9a954a..0000000000 --- a/services/platform/backend/core/governance/competence.ts +++ /dev/null @@ -1,54 +0,0 @@ -import type { MutationCtx, QueryCtx } from '../lib/ctx'; -/** Rows scanned per membership check — a member holds a handful of - * competences, not thousands; the cap bounds a pathological org. */ -import type { Doc } from '../lib/rows'; -const COMPETENCE_SCAN_CAP = 200; - -/** Whether the record vouches for its holder RIGHT NOW. */ -export function isCompetenceRecordActive( - record: Pick, 'expiresAt' | 'revokedAt'>, - now: number, -): boolean { - if (record.revokedAt !== undefined) return false; - if (record.expiresAt !== undefined && record.expiresAt <= now) return false; - return true; -} - -/** - * Whether `userId` holds EVERY competence in `required` through unexpired, - * unrevoked records. Returns the vouching record ids so the caller can stamp - * WHICH grants justified the decision (the review check outcome), and the - * missing slugs so a refusal can name what is lacking. An empty `required` - * trivially holds. - */ -export async function holdsAllCompetences( - ctx: QueryCtx | MutationCtx, - organizationId: string, - userId: string, - required: readonly string[], -): Promise<{ holdsAll: boolean; heldRecordIds: string[]; missing: string[] }> { - if (required.length === 0) { - return { holdsAll: true, heldRecordIds: [], missing: [] }; - } - const now = Date.now(); - const rows = await ctx.db - .query('competenceRecords') - .withIndex('by_org_user', (q) => - q.eq('organizationId', organizationId).eq('userId', userId), - ) - .take(COMPETENCE_SCAN_CAP); - const activeBySlug = new Map>(); - for (const row of rows) { - if (isCompetenceRecordActive(row, now)) { - activeBySlug.set(row.competence, row); - } - } - const heldRecordIds: string[] = []; - const missing: string[] = []; - for (const slug of new Set(required)) { - const record = activeBySlug.get(slug); - if (record === undefined) missing.push(slug); - else heldRecordIds.push(String(record._id)); - } - return { holdsAll: missing.length === 0, heldRecordIds, missing }; -} diff --git a/services/platform/backend/core/governance/review_policy.ts b/services/platform/backend/core/governance/review_policy.ts deleted file mode 100644 index 74f0611d5d..0000000000 --- a/services/platform/backend/core/governance/review_policy.ts +++ /dev/null @@ -1,42 +0,0 @@ -/** - * The `review_policy` governance read — who may sign off agent work parked - * at review. - * - * Reads exactly like `approval_policy` (approvals/gate.ts): the per-org JSON - * file under `$TALE_CONFIG_DIR//governance/review-policy.json` is the - * source of truth, mirrored into `configCache` for V8 reads - * (`readPolicyRow`), validated against `reviewPolicyConfigSchema`. A missing - * file means no extra requirement — today's behaviour exactly — and a - * malformed one falls back to absent with a logged warning (the - * approval-policy fallback stance: a broken governance file must not brick - * the review gate, and the warning is the operator's signal to fix it). - * - * Policy WRITES ride the generic governance write/audit path — the file - * store validates against `POLICY_SCHEMAS.review_policy` and the shared - * policy audit covers the change; nothing bespoke lives here. - */ - -import { - type ReviewPolicyConfig, - reviewPolicyConfigSchema, -} from '../../../lib/shared/schemas/governance'; -import type { DatabaseReader } from '../lib/ctx'; -import { readPolicyRow } from './helpers'; - -/** The org's effective review policy, or `null` when none is on file (or - * the file is malformed — logged, treated as absent). */ -export async function readReviewPolicy( - db: DatabaseReader, - organizationId: string, -): Promise { - const row = await readPolicyRow(db, organizationId, 'review_policy'); - if (row === null) return null; - const parsed = reviewPolicyConfigSchema.safeParse(row.config); - if (!parsed.success) { - console.warn( - `[governance] malformed review_policy for org '${organizationId}' — treating it as absent`, - ); - return null; - } - return parsed.data; -} diff --git a/services/platform/backend/core/lib/storage/blob_access.ts b/services/platform/backend/core/lib/storage/blob_access.ts index e08f947288..126f1c5a43 100644 --- a/services/platform/backend/core/lib/storage/blob_access.ts +++ b/services/platform/backend/core/lib/storage/blob_access.ts @@ -36,13 +36,11 @@ import { } from './blob_ref'; import { buildObjectKey, - DEFAULT_PRESIGN_TTL_SEC, resolveOrgObjectStore, s3DeleteObject, s3GetObjectBytes, s3HeadObject, s3PresignGetUrl, - s3PresignPutUrl, s3PutObject, type S3ObjectStore, } from './object_store'; @@ -142,68 +140,6 @@ export async function getBlobUrl( return await s3PresignGetUrl(store, parsed.key, { filename: opts.filename }); } -/** - * Upload handoff for the client: a presigned PUT plus the ref the client will - * bind (the key is known up front). The caller returns `{ url, method, s3Ref }` - * to the browser, which PUTs to `url` then binds `s3Ref`. The `method` union - * is the reused 0.4 wire shape; 0.5 only ever answers `PUT`. - */ -export async function generateBlobUpload( - _ctx: ActionCtx, - orgSlug: string, - opts: { contentType?: string } = {}, -): Promise<{ url: string; method: 'POST' | 'PUT'; s3Ref?: string }> { - const store = await resolveOrgObjectStore(orgSlug); - const key = buildObjectKey(store, orgSlug); - const url = await s3PresignPutUrl(store, key, { - contentType: opts.contentType, - }); - return { url, method: 'PUT', s3Ref: encodeS3Ref(key) }; -} - -export interface ReplacementBlobUploadHandoff { - url: string; - method: 'PUT'; - backend: 's3'; - uploadContentType: string; - uploadExpiresAt: number; - stagingRef: BlobRef; - finalRef: BlobRef; -} - -/** - * Mint a replacement-specific upload capability. - * - * S3 receives two keys: the browser can write only the staging key, while the - * final key is reserved for a create-only server PUT after attestation. The - * `intentNonce` is kept for the reused call shape. - */ -export async function generateReplacementBlobUpload( - _ctx: ActionCtx, - orgSlug: string, - _intentNonce: string, - contentType?: string, -): Promise { - const store = await resolveOrgObjectStore(orgSlug); - const baseContentType = contentType?.trim() || 'application/octet-stream'; - - const stagingKey = buildObjectKey(store, orgSlug); - const finalKey = buildObjectKey(store, orgSlug); - const uploadExpiresAt = Date.now() + DEFAULT_PRESIGN_TTL_SEC * 1000; - return { - url: await s3PresignPutUrl(store, stagingKey, { - contentType: baseContentType, - expiresInSec: DEFAULT_PRESIGN_TTL_SEC, - }), - method: 'PUT', - backend: 's3', - uploadContentType: baseContentType, - uploadExpiresAt, - stagingRef: encodeS3Ref(stagingKey), - finalRef: encodeS3Ref(finalKey), - }; -} - /** * Write attested bytes to a reserved S3 final reference exactly once. * diff --git a/services/platform/backend/core/notifications/actor_name.ts b/services/platform/backend/core/notifications/actor_name.ts deleted file mode 100644 index 01c36d8aa6..0000000000 --- a/services/platform/backend/core/notifications/actor_name.ts +++ /dev/null @@ -1,48 +0,0 @@ -/** - * Resolve a user id to a display name for notification copy. - * - * The resolved name is a proper noun (the person's name or email), so it is - * safe to interpolate into copy in ANY locale. When the id can't be resolved - * — a deleted user, a sentinel like 'system', or a malformed id — this returns - * null so the caller can fall back to an impersonal, fully-localized body - * rather than leaking an English fallback word into DE/FR copy. Never throws. - */ - -import { getUserById } from '../betterAuth/trusted_headers/get_user_by_id'; -import type { QueryCtx } from '../lib/ctx'; - -export async function resolveUserDisplayName( - ctx: QueryCtx, - userId: string | undefined | null, -): Promise { - if (!userId) return null; - try { - const user = await getUserById(ctx, userId); - const name = (user?.name ?? '').trim() || (user?.email ?? '').trim(); - return name || null; - } catch (err) { - console.warn( - `[resolveUserDisplayName] lookup failed for '${userId}': ${ - err instanceof Error ? err.message : String(err) - }`, - ); - return null; - } -} - -/** - * Resolve both the acting user (who filed/cancelled the request) and the - * data-subject (whose data the request concerns) for DSAR notifications. - * `named` is true only when BOTH resolve — the call site uses it to pick the - * fully-named copy variant, falling back to an impersonal (still localized) - * body when either name is unavailable. - */ -export async function resolveActorAndSubject( - ctx: QueryCtx, - actorId: string | undefined | null, - subjectUserId: string | undefined | null, -): Promise<{ actor: string | null; subject: string | null; named: boolean }> { - const actor = await resolveUserDisplayName(ctx, actorId); - const subject = await resolveUserDisplayName(ctx, subjectUserId); - return { actor, subject, named: Boolean(actor && subject) }; -} diff --git a/services/platform/backend/core/provisioning/provision_default_automations.test.ts b/services/platform/backend/core/provisioning/provision_default_automations.test.ts index 069829df1e..4f93d1b28b 100644 --- a/services/platform/backend/core/provisioning/provision_default_automations.test.ts +++ b/services/platform/backend/core/provisioning/provision_default_automations.test.ts @@ -15,7 +15,7 @@ import { fileURLToPath } from 'node:url'; import { describe, expect, it } from 'vitest'; -import { assertAutomationName } from '../automations/store'; +import { assertAutomationName } from '../../domains/automations/store.ts'; import { loadSeedablePacks } from './provision_default_automations'; const REPO_CATALOG = path.resolve( diff --git a/services/platform/backend/core/tasks/review_shared.ts b/services/platform/backend/core/tasks/review_shared.ts index 1418a86e1c..871b718018 100644 --- a/services/platform/backend/core/tasks/review_shared.ts +++ b/services/platform/backend/core/tasks/review_shared.ts @@ -1,515 +1,12 @@ /** - * Task-review shared core: reviewer resolution, the review-gate mint, and the - * gate's close-on-leave. - * - * `requestTaskReview` is the ONE door into the gate, for every way a task can - * reach `in_review` — a person moving the card (`mutations.updateTaskStatus`, - * `moveTask`, `bulkUpdateTasks`), an agent run's settle park - * (`internal_mutations.agentUpdateTaskStatus`), or an automation. - * `closePendingTaskReviewOnStatusLeave` is the matching exit: every way a task - * can LEAVE `in_review` closes the gate again, so a decided/mooted review never - * lingers as a bell or a Needs-my-review row. Both live apart from - * `review_mutations.ts` so the agent status mutation can mint/close - * transactionally with its transition without an import cycle - * (mutations → internal_mutations → here; review_mutations → here). + * The `review_policy` refusal codes the task-review gate throws. Shared with + * `domains/tasks/service.ts`, whose batch door (`bulkUpdateTasks`) skips a + * task on one of these instead of aborting the whole batch. The gate itself — + * reviewer resolution, the mint, the close-on-leave — lives in + * `domains/tasks/reviews.ts`. */ -import { AppError } from '../../../lib/shared/errors/app-error'; -import { isRecord } from '../../../lib/utils/type-utils'; -import { createAuditLog } from '../audit_logs/helpers'; -import { dismissReviewRequestNotifications } from '../collab/dismiss_review_notifications'; -import { - notifyTaskReviewRequested, - notifyTaskReviewResolved, - type TaskReviewSubmitter, -} from '../collab/notify_task_reviews'; -import { holdsAllCompetences } from '../governance/competence'; -import { readReviewPolicy } from '../governance/review_policy'; -import type { MutationCtx } from '../lib/ctx'; -import type { Doc, Id } from '../lib/rows'; -import { resolveProjectAccessForUser } from '../projects/resolve_project_access'; -import { recordActivity, TASK_METRIC_ACTIONS } from './helpers'; - -/** As much of an approval row as the two round/run readers below touch. */ -interface ApprovalMetadataFields { - metadata?: unknown; -} - -/** The review round an approval was minted for; rows predating the round key - * (or with malformed metadata) read as round 0. Exported for unit tests. */ -export function approvalRound(approval: ApprovalMetadataFields): number { - const metadata: unknown = approval.metadata; - if (!isRecord(metadata)) return 0; - return typeof metadata.round === 'number' ? metadata.round : 0; -} - -/** The agent run a workflow-free review was minted for (settle mints key - * their idempotency on this); absent on workflow-era rows. */ -export function approvalRunId( - approval: ApprovalMetadataFields, -): string | undefined { - const metadata: unknown = approval.metadata; - if (!isRecord(metadata)) return undefined; - return typeof metadata.runId === 'string' ? metadata.runId : undefined; -} - -/** - * Resolve who should review a task parked at `in_review`. Revalidated at - * every call (mint/read time) so a designee who lost project access falls - * through the chain instead of silently swallowing review requests: - * explicit `reviewerUserId` → human task creator → project creator — the - * first candidate who still holds project `canEdit` wins; else undefined. - * The durable field stores EXPLICIT designation only; this chain IS the - * default logic, evaluated at need and never persisted (a stale designation - * is left in place — lazy cleanup, no membership hooks). - */ -export async function resolveReviewer( - ctx: MutationCtx, - task: Doc<'tasks'>, -): Promise { - const project = await ctx.db.get(task.projectId); - const candidates = [ - task.reviewerUserId, - task.createdByType === 'user' ? task.createdBy : undefined, - project?.createdBy, - ]; - const seen = new Set(); - for (const candidate of candidates) { - if (candidate === undefined || seen.has(candidate)) continue; - seen.add(candidate); - const access = await resolveProjectAccessForUser(ctx, task.projectId, { - userId: candidate, - organizationId: task.organizationId, - }); - if (access.canEdit) return candidate; - } - return undefined; -} - -/** - * The driver's display name for review copy: the project agent's name, or the - * owning automation's store name. Undefined for human/unassigned drivers. - */ -async function resolveDriverDisplayName( - ctx: MutationCtx, - task: Doc<'tasks'>, -): Promise { - if (task.assigneeId === undefined) return undefined; - if (task.assigneeType === 'agent') { - const agentId = ctx.db.normalizeId('projectAgents', task.assigneeId); - if (agentId === null) return undefined; - const agent = await ctx.db.get(agentId); - return agent?.name; - } - if (task.assigneeType === 'app') return task.assigneeId; - return undefined; -} - -/** - * What put the task in front of a reviewer. The gate belongs to the STATE, not - * to the worker: an agent run's settle park, a person moving the card, and an - * automation all open the same gate — the trigger only decides the idempotency - * key and whose name the request copy carries. - */ -export type TaskReviewTrigger = - | { kind: 'agent_run'; runId: Id<'projectAgentRuns'> } - | { kind: 'human'; actorId: string } - | { kind: 'automation'; slug?: string }; - -/** - * Open the review gate on a task that just reached `in_review`. MUST run in the - * same transaction as the status flip: the agent settle action's burned-claim - * fallback can replay the sequence, and the find-or-insert below is what keeps - * two racers from minting twice. - * - * Idempotency depends on the trigger. An agent run keys on its `runId`, so a - * replayed settle returns the existing row; a human/automation submission keys - * on "is a review already pending for this task" — one open gate per task, so - * a re-designation or re-submission on a task SITTING at `in_review` does not - * mint a second request (nor a second bell). A leave from `in_review` closes - * the gate (`closePendingTaskReviewOnStatusLeave` below), so a bounce out and - * back in withdraws the old request and mints a fresh one. - * - * A fresh mint SUPERSEDES any older pending review on the task (rejected + - * `supersededBy`, bells dismissed) — newest submission wins — then notifies the - * resolved reviewer. When no reviewer resolves, the review is still minted - * (`requestedFor: null` — the board chip renders) and only the targeted - * request notification is skipped; watchers already get the generic - * status-change bell from the transition itself. - */ -export async function requestTaskReview( - ctx: MutationCtx, - args: { task: Doc<'tasks'>; trigger: TaskReviewTrigger }, -): Promise<{ approvalId: Id<'approvals'>; minted: boolean }> { - const { task, trigger } = args; - const runKey = trigger.kind === 'agent_run' ? trigger.runId : undefined; - - const prior: Doc<'approvals'>[] = []; - for await (const approval of ctx.db - .query('approvals') - .withIndex('by_resource', (q) => - q.eq('resourceType', 'task_review').eq('resourceId', String(task._id)), - )) { - prior.push(approval); - } - const existing = - runKey === undefined - ? prior.find((approval) => approval.status === 'pending') - : prior.find((approval) => approvalRunId(approval) === runKey); - if (existing) { - return { approvalId: existing._id, minted: false }; - } - - const reviewer = await resolveReviewer(ctx, task); - const driverName = await resolveDriverDisplayName(ctx, task); - const approvalId = await ctx.db.insert('approvals', { - organizationId: task.organizationId, - resourceType: 'task_review', - resourceId: String(task._id), - priority: 'high', - status: 'pending', - metadata: { - taskId: String(task._id), - projectId: String(task.projectId), - agentSlug: driverName ?? null, - requestedFor: reviewer ?? null, - round: prior.length, - // No stored question: readers render their own localized copy — - // persisting an English sentence here would ship untranslated. - question: null, - ...(runKey !== undefined ? { runId: runKey } : {}), - }, - }); - - const now = Date.now(); - for (const stale of prior) { - if (stale.status !== 'pending') continue; - await ctx.db.patch(stale._id, { - status: 'rejected', - reviewedAt: now, - metadata: { - ...(isRecord(stale.metadata) ? stale.metadata : {}), - supersededBy: approvalId, - }, - }); - await dismissReviewRequestNotifications(ctx, { - organizationId: task.organizationId, - approvalId: stale._id, - taskId: task._id, - }); - } - - if (reviewer !== undefined) { - await notifyTaskReviewRequested(ctx, { - task, - reviewerUserId: reviewer, - approvalId, - submitter: reviewSubmitter(trigger, driverName), - }); - } - - return { approvalId, minted: true }; -} - -/** - * Whose name the request copy carries. An agent-driven task names its driver - * even when a human moved the card (the work was the agent's); a human - * submission on a human/unassigned task names the person who submitted it. - */ -function reviewSubmitter( - trigger: TaskReviewTrigger, - driverName: string | undefined, -): TaskReviewSubmitter { - if (trigger.kind === 'human' && driverName === undefined) { - return { kind: 'user', userId: trigger.actorId }; - } - return { - kind: 'agent', - ...(driverName !== undefined ? { name: driverName } : {}), - }; -} - -/** The recorded decision on a `task_review` approval (`metadata.response`), - * written by `respondToTaskReview` and the leave-to-done close below. */ -export interface TaskReviewResponse { - decision: 'approve' | 'request_changes'; - feedback?: string; - respondedBy: string; - timestamp: number; - /** Check outcomes of the org's `review_policy`, recorded only when the - * policy demanded them (absent policy ⇒ absent fields — today's shape). */ - independentReviewer?: boolean; - competences?: { - required: string[]; - heldRecordIds: string[]; - checkedAt: number; - }; -} - export const REVIEW_POLICY_REFUSAL_CODES = [ 'REVIEW_INDEPENDENT_REVIEWER_REQUIRED', 'REVIEW_COMPETENCE_REQUIRED', ] as const; - -/** Whether a throw is one of the `review_policy` refusals — batch callers - * (`bulkUpdateTasks`) skip the task instead of aborting the whole batch. */ -export function isReviewPolicyRefusal(error: unknown): boolean { - if (!(error instanceof AppError)) return false; - const data: unknown = error.data; - if (!isRecord(data) || typeof data.code !== 'string') return false; - return (REVIEW_POLICY_REFUSAL_CODES as readonly string[]).includes(data.code); -} - -/** - * The org `review_policy` gate on WHO may decide a review — shared between - * `respondToTaskReview` and the leave-to-done close, so the status picker and - * the board drag cannot decide what the respond door would refuse. Pure check: - * throws a coded AppError on refusal, writes nothing, and returns the - * outcomes to stamp on the recorded response. Absent (or malformed — logged - * and treated as absent) policy means exactly the open behaviour: any project - * editor. - */ -export async function checkReviewPolicyForResponder( - ctx: MutationCtx, - args: { - approval: Doc<'approvals'>; - task: Doc<'tasks'>; - responderUserId: string; - now: number; - }, -): Promise> { - const { approval, task, responderUserId, now } = args; - const reviewPolicy = await readReviewPolicy(ctx.db, task.organizationId); - let independentReviewer: boolean | undefined; - let competences: TaskReviewResponse['competences']; - if (reviewPolicy?.requireIndependentReviewer === true) { - const settledRunKey = approvalRunId(approval); - const runId = - settledRunKey === undefined - ? null - : ctx.db.normalizeId('projectAgentRuns', settledRunKey); - const run = runId === null ? null : await ctx.db.get(runId); - if (run !== null && run.taskId === task._id) { - // The reviewed run's driver is the human who kicked it - // (`projectAgentRuns.startedBy` — every task-lane trigger is a - // person's act). Independence = the responder is someone else. - if (run.startedBy === responderUserId) { - throw new AppError({ - code: 'REVIEW_INDEPENDENT_REVIEWER_REQUIRED', - message: - 'This organization requires an independent reviewer: the person who started the run cannot approve its work.', - }); - } - } else if (task.createdBy === responderUserId) { - // Review rows without run linkage (human/automation mints, workflow-era - // rows, or a key that no longer resolves to a projectAgentRuns row) - // cannot recover the driver. Conservatively require the responder to - // differ from the task's creator — the closest proxy for the person - // whose work is under review. - throw new AppError({ - code: 'REVIEW_INDEPENDENT_REVIEWER_REQUIRED', - message: - "This organization requires an independent reviewer: the reviewed run's driver could not be resolved, so the task creator cannot respond.", - }); - } - independentReviewer = true; - } - const requiredCompetences = reviewPolicy?.requiredCompetences ?? []; - if (requiredCompetences.length > 0) { - const held = await holdsAllCompetences( - ctx, - task.organizationId, - responderUserId, - requiredCompetences, - ); - if (!held.holdsAll) { - throw new AppError({ - code: 'REVIEW_COMPETENCE_REQUIRED', - message: `Responding to this review requires the competence(s): ${held.missing.join(', ')}. Ask an org admin to grant them.`, - missing: held.missing, - }); - } - competences = { - required: [...requiredCompetences], - heldRecordIds: held.heldRecordIds, - checkedAt: now, - }; - } - return { - ...(independentReviewer !== undefined ? { independentReviewer } : {}), - ...(competences !== undefined ? { competences } : {}), - }; -} - -/** Unmuted human watchers of a task — the audience for a review outcome. */ -export async function collectTaskWatcherIds( - ctx: MutationCtx, - taskId: Id<'tasks'>, -): Promise { - const watcherIds = new Set(); - for await (const sub of ctx.db - .query('taskSubscriptions') - .withIndex('by_task', (q) => q.eq('taskId', taskId))) { - if (sub.subscriberType === 'user' && !sub.muted) { - watcherIds.add(sub.subscriberId); - } - } - return [...watcherIds]; -} - -/** Who moved the task out of `in_review`. A person's leave to `done` IS the - * approve; every other leave — any target for a system actor (an agent's - * `task_update_status`, an automation's transition, an external-sync close) — - * withdraws the request: no human decided, so no decision is recorded and the - * `review_policy` gate does not apply (it would wedge the non-human lane). */ -export type TaskReviewLeaveActor = - | { kind: 'user'; userId: string; email?: string } - | { kind: 'system'; actorId: string }; - -/** - * Close the review gate when a task leaves `in_review`. MUST run in the same - * transaction as the status write, from EVERY path that can move a task out of - * `in_review` — otherwise the pending approval keeps ringing bells and holding - * the Needs-my-review facet for work nobody is gated on any more. - * - * Self-guarding no-op unless the transition actually leaves `in_review` or the - * task holds a pending workflow-free review. Workflow-era rows (with a - * `wfExecutionId`) are left untouched: they belong to a paused execution's own - * request/respond protocol, not to the board state. - * - * A human leave to `done` records the SAME approve as `respondToTaskReview` - * (policy check, response metadata, `task.review_responded` audit, resolved - * notification, bell dismissal) — minus the status write, which is the - * caller's. Every other leave marks the row rejected with a `withdrawn` - * marker and dismisses the bells; the status change itself is the record. - * - * Ordering contract: ALL validation (the `review_policy` check) happens before - * ANY write, so a batch caller may catch `isReviewPolicyRefusal` errors and - * skip the task knowing nothing was half-written. - */ -export async function closePendingTaskReviewOnStatusLeave( - ctx: MutationCtx, - args: { - /** The task as loaded BEFORE the status write (status still `in_review`). */ - task: Doc<'tasks'>; - toStatus: Doc<'tasks'>['status']; - actor: TaskReviewLeaveActor; - }, -): Promise { - const { task, toStatus, actor } = args; - if (task.status !== 'in_review' || toStatus === 'in_review') return; - - const pending: Doc<'approvals'>[] = []; - for await (const approval of ctx.db - .query('approvals') - .withIndex('by_resource', (q) => - q.eq('resourceType', 'task_review').eq('resourceId', String(task._id)), - )) { - if (approval.status !== 'pending') continue; - if (approval.wfExecutionId !== undefined) continue; - pending.push(approval); - } - if (pending.length === 0) return; - - const now = Date.now(); - const approves = actor.kind === 'user' && toStatus === 'done'; - - if (approves) { - // Validate every row before writing anything (see the ordering contract). - const outcomes = new Map< - Id<'approvals'>, - Pick - >(); - for (const approval of pending) { - outcomes.set( - approval._id, - await checkReviewPolicyForResponder(ctx, { - approval, - task, - responderUserId: actor.userId, - now, - }), - ); - } - - for (const approval of pending) { - const outcome = outcomes.get(approval._id) ?? {}; - const response: TaskReviewResponse = { - decision: 'approve', - respondedBy: actor.userId, - timestamp: now, - ...outcome, - }; - const metadata = isRecord(approval.metadata) ? approval.metadata : {}; - await ctx.db.patch(approval._id, { - status: 'completed', - approvedBy: actor.userId, - reviewedAt: now, - metadata: { ...metadata, response }, - }); - await dismissReviewRequestNotifications(ctx, { - organizationId: task.organizationId, - approvalId: approval._id, - taskId: task._id, - }); - const settledRunId = approvalRunId(approval); - const auditMetadata: Record = { - ...(settledRunId !== undefined ? { runId: settledRunId } : {}), - ...outcome, - }; - await createAuditLog(ctx, { - organizationId: task.organizationId, - actorId: actor.userId, - actorEmail: actor.email, - actorType: 'user', - action: 'task.review_responded', - category: 'data', - resourceType: 'task', - resourceId: String(task._id), - resourceName: task.title, - newState: { decision: 'approve' }, - ...(Object.keys(auditMetadata).length > 0 - ? { metadata: auditMetadata } - : {}), - status: 'success', - }); - } - - // One metric and one watcher notification per gesture, however many rows - // the degenerate multi-pending case held. - await recordActivity(ctx, { - task, - actorType: 'user', - actorId: actor.userId, - action: TASK_METRIC_ACTIONS.reviewPassed, - }); - await notifyTaskReviewResolved(ctx, { - task, - decision: 'approve', - decidedByUserId: actor.userId, - recipientUserIds: await collectTaskWatcherIds(ctx, task._id), - }); - return; - } - - for (const approval of pending) { - const metadata = isRecord(approval.metadata) ? approval.metadata : {}; - await ctx.db.patch(approval._id, { - status: 'rejected', - reviewedAt: now, - metadata: { - ...metadata, - withdrawn: { - toStatus, - by: actor.kind === 'user' ? actor.userId : actor.actorId, - at: now, - }, - }, - }); - await dismissReviewRequestNotifications(ctx, { - organizationId: task.organizationId, - approvalId: approval._id, - taskId: task._id, - }); - } -} diff --git a/services/platform/env.sh b/services/platform/env.sh index 6a4a62c6a6..c78b4663e0 100644 --- a/services/platform/env.sh +++ b/services/platform/env.sh @@ -63,11 +63,3 @@ ensure_instance_secret() { export INSTANCE_SECRET="local-dev-insecure-secret" fi } - -# Tools that derive keys (generate-admin-key) need a real 64-hex secret. -ensure_hex_instance_secret() { - if ! echo "${INSTANCE_SECRET:-}" | grep -Eq '^[0-9a-fA-F]{64}$'; then - echo "Error: INSTANCE_SECRET must be a 64-character hex string. Set INSTANCE_SECRET in your .env." >&2 - exit 1 - fi -}