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 1/3] 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"]), From 313ae4d261a97e4562d75db379541882437714e8 Mon Sep 17 00:00:00 2001 From: Shivam Sharma <91240327+shivamhwp@users.noreply.github.com> Date: Wed, 29 Jul 2026 00:54:40 +0530 Subject: [PATCH 2/3] refactor(orchestration-v2): drop the redundant subagent observability backfill Every observability field the migration backfilled carries a decoding default equal to the value it wrote, so pre-upgrade rows already read back correctly without it. The projection schema version bump also fails startup verification, which deletes and replays those rows from the event log before anything reads them. Pin both halves: a contracts test decodes a stored payload missing all eight fields through the schema the projection actually uses, and the migration test asserts an existing row survives byte-identical. Co-Authored-By: Claude Opus 5 (1M context) --- ...chestrationV2SubagentObservability.test.ts | 196 +++--------------- ...44_OrchestrationV2SubagentObservability.ts | 30 +-- .../contracts/src/orchestrationV2.test.ts | 43 ++++ 3 files changed, 79 insertions(+), 190 deletions(-) diff --git a/apps/server/src/persistence/Migrations/044_OrchestrationV2SubagentObservability.test.ts b/apps/server/src/persistence/Migrations/044_OrchestrationV2SubagentObservability.test.ts index 81c7a2f25bc..ed637036bdd 100644 --- a/apps/server/src/persistence/Migrations/044_OrchestrationV2SubagentObservability.test.ts +++ b/apps/server/src/persistence/Migrations/044_OrchestrationV2SubagentObservability.test.ts @@ -9,10 +9,12 @@ 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", () => + 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, @@ -41,175 +43,12 @@ layer("044_OrchestrationV2SubagentObservability", (it) => { '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"}]}' + ${legacyPayload} ) `; 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 @@ -217,6 +56,33 @@ layer("044_OrchestrationV2SubagentObservability", (it) => { 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 index 215e9515abd..f30127b484a 100644 --- a/apps/server/src/persistence/Migrations/044_OrchestrationV2SubagentObservability.ts +++ b/apps/server/src/persistence/Migrations/044_OrchestrationV2SubagentObservability.ts @@ -22,29 +22,9 @@ export default Effect.gen(function* () { 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 - `; + // 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/packages/contracts/src/orchestrationV2.test.ts b/packages/contracts/src/orchestrationV2.test.ts index baaa4b87033..318617e2681 100644 --- a/packages/contracts/src/orchestrationV2.test.ts +++ b/packages/contracts/src/orchestrationV2.test.ts @@ -28,6 +28,7 @@ import { OrchestrationV2ShellSnapshot, OrchestrationV2Subagent, OrchestrationV2SubagentActivation, + OrchestrationV2SubagentJson, OrchestrationV2ThreadProjection, OrchestrationV2TurnItem, OrchestrationV2TurnItemJson, @@ -550,6 +551,48 @@ describe("orchestration V2 contracts", () => { 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 = Schema.decodeUnknownSync(Schema.fromJsonString(OrchestrationV2SubagentJson))( + 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", From f4037f3e7e95b62ace9fbf0c87401644a9b1e33d Mon Sep 17 00:00:00 2001 From: Shivam Sharma <91240327+shivamhwp@users.noreply.github.com> Date: Wed, 29 Jul 2026 01:52:21 +0530 Subject: [PATCH 3/3] refactor(orchestration-v2): hoist the new schema decoders to module scope The lint rule the CI Check job enforces flags a compiled decoder rebuilt on every call. The activation payload decoder and the subagent decoders this PR added were inline, which took the branch from the base's 67 lint warnings to 73 and failed the job. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/orchestration-v2/ProjectionStore.ts | 7 +++-- .../contracts/src/orchestrationV2.test.ts | 26 ++++++++++++------- 2 files changed, 20 insertions(+), 13 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 6c45ca903bc..214d8911db6 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -477,10 +477,9 @@ 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 decodeSubagentActivationPayload = Schema.decodeUnknownEffect( + Schema.fromJsonString(OrchestrationV2SubagentActivationJsonSchema), +); const decodeProviderSessionPayload = (json: string) => Schema.decodeUnknownEffect(Schema.fromJsonString(OrchestrationV2ProviderSessionJsonSchema))(json); const decodeProviderThreadPayload = (json: string) => diff --git a/packages/contracts/src/orchestrationV2.test.ts b/packages/contracts/src/orchestrationV2.test.ts index 318617e2681..cdf34e91212 100644 --- a/packages/contracts/src/orchestrationV2.test.ts +++ b/packages/contracts/src/orchestrationV2.test.ts @@ -47,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", () => { @@ -397,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", @@ -436,7 +446,7 @@ describe("orchestration V2 contracts", () => { completedAt: now, updatedAt: now, }); - const activation = Schema.decodeUnknownSync(OrchestrationV2SubagentActivation)({ + const activation = decodeOrchestrationV2SubagentActivation({ id: "node-subagent-1:activation:2", threadId: "thread-1", subagentId: subagent.id, @@ -460,8 +470,8 @@ describe("orchestration V2 contracts", () => { recentActivity: _recentActivity, ...legacySubagent } = subagent; - const decodedLegacySubagent = Schema.decodeUnknownSync(OrchestrationV2Subagent)(legacySubagent); - const decodedLegacyEvent = Schema.decodeUnknownSync(OrchestrationV2DomainEventJson)({ + const decodedLegacySubagent = decodeOrchestrationV2Subagent(legacySubagent); + const decodedLegacyEvent = decodeOrchestrationV2DomainEventJson({ id: "event-legacy-subagent", type: "subagent.updated", threadId: "thread-1", @@ -477,7 +487,7 @@ describe("orchestration V2 contracts", () => { updatedAt: DateTime.formatIso(now), }, }); - const decodedActivityEvent = Schema.decodeUnknownSync(OrchestrationV2DomainEventJson)({ + const decodedActivityEvent = decodeOrchestrationV2DomainEventJson({ id: "event-subagent-activity", type: "subagent.updated", threadId: "thread-1", @@ -579,9 +589,7 @@ describe("orchestration V2 contracts", () => { }); // The projection reads stored rows through exactly this schema. - const decoded = Schema.decodeUnknownSync(Schema.fromJsonString(OrchestrationV2SubagentJson))( - legacyPayloadJson, - ); + const decoded = decodeStoredOrchestrationV2Subagent(legacyPayloadJson); expect(decoded.kind).toBe("subagent"); expect(decoded.role).toEqual({ name: "general-purpose", source: "app_default" }); @@ -615,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();