diff --git a/.compliance/ropa.yaml b/.compliance/ropa.yaml index 95368450..09544fc7 100644 --- a/.compliance/ropa.yaml +++ b/.compliance/ropa.yaml @@ -466,6 +466,43 @@ processing_activities: - writes_staged_for_explicit_user_approval - plaintext_personnummer_blocked_from_staging + - id: mcp.async_task_handles + name: MCP-uppgiftshandtag för långkörande verktygsanrop + purpose: >- + Låta en behörig MCP-klient hämta resultatet av ett långkörande + verktygsanrop (t.ex. generering av arkivpaket) via ett hållbart + uppgifts-id i stället för att hålla anslutningen öppen. Raden lagrar + verktygets svar tills klienten hämtat det; endast den skapande + användaren kan läsa, och utgångna rader rensas löpande vid nya anrop. + lawful_basis: art_6_1_b + special_category_basis: null + controller: gnubok-tenant + processor: supabase + data_subjects: + - business_owner + - company_member + data_categories: + - user.financial + - user.activity_timestamp + recipients: + - name: Supabase + country: EU + role: processor + international_transfers: + applicable: false + mechanism: null + note: EU-only processor. + retention: + duration: 1h + basis: transient_tool_call_state_storage_limitation + stored_in: + - mcp_tasks + security_measures: + - rls_user_scoped_select # mcp_tasks SELECT: auth.uid() = user_id + - service_role_only_writes + - creator_scoped_task_polling # tasks/get requires the creating user + - expired_rows_purged_on_creation + - id: arsredovisning.signature_evidence name: Underskriftsbevis för årsredovisning purpose: >- diff --git a/extensions/general/mcp-server/__tests__/mcp-tasks.test.ts b/extensions/general/mcp-server/__tests__/mcp-tasks.test.ts new file mode 100644 index 00000000..ec6988dd --- /dev/null +++ b/extensions/general/mcp-server/__tests__/mcp-tasks.test.ts @@ -0,0 +1,300 @@ +/** + * Tests for the MCP Tasks extension (io.modelcontextprotocol/tasks): + * capability gating, CreateTaskResult, post-response completion writes, + * and the tasks/get / tasks/update / tasks/cancel methods. Does NOT re-test + * audit-package generation itself (covered by audit-package.test.ts). + */ +import { describe, it, expect, vi, beforeEach } from 'vitest' +import { eventBus } from '@/lib/events/bus' +import { isTaskCapableClient, TASKS_EXTENSION_ID } from '../tasks' + +const mocks = vi.hoisted(() => ({ + scopes: [] as string[], + taskInserts: [] as Record[], + taskUpdates: [] as Record[], + taskRow: null as Record | null, +})) + +vi.mock('@/lib/auth/api-keys', async (importOriginal) => { + const actual = await importOriginal() + mocks.scopes = [...actual.ALL_SCOPES] + + const membershipChain: unknown = new Proxy( + {}, + { + get(_t, prop) { + if (prop === 'then') { + return (resolve: (v: unknown) => void) => + resolve({ + data: { company_id: '11111111-1111-4111-8111-111111111111', role: 'owner' }, + error: null, + }) + } + return () => membershipChain + }, + } + ) + + const makeChain = (result: unknown): unknown => + new Proxy( + {}, + { + get(_t, prop) { + if (prop === 'then') { + return (resolve: (v: unknown) => void) => resolve(result) + } + if (prop === 'single') { + return () => Promise.resolve(result) + } + return () => makeChain(result) + }, + } + ) + + const tasksBuilder = { + delete: () => makeChain({ data: null, error: null }), + insert: (row: Record) => { + mocks.taskInserts.push(row) + return { + select: () => ({ + single: async () => ({ + data: { + id: 'task-1', + company_id: row.company_id, + user_id: row.user_id, + tool_name: row.tool_name, + status: 'working', + status_message: null, + result: null, + error: null, + poll_interval_ms: 2000, + ttl_ms: 3600000, + created_at: '2026-07-29T09:00:00.000Z', + }, + error: null, + }), + }), + } + }, + update: (patch: Record) => { + mocks.taskUpdates.push(patch) + return makeChain({ data: null, error: null }) + }, + select: () => + makeChain( + mocks.taskRow + ? { data: mocks.taskRow, error: null } + : { data: null, error: { message: 'not found' } } + ), + } + + return { + ...actual, + extractBearerToken: vi.fn().mockReturnValue('test-token'), + validateApiKey: vi.fn().mockResolvedValue({ + userId: 'user-1', + companyId: '11111111-1111-4111-8111-111111111111', + scopes: mocks.scopes, + apiKeyId: 'key-1', + apiKeyName: 'Test key', + mode: 'live', + }), + createServiceClientNoCookies: vi.fn(() => ({ + from: (table: string) => { + if (table === 'mcp_tasks') return tasksBuilder + if (table === 'company_members') return membershipChain + // fiscal_periods (and everything else) resolves to no rows, so the + // audit-package execution fails fast with "Fiscal period not found". + return makeChain({ data: null, error: { message: 'not found' }, count: 0 }) + }, + })), + } +}) + +import { handleMcpRequest } from '../server' + +const TASK_META = { + 'io.modelcontextprotocol/protocolVersion': '2026-07-28', + 'io.modelcontextprotocol/clientCapabilities': { + extensions: { [TASKS_EXTENSION_ID]: {} }, + }, +} + +function mcpRequest(method: string, params?: Record): Request { + return new Request('http://localhost:3000/api/extensions/ext/mcp-server/mcp', { + method: 'POST', + headers: { 'Content-Type': 'application/json', Authorization: 'Bearer test-token' }, + body: JSON.stringify({ jsonrpc: '2.0', id: 1, method, ...(params ? { params } : {}) }), + }) +} + +async function readBody(request: Request): Promise<{ + result?: Record + error?: { code: number; message: string } +}> { + const response = await handleMcpRequest(request) + const body = await response.json() + return { result: body.result, error: body.error } +} + +/** The after() fallback runs the job as a floating promise: let it settle. */ +async function settleBackgroundWork(): Promise { + await new Promise((resolve) => setTimeout(resolve, 0)) + await new Promise((resolve) => setTimeout(resolve, 0)) +} + +describe('MCP Tasks extension', () => { + beforeEach(() => { + vi.clearAllMocks() + eventBus.clear() + mocks.taskInserts.length = 0 + mocks.taskUpdates.length = 0 + mocks.taskRow = null + }) + + describe('capability detection', () => { + it('detects the tasks extension in per-request capabilities', () => { + expect(isTaskCapableClient(TASK_META as Record)).toBe(true) + expect(isTaskCapableClient({})).toBe(false) + expect( + isTaskCapableClient({ + 'io.modelcontextprotocol/clientCapabilities': { extensions: {} }, + }) + ).toBe(false) + }) + + it('advertises the tasks extension in server/discover', async () => { + const { result } = await readBody(mcpRequest('server/discover')) + const capabilities = result?.capabilities as { extensions: Record } + expect(capabilities.extensions[TASKS_EXTENSION_ID]).toEqual({}) + }) + }) + + describe('CreateTaskResult', () => { + it('returns a task handle to a task-capable client and completes after the response', async () => { + const { result } = await readBody( + mcpRequest('tools/call', { + name: 'gnubok_audit_package', + arguments: { fiscal_period_id: '22222222-2222-4222-8222-222222222222' }, + _meta: TASK_META, + }) + ) + expect(result?.resultType).toBe('task') + const task = result?.task as Record + expect(task.taskId).toBe('task-1') + expect(task.status).toBe('working') + expect(typeof task.pollIntervalMs).toBe('number') + expect(typeof task.ttlMs).toBe('number') + expect(mocks.taskInserts).toHaveLength(1) + + await settleBackgroundWork() + // The mocked DB has no fiscal period, so the execution fails and the + // task completes with the standard isError envelope. + expect(mocks.taskUpdates).toHaveLength(1) + expect(mocks.taskUpdates[0].status).toBe('completed') + const stored = mocks.taskUpdates[0].result as Record + expect(stored.isError).toBe(true) + expect(stored.resultType).toBe('complete') + }) + + it('never returns a task to a client that did not declare the extension', async () => { + const { result } = await readBody( + mcpRequest('tools/call', { + name: 'gnubok_audit_package', + arguments: { fiscal_period_id: '22222222-2222-4222-8222-222222222222' }, + _meta: { 'io.modelcontextprotocol/protocolVersion': '2026-07-28' }, + }) + ) + expect(result?.resultType).not.toBe('task') + expect(result?.isError).toBe(true) + expect(mocks.taskInserts).toHaveLength(0) + }) + + it('keeps estimate_only synchronous even for task-capable clients', async () => { + const { result } = await readBody( + mcpRequest('tools/call', { + name: 'gnubok_audit_package', + arguments: { + fiscal_period_id: '22222222-2222-4222-8222-222222222222', + estimate_only: true, + }, + _meta: TASK_META, + }) + ) + expect(result?.resultType).not.toBe('task') + expect(mocks.taskInserts).toHaveLength(0) + }) + }) + + describe('tasks/get', () => { + it('returns the working state while the task runs', async () => { + mocks.taskRow = { + id: 'task-1', + company_id: '11111111-1111-4111-8111-111111111111', + user_id: 'user-1', + tool_name: 'gnubok_audit_package', + status: 'working', + status_message: null, + result: null, + error: null, + poll_interval_ms: 2000, + ttl_ms: 3600000, + created_at: '2026-07-29T09:00:00.000Z', + } + const { result } = await readBody(mcpRequest('tasks/get', { taskId: 'task-1' })) + expect(result?.status).toBe('working') + expect(result?.result).toBeUndefined() + expect(result?.resultType).toBe('complete') + }) + + it('returns the stored tool result on completion', async () => { + const storedResult = { + resultType: 'complete', + content: [{ type: 'text', text: '{"file_name":"arkiv.zip"}' }], + structuredContent: { file_name: 'arkiv.zip' }, + } + mocks.taskRow = { + id: 'task-1', + company_id: '11111111-1111-4111-8111-111111111111', + user_id: 'user-1', + tool_name: 'gnubok_audit_package', + status: 'completed', + status_message: null, + result: storedResult, + error: null, + poll_interval_ms: 2000, + ttl_ms: 3600000, + created_at: '2026-07-29T09:00:00.000Z', + } + const { result } = await readBody(mcpRequest('tasks/get', { taskId: 'task-1' })) + expect(result?.status).toBe('completed') + expect(result?.result).toEqual(storedResult) + }) + + it('rejects unknown task ids with invalid params', async () => { + const { error } = await readBody(mcpRequest('tasks/get', { taskId: 'nope' })) + expect(error?.code).toBe(-32602) + }) + + it('requires taskId', async () => { + const { error } = await readBody(mcpRequest('tasks/get', {})) + expect(error?.code).toBe(-32602) + }) + }) + + describe('tasks/cancel and tasks/update', () => { + it('acknowledges cancellation and flips a working row', async () => { + const { result } = await readBody(mcpRequest('tasks/cancel', { taskId: 'task-1' })) + expect(result?.resultType).toBe('complete') + expect(mocks.taskUpdates).toHaveLength(1) + expect(mocks.taskUpdates[0].status).toBe('cancelled') + }) + + it('acknowledges tasks/update as a no-op (no input_required flows yet)', async () => { + const { result } = await readBody( + mcpRequest('tasks/update', { taskId: 'task-1', inputResponses: {} }) + ) + expect(result?.resultType).toBe('complete') + }) + }) +}) diff --git a/extensions/general/mcp-server/__tests__/protocol-2026-07-28.test.ts b/extensions/general/mcp-server/__tests__/protocol-2026-07-28.test.ts index ba5add9b..d3fbe0a8 100644 --- a/extensions/general/mcp-server/__tests__/protocol-2026-07-28.test.ts +++ b/extensions/general/mcp-server/__tests__/protocol-2026-07-28.test.ts @@ -100,7 +100,7 @@ describe('MCP spec revision 2026-07-28', () => { ]) const capabilities = result?.capabilities as Record expect(capabilities.tools).toEqual({ listChanged: false }) - expect(capabilities.extensions).toEqual({ 'io.modelcontextprotocol/ui': {} }) + expect(capabilities.extensions).toMatchObject({ 'io.modelcontextprotocol/ui': {} }) expect(typeof result?.ttlMs).toBe('number') expect(result?.cacheScope).toBe('private') const meta = result?._meta as Record> @@ -317,7 +317,7 @@ describe('MCP spec revision 2026-07-28', () => { expect(result?.protocolVersion).toBe('2025-06-18') expect((result?.serverInfo as Record).name).toBe('gnubok') const capabilities = result?.capabilities as Record - expect(capabilities.extensions).toEqual({ 'io.modelcontextprotocol/ui': {} }) + expect(capabilities.extensions).toMatchObject({ 'io.modelcontextprotocol/ui': {} }) }) it('negotiates an initialize requesting 2026-07-28 down to the handshake default', async () => { diff --git a/extensions/general/mcp-server/server.ts b/extensions/general/mcp-server/server.ts index a3a398ee..a448eb60 100644 --- a/extensions/general/mcp-server/server.ts +++ b/extensions/general/mcp-server/server.ts @@ -1,4 +1,12 @@ import { NextResponse, after } from 'next/server' +import { + TASKS_EXTENSION_ID, + isTaskCapableClient, + createMcpTask, + resolveMcpTask, + taskToWire, + type McpTaskRow, +} from './tasks' import { extractBearerToken, validateApiKey, @@ -244,6 +252,11 @@ interface McpTool { // _meta.ui.resourceUri on the RESULT, so the host renders the widget only when // asked. (Contrast _meta above, on the definition, which renders on every call.) uiResourceUri?: string + // Tasks extension: when this predicate returns true for a call from a + // task-capable client, the dispatcher returns a CreateTaskResult and runs + // execute() after the response instead of blocking on it. Not serialized + // into tools/list. + shouldRunAsTask?: (args: Record) => boolean execute: ( args: Record, companyId: string, @@ -12234,6 +12247,10 @@ export const tools: McpTool[] = [ idempotentHint: true, // repeat calls produce equivalent archives, fresh URL openWorldHint: false, }, + // Archive generation is the one genuinely long-running synchronous call + // in the catalog: task-capable clients get a durable handle instead of a + // multi-minute blocking response. Size estimates stay synchronous. + shouldRunAsTask: (args) => args.estimate_only !== true, async execute(args, companyId, userId, supabase) { const fiscalPeriodId = args.fiscal_period_id as string if (!fiscalPeriodId) throw new Error('fiscal_period_id is required') @@ -15463,9 +15480,13 @@ const SERVER_CAPABILITIES = { tools: { listChanged: false }, resources: { listChanged: false }, prompts: { listChanged: false }, - // MCP Apps (ratified extension): widgets are served as ui:// resources and - // referenced from tool _meta.ui.resourceUri (see widgets/). - extensions: { 'io.modelcontextprotocol/ui': {} }, + extensions: { + // MCP Apps (ratified extension): widgets are served as ui:// resources + // and referenced from tool _meta.ui.resourceUri (see widgets/). + 'io.modelcontextprotocol/ui': {}, + // MCP Tasks: durable handles for long-running tool calls (see tasks.ts). + [TASKS_EXTENSION_ID]: {}, + }, } /** @@ -15858,6 +15879,9 @@ export async function handleMcpRequest(request: Request): Promise { // Revisions are ISO dates, so string comparison orders them correctly. const statelessClient = typeof metaVersion === 'string' && metaVersion >= STATELESS_PROTOCOL_VERSION + // Tasks extension: only a client that declared it in THIS request's + // capabilities may ever receive a CreateTaskResult. + const taskCapable = statelessClient && isTaskCapableClient(requestMeta) // Standard request headers (2026-07-28): when present they must agree with // the JSON-RPC body. Absence stays accepted: this server supports @@ -16238,6 +16262,90 @@ export async function handleMcpRequest(request: Request): Promise { // mcp.next_hint_followed when the agent's behaviour matches the hint. checkAndEmitNextHintFollowed(sessionId, toolName, actor, userId, effectiveCompanyId) + // ── Tasks extension (io.modelcontextprotocol/tasks) ── + // Long-running tools return a durable handle immediately to a client + // that declared the extension; the work completes after the response + // (after() keeps the function alive) and the result lands in mcp_tasks + // for tasks/get polling. Runs after every auth/scope/capability guard + // so nothing is ever started for a call that would have been refused. + if (taskCapable && tool.shouldRunAsTask?.(toolArgs)) { + const task = await createMcpTask(supabase, { + companyId: effectiveCompanyId, + userId, + apiKeyId, + toolName, + }) + const taskStartedAt = Date.now() + emitAfterResponse(async () => { + try { + const rawResult = await tool.execute(toolArgs, effectiveCompanyId, userId, supabase, actor) + const canonicalResult = addCompanyToTopLevelNext(rawResult, effectiveCompanyId) + const result = projectMcpPayload(canonicalResult, toolNamespace) + const stored: Record = { + resultType: 'complete', + content: [{ type: 'text', text: JSON.stringify(result, null, 2) }], + } + if (result !== null && result !== undefined) { + stored.structuredContent = + typeof result === 'object' && !Array.isArray(result) ? result : { value: result } + } + await resolveMcpTask(supabase, task.id, { status: 'completed', result: stored }) + emitToolCallTelemetry({ + tool: toolName, + requiredScope: requiredScope ?? null, + actor, + latencyMs: Date.now() - taskStartedAt, + success: true, + isError: false, + errorCode: null, + errorKind: null, + errorMessage: null, + requestId: id ?? null, + userId, + companyId: effectiveCompanyId, + }) + } catch (err) { + // Tool failures complete the task with the standard isError + // envelope: exactly what the synchronous call would have + // returned. `failed` stays reserved for infrastructure errors. + const structured = toToolError(err, { toolName }) + const publicStructured = projectMcpPayload(structured, toolNamespace) + await resolveMcpTask(supabase, task.id, { + status: 'completed', + result: { + resultType: 'complete', + content: [{ type: 'text', text: JSON.stringify(publicStructured, null, 2) }], + isError: true, + }, + statusMessage: structured.error.message_sv, + }).catch((updateErr) => { + log.error('Failed to store MCP task failure result', { taskId: task.id, updateErr }) + }) + emitToolCallTelemetry({ + tool: toolName, + requiredScope: requiredScope ?? null, + actor, + latencyMs: Date.now() - taskStartedAt, + success: false, + isError: true, + errorCode: structured.error.code, + errorKind: 'execution', + errorMessage: structured.error.message_sv, + requestId: id ?? null, + userId, + companyId: effectiveCompanyId, + }) + } + }) + return NextResponse.json( + jsonRpc(id ?? null, { + resultType: 'task', + task: taskToWire(task), + _meta: { [META_SERVER_INFO]: SERVER_INFO_BY_NAMESPACE[toolNamespace] }, + }) + ) + } + const callStartedAt = Date.now() try { // gnubok_search_tools needs the caller's scopes to filter results to @@ -16537,6 +16645,51 @@ export async function handleMcpRequest(request: Request): Promise { ) } + case 'tasks/get': { + const taskId = (params as Record)?.taskId + if (typeof taskId !== 'string' || !taskId) { + return NextResponse.json(jsonRpcError(id ?? null, -32602, 'taskId is required')) + } + // Scoped to the creating user: an API key can only poll its own tasks. + const { data: taskRow } = await supabase + .from('mcp_tasks') + .select('*') + .eq('id', taskId) + .eq('user_id', userId) + .single() + if (!taskRow) { + return NextResponse.json(jsonRpcError(id ?? null, -32602, `Task not found: "${taskId}"`)) + } + const row = taskRow as McpTaskRow + const wire: Record = { resultType: 'complete', ...taskToWire(row) } + if (row.status === 'completed' && row.result) wire.result = row.result + if (row.status === 'failed' && row.error) wire.error = row.error + wire._meta = { [META_SERVER_INFO]: SERVER_INFO_BY_NAMESPACE[toolNamespace] } + return NextResponse.json(jsonRpc(id ?? null, wire)) + } + + case 'tasks/update': + // No input_required flows exist yet: acknowledge and ignore unknown or + // already-satisfied inputResponses, as the extension spec instructs. + return NextResponse.json(jsonRpc(id ?? null, { resultType: 'complete' })) + + case 'tasks/cancel': { + const taskId = (params as Record)?.taskId + if (typeof taskId !== 'string' || !taskId) { + return NextResponse.json(jsonRpcError(id ?? null, -32602, 'taskId is required')) + } + // Cooperative cancellation: flip a still-working row; an in-flight + // execution is not interrupted, and its late completion becomes a + // no-op against the now-terminal row. + await supabase + .from('mcp_tasks') + .update({ status: 'cancelled' }) + .eq('id', taskId) + .eq('user_id', userId) + .eq('status', 'working') + return NextResponse.json(jsonRpc(id ?? null, { resultType: 'complete' })) + } + default: return NextResponse.json( jsonRpcError(id ?? null, -32601, `Method not found: "${method}"`) diff --git a/extensions/general/mcp-server/tasks.ts b/extensions/general/mcp-server/tasks.ts new file mode 100644 index 00000000..c4931dbb --- /dev/null +++ b/extensions/general/mcp-server/tasks.ts @@ -0,0 +1,121 @@ +import type { SupabaseClient } from '@supabase/supabase-js' + +/** + * MCP Tasks extension (io.modelcontextprotocol/tasks). + * + * Durable handles for long-running tool calls: a task-capable client gets a + * CreateTaskResult (resultType: "task") immediately and polls tasks/get until + * a terminal status. Rows live in mcp_tasks (service-role writes only) so + * handles survive disconnects and serverless instance turnover. + * + * Failure mapping: a tool execution failure is stored as a COMPLETED task + * whose result carries the standard isError envelope, because that is exactly + * what the synchronous call would have returned. The `failed` status (and the + * `error` column) is reserved for infrastructure failures where no tool + * result exists. + */ + +export const TASKS_EXTENSION_ID = 'io.modelcontextprotocol/tasks' + +const DEFAULT_POLL_INTERVAL_MS = 2000 +const DEFAULT_TTL_MS = 3_600_000 + +export interface McpTaskRow { + id: string + company_id: string + user_id: string + tool_name: string + status: 'working' | 'input_required' | 'completed' | 'failed' | 'cancelled' + status_message: string | null + result: Record | null + error: Record | null + poll_interval_ms: number + ttl_ms: number + created_at: string +} + +/** + * Per the extension spec, a server must never return a task to a client that + * did not declare the extension in this request's capabilities. + */ +export function isTaskCapableClient(requestMeta: Record): boolean { + const caps = requestMeta['io.modelcontextprotocol/clientCapabilities'] + if (!caps || typeof caps !== 'object') return false + const extensions = (caps as Record).extensions + if (!extensions || typeof extensions !== 'object') return false + return TASKS_EXTENSION_ID in (extensions as Record) +} + +export async function createMcpTask( + supabase: SupabaseClient, + params: { companyId: string; userId: string; apiKeyId?: string | null; toolName: string } +): Promise { + // Storage limitation (GDPR Art. 5(1)(e)): opportunistically purge expired + // rows on every creation so the 1-hour retention is enforced without + // dedicated cron infrastructure (cheap via idx_mcp_tasks_expires). + // Best-effort: a failed sweep must never block the new task. + try { + await supabase.from('mcp_tasks').delete().lt('expires_at', new Date().toISOString()) + } catch { + // Ignore: the next creation retries the sweep. + } + const { data, error } = await supabase + .from('mcp_tasks') + .insert({ + company_id: params.companyId, + user_id: params.userId, + api_key_id: params.apiKeyId ?? null, + tool_name: params.toolName, + status: 'working', + poll_interval_ms: DEFAULT_POLL_INTERVAL_MS, + ttl_ms: DEFAULT_TTL_MS, + }) + .select('*') + .single() + if (error || !data) { + throw new Error(`Failed to create MCP task: ${error?.message ?? 'no row returned'}`) + } + return data as McpTaskRow +} + +/** + * Move a still-working task to a terminal state. The status='working' guard + * makes terminal states immutable (spec) and lets a tasks/cancel that raced + * the execution win: the late completion becomes a no-op. + */ +export async function resolveMcpTask( + supabase: SupabaseClient, + taskId: string, + terminal: { + status: 'completed' | 'failed' | 'cancelled' + result?: Record + error?: Record + statusMessage?: string + } +): Promise { + // Literal payload (no conditional spreads) so the phantom-column guard can + // resolve every column. Writing null for absent terminal fields is correct: + // the transition sets the complete terminal state. + await supabase + .from('mcp_tasks') + .update({ + status: terminal.status, + result: terminal.result ?? null, + error: terminal.error ?? null, + status_message: terminal.statusMessage ?? null, + }) + .eq('id', taskId) + .eq('status', 'working') +} + +/** Map a row to the wire Task object shared by CreateTaskResult and tasks/get. */ +export function taskToWire(row: McpTaskRow): Record { + return { + taskId: row.id, + status: row.status, + createdAt: row.created_at, + ttlMs: Number(row.ttl_ms), + pollIntervalMs: row.poll_interval_ms, + ...(row.status_message ? { statusMessage: row.status_message } : {}), + } +} diff --git a/lib/reports/full-archive-export.ts b/lib/reports/full-archive-export.ts index 0a97c540..07c3061c 100644 --- a/lib/reports/full-archive-export.ts +++ b/lib/reports/full-archive-export.ts @@ -961,6 +961,7 @@ export const ARCHIVE_EXCLUDED_TABLES: Record = { idempotency_keys: 'infrastructure', inbox_rate_counters: 'infrastructure', invoice_inbox_items: 'inbox workflow state; the files live in document_attachments', + mcp_tasks: 'MCP task handles: transient tool-call state with a 1-hour TTL', metered_events: 'billing telemetry', notification_log: 'notification dedup log', operations: 'staged-operation workflow state', diff --git a/supabase/migrations/20260729094000_create_mcp_tasks.sql b/supabase/migrations/20260729094000_create_mcp_tasks.sql new file mode 100644 index 00000000..b8ea4538 --- /dev/null +++ b/supabase/migrations/20260729094000_create_mcp_tasks.sql @@ -0,0 +1,59 @@ +-- Migration: MCP Tasks (io.modelcontextprotocol/tasks extension) +-- Durable handles for long-running MCP tool calls: the tool call returns a +-- task handle immediately and the work completes after the response; clients +-- poll tasks/get until a terminal status. Writes go through the service-role +-- MCP handler only (mirrors pending_operations); only the creating user may +-- read. Retention (1 hour via expires_at) is enforced by an opportunistic +-- sweep in the MCP handler: createMcpTask deletes expired rows on every +-- task creation (GDPR Art. 5(1)(e)). + +CREATE TABLE public.mcp_tasks ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + company_id UUID NOT NULL REFERENCES public.companies(id) ON DELETE CASCADE, + user_id UUID NOT NULL REFERENCES auth.users(id) ON DELETE CASCADE, + -- Attribution only (which API key created the task); deliberately no FK so + -- key rotation or deletion never breaks task history. + api_key_id UUID, + tool_name TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'working' CHECK (status IN ( + 'working', 'input_required', 'completed', 'failed', 'cancelled' + )), + status_message TEXT, + -- Terminal payloads: `result` holds exactly what the synchronous tool call + -- would have returned (a CallToolResult, including isError envelopes); + -- `error` holds a JSON-RPC error object for infrastructure failures. + result JSONB, + error JSONB, + poll_interval_ms INTEGER NOT NULL DEFAULT 2000, + ttl_ms BIGINT NOT NULL DEFAULT 3600000, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + expires_at TIMESTAMPTZ NOT NULL DEFAULT now() + interval '1 hour' +); + +-- tasks/get looks up by id + creator; expiry cleanup scans expires_at. +CREATE INDEX idx_mcp_tasks_user ON public.mcp_tasks (user_id, status); +CREATE INDEX idx_mcp_tasks_company ON public.mcp_tasks (company_id); +CREATE INDEX idx_mcp_tasks_expires ON public.mcp_tasks (expires_at); + +ALTER TABLE public.mcp_tasks ENABLE ROW LEVEL SECURITY; + +-- Reads: the creating user only. tasks/get in the MCP handler scopes to the +-- creator, and task results carry whatever the underlying tool returned; +-- the DB grant must not be broader than that application contract +-- (data minimisation, GDPR Art. 5(1)(c)). Mirrors pending_operations. +CREATE POLICY "mcp_tasks_select_own" ON public.mcp_tasks + FOR SELECT USING (auth.uid() = user_id); + +-- No INSERT/UPDATE/DELETE policies: all writes go through the service-role +-- MCP handler (mirrors pending_operations). Rows age out via expires_at. + +CREATE TRIGGER mcp_tasks_updated_at + BEFORE UPDATE ON public.mcp_tasks + FOR EACH ROW EXECUTE FUNCTION public.update_updated_at_column(); + +-- No audit trigger: operational task plumbing, not business records. The +-- underlying tool executions already emit mcp.tool_called telemetry, and any +-- committed bookkeeping effects carry their own audit trail. + +NOTIFY pgrst, 'reload schema'; diff --git a/tests/pg/mcp-tasks.pg.test.ts b/tests/pg/mcp-tasks.pg.test.ts new file mode 100644 index 00000000..5e7deb51 --- /dev/null +++ b/tests/pg/mcp-tasks.pg.test.ts @@ -0,0 +1,109 @@ +import { randomUUID } from 'node:crypto' +import { describe, expect, it } from 'vitest' +import { getPool, withUserContext } from './setup' +import { seedCompany, insertAuthUser, insertCompanyMember } from './fixtures' + +// pg-real coverage for 20260729094000_create_mcp_tasks.sql: the status CHECK, +// creator-only SELECT RLS, the deliberate absence of authenticated write +// policies (writes are service-role only), and the updated_at trigger. + +async function insertTask(params: { + companyId: string + userId: string + status?: string + toolName?: string +}): Promise { + const id = randomUUID() + await getPool().query( + `INSERT INTO public.mcp_tasks (id, company_id, user_id, tool_name, status) + VALUES ($1, $2, $3, $4, $5)`, + [id, params.companyId, params.userId, params.toolName ?? 'gnubok_audit_package', params.status ?? 'working'], + ) + return id +} + +describe('mcp_tasks', () => { + it('rejects statuses outside the task lifecycle', async () => { + const { companyId, userId } = await seedCompany() + await expect( + insertTask({ companyId, userId, status: 'running' }), + ).rejects.toThrow(/mcp_tasks_status_check/) + }) + + it('only the creating user can read a task (not other members, not other companies)', async () => { + const a = await seedCompany() + const b = await seedCompany() + const taskId = await insertTask({ companyId: a.companyId, userId: a.userId }) + + const mine = await withUserContext(a.userId, (client) => + client.query('SELECT id, status FROM public.mcp_tasks WHERE id = $1', [taskId]), + ) + expect(mine.rows).toHaveLength(1) + expect(mine.rows[0].status).toBe('working') + + // A second member of the SAME company must not see the row: task results + // carry raw tool output, so the grant is creator-only (Art. 5(1)(c)). + const colleagueId = await insertAuthUser() + await insertCompanyMember({ companyId: a.companyId, userId: colleagueId, role: 'member' }) + const colleague = await withUserContext(colleagueId, (client) => + client.query('SELECT id FROM public.mcp_tasks WHERE id = $1', [taskId]), + ) + expect(colleague.rows).toHaveLength(0) + + const theirs = await withUserContext(b.userId, (client) => + client.query('SELECT id FROM public.mcp_tasks WHERE id = $1', [taskId]), + ) + expect(theirs.rows).toHaveLength(0) + }) + + it('authenticated users cannot insert, update, or delete tasks (service-role only)', async () => { + const { companyId, userId } = await seedCompany() + const taskId = await insertTask({ companyId, userId }) + + await expect( + withUserContext(userId, (client) => + client.query( + `INSERT INTO public.mcp_tasks (company_id, user_id, tool_name) + VALUES ($1, $2, 'gnubok_audit_package')`, + [companyId, userId], + ), + ), + ).rejects.toThrow(/row-level security/) + + // UPDATE and DELETE have no policies: RLS silently filters all rows, + // so the statements succeed but affect nothing. + const upd = await withUserContext(userId, (client) => + client.query(`UPDATE public.mcp_tasks SET status = 'cancelled' WHERE id = $1`, [taskId]), + ) + expect(upd.rowCount).toBe(0) + + const del = await withUserContext(userId, (client) => + client.query('DELETE FROM public.mcp_tasks WHERE id = $1', [taskId]), + ) + expect(del.rowCount).toBe(0) + }) + + it('bumps updated_at on status transitions', async () => { + const { companyId, userId } = await seedCompany() + const taskId = await insertTask({ companyId, userId }) + + const before = await getPool().query( + 'SELECT updated_at FROM public.mcp_tasks WHERE id = $1', + [taskId], + ) + // clock_timestamp()-based trigger: force a measurable gap. + await getPool().query('SELECT pg_sleep(0.05)') + await getPool().query( + `UPDATE public.mcp_tasks SET status = 'completed', result = '{"ok":true}'::jsonb WHERE id = $1`, + [taskId], + ) + const after = await getPool().query( + 'SELECT updated_at, status FROM public.mcp_tasks WHERE id = $1', + [taskId], + ) + expect(after.rows[0].status).toBe('completed') + expect(new Date(after.rows[0].updated_at).getTime()).toBeGreaterThanOrEqual( + new Date(before.rows[0].updated_at).getTime(), + ) + }) +})