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..214d8911db6 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,9 @@ const decodeNodePayload = (json: string) => Schema.decodeUnknownEffect(Schema.fromJsonString(OrchestrationV2ExecutionNodeJsonSchema))(json); const decodeSubagentPayload = (json: string) => Schema.decodeUnknownEffect(Schema.fromJsonString(OrchestrationV2SubagentJsonSchema))(json); +const decodeSubagentActivationPayload = Schema.decodeUnknownEffect( + Schema.fromJsonString(OrchestrationV2SubagentActivationJsonSchema), +); const decodeProviderSessionPayload = (json: string) => Schema.decodeUnknownEffect(Schema.fromJsonString(OrchestrationV2ProviderSessionJsonSchema))(json); const decodeProviderThreadPayload = (json: string) => @@ -1266,6 +1279,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 +1898,7 @@ export const layer: Layer.Layer = attemptRows, nodeRows, subagentRows, + subagentActivationRows, providerSessionRows, providerThreadRows, providerTurnRows, @@ -1877,6 +1936,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 +2028,7 @@ export const layer: Layer.Layer = attempts, nodes, subagents, + subagentActivations, providerSessions, providerThreads, providerTurns, @@ -1979,6 +2045,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 +2065,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..ed637036bdd --- /dev/null +++ b/apps/server/src/persistence/Migrations/044_OrchestrationV2SubagentObservability.test.ts @@ -0,0 +1,88 @@ +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("installs activation persistence and leaves existing subagents alone", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* runMigrations({ toMigrationInclusive: 43 }); + + const legacyPayload = '{"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: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', + ${legacyPayload} + ) + `; + + yield* runMigrations({ toMigrationInclusive: 44 }); + + 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); + + const indexes = yield* sql<{ readonly name: string }>` + SELECT name + FROM sqlite_master + WHERE type = 'index' + AND tbl_name = 'orchestration_v2_projection_subagent_activations' + AND sql IS NOT NULL + ORDER BY name ASC + `; + assert.deepStrictEqual( + indexes.map((row) => row.name), + [ + "orchestration_v2_projection_subagent_activations_run_idx", + "orchestration_v2_projection_subagent_activations_thread_idx", + ], + ); + + // Deliberately no backfill: the observability fields all carry decoding + // defaults, and the projection schema version bump replays these rows + // from the event log anyway. Rewriting them here would be dead work, so + // the row must come through byte-identical. + const rows = yield* sql<{ readonly payload_json: string }>` + SELECT payload_json + FROM orchestration_v2_projection_subagents + WHERE subagent_id = 'node:legacy-subagent' + `; + assert.deepStrictEqual(rows, [{ payload_json: legacyPayload }]); + }), + ); +}); 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..f30127b484a --- /dev/null +++ b/apps/server/src/persistence/Migrations/044_OrchestrationV2SubagentObservability.ts @@ -0,0 +1,30 @@ +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)`; + + // No backfill of orchestration_v2_projection_subagents. Every observability + // field this schema adds carries a decoding default equal to what a backfill + // would write, so pre-upgrade rows already read back correctly; and the + // projection schema version bump makes startup verification fail, which + // deletes and replays those rows from the event log regardless. +}); 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..cdf34e91212 100644 --- a/packages/contracts/src/orchestrationV2.test.ts +++ b/packages/contracts/src/orchestrationV2.test.ts @@ -24,10 +24,16 @@ import { OrchestrationV2CheckpointScope, OrchestrationV2Command, OrchestrationV2DomainEvent, + OrchestrationV2DomainEventJson, OrchestrationV2ShellSnapshot, OrchestrationV2Subagent, + OrchestrationV2SubagentActivation, + OrchestrationV2SubagentJson, OrchestrationV2ThreadProjection, OrchestrationV2TurnItem, + OrchestrationV2TurnItemJson, + OrchestrationV2TurnItemStatus, + orchestrationV2SubagentStatusAsTurnItemStatus, } from "./orchestrationV2.ts"; const now = DateTime.makeUnsafe("2026-04-20T00:00:00.000Z"); @@ -41,6 +47,16 @@ const LegacyShellStreamItem = Schema.Union([ const decodeLegacyShellStreamItem = Schema.decodeUnknownSync(LegacyShellStreamItem); const decodeOrchestrationV2Command = Schema.decodeUnknownSync(OrchestrationV2Command); const decodeOrchestrationV2TurnItem = Schema.decodeUnknownSync(OrchestrationV2TurnItem); +const decodeOrchestrationV2Subagent = Schema.decodeUnknownSync(OrchestrationV2Subagent); +const decodeOrchestrationV2SubagentActivation = Schema.decodeUnknownSync( + OrchestrationV2SubagentActivation, +); +const decodeOrchestrationV2DomainEventJson = Schema.decodeUnknownSync( + OrchestrationV2DomainEventJson, +); +const decodeStoredOrchestrationV2Subagent = Schema.decodeUnknownSync( + Schema.fromJsonString(OrchestrationV2SubagentJson), +); describe("orchestration V2 contracts", () => { it("lets legacy snapshot decoders ignore enrichment metadata", () => { @@ -391,7 +407,7 @@ describe("orchestration V2 contracts", () => { }); it("decodes provider-native subagent lifecycle records and timeline items", () => { - const subagent = Schema.decodeUnknownSync(OrchestrationV2Subagent)({ + const subagent = decodeOrchestrationV2Subagent({ id: "node-subagent-1", threadId: "thread-1", runId: "run-1", @@ -410,13 +426,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 = decodeOrchestrationV2SubagentActivation({ + 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 = decodeOrchestrationV2Subagent(legacySubagent); + const decodedLegacyEvent = decodeOrchestrationV2DomainEventJson({ + 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 = decodeOrchestrationV2DomainEventJson({ + 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,11 +538,69 @@ 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"); }); + it("defaults every observability field a pre-upgrade row omits", () => { + // Rows written before this schema carry none of the observability fields. + // Each default below is what the projection reads back for such a row, so + // no migration has to rewrite them in place: a stored row missing all + // eight decodes to exactly this. + const legacyPayloadJson = JSON.stringify({ + id: "node-subagent-legacy", + threadId: "thread-1", + runId: "run-1", + parentNodeId: "node-root-1", + origin: "provider_native", + createdBy: "agent", + driver: "codex", + providerInstanceId: "codex", + providerThreadId: null, + childThreadId: null, + nativeTaskRef: null, + prompt: "Audit the parser.", + title: "Parser audit", + model: null, + status: "completed", + result: null, + startedAt: DateTime.formatIso(now), + completedAt: DateTime.formatIso(now), + updatedAt: DateTime.formatIso(now), + }); + + // The projection reads stored rows through exactly this schema. + const decoded = decodeStoredOrchestrationV2Subagent(legacyPayloadJson); + + expect(decoded.kind).toBe("subagent"); + expect(decoded.role).toEqual({ name: "general-purpose", source: "app_default" }); + expect(decoded.usage).toBeNull(); + expect(decoded.currentActivationId).toBeNull(); + expect(decoded.activationCount).toBe(1); + expect(decoded.workflow).toBeNull(); + expect(decoded.workflowMembership).toBeNull(); + expect(decoded.recentActivity).toEqual([]); + }); + it("decodes app-owned subagent parent-wake policies", () => { const appOwnedSubagent = { id: "node-subagent-2", @@ -473,7 +623,7 @@ describe("orchestration V2 contracts", () => { completedAt: null, updatedAt: now, }; - const decode = Schema.decodeUnknownSync(OrchestrationV2Subagent); + const decode = decodeOrchestrationV2Subagent; // Legacy records predate the field and must behave as settled_only. expect(decode(appOwnedSubagent).completionWake).toBeUndefined(); @@ -514,6 +664,7 @@ describe("orchestration V2 contracts", () => { attempts: [], nodes: [], subagents: [], + subagentActivations: [], providerSessions: [], providerThreads: [], providerTurns: [], @@ -654,6 +805,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"]),