Skip to content
Closed
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions packages/ai/src/ai/modules/task/task_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -1315,6 +1315,9 @@ async def stop_task(self, token: str):
# Only terminate tasks that were launched or executed directly
if control.launch_type in (LAUNCH_TYPE.LAUNCH, LAUNCH_TYPE.EXECUTE):
await control.task.stop_task()
# Free the slot so a subsequent use() with the same token
# does not race against a phantom registry entry.
self._task_control.pop(token, None)
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Outdated
self.debug_message(f'Task "{control.id}" stopped on request')

except Exception as e:
Expand Down
46 changes: 25 additions & 21 deletions packages/client-typescript/tests/RocketRideClient.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -150,31 +150,35 @@ describe('RocketRideClient Integration Tests', () => {
TEST_CONFIG.timeout
);

it('should get pipeline status', async () => {
const result = await client.use({
pipeline: getEchoPipeline(),
token: PIPELINE_TOKEN,
});
it(
'should get pipeline status',
async () => {
const result = await client.use({
pipeline: getEchoPipeline(),
token: PIPELINE_TOKEN,
});

// Retry a few times in case server is busy (tests may run in parallel)
const maxAttempts = 5;
const delayMs = 2000;
let status: Awaited<ReturnType<typeof client.getTaskStatus>> | null = null;
for (let attempt = 1; attempt <= maxAttempts; attempt++) {
try {
status = await client.getTaskStatus(result.token);
break;
} catch (e) {
if (attempt === maxAttempts) throw e;
await new Promise((r) => setTimeout(r, delayMs));
// Retry a few times in case server is busy (tests may run in parallel)
const maxAttempts = 5;
const delayMs = 2000;
let status: Awaited<ReturnType<typeof client.getTaskStatus>> | null = null;
for (let attempt = 1; attempt <= maxAttempts; attempt++) {
try {
status = await client.getTaskStatus(result.token);
break;
} catch (e) {
if (attempt === maxAttempts) throw e;
await new Promise((r) => setTimeout(r, delayMs));
}
}
}

expect(status).toHaveProperty('state');
expect(Object.values(TASK_STATE)).toContain(status!.state);
expect(status).toHaveProperty('state');
expect(Object.values(TASK_STATE)).toContain(status!.state);

await client.terminate(result.token);
}, 90000);
await client.terminate(result.token);
},
TEST_CONFIG.timeout
);

it(
'should terminate a pipeline',
Expand Down
Loading