diff --git a/apps/mobile/src/lib/threadActivity.ts b/apps/mobile/src/lib/threadActivity.ts index 9f79a90550d..83ca2312b33 100644 --- a/apps/mobile/src/lib/threadActivity.ts +++ b/apps/mobile/src/lib/threadActivity.ts @@ -238,6 +238,7 @@ function deriveWorkLogEntries( if (activity.kind === "tool.started") continue; if (activity.kind === "task.started") continue; if (activity.kind === "context-window.updated") continue; + if (activity.kind === "agent.snapshot") continue; if (activity.summary === "Checkpoint captured") continue; if (isPlanBoundaryToolActivity(activity)) continue; entries.push(toDerivedWorkLogEntry(activity)); diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 20611e1ee75..3462955fccf 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -9,6 +9,8 @@ import { ProviderRuntimeEvent, ProviderSession, ProviderInstanceId, + RuntimeTaskId, + type ThreadAgentSnapshot, } from "@t3tools/contracts"; import { ApprovalRequestId, @@ -43,7 +45,11 @@ import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityRes import { OrchestrationEngineLive } from "./OrchestrationEngine.ts"; import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts"; import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts"; -import { ProviderRuntimeIngestionLive } from "./ProviderRuntimeIngestion.ts"; +import { + foldTaskAgentEvent, + ProviderRuntimeIngestionLive, + pruneSettledAgents, +} from "./ProviderRuntimeIngestion.ts"; import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeIngestion.ts"; import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"; @@ -318,6 +324,133 @@ describe("ProviderRuntimeIngestion", () => { }; } + it("folds material transitions, reactivations, and settled retention", () => { + const agents = new Map(); + const started: Extract = { + type: "task.started", + eventId: asEventId("agent-started"), + provider: ProviderDriverKind.make("claudeAgent"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:00.000Z", + payload: { + taskId: RuntimeTaskId.make("agent-1"), + description: "Reviewer", + taskType: "local_agent", + }, + }; + expect(foldTaskAgentEvent(agents, started)).toBe(true); + const usageTick: Extract = { + ...started, + type: "task.progress", + eventId: asEventId("agent-progress"), + createdAt: "2026-01-01T00:00:01.000Z", + payload: { + taskId: RuntimeTaskId.make("agent-1"), + description: "Reviewer", + usage: { totalTokens: 10 }, + }, + }; + expect(foldTaskAgentEvent(agents, usageTick)).toBe(false); + const completed: Extract = { + ...started, + type: "task.completed", + eventId: asEventId("agent-completed"), + createdAt: "2026-01-01T00:00:02.000Z", + payload: { taskId: RuntimeTaskId.make("agent-1"), status: "completed" }, + }; + expect(foldTaskAgentEvent(agents, completed)).toBe(true); + expect( + foldTaskAgentEvent(agents, { ...started, eventId: asEventId("agent-reactivated") }), + ).toBe(true); + expect(agents.get("agent-1")?.activationCount).toBe(2); + + for (let index = 0; index < 51; index += 1) { + agents.set(`settled-${index}`, { + agentId: `settled-${index}`, + provider: ProviderDriverKind.make("claudeAgent"), + kind: "subagent", + name: `Agent ${index}`, + status: "completed", + firstStartedAt: `2026-01-01T00:00:${String(index).padStart(2, "0")}.000Z`, + lastActivityAt: `2026-01-01T00:00:${String(index).padStart(2, "0")}.000Z`, + activationCount: 1, + recentActivity: [], + updatedAt: `2026-01-01T00:00:${String(index).padStart(2, "0")}.000Z`, + }); + } + pruneSettledAgents(agents); + expect( + Array.from(agents.values()).filter((agent) => agent.status === "completed"), + ).toHaveLength(50); + expect(agents.has("settled-0")).toBe(false); + }); + + it("appends latest-wins agent rosters only on material changes and sweeps orphans", async () => { + const harness = await createHarness(); + harness.emit({ + type: "task.started", + eventId: asEventId("evt-agent-start"), + provider: ProviderDriverKind.make("claudeAgent"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:01.000Z", + payload: { taskId: "agent-1", description: "Reviewer", taskType: "local_agent" }, + }); + await harness.drain(); + harness.emit({ + type: "task.progress", + eventId: asEventId("evt-agent-usage"), + provider: ProviderDriverKind.make("claudeAgent"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:02.000Z", + payload: { taskId: "agent-1", description: "Reviewer", usage: { totalTokens: 10 } }, + }); + await harness.drain(); + let thread = (await harness.readModel()).threads.find((entry) => entry.id === "thread-1"); + expect( + thread?.activities.filter((activity) => activity.kind === "agent.snapshot"), + ).toHaveLength(1); + + harness.emit({ + type: "session.exited", + eventId: asEventId("evt-session-exit-agents"), + provider: ProviderDriverKind.make("claudeAgent"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:03.000Z", + payload: { reason: "process_exit" }, + }); + await harness.drain(); + thread = (await harness.readModel()).threads.find((entry) => entry.id === "thread-1"); + const snapshots = + thread?.activities.filter((activity) => activity.kind === "agent.snapshot") ?? []; + expect(snapshots).toHaveLength(2); + const latest = snapshots.at(-1)?.payload as { agents: Array } | undefined; + expect(latest?.agents).toHaveLength(1); + expect(latest?.agents[0]?.status).toBe("stopped"); + expect(latest?.agents[0]?.errorMessage).toBe("orphaned on session exit"); + + // The session-exit sweep releases the in-memory roster; a later session on + // the same thread re-hydrates from the persisted snapshot instead of the + // evicted map (agent-1 stays stopped, not resurrected). + harness.emit({ + type: "task.started", + eventId: asEventId("evt-agent-restart"), + provider: ProviderDriverKind.make("claudeAgent"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:04.000Z", + payload: { taskId: "agent-2", description: "Follow-up", taskType: "local_agent" }, + }); + await harness.drain(); + thread = (await harness.readModel()).threads.find((entry) => entry.id === "thread-1"); + const rehydrated = + thread?.activities.filter((activity) => activity.kind === "agent.snapshot") ?? []; + expect(rehydrated).toHaveLength(3); + const roster = rehydrated.at(-1)?.payload as { agents: Array } | undefined; + expect(roster?.agents.map((agent) => [agent.agentId, agent.status])).toEqual([ + ["agent-1", "stopped"], + ["agent-2", "running"], + ]); + }); + it("maps turn started/completed events into thread session updates", 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 09566feb2b2..6c4b81fcffb 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -2,6 +2,7 @@ import { ApprovalRequestId, type AssistantDeliveryMode, CommandId, + EventId, MessageId, type OrchestrationEvent, type OrchestrationMessage, @@ -9,6 +10,12 @@ import { CheckpointRef, isToolLifecycleItemType, ThreadId, + THREAD_AGENTS_ACTIVITY_KIND, + THREAD_AGENT_RECENT_ACTIVITY_LIMIT, + THREAD_AGENT_TERMINAL_STATUSES, + type ThreadAgentSnapshot, + type ThreadAgentStatus, + type ThreadAgentUsage, type ThreadTokenUsageSnapshot, TurnId, type OrchestrationCheckpointSummary, @@ -41,6 +48,269 @@ import { ServerSettingsService } from "../../serverSettings.ts"; const providerTurnKey = (threadId: ThreadId, turnId: TurnId) => `${threadId}:${turnId}`; const providerTaskKey = (threadId: ThreadId, taskId: string) => `${threadId}:${taskId}`; +const AGENT_SETTLED_RETENTION_LIMIT = 50; +const AGENT_IDLE_RETENTION_LIMIT = 25; +const AGENT_USAGE_MATERIAL_STEP = 25_000; + +// Fields must satisfy the contract's NonNegativeInt or the persisted roster +// row becomes undecodable on the client. +function usageCount(value: unknown): number | undefined { + return typeof value === "number" && Number.isInteger(value) && value >= 0 ? value : undefined; +} + +function taskUsage(value: unknown): ThreadAgentUsage | undefined { + if (value === null || typeof value !== "object" || Array.isArray(value)) return undefined; + const usage = value as Record; + const totalTokens = usageCount(usage.totalTokens); + if (totalTokens === undefined) return undefined; + const inputTokens = usageCount(usage.inputTokens); + const cachedInputTokens = usageCount(usage.cachedInputTokens); + const outputTokens = usageCount(usage.outputTokens); + const reasoningOutputTokens = usageCount(usage.reasoningOutputTokens); + const toolUses = usageCount(usage.toolUses); + return { + totalTokens, + ...(inputTokens !== undefined ? { inputTokens } : {}), + ...(cachedInputTokens !== undefined ? { cachedInputTokens } : {}), + ...(outputTokens !== undefined ? { outputTokens } : {}), + ...(reasoningOutputTokens !== undefined ? { reasoningOutputTokens } : {}), + ...(toolUses !== undefined ? { toolUses } : {}), + }; +} + +function taskStatus(event: ProviderRuntimeEvent): ThreadAgentStatus | undefined { + if (event.type === "task.started") return "running"; + if (event.type === "task.completed") return event.payload.status; + if (event.type !== "task.updated" || !event.payload.status) return undefined; + return event.payload.status === "killed" + ? "stopped" + : event.payload.status === "paused" + ? "idle" + : event.payload.status; +} + +function taskKind(taskType: string | undefined, parentTaskId: string | undefined) { + if (taskType === "local_agent") return "subagent" as const; + if (taskType === "local_workflow") return "workflow" as const; + if (taskType === "workflow_agent") return "workflow_agent" as const; + if (taskType === "local_bash" || taskType === "shell") return "shell" as const; + if (taskType === "monitor") return "monitor" as const; + // Parent linkage is only a workflow-membership hint when the type itself is + // unknown — a nested shell/monitor task keeps its explicit kind above. + if (parentTaskId) return "workflow_agent" as const; + return "other" as const; +} + +function isTaskAgentEvent( + event: ProviderRuntimeEvent, +): event is Extract< + ProviderRuntimeEvent, + { type: "task.started" | "task.progress" | "task.updated" | "task.completed" } +> { + return ( + event.type === "task.started" || + event.type === "task.progress" || + event.type === "task.updated" || + event.type === "task.completed" + ); +} + +function workflowPhasesEqual( + left: ThreadAgentSnapshot["phases"], + right: ThreadAgentSnapshot["phases"], +): boolean { + return ( + left === right || + (left !== undefined && + right !== undefined && + left.length === right.length && + left.every( + (phase, index) => + phase.index === right[index]?.index && phase.title === right[index]?.title, + )) + ); +} + +const AGENT_TEXT_LIMIT = 180; + +/** Roster snapshots duplicate these strings on every append — keep them small. */ +function boundAgentText(value: string): string { + return value.length > AGENT_TEXT_LIMIT ? `${value.slice(0, AGENT_TEXT_LIMIT - 1)}…` : value; +} + +export function foldTaskAgentEvent( + agents: Map, + event: Extract< + ProviderRuntimeEvent, + { type: "task.started" | "task.progress" | "task.updated" | "task.completed" } + >, +): boolean { + const payload = event.payload; + const previous = agents.get(payload.taskId); + const status = taskStatus(event) ?? previous?.status ?? "running"; + const description = "description" in payload ? payload.description : undefined; + const summary = "summary" in payload ? payload.summary : undefined; + const nextUsage = "usage" in payload ? taskUsage(payload.usage) : undefined; + const lastToolName = "lastToolName" in payload ? payload.lastToolName : undefined; + const settledNow = status === "idle" || THREAD_AGENT_TERMINAL_STATUSES.has(status); + // A settled agent's card shows its result/error, not a stale in-flight + // activity line ("Reading files…" on a failed card). + const rawCurrentActivity = settledNow + ? undefined + : (summary ?? lastToolName ?? previous?.currentActivity); + const currentActivity = + rawCurrentActivity === undefined ? undefined : boundAgentText(rawCurrentActivity); + const activityChanged = + (summary !== undefined && summary !== previous?.currentActivity) || + (lastToolName !== undefined && lastToolName !== previous?.lastToolName); + const recentActivity = activityChanged + ? [ + ...(previous?.recentActivity ?? []), + { + at: event.createdAt, + // Bounded: Codex child item summaries can carry full command/item + // text, and each entry is duplicated into every roster snapshot. + summary: boundAgentText(summary ?? lastToolName ?? "Activity updated"), + }, + ].slice(-THREAD_AGENT_RECENT_ACTIVITY_LIMIT) + : (previous?.recentActivity ?? []); + const wasSettled = + previous !== undefined && + (previous.status === "idle" || THREAD_AGENT_TERMINAL_STATUSES.has(previous.status)); + // Any non-settled status counts as a fresh activation — an idle Codex agent + // resumed via a waiting approval still increments, not just idle→running. + const reactivated = + wasSettled && status !== "idle" && !THREAD_AGENT_TERMINAL_STATUSES.has(status); + const explicitEndTime = event.type === "task.updated" ? event.payload.endTime : undefined; + const terminal = THREAD_AGENT_TERMINAL_STATUSES.has(status); + // Re-derive kind whenever this event carries a taskType/parent — a child + // registered from an early notification (before its parent linkage arrived) + // must not stay pinned to a first-guess "other". + const eventKind = taskKind( + "taskType" in payload ? payload.taskType : undefined, + payload.parentTaskId, + ); + const next: ThreadAgentSnapshot = { + agentId: payload.taskId, + provider: event.provider, + kind: eventKind !== "other" ? eventKind : (previous?.kind ?? eventKind), + name: payload.name ?? description ?? payload.workflowName ?? previous?.name ?? payload.taskId, + ...((payload.agentType ?? previous?.agentType) + ? { agentType: payload.agentType ?? previous?.agentType } + : {}), + ...((payload.model ?? previous?.model) ? { model: payload.model ?? previous?.model } : {}), + status, + ...(currentActivity ? { currentActivity } : {}), + ...((lastToolName ?? previous?.lastToolName) + ? { lastToolName: lastToolName ?? previous?.lastToolName } + : {}), + // Merge field-wise: a payload carrying only totalTokens (e.g. a terminal + // Claude task_notification) must not wipe previously known breakdowns. + ...(nextUsage || previous?.usage + ? { usage: { ...previous?.usage, ...nextUsage } as ThreadAgentUsage } + : {}), + firstStartedAt: previous?.firstStartedAt ?? event.createdAt, + // Reset per activation so live timers exclude idle gaps between runs. + lastStartedAt: reactivated + ? event.createdAt + : (previous?.lastStartedAt ?? previous?.firstStartedAt ?? event.createdAt), + lastActivityAt: event.createdAt, + // Prefer the provider's explicit end time, then an already-recorded one; + // only stamp ingestion time on the FIRST terminal transition so duplicate + // terminal events don't keep sliding the end forward. + ...(!reactivated && + (explicitEndTime ?? previous?.endedAt ?? (terminal ? event.createdAt : undefined)) + ? { + endedAt: explicitEndTime ?? previous?.endedAt ?? (terminal ? event.createdAt : undefined), + } + : {}), + activationCount: previous ? previous.activationCount + (reactivated ? 1 : 0) : 1, + ...((event.turnId ?? previous?.lastTurnId) + ? { lastTurnId: TurnId.make(String(event.turnId ?? previous?.lastTurnId)) } + : {}), + ...((payload.parentTaskId ?? previous?.parentAgentId) + ? { parentAgentId: payload.parentTaskId ?? previous?.parentAgentId } + : {}), + ...(payload.phaseIndex !== undefined || previous?.phaseIndex !== undefined + ? { phaseIndex: payload.phaseIndex ?? previous?.phaseIndex } + : {}), + ...((payload.phaseTitle ?? previous?.phaseTitle) + ? { phaseTitle: payload.phaseTitle ?? previous?.phaseTitle } + : {}), + ...((payload.phases ?? previous?.phases) ? { phases: payload.phases ?? previous?.phases } : {}), + ...((payload.scriptPath ?? previous?.scriptPath) + ? { scriptPath: payload.scriptPath ?? previous?.scriptPath } + : {}), + ...((payload.runId ?? previous?.runId) ? { runId: payload.runId ?? previous?.runId } : {}), + ...((payload.outputFile ?? previous?.outputFile) + ? { outputFile: payload.outputFile ?? previous?.outputFile } + : {}), + // A re-activated agent starts a fresh run: prior result/error text would + // read as the live run's output, so clear both unless this event sets them. + ...(event.type === "task.completed" && event.payload.summary + ? { resultSummary: event.payload.summary } + : !reactivated && previous?.resultSummary + ? { resultSummary: previous.resultSummary } + : {}), + ...(event.type === "task.updated" && event.payload.errorMessage + ? { errorMessage: event.payload.errorMessage } + : !reactivated && previous?.errorMessage + ? { errorMessage: previous.errorMessage } + : {}), + recentActivity, + updatedAt: event.createdAt, + }; + agents.set(payload.taskId, next); + if (!previous) return true; + const usageStepChanged = + Math.floor((previous.usage?.totalTokens ?? 0) / AGENT_USAGE_MATERIAL_STEP) !== + Math.floor((next.usage?.totalTokens ?? 0) / AGENT_USAGE_MATERIAL_STEP); + return ( + previous.status !== next.status || + previous.kind !== next.kind || + previous.parentAgentId !== next.parentAgentId || + previous.name !== next.name || + previous.model !== next.model || + previous.phaseIndex !== next.phaseIndex || + previous.phaseTitle !== next.phaseTitle || + !workflowPhasesEqual(previous.phases, next.phases) || + previous.currentActivity !== next.currentActivity || + activityChanged || + usageStepChanged || + previous.endedAt !== next.endedAt || + previous.resultSummary !== next.resultSummary || + previous.errorMessage !== next.errorMessage || + previous.scriptPath !== next.scriptPath || + previous.runId !== next.runId || + previous.outputFile !== next.outputFile + ); +} + +export function pruneSettledAgents(agents: Map): void { + // Terminal agents are history and compete for the standard retention pool. + const settled = Array.from(agents.values()) + .filter((agent) => THREAD_AGENT_TERMINAL_STATUSES.has(agent.status)) + .sort((left, right) => left.lastActivityAt.localeCompare(right.lastActivityAt)); + for (const agent of settled.slice( + 0, + Math.max(0, settled.length - AGENT_SETTLED_RETENTION_LIMIT), + )) { + agents.delete(agent.agentId); + } + // Idle agents are resumable identities, so they get their own (generous) + // cap rather than sharing the terminal pool — but Codex children end every + // run idle and never terminal until close/interrupt, so without a bound a + // long session accumulates them all and every material append persists the + // full roster. Evicting oldest-idle is safe: a resumed identity + // re-registers from its next event (the activation counter tolerates the + // restart). + const idle = Array.from(agents.values()) + .filter((agent) => agent.status === "idle") + .sort((left, right) => left.lastActivityAt.localeCompare(right.lastActivityAt)); + for (const agent of idle.slice(0, Math.max(0, idle.length - AGENT_IDLE_RETENTION_LIMIT))) { + agents.delete(agent.agentId); + } +} // Fallback when the in-memory description cache no longer has the task name // (server restart, session-exit sweep, TTL/capacity eviction): earlier @@ -94,9 +364,9 @@ 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"; -type TurnStartRequestedDomainEvent = Extract< +type IngestionDomainEvent = Extract< OrchestrationEvent, - { type: "thread.turn-start-requested" } + { type: "thread.turn-start-requested" | "thread.reverted" } >; type RuntimeIngestionInput = @@ -106,7 +376,7 @@ type RuntimeIngestionInput = } | { source: "domain"; - event: TurnStartRequestedDomainEvent; + event: IngestionDomainEvent; }; function toTurnId(value: TurnId | string | undefined): TurnId | undefined { @@ -310,6 +580,12 @@ export function runtimeEventToActivities( event: ProviderRuntimeEvent, taskTitle?: string, ): ReadonlyArray { + if ( + (isTaskAgentEvent(event) && event.payload.timelineBypass === true) || + event.type === "task.updated" + ) { + return []; + } const maybeSequence = (() => { const eventWithSequence = event as ProviderRuntimeEvent & { sessionSequence?: number }; return eventWithSequence.sessionSequence !== undefined @@ -693,6 +969,23 @@ const make = Effect.gen(function* () { const providerService = yield* ProviderService; const projectionTurnRepository = yield* ProjectionTurnRepository; const serverSettingsService = yield* ServerSettingsService; + const agentsByThread = new Map>(); + const hydratedAgentThreads = new Set(); + // Threads whose latest roster failed to persist: the in-memory state is + // newer than the last agent.snapshot activity, so the next event must + // re-dispatch even if it is not material on its own. + const pendingRosterRedispatch = new Set(); + // Activities appended per thread since its last agent.snapshot. The + // activities projection keeps only the newest 500 rows; refreshing the + // roster before the last snapshot can age out guarantees hydration (and the + // client's latest-wins scan) always finds one while agents exist. + const activitiesSinceAgentSnapshot = new Map(); + const AGENT_SNAPSHOT_REFRESH_ACTIVITY_COUNT = 400; + // Monotonic per-thread roster revision. Snapshot activities carry it so the + // client picks the newest roster deterministically even when two appends + // share a createdAt millisecond. Seeded from the hydrated snapshot. + const agentRosterRevision = new Map(); + let agentSnapshotDispatchCount = 0; const providerCommandId = (event: ProviderRuntimeEvent, tag: string) => crypto.randomUUIDv4.pipe( Effect.map((uuid) => CommandId.make(`provider:${event.eventId}:${tag}:${uuid}`)), @@ -1305,6 +1598,181 @@ const make = Effect.gen(function* () { return loadedThreadDetail; }); + // Hydrate lazily: on agent-touching events, or amortized once ordinary + // traffic has appended enough activities that a persisted snapshot + // could be approaching the 500-row cap. Never on every first event — + // message deltas after a restart must not each trigger a full + // thread-detail load through the serialized ingestion worker. + const eventTouchesAgents = isTaskAgentEvent(event) || event.type === "session.exited"; + const activityPressure = + (activitiesSinceAgentSnapshot.get(thread.id) ?? 0) >= AGENT_SNAPSHOT_REFRESH_ACTIVITY_COUNT; + if ((eventTouchesAgents || activityPressure) && !hydratedAgentThreads.has(thread.id)) { + const detail = yield* getLoadedThreadDetail(); + const activityList = detail?.activities ?? []; + // Select the snapshot the same way the client does — highest revision + // wins, list position breaks ties — so the reducer never resumes from + // a stale lower-revision roster that landed later in the list. + let latestSnapshotIndex = -1; + let bestRevision = -1; + for (let index = 0; index < activityList.length; index += 1) { + const activity = activityList[index]; + if (!activity || activity.kind !== THREAD_AGENTS_ACTIVITY_KIND) continue; + const revision = + activity.payload !== null && + typeof activity.payload === "object" && + typeof (activity.payload as { revision?: unknown }).revision === "number" + ? ((activity.payload as { revision: number }).revision ?? -1) + : -1; + if (revision >= bestRevision) { + bestRevision = revision; + latestSnapshotIndex = index; + } + } + const latestSnapshot = + latestSnapshotIndex >= 0 ? activityList[latestSnapshotIndex] : undefined; + // Seed the refresh counter with the snapshot's actual age in the + // capped projection — after a restart it may already sit hundreds of + // rows deep, and starting from zero would let it age out before the + // first refresh. + activitiesSinceAgentSnapshot.set( + thread.id, + latestSnapshotIndex >= 0 ? activityList.length - 1 - latestSnapshotIndex : 0, + ); + if (bestRevision >= 0) { + agentRosterRevision.set(thread.id, bestRevision); + } + const payload = + latestSnapshot?.payload !== null && typeof latestSnapshot?.payload === "object" + ? (latestSnapshot.payload as { agents?: unknown }) + : undefined; + const agents = new Map(); + if (Array.isArray(payload?.agents)) { + for (const candidate of payload.agents) { + if ( + candidate !== null && + typeof candidate === "object" && + typeof (candidate as { agentId?: unknown }).agentId === "string" + ) { + const agent = candidate as ThreadAgentSnapshot; + agents.set(agent.agentId, agent); + } + } + } + agentsByThread.set(thread.id, agents); + hydratedAgentThreads.add(thread.id); + } + + const threadAgents = agentsByThread.get(thread.id) ?? new Map(); + agentsByThread.set(thread.id, threadAgents); + let agentRosterMaterial = false; + if (isTaskAgentEvent(event)) { + agentRosterMaterial = foldTaskAgentEvent(threadAgents, event); + } else if (event.type === "session.exited") { + for (const [agentId, agent] of threadAgents) { + if (THREAD_AGENT_TERMINAL_STATUSES.has(agent.status)) continue; + threadAgents.set(agentId, { + ...agent, + status: "stopped", + endedAt: event.createdAt, + lastActivityAt: event.createdAt, + errorMessage: "orphaned on session exit", + updatedAt: event.createdAt, + }); + agentRosterMaterial = true; + } + } + if (pendingRosterRedispatch.has(thread.id) && threadAgents.size > 0) { + agentRosterMaterial = true; + } + if ( + !agentRosterMaterial && + threadAgents.size > 0 && + (activitiesSinceAgentSnapshot.get(thread.id) ?? 0) >= AGENT_SNAPSHOT_REFRESH_ACTIVITY_COUNT + ) { + agentRosterMaterial = true; + } + if (agentRosterMaterial) { + pruneSettledAgents(threadAgents); + const roster = Array.from(threadAgents.values()); + // "active" covers everything non-settled (running/pending/waiting) so a + // roster of blocked agents doesn't read as "0 agents settled". + const active = roster.filter( + (agent) => agent.status !== "idle" && !THREAD_AGENT_TERMINAL_STATUSES.has(agent.status), + ).length; + const settled = roster.length - active; + const count = active > 0 ? active : settled; + const summary = `${count} ${count === 1 ? "agent" : "agents"} ${active > 0 ? "active" : "settled"}`; + const nextRevision = (agentRosterRevision.get(thread.id) ?? 0) + 1; + const snapshotUuid = yield* crypto.randomUUIDv4; + const activity: OrchestrationThreadActivity = { + id: EventId.make(`${event.eventId}:agent-snapshot:${snapshotUuid}`), + createdAt: event.createdAt, + tone: "info", + kind: THREAD_AGENTS_ACTIVITY_KIND, + summary, + payload: { agents: roster, revision: nextRevision }, + turnId: toTurnId(event.turnId) ?? null, + }; + // The in-memory roster is authoritative (already folded); if the + // append fails, keep it and flag the thread so the NEXT event + // re-dispatches the full roster — discarding memory here would lose + // transitions newer than the last persisted snapshot. The failure is + // swallowed (logged) rather than raised: agent observability must + // never block the rest of this event's ingestion (message deltas, + // session status, timeline activities). Interrupts still propagate. + pendingRosterRedispatch.add(thread.id); + const appendSnapshot = orchestrationEngine + .dispatch({ + type: "thread.activity.append", + commandId: yield* providerCommandId(event, "agent-snapshot-append"), + threadId: thread.id, + activity, + createdAt: activity.createdAt, + }) + .pipe( + Effect.andThen( + Effect.gen(function* () { + pendingRosterRedispatch.delete(thread.id); + activitiesSinceAgentSnapshot.set(thread.id, 0); + agentRosterRevision.set(thread.id, nextRevision); + agentSnapshotDispatchCount += 1; + yield* Effect.logDebug("provider agent snapshot appended", { + threadId: thread.id, + agentCount: roster.length, + dispatchCount: agentSnapshotDispatchCount, + }); + }), + ), + ); + // session.exited is often the thread's LAST event — the usual + // "re-dispatch on the next event" recovery never fires, so retry the + // final orphan-sweep roster once before giving up on it. + yield* appendSnapshot.pipe( + event.type === "session.exited" + ? Effect.retry({ times: 1 }) + : (effect: typeof appendSnapshot) => effect, + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause) + : Effect.logWarning("provider agent snapshot append failed; will re-dispatch", { + threadId: thread.id, + agentCount: roster.length, + cause: Cause.pretty(cause), + }), + ), + ); + } + + // Sessions are done producing agent events once they exit; release the + // per-thread reducer state. A later session re-hydrates from the + // persisted snapshot (the exit sweep above just wrote the final roster). + if (event.type === "session.exited" && !pendingRosterRedispatch.has(thread.id)) { + agentsByThread.delete(thread.id); + hydratedAgentThreads.delete(thread.id); + activitiesSinceAgentSnapshot.delete(thread.id); + agentRosterRevision.delete(thread.id); + } + const now = event.createdAt; const eventTurnId = toTurnId(event.turnId); const activeTurnId = thread.session?.activeTurnId ?? null; @@ -1759,6 +2227,19 @@ const make = Effect.gen(function* () { } const activities = runtimeEventToActivities(event, taskTitle); + // Count for hydrated threads with a live roster AND for not-yet-hydrated + // threads: the counter is what triggers amortized hydration under + // ordinary traffic after a restart, so a persisted snapshot gets + // refreshed before it can age out of the capped projection. + if ( + activities.length > 0 && + (!hydratedAgentThreads.has(thread.id) || agentsByThread.get(thread.id)?.size) + ) { + activitiesSinceAgentSnapshot.set( + thread.id, + (activitiesSinceAgentSnapshot.get(thread.id) ?? 0) + activities.length, + ); + } yield* Effect.forEach(activities, (activity) => providerCommandId(event, "thread-activity-append").pipe( Effect.flatMap((commandId) => @@ -1774,7 +2255,22 @@ const make = Effect.gen(function* () { ).pipe(Effect.asVoid); }); - const processDomainEvent = (_event: TurnStartRequestedDomainEvent) => Effect.void; + const processDomainEvent = (event: IngestionDomainEvent) => + Effect.sync(() => { + if (event.type !== "thread.reverted") { + return; + } + // A checkpoint revert pruned reverted-turn activities (including agent + // snapshots) from the projection. Drop the cached roster so the next + // agent event rehydrates from post-revert state instead of folding a + // stale map and resurrecting reverted agents. + const threadId = event.payload.threadId; + agentsByThread.delete(threadId); + hydratedAgentThreads.delete(threadId); + activitiesSinceAgentSnapshot.delete(threadId); + agentRosterRevision.delete(threadId); + pendingRosterRedispatch.delete(threadId); + }); const processInput = (input: RuntimeIngestionInput) => input.source === "runtime" ? processRuntimeEvent(input.event) : processDomainEvent(input.event); @@ -1805,7 +2301,7 @@ const make = Effect.gen(function* () { ); yield* Effect.forkScoped( Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) => { - if (event.type !== "thread.turn-start-requested") { + if (event.type !== "thread.turn-start-requested" && event.type !== "thread.reverted") { return Effect.void; } return worker.enqueue({ source: "domain", event }); diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index 2735b69a187..bc6aa5886fe 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -1704,6 +1704,124 @@ describe("ClaudeAdapterLive", () => { "Code reviewer checked the migration edge cases.", ); assert.equal(progressEvent.payload.description, "Running background teammate"); + assert.deepEqual(progressEvent.payload.usage, { + totalTokens: 123, + toolUses: 4, + durationMs: 987, + }); + } + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + + it.effect("carries Claude task linkage, updates, and workflow progress", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const runtimeEvents: Array = []; + const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + Effect.sync(() => runtimeEvents.push(event)), + ).pipe(Effect.forkChild); + + yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); + harness.query.emit({ + type: "system", + subtype: "task_started", + task_id: "workflow-1", + description: "Run checks", + task_type: "local_workflow", + subagent_type: "Workflow", + tool_use_id: "tool-1", + workflow_name: "Verification", + prompt: "Check the repository", + session_id: "session", + uuid: "started", + } as unknown as SDKMessage); + harness.query.emit({ + type: "system", + subtype: "task_progress", + task_id: "workflow-1", + description: "Run checks", + subagent_type: "Workflow", + tool_use_id: "tool-1", + usage: { total_tokens: 12, tool_uses: 2, duration_ms: 30 }, + workflow_progress: [ + { type: "workflow_phase", index: 0, title: "Inspect" }, + { + type: "workflow_agent", + index: 1, + label: "Reviewer", + phaseIndex: 0, + phaseTitle: "Inspect", + agentId: "agent-1", + model: "claude-sonnet", + state: "start", + attempt: 1, + tokens: 7, + toolCalls: 1, + lastToolName: "Read", + }, + ], + session_id: "session", + uuid: "progress", + } as unknown as SDKMessage); + harness.query.emit({ + type: "system", + subtype: "task_updated", + task_id: "workflow-1", + patch: { + status: "paused", + end_time: 1_700_000_000_000, + is_backgrounded: true, + error: "waiting", + }, + session_id: "session", + uuid: "updated", + } as unknown as SDKMessage); + yield* Effect.yieldNow; + yield* Effect.yieldNow; + yield* Effect.yieldNow; + runtimeEventsFiber.interruptUnsafe(); + + const started = runtimeEvents.find((event) => event.type === "task.started"); + assert.equal(started?.type, "task.started"); + if (started?.type === "task.started") { + assert.equal(started.payload.name, "Run checks"); + assert.equal(started.payload.agentType, "Workflow"); + assert.equal(started.payload.toolUseId, "tool-1"); + assert.equal(started.payload.workflowName, "Verification"); + assert.equal(started.payload.prompt, "Check the repository"); + } + const phasePatch = runtimeEvents.find( + (event) => event.type === "task.updated" && event.payload.phases !== undefined, + ); + assert.equal(phasePatch?.type, "task.updated"); + if (phasePatch?.type === "task.updated") { + assert.deepEqual(phasePatch.payload.phases, [{ index: 0, title: "Inspect" }]); + } + const child = runtimeEvents.find( + (event) => event.type === "task.started" && event.payload.taskId === "workflow-1:wf:1", + ); + assert.equal(child?.type, "task.started"); + if (child?.type === "task.started") { + assert.equal(child.payload.parentTaskId, "workflow-1"); + assert.equal(child.payload.timelineBypass, true); + assert.equal(child.payload.taskType, "workflow_agent"); + } + const statusPatch = runtimeEvents.find( + (event) => event.type === "task.updated" && event.payload.status === "paused", + ); + assert.equal(statusPatch?.type, "task.updated"); + if (statusPatch?.type === "task.updated") { + assert.equal(statusPatch.payload.endTime, "2023-11-14T22:13:20.000Z"); + assert.equal(statusPatch.payload.isBackgrounded, true); + assert.equal(statusPatch.payload.errorMessage, "waiting"); } }).pipe( Effect.provideService(Random.Random, makeDeterministicRandomService()), diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index 757d7a00eb2..a40d8e2e9c0 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -704,6 +704,38 @@ function readString(value: unknown): string | undefined { return typeof value === "string" && value.trim().length > 0 ? value.trim() : undefined; } +function readNonNegativeInteger(value: unknown): number | undefined { + return typeof value === "number" && Number.isInteger(value) && value >= 0 ? value : undefined; +} + +function epochMillisToIso(value: number): string { + return DateTime.formatIso(DateTime.makeUnsafe(value)); +} + +function normalizeRuntimeTaskUsage(value: unknown) { + if (value === null || typeof value !== "object" || Array.isArray(value)) { + return undefined; + } + const usage = value as Record; + const totalTokens = readNonNegativeInteger(usage.total_tokens); + if (totalTokens === undefined) { + return undefined; + } + const toolUses = readNonNegativeInteger(usage.tool_uses); + const durationMs = readNonNegativeInteger(usage.duration_ms); + const inputTokens = readNonNegativeInteger(usage.input_tokens); + const outputTokens = readNonNegativeInteger(usage.output_tokens); + const cachedInputTokens = readNonNegativeInteger(usage.cache_read_input_tokens); + return { + totalTokens, + ...(inputTokens !== undefined ? { inputTokens } : {}), + ...(outputTokens !== undefined ? { outputTokens } : {}), + ...(cachedInputTokens !== undefined ? { cachedInputTokens } : {}), + ...(toolUses !== undefined ? { toolUses } : {}), + ...(durationMs !== undefined ? { durationMs } : {}), + }; +} + function readStringArray(value: unknown): Array { return Array.isArray(value) ? value.filter((entry): entry is string => typeof entry === "string" && entry.length > 0) @@ -2388,6 +2420,33 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }, }); + if (!toolResult.isError && tool.toolName === "Workflow" && toolUseResult) { + const taskId = readString(toolUseResult.taskId); + if (taskId) { + const stamp = yield* makeEventStamp(); + yield* offerRuntimeEvent({ + type: "task.updated", + eventId: stamp.eventId, + provider: PROVIDER, + createdAt: stamp.createdAt, + threadId: context.session.threadId, + ...(context.turnState ? { turnId: context.turnState.turnId } : {}), + payload: { + taskId: RuntimeTaskId.make(taskId), + ...(readString(toolUseResult.scriptPath) + ? { scriptPath: readString(toolUseResult.scriptPath) } + : {}), + ...(readString(toolUseResult.runId) + ? { runId: readString(toolUseResult.runId) } + : {}), + ...(readString(toolUseResult.workflowName) + ? { workflowName: readString(toolUseResult.workflowName) } + : {}), + }, + }); + } + } + const streamKind = toolResultStreamKind(tool.itemType); if (streamKind && toolResult.text.length > 0 && context.turnState) { const deltaStamp = yield* makeEventStamp(); @@ -2676,18 +2735,33 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }, }); return; - case "task_started": + case "task_started": { + const record = message as unknown as Record; + const agentType = readString(record.subagent_type); + const toolUseId = readString(record.tool_use_id); + const workflowName = readString(record.workflow_name); + const prompt = readString(record.prompt); yield* offerRuntimeEvent({ ...base, type: "task.started", payload: { taskId: RuntimeTaskId.make(message.task_id), description: message.description, + name: message.description, ...(message.task_type ? { taskType: message.task_type } : {}), + ...(agentType ? { agentType } : {}), + ...(toolUseId ? { toolUseId } : {}), + ...(workflowName ? { workflowName } : {}), + ...(prompt ? { prompt } : {}), }, }); return; - case "task_progress": + } + case "task_progress": { + const record = message as unknown as Record; + const agentType = readString(record.subagent_type); + const toolUseId = readString(record.tool_use_id); + const usage = normalizeRuntimeTaskUsage(record.usage); yield* emitThreadTokenUsage( context, normalizeClaudeTaskProgressTokenUsage(message.usage, context), @@ -2703,17 +2777,165 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( taskId: RuntimeTaskId.make(message.task_id), description: message.description, ...(message.summary ? { summary: message.summary } : {}), - ...(message.usage ? { usage: message.usage } : {}), + ...(usage ? { usage } : {}), ...(message.last_tool_name ? { lastToolName: message.last_tool_name } : {}), + ...(agentType ? { agentType } : {}), + ...(toolUseId ? { toolUseId } : {}), }, }); + const workflowProgress = record.workflow_progress; + if (Array.isArray(workflowProgress)) { + const phases = workflowProgress.flatMap((entry) => { + if (entry === null || typeof entry !== "object" || Array.isArray(entry)) return []; + const item = entry as Record; + if (item.type !== "workflow_phase") return []; + const index = readNonNegativeInteger(item.index); + const title = readString(item.title); + return index !== undefined && title ? [{ index, title }] : []; + }); + if (phases.length > 0) { + const stamp = yield* makeEventStamp(); + yield* offerRuntimeEvent({ + ...base, + eventId: stamp.eventId, + createdAt: stamp.createdAt, + type: "task.updated", + payload: { + taskId: RuntimeTaskId.make(message.task_id), + phases, + timelineBypass: true, + }, + }); + } + for (const entry of workflowProgress) { + if (entry === null || typeof entry !== "object" || Array.isArray(entry)) continue; + const item = entry as Record; + if (item.type !== "workflow_agent") continue; + // `index` is the stable identity: queued/blocked/pre-spawn-error + // events carry no agentId yet, and retries keep the index while + // minting a new agentId per attempt. Keying on agentId would drop + // queued agents and duplicate cards across retries. + const index = readNonNegativeInteger(item.index); + const label = readString(item.label); + if (index === undefined || !label) continue; + const state = readString(item.state); + const childUsage = readNonNegativeInteger(item.tokens); + const childToolUses = readNonNegativeInteger(item.toolCalls); + const childPayload = { + taskId: RuntimeTaskId.make(`${message.task_id}:wf:${index}`), + description: label, + name: label, + taskType: "workflow_agent", + parentTaskId: RuntimeTaskId.make(message.task_id), + timelineBypass: true as const, + ...(readString(item.model) ? { model: readString(item.model) } : {}), + ...(readNonNegativeInteger(item.phaseIndex) !== undefined + ? { phaseIndex: readNonNegativeInteger(item.phaseIndex) } + : {}), + ...(readString(item.phaseTitle) ? { phaseTitle: readString(item.phaseTitle) } : {}), + ...(readNonNegativeInteger(item.attempt) !== undefined + ? { attempt: readNonNegativeInteger(item.attempt) } + : {}), + ...(readString(item.lastToolName) + ? { lastToolName: readString(item.lastToolName) } + : {}), + ...(childUsage !== undefined + ? { + usage: { + totalTokens: childUsage, + ...(childToolUses !== undefined ? { toolUses: childToolUses } : {}), + }, + } + : {}), + }; + const stamp = yield* makeEventStamp(); + if (state === "start") { + yield* offerRuntimeEvent({ + ...base, + eventId: stamp.eventId, + createdAt: stamp.createdAt, + type: "task.started", + payload: childPayload, + }); + // The CLI emits state:"start" both when queueing (queuedAt only) + // and when execution begins (startedAt set). A queued agent must + // not show as running in the panel. + if (readNonNegativeInteger(item.startedAt) === undefined) { + const pendingStamp = yield* makeEventStamp(); + yield* offerRuntimeEvent({ + ...base, + eventId: pendingStamp.eventId, + createdAt: pendingStamp.createdAt, + type: "task.updated", + payload: { + taskId: childPayload.taskId, + status: "pending", + timelineBypass: true, + }, + }); + } + } else if (state === "done" || state === "error") { + yield* offerRuntimeEvent({ + ...base, + eventId: stamp.eventId, + createdAt: stamp.createdAt, + type: "task.completed", + payload: { + ...childPayload, + status: state === "done" ? "completed" : "failed", + ...(readString(item.error) ? { summary: readString(item.error) } : {}), + }, + }); + } else { + yield* offerRuntimeEvent({ + ...base, + eventId: stamp.eventId, + createdAt: stamp.createdAt, + type: "task.progress", + payload: childPayload, + }); + } + } + } return; - // Task state patch (status/backgrounded/end_time). No runtime mapping - // yet — the terminal task_notification reports the outcome — but it - // must not surface as an unknown-subtype warning row. - case "task_updated": + } + case "task_updated": { + const record = message as unknown as Record; + // SDK shape: {task_id, patch: {status?, description?, end_time?, is_backgrounded?, error?}} + const patch = + record.patch !== null && typeof record.patch === "object" && !Array.isArray(record.patch) + ? (record.patch as Record) + : {}; + const status = readString(patch.status); + const patchedDescription = readString(patch.description); + const endTime = + typeof patch.end_time === "number" ? epochMillisToIso(patch.end_time) : undefined; + yield* offerRuntimeEvent({ + ...base, + type: "task.updated", + payload: { + taskId: RuntimeTaskId.make(String(record.task_id)), + ...(status === "pending" || + status === "running" || + status === "completed" || + status === "failed" || + status === "killed" || + status === "paused" + ? { status } + : {}), + ...(endTime ? { endTime } : {}), + ...(patchedDescription ? { name: patchedDescription } : {}), + ...(typeof patch.is_backgrounded === "boolean" + ? { isBackgrounded: patch.is_backgrounded } + : {}), + ...(readString(patch.error) ? { errorMessage: readString(patch.error) } : {}), + }, + }); return; - case "task_notification": + } + case "task_notification": { + const record = message as unknown as Record; + const usage = normalizeRuntimeTaskUsage(record.usage); yield* emitThreadTokenUsage( context, normalizeClaudeTaskProgressTokenUsage(message.usage, context), @@ -2729,10 +2951,17 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( taskId: RuntimeTaskId.make(message.task_id), status: message.status, ...(message.summary ? { summary: message.summary } : {}), - ...(message.usage ? { usage: message.usage } : {}), + ...(usage ? { usage } : {}), + ...(readString(record.output_file) + ? { outputFile: readString(record.output_file) } + : {}), + ...(readString(record.tool_use_id) + ? { toolUseId: readString(record.tool_use_id) } + : {}), }, }); return; + } case "files_persisted": yield* offerRuntimeEvent({ ...base, diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 4ae654a5187..9a5c820b05d 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -555,6 +555,129 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => { }), ); + it.effect("maps collab/agentActivity turn lifecycle to timeline-bypassing task events", () => + Effect.gen(function* () { + const { adapter, runtime } = yield* startLifecycleRuntime(); + const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild); + + yield* runtime.emit({ + id: asEventId("evt-collab-turn-start"), + kind: "notification", + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:00.000Z", + method: "collab/agentActivity", + threadId: asThreadId("thread-1"), + payload: { + agentThreadId: "child-thread-1", + agentPath: "/root/marlow", + method: "turn/started", + params: { threadId: "child-thread-1", turn: { id: "child-turn-1" } }, + }, + }); + const firstEvent = yield* Fiber.join(firstEventFiber); + + NodeAssert.equal(firstEvent._tag, "Some"); + if (firstEvent._tag !== "Some") { + return; + } + NodeAssert.equal(firstEvent.value.type, "task.started"); + if (firstEvent.value.type !== "task.started") { + return; + } + NodeAssert.equal(firstEvent.value.payload.taskId, "child-thread-1"); + NodeAssert.equal(firstEvent.value.payload.name, "marlow"); + NodeAssert.equal(firstEvent.value.payload.timelineBypass, true); + }), + ); + + it.effect("maps collab child turn/completed to an idle task.updated", () => + Effect.gen(function* () { + const { adapter, runtime } = yield* startLifecycleRuntime(); + const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild); + + yield* runtime.emit({ + id: asEventId("evt-collab-turn-complete"), + kind: "notification", + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:01.000Z", + method: "collab/agentActivity", + threadId: asThreadId("thread-1"), + payload: { + agentThreadId: "child-thread-1", + agentPath: "/root/marlow", + method: "turn/completed", + params: { threadId: "child-thread-1", turn: { id: "child-turn-1" } }, + }, + }); + const firstEvent = yield* Fiber.join(firstEventFiber); + + NodeAssert.equal(firstEvent._tag, "Some"); + if (firstEvent._tag !== "Some") { + return; + } + NodeAssert.equal(firstEvent.value.type, "task.updated"); + if (firstEvent.value.type !== "task.updated") { + return; + } + NodeAssert.equal(firstEvent.value.payload.status, "idle"); + NodeAssert.equal(firstEvent.value.payload.timelineBypass, true); + }), + ); + + it.effect("maps collab child token usage to task.progress with normalized usage", () => + Effect.gen(function* () { + const { adapter, runtime } = yield* startLifecycleRuntime(); + const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild); + + yield* runtime.emit({ + id: asEventId("evt-collab-usage"), + kind: "notification", + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:02.000Z", + method: "collab/agentActivity", + threadId: asThreadId("thread-1"), + payload: { + agentThreadId: "child-thread-1", + agentPath: "/root/marlow", + method: "thread/tokenUsage/updated", + params: { + threadId: "child-thread-1", + turnId: "child-turn-1", + tokenUsage: { + total: { + totalTokens: 4200, + inputTokens: 4000, + cachedInputTokens: 1000, + outputTokens: 200, + reasoningOutputTokens: 50, + }, + last: { + totalTokens: 4200, + inputTokens: 4000, + cachedInputTokens: 1000, + outputTokens: 200, + reasoningOutputTokens: 50, + }, + modelContextWindow: 272_000, + }, + }, + }, + }); + const firstEvent = yield* Fiber.join(firstEventFiber); + + NodeAssert.equal(firstEvent._tag, "Some"); + if (firstEvent._tag !== "Some") { + return; + } + NodeAssert.equal(firstEvent.value.type, "task.progress"); + if (firstEvent.value.type !== "task.progress") { + return; + } + const usage = firstEvent.value.payload.usage as { totalTokens?: number } | undefined; + NodeAssert.equal(usage?.totalTokens, 4200); + }), + ); + it.effect("labels MCP lifecycle entries with server and tool names", () => Effect.gen(function* () { const { adapter, runtime } = yield* startLifecycleRuntime(); diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 38a5887cdc3..50a6e17b30a 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -20,6 +20,7 @@ import { type ProviderUserInputAnswers, RuntimeItemId, RuntimeRequestId, + RuntimeTaskId, ProviderApprovalDecision, ThreadId, ProviderSendTurnInput, @@ -494,10 +495,212 @@ function mapItemLifecycle( }; } +/** + * Codex v2 collab child threads have no task lifecycle of their own — the + * session runtime intercepts their notifications and forwards them wrapped as + * `collab/agentActivity`. Translate into the shared `task.*` family so the + * ingestion agent reducer treats Codex children like Claude subagents. All + * synthesized events set `timelineBypass` so they never become timeline rows. + * + * Probe-verified mapping (codex-cli 0.144.1): children emit no + * `thread/started`; their first event is `thread/status/changed`. Nicknames + * arrive via the parent's `subAgentActivity.agentPath` (`/root/`). + * `turn/completed` means idle-and-resumable, not terminal. + */ +function mapCollabAgentActivity( + event: ProviderEvent, + canonicalThreadId: ThreadId, +): ReadonlyArray { + const wrapper = + event.payload !== null && typeof event.payload === "object" + ? (event.payload as { + agentThreadId?: unknown; + agentPath?: unknown; + method?: unknown; + params?: unknown; + }) + : undefined; + if (!wrapper || typeof wrapper.agentThreadId !== "string" || typeof wrapper.method !== "string") { + return []; + } + const agentThreadId = wrapper.agentThreadId; + const agentPath = typeof wrapper.agentPath === "string" ? wrapper.agentPath : undefined; + // "/root/marlow" → "marlow" + const nickname = agentPath?.split("/").filter(Boolean).at(-1); + const inner: ProviderEvent = { + ...event, + method: wrapper.method, + ...(wrapper.params !== undefined ? { payload: wrapper.params } : {}), + }; + const base = { + ...runtimeEventBase(event, canonicalThreadId), + payload: { + taskId: RuntimeTaskId.make(agentThreadId), + description: nickname ?? agentThreadId, + name: nickname ?? agentThreadId, + taskType: "local_agent", + timelineBypass: true, + }, + } as const; + + switch (wrapper.method) { + case "turn/started": + return [{ ...base, type: "task.started", payload: { ...base.payload } }]; + case "turn/completed": { + const payload = readPayload(EffectCodexSchema.V2TurnCompletedNotification, inner.payload); + const turnStatus = payload?.turn.status; + const errorMessage = payload?.turn.error?.message; + if (turnStatus === "failed") { + return [ + { + ...base, + type: "task.completed", + payload: { + taskId: base.payload.taskId, + status: "failed", + ...(errorMessage ? { summary: errorMessage } : {}), + timelineBypass: true, + }, + }, + ]; + } + if (turnStatus === "interrupted") { + return [ + { + ...base, + type: "task.completed", + payload: { taskId: base.payload.taskId, status: "stopped", timelineBypass: true }, + }, + ]; + } + // completed (or unparseable payload): the identity is idle-and-resumable. + return [ + { + ...base, + type: "task.updated", + payload: { taskId: base.payload.taskId, status: "idle", timelineBypass: true }, + }, + ]; + } + case "subAgent/interrupted": + return [ + { + ...base, + type: "task.completed", + payload: { taskId: base.payload.taskId, status: "stopped", timelineBypass: true }, + }, + ]; + case "thread/status/changed": { + const payload = readPayload( + EffectCodexSchema.V2ThreadStatusChangedNotification, + inner.payload, + ); + const status = payload?.status; + if (!status) return []; + if (status.type === "active") { + // Flags (waitingOnApproval/waitingOnUserInput) mean blocked; active + // with no flags means the wait resolved and the agent is running again. + return [ + { + ...base, + type: "task.updated", + payload: { + taskId: base.payload.taskId, + status: status.activeFlags.length > 0 ? "waiting" : "running", + timelineBypass: true, + }, + }, + ]; + } + if (status.type === "systemError") { + return [ + { + ...base, + type: "task.completed", + payload: { + taskId: base.payload.taskId, + status: "failed", + summary: "Codex reported a system error for this agent", + timelineBypass: true, + }, + }, + ]; + } + // notLoaded/idle are covered by turn lifecycle events. + return []; + } + case "thread/closed": + return [ + { + ...base, + type: "task.completed", + payload: { taskId: base.payload.taskId, status: "stopped", timelineBypass: true }, + }, + ]; + case "thread/tokenUsage/updated": { + const payload = readPayload( + EffectCodexSchema.V2ThreadTokenUsageUpdatedNotification, + inner.payload, + ); + // All fields from the cumulative `total` breakdown — mixing the + // cumulative total with `last` (per-turn) breakdowns would make the + // numbers internally inconsistent, and `last` can shrink on follow-ups. + const total = payload?.tokenUsage.total; + if (!total || total.totalTokens <= 0) return []; + return [ + { + ...base, + type: "task.progress", + payload: { + ...base.payload, + usage: { + totalTokens: total.totalTokens, + inputTokens: total.inputTokens, + cachedInputTokens: total.cachedInputTokens, + outputTokens: total.outputTokens, + reasoningOutputTokens: total.reasoningOutputTokens, + }, + }, + }, + ]; + } + case "item/started": + case "item/completed": { + const payload = + readPayload(EffectCodexSchema.V2ItemStartedNotification, inner.payload) ?? + readPayload(EffectCodexSchema.V2ItemCompletedNotification, inner.payload); + const item = payload?.item; + if (!item) return []; + const itemType = toCanonicalItemType(item.type); + const title = itemTitle(itemType, item); + const detail = itemDetail(itemType, item); + const summary = detail ?? title; + if (!summary) return []; + return [ + { + ...base, + type: "task.progress", + payload: { + ...base.payload, + description: summary, + summary, + ...(title ? { lastToolName: title } : {}), + }, + }, + ]; + } + default: + return []; + } +} + function mapToRuntimeEvents( event: ProviderEvent, canonicalThreadId: ThreadId, ): ReadonlyArray { + if (event.kind === "notification" && event.method === "collab/agentActivity") { + return mapCollabAgentActivity(event, canonicalThreadId); + } if (event.kind === "error") { if (!event.message) { return []; diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 5a81e915e34..c66e1321226 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -585,6 +585,41 @@ function readRouteFields(notification: CodexServerNotification): { } } +interface CollabChildAgent { + readonly agentPath: string | undefined; + readonly interrupted: boolean; +} + +/** + * v2 collab children announce themselves only via `subAgentActivity` items on + * the parent stream (the probe shows no child `thread/started`, and v2 + * `collabAgentToolCall` items carry empty receiver lists). Track them so their + * notifications can be folded into agent telemetry instead of leaking into the + * parent timeline as unattributed items. Returns the interruption transition + * when this notification carries one, so the caller can emit a terminal event. + */ +function rememberSubAgentActivity( + childAgents: Map, + notification: CodexServerNotification, +): { readonly agentThreadId: string; readonly agentPath: string } | undefined { + if (notification.method !== "item/started" && notification.method !== "item/completed") { + return undefined; + } + const item = notification.params.item; + if (item.type !== "subAgentActivity") { + return undefined; + } + const interrupted = item.kind === "interrupted"; + const previous = childAgents.get(item.agentThreadId); + childAgents.set(item.agentThreadId, { agentPath: item.agentPath, interrupted }); + return interrupted && !previous?.interrupted + ? { agentThreadId: item.agentThreadId, agentPath: item.agentPath } + : undefined; +} + +/** Synthetic method carrying an intercepted child-thread notification. */ +export const COLLAB_AGENT_ACTIVITY_METHOD = "collab/agentActivity"; + function rememberCollabReceiverTurns( collabReceiverTurns: Map, notification: CodexServerNotification, @@ -708,6 +743,7 @@ export const makeCodexSessionRuntime = ( const approvalCorrelationsRef = yield* Ref.make(new Map()); const pendingUserInputsRef = yield* Ref.make(new Map()); const collabReceiverTurnsRef = yield* Ref.make(new Map()); + const collabChildAgentsRef = yield* Ref.make(new Map()); const closedRef = yield* Ref.make(false); // `~` is not shell-expanded when env vars are set via @@ -829,14 +865,76 @@ export const makeCodexSessionRuntime = ( const payload = notification.params; const route = readRouteFields(notification); const collabReceiverTurns = yield* Ref.get(collabReceiverTurnsRef); - const childParentTurnId = (() => { - const providerConversationId = readNotificationThreadId(notification); - return providerConversationId - ? collabReceiverTurns.get(providerConversationId) - : undefined; - })(); + const collabChildAgents = yield* Ref.get(collabChildAgentsRef); + const notificationThreadId = readNotificationThreadId(notification); + const childParentTurnId = notificationThreadId + ? collabReceiverTurns.get(notificationThreadId) + : undefined; rememberCollabReceiverTurns(collabReceiverTurns, notification, route.turnId); + const interruption = rememberSubAgentActivity(collabChildAgents, notification); + yield* Ref.set(collabChildAgentsRef, collabChildAgents); + if (interruption) { + // Surface the interruption as its own child event so the agent + // tracker can terminalize the child (the child thread itself goes + // silent when interrupted). + yield* emitEvent({ + kind: "notification", + threadId: options.threadId, + method: COLLAB_AGENT_ACTIVITY_METHOD, + payload: { + agentThreadId: interruption.agentThreadId, + agentPath: interruption.agentPath, + method: "subAgent/interrupted", + }, + }); + } + + // v2 collab child threads: their notifications arrive on this + // connection keyed by the child threadId. Divert the whole stream into + // a synthetic collab/agentActivity event for the agent tracker and keep + // it out of the parent timeline. A child's first event can arrive + // BEFORE the parent's subAgentActivity registers it (probe: child + // thread/status/changed lands first), so any thread that is neither + // the session's own provider thread nor a v1 collab receiver is + // treated as an unregistered child. + const providerThreadId = currentProviderThreadId(yield* Ref.get(sessionRef)); + // Re-check v1 receiver registration on every notification: if a + // thread was provisionally classified as a v2 child before the + // parent's collabAgentToolCall arrived, hand it back to the v1 + // suppression path once the receiver registration lands. + const knownChild = + notificationThreadId && !collabReceiverTurns.has(notificationThreadId) + ? collabChildAgents.get(notificationThreadId) + : undefined; + const isForeignThread = + notificationThreadId !== undefined && + providerThreadId !== undefined && + notificationThreadId !== providerThreadId && + !collabReceiverTurns.has(notificationThreadId); + if (notificationThreadId && (knownChild || isForeignThread)) { + if (!knownChild) { + yield* Ref.update(collabChildAgentsRef, (current) => { + const next = new Map(current); + next.set(notificationThreadId, { agentPath: undefined, interrupted: false }); + return next; + }); + } + yield* Ref.set(collabReceiverTurnsRef, collabReceiverTurns); + yield* emitEvent({ + kind: "notification", + threadId: options.threadId, + method: COLLAB_AGENT_ACTIVITY_METHOD, + payload: { + agentThreadId: notificationThreadId, + ...(knownChild?.agentPath ? { agentPath: knownChild.agentPath } : {}), + method: notification.method, + params: payload, + }, + }); + return; + } + if (childParentTurnId && shouldSuppressChildConversationNotification(notification.method)) { yield* Ref.set(collabReceiverTurnsRef, collabReceiverTurns); return; diff --git a/apps/web/src/components/AgentsPanel.tsx b/apps/web/src/components/AgentsPanel.tsx new file mode 100644 index 00000000000..7b6752d40bd --- /dev/null +++ b/apps/web/src/components/AgentsPanel.tsx @@ -0,0 +1,348 @@ +import { memo, useEffect, useRef, useState } from "react"; +import type { ThreadAgentSnapshot } from "@t3tools/contracts"; +import { + deriveAgentPanelState, + formatAgentTokenCount, + isTerminalAgentStatus, + type AgentPanelGroup, + type AgentPanelPhase, +} from "@t3tools/client-runtime/state/thread-agents"; +import { BotIcon, BracesIcon, CheckIcon, ChevronDownIcon, ChevronRightIcon } from "lucide-react"; +import { cn } from "~/lib/utils"; +import { Badge } from "./ui/badge"; +import { ScrollArea } from "./ui/scroll-area"; +import { formatDuration } from "../session-logic"; +import { parseTimestampDate } from "../timestampFormat"; + +interface AgentsPanelProps { + agents: ReadonlyArray; + onOpenScript?: (scriptPath: string) => void; + mode?: "sheet" | "sidebar" | "embedded"; +} + +const STATUS_DOT_CLASS: Record = { + pending: "bg-muted-foreground/40", + running: "bg-sky-500 animate-status-pulse", + waiting: "bg-warning animate-status-pulse", + idle: "bg-sky-500/50", + completed: "bg-success", + failed: "bg-destructive", + stopped: "bg-muted-foreground/60", +}; + +const STATUS_LABEL: Record = { + pending: "Queued", + running: "Running", + waiting: "Waiting", + idle: "Idle · resumable", + completed: "Completed", + failed: "Failed", + stopped: "Stopped", +}; + +function AgentStatusDot({ status }: { status: ThreadAgentSnapshot["status"] }) { + return ( + + ); +} + +/** + * Self-ticking elapsed label (WorkingTimer pattern): writes its own text node + * so per-second updates never cause React commits. Frozen once `endedAt` is + * set or the agent settles. + */ +function AgentElapsed({ agent }: { agent: ThreadAgentSnapshot }) { + const textRef = useRef(null); + const settled = isTerminalAgentStatus(agent.status) || agent.status === "idle"; + // Current-activation start (falls back to firstStartedAt for pre-field + // snapshots) so a resumed agent's timer excludes prior runs and idle gaps. + const startMs = + parseTimestampDate(agent.lastStartedAt ?? agent.firstStartedAt)?.getTime() ?? null; + const endMs = + (agent.endedAt ? parseTimestampDate(agent.endedAt)?.getTime() : null) ?? + (settled ? (parseTimestampDate(agent.lastActivityAt)?.getTime() ?? null) : null); + const initialText = + startMs === null ? null : formatDuration(Math.max(0, (endMs ?? Date.now()) - startMs)); + + useEffect(() => { + if (startMs === null || endMs !== null) return; + const update = () => { + if (textRef.current) { + textRef.current.textContent = formatDuration(Math.max(0, Date.now() - startMs)); + } + }; + update(); + const id = setInterval(update, 1000); + return () => clearInterval(id); + }, [startMs, endMs]); + + if (initialText === null) { + return null; + } + return ( + + {initialText} + + ); +} + +function AgentCard({ agent }: { agent: ThreadAgentSnapshot }) { + const [expanded, setExpanded] = useState(false); + const settled = isTerminalAgentStatus(agent.status); + // Settled cards lead with outcome (error first); live cards with activity. + const activity = + agent.status === "waiting" + ? "Waiting on approval or input" + : settled || agent.status === "idle" + ? (agent.errorMessage ?? + agent.resultSummary ?? + agent.currentActivity ?? + (agent.lastToolName ? `▸ ${agent.lastToolName}` : null)) + : (agent.currentActivity ?? + (agent.lastToolName ? `▸ ${agent.lastToolName}` : null) ?? + agent.resultSummary ?? + agent.errorMessage); + const hasFeed = agent.recentActivity.length > 0; + + return ( + + ); +} + +function PhaseHeader({ phase }: { phase: AgentPanelPhase }) { + // "active" (not just status==="running"): a phase whose agents are all + // pending/waiting is still in progress and must not read "0 running". + const doneCount = phase.agents.filter( + (agent) => agent.status === "idle" || isTerminalAgentStatus(agent.status), + ).length; + const activeCount = phase.agents.length - doneCount; + return ( +
+ + {phase.status === "done" ? "✓ " : ""} + {phase.title} + + {phase.status === "running" ? ( + + {activeCount} active{doneCount > 0 ? ` · ${doneCount} done` : ""} + + ) : phase.status === "pending" ? ( + pending + ) : null} + +
+ ); +} + +function AgentGroup({ + group, + onOpenScript, +}: { + group: AgentPanelGroup; + onOpenScript?: ((scriptPath: string) => void) | undefined; +}) { + const scriptPath = group.workflow?.scriptPath; + return ( +
+
+ {group.workflow ? ( + <> + Workflow · {group.workflow.name} + {scriptPath && onOpenScript ? ( + { + event.stopPropagation(); + onOpenScript(scriptPath); + }} + onKeyDown={(event) => { + if (event.key === "Enter" || event.key === " ") { + event.stopPropagation(); + onOpenScript(scriptPath); + } + }} + className="inline-flex cursor-pointer items-center gap-1 font-mono text-[10px] font-medium tracking-normal text-primary normal-case hover:underline" + > + script + + ) : null} + + ) : ( + Direct spawns + )} + +
+ {/* A failed/errored workflow with no member rows would otherwise render + as a bare header — surface the container itself so its status and + error are visible. */} + {group.workflow && + group.rest.length === 0 && + group.phases.every((phase) => phase.agents.length === 0) && + (group.workflow.status === "failed" || + group.workflow.status === "stopped" || + group.workflow.errorMessage) ? ( +
+ +
+ ) : null} + {group.phases.map((phase) => ( +
+ +
+ {phase.agents.map((agent) => ( + + ))} +
+
+ ))} + {group.rest.length > 0 ? ( +
0 && "pt-2")}> + {group.rest.map((agent) => ( + + ))} +
+ ) : null} +
+ ); +} + +const AgentsPanel = memo(function AgentsPanel({ agents, onOpenScript, mode }: AgentsPanelProps) { + const state = deriveAgentPanelState(agents); + + if (agents.length === 0) { + return ( +
+ +

No agents yet

+

+ When this thread spawns subagents or runs a workflow, they show up here with live status + and token usage. +

+
+ ); + } + + return ( +
+ +
+ {state.groups.map((group, index) => ( + + ))} +
+ +
+ {state.runningCount > 0 ? ( + + + {state.runningCount} running + + ) : null} + {state.waitingCount > 0 ? {state.waitingCount} waiting : null} + {state.settledCount > 0 ? {state.settledCount} settled : null} + + Σ {formatAgentTokenCount(state.totalTokens)} tok + +
+
+ ); +}); + +export default AgentsPanel; diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index c296c717066..aabf86521a1 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -85,6 +85,7 @@ import { hasActionableProposedPlan, isLatestTurnSettled, } from "../session-logic"; +import { deriveLatestAgentSnapshot } from "@t3tools/client-runtime/state/thread-agents"; import { type LegendListRef } from "@legendapp/list/react"; import { getAnchoredTurnMetrics, type TimelineScrollMode } from "./chat/timelineScrollAnchoring"; import { @@ -136,6 +137,8 @@ import { RightPanelTabs } from "./RightPanelTabs"; import { DiffWorkerPoolProvider } from "./DiffWorkerPoolProvider"; import { BranchToolbar } from "./BranchToolbar"; import { resolveShortcutCommand, shortcutLabelForCommand } from "../keybindings"; +import AgentsPanel from "./AgentsPanel"; +import AgentsLiveStrip from "./chat/AgentsLiveStrip"; import PlanSidebar from "./PlanSidebar"; import ThreadTerminalDrawer from "./ThreadTerminalDrawer"; import { ChevronDownIcon, TriangleAlertIcon, WifiOffIcon } from "lucide-react"; @@ -1862,6 +1865,20 @@ function ChatViewContent(props: ChatViewProps) { const phase = derivePhase(activeThread?.session ?? null); const threadActivities = activeThread?.activities ?? EMPTY_ACTIVITIES; const workLogEntries = useMemo(() => deriveWorkLogEntries(threadActivities), [threadActivities]); + const threadAgents = useMemo( + () => deriveLatestAgentSnapshot(threadActivities), + [threadActivities], + ); + // Mirrors AgentsLiveStrip's own visibility rule so the wrapper doesn't + // reserve space (mb-1.5) after all agents settle. + const hasLiveAgents = useMemo( + () => + threadAgents.some( + (agent) => + agent.status === "running" || agent.status === "pending" || agent.status === "waiting", + ), + [threadAgents], + ); const pendingApprovals = useMemo( () => derivePendingApprovals(threadActivities), [threadActivities], @@ -2926,6 +2943,10 @@ function ChatViewContent(props: ChatViewProps) { if (!activeThreadRef || !activeProject) return; useRightPanelStore.getState().open(activeThreadRef, "files"); }, [activeProject, activeThreadRef]); + const addAgentsSurface = useCallback(() => { + if (!activeThreadRef) return; + useRightPanelStore.getState().open(activeThreadRef, "agents"); + }, [activeThreadRef]); const openFileSurface = useCallback( (relativePath: string) => { if (!activeThreadRef || !activeProject) return; @@ -2933,6 +2954,21 @@ function ChatViewContent(props: ChatViewProps) { }, [activeProject, activeThreadRef], ); + // Workflow scripts live in the provider's session directory, outside the + // workspace root the file surface can read. Open in-workspace paths in the + // file surface; copy the absolute path otherwise. + const openAgentScript = useCallback( + (scriptPath: string) => { + if (activeWorkspaceRoot && scriptPath.startsWith(`${activeWorkspaceRoot}/`)) { + openFileSurface(scriptPath.slice(activeWorkspaceRoot.length + 1)); + return; + } + void navigator.clipboard.writeText(scriptPath).then(() => { + toastManager.add({ title: "Script path copied", description: scriptPath }); + }); + }, + [activeWorkspaceRoot, openFileSurface], + ); const togglePreviewPanel = useCallback(() => { if (!activeThreadRef || !isPreviewSupportedInRuntime()) return; if (previewPanelOpen) { @@ -5150,6 +5186,8 @@ function ChatViewContent(props: ChatViewProps) { initialGitScope={initialDiffPanelGitScope} /> + ) : activeRightPanelSurface?.kind === "agents" ? ( + ) : activeRightPanelSurface?.kind === "plan" ? ( )} + {hasLiveAgents ? ( +
+ +
+ ) : null}
void; onAddDiff: () => void; onAddFiles: () => void; + onAddAgents: () => void; browserAvailable: boolean; diffAvailable: boolean; filesAvailable: boolean; @@ -91,6 +92,7 @@ function RightPanelEmptyState(props: { onAddTerminal: () => void; onAddDiff: () => void; onAddFiles: () => void; + onAddAgents: () => void; browserAvailable: boolean; diffAvailable: boolean; filesAvailable: boolean; @@ -128,6 +130,14 @@ function RightPanelEmptyState(props: { disabledReason: SURFACE_DISABLED_REASONS.diff, onClick: props.onAddDiff, }, + { + label: "Agents", + description: "Watch subagents and workflows run.", + icon: Bot, + available: true, + disabledReason: null, + onClick: props.onAddAgents, + }, ] as const; return ( @@ -205,6 +215,8 @@ function surfaceTitle( ); case "plan": return "Plan"; + case "agents": + return "Agents"; case "preview": { const snapshot = surface.resourceId ? sessions[surface.resourceId] : null; if (!snapshot || snapshot.navStatus._tag === "Idle") return "Browser"; @@ -266,6 +278,8 @@ function SurfaceIcon({ return ; case "plan": return ; + case "agents": + return ; } } @@ -470,6 +484,10 @@ export function RightPanelTabs(props: RightPanelTabsProps) { Diff + + + Agents + ) : null} @@ -484,6 +502,7 @@ export function RightPanelTabs(props: RightPanelTabsProps) { onAddTerminal={props.onAddTerminal} onAddDiff={props.onAddDiff} onAddFiles={props.onAddFiles} + onAddAgents={props.onAddAgents} browserAvailable={props.browserAvailable} diffAvailable={props.diffAvailable} filesAvailable={props.filesAvailable} diff --git a/apps/web/src/components/chat/AgentsLiveStrip.tsx b/apps/web/src/components/chat/AgentsLiveStrip.tsx new file mode 100644 index 00000000000..81aee88627f --- /dev/null +++ b/apps/web/src/components/chat/AgentsLiveStrip.tsx @@ -0,0 +1,70 @@ +import { memo } from "react"; +import type { ThreadAgentSnapshot } from "@t3tools/contracts"; +import { + deriveAgentPanelState, + formatAgentTokenCount, +} from "@t3tools/client-runtime/state/thread-agents"; +import { BotIcon, ChevronRightIcon } from "lucide-react"; +import { cn } from "~/lib/utils"; + +/** + * Collapsed one-line agent roster shown near the composer while agents are + * live. Clicking opens the Agents panel — this strip is awareness only. + */ +const AgentsLiveStrip = memo(function AgentsLiveStrip({ + agents, + onOpen, +}: { + agents: ReadonlyArray; + onOpen: () => void; +}) { + const state = deriveAgentPanelState(agents); + const liveCount = state.runningCount + state.waitingCount; + if (liveCount === 0) { + // Parent wrappers must not reserve space when nothing renders (the + // caller keys visibility off hasLiveAgents with the same derivation). + return null; + } + + const runningPhase = state.groups + .flatMap((group) => group.phases) + .find((phase) => phase.status === "running"); + + return ( + + ); +}); + +export default AgentsLiveStrip; diff --git a/apps/web/src/rightPanelStore.ts b/apps/web/src/rightPanelStore.ts index 70d163306cc..cccb7238ca8 100644 --- a/apps/web/src/rightPanelStore.ts +++ b/apps/web/src/rightPanelStore.ts @@ -14,7 +14,15 @@ import { createJSONStorage, persist } from "zustand/middleware"; import { resolveStorage } from "./lib/storage"; -export const RIGHT_PANEL_KINDS = ["plan", "diff", "files", "file", "preview", "terminal"] as const; +export const RIGHT_PANEL_KINDS = [ + "plan", + "diff", + "files", + "file", + "preview", + "terminal", + "agents", +] as const; export type RightPanelKind = (typeof RIGHT_PANEL_KINDS)[number]; export type RightPanelSurface = @@ -37,10 +45,11 @@ export type RightPanelSurface = revealLine: number | null; revealRequestId: number; } - | { id: "plan"; kind: "plan" }; + | { id: "plan"; kind: "plan" } + | { id: "agents"; kind: "agents" }; const RIGHT_PANEL_STORAGE_KEY = "t3code:right-panel-state:v2"; -const RIGHT_PANEL_STORAGE_VERSION = 7; +const RIGHT_PANEL_STORAGE_VERSION = 8; export interface ThreadRightPanelState { isOpen: boolean; @@ -92,6 +101,8 @@ const singletonSurface = ( return { id: "files", kind }; case "plan": return { id: "plan", kind }; + case "agents": + return { id: "agents", kind }; } }; diff --git a/apps/web/src/session-logic.ts b/apps/web/src/session-logic.ts index 5d5051f748e..8ee996c84db 100644 --- a/apps/web/src/session-logic.ts +++ b/apps/web/src/session-logic.ts @@ -633,6 +633,7 @@ export function deriveWorkLogEntries( if (activity.kind === "tool.started") continue; if (activity.kind === "task.started") continue; if (activity.kind === "context-window.updated") continue; + if (activity.kind === "agent.snapshot") continue; if (activity.summary === "Checkpoint captured") continue; if (isPlanBoundaryToolActivity(activity)) continue; entries.push(toDerivedWorkLogEntry(activity)); diff --git a/apps/web/src/timestampFormat.ts b/apps/web/src/timestampFormat.ts index 31cc41798eb..cce5b141c63 100644 --- a/apps/web/src/timestampFormat.ts +++ b/apps/web/src/timestampFormat.ts @@ -40,7 +40,7 @@ function getTimestampFormatter( return formatter; } -function parseTimestampDate(isoDate: string): Date | null { +export function parseTimestampDate(isoDate: string): Date | null { const date = new Date(isoDate); return Number.isNaN(date.getTime()) ? null : date; } diff --git a/packages/client-runtime/package.json b/packages/client-runtime/package.json index d9e19889721..07707217c9e 100644 --- a/packages/client-runtime/package.json +++ b/packages/client-runtime/package.json @@ -119,6 +119,10 @@ "types": "./src/state/terminal.ts", "default": "./src/state/terminal.ts" }, + "./state/thread-agents": { + "types": "./src/state/threadAgents.ts", + "default": "./src/state/threadAgents.ts" + }, "./state/threads": { "types": "./src/state/threads.ts", "default": "./src/state/threads.ts" diff --git a/packages/client-runtime/src/state/threadAgents.test.ts b/packages/client-runtime/src/state/threadAgents.test.ts new file mode 100644 index 00000000000..b42f0b939a2 --- /dev/null +++ b/packages/client-runtime/src/state/threadAgents.test.ts @@ -0,0 +1,134 @@ +import { describe, expect, it } from "@effect/vitest"; +import type { OrchestrationThreadActivity, ThreadAgentSnapshot } from "@t3tools/contracts"; +import { + deriveAgentPanelState, + deriveLatestAgentSnapshot, + formatAgentTokenCount, +} from "./threadAgents.ts"; + +// Shape captured from a real server run (integrated verification, 2026-07-20). +const persistedAgent = { + agentId: "a732c41b4b7ba7742", + provider: "claudeAgent", + kind: "subagent", + name: "Say hi", + agentType: "general-purpose", + status: "completed", + usage: { totalTokens: 22_798, toolUses: 0 }, + firstStartedAt: "2026-07-21T03:52:02.264Z", + lastActivityAt: "2026-07-21T03:52:03.936Z", + endedAt: "2026-07-21T03:52:03.936Z", + activationCount: 1, + lastTurnId: "45609775-4b3b-4444-879d-1d9f25a1b954", + resultSummary: "Hi! How can I help you today?", + recentActivity: [], + updatedAt: "2026-07-21T03:52:03.936Z", +}; + +function activity(kind: string, payload: unknown, sequence: number): OrchestrationThreadActivity { + return { + id: `evt-${sequence}`, + tone: "info", + kind, + summary: "agents", + payload, + turnId: null, + sequence, + createdAt: "2026-07-21T03:52:03.936Z", + } as OrchestrationThreadActivity; +} + +describe("deriveLatestAgentSnapshot", () => { + it("decodes a persisted server payload and returns the newest roster", () => { + const agents = deriveLatestAgentSnapshot([ + activity("agent.snapshot", { agents: [{ ...persistedAgent, status: "running" }] }, 1), + activity("context-window.updated", { usedTokens: 10 }, 2), + activity("agent.snapshot", { agents: [persistedAgent] }, 3), + ]); + expect(agents).toHaveLength(1); + expect(agents[0]?.status).toBe("completed"); + expect(agents[0]?.usage?.totalTokens).toBe(22_798); + expect(agents[0]?.resultSummary).toBe("Hi! How can I help you today?"); + }); + + it("treats the newest agents array as authoritative even when its rows fail to decode", () => { + // Falling back to the older roster would resurrect a stale "running" + // snapshot; an undecodable newest roster must yield an empty panel. + const agents = deriveLatestAgentSnapshot([ + activity("agent.snapshot", { agents: [persistedAgent], revision: 1 }, 1), + activity("agent.snapshot", { agents: [{ bogus: true }], revision: 2 }, 2), + ]); + expect(agents).toHaveLength(0); + }); + + it("skips bad rows within a roster while keeping decodable ones", () => { + const agents = deriveLatestAgentSnapshot([ + activity("agent.snapshot", { agents: [persistedAgent, { bogus: true }], revision: 1 }, 1), + ]); + expect(agents).toHaveLength(1); + expect(agents[0]?.agentId).toBe(persistedAgent.agentId); + }); + + it("selects the highest revision regardless of list position", () => { + const older = { ...persistedAgent, status: "running" }; + const agents = deriveLatestAgentSnapshot([ + activity("agent.snapshot", { agents: [persistedAgent], revision: 5 }, 1), + activity("agent.snapshot", { agents: [older], revision: 4 }, 2), + ]); + expect(agents[0]?.status).toBe("completed"); + }); + + it("returns an empty roster when no snapshot activity exists", () => { + expect(deriveLatestAgentSnapshot([activity("task.progress", {}, 1)])).toHaveLength(0); + }); +}); + +describe("deriveAgentPanelState", () => { + const base = deriveLatestAgentSnapshot([ + activity("agent.snapshot", { agents: [persistedAgent] }, 1), + ]); + + it("counts settled agents and sums tokens", () => { + const state = deriveAgentPanelState(base); + expect(state.settledCount).toBe(1); + expect(state.runningCount).toBe(0); + expect(state.totalTokens).toBe(22_798); + expect(state.groups).toHaveLength(1); + expect(state.groups[0]?.workflow).toBeNull(); + }); + + it("groups workflow members under declared phases with derived status", () => { + const workflow: ThreadAgentSnapshot = { + ...(base[0] as ThreadAgentSnapshot), + agentId: "wf-1", + kind: "workflow", + name: "audit", + status: "running", + phases: [ + { index: 0, title: "Audit" }, + { index: 1, title: "Verify" }, + ], + }; + const member: ThreadAgentSnapshot = { + ...(base[0] as ThreadAgentSnapshot), + agentId: "wa-1", + kind: "workflow_agent", + parentAgentId: "wf-1", + phaseIndex: 0, + status: "completed", + }; + const state = deriveAgentPanelState([workflow, member]); + const group = state.groups[0]; + expect(group?.workflow?.agentId).toBe("wf-1"); + expect(group?.phases[0]?.status).toBe("done"); + expect(group?.phases[1]?.status).toBe("pending"); + }); +}); + +describe("formatAgentTokenCount", () => { + it("formats counts at k/M scale", () => { + expect(formatAgentTokenCount(950)).toBe("950"); + expect(formatAgentTokenCount(22_798)).toBe("22.8k"); + expect(formatAgentTokenCount(1_200_000)).toBe("1.2M"); + }); +}); diff --git a/packages/client-runtime/src/state/threadAgents.ts b/packages/client-runtime/src/state/threadAgents.ts new file mode 100644 index 00000000000..0bdf496bc0f --- /dev/null +++ b/packages/client-runtime/src/state/threadAgents.ts @@ -0,0 +1,197 @@ +/** + * Derivation helpers for the thread agent roster. + * + * The server ships the full per-thread roster latest-wins in the payload of + * `agent.snapshot` activities (see `@t3tools/contracts` ThreadAgentsActivityPayload). + * Mirrors the `context-window.updated` pattern: select the newest roster + * (highest revision), decode tolerantly, skip rows that fail to decode. + */ +import { + THREAD_AGENT_TERMINAL_STATUSES, + THREAD_AGENTS_ACTIVITY_KIND, + ThreadAgentSnapshot, + type OrchestrationThreadActivity, +} from "@t3tools/contracts"; +import * as Schema from "effect/Schema"; + +const decodeAgent = Schema.decodeUnknownOption(ThreadAgentSnapshot); + +interface RosterCandidate { + readonly payload: { readonly agents: ReadonlyArray }; + readonly revision: number; +} + +function peekRoster(payload: unknown): RosterCandidate | undefined { + if (payload === null || typeof payload !== "object") { + return undefined; + } + const record = payload as { agents?: unknown; revision?: unknown }; + if (!Array.isArray(record.agents)) { + return undefined; + } + return { + payload: record as RosterCandidate["payload"], + // Missing revision ranks lowest (-1), mirroring server hydration: a + // revision-less legacy roster never beats a revisioned one, and among + // equals the later list position wins. + revision: + typeof record.revision === "number" && Number.isInteger(record.revision) + ? record.revision + : -1, + }; +} + +export function deriveLatestAgentSnapshot( + activities: ReadonlyArray, +): ReadonlyArray { + // Two passes so only ONE roster is schema-decoded per recompute: rosters + // can dominate the 500-row activity window in busy sessions, and this runs + // inside a useMemo that re-fires on every activities change. Winner = + // highest revision, later list position breaking ties. + let best: RosterCandidate | undefined; + for (const activity of activities) { + if (activity.kind !== THREAD_AGENTS_ACTIVITY_KIND) { + continue; + } + const candidate = peekRoster(activity.payload); + if (candidate && (!best || candidate.revision >= best.revision)) { + best = candidate; + } + } + if (!best) { + return []; + } + // The winning roster is authoritative: rows decode per-element (one bad row + // is skipped, the rest kept), and a fully undecodable roster yields an + // empty panel rather than resurrecting an older snapshot. + const decoded: ThreadAgentSnapshot[] = []; + for (const candidate of best.payload.agents) { + const result = decodeAgent(candidate); + if (result._tag === "Some") { + decoded.push(result.value); + } + } + return decoded; +} + +export function isTerminalAgentStatus(status: ThreadAgentSnapshot["status"]): boolean { + return THREAD_AGENT_TERMINAL_STATUSES.has(status); +} + +export interface AgentPanelGroup { + /** The workflow snapshot this group belongs to, or null for direct spawns. */ + readonly workflow: ThreadAgentSnapshot | null; + /** Phase sections in declared order; agents without a phase land in `rest`. */ + readonly phases: ReadonlyArray; + readonly rest: ReadonlyArray; +} + +export interface AgentPanelPhase { + readonly index: number; + readonly title: string; + readonly status: "pending" | "running" | "done"; + readonly agents: ReadonlyArray; +} + +export interface AgentPanelState { + readonly groups: ReadonlyArray; + readonly runningCount: number; + readonly waitingCount: number; + readonly settledCount: number; + readonly totalTokens: number; +} + +function isSettledAgentStatus(status: ThreadAgentSnapshot["status"]): boolean { + // idle counts as settled for phase/summary purposes: the run finished even + // though the agent identity could be resumed. + return status === "idle" || isTerminalAgentStatus(status); +} + +function phaseStatus(agents: ReadonlyArray): "pending" | "running" | "done" { + if (agents.length === 0) return "pending"; + if (agents.every((agent) => isSettledAgentStatus(agent.status))) return "done"; + return "running"; +} + +export function deriveAgentPanelState(agents: ReadonlyArray): AgentPanelState { + const workflows = agents.filter((agent) => agent.kind === "workflow"); + const byParent = new Map(); + const direct: ThreadAgentSnapshot[] = []; + for (const agent of agents) { + if (agent.kind === "workflow") continue; + if (agent.parentAgentId) { + const list = byParent.get(agent.parentAgentId) ?? []; + list.push(agent); + byParent.set(agent.parentAgentId, list); + } else { + direct.push(agent); + } + } + + const groups: AgentPanelGroup[] = []; + for (const workflow of workflows) { + const members = byParent.get(workflow.agentId) ?? []; + byParent.delete(workflow.agentId); + const declaredPhases = workflow.phases ?? []; + const phases: AgentPanelPhase[] = declaredPhases.map((phase) => { + const phaseAgents = members.filter((agent) => agent.phaseIndex === phase.index); + return { + index: phase.index, + title: phase.title, + status: phaseStatus(phaseAgents), + agents: phaseAgents, + }; + }); + const inDeclaredPhase = new Set( + phases.flatMap((phase) => phase.agents.map((agent) => agent.agentId)), + ); + groups.push({ + workflow, + phases, + rest: members.filter((agent) => !inDeclaredPhase.has(agent.agentId)), + }); + } + // Orphaned parent groups (parent never materialized) fold into direct spawns. + for (const list of byParent.values()) { + direct.push(...list); + } + if (direct.length > 0) { + groups.push({ workflow: null, phases: [], rest: direct }); + } + + // Workflow container rows are grouping chrome, not workers: they are + // excluded from worker counts, and a container's own usage only counts when + // it has no member rows to avoid double-counting the same tokens. + const workflowsWithMembers = new Set( + agents.flatMap((agent) => + agent.kind !== "workflow" && agent.parentAgentId ? [agent.parentAgentId] : [], + ), + ); + let runningCount = 0; + let waitingCount = 0; + let settledCount = 0; + let totalTokens = 0; + for (const agent of agents) { + const isContainer = agent.kind === "workflow"; + if (!isContainer) { + if (agent.status === "running" || agent.status === "pending") runningCount += 1; + else if (agent.status === "waiting") waitingCount += 1; + else settledCount += 1; // idle + terminal + } + if (!isContainer || !workflowsWithMembers.has(agent.agentId)) { + totalTokens += agent.usage?.totalTokens ?? 0; + } + } + + return { groups, runningCount, waitingCount, settledCount, totalTokens }; +} + +export function formatAgentTokenCount(totalTokens: number): string { + if (totalTokens >= 1_000_000) { + return `${(totalTokens / 1_000_000).toFixed(1).replace(/\.0$/, "")}M`; + } + if (totalTokens >= 1_000) { + return `${(totalTokens / 1_000).toFixed(1).replace(/\.0$/, "")}k`; + } + return `${totalTokens}`; +} diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 43270efdec7..ebf6ffbf00c 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -18,6 +18,7 @@ export * from "./git.ts"; export * from "./vcs.ts"; export * from "./sourceControl.ts"; export * from "./orchestration.ts"; +export * from "./threadAgents.ts"; export * from "./editor.ts"; export * from "./project.ts"; export * from "./filesystem.ts"; diff --git a/packages/contracts/src/providerRuntime.ts b/packages/contracts/src/providerRuntime.ts index eb2563eff00..99ca1050f43 100644 --- a/packages/contracts/src/providerRuntime.ts +++ b/packages/contracts/src/providerRuntime.ts @@ -176,6 +176,7 @@ const ProviderRuntimeEventType = Schema.Literals([ "user-input.resolved", "task.started", "task.progress", + "task.updated", "task.completed", "hook.started", "hook.progress", @@ -226,6 +227,7 @@ const UserInputRequestedType = Schema.Literal("user-input.requested"); const UserInputResolvedType = Schema.Literal("user-input.resolved"); const TaskStartedType = Schema.Literal("task.started"); const TaskProgressType = Schema.Literal("task.progress"); +const TaskUpdatedType = Schema.Literal("task.updated"); const TaskCompletedType = Schema.Literal("task.completed"); const HookStartedType = Schema.Literal("hook.started"); const HookProgressType = Schema.Literal("hook.progress"); @@ -459,10 +461,56 @@ const UserInputResolvedPayload = Schema.Struct({ }); export type UserInputResolvedPayload = typeof UserInputResolvedPayload.Type; +/** + * Typed task usage rollup. `usage` on task events was historically + * `Schema.Unknown` (raw Claude SDK shape); emitters now normalize to this. + */ +export const RuntimeTaskUsage = Schema.Struct({ + totalTokens: NonNegativeInt, + inputTokens: Schema.optional(NonNegativeInt), + cachedInputTokens: Schema.optional(NonNegativeInt), + outputTokens: Schema.optional(NonNegativeInt), + reasoningOutputTokens: Schema.optional(NonNegativeInt), + toolUses: Schema.optional(NonNegativeInt), + durationMs: Schema.optional(NonNegativeInt), +}); +export type RuntimeTaskUsage = typeof RuntimeTaskUsage.Type; + +/** + * Agent-linkage fields shared by the task lifecycle payloads. Optional and + * additive: pre-existing emitters that only set taskId/description remain + * valid. + * + * `timelineBypass: true` marks events synthesized purely to feed the agent + * reducer (e.g. Codex child-thread activity): ingestion must skip the + * activity-timeline projection for them, otherwise child fleets would spray + * a row per tool call into the transcript. + */ +const TaskAgentLinkage = { + name: Schema.optional(TrimmedNonEmptyStringSchema), + agentType: Schema.optional(TrimmedNonEmptyStringSchema), + model: Schema.optional(TrimmedNonEmptyStringSchema), + toolUseId: Schema.optional(TrimmedNonEmptyStringSchema), + parentTaskId: Schema.optional(RuntimeTaskId), + workflowName: Schema.optional(TrimmedNonEmptyStringSchema), + phaseIndex: Schema.optional(NonNegativeInt), + phaseTitle: Schema.optional(TrimmedNonEmptyStringSchema), + phases: Schema.optional( + Schema.Array(Schema.Struct({ index: NonNegativeInt, title: TrimmedNonEmptyStringSchema })), + ), + attempt: Schema.optional(NonNegativeInt), + scriptPath: Schema.optional(TrimmedNonEmptyStringSchema), + runId: Schema.optional(TrimmedNonEmptyStringSchema), + outputFile: Schema.optional(TrimmedNonEmptyStringSchema), + timelineBypass: Schema.optional(Schema.Boolean), +} as const; + const TaskStartedPayload = Schema.Struct({ taskId: RuntimeTaskId, description: Schema.optional(TrimmedNonEmptyStringSchema), taskType: Schema.optional(TrimmedNonEmptyStringSchema), + prompt: Schema.optional(TrimmedNonEmptyStringSchema), + ...TaskAgentLinkage, }); export type TaskStartedPayload = typeof TaskStartedPayload.Type; @@ -470,16 +518,45 @@ const TaskProgressPayload = Schema.Struct({ taskId: RuntimeTaskId, description: TrimmedNonEmptyStringSchema, summary: Schema.optional(TrimmedNonEmptyStringSchema), - usage: Schema.optional(Schema.Unknown), + usage: Schema.optional(RuntimeTaskUsage), lastToolName: Schema.optional(TrimmedNonEmptyStringSchema), + ...TaskAgentLinkage, }); export type TaskProgressPayload = typeof TaskProgressPayload.Type; +/** + * Non-terminal state patch (Claude `task_updated`): running/paused flips, + * backgrounding, error text, end time. + */ +const TaskUpdatedPayload = Schema.Struct({ + taskId: RuntimeTaskId, + // "idle" = finished a run but resumable (Codex agents between activations); + // "waiting" = blocked on approval/user input. + status: Schema.optional( + Schema.Literals([ + "pending", + "running", + "completed", + "failed", + "killed", + "paused", + "idle", + "waiting", + ]), + ), + endTime: Schema.optional(IsoDateTime), + isBackgrounded: Schema.optional(Schema.Boolean), + errorMessage: Schema.optional(TrimmedNonEmptyStringSchema), + ...TaskAgentLinkage, +}); +export type TaskUpdatedPayload = typeof TaskUpdatedPayload.Type; + const TaskCompletedPayload = Schema.Struct({ taskId: RuntimeTaskId, status: Schema.Literals(["completed", "failed", "stopped"]), summary: Schema.optional(TrimmedNonEmptyStringSchema), - usage: Schema.optional(Schema.Unknown), + usage: Schema.optional(RuntimeTaskUsage), + ...TaskAgentLinkage, }); export type TaskCompletedPayload = typeof TaskCompletedPayload.Type; @@ -835,6 +912,13 @@ const ProviderRuntimeTaskProgressEvent = Schema.Struct({ }); export type ProviderRuntimeTaskProgressEvent = typeof ProviderRuntimeTaskProgressEvent.Type; +const ProviderRuntimeTaskUpdatedEvent = Schema.Struct({ + ...ProviderRuntimeEventBase.fields, + type: TaskUpdatedType, + payload: TaskUpdatedPayload, +}); +export type ProviderRuntimeTaskUpdatedEvent = typeof ProviderRuntimeTaskUpdatedEvent.Type; + const ProviderRuntimeTaskCompletedEvent = Schema.Struct({ ...ProviderRuntimeEventBase.fields, type: TaskCompletedType, @@ -995,6 +1079,7 @@ export const ProviderRuntimeEventV2 = Schema.Union([ ProviderRuntimeUserInputResolvedEvent, ProviderRuntimeTaskStartedEvent, ProviderRuntimeTaskProgressEvent, + ProviderRuntimeTaskUpdatedEvent, ProviderRuntimeTaskCompletedEvent, ProviderRuntimeHookStartedEvent, ProviderRuntimeHookProgressEvent, diff --git a/packages/contracts/src/threadAgents.ts b/packages/contracts/src/threadAgents.ts new file mode 100644 index 00000000000..02d5021ac0d --- /dev/null +++ b/packages/contracts/src/threadAgents.ts @@ -0,0 +1,145 @@ +import * as Schema from "effect/Schema"; +import { + ApprovalRequestId, + IsoDateTime, + NonNegativeInt, + TrimmedNonEmptyString, + TurnId, +} from "./baseSchemas.ts"; +import { ProviderDriverKind } from "./providerInstance.ts"; + +/** + * `ThreadAgentSnapshot` — one delegated worker (subagent, workflow, workflow + * agent, background shell, monitor) observed on a thread. + * + * The snapshot models a stable *identity* with rollups across its activations: + * a Codex collab agent is a durable, re-activatable child thread (follow-ups + * re-run it), while a Claude task is a one-shot execution. `status: "idle"` + * captures the Codex "finished a run but resumable" state, which is distinct + * from the terminal `completed`. + * + * Transport: the full per-thread roster is carried latest-wins in the payload + * of an `agent.snapshot` thread activity (see `ThreadAgentsActivityPayload`). + * There is no dedicated wire event — the activity channel's open `kind` string + * and unknown payload are the compatibility surface, mirroring how + * `context-window.updated` ships token snapshots today. + */ +export const ThreadAgentStatus = Schema.Literals([ + // Not yet running (workflow agent queued behind a phase, Codex pendingInit). + "pending", + "running", + // Blocked on an approval or user-input request. + "waiting", + // Finished a run but resumable (Codex agents between activations; Claude paused). + "idle", + "completed", + "failed", + "stopped", +]); +export type ThreadAgentStatus = typeof ThreadAgentStatus.Type; + +export const THREAD_AGENT_TERMINAL_STATUSES: ReadonlySet = new Set([ + "completed", + "failed", + "stopped", +]); + +export const ThreadAgentKind = Schema.Literals([ + "subagent", + "workflow", + "workflow_agent", + "shell", + "monitor", + "other", +]); +export type ThreadAgentKind = typeof ThreadAgentKind.Type; + +/** Cumulative usage across all of an agent's activations. */ +export const ThreadAgentUsage = Schema.Struct({ + totalTokens: NonNegativeInt, + inputTokens: Schema.optional(NonNegativeInt), + cachedInputTokens: Schema.optional(NonNegativeInt), + outputTokens: Schema.optional(NonNegativeInt), + reasoningOutputTokens: Schema.optional(NonNegativeInt), + toolUses: Schema.optional(NonNegativeInt), +}); +export type ThreadAgentUsage = typeof ThreadAgentUsage.Type; + +export const ThreadAgentActivityEntry = Schema.Struct({ + at: IsoDateTime, + summary: TrimmedNonEmptyString, +}); +export type ThreadAgentActivityEntry = typeof ThreadAgentActivityEntry.Type; + +/** Ordered workflow phase, present only on `kind: "workflow"` snapshots. */ +export const ThreadAgentWorkflowPhase = Schema.Struct({ + index: NonNegativeInt, + title: TrimmedNonEmptyString, +}); +export type ThreadAgentWorkflowPhase = typeof ThreadAgentWorkflowPhase.Type; + +export const THREAD_AGENT_RECENT_ACTIVITY_LIMIT = 6; + +export const ThreadAgentSnapshot = Schema.Struct({ + // Stable identity: Claude task_id, Codex child thread id. Provider + // discriminates the id space so cross-provider collisions are impossible. + agentId: TrimmedNonEmptyString, + provider: ProviderDriverKind, + kind: ThreadAgentKind, + name: TrimmedNonEmptyString, + // Provider agent type: Claude subagent_type ("Explore"), Codex agent_role + // ("explorer"). + agentType: Schema.optional(TrimmedNonEmptyString), + model: Schema.optional(TrimmedNonEmptyString), + status: ThreadAgentStatus, + currentActivity: Schema.optional(TrimmedNonEmptyString), + lastToolName: Schema.optional(TrimmedNonEmptyString), + usage: Schema.optional(ThreadAgentUsage), + firstStartedAt: IsoDateTime, + // Start of the CURRENT activation. Live elapsed timers derive from this so + // idle gaps between resumptions don't inflate the display; absent on + // pre-field snapshots (clients fall back to firstStartedAt). + lastStartedAt: Schema.optional(IsoDateTime), + lastActivityAt: IsoDateTime, + // Cleared when a re-activation (Codex follow-up) starts a new run. + endedAt: Schema.optional(IsoDateTime), + // Codex follow-ups and Claude workflow retry attempts. + activationCount: NonNegativeInt, + spawnTurnId: Schema.optional(TurnId), + lastTurnId: Schema.optional(TurnId), + parentAgentId: Schema.optional(TrimmedNonEmptyString), + // Workflow phase membership. Index is authoritative; title is display-only + // (titles can repeat across phases). + phaseIndex: Schema.optional(NonNegativeInt), + phaseTitle: Schema.optional(TrimmedNonEmptyString), + // kind === "workflow" only. + phases: Schema.optional(Schema.Array(ThreadAgentWorkflowPhase)), + scriptPath: Schema.optional(TrimmedNonEmptyString), + runId: Schema.optional(TrimmedNonEmptyString), + // Reserved for the approval deep-link milestone; no emitter populates it yet. + approvalRequestId: Schema.optional(ApprovalRequestId), + // Claude transcript/output path; reserved for the transcript drill-in milestone. + outputFile: Schema.optional(TrimmedNonEmptyString), + resultSummary: Schema.optional(TrimmedNonEmptyString), + errorMessage: Schema.optional(TrimmedNonEmptyString), + recentActivity: Schema.Array(ThreadAgentActivityEntry), + // Staleness watermark: how fresh this record is, independent of transport + // state. Clients compare against their sync watermark. + updatedAt: IsoDateTime, +}); +export type ThreadAgentSnapshot = typeof ThreadAgentSnapshot.Type; + +export const THREAD_AGENTS_ACTIVITY_KIND = "agent.snapshot"; + +/** + * Payload of an `agent.snapshot` thread activity: the complete latest-wins + * roster for the thread. Decoded tolerantly on the client — a row that fails + * to decode is ignored, never fatal. + */ +export const ThreadAgentsActivityPayload = Schema.Struct({ + agents: Schema.Array(ThreadAgentSnapshot), + // Monotonic per-thread revision: latest-wins selection uses the highest + // revision (not list position), making same-timestamp appends deterministic. + revision: Schema.optional(NonNegativeInt), +}); +export type ThreadAgentsActivityPayload = typeof ThreadAgentsActivityPayload.Type;