From a0c3de6df9548f360610184ee5db57eb4c4aa999 Mon Sep 17 00:00:00 2001 From: Adam Firestone Date: Tue, 18 Aug 2026 13:12:00 -0500 Subject: [PATCH] fix: preserve dev server background task classification - Recover task identity from persisted lifecycle activities after restart - Prevent contradictory recovery events from promoting background tasks to agents --- .../Layers/ProviderRuntimeIngestion.test.ts | 59 +++++++ .../Layers/ProviderRuntimeIngestion.ts | 161 +++++++++++++----- .../src/state/subagentRuntime.test.ts | 19 +++ .../src/state/subagentRuntime.ts | 34 +++- 4 files changed, 229 insertions(+), 44 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 72fe08e84531..b4c22bc290f3 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -51,6 +51,7 @@ import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQu import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts"; import * as ThreadPlanProgress from "../ThreadPlanProgress.ts"; import { + inheritTaskIdentityFromActivities, ProviderRuntimeIngestionLive, runtimeEventToActivities, } from "./ProviderRuntimeIngestion.ts"; @@ -117,6 +118,64 @@ describe("runtimeEventToActivities", () => { }); }); + it("recovers background identity for a sparse completion after restart", () => { + const createdAt = "2026-08-17T22:16:52.640Z"; + const [started] = runtimeEventToActivities({ + type: "task.started", + eventId: asEventId("evt-dev-server-started"), + provider: ProviderDriverKind.make("claudeAgent"), + createdAt, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-1"), + payload: { + taskId: RuntimeTaskId.make("dev-server"), + description: "Start dev server in background", + taskType: "local_bash", + title: "Start dev server in background", + }, + }); + expect(started?.payload).toMatchObject({ + taskId: "dev-server", + taskType: "local_bash", + agentKind: "background", + }); + + const contradictoryRecovery = { + ...started!, + id: asEventId("evt-bad-recovery"), + kind: "task.completed", + payload: { taskId: "dev-server", status: "stopped", agentKind: "agent" }, + }; + const sparsePayload = { taskId: "dev-server", status: "stopped", agentKind: "agent" }; + const resolvedPayload = inheritTaskIdentityFromActivities(sparsePayload, [ + started!, + contradictoryRecovery, + ]); + const [completed] = runtimeEventToActivities( + { + type: "task.completed", + eventId: asEventId("evt-dev-server-completed"), + provider: ProviderDriverKind.make("claudeAgent"), + createdAt: "2026-08-18T17:52:10.469Z", + threadId: asThreadId("thread-1"), + payload: { + taskId: RuntimeTaskId.make("dev-server"), + status: "stopped", + }, + }, + undefined, + resolvedPayload, + ); + + expect(completed?.payload).toMatchObject({ + taskId: "dev-server", + status: "stopped", + taskType: "local_bash", + agentKind: "background", + title: "Start dev server in background", + }); + }); + it("persists prompt suggestions as hidden turn-scoped composer metadata", () => { const activities = runtimeEventToActivities({ type: "thread.metadata.updated", diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 8fd75cd84335..dee62e44cb3e 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -26,6 +26,7 @@ import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as Predicate from "effect/Predicate"; import * as Stream from "effect/Stream"; import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; @@ -51,39 +52,93 @@ import { canReplaceThreadTitle } from "../threadTitles.ts"; const providerTurnKey = (threadId: ThreadId, turnId: TurnId) => `${threadId}:${turnId}`; const providerTaskKey = (threadId: ThreadId, taskId: string) => `${threadId}:${taskId}`; -// Fallback when the in-memory description cache no longer has the task name -// (server restart, session-exit sweep, TTL/capacity eviction): earlier -// task.started/task.progress activities for the task are persisted with it. -function findTaskTitleInActivities( +const TASK_IDENTITY_KEYS = [ + "agentKind", + "taskType", + "agentId", + "title", + "role", + "model", + "effort", + "toolUseId", + "parentAgentId", + "workflowName", + "agentIndex", + "phaseIndex", + "phaseTitle", + "phases", + "attempt", + "runHandles", + "outputFile", + "agentPath", + "timelineBypass", +] as const; + +// Fallback when provider-local task state no longer exists (server restart, +// session recycle, TTL/capacity eviction): earlier task lifecycle activities +// persist the identity needed to classify and label a later sparse event. +// The first retained start is authoritative for classification; later rows +// may be recovery notifications that lost their provider-local linkage. +export function inheritTaskIdentityFromActivities( + payload: Record, activities: ReadonlyArray | undefined, - taskId: string, -): string | undefined { - if (!activities) { - return undefined; +): Record { + const taskId = Predicate.isString(payload.taskId) ? payload.taskId : undefined; + if (!taskId || !activities) { + return payload; } + const inherited: Record = {}; + let startedClassification: Record | undefined; + for (let index = activities.length - 1; index >= 0; index -= 1) { const activity = activities[index]; - if (!activity || (activity.kind !== "task.started" && activity.kind !== "task.progress")) { - continue; + if (!activity || !activity.kind.startsWith("task.")) continue; + const activityPayload = Predicate.isObject(activity.payload) + ? (activity.payload as Record) + : undefined; + if (activityPayload?.taskId !== taskId) continue; + for (const key of TASK_IDENTITY_KEYS) { + if (inherited[key] === undefined && activityPayload[key] !== undefined) { + inherited[key] = activityPayload[key]; + } } - const payload = - activity.payload && typeof activity.payload === "object" - ? (activity.payload as { taskId?: unknown; title?: unknown; detail?: unknown }) - : undefined; - if (payload?.taskId !== taskId) { - continue; + if ( + activity.kind === "task.started" && + startedClassification === undefined && + (activityPayload.agentKind === "agent" || activityPayload.agentKind === "background") + ) { + startedClassification = { + agentKind: activityPayload.agentKind, + ...(activityPayload.taskType !== undefined ? { taskType: activityPayload.taskType } : {}), + ...(activityPayload.agentId !== undefined ? { agentId: activityPayload.agentId } : {}), + }; } - const title = - typeof payload.title === "string" - ? payload.title - : activity.kind === "task.started" && typeof payload.detail === "string" - ? payload.detail - : undefined; - if (title && title.trim().length > 0) { - return title; + if ( + inherited.title === undefined && + activity.kind === "task.started" && + Predicate.isString(activityPayload.detail) + ) { + inherited.title = activityPayload.detail; } } - return undefined; + + return { ...inherited, ...payload, ...startedClassification }; +} + +function taskPayloadLacksClassification(payload: Record): boolean { + return ( + payload.agentKind !== "agent" && + payload.agentKind !== "background" && + !Predicate.isString(payload.taskType) && + !Predicate.isString(payload.agentId) && + !Predicate.isString(payload.parentAgentId) && + payload.timelineBypass !== true + ); +} + +function taskTitleFromPayload(payload: Record): string | undefined { + const title = payload.title; + return Predicate.isString(title) && title.trim().length > 0 ? title : undefined; } interface AssistantSegmentState { @@ -370,6 +425,7 @@ function taskLinkageActivityFields(payload: Record): Record, ): ReadonlyArray { const maybeSequence = (() => { const eventWithSequence = event as ProviderRuntimeEvent & { sessionSequence?: number }; @@ -587,7 +643,9 @@ export function runtimeEventToActivities( ...(event.payload.description ? { detail: truncateDetail(event.payload.description) } : {}), - ...taskLinkageActivityFields(event.payload as Record), + ...taskLinkageActivityFields( + resolvedTaskPayload ?? (event.payload as Record), + ), }, turnId: toTurnId(event.turnId) ?? null, ...maybeSequence, @@ -596,7 +654,9 @@ export function runtimeEventToActivities( } case "task.progress": { - const linkage = taskLinkageActivityFields(event.payload as Record); + const linkage = taskLinkageActivityFields( + resolvedTaskPayload ?? (event.payload as Record), + ); // Usage and activity are independent latest-state streams. Keeping them // under separate stable ids prevents a command/reasoning update from // replacing the last known token count (and prevents a usage-only tick @@ -695,7 +755,9 @@ export function runtimeEventToActivities( ...(event.payload.isBackgrounded !== undefined ? { isBackgrounded: event.payload.isBackgrounded } : {}), - ...taskLinkageActivityFields(event.payload as Record), + ...taskLinkageActivityFields( + resolvedTaskPayload ?? (event.payload as Record), + ), }, turnId: toTurnId(event.turnId) ?? null, ...maybeSequence, @@ -763,7 +825,9 @@ export function runtimeEventToActivities( } : {}), ...(event.payload.usage !== undefined ? { usage: event.payload.usage } : {}), - ...taskLinkageActivityFields(event.payload as Record), + ...taskLinkageActivityFields( + resolvedTaskPayload ?? (event.payload as Record), + ), }, turnId: toTurnId(event.turnId) ?? null, ...maybeSequence, @@ -2003,6 +2067,32 @@ const make = Effect.gen(function* () { yield* rememberTaskDescription(thread.id, event.payload.taskId, description); } } + let taskTitle: string | undefined; + let resolvedTaskPayload: Record | undefined; + if ( + event.type === "task.progress" || + event.type === "task.updated" || + event.type === "task.completed" + ) { + const payload = event.payload as Record; + resolvedTaskPayload = payload; + if (event.type === "task.completed") { + taskTitle = + taskTitleFromPayload(payload) ?? + (yield* lookupTaskDescription(thread.id, event.payload.taskId)); + } + if ( + (event.type === "task.completed" && !taskTitle) || + taskPayloadLacksClassification(payload) + ) { + const threadDetail = yield* getLoadedThreadDetail(); + resolvedTaskPayload = inheritTaskIdentityFromActivities( + payload, + threadDetail?.activities, + ); + taskTitle = taskTitle ?? taskTitleFromPayload(resolvedTaskPayload); + } + } // Working-indicator plan progress: current step while the turn runs, // cleared on settle so a finished plan never lingers as stale UI. // Events carrying a turn id that conflicts with the active turn are @@ -2025,7 +2115,7 @@ const make = Effect.gen(function* () { case "task.progress": case "task.updated": case "task.completed": { - const payload = event.payload as { + const payload = (resolvedTaskPayload ?? event.payload) as { taskId: string; taskType?: string; status?: string; @@ -2057,16 +2147,7 @@ const make = Effect.gen(function* () { break; } - let taskTitle: string | undefined; - if (event.type === "task.completed") { - taskTitle = yield* lookupTaskDescription(thread.id, event.payload.taskId); - if (!taskTitle) { - const threadDetail = yield* getLoadedThreadDetail(); - taskTitle = findTaskTitleInActivities(threadDetail?.activities, event.payload.taskId); - } - } - - const activities = runtimeEventToActivities(event, taskTitle); + const activities = runtimeEventToActivities(event, taskTitle, resolvedTaskPayload); yield* Effect.forEach(activities, (activity) => providerCommandId(event, "thread-activity-append").pipe( Effect.flatMap((commandId) => diff --git a/packages/client-runtime/src/state/subagentRuntime.test.ts b/packages/client-runtime/src/state/subagentRuntime.test.ts index 84cdd6be89a0..978a2f88040e 100644 --- a/packages/client-runtime/src/state/subagentRuntime.test.ts +++ b/packages/client-runtime/src/state/subagentRuntime.test.ts @@ -622,6 +622,25 @@ describe("background task exclusion", () => { expect(agents.map((agent) => agent.id)).toEqual(["agent-1"]); }); + it("never promotes a background task from a contradictory recovery completion", () => { + const agents = fold([ + activity("task.started", { + taskId: "dev-server", + taskType: "local_bash", + agentKind: "background", + title: "Start dev server in background", + }), + activity("task.completed", { + taskId: "dev-server", + agentKind: "agent", + status: "stopped", + title: "Start dev server in background", + }), + ]); + + expect(agents).toHaveLength(0); + }); + it("includes an explicitly promoted external agent even when its SDK task is a shell", () => { const agents = fold([ activity("task.started", { diff --git a/packages/client-runtime/src/state/subagentRuntime.ts b/packages/client-runtime/src/state/subagentRuntime.ts index c1ea1cc2b15d..e54668a68353 100644 --- a/packages/client-runtime/src/state/subagentRuntime.ts +++ b/packages/client-runtime/src/state/subagentRuntime.ts @@ -461,6 +461,32 @@ export function foldSubagentActivities( options?: { readonly sessionLive?: boolean }, ): ReadonlyArray { const agents = new Map(); + const taskClassifications = new Map< + string, + { readonly kind: "agent" | "background"; readonly fromStart: boolean } + >(); + + // Classification belongs to task identity, not to an individual lifecycle + // row. Prefer task.started even when it arrives late; otherwise keep the + // first retained stamp. This prevents sparse restart-recovery completions + // from promoting a known shell into the agent roster. + for (const activity of activities) { + if (!activity.kind.startsWith("task.")) continue; + if (typeof activity.payload !== "object" || activity.payload === null) continue; + const payload = activity.payload as Record; + const taskId = asString(payload.taskId); + if (!taskId) continue; + const fromStart = activity.kind === "task.started"; + const existing = taskClassifications.get(taskId); + if (existing?.fromStart || (!fromStart && existing)) continue; + taskClassifications.set(taskId, { + kind: isBackgroundTaskActivity(payload) ? "background" : "agent", + fromStart, + }); + } + + const isAgentTask = (taskId: string): boolean => + taskClassifications.get(taskId)?.kind === "agent"; for (const activity of activities) { if (typeof activity.payload !== "object" || activity.payload === null) { @@ -476,7 +502,7 @@ export function foldSubagentActivities( // Only real agents join the roster. Shells, monitors, and plan-mode // tasks are background work — they render in the ordinary work log, // not the Agents surface (a "Run 12s stall" shell is not a subagent). - if (isBackgroundTaskActivity(payload)) break; + if (!isAgentTask(taskId)) break; const agent = getOrCreate(agents, taskId, payload, at); fillMetadata(agent, payload); // Order-robustness: a start row arriving after a terminal state is a @@ -505,7 +531,7 @@ export function foldSubagentActivities( // rows often carry only taskId+status, no marker fields) inherit the // first row's classification instead of being re-judged. const existed = agents.has(taskId); - if (!existed && isBackgroundTaskActivity(payload)) break; + if (!isAgentTask(taskId)) break; const agent = getOrCreate(agents, taskId, payload, at); fillMetadata(agent, payload); if (agent.activationCount === 0) agent.activationCount = 1; @@ -543,7 +569,7 @@ export function foldSubagentActivities( // Membership is sticky per taskId: rows after the first (terminal // rows often carry only taskId+status, no marker fields) inherit the // first row's classification instead of being re-judged. - if (!agents.has(taskId) && isBackgroundTaskActivity(payload)) break; + if (!isAgentTask(taskId)) break; const agent = getOrCreate(agents, taskId, payload, at); fillMetadata(agent, payload); // A task first seen via task.updated (start row aged out) has run at @@ -571,7 +597,7 @@ export function foldSubagentActivities( // Membership is sticky per taskId: rows after the first (terminal // rows often carry only taskId+status, no marker fields) inherit the // first row's classification instead of being re-judged. - if (!agents.has(taskId) && isBackgroundTaskActivity(payload)) break; + if (!isAgentTask(taskId)) break; const agent = getOrCreate(agents, taskId, payload, at); fillMetadata(agent, payload); if (agent.activationCount === 0) agent.activationCount = 1;