diff --git a/apps/mobile/src/lib/threadActivity.ts b/apps/mobile/src/lib/threadActivity.ts index dd568e6f045..4b0908c9539 100644 --- a/apps/mobile/src/lib/threadActivity.ts +++ b/apps/mobile/src/lib/threadActivity.ts @@ -80,6 +80,8 @@ interface WorkLogEntry { interface DerivedWorkLogEntry extends WorkLogEntry { activityKind: OrchestrationThreadActivity["kind"]; collapseKey?: string; + /** Grouping key for subagent lifecycle rows (one row per agent). */ + taskId?: string; } type RawThreadFeedEntry = @@ -235,6 +237,57 @@ function resolvePendingUserInputAnswer( return normalizeDraftAnswer(draft?.selectedOptionLabel); } +/** Codex children settle via task.updated (idle/failed/interrupted), never + * task.completed — these rows are mobile's only terminal signal for them. */ +const MOBILE_TERMINAL_UPDATE_STATUSES: ReadonlySet = new Set([ + "idle", + "completed", + "failed", + "cancelled", + "interrupted", +]); + +function isTerminalBypassUpdate(activity: OrchestrationThreadActivity): boolean { + if (activity.kind !== "task.updated") { + return false; + } + const payload = + activity.payload && typeof activity.payload === "object" + ? (activity.payload as Record) + : null; + return ( + payload?.timelineBypass === true && + typeof payload.status === "string" && + MOBILE_TERMINAL_UPDATE_STATUSES.has(payload.status) + ); +} + +/** + * Quiet-timeline guarantee (mirrors web's session-logic): agent-internal + * activity lives in the Agents sheet, not the work log. Terminal rows are + * kept — with no Agents surface on mobile they are the terminal signal + * (a surface that hides rows must keep its own terminal signal). That means + * task.completed (Claude) AND terminal bypassed task.updated (Codex, whose + * children never emit task.completed — review finding). + */ +function isAgentInternalActivity(activity: OrchestrationThreadActivity): boolean { + const payload = + activity.payload && typeof activity.payload === "object" + ? (activity.payload as Record) + : null; + if (!payload) { + return false; + } + if ( + payload.timelineBypass === true && + activity.kind !== "task.completed" && + !isTerminalBypassUpdate(activity) + ) { + return true; + } + return typeof payload.agentId === "string" && payload.agentId.trim().length > 0; +} + function deriveWorkLogEntries( activities: ReadonlyArray, ): DerivedWorkLogEntry[] { @@ -243,9 +296,13 @@ function deriveWorkLogEntries( for (const activity of ordered) { if (activity.kind === "tool.started") continue; if (activity.kind === "task.started") continue; + // Terminal bypassed updates pass: Codex children's only terminal signal. + if (activity.kind === "task.updated" && !isTerminalBypassUpdate(activity)) continue; + if (activity.kind === "tool.progress") continue; if (activity.kind === "context-window.updated") continue; if (activity.summary === "Checkpoint captured") continue; if (isPlanBoundaryToolActivity(activity)) continue; + if (isAgentInternalActivity(activity)) continue; entries.push(toDerivedWorkLogEntry(activity)); } return collapseDerivedWorkLogEntries(entries); @@ -271,7 +328,13 @@ function toDerivedWorkLogEntry(activity: OrchestrationThreadActivity): DerivedWo const commandPreview = extractToolCommand(payload); const changedFiles = extractChangedFiles(payload); const title = extractToolTitle(payload); - const isTaskActivity = activity.kind === "task.progress" || activity.kind === "task.completed"; + // task.updated included: terminal bypassed updates (Codex children's only + // terminal signal) must carry task identity so they collapse per child + // instead of stacking anonymous "Task idle" rows. + const isTaskActivity = + activity.kind === "task.progress" || + activity.kind === "task.completed" || + activity.kind === "task.updated"; const taskSummary = isTaskActivity && typeof payload?.summary === "string" && payload.summary.length > 0 ? payload.summary @@ -284,10 +347,15 @@ function toDerivedWorkLogEntry(activity: OrchestrationThreadActivity): DerivedWo ? payload.detail : null; const taskLabel = taskSummary || taskDetailAsLabel; + const taskId = + isTaskActivity && typeof payload?.taskId === "string" && payload.taskId.length > 0 + ? payload.taskId + : undefined; const entry: DerivedWorkLogEntry = { id: activity.id, createdAt: activity.createdAt, turnId: activity.turnId, + ...(taskId ? { taskId } : {}), label: taskLabel || activity.summary, tone: activity.kind === "task.progress" @@ -352,7 +420,25 @@ function collapseDerivedWorkLogEntries( entries: ReadonlyArray, ): DerivedWorkLogEntry[] { const collapsed: DerivedWorkLogEntry[] = []; + // Subagent rows collapse by identity, not adjacency (quiet-timeline + // guarantee; mirrors web's session-logic). + const taskRowIndex = new Map(); for (const entry of entries) { + const isTaskRow = + entry.taskId !== undefined && + (entry.activityKind === "task.progress" || + entry.activityKind === "task.completed" || + entry.activityKind === "task.updated"); + if (isTaskRow && entry.taskId !== undefined) { + const existingIndex = taskRowIndex.get(entry.taskId); + if (existingIndex !== undefined) { + collapsed[existingIndex] = mergeDerivedWorkLogEntries(collapsed[existingIndex]!, entry); + continue; + } + taskRowIndex.set(entry.taskId, collapsed.length); + collapsed.push(entry); + continue; + } const previous = collapsed.at(-1); if (previous && shouldCollapseToolLifecycleEntries(previous, entry)) { collapsed[collapsed.length - 1] = mergeDerivedWorkLogEntries(previous, entry); diff --git a/apps/server/integration/OrchestrationEngineHarness.integration.ts b/apps/server/integration/OrchestrationEngineHarness.integration.ts index ebc4f984b86..c3f77d677b1 100644 --- a/apps/server/integration/OrchestrationEngineHarness.integration.ts +++ b/apps/server/integration/OrchestrationEngineHarness.integration.ts @@ -50,6 +50,7 @@ import * as RepositoryIdentityResolver from "../src/project/RepositoryIdentityRe import { OrchestrationEngineLive } from "../src/orchestration/Layers/OrchestrationEngine.ts"; import { OrchestrationProjectionPipelineLive } from "../src/orchestration/Layers/ProjectionPipeline.ts"; import { OrchestrationProjectionSnapshotQueryLive } from "../src/orchestration/Layers/ProjectionSnapshotQuery.ts"; +import * as ThreadBackgroundLiveness from "../src/orchestration/ThreadBackgroundLiveness.ts"; import { RuntimeReceiptBusTest } from "../src/orchestration/Layers/RuntimeReceiptBus.ts"; import { OrchestrationReactorLive } from "../src/orchestration/Layers/OrchestrationReactor.ts"; import { ProviderCommandReactorLive } from "../src/orchestration/Layers/ProviderCommandReactor.ts"; @@ -305,7 +306,7 @@ export const makeOrchestrationIntegrationHarness = ( checkpointStoreLayer, providerLayer, RuntimeReceiptBusTest, - ); + ).pipe(Layer.provideMerge(ThreadBackgroundLiveness.layer)); const serverSettingsLayer = ServerSettingsService.layerTest(); const runtimeIngestionLayer = ProviderRuntimeIngestionLive.pipe( Layer.provideMerge(runtimeServicesLayer), diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index fb753b9aa4b..d6b64fec602 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -22,6 +22,7 @@ type WsRpcMethod = RpcGroup.Rpcs["_tag"]; */ export const RPC_REQUIRED_SCOPES = { [ORCHESTRATION_WS_METHODS.dispatchCommand]: AuthOrchestrationOperateScope, + [ORCHESTRATION_WS_METHODS.getWorkflowScript]: AuthOrchestrationReadScope, [ORCHESTRATION_WS_METHODS.getTurnDiff]: AuthOrchestrationReadScope, [ORCHESTRATION_WS_METHODS.getFullThreadDiff]: AuthOrchestrationReadScope, [ORCHESTRATION_WS_METHODS.searchThreads]: AuthOrchestrationReadScope, diff --git a/apps/server/src/orchestration/ActivityPayloadProjection.test.ts b/apps/server/src/orchestration/ActivityPayloadProjection.test.ts new file mode 100644 index 00000000000..7ea1e3ea0ed --- /dev/null +++ b/apps/server/src/orchestration/ActivityPayloadProjection.test.ts @@ -0,0 +1,63 @@ +import { describe, expect, it } from "vite-plus/test"; +import type { OrchestrationThreadActivity } from "@t3tools/contracts"; +import { projectActivityPayload } from "./ActivityPayloadProjection.ts"; + +function activity(payload: Record): OrchestrationThreadActivity { + return { + id: "activity-1", + tone: "tool", + kind: "tool.completed", + summary: "Tool", + payload, + turnId: null, + createdAt: "2026-08-01T10:00:00.000Z", + } as unknown as OrchestrationThreadActivity; +} + +/** + * Wire-survival regression: the slimming pass rewrites payload.data but must + * never strip the top-level per-agent fields the subagent fold depends on. + * If slimming ever moves to an allowlist over the whole payload, these + * assertions are the tripwire. + */ +describe("projectActivityPayload agent-field survival", () => { + it("preserves tool attribution (agentId/parentToolUseId) through data slimming", () => { + const projected = projectActivityPayload( + activity({ + itemType: "command_execution", + agentId: "task-123", + parentToolUseId: "toolu_abc", + data: { + toolName: "Bash", + input: { command: "ls" }, + command: "ls", + rawOutput: { content: "x".repeat(10) }, + somethingClientNeverReads: { big: "blob" }, + }, + }), + ); + const payload = projected.payload as Record; + expect(payload.agentId).toBe("task-123"); + expect(payload.parentToolUseId).toBe("toolu_abc"); + // Slimming itself still applies to data. + const data = payload.data as Record; + expect(data.somethingClientNeverReads).toBeUndefined(); + }); + + it("passes task lifecycle payloads (no data field) through untouched", () => { + const source = activity({ + taskId: "task-9", + title: "Audit auth", + role: "explorer", + model: "opus", + effort: "high", + workflowName: "audit-flow", + phases: [{ index: 0, title: "Audit" }], + typedUsage: { totalTokens: 1200 }, + runHandles: { runId: "run-1", scriptPath: "/tmp/wf.js" }, + timelineBypass: true, + }); + const projected = projectActivityPayload(source); + expect(projected.payload).toEqual(source.payload); + }); +}); diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts index e72a100069f..ddb525cd547 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts @@ -39,6 +39,7 @@ import { CheckpointReactorLive } from "./CheckpointReactor.ts"; import { OrchestrationEngineLive } from "./OrchestrationEngine.ts"; import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts"; import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts"; +import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts"; import { RuntimeReceiptBusLive } from "./RuntimeReceiptBus.ts"; import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts"; import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts"; @@ -294,6 +295,7 @@ describe("CheckpointReactor", () => { ); const orchestrationLayer = OrchestrationEngineLive.pipe( Layer.provide(OrchestrationProjectionSnapshotQueryLive), + Layer.provide(ThreadBackgroundLiveness.layer), Layer.provide(OrchestrationProjectionPipelineLive), Layer.provide(OrchestrationEventStoreLive), Layer.provide(OrchestrationCommandReceiptRepositoryLive), @@ -301,6 +303,7 @@ describe("CheckpointReactor", () => { Layer.provide(SqlitePersistenceMemory), ); const projectionSnapshotLayer = OrchestrationProjectionSnapshotQueryLive.pipe( + Layer.provide(ThreadBackgroundLiveness.layer), Layer.provide(RepositoryIdentityResolver.layer), Layer.provide(SqlitePersistenceMemory), ); diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts index 9ffe50d1341..857f5b887fc 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -31,6 +31,7 @@ import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityRes import { OrchestrationEngineLive } from "./OrchestrationEngine.ts"; import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts"; import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts"; +import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts"; import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; import { OrchestrationProjectionPipeline, @@ -55,6 +56,7 @@ async function createOrchestrationSystem() { ), OrchestrationProjectionSnapshotQueryLive, ).pipe( + Layer.provide(ThreadBackgroundLiveness.layer), Layer.provide(OrchestrationEventStoreLive), Layer.provide(OrchestrationCommandReceiptRepositoryLive), Layer.provide(RepositoryIdentityResolver.layer), @@ -817,6 +819,7 @@ describe("OrchestrationEngine", () => { const runtime = ManagedRuntime.make( OrchestrationEngineLive.pipe( Layer.provide(OrchestrationProjectionSnapshotQueryLive), + Layer.provide(ThreadBackgroundLiveness.layer), Layer.provide(OrchestrationProjectionPipelineLive), Layer.provide(Layer.succeed(OrchestrationEventStore, flakyStore)), Layer.provide(OrchestrationCommandReceiptRepositoryLive), @@ -922,6 +925,7 @@ describe("OrchestrationEngine", () => { const runtime = ManagedRuntime.make( OrchestrationEngineLive.pipe( Layer.provide(OrchestrationProjectionSnapshotQueryLive), + Layer.provide(ThreadBackgroundLiveness.layer), Layer.provide(Layer.succeed(OrchestrationProjectionPipeline, flakyProjectionPipeline)), Layer.provide(OrchestrationEventStoreLive), Layer.provide(OrchestrationCommandReceiptRepositoryLive), @@ -1065,6 +1069,7 @@ describe("OrchestrationEngine", () => { const runtime = ManagedRuntime.make( OrchestrationEngineLive.pipe( Layer.provide(OrchestrationProjectionSnapshotQueryLive), + Layer.provide(ThreadBackgroundLiveness.layer), Layer.provide(Layer.succeed(OrchestrationProjectionPipeline, flakyProjectionPipeline)), Layer.provide(Layer.succeed(OrchestrationEventStore, nonTransactionalStore)), Layer.provide(OrchestrationCommandReceiptRepositoryLive), diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index 926182a3ef0..9c4caf4c97d 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -31,6 +31,7 @@ import { OrchestrationProjectionPipelineLive, } from "./ProjectionPipeline.ts"; import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts"; +import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts"; import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; import { OrchestrationProjectionPipeline } from "../Services/ProjectionPipeline.ts"; import { ServerConfig } from "../../config.ts"; @@ -2664,6 +2665,7 @@ it.effect("restores pending turn-start metadata across projection pipeline resta const engineLayer = it.layer( OrchestrationEngineLive.pipe( Layer.provide(OrchestrationProjectionSnapshotQueryLive), + Layer.provide(ThreadBackgroundLiveness.layer), Layer.provide(OrchestrationProjectionPipelineLive), Layer.provide(OrchestrationEventStoreLive), Layer.provide(OrchestrationCommandReceiptRepositoryLive), diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index 6fe7f831a03..22e6dbbe5e7 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -17,6 +17,7 @@ import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityResolver.ts"; import { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts"; import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts"; +import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts"; import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"; const asProjectId = (value: string): ProjectId => ProjectId.make(value); @@ -27,6 +28,7 @@ const asCheckpointRef = (value: string): CheckpointRef => CheckpointRef.make(val const projectionSnapshotLayer = it.layer( OrchestrationProjectionSnapshotQueryLive.pipe( + Layer.provide(ThreadBackgroundLiveness.layer), Layer.provideMerge(RepositoryIdentityResolver.layer), Layer.provideMerge(SqlitePersistenceMemory), Layer.provideMerge(NodeServices.layer), @@ -441,6 +443,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { hasPendingApprovals: true, hasPendingUserInput: false, hasActionableProposedPlan: false, + backgroundLiveness: null, }, ]); @@ -1823,6 +1826,7 @@ it.effect( () => { const resolveCalls: string[] = []; const layer = OrchestrationProjectionSnapshotQueryLive.pipe( + Layer.provide(ThreadBackgroundLiveness.layer), Layer.provideMerge( Layer.succeed(RepositoryIdentityResolver.RepositoryIdentityResolver, { resolve: (cwd: string) => diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 4dcc43913c4..cc41c818196 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -43,6 +43,7 @@ import { type ProjectionRepositoryError, } from "../../persistence/Errors.ts"; import { ProjectionCheckpoint } from "../../persistence/Services/ProjectionCheckpoints.ts"; +import { ThreadBackgroundLivenessService } from "../ThreadBackgroundLiveness.ts"; import { ProjectionProject } from "../../persistence/Services/ProjectionProjects.ts"; import { ProjectionState } from "../../persistence/Services/ProjectionState.ts"; import { ProjectionThreadActivity } from "../../persistence/Services/ProjectionThreadActivities.ts"; @@ -308,6 +309,7 @@ function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: st } const makeProjectionSnapshotQuery = Effect.gen(function* () { + const threadBackgroundLiveness = yield* ThreadBackgroundLivenessService; const sql = yield* SqlClient.SqlClient; const repositoryIdentityResolver = yield* RepositoryIdentityResolver.RepositoryIdentityResolver; const repositoryIdentityResolutionConcurrency = 4; @@ -1671,6 +1673,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { hasPendingApprovals: row.pendingApprovalCount > 0, hasPendingUserInput: row.pendingUserInputCount > 0, hasActionableProposedPlan: row.hasActionableProposedPlan > 0, + backgroundLiveness: threadBackgroundLiveness.getThreadBackgroundLiveness( + row.threadId, + ), } satisfies OrchestrationThreadShell) : Result.failVoid, ), @@ -1810,6 +1815,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { hasPendingApprovals: row.pendingApprovalCount > 0, hasPendingUserInput: row.pendingUserInputCount > 0, hasActionableProposedPlan: row.hasActionableProposedPlan > 0, + backgroundLiveness: threadBackgroundLiveness.getThreadBackgroundLiveness( + row.threadId, + ), }), ), updatedAt: updatedAt ?? "1970-01-01T00:00:00.000Z", @@ -2081,6 +2089,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { hasPendingApprovals: threadRow.value.pendingApprovalCount > 0, hasPendingUserInput: threadRow.value.pendingUserInputCount > 0, hasActionableProposedPlan: threadRow.value.hasActionableProposedPlan > 0, + backgroundLiveness: threadBackgroundLiveness.getThreadBackgroundLiveness( + threadRow.value.threadId, + ), } satisfies OrchestrationThreadShell); }); diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index e4661061b23..8a4c704ef41 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -48,6 +48,7 @@ import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityRes import { OrchestrationEngineLive } from "./OrchestrationEngine.ts"; import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts"; import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts"; +import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts"; import { providerErrorLabel, providerErrorLabelFromInstanceHint, @@ -345,6 +346,7 @@ describe("ProviderCommandReactor", () => { const orchestrationLayer = OrchestrationEngineLive.pipe( Layer.provide(OrchestrationProjectionSnapshotQueryLive), + Layer.provide(ThreadBackgroundLiveness.layer), Layer.provide(OrchestrationProjectionPipelineLive), Layer.provide(OrchestrationEventStoreLive), Layer.provide(OrchestrationCommandReceiptRepositoryLive), @@ -352,6 +354,7 @@ describe("ProviderCommandReactor", () => { Layer.provide(SqlitePersistenceMemory), ); const projectionSnapshotLayer = OrchestrationProjectionSnapshotQueryLive.pipe( + Layer.provide(ThreadBackgroundLiveness.layer), Layer.provide(RepositoryIdentityResolver.layer), Layer.provide(SqlitePersistenceMemory), ); diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 74ece50cd31..c3bdd02e044 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -44,6 +44,7 @@ import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityRes import { OrchestrationEngineLive } from "./OrchestrationEngine.ts"; import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts"; import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts"; +import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts"; import { ProviderRuntimeIngestionLive } from "./ProviderRuntimeIngestion.ts"; import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeIngestion.ts"; @@ -237,6 +238,9 @@ describe("ProviderRuntimeIngestion", () => { const layer = ProviderRuntimeIngestionLive.pipe( Layer.provideMerge(orchestrationLayer), Layer.provideMerge(projectionSnapshotLayer), + // Single shared liveness instance across ingestion (writer), the + // engine, and the snapshot query (reader). + Layer.provideMerge(ThreadBackgroundLiveness.layer), Layer.provideMerge(SqlitePersistenceMemory), Layer.provideMerge(Layer.succeed(ProviderService, provider.service)), Layer.provideMerge(makeTestServerSettingsLayer(options?.serverSettings)), @@ -3143,7 +3147,7 @@ describe("ProviderRuntimeIngestion", () => { (activity: ProviderRuntimeTestActivity) => activity.id === "evt-task-started", ); const progress = thread.activities.find( - (activity: ProviderRuntimeTestActivity) => activity.id === "evt-task-progress", + (activity: ProviderRuntimeTestActivity) => activity.id === "task-progress:turn-task-1", ); const completed = thread.activities.find( (activity: ProviderRuntimeTestActivity) => activity.id === "evt-task-completed", @@ -3227,7 +3231,7 @@ describe("ProviderRuntimeIngestion", () => { ); const progress = thread.activities.find( - (activity: ProviderRuntimeTestActivity) => activity.id === "evt-named-task-progress", + (activity: ProviderRuntimeTestActivity) => activity.id === "task-progress:named-task-1", ); const completed = thread.activities.find( (activity: ProviderRuntimeTestActivity) => activity.id === "evt-named-task-completed", @@ -3318,7 +3322,7 @@ describe("ProviderRuntimeIngestion", () => { await waitForThread(harness.readModel, (entry) => entry.activities.some( - (activity: ProviderRuntimeTestActivity) => activity.id === "evt-swept-task-progress", + (activity: ProviderRuntimeTestActivity) => activity.id === "task-progress:swept-task-1", ), ); diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index c8d619270d3..fed56364658 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -7,6 +7,8 @@ import { type OrchestrationMessage, type OrchestrationProposedPlanId, CheckpointRef, + classifyTaskAgentKind, + EventId, isToolLifecycleItemType, ThreadId, type ThreadTokenUsageSnapshot, @@ -32,6 +34,7 @@ import { ProjectionTurnRepository } from "../../persistence/Services/ProjectionT import { ProjectionTurnRepositoryLive } from "../../persistence/Layers/ProjectionTurns.ts"; import { isGitRepository } from "../../git/Utils.ts"; import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; +import { ThreadBackgroundLivenessService } from "../ThreadBackgroundLiveness.ts"; import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"; import { ProviderRuntimeIngestionService, @@ -307,6 +310,52 @@ function requestKindFromCanonicalRequestType( } } +/** + * Copies the optional TaskAgentLinkage bundle from a task.* runtime payload + * into the persisted activity payload. Identity fields ride on every row so + * client folds survive activity retention; absent fields stay absent. + */ +function taskLinkageActivityFields(payload: Record): Record { + const fields: Record = { + // Server-stamped classification: persisted rows are self-describing, so + // clients trust the stamp instead of re-deriving agent-vs-background + // from taskType denylists and marker heuristics (legacy rows without a + // stamp keep the client fallback). + agentKind: classifyTaskAgentKind({ + taskType: typeof payload.taskType === "string" ? payload.taskType : undefined, + agentId: typeof payload.agentId === "string" ? payload.agentId : undefined, + }), + }; + for (const key of [ + "taskType", + "agentId", + "title", + "role", + "model", + "effort", + "toolUseId", + "parentAgentId", + "workflowName", + "agentIndex", + "phaseIndex", + "phaseTitle", + "phases", + "attempt", + "runHandles", + "outputFile", + "agentPath", + "timelineBypass", + "typedUsage", + "status", + "error", + ] as const) { + if (payload[key] !== undefined) { + fields[key] = payload[key]; + } + } + return fields; +} + export function runtimeEventToActivities( event: ProviderRuntimeEvent, taskTitle?: string, @@ -505,6 +554,7 @@ export function runtimeEventToActivities( ...(event.payload.description ? { detail: truncateDetail(event.payload.description) } : {}), + ...taskLinkageActivityFields(event.payload as Record), }, turnId: toTurnId(event.turnId) ?? null, ...maybeSequence, @@ -515,7 +565,13 @@ export function runtimeEventToActivities( case "task.progress": { return [ { - id: event.eventId, + // Stable per-task id: progress is "latest state", not history, so + // each tick REPLACES the last via the activity upsert (PK + the + // replace-by-id apply in projector and client reducer). Keeps one + // progress row per task instead of thousands, so a large fleet's + // ticks can no longer evict its own start/terminal rows out of + // the 500-row retention window. + id: EventId.make(`task-progress:${event.payload.taskId}`), createdAt: event.createdAt, tone: "info", kind: "task.progress", @@ -532,6 +588,70 @@ export function runtimeEventToActivities( ...(event.payload.summary ? { summary: truncateDetail(event.payload.summary) } : {}), ...(event.payload.lastToolName ? { lastToolName: event.payload.lastToolName } : {}), ...(event.payload.usage !== undefined ? { usage: event.payload.usage } : {}), + ...taskLinkageActivityFields(event.payload as Record), + }, + turnId: toTurnId(event.turnId) ?? null, + ...maybeSequence, + }, + ]; + } + + case "task.updated": { + return [ + { + id: event.eventId, + createdAt: event.createdAt, + tone: event.payload.status === "failed" ? "error" : "info", + kind: "task.updated", + summary: + event.payload.status === "failed" + ? "Task failed" + : event.payload.status + ? `Task ${event.payload.status}` + : "Task updated", + payload: { + taskId: event.payload.taskId, + ...(event.payload.description + ? { detail: truncateDetail(event.payload.description) } + : {}), + ...(event.payload.endedAt ? { endedAt: event.payload.endedAt } : {}), + ...(event.payload.isBackgrounded !== undefined + ? { isBackgrounded: event.payload.isBackgrounded } + : {}), + ...taskLinkageActivityFields(event.payload as Record), + }, + turnId: toTurnId(event.turnId) ?? null, + ...maybeSequence, + }, + ]; + } + + case "tool.progress": { + // Only agent-owned heartbeats are persisted: they feed the owning + // agent's activity line. Parent-conversation tool progress stays + // ephemeral (item lifecycle already covers it). + if (event.payload.taskId === undefined) { + return []; + } + return [ + { + // Same stable-id treatment as task.progress: a heartbeat is + // "what is this agent doing right now", so one row per task. + id: EventId.make(`tool-progress:${event.payload.taskId}`), + createdAt: event.createdAt, + tone: "info", + kind: "tool.progress", + summary: event.payload.toolName ?? "Tool progress", + payload: { + taskId: event.payload.taskId, + ...(event.payload.toolName ? { toolName: event.payload.toolName } : {}), + ...(event.payload.toolUseId ? { toolUseId: event.payload.toolUseId } : {}), + ...(event.payload.elapsedSeconds !== undefined + ? { elapsedSeconds: event.payload.elapsedSeconds } + : {}), + ...(event.payload.parentToolUseId + ? { parentToolUseId: event.payload.parentToolUseId } + : {}), }, turnId: toTurnId(event.turnId) ?? null, ...maybeSequence, @@ -565,6 +685,7 @@ export function runtimeEventToActivities( } : {}), ...(event.payload.usage !== undefined ? { usage: event.payload.usage } : {}), + ...taskLinkageActivityFields(event.payload as Record), }, turnId: toTurnId(event.turnId) ?? null, ...maybeSequence, @@ -630,6 +751,10 @@ export function runtimeEventToActivities( ...(event.payload.status ? { status: event.payload.status } : {}), ...(event.payload.detail ? { detail: truncateDetail(event.payload.detail) } : {}), ...(event.payload.data !== undefined ? { data: event.payload.data } : {}), + ...(event.payload.agentId ? { agentId: event.payload.agentId } : {}), + ...(event.payload.parentToolUseId + ? { parentToolUseId: event.payload.parentToolUseId } + : {}), }, turnId: toTurnId(event.turnId) ?? null, ...maybeSequence, @@ -652,6 +777,10 @@ export function runtimeEventToActivities( itemType: event.payload.itemType, ...(event.payload.detail ? { detail: truncateDetail(event.payload.detail) } : {}), ...(event.payload.data !== undefined ? { data: event.payload.data } : {}), + ...(event.payload.agentId ? { agentId: event.payload.agentId } : {}), + ...(event.payload.parentToolUseId + ? { parentToolUseId: event.payload.parentToolUseId } + : {}), }, turnId: toTurnId(event.turnId) ?? null, ...maybeSequence, @@ -673,6 +802,10 @@ export function runtimeEventToActivities( payload: { itemType: event.payload.itemType, ...(event.payload.detail ? { detail: truncateDetail(event.payload.detail) } : {}), + ...(event.payload.agentId ? { agentId: event.payload.agentId } : {}), + ...(event.payload.parentToolUseId + ? { parentToolUseId: event.payload.parentToolUseId } + : {}), }, turnId: toTurnId(event.turnId) ?? null, ...maybeSequence, @@ -688,6 +821,7 @@ export function runtimeEventToActivities( } const make = Effect.gen(function* () { + const threadBackgroundLiveness = yield* ThreadBackgroundLivenessService; const crypto = yield* Crypto.Crypto; const orchestrationEngine = yield* OrchestrationEngineService; const projectionSnapshotQuery = yield* ProjectionSnapshotQuery; @@ -1756,6 +1890,43 @@ const make = Effect.gen(function* () { yield* rememberTaskDescription(thread.id, event.payload.taskId, description); } } + // Sidebar background liveness: fed from the same lifecycle stream, + // read by the shell query at mapping time (no persistence). + switch (event.type) { + case "task.started": + case "task.progress": + case "task.updated": + case "task.completed": { + const payload = event.payload as { + taskId: string; + taskType?: string; + status?: string; + agentId?: string; + }; + threadBackgroundLiveness.recordTaskLiveness({ + threadId: thread.id, + taskId: payload.taskId, + taskType: payload.taskType, + status: payload.status, + agentId: payload.agentId, + kind: + event.type === "task.started" + ? "started" + : event.type === "task.progress" + ? "progress" + : event.type === "task.updated" + ? "updated" + : "completed", + }); + break; + } + case "session.exited": + threadBackgroundLiveness.clearThreadLiveness(thread.id); + break; + default: + break; + } + let taskTitle: string | undefined; if (event.type === "task.completed") { taskTitle = yield* lookupTaskDescription(thread.id, event.payload.taskId); diff --git a/apps/server/src/orchestration/ThreadBackgroundLiveness.test.ts b/apps/server/src/orchestration/ThreadBackgroundLiveness.test.ts new file mode 100644 index 00000000000..0c4841e8119 --- /dev/null +++ b/apps/server/src/orchestration/ThreadBackgroundLiveness.test.ts @@ -0,0 +1,173 @@ +import { describe, expect, it } from "vite-plus/test"; +import * as ThreadBackgroundLiveness from "./ThreadBackgroundLiveness.ts"; + +describe("ThreadBackgroundLiveness", () => { + it("agents present as working; monitors as monitoring; agents win", () => { + const liveness = ThreadBackgroundLiveness.make(); + const threadId = "t-live-1"; + liveness.recordTaskLiveness({ + threadId, + taskId: "m1", + taskType: "local_bash", + status: undefined, + kind: "started", + }); + expect(liveness.getThreadBackgroundLiveness(threadId)).toBe("monitoring"); + liveness.recordTaskLiveness({ + threadId, + taskId: "a1", + taskType: "subagent", + status: undefined, + kind: "started", + }); + expect(liveness.getThreadBackgroundLiveness(threadId)).toBe("working"); + liveness.recordTaskLiveness({ + threadId, + taskId: "a1", + taskType: "subagent", + status: "completed", + kind: "completed", + }); + expect(liveness.getThreadBackgroundLiveness(threadId)).toBe("monitoring"); + liveness.recordTaskLiveness({ + threadId, + taskId: "m1", + taskType: "local_bash", + status: "completed", + kind: "completed", + }); + expect(liveness.getThreadBackgroundLiveness(threadId)).toBeNull(); + }); + + it("terminal rows without a taskType still clear monitor entries", () => { + const liveness = ThreadBackgroundLiveness.make(); + const threadId = "t-live-2"; + liveness.recordTaskLiveness({ + threadId, + taskId: "m1", + taskType: "local_bash", + status: undefined, + kind: "started", + }); + // Terminal tick arrives with no taskType (common on task.completed). + liveness.recordTaskLiveness({ + threadId, + taskId: "m1", + taskType: undefined, + status: "completed", + kind: "completed", + }); + expect(liveness.getThreadBackgroundLiveness(threadId)).toBeNull(); + }); + + it("nested agents (agentId + agent taskType) still count toward liveness", () => { + const liveness = ThreadBackgroundLiveness.make(); + const threadId = "t-live-nested"; + liveness.recordTaskLiveness({ + threadId, + taskId: "n1", + taskType: "local_agent", + status: undefined, + kind: "started", + agentId: "owner", + }); + expect(liveness.getThreadBackgroundLiveness(threadId)).toBe("working"); + liveness.recordTaskLiveness({ + threadId, + taskId: "n1", + taskType: "local_agent", + status: "completed", + kind: "completed", + agentId: "owner", + }); + expect(liveness.getThreadBackgroundLiveness(threadId)).toBeNull(); + }); + + it("untyped rows count as agents; idle is not live; agent-owned tasks are ignored", () => { + const liveness = ThreadBackgroundLiveness.make(); + const threadId = "t-live-3"; + liveness.recordTaskLiveness({ + threadId, + taskId: "wf:1", + taskType: undefined, + status: "running", + kind: "progress", + }); + expect(liveness.getThreadBackgroundLiveness(threadId)).toBe("working"); + liveness.recordTaskLiveness({ + threadId, + taskId: "wf:1", + taskType: undefined, + status: "idle", + kind: "updated", + }); + expect(liveness.getThreadBackgroundLiveness(threadId)).toBeNull(); + liveness.recordTaskLiveness({ + threadId, + taskId: "sh:1", + taskType: "local_bash", + status: undefined, + kind: "started", + agentId: "owner", + }); + expect(liveness.getThreadBackgroundLiveness(threadId)).toBeNull(); + }); + + it("reclassification moves a task between buckets instead of duplicating it", () => { + const liveness = ThreadBackgroundLiveness.make(); + const threadId = "t-live-reclass"; + // First seen without a taskType: counts as an agent. + liveness.recordTaskLiveness({ + threadId, + taskId: "x1", + taskType: undefined, + status: "running", + kind: "started", + }); + expect(liveness.getThreadBackgroundLiveness(threadId)).toBe("working"); + // Later transition reveals it's a shell: downgrade to monitoring, not + // a stale duplicate pinning "working". + liveness.recordTaskLiveness({ + threadId, + taskId: "x1", + taskType: "local_bash", + status: "running", + kind: "progress", + }); + expect(liveness.getThreadBackgroundLiveness(threadId)).toBe("monitoring"); + // Turning out to be inert or agent-owned drops the prior entry too. + liveness.recordTaskLiveness({ + threadId, + taskId: "x1", + taskType: "local_bash", + status: "running", + kind: "progress", + agentId: "owner", + }); + expect(liveness.getThreadBackgroundLiveness(threadId)).toBeNull(); + }); + + it("plan tasks are inert; clear removes everything; instances are isolated", () => { + const a = ThreadBackgroundLiveness.make(); + const b = ThreadBackgroundLiveness.make(); + a.recordTaskLiveness({ + threadId: "t", + taskId: "p1", + taskType: "plan", + status: undefined, + kind: "started", + }); + expect(a.getThreadBackgroundLiveness("t")).toBeNull(); + a.recordTaskLiveness({ + threadId: "t", + taskId: "a1", + taskType: "local_workflow", + status: undefined, + kind: "started", + }); + expect(a.getThreadBackgroundLiveness("t")).toBe("working"); + expect(b.getThreadBackgroundLiveness("t")).toBeNull(); + a.clearThreadLiveness("t"); + expect(a.getThreadBackgroundLiveness("t")).toBeNull(); + }); +}); diff --git a/apps/server/src/orchestration/ThreadBackgroundLiveness.ts b/apps/server/src/orchestration/ThreadBackgroundLiveness.ts new file mode 100644 index 00000000000..8563e7665fb --- /dev/null +++ b/apps/server/src/orchestration/ThreadBackgroundLiveness.ts @@ -0,0 +1,160 @@ +/** + * ThreadBackgroundLivenessService - in-memory per-thread background liveness + * for the sidebar status pill. + * + * The turn can settle while native background work runs on (subagent fleets, + * workflow runs, Monitor watch loops); the shell previously showed nothing. + * Ingestion records task lifecycle transitions and the shell query reads the + * derived state at mapping time — no persistence, no migration. After a + * server restart the registry is empty until new task events arrive, which + * matches reality: orphaned background work is not live. + * + * "monitoring" is reserved for watch loops (monitor tasks and background + * shells) when they are the ONLY live work; any agent work presents as + * "working". + * + * @module ThreadBackgroundLivenessService + */ +import { INERT_TASK_TYPES, MONITOR_TASK_TYPES } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; + +export type ThreadBackgroundLiveness = "working" | "monitoring" | null; + +interface ThreadLivenessState { + readonly agents: Set; + readonly monitors: Set; +} + +// Classification sets are the shared contracts copies (MONITOR_TASK_TYPES: +// watch loops — monitor tasks plus background shells, which in practice are +// PR babysitting/log tails since pacing sleeps complete inside the turn; +// INERT_TASK_TYPES: plan-mode bookkeeping) so this registry, ingestion's +// agentKind stamp, and the client fold can never drift apart. + +const TERMINAL_STATUSES: ReadonlySet = new Set([ + "completed", + "failed", + "stopped", + "cancelled", + "interrupted", +]); + +export class ThreadBackgroundLivenessService extends Context.Service< + ThreadBackgroundLivenessService, + { + /** + * Feed one task lifecycle transition. taskType may be absent on + * synthesized rows (workflow members, Codex children) — those count as + * agents. agentId marks a task launched from inside a subagent: its + * internal shells are covered by the owning agent's liveness, but a + * NESTED AGENT (agentId + agent-flavored taskType) still counts — it + * can outlive its parent and must keep the thread Working. + */ + readonly recordTaskLiveness: (input: { + readonly threadId: string; + readonly taskId: string; + readonly taskType: string | undefined; + readonly status: string | undefined; + readonly kind: "started" | "progress" | "updated" | "completed"; + readonly agentId?: string | undefined; + }) => void; + + /** Session death orphans all of a thread's background work. */ + readonly clearThreadLiveness: (threadId: string) => void; + + /** + * Two-state vocabulary by design: any live agent work is "working"; + * "monitoring" only when watch loops are the ONLY live work. + */ + readonly getThreadBackgroundLiveness: (threadId: string) => ThreadBackgroundLiveness; + } +>()("t3/orchestration/ThreadBackgroundLiveness/ThreadBackgroundLivenessService") {} + +export function make(): ThreadBackgroundLivenessService["Service"] { + const stateByThreadId = new Map(); + + const stateFor = (threadId: string): ThreadLivenessState => { + const existing = stateByThreadId.get(threadId); + if (existing) { + return existing; + } + const created: ThreadLivenessState = { agents: new Set(), monitors: new Set() }; + stateByThreadId.set(threadId, created); + return created; + }; + + // Classification is per-transition, not sticky: a task first seen without + // a taskType may later reveal itself as a shell, become inert, or turn out + // to be agent-owned. Every path drops any prior entry for the taskId so a + // stale bucket assignment can't pin the thread's status (review finding). + const drop = (threadId: string, taskId: string) => { + const state = stateByThreadId.get(threadId); + if (!state) { + return; + } + state.agents.delete(taskId); + state.monitors.delete(taskId); + if (state.agents.size === 0 && state.monitors.size === 0) { + stateByThreadId.delete(threadId); + } + }; + + return { + recordTaskLiveness: (input) => { + const taskType = input.taskType; + if (taskType !== undefined && INERT_TASK_TYPES.has(taskType)) { + drop(input.threadId, input.taskId); + return; + } + // A subagent's internal non-agent work (its own shells/monitors) is + // covered by the owning agent's liveness. Nested agents fall through: + // they can outlive their parent (review finding). + if ( + input.agentId !== undefined && + (taskType === undefined || MONITOR_TASK_TYPES.has(taskType)) + ) { + drop(input.threadId, input.taskId); + return; + } + + // Idle counts as not-live: a resting (resumable) Codex child isn't + // doing anything, and an all-idle fleet must not pin Working. + const terminal = + input.kind === "completed" || + input.status === "idle" || + (input.status !== undefined && TERMINAL_STATUSES.has(input.status)); + if (terminal) { + drop(input.threadId, input.taskId); + return; + } + + drop(input.threadId, input.taskId); + const state = stateFor(input.threadId); + const bucket = + taskType !== undefined && MONITOR_TASK_TYPES.has(taskType) ? state.monitors : state.agents; + bucket.add(input.taskId); + }, + + clearThreadLiveness: (threadId) => { + stateByThreadId.delete(threadId); + }, + + getThreadBackgroundLiveness: (threadId) => { + const state = stateByThreadId.get(threadId); + if (!state) { + return null; + } + if (state.agents.size > 0) { + return "working"; + } + if (state.monitors.size > 0) { + return "monitoring"; + } + return null; + }, + }; +} + +export const layer = Layer.effect(ThreadBackgroundLivenessService, Effect.sync(make)); diff --git a/apps/server/src/orchestration/runtimeLayer.ts b/apps/server/src/orchestration/runtimeLayer.ts index a2ed5875950..0bc624ec365 100644 --- a/apps/server/src/orchestration/runtimeLayer.ts +++ b/apps/server/src/orchestration/runtimeLayer.ts @@ -5,6 +5,7 @@ import { OrchestrationEventStoreLive } from "../persistence/Layers/Orchestration import { OrchestrationEngineLive } from "./Layers/OrchestrationEngine.ts"; import { OrchestrationProjectionPipelineLive } from "./Layers/ProjectionPipeline.ts"; import { OrchestrationProjectionSnapshotQueryLive } from "./Layers/ProjectionSnapshotQuery.ts"; +import * as ThreadBackgroundLiveness from "./ThreadBackgroundLiveness.ts"; export const OrchestrationEventInfrastructureLayerLive = Layer.mergeAll( OrchestrationEventStoreLive, @@ -19,7 +20,10 @@ export const OrchestrationInfrastructureLayerLive = Layer.mergeAll( OrchestrationProjectionSnapshotQueryLive, OrchestrationEventInfrastructureLayerLive, OrchestrationProjectionPipelineLayerLive, -); + // Shared background-liveness registry: written by runtime ingestion, + // read by the snapshot query. provideMerge feeds the same instance to + // the snapshot query here and re-exports it for runtime ingestion. +).pipe(Layer.provideMerge(ThreadBackgroundLiveness.layer)); export const OrchestrationLayerLive = Layer.mergeAll( OrchestrationInfrastructureLayerLive, diff --git a/apps/server/src/orchestration/workflowScriptQuery.test.ts b/apps/server/src/orchestration/workflowScriptQuery.test.ts new file mode 100644 index 00000000000..d1031458b30 --- /dev/null +++ b/apps/server/src/orchestration/workflowScriptQuery.test.ts @@ -0,0 +1,58 @@ +// @effect-diagnostics nodeBuiltinImport:off +import * as NodeFS from "node:fs"; +import * as NodeOS from "node:os"; +import * as NodePath from "node:path"; +import { it as effectIt } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import { afterAll, assert, describe } from "vite-plus/test"; +import { readWorkflowScript } from "./workflowScriptQuery.ts"; + +const root = NodePath.join(NodeOS.homedir(), ".claude", "projects", "__wf_script_test__"); +NodeFS.mkdirSync(root, { recursive: true }); +const scriptPath = NodePath.join(root, "run.js"); +NodeFS.writeFileSync(scriptPath, "export const meta = {};\n"); +const outside = NodePath.join(NodeOS.tmpdir(), "wf-outside.js"); +NodeFS.writeFileSync(outside, "evil\n"); +const link = NodePath.join(root, "sneaky.js"); +try { + NodeFS.symlinkSync(outside, link); +} catch { + // pre-existing from a prior run +} + +afterAll(() => { + NodeFS.rmSync(root, { recursive: true, force: true }); + NodeFS.rmSync(outside, { force: true }); +}); + +describe("readWorkflowScript containment", () => { + effectIt.effect("serves a real script under the projects root", () => + Effect.gen(function* () { + const result = yield* readWorkflowScript({ scriptPath }); + assert.include(result.contents, "export const meta"); + assert.equal(result.truncated, false); + }), + ); + + effectIt.effect("rejects relative and non-js paths", () => + Effect.gen(function* () { + const relative = yield* Effect.exit(readWorkflowScript({ scriptPath: "run.js" })); + assert.equal(relative._tag, "Failure"); + const nonJs = yield* Effect.exit( + readWorkflowScript({ scriptPath: scriptPath.replace(".js", ".ts") }), + ); + assert.equal(nonJs._tag, "Failure"); + }), + ); + + effectIt.effect("rejects paths outside the root and symlink escapes", () => + Effect.gen(function* () { + const escaped = yield* Effect.exit(readWorkflowScript({ scriptPath: outside })); + assert.equal(escaped._tag, "Failure"); + // A symlink INSIDE the root pointing outside must also fail (realpath + // re-containment of the leaf). + const sneaky = yield* Effect.exit(readWorkflowScript({ scriptPath: link })); + assert.equal(sneaky._tag, "Failure"); + }), + ); +}); diff --git a/apps/server/src/orchestration/workflowScriptQuery.ts b/apps/server/src/orchestration/workflowScriptQuery.ts new file mode 100644 index 00000000000..b02418e32a8 --- /dev/null +++ b/apps/server/src/orchestration/workflowScriptQuery.ts @@ -0,0 +1,115 @@ +// @effect-diagnostics nodeBuiltinImport:off +/** + * Read-only access to persisted workflow scripts for the Agents surface's + * "{} script" affordance. + * + * Containment rules (lifted from the reviewed #3650 inspection service): + * - the resolved realpath must live under ~/.claude/projects (where the + * Claude harness persists workflow scripts) — realpath re-containment + * defeats symlink escapes, including a symlinked leaf file; + * - only .js leaf files are served; + * - reads are size-capped rather than failed, with a truncation marker. + * + * The client-supplied path is a hint from the workflow's runHandles; it is + * never trusted beyond these checks. + */ +import * as NodeFSP from "node:fs/promises"; +import * as NodeOS from "node:os"; +import * as NodePath from "node:path"; + +import { OrchestrationGetWorkflowScriptError } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; + +const SCRIPT_BYTE_CAP = 256 * 1024; + +function scriptsRoot(): string { + return NodePath.join(NodeOS.homedir(), ".claude", "projects"); +} + +export const readWorkflowScript = Effect.fn("orchestration.readWorkflowScript")(function* (input: { + readonly scriptPath: string; +}) { + const requested = input.scriptPath; + + if (!NodePath.isAbsolute(requested) || NodePath.extname(requested) !== ".js") { + return yield* Effect.fail( + new OrchestrationGetWorkflowScriptError({ reason: "invalid-path", scriptPath: requested }), + ); + } + + const root = yield* Effect.tryPromise({ + try: () => NodeFSP.realpath(scriptsRoot()), + catch: (cause) => + new OrchestrationGetWorkflowScriptError({ + reason: "root-unavailable", + scriptPath: requested, + cause, + }), + }); + + // Realpath the FILE itself (not just its directory): a symlink named + // like a script inside a contained directory must not escape. + const resolved = yield* Effect.tryPromise({ + try: () => NodeFSP.realpath(requested), + catch: (cause) => + new OrchestrationGetWorkflowScriptError({ + reason: "not-found", + scriptPath: requested, + cause, + }), + }); + + if (resolved !== root && !resolved.startsWith(`${root}${NodePath.sep}`)) { + return yield* Effect.fail( + new OrchestrationGetWorkflowScriptError({ reason: "outside-root", scriptPath: resolved }), + ); + } + if (NodePath.extname(resolved) !== ".js") { + return yield* Effect.fail( + new OrchestrationGetWorkflowScriptError({ reason: "not-js", scriptPath: resolved }), + ); + } + + // TOCTOU-safe read (review finding): open FIRST, then verify what was + // actually opened via the file descriptor. Re-checking the path after + // open would race against a swap; fstat on the handle cannot. + const read = yield* Effect.tryPromise({ + try: async () => { + const handle = await NodeFSP.open(resolved, "r"); + try { + const stat = await handle.stat(); + if (!stat.isFile()) { + throw new Error("not a regular file"); + } + // The opened inode must be the same one realpath resolved to: a + // process swapping the path between realpath and open changes the + // inode, which this comparison catches. + const pathStat = await NodeFSP.lstat(resolved); + if (stat.ino !== pathStat.ino || stat.dev !== pathStat.dev) { + throw new Error("file changed between resolution and open"); + } + const truncated = stat.size > SCRIPT_BYTE_CAP; + const buffer = Buffer.alloc(Math.min(stat.size, SCRIPT_BYTE_CAP)); + const { bytesRead } = await handle.read(buffer, 0, buffer.length, 0); + return { + contents: buffer.subarray(0, bytesRead).toString("utf8"), + truncated, + }; + } finally { + await handle.close(); + } + }, + catch: (cause) => + new OrchestrationGetWorkflowScriptError({ + reason: "read-failed", + scriptPath: resolved, + cause, + }), + }); + + return { + scriptPath: resolved, + contents: read.contents, + truncated: read.truncated, + }; +}); diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index 760f0e7fbab..c4d6ded5097 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -55,6 +55,7 @@ class FakeClaudeQuery implements AsyncIterable { private failure: unknown | undefined; public readonly interruptCalls: Array = []; + public readonly stopTaskCalls: Array = []; public readonly setModelCalls: Array = []; public readonly setPermissionModeCalls: Array = []; public readonly setMaxThinkingTokensCalls: Array = []; @@ -98,6 +99,10 @@ class FakeClaudeQuery implements AsyncIterable { this.interruptCalls.push(undefined); }; + readonly stopTask = async (taskId: string): Promise => { + this.stopTaskCalls.push(taskId); + }; + readonly setModel = async (model?: string): Promise => { this.setModelCalls.push(model); }; @@ -1445,6 +1450,155 @@ describe("ClaudeAdapterLive", () => { ); }); + it.effect("interruptTurn stops every live task before interrupting the turn", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + + // Wait for the three task.* runtime events to prove the lifecycle + // handlers processed the emissions (no wall-clock sleeps under the + // test clock). + const taskEventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.type.startsWith("task.")), + Stream.take(3), + Stream.runCollect, + Effect.forkChild, + ); + + const session = yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); + yield* adapter.sendTurn({ + threadId: session.threadId, + input: "spawn agents", + attachments: [], + }); + + harness.query.emit({ + type: "system", + subtype: "task_started", + task_id: "task-live", + description: "Agent A", + task_type: "local_agent", + uuid: "task-live-uuid", + session_id: "sdk-session", + } as unknown as SDKMessage); + harness.query.emit({ + type: "system", + subtype: "task_started", + task_id: "task-settled", + description: "Agent B", + task_type: "local_agent", + uuid: "task-settled-uuid", + session_id: "sdk-session", + } as unknown as SDKMessage); + harness.query.emit({ + type: "system", + subtype: "task_notification", + task_id: "task-settled", + status: "completed", + output_file: "/tmp/task-settled.jsonl", + summary: "done", + uuid: "task-settled-done-uuid", + session_id: "sdk-session", + } as unknown as SDKMessage); + + yield* Fiber.join(taskEventsFiber); + + yield* adapter.interruptTurn(session.threadId); + + // Only the still-live task is stopped; interrupt always fires after. + assert.deepEqual(harness.query.stopTaskCalls, ["task-live"]); + assert.equal(harness.query.interruptCalls.length, 1); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + + it.effect("task.started carries model/effort; subagent snapshots refine the model", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + + const taskEventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.type.startsWith("task.")), + Stream.take(2), + Stream.runCollect, + Effect.forkChild, + ); + + const session = yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + modelSelection: createModelSelection( + ProviderInstanceId.make("claudeAgent"), + "claude-opus-4-6", + [{ id: "effort", value: "max" }], + ), + runtimeMode: "full-access", + }); + yield* adapter.sendTurn({ + threadId: session.threadId, + input: "spawn an agent", + attachments: [], + }); + + // No explicit model/effort on the launch input: the task inherits the + // session's selection. + harness.query.emit({ + type: "system", + subtype: "task_started", + task_id: "task-model", + description: "Agent M", + task_type: "local_agent", + tool_use_id: "toolu_agent_m", + uuid: "task-model-uuid", + session_id: "sdk-session", + } as unknown as SDKMessage); + // The subagent's assistant snapshot carries the authoritative API + // model id, which refines the linkage on later rows. + harness.query.emit({ + type: "assistant", + parent_tool_use_id: "toolu_agent_m", + message: { + model: "claude-sonnet-5[1m]", + content: [], + }, + uuid: "subagent-snapshot-uuid", + session_id: "sdk-session", + } as unknown as SDKMessage); + harness.query.emit({ + type: "system", + subtype: "task_progress", + task_id: "task-model", + description: "Agent M", + usage: { total_tokens: 100, tool_uses: 1, duration_ms: 10 }, + uuid: "task-model-progress-uuid", + session_id: "sdk-session", + } as unknown as SDKMessage); + + const taskEvents = Array.from(yield* Fiber.join(taskEventsFiber)); + const started = taskEvents[0]; + assert.equal(started?.type, "task.started"); + if (started?.type === "task.started") { + assert.equal(started.payload.model, "claude-opus-4-6"); + assert.equal(started.payload.effort, "max"); + } + const progress = taskEvents[1]; + assert.equal(progress?.type, "task.progress"); + if (progress?.type === "task.progress") { + assert.equal(progress.payload.model, "claude-sonnet-5[1m]"); + assert.equal(progress.payload.effort, "max"); + } + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + it.effect("closes the session when the Claude stream aborts after a turn starts", () => { const harness = makeHarness(); return Effect.gen(function* () { diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index f87d5be7446..713f9bfd74f 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -42,6 +42,10 @@ import { RuntimeItemId, RuntimeRequestId, RuntimeTaskId, + type RuntimeTaskStatus, + type RuntimeTaskUsage, + type TaskAgentLinkage, + type TaskRunHandles, ThreadId, TurnId, type UserInputQuestion, @@ -169,6 +173,9 @@ interface ToolInFlight { readonly input: Record; readonly partialInputJson: string; readonly lastEmittedInputFingerprint?: string; + /** Owning agent when this tool ran inside a subagent (see attribution note). */ + readonly agentId?: string; + readonly parentToolUseId?: string; } interface ClaudeTaskState { @@ -178,6 +185,28 @@ interface ClaudeTaskState { readonly blockedBy: Set; } +/** + * Agent identity captured from task_started and repeated on every subsequent + * task.* payload, so client folds can reconstruct an agent even when its + * start row aged out of activity retention. + */ +interface ClaudeTaskAgentState { + readonly taskId: string; + toolUseId: string | undefined; + description: string | undefined; + subagentType: string | undefined; + taskType: string | undefined; + workflowName: string | undefined; + skipTranscript: boolean; + runHandles: TaskRunHandles | undefined; + /** Set when this task was launched from inside a subagent. */ + owningAgentId: string | undefined; + /** Seeded from the launching tool's input; refined by the subagent's own + * assistant snapshots (authoritative API model). */ + model: string | undefined; + effort: string | undefined; +} + interface ClaudeSessionContext { session: ProviderSession; readonly promptQueue: Queue.Queue; @@ -186,6 +215,9 @@ interface ClaudeSessionContext { readonly startedAt: string; readonly basePermissionMode: PermissionMode | undefined; currentApiModelId: string | undefined; + /** Effective effort for the session's turns; subagents without an explicit + * effort override inherit this. */ + currentEffort: string | undefined; resumeSessionId: string | undefined; readonly pendingApprovals: Map; readonly pendingUserInputs: Map; @@ -195,6 +227,9 @@ interface ClaudeSessionContext { }>; readonly inFlightTools: Map; readonly claudeTasks: Map; + readonly taskAgents: Map; + /** Task ids that have started and not yet reached a terminal state. */ + readonly liveTaskIds: Set; turnState: ClaudeTurnState | undefined; lastKnownContextWindow: number | undefined; lastKnownTokenUsage: ThreadTokenUsageSnapshot | undefined; @@ -206,6 +241,8 @@ interface ClaudeSessionContext { interface ClaudeQueryRuntime extends AsyncIterable { readonly interrupt: () => Promise; + /** SDK Query.stopTask — present on real queries; optional for test doubles. */ + readonly stopTask?: (taskId: string) => Promise; readonly setModel: (model?: string) => Promise; readonly setPermissionMode: (mode: PermissionMode) => Promise; readonly setMaxThinkingTokens: (maxThinkingTokens: number | null) => Promise; @@ -826,6 +863,225 @@ function planStepsFromClaudeTasks(tasks: Map): PlanStep }); } +/** Only http/https survive; anything else (javascript:, file:, …) is dropped. */ +function sanitizeSessionUrl(value: unknown): string | undefined { + if (typeof value !== "string") { + return undefined; + } + const trimmed = value.trim(); + if (!/^https?:\/\//i.test(trimmed)) { + return undefined; + } + return trimmed; +} + +function nonNegativeInt(value: unknown): number | undefined { + return typeof value === "number" && Number.isFinite(value) && value >= 0 + ? Math.floor(value) + : undefined; +} + +function trimmedString(value: unknown): string | undefined { + if (typeof value !== "string") { + return undefined; + } + const trimmed = value.trim(); + return trimmed.length > 0 ? trimmed : undefined; +} + +/** + * SDK task usage ({total_tokens, tool_uses, duration_ms}, sometimes with + * input/output/cache breakdowns) → the typed contract shape. Unknown or + * malformed input yields undefined rather than a partial guess. + */ +function normalizeTaskUsage(usage: unknown): RuntimeTaskUsage | undefined { + if (typeof usage !== "object" || usage === null) { + return undefined; + } + const record = usage as Record; + const totalTokens = nonNegativeInt(record.total_tokens); + if (totalTokens === undefined) { + return undefined; + } + const inputTokens = nonNegativeInt(record.input_tokens); + const cachedInputTokens = nonNegativeInt(record.cache_read_input_tokens); + const outputTokens = nonNegativeInt(record.output_tokens); + const toolUses = nonNegativeInt(record.tool_uses); + const durationMs = nonNegativeInt(record.duration_ms); + return { + totalTokens, + ...(inputTokens !== undefined ? { inputTokens } : {}), + ...(cachedInputTokens !== undefined ? { cachedInputTokens } : {}), + ...(outputTokens !== undefined ? { outputTokens } : {}), + ...(toolUses !== undefined ? { toolUses } : {}), + ...(durationMs !== undefined ? { durationMs } : {}), + }; +} + +/** SDK task_updated patch status → the shared wire vocabulary. */ +const CLAUDE_TASK_PATCH_STATUS: Record = { + pending: "pending", + running: "running", + completed: "completed", + failed: "failed", + killed: "cancelled", + paused: "idle", +}; + +/** + * Resolves a stream message's parent_tool_use_id to the owning agent's + * taskId. The Task tool's tool_use_id is remembered on task_started; any + * subagent-forwarded block carries that id as its parent. Returns undefined + * for parent-conversation traffic. + */ +function agentIdForParentToolUse( + agents: Map, + parentToolUseId: string | null | undefined, +): string | undefined { + if (parentToolUseId === null || parentToolUseId === undefined) { + return undefined; + } + for (const agent of agents.values()) { + if (agent.toolUseId === parentToolUseId) { + return agent.taskId; + } + } + return undefined; +} + +/** + * Linkage bundle repeated on every task.* payload for `taskId`. Reads the + * remembered identity (from task_started) so progress/terminal rows are + * self-describing even when the start row ages out of activity retention. + */ +function taskLinkageFor( + agents: Map, + taskId: string, +): TaskAgentLinkage { + const agent = agents.get(taskId); + if (!agent) { + return {}; + } + return { + ...(agent.taskType ? { taskType: agent.taskType } : {}), + ...(agent.owningAgentId ? { agentId: agent.owningAgentId } : {}), + ...(agent.description ? { title: agent.description } : {}), + ...(agent.subagentType ? { role: agent.subagentType } : {}), + ...(agent.model ? { model: agent.model } : {}), + ...(agent.effort ? { effort: agent.effort } : {}), + ...(agent.toolUseId ? { toolUseId: agent.toolUseId } : {}), + ...(agent.workflowName ? { workflowName: agent.workflowName } : {}), + ...(agent.runHandles ? { runHandles: agent.runHandles } : {}), + }; +} + +const WORKFLOW_PHASE_CAP = 64; +const WORKFLOW_AGENT_CAP = 100; + +interface ClaudeWorkflowAgentEntry { + readonly index: number; + readonly state: string; + readonly label: string | undefined; + readonly phaseIndex: number | undefined; + readonly phaseTitle: string | undefined; + readonly model: string | undefined; + readonly attempt: number | undefined; + readonly lastToolName: string | undefined; + readonly startedAt: string | undefined; + readonly error: string | undefined; + readonly tokens: number | undefined; + readonly toolCalls: number | undefined; +} + +interface ClaudeWorkflowProgress { + readonly phases: ReadonlyArray<{ index: number; title: string }>; + readonly agents: ReadonlyArray; +} + +/** + * Defensive parse of the SDK's undeclared-but-real workflow_progress array on + * task_progress messages (wire-confirmed; absent from sdk.d.ts). Unknown + * shapes are skipped per-entry; phases and agents dedupe by index before + * caps; a vanished field never throws. If the array disappears upstream the + * caller keeps the coordinator row and plain task lifecycle. + */ +function parseWorkflowProgress(value: unknown): ClaudeWorkflowProgress | undefined { + if (!Array.isArray(value) || value.length === 0) { + return undefined; + } + const phasesByIndex = new Map(); + const agentsByIndex = new Map(); + for (const entry of value) { + if (typeof entry !== "object" || entry === null) { + continue; + } + const record = entry as Record; + const entryType = trimmedString(record.type); + if (entryType === "workflow_phase") { + const index = nonNegativeInt(record.index); + const title = trimmedString(record.title); + if (index !== undefined && title && !phasesByIndex.has(index)) { + phasesByIndex.set(index, title); + } + continue; + } + if (entryType !== "workflow_agent") { + continue; + } + const index = nonNegativeInt(record.index); + const state = trimmedString(record.state); + if (index === undefined || !state || agentsByIndex.has(index)) { + continue; + } + agentsByIndex.set(index, { + index, + state, + label: trimmedString(record.label), + phaseIndex: nonNegativeInt(record.phaseIndex), + phaseTitle: trimmedString(record.phaseTitle), + model: trimmedString(record.model), + attempt: nonNegativeInt(record.attempt), + lastToolName: trimmedString(record.lastToolName), + startedAt: trimmedString(record.startedAt), + error: trimmedString(record.error), + tokens: nonNegativeInt(record.tokens), + toolCalls: nonNegativeInt(record.toolCalls), + }); + } + if (phasesByIndex.size === 0 && agentsByIndex.size === 0) { + return undefined; + } + const phases = Array.from(phasesByIndex.entries()) + .map(([index, title]) => ({ index, title })) + .toSorted((a, b) => a.index - b.index) + .slice(0, WORKFLOW_PHASE_CAP); + const agents = Array.from(agentsByIndex.values()) + .toSorted((a, b) => a.index - b.index) + .slice(0, WORKFLOW_AGENT_CAP); + return { phases, agents }; +} + +/** + * Workflow member states from workflow_progress → shared task status. + * Unknown states read running after startedAt, pending before it. + */ +function workflowAgentStatus(entry: ClaudeWorkflowAgentEntry): RuntimeTaskStatus { + switch (entry.state) { + case "queued": + case "pending": + return "pending"; + case "start": + case "running": + return entry.startedAt === undefined ? "pending" : "running"; + case "done": + return "completed"; + case "error": + return "failed"; + default: + return entry.startedAt === undefined ? "pending" : "running"; + } +} + function summarizeToolRequest(toolName: string, input: Record): string { const commandValue = input.command ?? input.cmd; const command = typeof commandValue === "string" ? commandValue : undefined; @@ -833,17 +1089,17 @@ function summarizeToolRequest(toolName: string, input: Record): return `${toolName}: ${command.trim().slice(0, 400)}`; } - // For agent/subagent tools, prefer human-readable description or prompt over raw JSON + // For agent/subagent tools, prefer the human-readable description or prompt + // over raw JSON. The structured subagent_type is carried separately on the + // task.* payloads (role) — the label is display-only. const itemType = classifyToolItemType(toolName); if (itemType === "collab_agent_tool_call") { const description = typeof input.description === "string" ? input.description.trim() : undefined; const prompt = typeof input.prompt === "string" ? input.prompt.trim() : undefined; - const subagentType = - typeof input.subagent_type === "string" ? input.subagent_type.trim() : undefined; const label = description || (prompt ? prompt.slice(0, 200) : undefined); if (label) { - return subagentType ? `${subagentType}: ${label}` : label; + return label; } } @@ -2078,6 +2334,32 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const { event } = message; + // Subagent-owned stream traffic (parent_tool_use_id set) must not write + // into the parent transcript: with forwardSubagentText off the SDK still + // forwards subagent tool_use/tool_result blocks and their wrapping + // text/thinking deltas, and emitting them interleaved N subagents' + // narration into the chat (live-test finding). Their results reach the + // UI via the task.* lifecycle; their tool blocks are attributed and + // re-homed by the quiet-timeline filter. + const streamParentToolUseId = (message as { parent_tool_use_id?: string | null }) + .parent_tool_use_id; + if (streamParentToolUseId !== null && streamParentToolUseId !== undefined) { + // Drop only the subagent's narration (text/thinking); tool_use blocks + // and their input_json_delta frames must flow so attributed tool items + // keep their inputs (review finding: dropping deltas emptied inputs). + const dropStart = + event.type === "content_block_start" && + event.content_block.type !== "tool_use" && + event.content_block.type !== "server_tool_use" && + event.content_block.type !== "mcp_tool_use"; + const dropDelta = + event.type === "content_block_delta" && + (event.delta.type === "text_delta" || event.delta.type === "thinking_delta"); + if (dropStart || dropDelta) { + return; + } + } + if (event.type === "message_delta") { if (message.parent_tool_use_id !== null && message.parent_tool_use_id !== undefined) { return; @@ -2205,6 +2487,8 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( status: "inProgress", title: nextTool.title, ...(nextTool.detail ? { detail: nextTool.detail } : {}), + ...(nextTool.agentId ? { agentId: nextTool.agentId } : {}), + ...(nextTool.parentToolUseId ? { parentToolUseId: nextTool.parentToolUseId } : {}), data: { toolName: nextTool.toolName, input: nextTool.input, @@ -2274,6 +2558,14 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const inputFingerprint = Object.keys(toolInput).length > 0 ? toolInputFingerprint(toolInput) : undefined; + // Attribute tools that ran inside a subagent to their owning agent so + // clients can re-home them out of the main timeline (quiet-timeline + // guarantee): the SDK forwards subagent tool_use blocks tagged with the + // spawning Task tool's id as parent_tool_use_id. + const parentToolUseId = + (message as { parent_tool_use_id?: string | null }).parent_tool_use_id ?? undefined; + const owningAgentId = agentIdForParentToolUse(context.taskAgents, parentToolUseId); + const tool: ToolInFlight = { itemId, itemType, @@ -2283,6 +2575,8 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( input: toolInput, partialInputJson: "", ...(inputFingerprint ? { lastEmittedInputFingerprint: inputFingerprint } : {}), + ...(owningAgentId ? { agentId: owningAgentId } : {}), + ...(parentToolUseId ? { parentToolUseId } : {}), }; context.inFlightTools.set(index, tool); @@ -2300,6 +2594,8 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( status: "inProgress", title: tool.title, ...(tool.detail ? { detail: tool.detail } : {}), + ...(tool.agentId ? { agentId: tool.agentId } : {}), + ...(tool.parentToolUseId ? { parentToolUseId: tool.parentToolUseId } : {}), data: { toolName: tool.toolName, input: toolInput, @@ -2378,6 +2674,8 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( status: toolResult.isError ? "failed" : "inProgress", title: tool.title, ...(tool.detail ? { detail: tool.detail } : {}), + ...(tool.agentId ? { agentId: tool.agentId } : {}), + ...(tool.parentToolUseId ? { parentToolUseId: tool.parentToolUseId } : {}), data: toolData, }, providerRefs: nativeProviderRefs(context, { @@ -2430,6 +2728,8 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( status: itemStatus, title: tool.title, ...(tool.detail ? { detail: tool.detail } : {}), + ...(tool.agentId ? { agentId: tool.agentId } : {}), + ...(tool.parentToolUseId ? { parentToolUseId: tool.parentToolUseId } : {}), data: toolData, }, providerRefs: nativeProviderRefs(context, { @@ -2442,6 +2742,43 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }, }); + // The Workflow tool's result carries the run handles (runId, scriptPath, + // transcriptDir, sessionUrl). Attach them to the workflow's task agent so + // the next task.* payload advertises them to clients. + if (!toolResult.isError && tool.toolName.toLowerCase() === "workflow" && toolUseResult) { + const workflowTaskId = trimmedString(toolUseResult.taskId); + if (workflowTaskId) { + const runHandles: TaskRunHandles = { + ...(trimmedString(toolUseResult.runId) + ? { runId: trimmedString(toolUseResult.runId) } + : {}), + ...(trimmedString(toolUseResult.scriptPath) + ? { scriptPath: trimmedString(toolUseResult.scriptPath) } + : {}), + ...(trimmedString(toolUseResult.transcriptDir) + ? { transcriptDir: trimmedString(toolUseResult.transcriptDir) } + : {}), + ...(sanitizeSessionUrl(toolUseResult.sessionUrl) + ? { sessionUrl: sanitizeSessionUrl(toolUseResult.sessionUrl) } + : {}), + }; + const existing = context.taskAgents.get(workflowTaskId); + context.taskAgents.set(workflowTaskId, { + taskId: workflowTaskId, + toolUseId: existing?.toolUseId ?? tool.itemId, + description: existing?.description, + subagentType: existing?.subagentType, + taskType: existing?.taskType ?? "local_workflow", + workflowName: existing?.workflowName, + skipTranscript: existing?.skipTranscript ?? false, + runHandles, + owningAgentId: existing?.owningAgentId, + model: existing?.model, + effort: existing?.effort, + }); + } + } + if ( !toolResult.isError && applyClaudeTaskToolResult(context.claudeTasks, tool, toolUseResult) @@ -2465,6 +2802,26 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( return; } + // Subagent-owned assistant snapshots (parent_tool_use_id set) are the + // subagent's own conversation, not the parent's. Emitting them created + // interleaved "Agent N done"-adjacent leak messages and spawned synthetic + // turns per subagent completion (which also reset the Working timer). + const assistantParentToolUseId = (message as { parent_tool_use_id?: string | null }) + .parent_tool_use_id; + if (assistantParentToolUseId !== null && assistantParentToolUseId !== undefined) { + // The snapshot's message.model is the authoritative API model the + // subagent actually ran on — refine the seeded launch-time value. + const owningTaskId = agentIdForParentToolUse(context.taskAgents, assistantParentToolUseId); + const snapshotModel = trimmedString(message.message.model); + const owningAgent = owningTaskId ? context.taskAgents.get(owningTaskId) : undefined; + if (owningAgent && snapshotModel) { + owningAgent.model = snapshotModel; + } + context.lastAssistantUuid = message.uuid; + yield* updateResumeCursor(context); + return; + } + // Auto-start a synthetic turn for assistant messages that arrive without // an active turn (e.g., background agent/subagent responses between user prompts). if (!context.turnState) { @@ -2563,6 +2920,79 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( yield* completeTurn(context, status, errorMessage, message); }); + /** + * Synthesizes per-member task.progress rows from the coordinator's + * workflow_progress array. Member identity is the stable slot + * `:wf:` (never the per-attempt agent id, which + * changes on retry and would split one member into duplicate rows). + * timelineBypass keeps these out of the parent chat; the Agents surface and + * workflow card consume them. + */ + const emitWorkflowMemberProgress = Effect.fn("emitWorkflowMemberProgress")(function* ( + context: ClaudeSessionContext, + base: Omit, + message: Extract, + ) { + const progress = parseWorkflowProgress( + (message as unknown as Record).workflow_progress, + ); + if (!progress) { + return; + } + const coordinatorId = message.task_id; + const coordinatorLinkage = taskLinkageFor(context.taskAgents, coordinatorId); + if (progress.phases.length > 0) { + // Phases ride on a coordinator-addressed progress row. + const stamp = yield* makeEventStamp(); + yield* offerRuntimeEvent({ + ...base, + eventId: stamp.eventId, + createdAt: stamp.createdAt, + type: "task.progress", + payload: { + taskId: RuntimeTaskId.make(coordinatorId), + description: message.description, + phases: progress.phases, + ...coordinatorLinkage, + }, + }); + } + for (const entry of progress.agents) { + const memberTaskId = `${coordinatorId}:wf:${entry.index}`; + const status = workflowAgentStatus(entry); + const stamp = yield* makeEventStamp(); + yield* offerRuntimeEvent({ + ...base, + eventId: stamp.eventId, + createdAt: stamp.createdAt, + type: "task.progress", + payload: { + taskId: RuntimeTaskId.make(memberTaskId), + description: entry.label ?? `agent ${entry.index}`, + status, + ...(entry.error ? { error: entry.error } : {}), + ...(entry.label ? { title: entry.label } : {}), + ...(entry.model ? { model: entry.model } : {}), + ...(entry.lastToolName ? { lastToolName: entry.lastToolName } : {}), + ...(entry.tokens !== undefined + ? { + typedUsage: { + totalTokens: entry.tokens, + ...(entry.toolCalls !== undefined ? { toolUses: entry.toolCalls } : {}), + }, + } + : {}), + parentAgentId: coordinatorId, + agentIndex: entry.index, + ...(entry.phaseIndex !== undefined ? { phaseIndex: entry.phaseIndex } : {}), + ...(entry.phaseTitle ? { phaseTitle: entry.phaseTitle } : {}), + ...(entry.attempt !== undefined ? { attempt: entry.attempt } : {}), + timelineBypass: true, + }, + }); + } + }); + const handleSystemMessage = Effect.fn("handleSystemMessage")(function* ( context: ClaudeSessionContext, message: SDKMessage, @@ -2678,7 +3108,46 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }, }); return; - case "task_started": + case "task_started": { + // A task launched by a tool that itself ran inside a subagent (the + // in-flight tool carries agentId from parent_tool_use_id) is + // agent-internal: a subagent's background shell, not parent work. + const launchingTool = message.tool_use_id + ? Array.from(context.inFlightTools.values()).find( + (tool) => tool.itemId === message.tool_use_id, + ) + : undefined; + const owningAgentId = launchingTool?.agentId; + // Model/effort: the Agent tool's input carries explicit overrides; + // absent ones inherit the session's selection (SDK behavior). + // Subagent assistant snapshots later refine model with the + // authoritative API id. AgentInput.effort may be a named level or an + // integer. + const launchInput = launchingTool?.input; + const model = + trimmedString(launchInput?.model) ?? trimmedString(context.session.model ?? undefined); + const rawLaunchEffort = launchInput?.effort; + const effort = + trimmedString(rawLaunchEffort) ?? + (typeof rawLaunchEffort === "number" && Number.isFinite(rawLaunchEffort) + ? String(rawLaunchEffort) + : context.currentEffort); + // Remember the agent identity so every later task.* payload for this + // taskId is self-describing (identity must survive activity retention). + context.taskAgents.set(message.task_id, { + taskId: message.task_id, + toolUseId: message.tool_use_id, + description: message.description, + subagentType: message.subagent_type, + taskType: message.task_type, + workflowName: message.workflow_name, + skipTranscript: message.skip_transcript === true, + runHandles: context.taskAgents.get(message.task_id)?.runHandles, + owningAgentId, + model, + effort, + }); + context.liveTaskIds.add(message.task_id); yield* offerRuntimeEvent({ ...base, type: "task.started", @@ -2686,10 +3155,18 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( taskId: RuntimeTaskId.make(message.task_id), description: message.description, ...(message.task_type ? { taskType: message.task_type } : {}), + ...(owningAgentId ? { agentId: owningAgentId } : {}), + ...(message.description ? { title: message.description } : {}), + ...(message.subagent_type ? { role: message.subagent_type } : {}), + ...(model ? { model } : {}), + ...(effort ? { effort } : {}), + ...(message.tool_use_id ? { toolUseId: message.tool_use_id } : {}), + ...(message.workflow_name ? { workflowName: message.workflow_name } : {}), }, }); return; - case "task_progress": + } + case "task_progress": { yield* emitThreadTokenUsage( context, normalizeClaudeTaskProgressTokenUsage(message.usage, context), @@ -2698,6 +3175,8 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( rawPayload: message, }, ); + const linkage = taskLinkageFor(context.taskAgents, message.task_id); + const typedUsage = normalizeTaskUsage(message.usage); yield* offerRuntimeEvent({ ...base, type: "task.progress", @@ -2706,16 +3185,47 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( description: message.description, ...(message.summary ? { summary: message.summary } : {}), ...(message.usage ? { usage: message.usage } : {}), + ...(typedUsage ? { typedUsage } : {}), ...(message.last_tool_name ? { lastToolName: message.last_tool_name } : {}), + ...linkage, + ...(message.subagent_type ? { role: message.subagent_type } : {}), }, }); + yield* emitWorkflowMemberProgress(context, base, message); 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": { + // Status patch (killed/paused/backgrounded/end_time/error) — main + // previously dropped this on the floor, losing all transitions. + const patch = message.patch; + const status = + patch.status !== undefined ? CLAUDE_TASK_PATCH_STATUS[patch.status] : undefined; + if (status === "completed" || status === "failed" || status === "cancelled") { + context.liveTaskIds.delete(message.task_id); + } + const endedAt = + typeof patch.end_time === "number" && Number.isFinite(patch.end_time) + ? DateTime.formatIso(DateTime.makeUnsafe(patch.end_time)) + : undefined; + yield* offerRuntimeEvent({ + ...base, + type: "task.updated", + payload: { + taskId: RuntimeTaskId.make(message.task_id), + ...(status ? { status } : {}), + ...(patch.description ? { description: patch.description } : {}), + ...(patch.error ? { error: patch.error } : {}), + ...(endedAt ? { endedAt } : {}), + ...(patch.is_backgrounded !== undefined + ? { isBackgrounded: patch.is_backgrounded } + : {}), + ...taskLinkageFor(context.taskAgents, message.task_id), + }, + }); return; - case "task_notification": + } + case "task_notification": { + context.liveTaskIds.delete(message.task_id); yield* emitThreadTokenUsage( context, normalizeClaudeTaskProgressTokenUsage(message.usage, context), @@ -2724,6 +3234,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( rawPayload: message, }, ); + const typedUsage = normalizeTaskUsage(message.usage); yield* offerRuntimeEvent({ ...base, type: "task.completed", @@ -2732,9 +3243,13 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( status: message.status, ...(message.summary ? { summary: message.summary } : {}), ...(message.usage ? { usage: message.usage } : {}), + ...(typedUsage ? { typedUsage } : {}), + ...(message.output_file ? { outputFile: message.output_file } : {}), + ...taskLinkageFor(context.taskAgents, message.task_id), }, }); return; + } case "files_persisted": yield* offerRuntimeEvent({ ...base, @@ -2870,7 +3385,10 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( toolUseId: message.tool_use_id, toolName: message.tool_name, elapsedSeconds: message.elapsed_time_seconds, - ...(message.task_id ? { summary: `task:${message.task_id}` } : {}), + ...(message.task_id ? { taskId: RuntimeTaskId.make(message.task_id) } : {}), + ...(message.parent_tool_use_id !== null + ? { parentToolUseId: message.parent_tool_use_id } + : {}), }, }); return; @@ -3192,6 +3710,8 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const pendingUserInputs = new Map(); const inFlightTools = new Map(); const claudeTasks = new Map(); + const taskAgents = new Map(); + const liveTaskIds = new Set(); const contextRef = yield* Ref.make(undefined); @@ -3628,12 +4148,15 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( startedAt, basePermissionMode: permissionMode, currentApiModelId: apiModelId, + currentEffort: effectiveEffort ?? undefined, resumeSessionId: sessionId, pendingApprovals, pendingUserInputs, turns: [], inFlightTools, claudeTasks, + taskAgents, + liveTaskIds, turnState: undefined, lastKnownContextWindow: initialContextWindow, lastKnownTokenUsage: undefined, @@ -3750,6 +4273,13 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( ...context.session, model: modelSelection.model, }; + const turnCaps = getClaudeModelCapabilities(modelSelection.model); + const turnEffort = resolveClaudeEffort( + turnCaps, + getModelSelectionStringOptionValue(modelSelection, "effort"), + ); + context.currentEffort = + getEffectiveClaudeAgentEffort(turnEffort ?? null, modelSelection.model) ?? undefined; } // Apply interaction mode by switching the SDK's permission mode. @@ -3825,6 +4355,24 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const interruptTurn: ClaudeAdapterShape["interruptTurn"] = Effect.fn("interruptTurn")( function* (threadId, _turnId) { const context = yield* requireSession(threadId); + // Stop-everything semantics: users reach for Stop precisely when a + // fleet ran away. interrupt() alone only ends the parent turn — + // background subagents/shells keep running and keep burning tokens. + // Stop every live task first (best-effort per task: one refusal must + // not strand the rest or block the turn interrupt), then interrupt. + if (context.query.stopTask && context.liveTaskIds.size > 0) { + const liveIds = Array.from(context.liveTaskIds); + yield* Effect.forEach( + liveIds, + (taskId) => + Effect.tryPromise({ + // Invoke through the query object: SDK methods rely on `this`. + try: () => context.query.stopTask!(taskId), + catch: () => undefined, + }).pipe(Effect.ignore), + { concurrency: 8, discard: true }, + ); + } yield* Effect.tryPromise({ try: () => context.query.interrupt(), catch: (cause) => toRequestError(threadId, "turn/interrupt", cause), diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 4146121b147..564865e1447 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -20,6 +20,8 @@ import { type ProviderUserInputAnswers, RuntimeItemId, RuntimeRequestId, + RuntimeTaskId, + type RuntimeTaskUsage, ProviderApprovalDecision, ThreadId, ProviderSendTurnInput, @@ -494,10 +496,274 @@ function mapItemLifecycle( }; } +/** + * Maps the session runtime's synthetic `collabAgent/*` events (native + * multi-agent v2 child-thread signals) into the shared task.* lifecycle. + * Agent identity = child thread id; nickname is the display title, role is + * agentRole (fallback: last agentPath segment, then "general-purpose"). + * A completed child turn is idle (resumable), not terminal. timelineBypass + * keeps these rows out of the parent chat. + */ +function mapCollabAgentEvent( + event: ProviderEvent, + canonicalThreadId: ThreadId, +): ReadonlyArray { + const payload = + typeof event.payload === "object" && event.payload !== null + ? (event.payload as Record) + : undefined; + const agentThreadId = typeof payload?.agentThreadId === "string" ? payload.agentThreadId : ""; + if (!payload || agentThreadId.length === 0) { + return []; + } + const base = runtimeEventBase(event, canonicalThreadId); + const taskId = RuntimeTaskId.make(agentThreadId); + const agentPath = typeof payload.agentPath === "string" ? payload.agentPath : undefined; + const pathLeaf = agentPath?.split("/").findLast((segment) => segment.length > 0); + const nickname = typeof payload.nickname === "string" ? payload.nickname : undefined; + const role = + (typeof payload.role === "string" ? payload.role : undefined) ?? pathLeaf ?? "general-purpose"; + // A bare thread id is not a name. Omitting the title lets the client fold + // keep the real one from task.started instead of clobbering it (probe + // finding: progress rows renamed math_one to its UUID). + const knownName = nickname ?? pathLeaf; + const title = knownName ?? agentThreadId; + // Identity repeated on every status patch so rows are self-describing when + // the start row ages out of activity retention (review finding: a + // reconstructed agent had a UUID name and no role/path). + const statusLinkage = { + role, + ...(knownName ? { title: knownName } : {}), + ...(agentPath ? { agentPath } : {}), + timelineBypass: true, + } as const; + + switch (event.method) { + case "collabAgent/started": + return [ + { + ...base, + type: "task.started", + payload: { + taskId, + description: title, + title, + role, + ...(agentPath ? { agentPath } : {}), + ...(typeof payload.parentThreadId === "string" + ? { parentAgentId: payload.parentThreadId } + : {}), + timelineBypass: true, + }, + }, + ]; + case "collabAgent/activity": { + const activityKind = typeof payload.activityKind === "string" ? payload.activityKind : ""; + if (activityKind === "interrupted") { + return [ + { + ...base, + type: "task.updated", + payload: { taskId, status: "interrupted", ...statusLinkage }, + }, + ]; + } + if (activityKind === "started") { + // Wire-probe finding: children often register via subAgentActivity + // alone (no thread/started with a spawn source), so this is the one + // shot at a task.started with a real name — agentPath leaf beats a + // bare thread-id title. + return [ + { + ...base, + type: "task.started", + payload: { + taskId, + description: title, + title, + role, + ...(agentPath ? { agentPath } : {}), + timelineBypass: true, + }, + }, + ]; + } + // interacted → the child is (again) actively driven. + return [ + { + ...base, + type: "task.updated", + payload: { taskId, status: "running", ...statusLinkage }, + }, + ]; + } + case "collabAgent/turnStarted": + return [ + { + ...base, + type: "task.updated", + payload: { taskId, status: "running", ...statusLinkage }, + }, + ]; + case "collabAgent/turnCompleted": { + // Idle, not terminal: the identity is resumable via sendInput/resume. + const turn = + typeof payload.turn === "object" && payload.turn !== null + ? (payload.turn as Record) + : undefined; + const turnStatus = typeof turn?.status === "string" ? turn.status : undefined; + const status = + turnStatus === "failed" + ? ("failed" as const) + : turnStatus === "interrupted" + ? ("interrupted" as const) + : ("idle" as const); + return [ + { + ...base, + type: "task.updated", + payload: { taskId, status, ...statusLinkage }, + }, + ]; + } + case "collabAgent/statusChanged": { + const status = + typeof payload.status === "object" && payload.status !== null + ? (payload.status as Record) + : undefined; + const statusType = typeof status?.type === "string" ? status.type : undefined; + if (statusType === "systemError") { + // Silently dropping this once left children stuck running forever. + return [ + { + ...base, + type: "task.updated", + payload: { taskId, status: "failed", ...statusLinkage }, + }, + ]; + } + if (statusType === "active") { + const flags = Array.isArray(status?.activeFlags) ? status.activeFlags : []; + const waiting = flags.some( + (flag) => flag === "waitingOnApproval" || flag === "waitingOnUserInput", + ); + return [ + { + ...base, + type: "task.updated", + payload: { taskId, status: waiting ? "waiting" : "running", ...statusLinkage }, + }, + ]; + } + if (statusType === "idle") { + return [ + { + ...base, + type: "task.updated", + payload: { taskId, status: "idle", ...statusLinkage }, + }, + ]; + } + return []; + } + case "collabAgent/tokenUsage": { + // Cumulative per child thread: always the `total` breakdown, never + // `last` (which shrinks on follow-ups). Client folds max-merge. + const tokenUsage = + typeof payload.tokenUsage === "object" && payload.tokenUsage !== null + ? (payload.tokenUsage as Record) + : undefined; + const total = + typeof tokenUsage?.total === "object" && tokenUsage.total !== null + ? (tokenUsage.total as Record) + : undefined; + const totalTokens = typeof total?.totalTokens === "number" ? total.totalTokens : undefined; + if (totalTokens === undefined) { + return []; + } + const count = (value: unknown): number | undefined => + typeof value === "number" && Number.isFinite(value) && value >= 0 ? value : undefined; + const typedUsage: RuntimeTaskUsage = { + totalTokens, + ...(count(total?.inputTokens) !== undefined + ? { inputTokens: count(total?.inputTokens) } + : {}), + ...(count(total?.cachedInputTokens) !== undefined + ? { cachedInputTokens: count(total?.cachedInputTokens) } + : {}), + ...(count(total?.outputTokens) !== undefined + ? { outputTokens: count(total?.outputTokens) } + : {}), + ...(count(total?.reasoningOutputTokens) !== undefined + ? { reasoningOutputTokens: count(total?.reasoningOutputTokens) } + : {}), + }; + return [ + { + ...base, + type: "task.progress", + payload: { + taskId, + description: title, + ...(knownName ? { title: knownName } : {}), + typedUsage, + timelineBypass: true, + }, + }, + ]; + } + case "collabAgent/item": { + const item = + typeof payload.item === "object" && payload.item !== null + ? (payload.item as Record) + : undefined; + const itemTypeRaw = typeof item?.type === "string" ? item.type : undefined; + if (!itemTypeRaw) { + return []; + } + // A loose summary from the raw item: the child stream is untyped at + // this boundary (synthetic event payload), so read best-effort fields + // rather than force a schema decode. + const looseSummary = + (typeof item?.command === "string" ? item.command : undefined) ?? + (typeof item?.title === "string" ? item.title : undefined) ?? + (typeof item?.query === "string" ? item.query : undefined); + const canonical = toCanonicalItemType(itemTypeRaw); + const summary = looseSummary ?? canonical.replaceAll("_", " "); + return [ + { + ...base, + type: "task.progress", + payload: { + taskId, + description: title, + ...(knownName ? { title: knownName } : {}), + summary, + timelineBypass: true, + }, + }, + ]; + } + case "collabAgent/closed": + return [ + { + ...base, + type: "task.updated", + payload: { taskId, status: "interrupted", ...statusLinkage }, + }, + ]; + default: + return []; + } +} + function mapToRuntimeEvents( event: ProviderEvent, canonicalThreadId: ThreadId, ): ReadonlyArray { + if (event.kind === "notification" && event.method.startsWith("collabAgent/")) { + return mapCollabAgentEvent(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 67108dd4dbb..2a65fb9d6dc 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -600,6 +600,71 @@ function readRouteFields(notification: CodexServerNotification): { } } +/** + * Native collab child-agent tracking (multi-agent v2). Under v2 subagents are + * full app-server threads: identity arrives on `thread/started` with + * source.subAgent.thread_spawn, lifecycle on `subAgentActivity` items and the + * child thread's own turn/status/tokenUsage notifications. The runtime + * registers children from those explicit signals, intercepts their + * notifications before parent-timeline mapping, and re-emits them as + * synthetic `collabAgent/*` provider events the adapter turns into task.* + * runtime events (timelineBypass keeps them out of the parent chat). + * + * WIP, probe-gated: registration is deliberately explicit-signals-only. The + * spec's "provisionally treat unknown foreign thread ids as v2 children" rule + * needs a live wire capture of the packaged binary before it lands — blind + * capture risks eating unrelated traffic. Until then a child whose first + * notification precedes registration passes through as today (no regression + * vs main, which passes everything through). + */ +interface CollabChildAgentState { + readonly agentThreadId: string; + readonly nickname: string | undefined; + readonly role: string | undefined; + readonly agentPath: string | undefined; + readonly depth: number | undefined; + readonly parentThreadId: string | undefined; + /** + * Parent canonical turn active when the child registered. Stamped on every + * synthetic collabAgent/* event so clients can batch a fleet by its spawn + * turn — without it, separate fleets in one thread collapsed into a single + * "direct:no-turn" CTA (review finding). + */ + readonly spawnTurnId: TurnId | undefined; +} + +function readThreadSpawnSource(thread: { readonly source: unknown }): + | { + nickname: string | undefined; + role: string | undefined; + agentPath: string | undefined; + depth: number | undefined; + parentThreadId: string | undefined; + } + | undefined { + const source = thread.source; + if (typeof source !== "object" || source === null || !("subAgent" in source)) { + return undefined; + } + const subAgent = (source as { subAgent: unknown }).subAgent; + if (typeof subAgent !== "object" || subAgent === null || !("thread_spawn" in subAgent)) { + return undefined; + } + const spawn = (subAgent as { thread_spawn: unknown }).thread_spawn; + if (typeof spawn !== "object" || spawn === null) { + return undefined; + } + const record = spawn as Record; + return { + nickname: typeof record.agent_nickname === "string" ? record.agent_nickname : undefined, + role: typeof record.agent_role === "string" ? record.agent_role : undefined, + agentPath: typeof record.agent_path === "string" ? record.agent_path : undefined, + depth: typeof record.depth === "number" ? record.depth : undefined, + parentThreadId: + typeof record.parent_thread_id === "string" ? record.parent_thread_id : undefined, + }; +} + function rememberCollabReceiverTurns( collabReceiverTurns: Map, notification: CodexServerNotification, @@ -723,6 +788,9 @@ 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()); + /** Child provider-thread id → its currently running provider turn id. */ + const collabChildLiveTurnsRef = yield* Ref.make(new Map()); const closedRef = yield* Ref.make(false); // `~` is not shell-expanded when env vars are set via @@ -839,6 +907,221 @@ export const makeCodexSessionRuntime = ( ), ); + /** + * Registers v2 collab children and re-emits their notifications as + * synthetic `collabAgent/*` events for the adapter's task.* synthesis. + * Returns true when the notification was fully handled (must not reach + * parent-timeline mapping). + */ + const interceptCollabChildNotification = (notification: CodexServerNotification) => + Effect.gen(function* () { + // Registration path 1: child thread announces itself with a + // subAgent thread_spawn source. + if (notification.method === "thread/started") { + const thread = notification.params.thread; + const spawn = readThreadSpawnSource(thread); + if (!spawn) { + return false; + } + const spawnTurnId = (yield* Ref.get(sessionRef)).activeTurnId ?? undefined; + const state: CollabChildAgentState = { + agentThreadId: thread.id, + nickname: spawn.nickname ?? thread.agentNickname ?? undefined, + role: spawn.role ?? thread.agentRole ?? undefined, + agentPath: spawn.agentPath, + depth: spawn.depth, + parentThreadId: spawn.parentThreadId ?? thread.parentThreadId ?? undefined, + spawnTurnId, + }; + yield* Ref.update(collabChildAgentsRef, (current) => { + const next = new Map(current); + next.set(thread.id, state); + return next; + }); + yield* emitEvent({ + kind: "notification", + threadId: options.threadId, + method: "collabAgent/started", + ...(state.spawnTurnId ? { turnId: state.spawnTurnId } : {}), + payload: { + agentThreadId: state.agentThreadId, + ...(state.nickname ? { nickname: state.nickname } : {}), + ...(state.role ? { role: state.role } : {}), + ...(state.agentPath ? { agentPath: state.agentPath } : {}), + ...(state.depth !== undefined ? { depth: state.depth } : {}), + ...(state.parentThreadId ? { parentThreadId: state.parentThreadId } : {}), + }, + }); + return true; + } + + // Registration path 2: parent-side subAgentActivity item names the + // child thread (may arrive before or after thread/started). + if ( + (notification.method === "item/started" || notification.method === "item/completed") && + notification.params.item.type === "subAgentActivity" + ) { + const item = notification.params.item; + // Never register the session's ROOT thread as its own child. The + // wire emits subAgentActivity {agentPath: "/root", interacted} + // about the root during collab runs; registering it intercepted + // every subsequent root notification — including the final + // assistant message and turn/completed — so the thread hung + // "working" after all subagents finished (live-probe finding). + const rootProviderThreadId = currentProviderThreadId(yield* Ref.get(sessionRef)); + if ( + item.agentThreadId === rootProviderThreadId || + item.agentPath === "/root" || + item.agentPath === "/" + ) { + return false; + } + const activitySpawnTurnId = (yield* Ref.get(sessionRef)).activeTurnId ?? undefined; + yield* Ref.update(collabChildAgentsRef, (current) => { + if (current.has(item.agentThreadId)) { + return current; + } + const next = new Map(current); + next.set(item.agentThreadId, { + agentThreadId: item.agentThreadId, + nickname: item.agentPath.split("/").findLast((segment) => segment.length > 0), + role: undefined, + agentPath: item.agentPath, + depth: undefined, + parentThreadId: undefined, + spawnTurnId: activitySpawnTurnId, + }); + return next; + }); + const registeredChild = (yield* Ref.get(collabChildAgentsRef)).get(item.agentThreadId); + yield* emitEvent({ + kind: "notification", + threadId: options.threadId, + method: "collabAgent/activity", + ...(registeredChild?.spawnTurnId ? { turnId: registeredChild.spawnTurnId } : {}), + payload: { + agentThreadId: item.agentThreadId, + agentPath: item.agentPath, + activityKind: item.kind, + }, + }); + return true; + } + + // Interception: notifications addressed to a registered child thread + // become agent-scoped synthetic events instead of parent chatter. + const providerConversationId = readNotificationThreadId(notification); + if (!providerConversationId) { + return false; + } + // Belt-and-braces: the root thread's traffic must never be + // intercepted, whatever the registry says. + const interceptRootId = currentProviderThreadId(yield* Ref.get(sessionRef)); + if (providerConversationId === interceptRootId) { + return false; + } + const children = yield* Ref.get(collabChildAgentsRef); + const child = children.get(providerConversationId); + if (!child) { + return false; + } + const childIdentity = { + agentThreadId: child.agentThreadId, + ...(child.nickname ? { nickname: child.nickname } : {}), + ...(child.role ? { role: child.role } : {}), + ...(child.agentPath ? { agentPath: child.agentPath } : {}), + }; + switch (notification.method) { + case "turn/started": { + const childTurnId = + typeof (notification.params as { turn?: { id?: unknown } }).turn?.id === "string" + ? ((notification.params as { turn: { id: string } }).turn.id as string) + : undefined; + if (childTurnId) { + yield* Ref.update(collabChildLiveTurnsRef, (current) => { + const next = new Map(current); + next.set(child.agentThreadId, childTurnId); + return next; + }); + } + yield* emitEvent({ + kind: "notification", + threadId: options.threadId, + ...(child.spawnTurnId ? { turnId: child.spawnTurnId } : {}), + method: "collabAgent/turnStarted", + payload: childIdentity, + }); + return true; + } + case "turn/completed": + yield* Ref.update(collabChildLiveTurnsRef, (current) => { + const next = new Map(current); + next.delete(child.agentThreadId); + return next; + }); + yield* emitEvent({ + kind: "notification", + threadId: options.threadId, + ...(child.spawnTurnId ? { turnId: child.spawnTurnId } : {}), + method: "collabAgent/turnCompleted", + payload: { + ...childIdentity, + turn: notification.params.turn, + }, + }); + return true; + case "thread/status/changed": + yield* emitEvent({ + kind: "notification", + threadId: options.threadId, + ...(child.spawnTurnId ? { turnId: child.spawnTurnId } : {}), + method: "collabAgent/statusChanged", + payload: { + ...childIdentity, + status: notification.params.status, + }, + }); + return true; + case "thread/tokenUsage/updated": + yield* emitEvent({ + kind: "notification", + threadId: options.threadId, + method: "collabAgent/tokenUsage", + payload: { + ...childIdentity, + tokenUsage: notification.params.tokenUsage, + }, + }); + return true; + case "item/started": + case "item/completed": + yield* emitEvent({ + kind: "notification", + threadId: options.threadId, + ...(child.spawnTurnId ? { turnId: child.spawnTurnId } : {}), + method: "collabAgent/item", + payload: { + ...childIdentity, + item: notification.params.item, + }, + }); + return true; + case "thread/closed": + yield* emitEvent({ + kind: "notification", + threadId: options.threadId, + ...(child.spawnTurnId ? { turnId: child.spawnTurnId } : {}), + method: "collabAgent/closed", + payload: childIdentity, + }); + return true; + default: + // Remaining child chatter (name updates, deltas, plan updates) + // stays out of the parent timeline and has no agent mapping yet. + return true; + } + }); + const handleRawNotification = (notification: CodexServerNotification) => Effect.gen(function* () { const payload = notification.params; @@ -857,6 +1140,11 @@ export const makeCodexSessionRuntime = ( return; } + if (yield* interceptCollabChildNotification(notification)) { + yield* Ref.set(collabReceiverTurnsRef, collabReceiverTurns); + return; + } + let requestId: ApprovalRequestId | undefined; let requestKind: ProviderRequestKind | undefined; let turnId = childParentTurnId ?? route.turnId; @@ -1331,6 +1619,21 @@ export const makeCodexSessionRuntime = ( Effect.gen(function* () { const providerThreadId = yield* readProviderThreadId; const session = yield* Ref.get(sessionRef); + // Stop-everything: children are full threads with their own turns; + // interrupting only the parent leaves the fleet running. Interrupt + // each live child turn first, best-effort per child. + const liveChildTurns = yield* Ref.get(collabChildLiveTurnsRef); + yield* Effect.forEach( + Array.from(liveChildTurns.entries()), + ([childThreadId, childTurnId]) => + client + .request("turn/interrupt", { + threadId: childThreadId, + turnId: childTurnId, + }) + .pipe(Effect.ignore), + { concurrency: 8, discard: true }, + ); const effectiveTurnId = turnId ?? session.activeTurnId; if (!effectiveTurnId) { return; diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 909a51a4cf5..f7a7e237a52 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -91,6 +91,7 @@ import { issueAssetUrl } from "./assets/AssetAccess.ts"; import * as PortScanner from "./preview/PortScanner.ts"; import * as WorkspaceEntries from "./workspace/WorkspaceEntries.ts"; import * as WorkspaceFileSystem from "./workspace/WorkspaceFileSystem.ts"; +import { readWorkflowScript } from "./orchestration/workflowScriptQuery.ts"; import * as WorkspacePaths from "./workspace/WorkspacePaths.ts"; import * as VcsStatusBroadcaster from "./vcs/VcsStatusBroadcaster.ts"; import * as VcsProvisioningService from "./vcs/VcsProvisioningService.ts"; @@ -1084,6 +1085,12 @@ const makeWsRpcLayer = ( ), { "rpc.aggregate": "orchestration" }, ), + [ORCHESTRATION_WS_METHODS.getWorkflowScript]: (input) => + observeRpcEffect( + ORCHESTRATION_WS_METHODS.getWorkflowScript, + readWorkflowScript({ scriptPath: input.scriptPath }), + { "rpc.aggregate": "orchestration" }, + ), [ORCHESTRATION_WS_METHODS.getTurnDiff]: (input) => observeRpcEffect( ORCHESTRATION_WS_METHODS.getTurnDiff, diff --git a/apps/web/src/components/AgentsPanel.tsx b/apps/web/src/components/AgentsPanel.tsx new file mode 100644 index 00000000000..169c662e585 --- /dev/null +++ b/apps/web/src/components/AgentsPanel.tsx @@ -0,0 +1,568 @@ +/** + * Agents right-panel surface: the fleet view over the native subagent fold, + * and the ONLY place the roster renders (the chat carries one CTA row per + * spawn batch). + * + * Visualization rules (from live-test feedback): + * - Live work first: running workflows and direct spawns sort above settled. + * - Rows are flat status lines — no expansion, no per-agent tool feeds. The + * row answers "who / what phase / how much"; anything deeper is a future + * drill-in, not an unfold. + * - A settled workflow run collapses to a single summary line; click it to + * show its member list inline (the one allowed toggle — run granularity, + * not agent granularity). + * - Static status dots, DOM-write elapsed timers, plain token counters. + */ +import { useAtomValue } from "@effect/atom-react"; +import type { + AgentPanelModel, + AgentPanelWorkflowGroup, + RuntimeSubagent, +} from "@t3tools/client-runtime/state/subagentRuntime"; +import { + formatSubagentModelLabel, + formatSubagentTokenCount, +} from "@t3tools/client-runtime/state/subagentRuntime"; +import type { EnvironmentId, ThreadId } from "@t3tools/contracts"; +import { Bot, Braces, Check, ChevronDown, ChevronRight, X } from "lucide-react"; +import { useEffect, useRef, useState } from "react"; + +import { cn } from "~/lib/utils"; +import { orchestrationEnvironment } from "~/state/orchestration"; +import { ScrollArea } from "~/components/ui/scroll-area"; + +/** + * In-flight states all present as Working (one steady state, per the + * monitoring-pill design: detail belongs in the activity sub-line, and a + * stalled/waiting/queued subagent is still the fleet doing its job, not a + * user problem). Only settled states differentiate. + */ +const STATUS_VISUALS: Record = { + pending: { dotClass: "bg-info", label: "Working" }, + running: { dotClass: "bg-info", label: "Working" }, + waiting: { dotClass: "bg-info", label: "Working" }, + // Idle reads as settled (muted, not sky): a resting Codex child looks done + // unless resumed — live-test: sky idle dots read as stuck in-progress. + idle: { dotClass: "bg-muted-foreground/50", label: "Idle · resumable" }, + completed: { dotClass: "bg-success", label: "Completed" }, + failed: { dotClass: "bg-destructive", label: "Failed" }, + cancelled: { dotClass: "bg-muted-foreground/60", label: "Stopped" }, + interrupted: { dotClass: "bg-muted-foreground/60", label: "Stopped" }, +}; + +function StatusDot({ status }: { status: RuntimeSubagent["status"] }) { + return ( + + ); +} + +function formatElapsedSeconds(totalSeconds: number): string { + const seconds = Math.max(0, Math.floor(totalSeconds)); + const minutes = Math.floor(seconds / 60); + if (minutes === 0) { + return `${seconds}s`; + } + const hours = Math.floor(minutes / 60); + if (hours === 0) { + return `${minutes}m ${String(seconds % 60).padStart(2, "0")}s`; + } + return `${hours}h ${String(minutes % 60).padStart(2, "0")}m`; +} + +function elapsedBetween(startedAt: string, endIso: string | null): string { + const start = Date.parse(startedAt); + const end = endIso ? Date.parse(endIso) : Date.now(); + if (Number.isNaN(start) || Number.isNaN(end)) { + return ""; + } + return formatElapsedSeconds((end - start) / 1000); +} + +/** + * Elapsed time for the current activation. Live agents self-tick via DOM + * writes (zero React commits per tick); settled agents freeze at completedAt. + */ +function AgentElapsed({ agent }: { agent: RuntimeSubagent }) { + const textRef = useRef(null); + const live = agent.status === "running" || agent.status === "waiting"; + const startedAt = agent.startedAt; + + useEffect(() => { + if (!live || !startedAt) { + return; + } + const update = () => { + if (textRef.current) { + textRef.current.textContent = elapsedBetween(startedAt, null); + } + }; + update(); + const id = setInterval(update, 1000); + return () => clearInterval(id); + }, [live, startedAt]); + + if (!startedAt) { + return null; + } + return ( + + {elapsedBetween(startedAt, live ? null : agent.completedAt)} + + ); +} + +/** + * Status-dependent activity line. Live rows lead with what is happening now; + * settled rows lead with the outcome. Errors are the only inline previews on + * failed rows because they explain a red row at a glance. + */ +function agentActivityText(agent: RuntimeSubagent): string | null { + const live = + agent.status === "running" || agent.status === "pending" || agent.status === "waiting"; + if (live) { + return ( + agent.progress ?? + (agent.lastToolName ? `▸ ${agent.lastToolName}` : null) ?? + agent.result ?? + agent.error + ); + } + return ( + agent.error ?? + agent.result ?? + agent.progress ?? + (agent.lastToolName ? `▸ ${agent.lastToolName}` : null) + ); +} + +/** Flat, non-interactive agent status line. No unfold. */ +function AgentRow({ agent }: { agent: RuntimeSubagent }) { + const visuals = STATUS_VISUALS[agent.status]; + const activity = agentActivityText(agent); + const modelLabel = formatSubagentModelLabel(agent.model, agent.effort); + + return ( +
+
+ + + + + + {agent.title} + {agent.role ? ( + + {agent.role} + + ) : null} + + + {agent.status === "completed" ? ( + + ) : null} + + + {activity ? ( + + {activity} + + ) : null} + + {modelLabel ? {modelLabel} : null} + {agent.usage ? ( + + {modelLabel ? "· " : ""} + {formatSubagentTokenCount(agent.usage.totalTokens)} tok + + ) : null} + {agent.usage?.toolUses !== undefined ? ( + · {agent.usage.toolUses} tools + ) : null} + {agent.activationCount > 1 ? · run {agent.activationCount} : null} + {visuals.label} + + +
+
+ ); +} + +function workflowIsLive(group: AgentPanelWorkflowGroup): boolean { + const status = group.workflow.status; + return ( + status !== "completed" && + status !== "failed" && + status !== "cancelled" && + status !== "interrupted" + ); +} + +function workflowMembers(group: AgentPanelWorkflowGroup): ReadonlyArray { + return [...group.phases.flatMap((phase) => phase.members), ...group.unphasedMembers]; +} + +/** + * Phase rail: the run's shape at a glance. One segment per phase in order, + * separated by chevrons; each segment shows title + one dot per member. + * The whole arc (done → live → pending) is visible without scrolling the + * member list. + */ +function PhaseRail({ group }: { group: AgentPanelWorkflowGroup }) { + if (group.phases.length === 0) { + return null; + } + return ( +
+ {group.phases.map((phase, index) => ( +
+ {index > 0 ? ( + + ) : null} +
+ + {phase.state === "done" ? "✓ " : ""} + {phase.title} + + + {phase.members.length === 0 ? ( + – + ) : ( + phase.members.map((member) => ) + )} + +
+
+ ))} +
+ ); +} + +/** + * Read-only workflow script viewer, fetched through the contained + * getWorkflowScript RPC (never a raw filesystem read from the client). + */ +function WorkflowScriptView({ + environmentId, + threadId, + scriptPath, + onClose, +}: { + environmentId: EnvironmentId; + threadId: ThreadId; + scriptPath: string; + onClose: () => void; +}) { + const result = useAtomValue( + orchestrationEnvironment.workflowScript({ environmentId, input: { threadId, scriptPath } }), + ); + return ( +
+
+ + + {scriptPath.split("/").at(-1)} + + +
+
+ {result._tag === "Success" ? ( + 
+            {result.value.contents}
+            {result.value.truncated ? "\n… (truncated)" : ""}
+
+ ) : result._tag === "Failure" ? ( +

Could not load the script.

+ ) : ( +

Loading…

+ )} +
+
+ ); +} + +/** + * Collapsible phase section (Claude Code Background-tasks pattern): live + * phases open by default, done phases collapsed to header + member dot row. + * User toggles override the default and stick for the phase's lifetime. + */ +function PhaseSection({ phase }: { phase: AgentPanelWorkflowGroup["phases"][number] }) { + const [userOpen, setUserOpen] = useState(null); + const open = userOpen ?? phase.state === "running"; + return ( +
+ + {open ? phase.members.map((member) => ) : null} +
+ ); +} + +/** Live workflow: phase rail + full phase tree. */ +function LiveWorkflowSection({ + group, + environmentId, + threadId, +}: { + group: AgentPanelWorkflowGroup; + environmentId: EnvironmentId | null; + threadId: ThreadId | null; +}) { + const [scriptOpen, setScriptOpen] = useState(false); + const members = workflowMembers(group); + const settled = members.filter( + (member) => + member.status === "completed" || + member.status === "failed" || + member.status === "cancelled" || + member.status === "interrupted", + ).length; + const scriptPath = group.workflow.runHandles?.scriptPath; + const canShowScript = scriptPath !== undefined && environmentId !== null && threadId !== null; + return ( +
+
+ + {group.workflow.workflowName ?? group.workflow.title} + {canShowScript ? ( + + ) : null} + + {settled}/{members.length} settled + +
+ + {scriptOpen && canShowScript ? ( + setScriptOpen(false)} + /> + ) : null} + {group.phases.map((phase) => ( + + ))} + {group.unphasedMembers.map((member) => ( + + ))} + {group.phases.length === 0 && group.unphasedMembers.length === 0 ? ( + + ) : null} +
+ ); +} + +/** + * Settled workflow: one summary line. Click toggles the member list — the + * only expansion in the panel, at run granularity. + */ +function SettledWorkflowSection({ group }: { group: AgentPanelWorkflowGroup }) { + const [open, setOpen] = useState(false); + const members = workflowMembers(group); + const failed = members.filter((member) => member.status === "failed").length; + // Coordinator usage may already aggregate members (panel-footer rule): + // count it only when there are no member rows to sum. + const totalTokens = members.reduce( + (sum, member) => sum + (member.usage?.totalTokens ?? 0), + members.length === 0 ? (group.workflow.usage?.totalTokens ?? 0) : 0, + ); + const elapsed = + group.workflow.startedAt && group.workflow.completedAt + ? elapsedBetween(group.workflow.startedAt, group.workflow.completedAt) + : null; + return ( +
+ + {open ? ( +
+ {members.map((member) => ( + + ))} +
+ ) : null} +
+ ); +} + +export function AgentsPanel({ + model, + environmentId = null, + threadId = null, +}: { + model: AgentPanelModel; + environmentId?: EnvironmentId | null; + threadId?: ThreadId | null; +}) { + if (!model.hasAgents) { + return ( +
+ +

No agents yet

+

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

+
+ ); + } + + const liveWorkflows = model.workflows.filter(workflowIsLive); + const settledWorkflows = model.workflows.filter((group) => !workflowIsLive(group)); + const liveDirect = model.directAgents.filter( + (agent) => + agent.status === "running" || agent.status === "pending" || agent.status === "waiting", + ); + const settledDirect = model.directAgents.filter( + (agent) => + agent.status !== "running" && agent.status !== "pending" && agent.status !== "waiting", + ); + + return ( +
+ +
+ {liveWorkflows.map((group) => ( + + ))} + {liveDirect.length > 0 ? ( +
+
+ Direct spawns +
+ {liveDirect.map((agent) => ( + + ))} +
+ ) : null} + {settledWorkflows.length > 0 || settledDirect.length > 0 ? ( +
+
+ Earlier +
+ {settledWorkflows.map((group) => ( + + ))} + {settledDirect.map((agent) => ( + + ))} +
+ ) : null} +
+ +
+ + {model.runningCount + model.waitingCount > 0 ? ( + + ● {model.runningCount + model.waitingCount} working + + ) : null} + {model.idleCount > 0 ? {model.idleCount} idle : null} + {model.settledCount > 0 ? {model.settledCount} settled : null} + + Σ {formatSubagentTokenCount(model.totalTokens)} tok +
+
+ ); +} diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 2b9eda1a787..7dd878184b8 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -139,6 +139,11 @@ import { usePreviewMiniPlayerStore, } from "../previewMiniPlayerStore"; import { RightPanelTabs } from "./RightPanelTabs"; +import { AgentsPanel } from "./AgentsPanel"; +import { + deriveAgentPanelModel, + foldSubagentActivities, +} from "@t3tools/client-runtime/state/subagentRuntime"; import { DiffWorkerPoolProvider } from "./DiffWorkerPoolProvider"; import { BranchToolbar } from "./BranchToolbar"; import { resolveShortcutCommand, shortcutLabelForCommand } from "../keybindings"; @@ -2050,6 +2055,18 @@ function ChatViewContent(props: ChatViewProps) { const phase = derivePhase(activeThread?.session ?? null); const threadActivities = activeThread?.activities ?? EMPTY_ACTIVITIES; const workLogEntries = useMemo(() => deriveWorkLogEntries(threadActivities), [threadActivities]); + // Native subagent fold: memoized by activity-list identity, shared by the + // Agents surface, live strip, and workflow cards. v2Projection is null + // until orchestration-v2 lands (source precedence lives in the derive). + // sessionLive derives interruption for agents orphaned by session death. + const agentSessionLive = phase !== "disconnected"; + const agentPanelModel = useMemo( + () => + deriveAgentPanelModel({ + agents: foldSubagentActivities(threadActivities, { sessionLive: agentSessionLive }), + }), + [agentSessionLive, threadActivities], + ); const pendingApprovals = useMemo( () => derivePendingApprovals(threadActivities), [threadActivities], @@ -3132,6 +3149,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; @@ -4165,6 +4186,80 @@ function ChatViewContent(props: ChatViewProps) { switchGitRef, updateThreadMetadata, ]); + // Background work (subagent fleets, workflow runs, watch loops) can outlive + // the turn; once it settles, the composer stop button is gone, so this + // banner is the only visible stop affordance. Stop routes through the + // stop-everything interrupt: it kills every live background task before + // interrupting, and works by session, so no active turn is needed. + const activeBackgroundLiveness = + !isWorking && activeThread ? (activeThreadShell?.backgroundLiveness ?? null) : null; + const [isStoppingBackgroundWork, setIsStoppingBackgroundWork] = useState(false); + useEffect(() => { + // "Stopping..." holds until the liveness clears; the interrupt command + // returning only means the request was accepted. + if (activeBackgroundLiveness === null) { + setIsStoppingBackgroundWork(false); + } + }, [activeBackgroundLiveness]); + useEffect(() => { + // Per-thread state: switching threads while A's stop is pending must not + // disable B's Stop button (review finding). + setIsStoppingBackgroundWork(false); + }, [activeThreadId]); + const handleStopBackgroundWork = useCallback(async () => { + if (!activeThread) return; + setIsStoppingBackgroundWork(true); + const result = await interruptThreadTurn({ + environmentId, + input: buildThreadTurnInterruptInput(activeThread), + }); + if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { + setIsStoppingBackgroundWork(false); + const error = squashAtomCommandFailure(result); + setThreadError( + activeThread.id, + error instanceof Error ? error.message : "Failed to stop background work.", + ); + } + }, [activeThread, environmentId, interruptThreadTurn, setThreadError]); + const backgroundLivenessBannerItem = useMemo(() => { + if (activeBackgroundLiveness === null || !activeThread) { + return null; + } + const working = activeBackgroundLiveness === "working"; + const liveCount = agentPanelModel.liveCount; + return { + id: `background-liveness:${activeThread.id}`, + variant: "default", + icon: ( +