diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.activity.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.activity.test.ts new file mode 100644 index 00000000000..93604103864 --- /dev/null +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.activity.test.ts @@ -0,0 +1,84 @@ +import { + EventId, + ProviderDriverKind, + RuntimeTaskId, + ThreadId, + type ProviderRuntimeEvent, +} from "@t3tools/contracts"; +import { describe, expect, it } from "vite-plus/test"; + +import { runtimeEventToActivities } from "./ProviderRuntimeIngestion.ts"; + +const base = { + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-08-06T00:00:00.000Z", + threadId: ThreadId.make("thread-1"), +}; + +describe("runtimeEventToActivities task progress", () => { + it("persists usage independently from replaceable activity", () => { + const taskId = RuntimeTaskId.make("agent-1"); + const usageOnly = { + ...base, + type: "task.progress", + eventId: EventId.make("evt-usage"), + payload: { + taskId, + description: "Agent one", + typedUsage: { totalTokens: 73_700_000 }, + }, + } satisfies ProviderRuntimeEvent; + const command = { + ...base, + type: "task.progress", + eventId: EventId.make("evt-command"), + payload: { + taskId, + description: "Agent one", + summary: "Running tests", + lastToolName: "exec_command", + }, + } satisfies ProviderRuntimeEvent; + + const usageActivities = runtimeEventToActivities(usageOnly); + const commandActivities = runtimeEventToActivities(command); + + expect(usageActivities.map((activity) => activity.id)).toEqual(["task-usage:thread-1:agent-1"]); + expect(commandActivities.map((activity) => activity.id)).toEqual([ + "task-progress:thread-1:agent-1", + ]); + const usagePayload = usageActivities[0]?.payload as Record | undefined; + expect(usagePayload?.typedUsage).toEqual({ totalTokens: 73_700_000 }); + expect(usagePayload?.usageSnapshot).toBe(true); + }); + + it("splits combined progress and usage into their independent snapshots", () => { + const event = { + ...base, + type: "task.progress", + eventId: EventId.make("evt-combined"), + payload: { + taskId: RuntimeTaskId.make("agent-2"), + description: "Agent two", + summary: "Inspecting the panel", + typedUsage: { totalTokens: 4_200, toolUses: 7 }, + status: "running", + }, + } satisfies ProviderRuntimeEvent; + + const activities = runtimeEventToActivities(event); + const progressPayload = activities[0]?.payload as Record; + const usagePayload = activities[1]?.payload as Record; + + expect(activities.map((activity) => activity.id)).toEqual([ + "task-progress:thread-1:agent-2", + "task-usage:thread-1:agent-2", + ]); + expect(progressPayload.summary).toBe("Inspecting the panel"); + expect(progressPayload.status).toBe("running"); + expect(progressPayload).not.toHaveProperty("typedUsage"); + expect(usagePayload.typedUsage).toEqual({ totalTokens: 4_200, toolUses: 7 }); + expect(usagePayload.usageSnapshot).toBe(true); + expect(usagePayload).not.toHaveProperty("status"); + }); +}); diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 0420420939e..189dd696106 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -563,39 +563,80 @@ export function runtimeEventToActivities( } case "task.progress": { + const linkage = taskLinkageActivityFields(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 + // from blanking the last meaningful activity). + const identityLinkage = { ...linkage }; + delete identityLinkage.typedUsage; + delete identityLinkage.status; + delete identityLinkage.error; + const title = + event.payload.description.trim().length > 0 + ? { title: truncateDetail(event.payload.description, 120) } + : {}; + const hasProgressState = + event.payload.typedUsage === undefined || + event.payload.summary !== undefined || + event.payload.lastToolName !== undefined || + event.payload.status !== undefined || + event.payload.error !== undefined; return [ - { - // Stable per-task id: progress is "latest state", not history, so - // each tick REPLACES the last via the activity upsert (PK + the - // replace-by-id apply in projector and client reducer). Keeps one - // progress row per task instead of thousands, so a large fleet's - // ticks can no longer evict its own start/terminal rows out of - // the 500-row retention window. Thread-scoped: activity_id is a - // GLOBAL primary key and Claude task ids are session-local, so a - // bare taskId could collide across threads and steal another - // thread's row (review finding). - id: EventId.make(`task-progress:${event.threadId}:${event.payload.taskId}`), - createdAt: event.createdAt, - tone: "info", - kind: "task.progress", - 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 } : {}), - ...(event.payload.usage !== undefined ? { usage: event.payload.usage } : {}), - ...taskLinkageActivityFields(event.payload as Record), - }, - turnId: toTurnId(event.turnId) ?? null, - ...maybeSequence, - }, + ...(hasProgressState + ? [ + { + // Stable per-task id: activity is "latest state", not + // history, so each meaningful tick replaces the last. This + // bounds a large fleet to one activity row per task. + id: EventId.make(`task-progress:${event.threadId}:${event.payload.taskId}`), + createdAt: event.createdAt, + tone: "info" as const, + kind: "task.progress" as const, + summary: + event.payload.description.trim().length > 0 + ? truncateDetail(event.payload.description, 120) + : "Reasoning update", + payload: { + taskId: event.payload.taskId, + ...title, + detail: truncateDetail(event.payload.summary ?? event.payload.description), + ...(event.payload.summary + ? { summary: truncateDetail(event.payload.summary) } + : {}), + ...(event.payload.lastToolName + ? { lastToolName: event.payload.lastToolName } + : {}), + ...(event.payload.status ? { status: event.payload.status } : {}), + ...(event.payload.error ? { error: event.payload.error } : {}), + ...(event.payload.usage !== undefined ? { usage: event.payload.usage } : {}), + ...identityLinkage, + }, + turnId: toTurnId(event.turnId) ?? null, + ...maybeSequence, + }, + ] + : []), + ...(event.payload.typedUsage !== undefined + ? [ + { + id: EventId.make(`task-usage:${event.threadId}:${event.payload.taskId}`), + createdAt: event.createdAt, + tone: "info" as const, + kind: "task.progress" as const, + summary: "Task usage updated", + payload: { + taskId: event.payload.taskId, + ...title, + ...identityLinkage, + usageSnapshot: true, + typedUsage: event.payload.typedUsage, + }, + turnId: toTurnId(event.turnId) ?? null, + ...maybeSequence, + }, + ] + : []), ]; } diff --git a/apps/web/src/components/AgentsPanel.tsx b/apps/web/src/components/AgentsPanel.tsx index 169c662e585..4eeff67ce5f 100644 --- a/apps/web/src/components/AgentsPanel.tsx +++ b/apps/web/src/components/AgentsPanel.tsx @@ -4,13 +4,11 @@ * spawn batch). * * Visualization rules (from live-test feedback): - * - Live work first: running workflows and direct spawns sort above settled. - * - Rows are flat status lines — no expansion, no per-agent tool feeds. The - * row answers "who / what phase / how much"; anything deeper is a future - * drill-in, not an unfold. - * - A settled workflow run collapses to a single summary line; click it to - * show its member list inline (the one allowed toggle — run granularity, - * not agent granularity). + * - Spawn order is stable. Activity and completion update rows in place. + * - Agent rows reserve three fixed lines for identity, activity, and metrics; + * changing data must never change their height. + * - Workflow expansion is presentation state. A live run stays expanded when + * it settles; older collapsed runs can still be opened at run granularity. * - Static status dots, DOM-write elapsed timers, plain token counters. */ import { useAtomValue } from "@effect/atom-react"; @@ -143,54 +141,50 @@ function AgentRow({ agent }: { agent: RuntimeSubagent }) { const visuals = STATUS_VISUALS[agent.status]; const activity = agentActivityText(agent); const modelLabel = formatSubagentModelLabel(agent.model, agent.effort); + const role = + agent.role?.trim().toLocaleLowerCase() === agent.title.trim().toLocaleLowerCase() + ? null + : agent.role; + const metadata = [ + modelLabel, + agent.usage ? `${formatSubagentTokenCount(agent.usage.totalTokens)} tok` : "— tok", + agent.usage?.toolUses !== undefined ? `${agent.usage.toolUses} tools` : null, + agent.activationCount > 1 ? `run ${agent.activationCount}` : null, + ].filter((value): value is string => value !== null); return ( -
-
- - - - - - {agent.title} - {agent.role ? ( - - {agent.role} - - ) : null} - - - {agent.status === "completed" ? ( - - ) : null} - +
+ + + + + {agent.title} + {role ? ( + + {role} - {activity ? ( - - {activity} - + ) : null} + + + + + {agent.status === "completed" ? ( + ) : null} - - {modelLabel ? {modelLabel} : null} - {agent.usage ? ( - - {modelLabel ? "· " : ""} - {formatSubagentTokenCount(agent.usage.totalTokens)} tok - - ) : null} - {agent.usage?.toolUses !== undefined ? ( - · {agent.usage.toolUses} tools - ) : null} - {agent.activationCount > 1 ? · run {agent.activationCount} : null} - {visuals.label} - -
+ + + {activity ?? visuals.label} + + + {metadata.join(" · ")} + + {visuals.label}
); } @@ -314,18 +308,32 @@ function WorkflowScriptView({ } /** - * Collapsible phase section (Claude Code Background-tasks pattern): live - * phases open by default, done phases collapsed to header + member dot row. - * User toggles override the default and stick for the phase's lifetime. + * Collapsible phase section. A phase opens when it becomes active, then keeps + * that shape as it settles so completion never yanks rows out from under the + * user. Manual toggles stick until a later activation begins. */ -function PhaseSection({ phase }: { phase: AgentPanelWorkflowGroup["phases"][number] }) { - const [userOpen, setUserOpen] = useState(null); - const open = userOpen ?? phase.state === "running"; +function PhaseSection({ + phase, + defaultOpen = false, +}: { + phase: AgentPanelWorkflowGroup["phases"][number]; + defaultOpen?: boolean; +}) { + const [open, setOpen] = useState(defaultOpen || phase.state === "running"); + const previousState = useRef(phase.state); + + useEffect(() => { + if (previousState.current !== "running" && phase.state === "running") { + setOpen(true); + } + previousState.current = phase.state; + }, [phase.state]); + return (
{scriptOpen && canShowScript ? ( @@ -416,7 +436,7 @@ function LiveWorkflowSection({ /> ) : null} {group.phases.map((phase) => ( - + ))} {group.unphasedMembers.map((member) => ( @@ -429,11 +449,16 @@ function LiveWorkflowSection({ } /** - * Settled workflow: one summary line. Click toggles the member list — the - * only expansion in the panel, at run granularity. + * Collapsed workflow: one summary line. The parent owns expansion so a live + * workflow keeps its shape when it settles. */ -function SettledWorkflowSection({ group }: { group: AgentPanelWorkflowGroup }) { - const [open, setOpen] = useState(false); +function CollapsedWorkflowSection({ + group, + onExpand, +}: { + group: AgentPanelWorkflowGroup; + onExpand: () => void; +}) { const members = workflowMembers(group); const failed = members.filter((member) => member.status === "failed").length; // Coordinator usage may already aggregate members (panel-footer rule): @@ -450,9 +475,9 @@ function SettledWorkflowSection({ group }: { group: AgentPanelWorkflowGroup }) {
- {open ? ( -
- {members.map((member) => ( - - ))} -
- ) : null}
); } +/** A workflow's open state is presentation state, not a status derivative. */ +function WorkflowSection({ + group, + environmentId, + threadId, +}: { + group: AgentPanelWorkflowGroup; + environmentId: EnvironmentId | null; + threadId: ThreadId | null; +}) { + const [open, setOpen] = useState(() => workflowIsLive(group)); + return open ? ( + setOpen(false)} + /> + ) : ( + setOpen(true)} /> + ); +} + export function AgentsPanel({ model, environmentId = null, @@ -503,48 +540,24 @@ export function AgentsPanel({ ); } - const liveWorkflows = model.workflows.filter(workflowIsLive); - const settledWorkflows = model.workflows.filter((group) => !workflowIsLive(group)); - const liveDirect = model.directAgents.filter( - (agent) => - agent.status === "running" || agent.status === "pending" || agent.status === "waiting", - ); - const settledDirect = model.directAgents.filter( - (agent) => - agent.status !== "running" && agent.status !== "pending" && agent.status !== "waiting", - ); - return (
- {liveWorkflows.map((group) => ( - ( + ))} - {liveDirect.length > 0 ? ( + {model.directAgents.length > 0 ? (
Direct spawns
- {liveDirect.map((agent) => ( - - ))} -
- ) : null} - {settledWorkflows.length > 0 || settledDirect.length > 0 ? ( -
-
- Earlier -
- {settledWorkflows.map((group) => ( - - ))} - {settledDirect.map((agent) => ( + {model.directAgents.map((agent) => ( ))}
diff --git a/packages/client-runtime/src/state/subagentRuntime.test.ts b/packages/client-runtime/src/state/subagentRuntime.test.ts index c6c758511a3..ceb40517550 100644 --- a/packages/client-runtime/src/state/subagentRuntime.test.ts +++ b/packages/client-runtime/src/state/subagentRuntime.test.ts @@ -186,6 +186,34 @@ describe("foldSubagentActivities", () => { expect(agents[0]!.usage).toEqual({ totalTokens: 900, inputTokens: 700 }); }); + it("usage snapshots enrich an existing agent without changing its status", () => { + const [agent] = fold([ + activity("task.started", { taskId: "usage-waiting", taskType: "local_agent" }), + activity("task.progress", { taskId: "usage-waiting", status: "waiting" }), + activity("task.progress", { + taskId: "usage-waiting", + usageSnapshot: true, + typedUsage: { totalTokens: 1_200 }, + }), + ]); + + expect(agent?.status).toBe("waiting"); + expect(agent?.usage?.totalTokens).toBe(1_200); + }); + + it("a retained usage snapshot can still reconstruct a running agent", () => { + const [agent] = fold([ + activity("task.progress", { + taskId: "usage-only", + usageSnapshot: true, + typedUsage: { totalTokens: 800 }, + }), + ]); + + expect(agent?.status).toBe("running"); + expect(agent?.usage?.totalTokens).toBe(800); + }); + it("partial terminal usage preserves known breakdown fields", () => { const agents = fold([ activity("task.started", { taskId: "task-6", taskType: "local_agent" }), @@ -362,6 +390,49 @@ describe("deriveAgentPanelModel", () => { ); }); + it("keeps direct spawns in first-seen order as their activity changes", () => { + const directRoster = fold([ + activity("task.started", { taskId: "direct-a", title: "First" }, "2026-08-01T11:00:00.000Z"), + activity("task.started", { taskId: "direct-b", title: "Second" }, "2026-08-01T11:00:01.000Z"), + activity( + "task.progress", + { taskId: "direct-a", summary: "Newest activity" }, + "2026-08-01T11:00:02.000Z", + ), + ]); + + expect( + deriveAgentPanelModel({ agents: directRoster }).directAgents.map((agent) => agent.id), + ).toEqual(["direct-a", "direct-b"]); + }); + + it("keeps first-seen order after the roster retention ranking runs", () => { + const starts = Array.from({ length: 101 }, (_, index) => + activity( + "task.started", + { taskId: `capped-${index}`, title: `Agent ${index}` }, + `2026-08-01T12:${String(Math.floor(index / 60)).padStart(2, "0")}:${String( + index % 60, + ).padStart(2, "0")}.000Z`, + ), + ); + const cappedRoster = fold([ + ...starts, + activity( + "task.progress", + { taskId: "capped-0", summary: "Newest activity" }, + "2026-08-01T12:02:00.000Z", + ), + ]); + + const ids = deriveAgentPanelModel({ agents: cappedRoster }).directAgents.map( + (agent) => agent.id, + ); + expect(ids).toHaveLength(100); + expect(ids.slice(0, 3)).toEqual(["capped-0", "capped-2", "capped-3"]); + expect(ids.at(-1)).toBe("capped-100"); + }); + it("a phase with only pending members never reads as running", () => { const pendingRoster = fold([ activity("task.started", { taskId: "wf-9", taskType: "local_workflow" }), diff --git a/packages/client-runtime/src/state/subagentRuntime.ts b/packages/client-runtime/src/state/subagentRuntime.ts index c81dd634140..e5f2b586b8c 100644 --- a/packages/client-runtime/src/state/subagentRuntime.ts +++ b/packages/client-runtime/src/state/subagentRuntime.ts @@ -80,6 +80,8 @@ export interface RuntimeSubagent { readonly phases: ReadonlyArray; readonly runHandles: SubagentRunHandles | null; readonly recentActivity: ReadonlyArray; + /** First retained observation, used as the roster's stable display order. */ + readonly firstSeenAt: string; readonly startedAt: string | null; readonly completedAt: string | null; readonly updatedAt: string; @@ -247,6 +249,7 @@ interface MutableAgent { phases: ReadonlyArray; runHandles: SubagentRunHandles | null; recentActivity: ReadonlyArray; + firstSeenAt: string; startedAt: string | null; completedAt: string | null; updatedAt: string; @@ -300,6 +303,7 @@ function getOrCreate( phases: [], runHandles: null, recentActivity: [], + firstSeenAt: at, startedAt: null, completedAt: null, updatedAt: at, @@ -500,14 +504,19 @@ 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; + const existed = agents.has(taskId); + if (!existed && isBackgroundTaskActivity(payload)) break; const agent = getOrCreate(agents, taskId, payload, at); fillMetadata(agent, payload); if (agent.activationCount === 0) agent.activationCount = 1; const explicitStatus = asRuntimeStatus(payload.status); if (explicitStatus) { applyStatus(agent, explicitStatus, at); - } else if (!isTerminalSubagentStatus(agent.status) && agent.status !== "idle") { + } else if ( + (payload.usageSnapshot !== true || !existed) && + !isTerminalSubagentStatus(agent.status) && + agent.status !== "idle" + ) { applyStatus(agent, "running", at); } const summary = asString(payload.summary); @@ -726,7 +735,10 @@ export function deriveAgentPanelModel({ return EMPTY_PANEL_MODEL; } - const workflows = source.filter((agent) => agent.kind === "workflow"); + const workflows = source + .filter((agent) => agent.kind === "workflow") + .slice() + .sort((a, b) => a.firstSeenAt.localeCompare(b.firstSeenAt) || a.id.localeCompare(b.id)); const workflowIds = new Set(workflows.map((workflow) => workflow.id)); const members = new Map(); const direct: RuntimeSubagent[] = []; @@ -827,7 +839,11 @@ export function deriveAgentPanelModel({ return { workflows: workflowGroups, - directAgents: direct.slice().sort((a, b) => b.updatedAt.localeCompare(a.updatedAt)), + // Updates and the >100-agent retention ranking must never reshuffle rows + // that remain visible. + directAgents: direct + .slice() + .sort((a, b) => a.firstSeenAt.localeCompare(b.firstSeenAt) || a.id.localeCompare(b.id)), runningCount, waitingCount, idleCount,