recover still-running background tasks on reattach
A task row is officer's own invention, synthesised from the harness's system.task_started, and nothing corresponding to it is ever written to Claude's transcript. So rebuildTranscript can only produce user/tool/assistant rows, and sync:live deliberately carries no messages — which left the background-task tray empty after a mid-task refresh even though the work was still running. Fold the durable log on attach into started-minus-notified and hand that back on sync:live. The same read now supplies the cursor, so this costs one query rather than two. Finished tasks are excluded: replaying those would resurrect rows already seen to resolve. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -120,6 +120,9 @@ export type ClientMessage =
|
|||||||
// inside one bubble. Attributing them lets the UI file them under the Task row that spawned them.
|
// inside one bubble. Attributing them lets the UI file them under the Task row that spawned them.
|
||||||
type Parented = { parentToolUseId?: string };
|
type Parented = { parentToolUseId?: string };
|
||||||
|
|
||||||
|
/** A background task with no terminal notification yet. Same fields as `task:started`, by construction. */
|
||||||
|
export type RunningTask = { taskId: string; description: string; taskType?: string };
|
||||||
|
|
||||||
export type ServerMessage =
|
export type ServerMessage =
|
||||||
| {
|
| {
|
||||||
type: 'session:init';
|
type: 'session:init';
|
||||||
@@ -195,6 +198,16 @@ export type ServerMessage =
|
|||||||
* conversation.
|
* conversation.
|
||||||
*/
|
*/
|
||||||
cursor: number;
|
cursor: number;
|
||||||
|
/**
|
||||||
|
* Background tasks that were started and have not reported a terminal status. The one exception to
|
||||||
|
* "carries no messages", and for a concrete reason: a task row is officer's own invention — it is
|
||||||
|
* synthesised from the harness's `system.task_started`, and nothing corresponding to it is ever
|
||||||
|
* written to Claude's transcript. So unlike every other message, rebuilding the transcript cannot
|
||||||
|
* produce it, and a refresh mid-task left the tray empty while the work was still running.
|
||||||
|
* Finished tasks are deliberately excluded: replaying those would resurrect rows the user has
|
||||||
|
* already seen resolve, and only the running ones are still telling you anything.
|
||||||
|
*/
|
||||||
|
runningTasks: RunningTask[];
|
||||||
}
|
}
|
||||||
| {
|
| {
|
||||||
type: 'error';
|
type: 'error';
|
||||||
|
|||||||
@@ -1,6 +1,14 @@
|
|||||||
import type { ServerWebSocket } from 'bun';
|
import type { ServerWebSocket } from 'bun';
|
||||||
import { randomUUID } from 'crypto';
|
import { randomUUID } from 'crypto';
|
||||||
import type { ClientMessage, ServerMessage, Message, PromptImage, TurnMessage, UserSession } from './types';
|
import type {
|
||||||
|
ClientMessage,
|
||||||
|
ServerMessage,
|
||||||
|
Message,
|
||||||
|
PromptImage,
|
||||||
|
RunningTask,
|
||||||
|
TurnMessage,
|
||||||
|
UserSession,
|
||||||
|
} from './types';
|
||||||
import { sessionManager } from './session-manager';
|
import { sessionManager } from './session-manager';
|
||||||
import { sendClaudeCodeStreaming } from '@@/channels/send-claude-code';
|
import { sendClaudeCodeStreaming } from '@@/channels/send-claude-code';
|
||||||
import { sendOpenCodeStreaming } from '@@/channels/send-opencode';
|
import { sendOpenCodeStreaming } from '@@/channels/send-opencode';
|
||||||
@@ -8,7 +16,7 @@ import { ensureGeneralChatSessionsCwd } from './claude-sessions';
|
|||||||
import * as sidecar from '@@/sidecar-registry';
|
import * as sidecar from '@@/sidecar-registry';
|
||||||
import { join } from 'path';
|
import { join } from 'path';
|
||||||
import { getOwnerHomeDir, getEmailAccountsDir } from '../../../servers/data-path';
|
import { getOwnerHomeDir, getEmailAccountsDir } from '../../../servers/data-path';
|
||||||
import { getUserSettings, getEmailAccounts, getChatEventsSince, getLastChatEventSeq, appendChatEvent } from 'officerdb';
|
import { getUserSettings, getEmailAccounts, getChatEventsSince, appendChatEvent } from 'officerdb';
|
||||||
import { mkdirSync } from 'node:fs';
|
import { mkdirSync } from 'node:fs';
|
||||||
import { logger } from './logger';
|
import { logger } from './logger';
|
||||||
|
|
||||||
@@ -605,6 +613,30 @@ async function handleResumeCursor(
|
|||||||
if (msg.generating) await endTurnIfAgentIsGone(ws, sessionId, model);
|
if (msg.generating) await endTurnIfAgentIsGone(ws, sessionId, model);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Which background tasks are still outstanding, by replaying the durable log against itself.
|
||||||
|
*
|
||||||
|
* A task's whole life is two events — `task:started` and, eventually, `task:notification` with a terminal
|
||||||
|
* status — so started-minus-notified is the answer, and a Map keyed by taskId keeps the last word on each.
|
||||||
|
* There is no third event: a task the agent abandoned without notifying stays here until the log is pruned,
|
||||||
|
* which is the honest reading of the record rather than a bug to paper over.
|
||||||
|
*/
|
||||||
|
function collectRunningTasks(events: ServerMessage[]): RunningTask[] {
|
||||||
|
const running = new Map<string, RunningTask>();
|
||||||
|
for (const event of events) {
|
||||||
|
if (event.type === 'task:started') {
|
||||||
|
running.set(event.taskId, {
|
||||||
|
taskId: event.taskId,
|
||||||
|
description: event.description,
|
||||||
|
taskType: event.taskType,
|
||||||
|
});
|
||||||
|
} else if (event.type === 'task:notification') {
|
||||||
|
running.delete(event.taskId);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return [...running.values()];
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Re-bind a socket that knows only Claude's transcript uuid.
|
* Re-bind a socket that knows only Claude's transcript uuid.
|
||||||
*
|
*
|
||||||
@@ -656,11 +688,17 @@ async function handleAttach(ws: ServerWebSocket<WSData>, msg: { claudeSessionId:
|
|||||||
const isGenerating = existing ? session.isGenerating : await sidecar.isClaudeGenerating(sessionId);
|
const isGenerating = existing ? session.isGenerating : await sidecar.isClaudeGenerating(sessionId);
|
||||||
session.isGenerating = isGenerating;
|
session.isGenerating = isGenerating;
|
||||||
|
|
||||||
|
// One read serves both answers: the head of the log is the cursor, and folding the whole log gives the
|
||||||
|
// tasks still outstanding. Reading it all is affordable because attach happens once per socket and this
|
||||||
|
// server has one user; a `getLastChatEventSeq` would only have saved a second round trip.
|
||||||
let cursor = 0;
|
let cursor = 0;
|
||||||
|
let runningTasks: RunningTask[] = [];
|
||||||
try {
|
try {
|
||||||
cursor = (await getLastChatEventSeq(sessionId)) ?? 0;
|
const events = await getChatEventsSince(sessionId, 0);
|
||||||
|
cursor = events.at(-1)?.id ?? 0;
|
||||||
|
runningTasks = collectRunningTasks(events.map((e) => e.event as ServerMessage));
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
logger.error('Failed to read chat event head on attach', { sessionId, error: String(err) });
|
logger.error('Failed to read the durable log on attach', { sessionId, error: String(err) });
|
||||||
}
|
}
|
||||||
|
|
||||||
sendToClient(ws, {
|
sendToClient(ws, {
|
||||||
@@ -668,6 +706,7 @@ async function handleAttach(ws: ServerWebSocket<WSData>, msg: { claudeSessionId:
|
|||||||
sessionId,
|
sessionId,
|
||||||
isGenerating,
|
isGenerating,
|
||||||
cursor,
|
cursor,
|
||||||
|
runningTasks,
|
||||||
// Whatever the agent had typed but not yet finished as a message. The transcript on disk cannot
|
// Whatever the agent had typed but not yet finished as a message. The transcript on disk cannot
|
||||||
// supply it — the harness writes an assistant message only once it is complete — so this is the one
|
// supply it — the harness writes an assistant message only once it is complete — so this is the one
|
||||||
// piece of the turn a refresh would otherwise genuinely lose.
|
// piece of the turn a refresh would otherwise genuinely lose.
|
||||||
|
|||||||
@@ -71,6 +71,8 @@ export type ChatMessage =
|
|||||||
done?: boolean;
|
done?: boolean;
|
||||||
};
|
};
|
||||||
|
|
||||||
|
export type RunningTask = { taskId: string; description: string; taskType?: string };
|
||||||
|
|
||||||
export type TaskInfo = {
|
export type TaskInfo = {
|
||||||
taskName: string;
|
taskName: string;
|
||||||
taskDirName: string;
|
taskDirName: string;
|
||||||
@@ -105,7 +107,20 @@ export type ServerMessage =
|
|||||||
* transcript came over HTTP a moment ago and there is no shared id to reconcile the two records by, so
|
* transcript came over HTTP a moment ago and there is no shared id to reconcile the two records by, so
|
||||||
* this hands over the rest of the turn and the half-written paragraph, and nothing that would double up.
|
* this hands over the rest of the turn and the half-written paragraph, and nothing that would double up.
|
||||||
*/
|
*/
|
||||||
| { type: 'sync:live'; sessionId: string; isGenerating: boolean; streamingText: string; cursor: number }
|
| {
|
||||||
|
type: 'sync:live';
|
||||||
|
sessionId: string;
|
||||||
|
isGenerating: boolean;
|
||||||
|
streamingText: string;
|
||||||
|
cursor: number;
|
||||||
|
/**
|
||||||
|
* Background tasks started and not yet finished. The single exception to "carries no messages": a
|
||||||
|
* task row has no counterpart in Claude's transcript — officer synthesises it — so rebuilding from
|
||||||
|
* the file cannot bring one back, and the tray came up empty on a refresh while work was still
|
||||||
|
* running. Finished tasks are excluded server-side.
|
||||||
|
*/
|
||||||
|
runningTasks: RunningTask[];
|
||||||
|
}
|
||||||
| { type: 'error'; message: string; errorCode?: string }
|
| { type: 'error'; message: string; errorCode?: string }
|
||||||
| { type: 'stopped' }
|
| { type: 'stopped' }
|
||||||
/** The agent process went away mid-turn. Recoverable — see the `cutoff` message role. */
|
/** The agent process went away mid-turn. Recoverable — see the `cutoff` message role. */
|
||||||
|
|||||||
@@ -355,6 +355,19 @@ export function useChat(initialSessionId?: string, initialModel?: string | null,
|
|||||||
streamingRef.current = msg.streamingText;
|
streamingRef.current = msg.streamingText;
|
||||||
flushStreaming();
|
flushStreaming();
|
||||||
}
|
}
|
||||||
|
// The one thing the transcript genuinely cannot supply. Appended rather than spliced into place:
|
||||||
|
// a task row carries no timestamp to sort by, so a recovered one lands at the end of the
|
||||||
|
// conversation instead of where it was started. The tray reads by taskId and renders correctly
|
||||||
|
// either way, and the alternative is inventing an ordering the record does not contain.
|
||||||
|
if (msg.runningTasks?.length) {
|
||||||
|
setMessages((prev) => {
|
||||||
|
const known = new Set(prev.filter((m) => m.role === 'task').map((m) => m.taskId));
|
||||||
|
const recovered = msg.runningTasks
|
||||||
|
.filter((t) => !known.has(t.taskId))
|
||||||
|
.map((t): ChatMessage => ({ role: 'task', ...t }));
|
||||||
|
return recovered.length ? [...prev, ...recovered] : prev;
|
||||||
|
});
|
||||||
|
}
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user