/** * Native-provider subagent observability: a tolerant fold over persisted * task.* / tool.* thread activities into orchestration-v2-shaped subagent * state, plus the source-neutral panel model every client renders. * * This module is deliberately legacy-bridge code. When orchestration-v2's * subagent projection is available for a thread, deriveAgentPanelModel * prefers it (see the v2Projection parameter) and the fold is skipped; when * the v1 orchestrator is retired this file is deleted. Field names and * transition semantics copy the v2 stack (#4779) exactly so that swap is * mechanical. * * Invariants encoded here trace to shipped bugs in the prior PRs (#4220, * #3650, #4662): reusable identity vs one-shot activations, idle as a real * nonterminal state, provider-specific usage merges, first-write terminal * timestamps, reactivation clearing terminal detail, and order-robust * folding (completion can create an agent; a late start only fills * metadata). */import * as DateTime from "effect/DateTime";import type { OrchestrationThreadActivity, OrchestrationV2Subagent } from "@t3tools/contracts"; export type RuntimeSubagentStatus =  | "pending"  | "running"  | "waiting"  | "idle"  | "completed"  | "failed"  | "cancelled"  | "interrupted"; export interface SubagentUsage {  readonly totalTokens: number;  readonly inputTokens?: number;  readonly cachedInputTokens?: number;  readonly outputTokens?: number;  readonly reasoningOutputTokens?: number;  readonly toolUses?: number;  readonly durationMs?: number;} export interface SubagentActivityEntry {  readonly at: string;  readonly summary: string;} export interface SubagentWorkflowPhase {  readonly index: number;  readonly title: string;} export interface SubagentRunHandles {  readonly runId?: string;  readonly scriptPath?: string;  readonly transcriptDir?: string;  readonly sessionUrl?: string;} export interface RuntimeSubagent {  readonly id: string;  readonly kind: "subagent" | "subagent_batch" | "workflow" | "workflow_agent";  readonly title: string;  readonly role: string | null;  readonly model: string | null;  readonly effort: string | null;  readonly status: RuntimeSubagentStatus;  readonly activationCount: number;  readonly usage: SubagentUsage | null;  readonly progress: string | null;  readonly lastToolName: string | null;  readonly result: string | null;  readonly error: string | null;  readonly outputFile: string | null;  readonly parentAgentId: string | null;  readonly agentIndex: number | null;  readonly phaseIndex: number | null;  readonly phaseTitle: string | null;  readonly attempt: number | null;  readonly workflowName: string | null;  readonly phases: ReadonlyArray<SubagentWorkflowPhase>;  readonly runHandles: SubagentRunHandles | null;  readonly recentActivity: ReadonlyArray<SubagentActivityEntry>;  /** First retained observation, used as the roster's stable display order. */  readonly firstSeenAt: string;  readonly startedAt: string | null;  readonly completedAt: string | null;  readonly updatedAt: string;} const TERMINAL_STATUSES: ReadonlySet<RuntimeSubagentStatus> = new Set([  "completed",  "failed",  "cancelled",  "interrupted",]); export function isTerminalSubagentStatus(status: RuntimeSubagentStatus): boolean {  return TERMINAL_STATUSES.has(status);} /** Active = the user may still need to care while it runs. Idle is settled-ish * but resumable; waiting counts as active because it needs the user. */export function isActiveSubagentStatus(status: RuntimeSubagentStatus): boolean {  return status === "pending" || status === "running" || status === "waiting";} const RECENT_ACTIVITY_LIMIT = 6;const SUMMARY_CHAR_LIMIT = 180;const ROSTER_LIMIT = 100; /** * True when this activity's payload does NOT belong on the Agents surface. * Classification happens exactly once, server-side at ingestion * (classifyTaskAgentKind → the persisted agentKind stamp); the client only * reads it. Rows without a stamp — legacy threads, pre-stamp servers — are * background by definition: they render in the ordinary work log, exactly * as they did before this feature existed. */export function isBackgroundTaskActivity(payload: Record<string, unknown>): boolean {  return payload.agentKind !== "agent";} function bounded(value: string): string {  return value.length <= SUMMARY_CHAR_LIMIT ? value : `${value.slice(0, SUMMARY_CHAR_LIMIT - 1)}…`;} /** Appends to the ring buffer, deduping consecutive identical summaries. */function appendActivity(  entries: ReadonlyArray<SubagentActivityEntry>,  at: string,  summary: string,): ReadonlyArray<SubagentActivityEntry> {  const boundedSummary = bounded(summary);  if (entries.length > 0 && entries[entries.length - 1]?.summary === boundedSummary) {    return entries;  }  const next = [...entries, { at, summary: boundedSummary }];  return next.length > RECENT_ACTIVITY_LIMIT ? next.slice(-RECENT_ACTIVITY_LIMIT) : next;} function asString(value: unknown): string | undefined {  return typeof value === "string" && value.trim().length > 0 ? value.trim() : undefined;} function asCount(value: unknown): number | undefined {  return typeof value === "number" && Number.isFinite(value) && value >= 0 ? value : undefined;} function asUsage(value: unknown): SubagentUsage | undefined {  if (typeof value !== "object" || value === null) {    return undefined;  }  const record = value as Record<string, unknown>;  const totalTokens = asCount(record.totalTokens);  if (totalTokens === undefined) {    return undefined;  }  const usage: {    totalTokens: number;    inputTokens?: number;    cachedInputTokens?: number;    outputTokens?: number;    reasoningOutputTokens?: number;    toolUses?: number;    durationMs?: number;  } = { totalTokens };  const inputTokens = asCount(record.inputTokens);  if (inputTokens !== undefined) usage.inputTokens = inputTokens;  const cachedInputTokens = asCount(record.cachedInputTokens);  if (cachedInputTokens !== undefined) usage.cachedInputTokens = cachedInputTokens;  const outputTokens = asCount(record.outputTokens);  if (outputTokens !== undefined) usage.outputTokens = outputTokens;  const reasoningOutputTokens = asCount(record.reasoningOutputTokens);  if (reasoningOutputTokens !== undefined) usage.reasoningOutputTokens = reasoningOutputTokens;  const toolUses = asCount(record.toolUses);  if (toolUses !== undefined) usage.toolUses = toolUses;  const durationMs = asCount(record.durationMs);  if (durationMs !== undefined) usage.durationMs = durationMs;  return usage;} /** * Provider-specific usage merge (#4779 semantics, verbatim): * - max-merge (Codex-style cumulative frames): field-wise maximum, idempotent *   under duplicate or late frames. Cumulative totals never shrink. * - accumulate (Claude-style activation deltas): not needed at this layer — *   Claude's task_progress usage is itself cumulative per task, so the fold *   also max-merges. The distinction matters when v2 sums activations. * Field-wise: a terminal payload carrying only totalTokens must not wipe a * known breakdown. */function mergeUsageMax(  current: SubagentUsage | null,  incoming: SubagentUsage | undefined,): SubagentUsage | null {  if (!incoming) {    return current;  }  if (!current) {    return incoming;  }  const pick = (a: number | undefined, b: number | undefined): number | undefined =>    a === undefined ? b : b === undefined ? a : Math.max(a, b);  const merged: {    totalTokens: number;    inputTokens?: number;    cachedInputTokens?: number;    outputTokens?: number;    reasoningOutputTokens?: number;    toolUses?: number;    durationMs?: number;  } = { totalTokens: Math.max(current.totalTokens, incoming.totalTokens) };  const inputTokens = pick(current.inputTokens, incoming.inputTokens);  if (inputTokens !== undefined) merged.inputTokens = inputTokens;  const cachedInputTokens = pick(current.cachedInputTokens, incoming.cachedInputTokens);  if (cachedInputTokens !== undefined) merged.cachedInputTokens = cachedInputTokens;  const outputTokens = pick(current.outputTokens, incoming.outputTokens);  if (outputTokens !== undefined) merged.outputTokens = outputTokens;  const reasoningOutputTokens = pick(current.reasoningOutputTokens, incoming.reasoningOutputTokens);  if (reasoningOutputTokens !== undefined) merged.reasoningOutputTokens = reasoningOutputTokens;  const toolUses = pick(current.toolUses, incoming.toolUses);  if (toolUses !== undefined) merged.toolUses = toolUses;  const durationMs = pick(current.durationMs, incoming.durationMs);  if (durationMs !== undefined) merged.durationMs = durationMs;  return merged;} interface MutableAgent {  id: string;  kind: RuntimeSubagent["kind"];  title: string;  role: string | null;  model: string | null;  effort: string | null;  status: RuntimeSubagentStatus;  activationCount: number;  usage: SubagentUsage | null;  progress: string | null;  lastToolName: string | null;  result: string | null;  error: string | null;  outputFile: string | null;  parentAgentId: string | null;  agentIndex: number | null;  phaseIndex: number | null;  phaseTitle: string | null;  attempt: number | null;  workflowName: string | null;  phases: ReadonlyArray<SubagentWorkflowPhase>;  runHandles: SubagentRunHandles | null;  recentActivity: ReadonlyArray<SubagentActivityEntry>;  firstSeenAt: string;  startedAt: string | null;  completedAt: string | null;  updatedAt: string;} function kindFromPayload(  payload: Record<string, unknown>,  agentId: string,): RuntimeSubagent["kind"] {  if (payload.taskType === "subagent_batch") {    return "subagent_batch";  }  if (asString(payload.taskType) === "local_workflow") {    return "workflow";  }  if (payload.parentAgentId !== undefined || agentId.includes(":wf:")) {    return "workflow_agent";  }  return "subagent";} /** Completion can create an agent (its start may have aged out of retention). */function getOrCreate(  agents: Map<string, MutableAgent>,  id: string,  payload: Record<string, unknown>,  at: string,): MutableAgent {  const existing = agents.get(id);  if (existing) {    return existing;  }  const created: MutableAgent = {    id,    kind: kindFromPayload(payload, id),    title: asString(payload.title) ?? asString(payload.detail) ?? id,    role: asString(payload.role) ?? null,    model: asString(payload.model) ?? null,    effort: asString(payload.effort) ?? null,    status: "pending",    activationCount: 0,    usage: null,    progress: null,    lastToolName: null,    result: null,    error: null,    outputFile: null,    parentAgentId: asString(payload.parentAgentId) ?? null,    agentIndex: asCount(payload.agentIndex) ?? null,    phaseIndex: asCount(payload.phaseIndex) ?? null,    phaseTitle: asString(payload.phaseTitle) ?? null,    attempt: asCount(payload.attempt) ?? null,    workflowName: asString(payload.workflowName) ?? null,    phases: [],    runHandles: null,    recentActivity: [],    firstSeenAt: at,    startedAt: null,    completedAt: null,    updatedAt: at,  };  agents.set(id, created);  return created;} /** Metadata fill from any payload: never downgrades known values to null. */function fillMetadata(agent: MutableAgent, payload: Record<string, unknown>): void {  if (payload.taskType === "subagent_batch") agent.kind = "subagent_batch";  const title = asString(payload.title);  if (title) agent.title = title;  const role = asString(payload.role);  if (role) agent.role = role;  const model = asString(payload.model);  if (model) agent.model = model;  const effort = asString(payload.effort);  if (effort) agent.effort = effort;  const parentAgentId = asString(payload.parentAgentId);  if (parentAgentId) {    agent.parentAgentId = parentAgentId;    if (agent.kind === "subagent") agent.kind = "workflow_agent";  }  const workflowName = asString(payload.workflowName);  if (workflowName) agent.workflowName = workflowName;  if (asString(payload.taskType) === "local_workflow") agent.kind = "workflow";  const agentIndex = asCount(payload.agentIndex);  if (agentIndex !== undefined) agent.agentIndex = agentIndex;  const phaseIndex = asCount(payload.phaseIndex);  if (phaseIndex !== undefined) agent.phaseIndex = phaseIndex;  const phaseTitle = asString(payload.phaseTitle);  if (phaseTitle) agent.phaseTitle = phaseTitle;  const attempt = asCount(payload.attempt);  if (attempt !== undefined) {    // A new attempt on a workflow slot is a reactivation of the same    // identity: clear the previous attempt's terminal detail so the status    // transition (terminal → running, in applyStatus) reads as a fresh run.    // The activation bump lives ONLY in applyStatus — bumping here too    // counted every retry twice (review finding: two attempts read "run 3").    if (agent.attempt !== null && attempt > agent.attempt) {      agent.result = null;      agent.error = null;      agent.completedAt = null;    }    agent.attempt = attempt;  }  const outputFile = asString(payload.outputFile);  if (outputFile) agent.outputFile = outputFile;  if (Array.isArray(payload.phases)) {    const phases: SubagentWorkflowPhase[] = [];    for (const entry of payload.phases) {      if (typeof entry !== "object" || entry === null) continue;      const record = entry as Record<string, unknown>;      const index = asCount(record.index);      const phaseName = asString(record.title);      if (index !== undefined && phaseName) {        phases.push({ index, title: phaseName });      }    }    if (phases.length > 0) {      agent.phases = phases.slice().sort((a, b) => a.index - b.index);    }  }  if (typeof payload.runHandles === "object" && payload.runHandles !== null) {    const record = payload.runHandles as Record<string, unknown>;    const runHandles: {      runId?: string;      scriptPath?: string;      transcriptDir?: string;      sessionUrl?: string;    } = {};    const runId = asString(record.runId);    if (runId) runHandles.runId = runId;    const scriptPath = asString(record.scriptPath);    if (scriptPath) runHandles.scriptPath = scriptPath;    const transcriptDir = asString(record.transcriptDir);    if (transcriptDir) runHandles.transcriptDir = transcriptDir;    // Defense-in-depth: the adapter already sanitizes, but payloads are not    // schema-validated on the read path (shipped XSS lesson).    const sessionUrl = asString(record.sessionUrl);    if (sessionUrl && /^https?:\/\//i.test(sessionUrl)) runHandles.sessionUrl = sessionUrl;    if (Object.keys(runHandles).length > 0) {      agent.runHandles = { ...agent.runHandles, ...runHandles };    }  }} function applyStatus(agent: MutableAgent, status: RuntimeSubagentStatus, at: string): void {  const wasTerminal = isTerminalSubagentStatus(agent.status);  const isTerminal = isTerminalSubagentStatus(status);  if (wasTerminal && isTerminal) {    // Duplicate terminal events are idempotent: first write wins, timestamps    // don't slide.    return;  }  if ((wasTerminal || agent.status === "idle") && (status === "running" || status === "pending")) {    // Reactivation: same identity, new run. Clear the previous run's terminal    // detail so a live card never shows the prior run's output.    agent.activationCount += 1;    agent.result = null;    agent.error = null;    agent.completedAt = null;    if (status === "running") {      agent.startedAt = at;    }  }  if (status === "running" && agent.startedAt === null) {    agent.startedAt = at;  }  if (isTerminal && agent.completedAt === null) {    agent.completedAt = at;  }  agent.status = status;} // Map, not object literal: payloads aren't schema-validated on the read// path, so a status like "toString" must miss instead of resolving an// inherited Function through the prototype chain.const TASK_COMPLETED_STATUS: ReadonlyMap<string, RuntimeSubagentStatus> = new Map([  ["completed", "completed"],  ["failed", "failed"],  ["stopped", "interrupted"],]); const KNOWN_STATUSES: ReadonlySet<string> = new Set([  "pending",  "running",  "waiting",  "idle",  "completed",  "failed",  "cancelled",  "interrupted",]); function asRuntimeStatus(value: unknown): RuntimeSubagentStatus | undefined {  return typeof value === "string" && KNOWN_STATUSES.has(value)    ? (value as RuntimeSubagentStatus)    : undefined;} /** * Folds a thread's persisted activities into subagent state. Tolerant by * construction: malformed rows are skipped individually; unknown kinds are * ignored. Pure — memoize by activity-list identity at the atom layer. * * sessionLive=false derives interruption: background tasks die with their * provider session, so agents whose terminal rows were lost (server * restart, crash) must not read as running forever (review finding: a dead * session left a panel full of "Working" agents while the sidebar showed * nothing). Idle is preserved — a resumable Codex child stays resumable. */export function foldSubagentActivities(  activities: ReadonlyArray<OrchestrationThreadActivity>,  options?: { readonly sessionLive?: boolean },): ReadonlyArray<RuntimeSubagent> {  const agents = new Map<string, MutableAgent>();   for (const activity of activities) {    if (typeof activity.payload !== "object" || activity.payload === null) {      continue;    }    const payload = activity.payload as Record<string, unknown>;    const at = activity.createdAt;     switch (activity.kind) {      case "task.started": {        const taskId = asString(payload.taskId);        if (!taskId) break;        // Only real agents join the roster. Shells, monitors, and plan-mode        // tasks are background work — they render in the ordinary work log,        // not the Agents surface (a "Run 12s stall" shell is not a subagent).        if (isBackgroundTaskActivity(payload)) break;        const agent = getOrCreate(agents, taskId, payload, at);        fillMetadata(agent, payload);        // Order-robustness: a start row arriving after a terminal state is a        // late/out-of-order delivery and only fills metadata — it must not        // reopen the run. Reactivation comes exclusively from explicit        // status transitions (task.updated / progress status). Guard on the        // status itself, not activationCount: a task first seen via a        // terminal task.updated has zero activations but is still settled        // (review finding: a late start reopened a failed child).        if (agent.activationCount === 0 && !isTerminalSubagentStatus(agent.status)) {          agent.activationCount = 1;          agent.startedAt = agent.startedAt ?? at;          agent.status = "running";        } else if (agent.status === "idle") {          applyStatus(agent, "running", at);        }

Lines 1–500 of 1004. Load more