Skip to content

Commit 9a234ab

Browse files
committed
fix: ensure final statusMessage is set correctly during task result storage
1 parent 87017c7 commit 9a234ab

4 files changed

Lines changed: 148 additions & 31 deletions

File tree

src/mcp/server.ts

Lines changed: 28 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -80,7 +80,7 @@ import { getUserIdFromTokenCached } from '../utils/userid_cache.js';
8080
import { getPackageVersion } from '../utils/version.js';
8181
import { connectMCPClient } from './client.js';
8282
import { EXTERNAL_TOOL_CALL_TIMEOUT_MSEC, LOG_LEVEL_MAP } from './const.js';
83-
import { isTaskCancelled, processParamsGetTools } from './utils.js';
83+
import { isTaskCancelled, processParamsGetTools, storeTaskResultWithMessage } from './utils.js';
8484

8585
/** Mode → actor executor. Add new modes here. */
8686
const actorExecutorsByMode: Record<ServerMode, ActorExecutor> = {
@@ -1259,25 +1259,38 @@ export class ActorsMcpServer {
12591259
return;
12601260
}
12611261

1262-
// Store the result in the task store
1263-
log.debug('[executeToolAndUpdateTask] Storing completed result', {
1264-
taskId,
1265-
mcpSessionId,
1266-
});
1267-
// Set final statusMessage before transitioning to terminal state,
1268-
// because storeTaskResult does not accept a statusMessage parameter.
1269-
await this.taskStore.updateTaskStatus(taskId, 'working', `${getToolFullName(tool)}: completed`, mcpSessionId);
1270-
await this.taskStore.storeTaskResult(taskId, 'completed', result, mcpSessionId);
1271-
log.debug('Task completed successfully', { taskId, toolName: tool.name, mcpSessionId });
1262+
// Set final statusMessage + store result. Three paths:
1263+
// - SUCCEEDED: normal completion
1264+
// - Pre-flight 402: stored as 'completed' because the x402 payment payload is a valid
1265+
// structured result that clients parse to pay (same as non-task mode which returns it directly)
1266+
// - ABORTED: signal-based abort (client disconnect / notifications/cancelled) — no result to store,
1267+
// transition to 'cancelled' to match tasks/cancel behavior
1268+
log.debug('[executeToolAndUpdateTask] Storing task result', { taskId, toolStatus, mcpSessionId });
1269+
if (toolStatus === TOOL_STATUS.SUCCEEDED) {
1270+
await storeTaskResultWithMessage(this.taskStore, taskId, 'completed', result, `${getToolFullName(tool)}: completed`, mcpSessionId);
1271+
} else if (paymentRequiredResult) {
1272+
await storeTaskResultWithMessage(this.taskStore, taskId, 'completed', result, `${getToolFullName(tool)}: payment required`, mcpSessionId);
1273+
} else if (toolStatus === TOOL_STATUS.ABORTED) {
1274+
await this.taskStore.updateTaskStatus(taskId, 'cancelled', `${getToolFullName(tool)}: aborted by client`, mcpSessionId);
1275+
}
1276+
log.debug('Task execution finished', { taskId, toolName: tool.name, toolStatus, mcpSessionId });
12721277

12731278
finishTaskTracking(toolStatus, callDiagnostics);
12741279
} catch (error) {
12751280
// Handle 402 Payment Required — return structured x402 result so clients can auto-pay
12761281
const httpStatus = getHttpStatusCode(error);
12771282
if (httpStatus === HTTP_PAYMENT_REQUIRED) {
12781283
logHttpError(error, 'Payment required while calling tool (task mode)', { toolName: tool.name });
1279-
await this.taskStore.updateTaskStatus(taskId, 'working', `${getToolFullName(tool)}: payment required`, mcpSessionId);
1280-
await this.taskStore.storeTaskResult(taskId, 'completed', buildPaymentRequiredResponse(error), mcpSessionId);
1284+
// Guard: task may have been cancelled while the actor was running. The SDK throws
1285+
// if we try to transition from terminal 'cancelled' to 'working' (inside storeTaskResultWithMessage).
1286+
// Track as ABORTED — the cancellation is what matters, not the 402.
1287+
if (await isTaskCancelled(taskId, mcpSessionId, this.taskStore)) {
1288+
finishTaskTracking(TOOL_STATUS.ABORTED, { ...buildActorFields(actorName, actorId) });
1289+
return;
1290+
}
1291+
// Store as 'completed' — x402 payment payload is a valid result, not an error (see try-block comment)
1292+
const paymentResult = buildPaymentRequiredResponse(error);
1293+
await storeTaskResultWithMessage(this.taskStore, taskId, 'completed', paymentResult, `${getToolFullName(tool)}: payment required`, mcpSessionId);
12811294
finishTaskTracking(TOOL_STATUS.SOFT_FAIL, {
12821295
failure_category: FAILURE_CATEGORY.INVALID_INPUT,
12831296
failure_http_status: 402,
@@ -1341,15 +1354,14 @@ export class ActorsMcpServer {
13411354
taskId,
13421355
mcpSessionId,
13431356
});
1344-
await this.taskStore.updateTaskStatus(taskId, 'working', `${getToolFullName(tool)}: failed`, mcpSessionId);
1345-
await this.taskStore.storeTaskResult(taskId, 'failed', {
1357+
await storeTaskResultWithMessage(this.taskStore, taskId, 'failed', {
13461358
content: [{
13471359
type: 'text' as const,
13481360
text: userText,
13491361
}],
13501362
isError: true,
13511363
internalToolStatus: toolStatus,
1352-
}, mcpSessionId);
1364+
}, `${getToolFullName(tool)}: failed`, mcpSessionId);
13531365

13541366
finishTaskTracking(toolStatus, callDiagnostics);
13551367
}

src/mcp/utils.ts

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
import { parse } from 'node:querystring';
22

33
import type { TaskStore } from '@modelcontextprotocol/sdk/experimental/tasks/interfaces.js';
4+
import type { Result } from '@modelcontextprotocol/sdk/types.js';
45
import type { ApifyClient } from 'apify-client';
56

67
import { processInput } from '../input.js';
@@ -47,3 +48,24 @@ export async function isTaskCancelled(
4748
const task = await taskStore.getTask(taskId, mcpSessionId);
4849
return task?.status === 'cancelled';
4950
}
51+
52+
/**
53+
* Stores a task result with a final statusMessage in one logical step.
54+
*
55+
* WARNING: This is NOT atomic. The SDK's storeTaskResult() does not accept a statusMessage,
56+
* so we first call updateTaskStatus('working', message) then storeTaskResult(status, result).
57+
* If the process crashes between the two calls, the task stays 'working' with the final message
58+
* but never transitions to terminal state. This is acceptable because the same crash would leave
59+
* the task stuck regardless — the TTL cleanup will eventually remove it.
60+
*/
61+
export async function storeTaskResultWithMessage(
62+
taskStore: TaskStore,
63+
taskId: string,
64+
status: 'completed' | 'failed',
65+
result: Result,
66+
statusMessage: string,
67+
sessionId?: string,
68+
): Promise<void> {
69+
await taskStore.updateTaskStatus(taskId, 'working', statusMessage, sessionId);
70+
await taskStore.storeTaskResult(taskId, status, result, sessionId);
71+
}

tests/integration/suite.ts

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2446,6 +2446,42 @@ export function createIntegrationTestsSuite(
24462446
await assertStatusMessagePropagated(client, stream);
24472447
});
24482448

2449+
// Regression test for #638: final statusMessage must be ": completed" after task finishes,
2450+
// not a stale intermediate progress message like "Starting the crawler."
2451+
it('should set final statusMessage to "completed" after task finishes successfully', async () => {
2452+
client = await createClientFn({ tools: [ACTOR_PYTHON_EXAMPLE] });
2453+
2454+
const stream = client.experimental.tasks.callToolStream(
2455+
{
2456+
name: actorNameToToolName(ACTOR_PYTHON_EXAMPLE),
2457+
arguments: {
2458+
first_number: 7,
2459+
second_number: 8,
2460+
},
2461+
},
2462+
CallToolResultSchema,
2463+
{
2464+
task: {
2465+
ttl: 60000,
2466+
},
2467+
},
2468+
);
2469+
2470+
let taskId: string | null = null;
2471+
for await (const message of stream) {
2472+
if (message.type === 'taskCreated') {
2473+
taskId = message.task.taskId;
2474+
} else if (message.type === 'error') {
2475+
throw message.error;
2476+
}
2477+
}
2478+
2479+
expect(taskId).not.toBeNull();
2480+
const task = await client.experimental.tasks.getTask(taskId!);
2481+
expect(task.status).toBe('completed');
2482+
expect(task.statusMessage).toMatch(/: completed$/);
2483+
});
2484+
24492485
it.runIf(options.transport === 'stdio')('should use UI_MODE env var when CLI arg is not provided', async () => {
24502486
client = await createClientFn({ useEnv: true, uiMode: 'openai' });
24512487
const tools = await client.listTools();
Lines changed: 62 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,18 @@
11
import { InMemoryTaskStore } from '@modelcontextprotocol/sdk/experimental/tasks/stores/in-memory.js';
22
import { describe, expect, it } from 'vitest';
33

4+
import { storeTaskResultWithMessage } from '../../src/mcp/utils.js';
5+
46
/**
57
* Regression tests for #638: tasks/list shows stale statusMessage for completed tasks.
68
*
79
* storeTaskResult() does not accept a statusMessage parameter, so we must call
810
* updateTaskStatus() with the final message before calling storeTaskResult().
9-
* These tests verify that statusMessage survives the storeTaskResult() transition.
11+
*
12+
* These tests verify:
13+
* 1. statusMessage survives the storeTaskResult() transition (SDK invariant)
14+
* 2. The correct final message is set for each outcome (success, payment required, failed)
15+
* 3. Non-success paths do NOT get a misleading ": completed" message
1016
*/
1117
describe('Task statusMessage after terminal transition', () => {
1218
const REQUEST_ID = 'req-1';
@@ -17,39 +23,80 @@ describe('Task statusMessage after terminal transition', () => {
1723
return task.taskId;
1824
}
1925

20-
it('should retain final statusMessage after storeTaskResult completes the task', async () => {
26+
it('should set statusMessage and status atomically via storeTaskResultWithMessage', async () => {
2127
const store = new InMemoryTaskStore();
2228
const taskId = await createWorkingTask(store);
2329

24-
// Intermediate progress message (as set by ProgressTracker during execution)
25-
await store.updateTaskStatus(taskId, 'working', 'apify/rag-web-browser: Starting the crawler.');
26-
27-
// Final message set just before storeTaskResult (the fix)
28-
await store.updateTaskStatus(taskId, 'working', 'apify/rag-web-browser: completed');
29-
30-
await store.storeTaskResult(taskId, 'completed', {
30+
await storeTaskResultWithMessage(store, taskId, 'completed', {
3131
content: [{ type: 'text', text: 'result data' }],
32-
});
32+
}, 'apify/rag-web-browser: completed');
3333

3434
const task = await store.getTask(taskId);
3535
expect(task).not.toBeNull();
3636
expect(task!.status).toBe('completed');
3737
expect(task!.statusMessage).toBe('apify/rag-web-browser: completed');
3838
});
3939

40-
it('should retain final statusMessage after storeTaskResult marks task as failed', async () => {
40+
it('should set statusMessage for failed tasks via storeTaskResultWithMessage', async () => {
4141
const store = new InMemoryTaskStore();
4242
const taskId = await createWorkingTask(store);
4343

44-
await store.updateTaskStatus(taskId, 'working', 'my-tool: failed');
45-
46-
await store.storeTaskResult(taskId, 'failed', {
44+
await storeTaskResultWithMessage(store, taskId, 'failed', {
4745
content: [{ type: 'text', text: 'error' }],
4846
isError: true,
49-
});
47+
}, 'my-tool: failed');
5048

5149
const task = await store.getTask(taskId);
5250
expect(task!.status).toBe('failed');
5351
expect(task!.statusMessage).toBe('my-tool: failed');
5452
});
53+
54+
it('should set "payment required" statusMessage for pre-flight 402 path, not "completed"', async () => {
55+
const store = new InMemoryTaskStore();
56+
const taskId = await createWorkingTask(store);
57+
58+
// Simulates the pre-flight payment failure path in executeToolAndUpdateTask:
59+
// paymentRequiredResult is set, toolStatus = SOFT_FAIL, no actor execution happens.
60+
// The fix guards the updateTaskStatus call so it stamps ": payment required" instead of ": completed".
61+
await store.updateTaskStatus(taskId, 'working', 'apify/rag-web-browser: payment required');
62+
63+
await store.storeTaskResult(taskId, 'completed', {
64+
content: [{ type: 'text', text: 'Payment required' }],
65+
});
66+
67+
const task = await store.getTask(taskId);
68+
expect(task!.status).toBe('completed');
69+
expect(task!.statusMessage).toBe('apify/rag-web-browser: payment required');
70+
expect(task!.statusMessage).not.toContain('completed');
71+
});
72+
73+
it('should set status "cancelled" when task was aborted via signal, not "completed"', async () => {
74+
const store = new InMemoryTaskStore();
75+
const taskId = await createWorkingTask(store);
76+
77+
// Simulates the signal-based abort path: executeActorTool returns null,
78+
// toolStatus = ABORTED. The fix transitions to 'cancelled' instead of
79+
// storing as 'completed' with empty result.
80+
await store.updateTaskStatus(taskId, 'working', 'apify/rag-web-browser: Running the crawler.');
81+
await store.updateTaskStatus(taskId, 'cancelled', 'apify/rag-web-browser: aborted by client');
82+
83+
const task = await store.getTask(taskId);
84+
expect(task!.status).toBe('cancelled');
85+
expect(task!.statusMessage).toBe('apify/rag-web-browser: aborted by client');
86+
});
87+
88+
it('should not call updateTaskStatus on cancelled task (402 catch path)', async () => {
89+
const store = new InMemoryTaskStore();
90+
const taskId = await createWorkingTask(store);
91+
92+
// Simulate: task is cancelled while actor is running
93+
await store.updateTaskStatus(taskId, 'cancelled', 'Cancelled by client');
94+
95+
const task = await store.getTask(taskId);
96+
expect(task!.status).toBe('cancelled');
97+
98+
// Trying to transition from cancelled → working should throw (SDK enforces terminal state).
99+
// The fix adds an isTaskCancelled guard before updateTaskStatus in the 402 catch path.
100+
await expect(store.updateTaskStatus(taskId, 'working', 'some-tool: payment required')).rejects.toThrow();
101+
});
55102
});

0 commit comments

Comments
 (0)