diff --git a/.changeset/training-agent-force-task-completion.md b/.changeset/training-agent-force-task-completion.md new file mode 100644 index 0000000000..a5e04f0636 --- /dev/null +++ b/.changeset/training-agent-force-task-completion.md @@ -0,0 +1,19 @@ +--- +--- + +patch: training-agent implements force_task_completion + +Wires this repo's reference seller to the `force_task_completion` controller scenario from #3138 — the spec primitive that resolves a previously-submitted task to `completed` with a buyer-supplied result payload. Companion to #3115 (`force_create_media_buy_arm`); rebased onto #3191 (the `@adcp/client` 5.18.0 bump that ships `PostgresTaskStore.createTask({ taskId })`). + +**Controller scenario.** `handleComplyTestController` pre-dispatches `force_task_completion` before delegating to the SDK (the SDK's `CONTROLLER_SCENARIOS` enum is closed; new spec scenarios live in the wrapper until adopted upstream). Validates `task_id` (non-empty, ≤128 chars), `result` (plain object), and a soft 256 KB cap on the payload. Records `(task_id, result, ownerKey)` in a process-global Map so cross-account replays return `NOT_FOUND` (per the spec MUST), identical-params replays are idempotent no-ops, and diverging-params replays against a terminal task return `INVALID_TRANSITION`. `list_scenarios` is augmented to advertise the new scenario. + +**Why a local Map and not the SDK task store.** 5.18.0 ships `PostgresTaskStore.createTask({ taskId })` — meaningful progress, but two adjacent gaps still block SDK-backed roundtripping: + +1. `InMemoryTaskStore` (re-exported from `@modelcontextprotocol/sdk`) doesn't yet honor caller-supplied IDs. Training-agent CI without `DATABASE_URL` falls back to InMemory, so an SDK-backed implementation would silently get random IDs there. +2. The SDK auto-registers `tasks/get` with the MCP `Task` shape (`taskId`, status ∈ `working|completed|failed|input_required|cancelled`); the AdCP `tasks-get-response.json` schema requires the AdCP shape (`task_id`, status ∈ `submitted|working|input-required|completed|canceled|...`, plus `task_type`, `protocol`, `result`). A storyboard polling phase asserting AdCP fails against the SDK's auto-registered handler. + +Both tracked in adcp-client#994. The local-Map primitive ships the spec contract sellers must honor (cross-account NOT_FOUND, replay idempotency, terminal INVALID_TRANSITION) without depending on infrastructure that isn't there yet. Swapping the storage layer is mechanical when upstream lands. + +**Storyboard extension still deferred.** `create_media_buy_async.yaml` remains v1.0.0 (submitted-arm only). The polling phase needs Gap 2 to land plus storyboard-runner wiring to thread caller-supplied IDs through tool input. + +**Tests.** New `server/tests/unit/training-agent-force-task-completion.test.ts` (9 tests): directive registration, INVALID_PARAMS for missing/oversized fields, replay idempotency, diverging-replay INVALID_TRANSITION, cross-account NOT_FOUND, list_scenarios advertisement. `comply-test-controller.test.ts` length assertion bumped 8→9. diff --git a/server/src/training-agent/comply-test-controller.ts b/server/src/training-agent/comply-test-controller.ts index 0a0ceb8797..cf1ab7a1a9 100644 --- a/server/src/training-agent/comply-test-controller.ts +++ b/server/src/training-agent/comply-test-controller.ts @@ -465,7 +465,7 @@ function createStore(session: SessionState): TestControllerStore { * adcontextprotocol/adcp-client — the dedup below means it is safe to leave this * entry in place during the transition; remove once a release has landed and the * cross-impl tests no longer rely on it). */ -const LOCAL_SCENARIOS = ['force_create_media_buy_arm', 'seed_creative_format'] as const; +const LOCAL_SCENARIOS = ['force_create_media_buy_arm', 'force_task_completion', 'seed_creative_format'] as const; // ── Tool definition ─────────────────────────────────────────────── @@ -540,7 +540,8 @@ export async function handleComplyTestController(args: ToolArgs, ctx: TrainingCo }; } - const session = await getSession(sessionKeyFromArgs(args, ctx.mode, ctx.userId, ctx.moduleId)); + const sessionKey = sessionKeyFromArgs(args, ctx.mode, ctx.userId, ctx.moduleId); + const session = await getSession(sessionKey); // Pre-dispatch local scenarios the SDK doesn't know about yet. The SDK's // dispatcher would return UNKNOWN_SCENARIO for these, so handle them before @@ -549,6 +550,9 @@ export async function handleComplyTestController(args: ToolArgs, ctx: TrainingCo if (scenario === 'force_create_media_buy_arm') { return handleForceCreateMediaBuyArm(session, rawArgs); } + if (scenario === 'force_task_completion') { + return handleForceTaskCompletion(sessionKey, rawArgs); + } // seed_creative_format is a training-agent extension not in the SDK's // CONTROLLER_SCENARIOS. Handle it before the SDK dispatcher so the SDK // doesn't return UNKNOWN_SCENARIO. Idempotency (same ID + same fixture @@ -697,6 +701,137 @@ function handleForceCreateMediaBuyArm(session: SessionState, rawArgs: Record; ownerKey: string; completedAt: string }>(); +const MAX_FORCED_TASK_COMPLETIONS = 1000; + +/** Test-only: clear the forced-completion pool. */ +export function clearForcedTaskCompletions(): void { + FORCED_TASK_COMPLETIONS.clear(); +} + +/** Test-only: read the forced-completion pool. */ +export function getForcedTaskCompletions(): ReadonlyMap; ownerKey: string; completedAt: string }> { + return FORCED_TASK_COMPLETIONS; +} + +function handleForceTaskCompletion(sessionKey: string, rawArgs: Record): object { + const params = rawArgs.params as Record | undefined; + if (!params || typeof params !== 'object') { + return { + success: false, + error: 'INVALID_PARAMS', + error_detail: 'force_task_completion requires params', + }; + } + + const taskId = params.task_id; + if (typeof taskId !== 'string' || taskId.length === 0) { + return { + success: false, + error: 'INVALID_PARAMS', + error_detail: 'task_id is required', + }; + } + if (taskId.length > 128) { + return { + success: false, + error: 'INVALID_PARAMS', + error_detail: 'task_id exceeds maxLength 128', + }; + } + + const result = params.result; + if (!result || typeof result !== 'object' || Array.isArray(result)) { + return { + success: false, + error: 'INVALID_PARAMS', + error_detail: 'result is required and must be an object (validates against async-response-data.json)', + }; + } + + // Soft 256 KB cap on result payloads, per the spec's recommendation. Bounds + // sandbox-amplified storage/echo DoS against the seller's task store. + const resultBytes = JSON.stringify(result).length; + if (resultBytes > 256 * 1024) { + return { + success: false, + error: 'INVALID_PARAMS', + error_detail: `result payload exceeds 256 KB (${resultBytes} bytes)`, + }; + } + + const existing = FORCED_TASK_COMPLETIONS.get(taskId); + if (existing) { + // Cross-account check (spec MUST): NOT_FOUND for task_ids belonging to other + // accounts, conventional "not yours" → "doesn't exist" treatment. + if (existing.ownerKey !== sessionKey) { + return { + success: false, + error: 'NOT_FOUND', + error_detail: `Task "${taskId}" was not registered for this sandbox account`, + }; + } + // Idempotent replay: same params → no-op success. + if (JSON.stringify(existing.result) === JSON.stringify(result)) { + return { + success: true, + previous_state: 'completed', + current_state: 'completed', + message: `Task ${taskId} already completed with the same result`, + }; + } + // Diverging replay against a terminal task: INVALID_TRANSITION. + return { + success: false, + error: 'INVALID_TRANSITION', + error_detail: `Task "${taskId}" is already terminal (completed); cannot re-complete with diverging result`, + current_state: 'completed', + }; + } + + if (FORCED_TASK_COMPLETIONS.size >= MAX_FORCED_TASK_COMPLETIONS) { + return { + success: false, + error: 'INVALID_STATE', + error_detail: `Forced-completion cap reached (${MAX_FORCED_TASK_COMPLETIONS})`, + }; + } + + FORCED_TASK_COMPLETIONS.set(taskId, { + result: result as Record, + ownerKey: sessionKey, + completedAt: new Date().toISOString(), + }); + + return { + success: true, + previous_state: 'submitted', + current_state: 'completed', + message: `Task ${taskId} transitioned from submitted to completed`, + }; +} + // Module-level seed-fixture cache enforces the spec's same-ID-different- // fixture rejection rule across all seed calls in the process. Scoping per- // process keeps it aligned with the CONTROLLER_SCENARIOS list being static. diff --git a/server/tests/unit/comply-test-controller.test.ts b/server/tests/unit/comply-test-controller.test.ts index 1b0f8ec48d..b4c54f035d 100644 --- a/server/tests/unit/comply-test-controller.test.ts +++ b/server/tests/unit/comply-test-controller.test.ts @@ -176,11 +176,12 @@ describe('comply_test_controller', () => { // Local scenarios — see LOCAL_SCENARIOS in // server/src/training-agent/comply-test-controller.ts. 'force_create_media_buy_arm', + 'force_task_completion', 'seed_creative_format', ])); // Catch silent drift in either direction (entries removed, or new ones // not yet documented in this assertion). - expect(scenarios.length).toBe(8); + expect(scenarios.length).toBe(9); // Dedup invariant — see SCENARIO_ENUM dedup in the wrapper. expect(new Set(scenarios).size).toBe(scenarios.length); }); diff --git a/server/tests/unit/training-agent-force-task-completion.test.ts b/server/tests/unit/training-agent-force-task-completion.test.ts new file mode 100644 index 0000000000..a2a8196c0e --- /dev/null +++ b/server/tests/unit/training-agent-force-task-completion.test.ts @@ -0,0 +1,224 @@ +import { describe, it, expect, beforeEach } from 'vitest'; +import crypto from 'node:crypto'; +import { + createTrainingAgentServer, + invalidateCache, + clearTaskStore, +} from '../../src/training-agent/task-handlers.js'; +import { clearSessions } from '../../src/training-agent/state.js'; +import { MUTATING_TOOLS, clearIdempotencyCache } from '../../src/training-agent/idempotency.js'; +import { + clearForcedTaskCompletions, + getForcedTaskCompletions, +} from '../../src/training-agent/comply-test-controller.js'; +import type { TrainingContext } from '../../src/training-agent/types.js'; + +const DEFAULT_CTX: TrainingContext = { mode: 'open' }; +const ACCOUNT_A = { brand: { domain: 'force-completion-a.example' }, operator: 'tester-a', sandbox: true }; +const BRAND_A = { domain: 'force-completion-a.example' }; +const ACCOUNT_B = { brand: { domain: 'force-completion-b.example' }, operator: 'tester-b', sandbox: true }; +const BRAND_B = { domain: 'force-completion-b.example' }; + +const SAMPLE_RESULT = { + media_buy_id: 'mb_async_signed_io_q2', + status: 'active', + packages: [ + { package_id: 'pkg-0', product_id: 'async_signed_io_q2', budget: 30000 }, + ], +}; + +function withIdempotencyKey(toolName: string, args: Record): Record { + if (!MUTATING_TOOLS.has(toolName)) return args; + if (args.idempotency_key !== undefined) return args; + return { ...args, idempotency_key: `test-${crypto.randomUUID()}` }; +} + +async function callTool( + server: ReturnType, + toolName: string, + args: Record, +): Promise> { + const requestHandlers = (server as unknown as { _requestHandlers: Map })._requestHandlers; + const handler = requestHandlers.get('tools/call'); + if (!handler) throw new Error('CallTool handler not found'); + const response = await handler( + { method: 'tools/call', params: { name: toolName, arguments: withIdempotencyKey(toolName, args) } }, + {}, + ); + const text = response.content?.[0]?.text; + const parsed: Record = response.structuredContent + ? (response.structuredContent as Record) + : (text ? JSON.parse(text) : {}); + return (parsed.adcp_error as Record | undefined) ?? parsed; +} + +describe('force_task_completion', () => { + let server: ReturnType; + + beforeEach(async () => { + await clearSessions(); + clearIdempotencyCache(); + invalidateCache(); + clearTaskStore(); + clearForcedTaskCompletions(); + server = createTrainingAgentServer(DEFAULT_CTX); + }); + + describe('directive registration', () => { + it('registers a completion with task_id and result, returns StateTransitionSuccess', async () => { + const result = await callTool(server, 'comply_test_controller', { + scenario: 'force_task_completion', + params: { task_id: 'task_async_signed_io_q2', result: SAMPLE_RESULT }, + account: ACCOUNT_A, + brand: BRAND_A, + }); + + expect(result.success).toBe(true); + expect(result.previous_state).toBe('submitted'); + expect(result.current_state).toBe('completed'); + + // Recorded in the process-global pool. + const recorded = getForcedTaskCompletions().get('task_async_signed_io_q2'); + expect(recorded).toBeDefined(); + expect(recorded!.result).toEqual(SAMPLE_RESULT); + }); + + it('rejects missing task_id with INVALID_PARAMS', async () => { + const result = await callTool(server, 'comply_test_controller', { + scenario: 'force_task_completion', + params: { result: SAMPLE_RESULT }, + account: ACCOUNT_A, + brand: BRAND_A, + }); + + expect(result.success).toBe(false); + expect(result.error).toBe('INVALID_PARAMS'); + expect(result.error_detail).toMatch(/task_id/); + }); + + it('rejects missing result with INVALID_PARAMS', async () => { + const result = await callTool(server, 'comply_test_controller', { + scenario: 'force_task_completion', + params: { task_id: 'task_no_result' }, + account: ACCOUNT_A, + brand: BRAND_A, + }); + + expect(result.success).toBe(false); + expect(result.error).toBe('INVALID_PARAMS'); + expect(result.error_detail).toMatch(/result/); + }); + + it('rejects task_id over 128 chars', async () => { + const result = await callTool(server, 'comply_test_controller', { + scenario: 'force_task_completion', + params: { task_id: 'x'.repeat(129), result: SAMPLE_RESULT }, + account: ACCOUNT_A, + brand: BRAND_A, + }); + + expect(result.success).toBe(false); + expect(result.error).toBe('INVALID_PARAMS'); + expect(result.error_detail).toMatch(/task_id/); + }); + + it('rejects result over 256 KB', async () => { + // Build a result with one giant string field. JSON-stringified size > 256KB. + const huge = { media_buy_id: 'mb_huge', status: 'active', packages: [], filler: 'x'.repeat(260 * 1024) }; + const result = await callTool(server, 'comply_test_controller', { + scenario: 'force_task_completion', + params: { task_id: 'task_huge', result: huge }, + account: ACCOUNT_A, + brand: BRAND_A, + }); + + expect(result.success).toBe(false); + expect(result.error).toBe('INVALID_PARAMS'); + expect(result.error_detail).toMatch(/256 KB/); + }); + }); + + describe('replay semantics', () => { + it('replays with identical params are idempotent no-ops', async () => { + const args = { + scenario: 'force_task_completion', + params: { task_id: 'task_replay', result: SAMPLE_RESULT }, + account: ACCOUNT_A, + brand: BRAND_A, + }; + + const first = await callTool(server, 'comply_test_controller', args); + expect(first.success).toBe(true); + expect(first.previous_state).toBe('submitted'); + + const replay = await callTool(server, 'comply_test_controller', args); + expect(replay.success).toBe(true); + // Same-params replay reports both states as 'completed' — idempotent no-op. + expect(replay.previous_state).toBe('completed'); + expect(replay.current_state).toBe('completed'); + }); + + it('replays with diverging params return INVALID_TRANSITION', async () => { + await callTool(server, 'comply_test_controller', { + scenario: 'force_task_completion', + params: { task_id: 'task_diverge', result: SAMPLE_RESULT }, + account: ACCOUNT_A, + brand: BRAND_A, + }); + + const replay = await callTool(server, 'comply_test_controller', { + scenario: 'force_task_completion', + params: { task_id: 'task_diverge', result: { ...SAMPLE_RESULT, media_buy_id: 'mb_different' } }, + account: ACCOUNT_A, + brand: BRAND_A, + }); + + expect(replay.success).toBe(false); + expect(replay.error).toBe('INVALID_TRANSITION'); + expect(replay.current_state).toBe('completed'); + }); + }); + + describe('cross-account isolation', () => { + it('returns NOT_FOUND when account B tries to re-complete account A\'s task with diverging result', async () => { + // Account A registers the task. + await callTool(server, 'comply_test_controller', { + scenario: 'force_task_completion', + params: { task_id: 'task_cross_tenant', result: SAMPLE_RESULT }, + account: ACCOUNT_A, + brand: BRAND_A, + }); + + // Account B tries to overwrite. + const result = await callTool(server, 'comply_test_controller', { + scenario: 'force_task_completion', + params: { task_id: 'task_cross_tenant', result: { ...SAMPLE_RESULT, media_buy_id: 'mb_hijack' } }, + account: ACCOUNT_B, + brand: BRAND_B, + }); + + expect(result.success).toBe(false); + // Per spec MUST: cross-account → NOT_FOUND, not FORBIDDEN. + expect(result.error).toBe('NOT_FOUND'); + + // Original record unchanged. + const recorded = getForcedTaskCompletions().get('task_cross_tenant'); + expect(recorded?.result).toEqual(SAMPLE_RESULT); + }); + }); + + describe('list_scenarios advertisement', () => { + it('includes force_task_completion in the supported scenarios list', async () => { + const result = await callTool(server, 'comply_test_controller', { + scenario: 'list_scenarios', + account: ACCOUNT_A, + brand: BRAND_A, + }); + + expect(result.success).toBe(true); + const scenarios = (result as { scenarios: string[] }).scenarios; + expect(scenarios).toContain('force_task_completion'); + expect(scenarios).toContain('force_create_media_buy_arm'); + }); + }); +});