Skip to content

fix(cloud-task): refresh run status before emitting watcher error #2201

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
thmsobrmlr wants to merge 1 commit into from
+210 −11
Closed

fix(cloud-task): refresh run status before emitting watcher error #2201

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
143 changes: 141 additions & 2 deletions apps/code/src/main/services/cloud-task/service.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -659,8 +659,9 @@ describe("CloudTaskService", () => {

expect(mockStreamFetch.mock.calls.length).toBe(6);
// 2 bootstrap calls + 1 post-bootstrap status verification + 6
// handleStreamCompletion calls (one per stream error)
expect(mockNetFetch).toHaveBeenCalledTimes(9);
// handleStreamCompletion calls (one per stream error) + 1 final
// status refresh from failWatcher
expect(mockNetFetch).toHaveBeenCalledTimes(10);
expect(updates).toContainEqual({
taskId: "task-1",
runId: "run-1",
Expand All @@ -672,6 +673,144 @@ describe("CloudTaskService", () => {
});
});

it("emits a terminal status update before the error when the run finished while reconnecting", async () => {
vi.useFakeTimers();

const updates: unknown[] = [];
service.on(CloudTaskEvent.Update, (payload) => updates.push(payload));

const inProgressRun = () =>
createJsonResponse({
id: "run-1",
status: "in_progress",
stage: null,
output: null,
error_message: null,
branch: "main",
updated_at: "2026-01-01T00:00:00Z",
});

const completedRun = () =>
createJsonResponse({
id: "run-1",
status: "completed",
stage: null,
output: { pr_url: "https://example.com/pr/1" },
error_message: null,
branch: "main",
updated_at: "2026-01-01T00:01:00Z",
completed_at: "2026-01-01T00:01:00Z",
});

// Bootstrap + verifyPostBootstrapStatus + each handleStreamCompletion call
// still reports in_progress; only the final failWatcher refresh sees the
// completed terminal state.
let netCallCount = 0;
mockNetFetch.mockImplementation(() => {
netCallCount++;
if (netCallCount === 2) {
// bootstrap: fetchSessionLogs
return Promise.resolve(
createJsonResponse([], 200, { "X-Has-More": "false" }),
);
}
if (netCallCount === 10) {
// refreshStatusThenEmitError: run is now terminal
return Promise.resolve(completedRun());
}
return Promise.resolve(inProgressRun());
});

mockStreamFetch.mockImplementation(() =>
Promise.resolve(
createSseResponse(
'event: keepalive\ndata: {"type":"keepalive"}\n\nevent: error\ndata: {"error":"boom"}\n\n',
),
),
);

service.watch({
taskId: "task-1",
runId: "run-1",
apiHost: "https://app.example.com",
teamId: 2,
});

await waitFor(() => mockStreamFetch.mock.calls.length === 1);
await vi.advanceTimersByTimeAsync(70_000);
await waitFor(
() =>
updates.some(
(u) =>
typeof u === "object" &&
u !== null &&
(u as { kind?: string }).kind === "error",
),
10_000,
);

const statusIdx = updates.findIndex(
(u) =>
typeof u === "object" &&
u !== null &&
(u as { kind?: string }).kind === "status" &&
(u as { status?: string }).status === "completed",
);
const errorIdx = updates.findIndex(
(u) =>
typeof u === "object" &&
u !== null &&
(u as { kind?: string }).kind === "error",
);

expect(statusIdx).toBeGreaterThanOrEqual(0);
expect(errorIdx).toBeGreaterThanOrEqual(0);
expect(statusIdx).toBeLessThan(errorIdx);
expect(updates[statusIdx]).toMatchObject({
taskId: "task-1",
runId: "run-1",
kind: "status",
status: "completed",
output: { pr_url: "https://example.com/pr/1" },
});
});

it("skips the status refresh when the failure is an auth/access error", async () => {
const updates: unknown[] = [];
service.on(CloudTaskEvent.Update, (payload) => updates.push(payload));

mockNetFetch.mockResolvedValueOnce(
createJsonResponse({ detail: "Forbidden" }, 403),
);

service.watch({
taskId: "task-1",
runId: "run-1",
apiHost: "https://app.example.com",
Comment on lines +740 to +789
Copy link
Copy Markdown
Contributor

@greptile-apps greptile-apps Bot May 18, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 The auth-skip test only exercises the 403 path, but isAuthFailureError covers three distinct HTTP statuses (401, 403, 404). Per the team's preference for parameterised tests, all three cases should be verified — a regression in any one of them would go undetected. Consider using it.each([[401], [403], [404]]) (or Vitest's test.each) to run the same assertions against all three codes.

Context Used: Do not attempt to comment on incorrect alphabetica... (source)

Prompt To Fix With AI 
This is a comment left during a code review.
Path: apps/code/src/main/services/cloud-task/service.test.ts
Line: 740-789

Comment:
The auth-skip test only exercises the 403 path, but `isAuthFailureError` covers three distinct HTTP statuses (401, 403, 404). Per the team's preference for parameterised tests, all three cases should be verified — a regression in any one of them would go undetected. Consider using `it.each([[401], [403], [404]])` (or Vitest's `test.each`) to run the same assertions against all three codes.

**Context Used:** Do not attempt to comment on incorrect alphabetica... ([source](https://app.greptile.com/review/custom-context?memory=instruction-0))

How can I resolve this? If you propose a fix, please make it concise.

teamId: 2,
});

await waitFor(() =>
updates.some(
(u) =>
typeof u === "object" &&
u !== null &&
(u as { kind?: string }).kind === "error",
),
);

// Only the initial bootstrap fetchTaskRun — no extra silent refresh.
expect(mockNetFetch).toHaveBeenCalledTimes(1);
expect(
updates.some(
(u) =>
typeof u === "object" &&
u !== null &&
(u as { kind?: string }).kind === "status",
),
).toBe(false);
});

const guardedFetchStatusExpectations = [
[
401,
Expand Down
78 changes: 69 additions & 9 deletions apps/code/src/main/services/cloud-task/service.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -946,6 +946,41 @@ export class CloudTaskService extends TypedEventEmitter<CloudTaskEvents> {
watcher.sseAbortController?.abort();
watcher.sseAbortController = null;

void this.refreshStatusThenEmitError(watcher, error);
}

/**
* Refresh the run state one final time before emitting the watcher's error
* event. The stream typically fails *after* the run has already reached a
* terminal status (e.g. reconnect attempts exhausted, auth flicker), so a
* single REST poll often surfaces the real outcome. Clients can then move
* the task out of the "in_progress" UI even though the watcher is dead.
*
* Auth/access failures (401/403/404) are skipped because the same call
* would just fail again — and `fetchTaskRunSilent` would loop into another
* `failWatcher` via `shouldFailWatcherForFetchStatus` if we reused
* `fetchTaskRun` here.
*/
private async refreshStatusThenEmitError(
watcher: WatcherState,
error: CloudTaskConnectionError,
): Promise<void> {
if (!isAuthFailureError(error)) {
const { run } = await this.fetchTaskRunSilent(watcher);
if (run && this.applyTaskRunState(watcher, run)) {
this.emit(CloudTaskEvent.Update, {
taskId: watcher.taskId,
runId: watcher.runId,
kind: "status",
status: watcher.lastStatus ?? undefined,
stage: watcher.lastStage,
output: watcher.lastOutput,
errorMessage: watcher.lastErrorMessage,
branch: watcher.lastBranch,
});
}
}

this.emit(CloudTaskEvent.Update, {
taskId: watcher.taskId,
runId: watcher.runId,
Expand Down Expand Up @@ -1219,6 +1254,26 @@ export class CloudTaskService extends TypedEventEmitter<CloudTaskEvents> {
private async fetchTaskRun(
watcher: WatcherState,
): Promise<TaskRunResponse | null> {
const { run, httpStatus } = await this.fetchTaskRunSilent(watcher);
if (run === null && httpStatus !== null) {
if (shouldFailWatcherForFetchStatus(httpStatus)) {
this.failWatcher(
watcherKey(watcher.taskId, watcher.runId),
createStreamStatusError(httpStatus).details,
);
}
}
return run;
}

/**
* Same wire call as `fetchTaskRun` but never escalates a non-OK response
* into a watcher failure. Used from `failWatcher` so the final status poll
* can't re-enter watcher teardown.
*/
private async fetchTaskRunSilent(
watcher: WatcherState,
): Promise<{ run: TaskRunResponse | null; httpStatus: number | null }> {
const url = `${watcher.apiHost}/api/projects/${watcher.teamId}/tasks/${watcher.taskId}/runs/${watcher.runId}/`;

try {
Expand All @@ -1236,23 +1291,28 @@ export class CloudTaskService extends TypedEventEmitter<CloudTaskEvents> {
taskId: watcher.taskId,
runId: watcher.runId,
});
if (shouldFailWatcherForFetchStatus(authedResponse.status)) {
this.failWatcher(
watcherKey(watcher.taskId, watcher.runId),
createStreamStatusError(authedResponse.status).details,
);
}
return null;
return { run: null, httpStatus: authedResponse.status };
}

return (await authedResponse.json()) as TaskRunResponse;
return {
run: (await authedResponse.json()) as TaskRunResponse,
httpStatus: authedResponse.status,
};
} catch (error) {
log.warn("Cloud task status fetch error", {
taskId: watcher.taskId,
runId: watcher.runId,
error,
});
return null;
return { run: null, httpStatus: null };
}
}
}

function isAuthFailureError(error: CloudTaskConnectionError): boolean {
return (
error.title === "Cloud authentication expired" ||
error.title === "Cloud access denied" ||
error.title === "Cloud run not found"
);
}
Comment on lines +1312 to +1318
Copy link
Copy Markdown
Contributor

@greptile-apps greptile-apps Bot May 18, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 isAuthFailureError duplicates the exact title strings defined inside createStreamStatusError, violating OnceAndOnlyOnce. If any of those three titles are ever updated in createStreamStatusError, isAuthFailureError silently returns false for what is actually an auth failure, causing an unnecessary fetchTaskRunSilent call. Since CloudTaskConnectionError is an interface you own, the simplest fix is to add an optional httpStatus?: number field to it — createStreamStatusError already receives the status code and can populate it, and isAuthFailureError can then check [401, 403, 404].includes(error.httpStatus ?? 0) instead of comparing title strings.

Prompt To Fix With AI 
This is a comment left during a code review.
Path: apps/code/src/main/services/cloud-task/service.ts
Line: 1312-1318

Comment:
`isAuthFailureError` duplicates the exact title strings defined inside `createStreamStatusError`, violating OnceAndOnlyOnce. If any of those three titles are ever updated in `createStreamStatusError`, `isAuthFailureError` silently returns `false` for what is actually an auth failure, causing an unnecessary `fetchTaskRunSilent` call. Since `CloudTaskConnectionError` is an interface you own, the simplest fix is to add an optional `httpStatus?: number` field to it — `createStreamStatusError` already receives the status code and can populate it, and `isAuthFailureError` can then check `[401, 403, 404].includes(error.httpStatus ?? 0)` instead of comparing title strings.

How can I resolve this? If you propose a fix, please make it concise.

Loading