From 12b011144b57753938a5438466fb738ef163b0fc Mon Sep 17 00:00:00 2001 From: Shivam Sharma <91240327+shivamhwp@users.noreply.github.com> Date: Mon, 27 Jul 2026 15:12:53 +0530 Subject: [PATCH] feat(orchestration-v2): add subagent observability data model MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduces the schema, storage, and migration for richer subagent tracking. Nothing produces the new data yet — adapters populate neutral defaults so every subagent row remains valid — which keeps this change reviewable as a pure data-model step. Contracts gain a reusable subagent identity: per-activation records (SubagentActivationId, OrchestrationV2SubagentActivation), cumulative usage, a role, recent activity, and workflow grouping. A subagent may now also rest at "idle" between activations. Because a subagent status is not a turn item status, the two are bridged by an explicit mapping rather than a copy. The Record type makes that total: adding a subagent status without giving it a timeline meaning is a compile error. Without it, a producer can emit a turn item whose own schema cannot decode it, which is unrecoverable — the projection rebuild then fails on every subsequent startup and the server cannot boot. Migration 043 backfills existing rows so old events stay readable. Its riskiest path is a partially-migrated row, where already-present JSON is round-tripped through COALESCE; a test pins that structured values keep their JSON subtype rather than being rewritten as quoted strings. Also narrows event-envelope attribution in ProviderEventIngestor: a payload carrying an explicit null run or node id now keeps that null instead of inheriting the ambient run. Activations legitimately have no run id, and the envelope is persistence metadata only — routing reads payload fields — but this does change stored metadata for existing event types that already pass explicit nulls. --- .../orchestration-v2/Adapters/AcpAdapterV2.ts | 8 + .../Adapters/ClaudeAdapterV2.ts | 8 + .../Adapters/CodexAdapterV2.ts | 18 +- .../Adapters/CursorAdapterV2.ts | 8 + .../Adapters/OpenCodeAdapterV2.ts | 8 + .../FoundationPersistence.test.ts | 80 +++++++ .../src/orchestration-v2/Orchestrator.ts | 90 ++++++- .../orchestration-v2/ProjectionMaintenance.ts | 1 + .../src/orchestration-v2/ProjectionStore.ts | 71 +++++- .../src/orchestration-v2/ProviderAdapter.ts | 7 + .../ProviderEventIngestor.test.ts | 81 +++++++ .../orchestration-v2/ProviderEventIngestor.ts | 14 +- .../ProviderTurnControlService.test.ts | 1 + .../RunExecutionService.test.ts | 16 ++ .../orchestration-v2/RunExecutionService.ts | 2 + .../SubagentObservability.test.ts | 23 ++ .../orchestration-v2/SubagentObservability.ts | 13 + .../ThreadForkService.test.ts | 1 + apps/server/src/persistence/Migrations.ts | 2 + .../035_036_OrchestrationV2.test.ts | 7 +- ...chestrationV2SubagentObservability.test.ts | 222 ++++++++++++++++++ ...44_OrchestrationV2SubagentObservability.ts | 50 ++++ apps/web/src/test-fixtures.ts | 1 + .../state/orchestrationV2Projection.test.ts | 1 + .../src/state/orchestrationV2Projection.ts | 5 + .../src/state/orchestrationV2TestFixtures.ts | 1 + packages/contracts/src/baseSchemas.ts | 2 + .../contracts/src/orchestrationV2.test.ts | 157 ++++++++++++- packages/contracts/src/orchestrationV2.ts | 143 +++++++++++ 29 files changed, 1031 insertions(+), 10 deletions(-) create mode 100644 apps/server/src/orchestration-v2/SubagentObservability.test.ts create mode 100644 apps/server/src/orchestration-v2/SubagentObservability.ts create mode 100644 apps/server/src/persistence/Migrations/044_OrchestrationV2SubagentObservability.test.ts create mode 100644 apps/server/src/persistence/Migrations/044_OrchestrationV2SubagentObservability.ts diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.ts index aa3f8bb3364..2ae679fff3e 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.ts @@ -1963,6 +1963,14 @@ export function makeAcpAdapterV2(options: AcpAdapterV2Options): ProviderAdapterV prompt: update.prompt, title: update.title, model: update.model, + kind: "subagent", + role: { name: "general-purpose", source: "app_default" }, + usage: null, + currentActivationId: null, + activationCount: 1, + workflow: null, + workflowMembership: null, + recentActivity: [], result: null, startedAt: now, }), diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 11132ae7b82..c06c11e7c36 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -2321,6 +2321,14 @@ export function makeClaudeAdapterV2( prompt: input.prompt ?? "", title: input.title ?? null, model: input.context.input.modelSelection.model, + kind: "subagent" as const, + role: { name: "general-purpose", source: "app_default" as const }, + usage: null, + currentActivationId: null, + activationCount: 1, + workflow: null, + workflowMembership: null, + recentActivity: [], result: null, startedAt: now, }), diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 0ffd4232111..39b5a27b381 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -1,4 +1,9 @@ -import { CodexSettings, defaultInstanceIdForDriver, ProviderDriverKind } from "@t3tools/contracts"; +import { + CodexSettings, + defaultInstanceIdForDriver, + orchestrationV2SubagentStatusAsTurnItemStatus, + ProviderDriverKind, +} from "@t3tools/contracts"; import { HostProcessEnvironment } from "@t3tools/shared/hostProcess"; import { getModelSelectionStringOptionValue } from "@t3tools/shared/model"; import { resolveSpawnCommand } from "@t3tools/shared/shell"; @@ -100,6 +105,7 @@ import { makeSubagentConversationArtifacts, subagentThreadTitle, } from "../SubagentProjection.ts"; +import { defaultSubagentRole } from "../SubagentObservability.ts"; const CODEX_PROVIDER = ProviderDriverKind.make("codex"); export const CODEX_DRIVER_KIND = CODEX_PROVIDER; @@ -1721,7 +1727,7 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi nativeItemRef: task.nativeTaskRef, parentItemId: null, ordinal: input.subagent.turnItemOrdinal, - status: task.status, + status: orchestrationV2SubagentStatusAsTurnItemStatus[task.status], title: task.title, startedAt: task.startedAt, completedAt: task.completedAt, @@ -1968,8 +1974,16 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi prompt: input.prompt, title: input.title, model: input.model, + kind: "subagent", + role: defaultSubagentRole(), status: "running", result: null, + usage: null, + currentActivationId: null, + activationCount: 1, + workflow: null, + workflowMembership: null, + recentActivity: [], startedAt: now, completedAt: null, updatedAt: now, diff --git a/apps/server/src/orchestration-v2/Adapters/CursorAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CursorAdapterV2.ts index 30b7a36c4d9..01b144a0ab5 100644 --- a/apps/server/src/orchestration-v2/Adapters/CursorAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CursorAdapterV2.ts @@ -1497,6 +1497,14 @@ export function makeCursorAdapterV2( prompt: args.prompt, title: args.description, model: args.model ?? input.context.input.modelSelection.model, + kind: "subagent", + role: { name: "general-purpose", source: "app_default" }, + usage: null, + currentActivationId: null, + activationCount: 1, + workflow: null, + workflowMembership: null, + recentActivity: [], result: null, startedAt: now, }), diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.ts index 59ca6efb355..8254b8ea1e3 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.ts @@ -1210,6 +1210,14 @@ export function makeOpenCodeAdapterV2(options: OpenCodeAdapterV2Options): Provid prompt, title, model: context.model, + kind: "subagent", + role: { name: "general-purpose", source: "app_default" }, + usage: null, + currentActivationId: null, + activationCount: 1, + workflow: null, + workflowMembership: null, + recentActivity: [], status: subagentStatus, result: context.result, startedAt: context.startedAt, diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index c7303513308..1833f1705ea 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -19,6 +19,7 @@ import { ProviderThreadId, RunAttemptId, RunId, + SubagentActivationId, ThreadId, TurnItemId, } from "@t3tools/contracts"; @@ -269,6 +270,7 @@ it.layer(TestLayer)("orchestration V2 foundation persistence", (it) => { const eventSink = yield* EventSinkV2; const projectionStore = yield* ProjectionStoreV2; const maintenance = yield* ProjectionMaintenanceV2; + const sql = yield* SqlClient.SqlClient; const now = yield* DateTime.now; const parentThreadId = ThreadId.make("thread:foundation-cross-thread:parent"); const childThreadId = ThreadId.make("thread:foundation-cross-thread:child"); @@ -276,6 +278,9 @@ it.layer(TestLayer)("orchestration V2 foundation persistence", (it) => { "provider-thread:foundation-cross-thread:child", ); const subagentId = NodeId.make("subagent:foundation-cross-thread"); + const activationId = SubagentActivationId.make( + "subagent:foundation-cross-thread:activation:1", + ); const spawnTransferId = ContextTransferId.make("transfer:foundation-cross-thread:spawn"); const resultTransferId = ContextTransferId.make("transfer:foundation-cross-thread:result"); const parentThread = makeThread(parentThreadId, now); @@ -324,8 +329,37 @@ it.layer(TestLayer)("orchestration V2 foundation persistence", (it) => { prompt: "Inspect the child flow", title: "Cross-thread child", model: modelSelection.model, + kind: "subagent", + role: { name: "general-purpose", source: "app_default" }, status: "completed", result: "done", + usage: null, + currentActivationId: null, + activationCount: 1, + workflow: null, + workflowMembership: null, + recentActivity: [], + startedAt: now, + completedAt: now, + updatedAt: now, + }, + }, + { + id: EventId.make("event:foundation-cross-thread:subagent-activation"), + type: "subagent-activation.updated", + threadId: parentThreadId, + nodeId: subagentId, + providerInstanceId, + occurredAt: now, + payload: { + id: activationId, + threadId: parentThreadId, + subagentId, + runId: null, + providerTurnId: null, + ordinal: 1, + status: "completed", + usage: { totalTokens: 240, toolUses: 2 }, startedAt: now, completedAt: now, updatedAt: now, @@ -426,11 +460,57 @@ it.layer(TestLayer)("orchestration V2 foundation persistence", (it) => { [childProviderThreadId], ); assert.equal(child.thread.activeProviderThreadId, childProviderThreadId); + assert.deepEqual( + parent.subagentActivations.map((activation) => ({ + id: activation.id, + totalTokens: activation.usage?.totalTokens, + })), + [ + { + id: activationId, + totalTokens: 240, + }, + ], + ); }); yield* assertCrossThreadProjection; assert.isTrue((yield* maintenance.verify).valid); + yield* sql` + INSERT INTO orchestration_v2_projection_subagent_activations ( + activation_id, + thread_id, + subagent_id, + run_id, + provider_turn_id, + ordinal, + status, + started_at, + completed_at, + updated_at, + payload_json + ) + VALUES ( + 'activation:foundation-cross-thread:stale', + ${parentThreadId}, + ${subagentId}, + NULL, + NULL, + 99, + 'completed', + NULL, + NULL, + '2026-07-26T00:00:00.000Z', + '{}' + ) + `; assert.isTrue((yield* maintenance.rebuild).valid); + const staleActivationRows = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count + FROM orchestration_v2_projection_subagent_activations + WHERE activation_id = 'activation:foundation-cross-thread:stale' + `; + assert.equal(staleActivationRows[0]?.count, 0); yield* assertCrossThreadProjection; }), ); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 8292371fa69..f9280a13cff 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -18,6 +18,7 @@ import { type OrchestrationV2ThreadShellSnapshot, type OrchestrationV2StoredEvent, type OrchestrationV2Subagent, + type OrchestrationV2SubagentActivation, type OrchestrationV2ThreadProjection, type OrchestrationV2TurnItem, ProviderInstanceId, @@ -62,6 +63,7 @@ import { subagentResultForRun, subagentThreadTitle, } from "./SubagentProjection.ts"; +import { defaultSubagentRole, subagentActivationId } from "./SubagentObservability.ts"; import { ThreadForkServiceV2 } from "./ThreadForkService.ts"; export class OrchestratorDispatchError extends Schema.TaggedErrorClass()( @@ -286,7 +288,10 @@ function isTerminalDelegatedTaskStatus(status: OrchestrationV2Subagent["status"] function delegatedTaskTerminalStatus( status: OrchestrationV2Run["status"], -): OrchestrationV2Subagent["status"] | null { +): Extract< + OrchestrationV2Subagent["status"], + "completed" | "failed" | "cancelled" | "interrupted" +> | null { switch (status) { case "completed": case "failed": @@ -304,6 +309,12 @@ function delegatedTaskTerminalStatus( } } +const isTerminalSubagentActivation = (activation: OrchestrationV2SubagentActivation) => + activation.status === "completed" || + activation.status === "failed" || + activation.status === "cancelled" || + activation.status === "interrupted"; + function nextQueuedRun( projection: OrchestrationV2ThreadProjection, ): OrchestrationV2Run | undefined { @@ -3849,6 +3860,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio runtimeMode: command.runtimeMode, interactionMode: command.interactionMode, }; + const activationId = subagentActivationId(taskNodeId, 1); const task: OrchestrationV2Subagent = { id: taskNodeId, threadId: command.parentThreadId, @@ -3865,8 +3877,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio title: command.title ?? null, model: command.modelSelection.model, ...(command.completionWake === undefined ? {} : { completionWake: command.completionWake }), + kind: "subagent", + role: defaultSubagentRole("delegated-worker"), status: "running", result: null, + usage: null, + currentActivationId: activationId, + activationCount: 1, + workflow: null, + workflowMembership: null, + recentActivity: [], startedAt: now, completedAt: null, updatedAt: now, @@ -3889,6 +3909,19 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio completedAt: null, }; const parentProviderTurn = providerTurnForRun(parentProjection, parentRun); + const activation = { + id: activationId, + threadId: command.parentThreadId, + subagentId: taskNodeId, + runId: parentRun.id, + providerTurnId: parentProviderTurn?.id ?? null, + ordinal: 1, + status: "running", + usage: null, + startedAt: now, + completedAt: null, + updatedAt: now, + } satisfies OrchestrationV2SubagentActivation; const taskTurnItem: OrchestrationV2TurnItem = { id: taskTurnItemId, threadId: command.parentThreadId, @@ -3943,6 +3976,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio occurredAt: now, payload: task, }); + yield* emitEvent({ + type: "subagent-activation.updated", + threadId: command.parentThreadId, + runId: parentRun.id, + nodeId: taskNodeId, + driver: targetAdapter.driver, + providerInstanceId: command.modelSelection.instanceId, + occurredAt: now, + payload: activation, + }); yield* emitEvent({ type: "turn-item.updated", threadId: command.parentThreadId, @@ -5236,9 +5279,41 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio providerThreadId: childRun.providerThreadId, status: terminalStatus, result: result.text, + currentActivationId: null, completedAt: now, updatedAt: now, }; + const currentActivation = + task.currentActivationId === null + ? undefined + : parentProjection.subagentActivations.find( + (activation) => activation.id === task.currentActivationId, + ); + const terminalActivation = + task.currentActivationId === null || currentActivation?.status === terminalStatus + ? null + : currentActivation === undefined + ? ({ + id: task.currentActivationId, + threadId: task.threadId, + subagentId: task.id, + runId: task.runId, + providerTurnId: null, + ordinal: Math.max(1, task.activationCount), + status: terminalStatus, + usage: task.usage, + startedAt: task.startedAt, + completedAt: now, + updatedAt: now, + } satisfies OrchestrationV2SubagentActivation) + : isTerminalSubagentActivation(currentActivation) + ? null + : ({ + ...currentActivation, + status: terminalStatus, + completedAt: now, + updatedAt: now, + } satisfies OrchestrationV2SubagentActivation); const resultTransferId = yield* idAllocator.allocate.contextTransfer({ sourceThreadId: childThreadId, targetThreadId: parentThreadId, @@ -5323,6 +5398,19 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio occurredAt: now, payload: updatedTask, }, + ...(terminalActivation === null + ? [] + : [ + { + type: "subagent-activation.updated" as const, + threadId: parentThreadId, + ...(task.runId === null ? {} : { runId: task.runId }), + nodeId: task.id, + driver: task.driver, + occurredAt: now, + payload: terminalActivation, + }, + ]), ...(parentNode === undefined ? [] : [ diff --git a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts index 84260f5ac8a..1d26c8b574f 100644 --- a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts +++ b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts @@ -152,6 +152,7 @@ export const layer: Layer.Layer< yield* sql`DELETE FROM orchestration_v2_projection_provider_threads`; yield* sql`DELETE FROM orchestration_v2_projection_provider_session_bindings`; yield* sql`DELETE FROM orchestration_v2_projection_provider_sessions`; + yield* sql`DELETE FROM orchestration_v2_projection_subagent_activations`; yield* sql`DELETE FROM orchestration_v2_projection_subagents`; yield* sql`DELETE FROM orchestration_v2_projection_nodes`; yield* sql`DELETE FROM orchestration_v2_projection_run_attempts`; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 0b21547e6eb..6c45ca903bc 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -25,6 +25,7 @@ import { OrchestrationV2RunJson as OrchestrationV2RunJsonSchema, OrchestrationV2RuntimeRequestJson as OrchestrationV2RuntimeRequestJsonSchema, OrchestrationV2SubagentJson as OrchestrationV2SubagentJsonSchema, + OrchestrationV2SubagentActivationJson as OrchestrationV2SubagentActivationJsonSchema, OrchestrationV2TurnItemJson as OrchestrationV2TurnItemJsonSchema, RunId, ThreadId, @@ -121,7 +122,7 @@ export class ProjectionStoreV2 extends Context.Service(items: ReadonlyArray, next: T): Array { const index = items.findIndex((item) => item.id === next.id); @@ -143,6 +144,7 @@ export function emptyProjection( attempts: [], nodes: [], subagents: [], + subagentActivations: [], providerSessions: [], providerThreads: [], providerTurns: [], @@ -212,6 +214,11 @@ export function applyToProjection( ...base, subagents: upsertById(base.subagents, event.payload), }; + case "subagent-activation.updated": + return { + ...base, + subagentActivations: upsertById(base.subagentActivations, event.payload), + }; case "provider-session.attached": case "provider-session.updated": return { @@ -423,6 +430,9 @@ const encodeNodePayload = Schema.encodeEffect( const encodeSubagentPayload = Schema.encodeEffect( Schema.fromJsonString(OrchestrationV2SubagentJsonSchema), ); +const encodeSubagentActivationPayload = Schema.encodeEffect( + Schema.fromJsonString(OrchestrationV2SubagentActivationJsonSchema), +); const encodeProviderSessionPayload = Schema.encodeEffect( Schema.fromJsonString(OrchestrationV2ProviderSessionJsonSchema), ); @@ -467,6 +477,10 @@ const decodeNodePayload = (json: string) => Schema.decodeUnknownEffect(Schema.fromJsonString(OrchestrationV2ExecutionNodeJsonSchema))(json); const decodeSubagentPayload = (json: string) => Schema.decodeUnknownEffect(Schema.fromJsonString(OrchestrationV2SubagentJsonSchema))(json); +const decodeSubagentActivationPayload = (json: string) => + Schema.decodeUnknownEffect(Schema.fromJsonString(OrchestrationV2SubagentActivationJsonSchema))( + json, + ); const decodeProviderSessionPayload = (json: string) => Schema.decodeUnknownEffect(Schema.fromJsonString(OrchestrationV2ProviderSessionJsonSchema))(json); const decodeProviderThreadPayload = (json: string) => @@ -1266,6 +1280,51 @@ export const layer: Layer.Layer = `; break; } + case "subagent-activation.updated": { + const payloadJson = yield* encodeSubagentActivationPayload(event.payload); + const payload = parseEncodedPayload(payloadJson); + yield* sql` + INSERT INTO orchestration_v2_projection_subagent_activations ( + activation_id, + thread_id, + subagent_id, + run_id, + provider_turn_id, + ordinal, + status, + started_at, + completed_at, + updated_at, + payload_json + ) + VALUES ( + ${event.payload.id}, + ${event.payload.threadId}, + ${event.payload.subagentId}, + ${event.payload.runId}, + ${event.payload.providerTurnId}, + ${event.payload.ordinal}, + ${event.payload.status}, + ${nullableStringField(payload, "startedAt")}, + ${nullableStringField(payload, "completedAt")}, + ${stringField(payload, "updatedAt")}, + ${payloadJson} + ) + ON CONFLICT(activation_id) + DO UPDATE SET + thread_id = excluded.thread_id, + subagent_id = excluded.subagent_id, + run_id = excluded.run_id, + provider_turn_id = excluded.provider_turn_id, + ordinal = excluded.ordinal, + status = excluded.status, + started_at = excluded.started_at, + completed_at = excluded.completed_at, + updated_at = excluded.updated_at, + payload_json = excluded.payload_json + `; + break; + } case "provider-session.attached": case "provider-session.updated": { const payloadJson = yield* encodeProviderSessionPayload(event.payload); @@ -1840,6 +1899,7 @@ export const layer: Layer.Layer = attemptRows, nodeRows, subagentRows, + subagentActivationRows, providerSessionRows, providerThreadRows, providerTurnRows, @@ -1877,6 +1937,12 @@ export const layer: Layer.Layer = WHERE thread_id = ${threadId} ORDER BY COALESCE(started_at, ''), subagent_id ASC `, + sql` + SELECT payload_json + FROM orchestration_v2_projection_subagent_activations + WHERE thread_id = ${threadId} + ORDER BY subagent_id ASC, ordinal ASC + `, sql` SELECT sessions.payload_json FROM orchestration_v2_projection_provider_sessions AS sessions @@ -1963,6 +2029,7 @@ export const layer: Layer.Layer = attempts, nodes, subagents, + subagentActivations, providerSessions, providerThreads, providerTurns, @@ -1979,6 +2046,7 @@ export const layer: Layer.Layer = decodeRows(decodeRunAttemptPayload, threadId)(attemptRows), decodeRows(decodeNodePayload, threadId)(nodeRows), decodeRows(decodeSubagentPayload, threadId)(subagentRows), + decodeRows(decodeSubagentActivationPayload, threadId)(subagentActivationRows), decodeRows(decodeProviderSessionPayload, threadId)(providerSessionRows), decodeRows(decodeProviderThreadPayload, threadId)(providerThreadRows), decodeRows(decodeProviderTurnPayload, threadId)(providerTurnRows), @@ -1998,6 +2066,7 @@ export const layer: Layer.Layer = attempts, nodes, subagents, + subagentActivations, providerSessions, providerThreads, providerTurns, diff --git a/apps/server/src/orchestration-v2/ProviderAdapter.ts b/apps/server/src/orchestration-v2/ProviderAdapter.ts index d8f3595bcf3..c64db5e8b6c 100644 --- a/apps/server/src/orchestration-v2/ProviderAdapter.ts +++ b/apps/server/src/orchestration-v2/ProviderAdapter.ts @@ -15,6 +15,7 @@ import { OrchestrationV2ProviderTurn, OrchestrationV2RuntimeRequest, OrchestrationV2Subagent, + OrchestrationV2SubagentActivation, OrchestrationV2TurnItem, ProviderApprovalDecision, ProviderInteractionMode, @@ -30,6 +31,7 @@ import { RunAttemptId, RunId, ThreadId, + TurnItemId, } from "@t3tools/contracts"; import * as Context from "effect/Context"; import * as Schema from "effect/Schema"; @@ -103,6 +105,11 @@ export const ProviderAdapterV2Event = Schema.Union([ driver: ProviderDriverKind, subagent: OrchestrationV2Subagent, }), + Schema.Struct({ + type: Schema.Literal("subagent_activation.updated"), + driver: ProviderDriverKind, + activation: OrchestrationV2SubagentActivation, + }), Schema.Struct({ type: Schema.Literal("message.updated"), driver: ProviderDriverKind, diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts index ea2d2123e31..f663b8e4cd3 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts @@ -8,6 +8,8 @@ import { type OrchestrationV2ProviderThread, ProviderDriverKind, ProviderInstanceId, + RunId, + SubagentActivationId, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; @@ -367,4 +369,83 @@ layer("ProviderEventIngestorV2", (it) => { assert.equal(messageEvents[0]?.threadId, childThreadId); }), ); + + it.effect("does not replace explicit null run and node overrides with ambient ids", () => + Effect.gen(function* () { + const now = yield* DateTime.now; + const ingestor = yield* ProviderEventIngestorV2; + const idAllocator = yield* IdAllocatorV2; + const projectId = yield* idAllocator.allocate.project({ + fixtureName: "provider-event-null-overrides", + }); + const threadId = yield* idAllocator.allocate.thread({ + fixtureName: "provider-event-null-overrides", + projectId, + }); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + const ambientRunId = RunId.make("run:provider-event-null-overrides"); + const ambientNodeId = NodeId.make("node:provider-event-null-overrides"); + const subagentId = NodeId.make("node:provider-event-null-overrides:subagent"); + + const activationEvents = yield* ingestor.normalize({ + providerSessionId, + providerInstanceId: modelSelection.instanceId, + threadId, + runId: ambientRunId, + nodeId: ambientNodeId, + event: { + type: "subagent_activation.updated", + driver: CODEX_DRIVER, + activation: { + id: SubagentActivationId.make( + "node:provider-event-null-overrides:subagent:activation:1", + ), + threadId, + subagentId, + runId: null, + providerTurnId: null, + ordinal: 1, + status: "completed", + usage: null, + startedAt: now, + completedAt: now, + updatedAt: now, + }, + }, + }); + const messageEvents = yield* ingestor.normalize({ + providerSessionId, + providerInstanceId: modelSelection.instanceId, + threadId, + runId: ambientRunId, + nodeId: ambientNodeId, + event: { + type: "message.updated", + driver: CODEX_DRIVER, + message: { + createdBy: "agent", + creationSource: "provider", + id: MessageId.make("message:provider-event-null-overrides"), + threadId, + runId: null, + nodeId: null, + role: "assistant", + text: "Detached provider message", + attachments: [], + streaming: false, + createdAt: now, + updatedAt: now, + }, + }, + }); + + assert.isUndefined(activationEvents[0]?.runId); + assert.equal(activationEvents[0]?.nodeId, subagentId); + assert.isUndefined(messageEvents[0]?.runId); + assert.isUndefined(messageEvents[0]?.nodeId); + }), + ); }); diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts index 91e4a974390..f921472d81c 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts @@ -125,8 +125,8 @@ export const layer: Layer.Layer { + it("marks a fallback role as app-provided so provider roles stay distinguishable", () => { + expect(defaultSubagentRole()).toEqual({ name: "general-purpose", source: "app_default" }); + expect(defaultSubagentRole("workflow-worker")).toEqual({ + name: "workflow-worker", + source: "app_default", + }); + }); + + it("derives activation ids that are stable and unique per ordinal", () => { + const subagentId = NodeId.make("node:provider:codex:native-item:call_abc"); + expect(subagentActivationId(subagentId, 1)).toBe(`${subagentId}:activation:1`); + // Stable derivation means a replayed activation updates its own row rather + // than inserting a duplicate. + expect(subagentActivationId(subagentId, 2)).toBe(subagentActivationId(subagentId, 2)); + expect(subagentActivationId(subagentId, 1)).not.toBe(subagentActivationId(subagentId, 2)); + }); +}); diff --git a/apps/server/src/orchestration-v2/SubagentObservability.ts b/apps/server/src/orchestration-v2/SubagentObservability.ts new file mode 100644 index 00000000000..1da0281843d --- /dev/null +++ b/apps/server/src/orchestration-v2/SubagentObservability.ts @@ -0,0 +1,13 @@ +import { + SubagentActivationId, + type NodeId, + type OrchestrationV2SubagentRole, +} from "@t3tools/contracts"; + +export const defaultSubagentRole = (name = "general-purpose"): OrchestrationV2SubagentRole => ({ + name, + source: "app_default", +}); + +export const subagentActivationId = (subagentId: NodeId, ordinal: number) => + SubagentActivationId.make(`${subagentId}:activation:${ordinal}`); diff --git a/apps/server/src/orchestration-v2/ThreadForkService.test.ts b/apps/server/src/orchestration-v2/ThreadForkService.test.ts index 410ec53b576..ab49188f8d1 100644 --- a/apps/server/src/orchestration-v2/ThreadForkService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadForkService.test.ts @@ -92,6 +92,7 @@ it.effect("keeps a fork awake when its source thread is snoozed", () => attempts: [], nodes: [], subagents: [], + subagentActivations: [], providerSessions: [], providerThreads: [], providerTurns: [], diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 05ff25d81a2..8bddf734003 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -56,6 +56,7 @@ import Migration0040 from "./Migrations/040_ApplicationEventSource.ts"; import Migration0041 from "./Migrations/041_OrchestrationV2EffectCancellation.ts"; import Migration0042 from "./Migrations/042_ScheduledTasks.ts"; import Migration0043 from "./Migrations/043_LegacyV1ImportState.ts"; +import Migration0044 from "./Migrations/044_OrchestrationV2SubagentObservability.ts"; /** * Migration loader with all migrations defined inline. @@ -111,6 +112,7 @@ export const migrationEntries = [ [41, "OrchestrationV2EffectCancellation", Migration0041], [42, "ScheduledTasks", Migration0042], [43, "LegacyV1ImportState", Migration0043], + [44, "OrchestrationV2SubagentObservability", Migration0044], ] as const; export const makeMigrationLoader = (throughId?: number) => diff --git a/apps/server/src/persistence/Migrations/035_036_OrchestrationV2.test.ts b/apps/server/src/persistence/Migrations/035_036_OrchestrationV2.test.ts index 5b217fb9c70..6b903e524bc 100644 --- a/apps/server/src/persistence/Migrations/035_036_OrchestrationV2.test.ts +++ b/apps/server/src/persistence/Migrations/035_036_OrchestrationV2.test.ts @@ -13,7 +13,7 @@ layer("035_036_OrchestrationV2", (it) => { Effect.sync(() => { assert.deepStrictEqual( migrationEntries.map(([id]) => id), - Array.from({ length: 43 }, (_, index) => index + 1), + Array.from({ length: 44 }, (_, index) => index + 1), ); }), ); @@ -122,7 +122,7 @@ it.effect("upgrades a database already at released main migration 034", () => assert.ok(snoozeColumns.some((column) => column.name === "snoozed_until")); assert.ok(snoozeColumns.some((column) => column.name === "snoozed_at")); - yield* runMigrations({ toMigrationInclusive: 43 }); + yield* runMigrations({ toMigrationInclusive: 44 }); const migrations = yield* sql<{ readonly migration_id: number; @@ -130,7 +130,7 @@ it.effect("upgrades a database already at released main migration 034", () => }>` SELECT migration_id, name FROM effect_sql_migrations - WHERE migration_id BETWEEN 34 AND 43 + WHERE migration_id BETWEEN 34 AND 44 ORDER BY migration_id `; assert.deepStrictEqual( @@ -146,6 +146,7 @@ it.effect("upgrades a database already at released main migration 034", () => [41, "OrchestrationV2EffectCancellation"], [42, "ScheduledTasks"], [43, "LegacyV1ImportState"], + [44, "OrchestrationV2SubagentObservability"], ], ); diff --git a/apps/server/src/persistence/Migrations/044_OrchestrationV2SubagentObservability.test.ts b/apps/server/src/persistence/Migrations/044_OrchestrationV2SubagentObservability.test.ts new file mode 100644 index 00000000000..81c7a2f25bc --- /dev/null +++ b/apps/server/src/persistence/Migrations/044_OrchestrationV2SubagentObservability.test.ts @@ -0,0 +1,222 @@ +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import { runMigrations } from "../Migrations.ts"; +import * as NodeSqliteClient from "../NodeSqliteClient.ts"; + +const layer = it.layer(Layer.mergeAll(NodeSqliteClient.layerMemory())); + +layer("044_OrchestrationV2SubagentObservability", (it) => { + it.effect("backfills old subagents and installs activation persistence", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* runMigrations({ toMigrationInclusive: 43 }); + yield* sql` + INSERT INTO orchestration_v2_projection_subagents ( + subagent_id, + thread_id, + run_id, + parent_node_id, + provider, + provider_thread_id, + child_thread_id, + origin, + status, + started_at, + completed_at, + updated_at, + payload_json + ) VALUES ( + 'node:legacy-subagent', + 'thread:legacy-subagent', + 'run:legacy-subagent', + 'node:legacy-root', + 'codex', + NULL, + NULL, + 'provider_native', + 'completed', + '2026-07-26T00:00:00.000Z', + '2026-07-26T00:01:00.000Z', + '2026-07-26T00:01:00.000Z', + '{"id":"node:legacy-subagent","status":"completed"}' + ) + `; + yield* sql` + INSERT INTO orchestration_v2_projection_subagents ( + subagent_id, + thread_id, + run_id, + parent_node_id, + provider, + provider_thread_id, + child_thread_id, + origin, + status, + started_at, + completed_at, + updated_at, + payload_json + ) VALUES ( + 'node:current-subagent', + 'thread:current-subagent', + 'run:current-subagent', + 'node:current-root', + 'claudeAgent', + NULL, + NULL, + 'provider_native', + 'running', + '2026-07-26T00:00:00.000Z', + NULL, + '2026-07-26T00:01:00.000Z', + '{"id":"node:current-subagent","status":"running","kind":"workflow_agent","role":{"name":"researcher","source":"provider"},"usage":{"totalTokens":12},"currentActivationId":"activation:current","activationCount":2,"workflow":null,"workflowMembership":null,"recentActivity":[{"at":"2026-07-26T00:00:30.000Z","summary":"kept"}]}' + ) + `; + + // Half-migrated: `role` and `recentActivity` already exist as structured + // JSON, while `kind`, `usage`, and `activationCount` are still missing. + yield* sql` + INSERT INTO orchestration_v2_projection_subagents ( + subagent_id, + thread_id, + run_id, + parent_node_id, + provider, + provider_thread_id, + child_thread_id, + origin, + status, + started_at, + completed_at, + updated_at, + payload_json + ) VALUES ( + 'node:mixed-subagent', + 'thread:mixed-subagent', + 'run:mixed-subagent', + 'node:mixed-root', + 'codex', + NULL, + NULL, + 'provider_native', + 'running', + '2026-07-26T00:00:00.000Z', + NULL, + '2026-07-26T00:01:00.000Z', + '{"id":"node:mixed-subagent","status":"running","role":{"name":"reviewer","source":"provider"},"recentActivity":[{"at":"2026-07-26T00:00:30.000Z","summary":"kept"}]}' + ) + `; + + yield* runMigrations({ toMigrationInclusive: 44 }); + + const rows = yield* sql<{ + readonly kind: string; + readonly role_name: string; + readonly role_source: string; + readonly activation_count: number; + readonly usage_type: string; + readonly activity_count: number; + }>` + SELECT + json_extract(payload_json, '$.kind') AS kind, + json_extract(payload_json, '$.role.name') AS role_name, + json_extract(payload_json, '$.role.source') AS role_source, + json_extract(payload_json, '$.activationCount') AS activation_count, + json_type(payload_json, '$.usage') AS usage_type, + json_array_length(json_extract(payload_json, '$.recentActivity')) AS activity_count + FROM orchestration_v2_projection_subagents + WHERE subagent_id = 'node:legacy-subagent' + `; + assert.deepStrictEqual(rows, [ + { + kind: "subagent", + role_name: "general-purpose", + role_source: "app_default", + activation_count: 1, + usage_type: "null", + activity_count: 0, + }, + ]); + + const currentRows = yield* sql<{ + readonly kind: string; + readonly role_name: string; + readonly role_source: string; + readonly total_tokens: number; + readonly current_activation_id: string; + readonly activation_count: number; + readonly activity_summary: string; + }>` + SELECT + json_extract(payload_json, '$.kind') AS kind, + json_extract(payload_json, '$.role.name') AS role_name, + json_extract(payload_json, '$.role.source') AS role_source, + json_extract(payload_json, '$.usage.totalTokens') AS total_tokens, + json_extract(payload_json, '$.currentActivationId') AS current_activation_id, + json_extract(payload_json, '$.activationCount') AS activation_count, + json_extract(payload_json, '$.recentActivity[0].summary') AS activity_summary + FROM orchestration_v2_projection_subagents + WHERE subagent_id = 'node:current-subagent' + `; + assert.deepStrictEqual(currentRows, [ + { + kind: "workflow_agent", + role_name: "researcher", + role_source: "provider", + total_tokens: 12, + current_activation_id: "activation:current", + activation_count: 2, + activity_summary: "kept", + }, + ]); + + // A partially-migrated row is the only one that both matches the WHERE + // clause and round-trips already-present JSON through COALESCE. If the + // JSON subtype were lost there, structured values would be rewritten as + // quoted strings and the payload would stop decoding — the same + // unreadable-event failure this migration exists to avoid. + const mixedRows = yield* sql<{ + readonly role_type: string; + readonly role_name: string; + readonly activity_type: string; + readonly activity_summary: string; + readonly usage_type: string; + readonly kind: string; + readonly activation_count: number; + }>` + SELECT + json_type(payload_json, '$.role') AS role_type, + json_extract(payload_json, '$.role.name') AS role_name, + json_type(payload_json, '$.recentActivity') AS activity_type, + json_extract(payload_json, '$.recentActivity[0].summary') AS activity_summary, + json_type(payload_json, '$.usage') AS usage_type, + json_extract(payload_json, '$.kind') AS kind, + json_extract(payload_json, '$.activationCount') AS activation_count + FROM orchestration_v2_projection_subagents + WHERE subagent_id = 'node:mixed-subagent' + `; + assert.deepStrictEqual(mixedRows, [ + { + role_type: "object", + role_name: "reviewer", + activity_type: "array", + activity_summary: "kept", + usage_type: "null", + kind: "subagent", + activation_count: 1, + }, + ]); + + const activationTable = yield* sql<{ readonly name: string }>` + SELECT name + FROM sqlite_master + WHERE type = 'table' + AND name = 'orchestration_v2_projection_subagent_activations' + `; + assert.lengthOf(activationTable, 1); + }), + ); +}); diff --git a/apps/server/src/persistence/Migrations/044_OrchestrationV2SubagentObservability.ts b/apps/server/src/persistence/Migrations/044_OrchestrationV2SubagentObservability.ts new file mode 100644 index 00000000000..215e9515abd --- /dev/null +++ b/apps/server/src/persistence/Migrations/044_OrchestrationV2SubagentObservability.ts @@ -0,0 +1,50 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + CREATE TABLE orchestration_v2_projection_subagent_activations ( + activation_id TEXT PRIMARY KEY, + thread_id TEXT NOT NULL, + subagent_id TEXT NOT NULL, + run_id TEXT, + provider_turn_id TEXT, + ordinal INTEGER NOT NULL, + status TEXT NOT NULL, + started_at TEXT, + completed_at TEXT, + updated_at TEXT NOT NULL, + payload_json TEXT NOT NULL + ) + `; + yield* sql`CREATE INDEX orchestration_v2_projection_subagent_activations_thread_idx ON orchestration_v2_projection_subagent_activations(thread_id, subagent_id, ordinal)`; + yield* sql`CREATE INDEX orchestration_v2_projection_subagent_activations_run_idx ON orchestration_v2_projection_subagent_activations(run_id, status)`; + + yield* sql` + UPDATE orchestration_v2_projection_subagents + SET payload_json = json_set( + payload_json, + '$.kind', COALESCE(json_extract(payload_json, '$.kind'), 'subagent'), + '$.role', COALESCE( + json_extract(payload_json, '$.role'), + json_object('name', 'general-purpose', 'source', 'app_default') + ), + '$.usage', json_extract(payload_json, '$.usage'), + '$.currentActivationId', json_extract(payload_json, '$.currentActivationId'), + '$.activationCount', COALESCE(json_extract(payload_json, '$.activationCount'), 1), + '$.workflow', json_extract(payload_json, '$.workflow'), + '$.workflowMembership', json_extract(payload_json, '$.workflowMembership'), + '$.recentActivity', COALESCE(json_extract(payload_json, '$.recentActivity'), json_array()) + ) + WHERE json_type(payload_json, '$.kind') IS NULL + OR json_type(payload_json, '$.role') IS NULL + OR json_type(payload_json, '$.usage') IS NULL + OR json_type(payload_json, '$.currentActivationId') IS NULL + OR json_type(payload_json, '$.activationCount') IS NULL + OR json_type(payload_json, '$.workflow') IS NULL + OR json_type(payload_json, '$.workflowMembership') IS NULL + OR json_type(payload_json, '$.recentActivity') IS NULL + `; +}); diff --git a/apps/web/src/test-fixtures.ts b/apps/web/src/test-fixtures.ts index f4804690b82..cb0b87da307 100644 --- a/apps/web/src/test-fixtures.ts +++ b/apps/web/src/test-fixtures.ts @@ -50,6 +50,7 @@ export function makeThreadProjectionFixture(): OrchestrationV2ThreadProjection { attempts: [], nodes: [], subagents: [], + subagentActivations: [], providerSessions: [], providerThreads: [], providerTurns: [], diff --git a/packages/client-runtime/src/state/orchestrationV2Projection.test.ts b/packages/client-runtime/src/state/orchestrationV2Projection.test.ts index f804bd64761..07cf08f383f 100644 --- a/packages/client-runtime/src/state/orchestrationV2Projection.test.ts +++ b/packages/client-runtime/src/state/orchestrationV2Projection.test.ts @@ -85,6 +85,7 @@ const emptyProjection = { attempts: [], nodes: [], subagents: [], + subagentActivations: [], providerSessions: [], providerThreads: [], providerTurns: [], diff --git a/packages/client-runtime/src/state/orchestrationV2Projection.ts b/packages/client-runtime/src/state/orchestrationV2Projection.ts index db49437cf69..b89ee920a97 100644 --- a/packages/client-runtime/src/state/orchestrationV2Projection.ts +++ b/packages/client-runtime/src/state/orchestrationV2Projection.ts @@ -125,6 +125,11 @@ export function applyOrchestrationV2ProjectionEvent( return { ...base, nodes: upsertEntity(base.nodes, event.payload) }; case "subagent.updated": return { ...base, subagents: upsertEntity(base.subagents, event.payload) }; + case "subagent-activation.updated": + return { + ...base, + subagentActivations: upsertEntity(base.subagentActivations, event.payload), + }; case "provider-session.attached": case "provider-session.updated": return { diff --git a/packages/client-runtime/src/state/orchestrationV2TestFixtures.ts b/packages/client-runtime/src/state/orchestrationV2TestFixtures.ts index 3907c395b42..250cfc53264 100644 --- a/packages/client-runtime/src/state/orchestrationV2TestFixtures.ts +++ b/packages/client-runtime/src/state/orchestrationV2TestFixtures.ts @@ -84,6 +84,7 @@ export const v2Projection: OrchestrationV2ThreadProjection = { attempts: [], nodes: [], subagents: [], + subagentActivations: [], providerSessions: [], providerThreads: [], providerTurns: [], diff --git a/packages/contracts/src/baseSchemas.ts b/packages/contracts/src/baseSchemas.ts index 2a1764d82ce..13248712ef8 100644 --- a/packages/contracts/src/baseSchemas.ts +++ b/packages/contracts/src/baseSchemas.ts @@ -47,6 +47,8 @@ export const RunAttemptId = makeEntityId("RunAttemptId"); export type RunAttemptId = typeof RunAttemptId.Type; export const NodeId = makeEntityId("NodeId"); export type NodeId = typeof NodeId.Type; +export const SubagentActivationId = makeEntityId("SubagentActivationId"); +export type SubagentActivationId = typeof SubagentActivationId.Type; export const AuthSessionId = makeEntityId("AuthSessionId"); export type AuthSessionId = typeof AuthSessionId.Type; diff --git a/packages/contracts/src/orchestrationV2.test.ts b/packages/contracts/src/orchestrationV2.test.ts index 2e7b1080cd0..baaa4b87033 100644 --- a/packages/contracts/src/orchestrationV2.test.ts +++ b/packages/contracts/src/orchestrationV2.test.ts @@ -24,10 +24,15 @@ import { OrchestrationV2CheckpointScope, OrchestrationV2Command, OrchestrationV2DomainEvent, + OrchestrationV2DomainEventJson, OrchestrationV2ShellSnapshot, OrchestrationV2Subagent, + OrchestrationV2SubagentActivation, OrchestrationV2ThreadProjection, OrchestrationV2TurnItem, + OrchestrationV2TurnItemJson, + OrchestrationV2TurnItemStatus, + orchestrationV2SubagentStatusAsTurnItemStatus, } from "./orchestrationV2.ts"; const now = DateTime.makeUnsafe("2026-04-20T00:00:00.000Z"); @@ -410,13 +415,89 @@ describe("orchestration V2 contracts", () => { prompt: "Inspect package.json", title: "Package audit", model: "gpt-5.4", - status: "completed", + kind: "subagent", + role: { name: "reviewer", source: "provider" }, + status: "idle", progress: "Inspecting package metadata", result: "Package is private.", + usage: { + totalTokens: 1200, + inputTokens: 900, + outputTokens: 300, + toolUses: 2, + }, + currentActivationId: null, + activationCount: 2, + workflow: null, + workflowMembership: null, + recentActivity: [{ at: now, summary: "Inspected package metadata" }], + startedAt: now, + completedAt: now, + updatedAt: now, + }); + const activation = Schema.decodeUnknownSync(OrchestrationV2SubagentActivation)({ + id: "node-subagent-1:activation:2", + threadId: "thread-1", + subagentId: subagent.id, + runId: "run-1", + providerTurnId: "provider-turn-1", + ordinal: 2, + status: "completed", + usage: { totalTokens: 400 }, startedAt: now, completedAt: now, updatedAt: now, }); + const { + kind: _kind, + role: _role, + usage: _usage, + currentActivationId: _currentActivationId, + activationCount: _activationCount, + workflow: _workflow, + workflowMembership: _workflowMembership, + recentActivity: _recentActivity, + ...legacySubagent + } = subagent; + const decodedLegacySubagent = Schema.decodeUnknownSync(OrchestrationV2Subagent)(legacySubagent); + const decodedLegacyEvent = Schema.decodeUnknownSync(OrchestrationV2DomainEventJson)({ + id: "event-legacy-subagent", + type: "subagent.updated", + threadId: "thread-1", + runId: "run-1", + nodeId: "node-subagent-1", + driver: "codex", + providerInstanceId: "codex", + occurredAt: DateTime.formatIso(now), + payload: { + ...legacySubagent, + startedAt: DateTime.formatIso(now), + completedAt: DateTime.formatIso(now), + updatedAt: DateTime.formatIso(now), + }, + }); + const decodedActivityEvent = Schema.decodeUnknownSync(OrchestrationV2DomainEventJson)({ + id: "event-subagent-activity", + type: "subagent.updated", + threadId: "thread-1", + runId: "run-1", + nodeId: "node-subagent-1", + driver: "codex", + providerInstanceId: "codex", + occurredAt: DateTime.formatIso(now), + payload: { + ...subagent, + recentActivity: [ + { + at: DateTime.formatIso(now), + summary: "Inspected package metadata", + }, + ], + startedAt: DateTime.formatIso(now), + completedAt: DateTime.formatIso(now), + updatedAt: DateTime.formatIso(now), + }, + }); const turnItem = Schema.decodeUnknownSync(OrchestrationV2TurnItem)({ id: "turn-item-subagent-1", type: "subagent", @@ -446,6 +527,24 @@ describe("orchestration V2 contracts", () => { expect(subagent.origin).toBe("provider_native"); expect(subagent.progress).toBe("Inspecting package metadata"); expect(subagent.childThreadId).toBeNull(); + expect(subagent.role).toEqual({ name: "reviewer", source: "provider" }); + expect(subagent.status).toBe("idle"); + expect(activation.ordinal).toBe(2); + expect(decodedLegacySubagent.role).toEqual({ + name: "general-purpose", + source: "app_default", + }); + expect(decodedLegacySubagent.activationCount).toBe(1); + if (decodedLegacyEvent.type !== "subagent.updated") { + throw new Error("expected legacy subagent event"); + } + expect(decodedLegacyEvent.payload.recentActivity).toEqual([]); + if (decodedActivityEvent.type !== "subagent.updated") { + throw new Error("expected subagent activity event"); + } + expect(DateTime.formatIso(decodedActivityEvent.payload.recentActivity[0]!.at)).toBe( + DateTime.formatIso(now), + ); expect(turnItem.type).toBe("subagent"); if (turnItem.type !== "subagent") throw new Error("expected subagent item"); expect(turnItem.progress).toBe("Inspecting package metadata"); @@ -514,6 +613,7 @@ describe("orchestration V2 contracts", () => { attempts: [], nodes: [], subagents: [], + subagentActivations: [], providerSessions: [], providerThreads: [], providerTurns: [], @@ -654,6 +754,61 @@ describe("orchestration V2 contracts", () => { expect(fork.type).toBe("fork"); }); + it("projects every subagent status onto a decodable turn item status", () => { + // Regression guard for a bricked-server class of bug: a producer copied a + // raw subagent status onto a timeline row, emitting a "turn-item.updated" + // event that the domain schema could not decode. Nothing failed until the + // next startup, where the projection rebuild died on the stored event and + // the server could no longer boot at all. + const subagentStatuses = OrchestrationV2Subagent.fields.status.literals; + const turnItemStatuses = new Set(OrchestrationV2TurnItemStatus.literals); + + for (const status of subagentStatuses) { + const projected = orchestrationV2SubagentStatusAsTurnItemStatus[status]; + expect( + turnItemStatuses.has(projected), + `subagent status "${status}" projects to "${projected}", which a turn item cannot hold`, + ).toBe(true); + } + + // "idle" is the reusable-identity resting state and the one with no direct + // timeline equivalent; it must collapse onto the finished activation. + expect(subagentStatuses).toContain("idle"); + expect(turnItemStatuses.has("idle")).toBe(false); + expect(orchestrationV2SubagentStatusAsTurnItemStatus.idle).toBe("completed"); + + // Prove it end to end: the projected status decodes as a real event, and + // the raw one is rejected — exactly the event that could not be read back. + const base = { + id: "turn-item:subagent-status-projection", + threadId: "thread-1", + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + title: null, + startedAt: null, + completedAt: null, + updatedAt: "2026-07-27T00:00:00.000Z", + type: "subagent", + subagentId: "node-1", + origin: "provider_native", + driver: "codex", + providerInstanceId: "codex", + childThreadId: null, + prompt: "", + result: null, + }; + const decode = Schema.decodeUnknownSync(OrchestrationV2TurnItemJson); + expect(() => + decode({ ...base, status: orchestrationV2SubagentStatusAsTurnItemStatus.idle }), + ).not.toThrow(); + expect(() => decode({ ...base, status: "idle" })).toThrow(); + }); + it("exports the V2 branded ids through the public contracts entrypoint", () => { expect(ThreadId.make("thread-1")).toBe("thread-1"); expect(ProjectId.make("project-1")).toBe("project-1"); diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index bacaf670221..92a651e85e5 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -23,6 +23,7 @@ import { RunAttemptId, RunId, RuntimeRequestId, + SubagentActivationId, ThreadId, TrimmedNonEmptyString, TurnItemId, @@ -431,6 +432,35 @@ export const OrchestrationV2ExecutionNode = Schema.Struct({ }); export type OrchestrationV2ExecutionNode = typeof OrchestrationV2ExecutionNode.Type; +export const OrchestrationV2SubagentUsage = Schema.Struct({ + totalTokens: NonNegativeInt, + inputTokens: Schema.optional(NonNegativeInt), + cachedInputTokens: Schema.optional(NonNegativeInt), + outputTokens: Schema.optional(NonNegativeInt), + reasoningOutputTokens: Schema.optional(NonNegativeInt), + toolUses: Schema.optional(NonNegativeInt), + durationMs: Schema.optional(NonNegativeInt), +}); +export type OrchestrationV2SubagentUsage = typeof OrchestrationV2SubagentUsage.Type; + +export const OrchestrationV2SubagentRole = Schema.Struct({ + name: TrimmedNonEmptyString, + source: Schema.Literals(["provider", "app_default"]), +}); +export type OrchestrationV2SubagentRole = typeof OrchestrationV2SubagentRole.Type; + +export const OrchestrationV2SubagentActivity = Schema.Struct({ + at: Schema.DateTimeUtc, + summary: TrimmedNonEmptyString, +}); +export type OrchestrationV2SubagentActivity = typeof OrchestrationV2SubagentActivity.Type; + +export const OrchestrationV2WorkflowPhase = Schema.Struct({ + index: NonNegativeInt, + title: TrimmedNonEmptyString, +}); +export type OrchestrationV2WorkflowPhase = typeof OrchestrationV2WorkflowPhase.Type; + export const OrchestrationV2Subagent = Schema.Struct({ id: NodeId, threadId: ThreadId, @@ -452,10 +482,19 @@ export const OrchestrationV2Subagent = Schema.Struct({ // live run (wait-mode delegations, whose result returns through the // blocking tool call). Absent on legacy records; treated as settled_only. completionWake: Schema.optional(Schema.Literals(["always", "settled_only"])), + kind: Schema.Literals(["subagent", "workflow", "workflow_agent"]).pipe( + Schema.withDecodingDefault(Effect.succeed("subagent" as const)), + ), + role: OrchestrationV2SubagentRole.pipe( + Schema.withDecodingDefault( + Effect.succeed({ name: "general-purpose", source: "app_default" as const }), + ), + ), status: Schema.Literals([ "pending", "running", "waiting", + "idle", "completed", "failed", "cancelled", @@ -463,12 +502,58 @@ export const OrchestrationV2Subagent = Schema.Struct({ ]), progress: Schema.optional(Schema.String), result: Schema.NullOr(Schema.String), + usage: Schema.NullOr(OrchestrationV2SubagentUsage).pipe( + Schema.withDecodingDefault(Effect.succeed(null)), + ), + currentActivationId: Schema.NullOr(SubagentActivationId).pipe( + Schema.withDecodingDefault(Effect.succeed(null)), + ), + activationCount: NonNegativeInt.pipe(Schema.withDecodingDefault(Effect.succeed(1))), + workflow: Schema.NullOr( + Schema.Struct({ + phases: Schema.Array(OrchestrationV2WorkflowPhase), + }), + ).pipe(Schema.withDecodingDefault(Effect.succeed(null))), + workflowMembership: Schema.NullOr( + Schema.Struct({ + workflowSubagentId: NodeId, + agentIndex: NonNegativeInt, + phaseIndex: Schema.NullOr(NonNegativeInt), + attempt: PositiveInt, + }), + ).pipe(Schema.withDecodingDefault(Effect.succeed(null))), + recentActivity: Schema.Array(OrchestrationV2SubagentActivity).pipe( + Schema.withDecodingDefault(Effect.succeed([])), + ), startedAt: Schema.NullOr(Schema.DateTimeUtc), completedAt: Schema.NullOr(Schema.DateTimeUtc), updatedAt: Schema.DateTimeUtc, }); export type OrchestrationV2Subagent = typeof OrchestrationV2Subagent.Type; +export const OrchestrationV2SubagentActivation = Schema.Struct({ + id: SubagentActivationId, + threadId: ThreadId, + subagentId: NodeId, + runId: Schema.NullOr(RunId), + providerTurnId: Schema.NullOr(ProviderTurnId), + ordinal: PositiveInt, + status: Schema.Literals([ + "pending", + "running", + "waiting", + "completed", + "failed", + "cancelled", + "interrupted", + ]), + usage: Schema.NullOr(OrchestrationV2SubagentUsage), + startedAt: Schema.NullOr(Schema.DateTimeUtc), + completedAt: Schema.NullOr(Schema.DateTimeUtc), + updatedAt: Schema.DateTimeUtc, +}); +export type OrchestrationV2SubagentActivation = typeof OrchestrationV2SubagentActivation.Type; + export const OrchestrationV2CheckpointScope = Schema.Struct({ id: CheckpointScopeId, threadId: ThreadId, @@ -710,6 +795,33 @@ export const OrchestrationV2TurnItemStatus = Schema.Literals([ ]); export type OrchestrationV2TurnItemStatus = typeof OrchestrationV2TurnItemStatus.Type; +/** + * How a subagent's status projects onto its timeline row. + * + * A subagent identity is reusable and rests at "idle" between activations, but + * a turn item records one completed activation and has no such state. Producers + * must translate rather than copy: writing a raw subagent status onto a turn + * item can emit an event that its own schema cannot decode, which is + * unrecoverable because the projection rebuild fails on every later startup. + * + * The `Record` type is the safeguard — adding a subagent status without giving + * it a timeline meaning here is a compile error, not a runtime brick. + */ +export const orchestrationV2SubagentStatusAsTurnItemStatus: Record< + typeof OrchestrationV2Subagent.Type.status, + OrchestrationV2TurnItemStatus +> = { + pending: "pending", + running: "running", + waiting: "waiting", + // Resting between activations: the row for the activation that just ended. + idle: "completed", + completed: "completed", + failed: "failed", + cancelled: "cancelled", + interrupted: "interrupted", +}; + export const OrchestrationV2ProviderFailureClass = Schema.Literals([ "provider_error", "transport_error", @@ -1042,6 +1154,11 @@ export const OrchestrationV2DomainEvent = Schema.Union([ type: Schema.Literal("subagent.updated"), payload: OrchestrationV2Subagent, }), + Schema.Struct({ + ...OrchestrationV2EventBase.fields, + type: Schema.Literal("subagent-activation.updated"), + payload: OrchestrationV2SubagentActivation, + }), Schema.Struct({ ...OrchestrationV2EventBase.fields, type: Schema.Literals(["provider-session.attached", "provider-session.updated"]), @@ -1121,6 +1238,7 @@ export const OrchestrationV2ThreadProjection = Schema.Struct({ attempts: Schema.Array(OrchestrationV2RunAttempt), nodes: Schema.Array(OrchestrationV2ExecutionNode), subagents: Schema.Array(OrchestrationV2Subagent), + subagentActivations: Schema.Array(OrchestrationV2SubagentActivation), providerSessions: Schema.Array(OrchestrationV2ProviderSession), providerThreads: Schema.Array(OrchestrationV2ProviderThread), providerTurns: Schema.Array(OrchestrationV2ProviderTurn), @@ -1296,12 +1414,29 @@ export type OrchestrationV2ExecutionNodeJson = typeof OrchestrationV2ExecutionNo export const OrchestrationV2SubagentJson = OrchestrationV2Subagent.mapFields((fields) => ({ ...fields, + recentActivity: Schema.Array( + OrchestrationV2SubagentActivity.mapFields((activityFields) => ({ + ...activityFields, + at: Schema.DateTimeUtcFromString, + })), + ).pipe(Schema.withDecodingDefault(Effect.succeed([]))), startedAt: Schema.NullOr(Schema.DateTimeUtcFromString), completedAt: Schema.NullOr(Schema.DateTimeUtcFromString), updatedAt: Schema.DateTimeUtcFromString, })); export type OrchestrationV2SubagentJson = typeof OrchestrationV2SubagentJson.Type; +export const OrchestrationV2SubagentActivationJson = OrchestrationV2SubagentActivation.mapFields( + (fields) => ({ + ...fields, + startedAt: Schema.NullOr(Schema.DateTimeUtcFromString), + completedAt: Schema.NullOr(Schema.DateTimeUtcFromString), + updatedAt: Schema.DateTimeUtcFromString, + }), +); +export type OrchestrationV2SubagentActivationJson = + typeof OrchestrationV2SubagentActivationJson.Type; + export const OrchestrationV2CheckpointScopeJson = OrchestrationV2CheckpointScope.mapFields( (fields) => ({ ...fields, @@ -1587,6 +1722,9 @@ export const OrchestrationV2ThreadProjectionJson = OrchestrationV2ThreadProjecti attempts: Schema.Array(OrchestrationV2RunAttemptJson), nodes: Schema.Array(OrchestrationV2ExecutionNodeJson), subagents: Schema.Array(OrchestrationV2SubagentJson), + subagentActivations: Schema.Array(OrchestrationV2SubagentActivationJson).pipe( + Schema.withDecodingDefault(Effect.succeed([])), + ), providerSessions: Schema.Array(OrchestrationV2ProviderSessionJson), providerThreads: Schema.Array(OrchestrationV2ProviderThreadJson), providerTurns: Schema.Array(OrchestrationV2ProviderTurnJson), @@ -1711,6 +1849,11 @@ export const OrchestrationV2DomainEventJson = Schema.Union([ type: Schema.Literal("subagent.updated"), payload: OrchestrationV2SubagentJson, }), + Schema.Struct({ + ...OrchestrationV2JsonEventBaseFields, + type: Schema.Literal("subagent-activation.updated"), + payload: OrchestrationV2SubagentActivationJson, + }), Schema.Struct({ ...OrchestrationV2JsonEventBaseFields, type: Schema.Literals(["provider-session.attached", "provider-session.updated"]),