diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 1008d33dd1e0..ce7535a36d81 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1,6 +1,8 @@ import { threadPullRequestsOf } from "@t3tools/shared/threadPullRequests"; import type { OrchestrationV2AppThread, + OrchestrationV2PlanArtifact, + PlanId, OrchestrationV2ConversationMessage, OrchestrationV2DomainEvent, OrchestrationV2ProjectedTurnItem, @@ -243,6 +245,10 @@ export interface ProjectionStoreV2Shape { threadId: ThreadId, providerTurnId: ProviderTurnId, ) => Effect.Effect; + readonly getPlan: ( + threadId: ThreadId, + planId: PlanId, + ) => Effect.Effect; readonly getRuntimeRequest: ( threadId: ThreadId, requestId: RuntimeRequestId, @@ -3461,6 +3467,21 @@ export const layer: Layer.Layer = ) .pipe(Effect.mapError(controlReadError(threadId))); + const getPlan: ProjectionStoreV2Shape["getPlan"] = (threadId, planId) => + sql + .withTransaction( + Effect.gen(function* () { + yield* requireThread(threadId); + const rows = + yield* sql`SELECT payload_json FROM orchestration_v2_projection_plans + WHERE thread_id = ${threadId} AND plan_id = ${planId}`; + return rows[0] === undefined + ? undefined + : yield* decodePlanPayload(rows[0].payload_json); + }), + ) + .pipe(Effect.mapError(controlReadError(threadId))); + const getRuntimeRequest: ProjectionStoreV2Shape["getRuntimeRequest"] = (threadId, requestId) => sql .withTransaction( @@ -4380,6 +4401,7 @@ export const layer: Layer.Layer = getCheckpointContext, getPendingNativeUserInputs, getRuntimeRequest, + getPlan, getProviderControlContext, getRecoveryThreadIds, getUnreadableThreadIds, @@ -4514,6 +4536,13 @@ export const layerMemory: Layer.Layer = Layer.effect( ), }; }), + getPlan: (threadId, planId) => + Effect.gen(function* () { + const projection = (yield* Ref.get(replayState)).projections.get(threadId); + if (projection === undefined) + return yield* new ProjectionStoreThreadNotFoundError({ threadId }); + return projection.plans.find((plan) => plan.id === planId); + }), getRuntimeRequest: (threadId, requestId) => Effect.gen(function* () { const projection = (yield* Ref.get(replayState)).projections.get(threadId); diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts index 43c6339c910c..eb4f91daf7ce 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts @@ -6,12 +6,14 @@ import { type OrchestrationV2AppThread, type OrchestrationV2DomainEvent, type OrchestrationV2ExecutionNode, + type OrchestrationV2PlanArtifact, type OrchestrationV2RuntimeRequest, type OrchestrationV2ProviderThread, type OrchestrationV2Run, type OrchestrationV2TurnItem, ProviderDriverKind, ProviderInstanceId, + PlanId, RunAttemptId, RunId, RuntimeRequestId, @@ -23,6 +25,7 @@ import * as Fiber from "effect/Fiber"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts"; import { EventSinkV2, layer as eventSinkLayer } from "./EventSink.ts"; @@ -97,6 +100,8 @@ function threadCreatedEvent( interactionMode: "default", branch: null, worktreePath: null, + branchPullRequest: null, + activeOrderKey: null, activeProviderThreadId: providerThreadId, lineage: { parentThreadId: null, @@ -307,6 +312,119 @@ layer("ProviderEventIngestorV2", (it) => { }), ); + it.effect("carries plan-step durations through consecutive and restarted ingestion", () => + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse("2026-09-07T00:00:00.000Z")); + const now = yield* DateTime.now; + const eventSink = yield* EventSinkV2; + const projectionStore = yield* ProjectionStoreV2; + const ingestor = yield* ProviderEventIngestorV2; + const idAllocator = yield* IdAllocatorV2; + const threadEvent = yield* threadCreatedEvent(now); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId: threadEvent.threadId, + }); + const planId = PlanId.make("plan:provider-event-duration"); + const nodeId = NodeId.make("node:provider-event-duration"); + type TodoListPlan = Extract; + const plan = (steps: TodoListPlan["steps"]): TodoListPlan => ({ + id: planId, + threadId: threadEvent.threadId, + runId: null, + nodeId, + kind: "todo_list", + status: "active", + steps, + }); + const ingest = (service: ProviderEventIngestorV2["Service"], steps: TodoListPlan["steps"]) => + service.ingestNormalized({ + providerSessionId, + providerInstanceId: modelSelection.instanceId, + threadId: threadEvent.threadId, + event: { type: "plan.updated", driver: CODEX_DRIVER, plan: plan(steps) }, + }); + + yield* eventSink.write({ events: [threadEvent] }); + yield* ingest(ingestor, [ + { id: "duplicate-a", text: "Verify", status: "running" }, + { id: "duplicate-b", text: "Verify", status: "pending" }, + { id: "fallback", text: "Report", status: "pending" }, + ]); + yield* TestClock.adjust("3 seconds"); + yield* ingest(ingestor, [ + { id: "duplicate-a", text: "Verify", status: "completed" }, + { id: "duplicate-b", text: "Verify", status: "pending" }, + { id: "fallback", text: "Report", status: "pending" }, + ]); + + const restartedIngestor = yield* ProviderEventIngestorV2.pipe( + Effect.provide( + Layer.fresh(providerEventIngestorLayer).pipe( + Layer.provide( + Layer.succeed(ProjectionStoreV2, { + ...projectionStore, + getThreadProjection: () => Effect.die("Plan timing must not load thread history"), + }), + ), + ), + ), + ); + yield* TestClock.adjust("4 seconds"); + yield* ingest(restartedIngestor, [ + { id: "duplicate-a", text: "Verify", status: "completed" }, + { id: "duplicate-b", text: "Verify", status: "completed" }, + { id: "fallback", text: "Report", status: "pending" }, + ]); + yield* TestClock.adjust("5 seconds"); + yield* ingest(restartedIngestor, [ + { id: "duplicate-a", text: "Verify", status: "completed" }, + { id: "duplicate-b", text: "Verify", status: "completed" }, + { id: "fallback", text: "Report", status: "completed" }, + ]); + + const projection = yield* projectionStore.getThreadProjection(threadEvent.threadId); + const persisted = projection.plans.find( + (candidate): candidate is TodoListPlan => + candidate.kind === "todo_list" && candidate.id === planId, + ); + assert.deepEqual( + persisted?.steps.map(({ id, text, status, durationMs }) => ({ + id, + text, + status, + durationMs, + })), + [ + { id: "duplicate-a", text: "Verify", status: "completed", durationMs: 3_000 }, + { id: "duplicate-b", text: "Verify", status: "completed", durationMs: 4_000 }, + { id: "fallback", text: "Report", status: "completed", durationMs: 5_000 }, + ], + ); + + yield* ingest(restartedIngestor, [ + { id: "duplicate-a", text: "Inserted task", status: "running" }, + { id: "duplicate-b", text: "Verify", status: "completed" }, + { id: "fallback", text: "Different completed task", status: "completed" }, + ]); + yield* TestClock.adjust("2 seconds"); + yield* ingest(restartedIngestor, [ + { id: "duplicate-a", text: "Inserted task", status: "completed" }, + { id: "duplicate-b", text: "Verify", status: "completed" }, + { id: "fallback", text: "Different completed task", status: "completed" }, + ]); + const updated = yield* projectionStore.getThreadProjection(threadEvent.threadId); + const changedPlan = updated.plans.find( + (candidate): candidate is TodoListPlan => + candidate.kind === "todo_list" && candidate.id === planId, + ); + assert.deepEqual( + changedPlan?.steps.map((step) => step.durationMs), + [2_000, 4_000, undefined], + ); + }), + ); + it.effect( "treats successful provider terminal markers as non-persisted orchestration control signals", () => diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts index 4aa16e84d33b..e6764f1e9146 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts @@ -3,6 +3,7 @@ import { CommandId, OrchestrationV2DomainEvent, OrchestrationV2StoredEvent, + type OrchestrationV2PlanArtifact, type OrchestrationV2Run, type OrchestrationV2ProviderTurn, type ModelSelection, @@ -129,6 +130,70 @@ function providerTurnAnalyticsProperties(input: { }; } +type TodoListPlan = Extract; + +function withPlanStepDurations( + plan: TodoListPlan, + previous: TodoListPlan | undefined, + occurredAt: DateTime.Utc, +): TodoListPlan { + const occurredAtIso = DateTime.formatIso(occurredAt); + const occurredAtMs = DateTime.toEpochMillis(occurredAt); + const previousById = new Map(previous?.steps.map((step) => [step.id, step])); + // Provider step IDs may be positional. Changed text must not inherit another task's timing. + const previousStep = (step: TodoListPlan["steps"][number]) => { + const prior = previousById.get(step.id); + return prior?.text === step.text ? prior : undefined; + }; + const hasNewCompletion = plan.steps.some((step) => { + const prior = previousStep(step); + return step.status === "completed" && prior?.status !== "completed"; + }); + let fallbackCompletionConsumed = false; + + return { + ...plan, + steps: plan.steps.map((step) => { + const prior = previousStep(step); + const baseStep = { id: step.id, text: step.text, status: step.status }; + if (step.status === "completed") { + if (prior?.status === "completed") { + return { + ...baseStep, + ...(prior.durationMs === undefined ? {} : { durationMs: prior.durationMs }), + }; + } + const durationAnchorAt = + prior?.status === "running" || !fallbackCompletionConsumed + ? prior?.durationAnchorAt + : occurredAtIso; + fallbackCompletionConsumed = true; + const anchorMs = + durationAnchorAt === undefined ? occurredAtMs : Date.parse(durationAnchorAt); + const durationMs = Number.isFinite(anchorMs) ? Math.max(0, occurredAtMs - anchorMs) : 0; + return { + ...baseStep, + ...(durationMs > 0 ? { durationMs } : {}), + }; + } + if (step.status === "running") { + return { + ...baseStep, + durationAnchorAt: + prior?.status === "running" ? (prior.durationAnchorAt ?? occurredAtIso) : occurredAtIso, + }; + } + return { + ...baseStep, + durationAnchorAt: + hasNewCompletion || prior?.status !== "pending" + ? occurredAtIso + : (prior.durationAnchorAt ?? occurredAtIso), + }; + }), + }; +} + export interface ProviderEventIngestInput { readonly providerSessionId: ProviderSessionId; readonly providerInstanceId: ProviderInstanceId; @@ -202,6 +267,7 @@ export const layer: Layer.Layer< readonly threadId?: ThreadId; readonly runId?: RunId | null; readonly nodeId?: NodeId | null; + readonly occurredAt?: DateTime.Utc; }, ) => Effect.gen(function* () { @@ -210,7 +276,7 @@ export const layer: Layer.Layer< threadId, providerSessionId: input.providerSessionId, }); - const occurredAt = yield* DateTime.now; + const occurredAt = payloadInput.occurredAt ?? (yield* DateTime.now); return yield* decodeDomainEvent( compactUndefined({ id: eventId, @@ -365,16 +431,32 @@ export const layer: Layer.Layer< nodeId: input.event.runtimeRequest.nodeId, }), ]; - case "plan.updated": + case "plan.updated": { + const occurredAt = yield* DateTime.now; + const plan = input.event.plan; + const previous = + plan.kind === "todo_list" + ? yield* projections.getPlan(plan.threadId, plan.id) + : undefined; + const payload = + plan.kind === "todo_list" + ? withPlanStepDurations( + plan, + previous?.kind === "todo_list" ? previous : undefined, + occurredAt, + ) + : plan; return [ yield* makeDomainEvent(input, { type: "plan.updated", - threadId: input.event.plan.threadId, - payload: input.event.plan, - runId: input.event.plan.runId, - nodeId: input.event.plan.nodeId, + threadId: plan.threadId, + payload, + runId: plan.runId, + nodeId: plan.nodeId, + occurredAt, }), ]; + } case "turn.terminal": const dismissed = yield* dismissNativeUserInputs(input, input.event.providerTurnId); if (input.event.status !== "failed") { diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index 7c7e3c058acf..839a5992ef65 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -220,6 +220,7 @@ it.effect( getSettlementCandidates: () => Effect.die("unused getSettlementCandidates"), getThreadProjection: () => Effect.die("control effects must not load transcript"), getRuntimeRecoveryProjection: () => Effect.die("unused getRuntimeRecoveryProjection"), + getPlan: () => Effect.die("unused"), getRuntimeRequest: () => Effect.die("unused getRuntimeRequest"), getRunningTurnContext: () => Effect.die("unused getRunningTurnContext"), getThreadProviderContext: () => Effect.die("unused getThreadProviderContext"), diff --git a/apps/web/src/session-logic.test.ts b/apps/web/src/session-logic.test.ts index d302661121c0..f8e6493f857d 100644 --- a/apps/web/src/session-logic.test.ts +++ b/apps/web/src/session-logic.test.ts @@ -443,6 +443,31 @@ describe("V2 session presentation", () => { }, ); + it("preserves independently derived durations for repeated plan-step labels", () => { + const projection = makeThreadProjectionFixture(); + const runId = RunId.make("run-timed-tasks"); + const planId = PlanId.make("plan-timed-tasks"); + const plan = { + id: planId, + threadId: projection.thread.id, + runId, + nodeId: NodeId.make("node-timed-tasks"), + kind: "todo_list" as const, + status: "active" as const, + steps: [ + { id: "verify-a", text: "Verify", status: "completed" as const, durationMs: 3_000 }, + { id: "verify-b", text: "Verify", status: "completed" as const, durationMs: 4_000 }, + { id: "report", text: "Report", status: "pending" as const }, + ], + }; + + expect(deriveActivePlanState({ ...projection, plans: [plan] }, runId)?.steps).toEqual([ + { step: "Verify", status: "completed", durationMs: 3_000 }, + { step: "Verify", status: "completed", durationMs: 4_000 }, + { step: "Report", status: "pending" }, + ]); + }); + it("keeps failed tool items tool-toned so groups still summarize", () => { const failedCommand = { id: TurnItemId.make("item-failed-command"), diff --git a/apps/web/src/session-logic.ts b/apps/web/src/session-logic.ts index 63c03570fab2..3c68b11c6f2e 100644 --- a/apps/web/src/session-logic.ts +++ b/apps/web/src/session-logic.ts @@ -90,6 +90,7 @@ export interface ActivePlanState { readonly steps: Array<{ readonly step: string; readonly status: "pending" | "inProgress" | "completed"; + readonly durationMs?: number; }>; } @@ -240,9 +241,10 @@ export function deriveActivePlanState( createdAt: planItemTime(projection, plan.id), runId: plan.runId, explanation: plan.explanation ?? null, - steps: plan.steps.map(({ text, status }) => ({ + steps: plan.steps.map(({ text, status, durationMs }) => ({ step: text, status: status === "running" ? "inProgress" : status, + ...(durationMs === undefined ? {} : { durationMs }), })), }; } diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index b490f9ce454f..fc6db06f940e 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -801,6 +801,10 @@ export const OrchestrationV2PlanStep = Schema.Struct({ id: TrimmedNonEmptyString, text: TrimmedNonEmptyString, status: Schema.Literals(["pending", "running", "completed"]), + /** Durable server-owned boundary used to calculate elapsed time. */ + durationAnchorAt: Schema.optional(IsoDateTime), + /** Elapsed time for a completed step. */ + durationMs: Schema.optional(NonNegativeInt), }); export type OrchestrationV2PlanStep = typeof OrchestrationV2PlanStep.Type;