diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 001ba388949..72976768fec 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -2947,6 +2947,196 @@ describe("ProviderRuntimeIngestion", () => { ).toBe("# Plan title"); }); + it("titles task activities with the task description, including on completion", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + + harness.emit({ + type: "task.started", + eventId: asEventId("evt-named-task-started"), + provider: ProviderDriverKind.make("claudeAgent"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-named-task"), + payload: { + taskId: "named-task-1", + description: "Typecheck mobile app", + taskType: "local_bash", + }, + }); + + harness.emit({ + type: "task.progress", + eventId: asEventId("evt-named-task-progress"), + provider: ProviderDriverKind.make("claudeAgent"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-named-task"), + payload: { + taskId: "named-task-1", + description: "Typecheck mobile app", + summary: "Running tsc across the mobile workspace.", + }, + }); + + harness.emit({ + type: "task.completed", + eventId: asEventId("evt-named-task-completed"), + provider: ProviderDriverKind.make("claudeAgent"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-named-task"), + payload: { + taskId: "named-task-1", + status: "completed", + summary: "Typecheck finished without errors.", + }, + }); + + const thread = await waitForThread(harness.readModel, (entry) => + entry.activities.some( + (activity: ProviderRuntimeTestActivity) => activity.id === "evt-named-task-completed", + ), + ); + + const progress = thread.activities.find( + (activity: ProviderRuntimeTestActivity) => activity.id === "evt-named-task-progress", + ); + const completed = thread.activities.find( + (activity: ProviderRuntimeTestActivity) => activity.id === "evt-named-task-completed", + ); + + const progressPayload = + progress?.payload && typeof progress.payload === "object" + ? (progress.payload as Record) + : undefined; + const completedPayload = + completed?.payload && typeof completed.payload === "object" + ? (completed.payload as Record) + : undefined; + + expect(progress?.summary).toBe("Typecheck mobile app"); + expect(progressPayload?.title).toBe("Typecheck mobile app"); + expect(completed?.summary).toBe("Task completed"); + expect(completedPayload?.title).toBe("Typecheck mobile app"); + expect(completedPayload?.summary).toBe("Typecheck finished without errors."); + expect(completedPayload?.detail).toBe("Typecheck finished without errors."); + }); + + it("titles task completion from task.started when no progress event carried the name", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + + harness.emit({ + type: "task.started", + eventId: asEventId("evt-fast-task-started"), + provider: ProviderDriverKind.make("claudeAgent"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-fast-task"), + payload: { + taskId: "fast-task-1", + description: "wait for codex review to finish", + taskType: "local_bash", + }, + }); + + harness.emit({ + type: "task.completed", + eventId: asEventId("evt-fast-task-completed"), + provider: ProviderDriverKind.make("claudeAgent"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-fast-task"), + payload: { + taskId: "fast-task-1", + status: "completed", + }, + }); + + const thread = await waitForThread(harness.readModel, (entry) => + entry.activities.some( + (activity: ProviderRuntimeTestActivity) => activity.id === "evt-fast-task-completed", + ), + ); + + const completed = thread.activities.find( + (activity: ProviderRuntimeTestActivity) => activity.id === "evt-fast-task-completed", + ); + const completedPayload = + completed?.payload && typeof completed.payload === "object" + ? (completed.payload as Record) + : undefined; + + expect(completedPayload?.title).toBe("wait for codex review to finish"); + }); + + it("titles task completion from persisted activities after the description cache is swept", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + + harness.emit({ + type: "task.progress", + eventId: asEventId("evt-swept-task-progress"), + provider: ProviderDriverKind.make("claudeAgent"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-swept-task"), + payload: { + taskId: "swept-task-1", + description: "Watch round-3 CI and bots", + summary: "Polling CI checks.", + }, + }); + + await waitForThread(harness.readModel, (entry) => + entry.activities.some( + (activity: ProviderRuntimeTestActivity) => activity.id === "evt-swept-task-progress", + ), + ); + + // session.exited sweeps the in-memory description cache; the completion + // that follows must recover the name from persisted activities. + harness.emit({ + type: "session.exited", + eventId: asEventId("evt-swept-task-session-exited"), + provider: ProviderDriverKind.make("claudeAgent"), + createdAt: now, + threadId: asThreadId("thread-1"), + payload: {}, + }); + + harness.emit({ + type: "task.completed", + eventId: asEventId("evt-swept-task-completed"), + provider: ProviderDriverKind.make("claudeAgent"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-swept-task"), + payload: { + taskId: "swept-task-1", + status: "completed", + summary: "CI is green.", + }, + }); + + const thread = await waitForThread(harness.readModel, (entry) => + entry.activities.some( + (activity: ProviderRuntimeTestActivity) => activity.id === "evt-swept-task-completed", + ), + ); + + const completed = thread.activities.find( + (activity: ProviderRuntimeTestActivity) => activity.id === "evt-swept-task-completed", + ); + const completedPayload = + completed?.payload && typeof completed.payload === "object" + ? (completed.payload as Record) + : undefined; + + expect(completedPayload?.title).toBe("Watch round-3 CI and bots"); + }); + it("projects structured user input request and resolution as thread activities", async () => { const harness = await createHarness(); const now = "2026-01-01T00:00:00.000Z"; diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 3e5978f4846..ef3454348e4 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -40,6 +40,42 @@ import { import { ServerSettingsService } from "../../serverSettings.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( + activities: ReadonlyArray | undefined, + taskId: string, +): string | undefined { + if (!activities) { + return 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; + } + const payload = + activity.payload && typeof activity.payload === "object" + ? (activity.payload as { taskId?: unknown; title?: unknown; detail?: unknown }) + : undefined; + if (payload?.taskId !== taskId) { + continue; + } + 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; + } + } + return undefined; +} interface AssistantSegmentState { baseKey: string; @@ -53,6 +89,8 @@ const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_CACHE_CAPACITY = 20_000; const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_TTL = Duration.minutes(120); const BUFFERED_PROPOSED_PLAN_BY_ID_CACHE_CAPACITY = 10_000; const BUFFERED_PROPOSED_PLAN_BY_ID_TTL = Duration.minutes(120); +const TASK_DESCRIPTION_BY_TASK_CACHE_CAPACITY = 10_000; +const TASK_DESCRIPTION_BY_TASK_TTL = Duration.minutes(120); const MAX_BUFFERED_ASSISTANT_CHARS = 24_000; const STRICT_PROVIDER_LIFECYCLE_GUARD = process.env.T3CODE_STRICT_PROVIDER_LIFECYCLE_GUARD !== "0"; @@ -264,6 +302,7 @@ function requestKindFromCanonicalRequestType( function runtimeEventToActivities( event: ProviderRuntimeEvent, + taskTitle?: string, ): ReadonlyArray { const maybeSequence = (() => { const eventWithSequence = event as ProviderRuntimeEvent & { sessionSequence?: number }; @@ -473,9 +512,15 @@ function runtimeEventToActivities( createdAt: event.createdAt, tone: "info", kind: "task.progress", - summary: "Reasoning update", + summary: + event.payload.description.trim().length > 0 + ? truncateDetail(event.payload.description, 120) + : "Reasoning update", payload: { taskId: event.payload.taskId, + ...(event.payload.description.trim().length > 0 + ? { title: truncateDetail(event.payload.description, 120) } + : {}), detail: truncateDetail(event.payload.summary ?? event.payload.description), ...(event.payload.summary ? { summary: truncateDetail(event.payload.summary) } : {}), ...(event.payload.lastToolName ? { lastToolName: event.payload.lastToolName } : {}), @@ -503,7 +548,15 @@ function runtimeEventToActivities( payload: { taskId: event.payload.taskId, status: event.payload.status, - ...(event.payload.summary ? { detail: truncateDetail(event.payload.summary) } : {}), + ...(taskTitle ? { title: truncateDetail(taskTitle, 120) } : {}), + // summary + detail mirror task.progress: clients label the row from + // summary and keep detail for the preview/expanded body. + ...(event.payload.summary + ? { + summary: truncateDetail(event.payload.summary), + detail: truncateDetail(event.payload.summary), + } + : {}), ...(event.payload.usage !== undefined ? { usage: event.payload.usage } : {}), }, turnId: toTurnId(event.turnId) ?? null, @@ -666,6 +719,27 @@ const make = Effect.gen(function* () { lookup: () => Effect.succeed({ text: "", createdAt: "" }), }); + // Task names arrive on task.started/task.progress but not on task.completed, + // so remember them per task to title the completion activity. + const taskDescriptionByTaskKey = yield* Cache.make({ + capacity: TASK_DESCRIPTION_BY_TASK_CACHE_CAPACITY, + timeToLive: TASK_DESCRIPTION_BY_TASK_TTL, + lookup: () => Effect.succeed(""), + }); + + const rememberTaskDescription = (threadId: ThreadId, taskId: string, description: string) => + Cache.set(taskDescriptionByTaskKey, providerTaskKey(threadId, taskId), description); + + // Entries are left in place after completion so replayed or duplicate + // terminal events stay titled; TTL, capacity, and the session-exit sweep + // bound the cache. + const lookupTaskDescription = (threadId: ThreadId, taskId: string) => + Cache.getOption(taskDescriptionByTaskKey, providerTaskKey(threadId, taskId)).pipe( + Effect.map((description) => + Option.filter(description, (value) => value.length > 0).pipe(Option.getOrUndefined), + ), + ); + const resolveThreadDetail = Effect.fn("resolveThreadDetail")(function* (threadId: ThreadId) { return yield* projectionSnapshotQuery .getThreadDetailById(threadId) @@ -1090,6 +1164,7 @@ const make = Effect.gen(function* () { const turnKeys = Array.from(yield* Cache.keys(turnMessageIdsByTurnKey)); const assistantSegmentKeys = Array.from(yield* Cache.keys(assistantSegmentStateByTurnKey)); const proposedPlanKeys = Array.from(yield* Cache.keys(bufferedProposedPlanById)); + const taskDescriptionKeys = Array.from(yield* Cache.keys(taskDescriptionByTaskKey)); yield* Effect.forEach( turnKeys, (key) => @@ -1125,6 +1200,12 @@ const make = Effect.gen(function* () { : Effect.void, { concurrency: 1 }, ).pipe(Effect.asVoid); + yield* Effect.forEach( + taskDescriptionKeys, + (key) => + key.startsWith(prefix) ? Cache.invalidate(taskDescriptionByTaskKey, key) : Effect.void, + { concurrency: 1 }, + ).pipe(Effect.asVoid); }); const getSourceProposedPlanReferenceForPendingTurnStart = Effect.fn( @@ -1654,7 +1735,22 @@ const make = Effect.gen(function* () { } } - const activities = runtimeEventToActivities(event); + if (event.type === "task.started" || event.type === "task.progress") { + const description = event.payload.description?.trim(); + if (description) { + yield* rememberTaskDescription(thread.id, event.payload.taskId, description); + } + } + 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); yield* Effect.forEach(activities, (activity) => providerCommandId(event, "thread-activity-append").pipe( Effect.flatMap((commandId) =>