fix(orchestrator): preserve task-step elapsed time across restart

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<ProjectionPendingUserInputs, ProjectionStoreV2Error>;
+  readonly getPlan: (
+    threadId: ThreadId,
+    planId: PlanId,
+  ) => Effect.Effect<OrchestrationV2PlanArtifact | undefined, ProjectionStoreV2Error>;
   readonly getRuntimeRequest: (
     threadId: ThreadId,
     requestId: RuntimeRequestId,
@@ -3461,6 +3467,21 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
         )
         .pipe(Effect.mapError(controlReadError(threadId)));
 
+    const getPlan: ProjectionStoreV2Shape["getPlan"] = (threadId, planId) =>
+      sql
+        .withTransaction(
+          Effect.gen(function* () {
+            yield* requireThread(threadId);
+            const rows =
+              yield* sql<PayloadRow>`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<ProjectionStoreV2, never, SqlClient.SqlClient> =
       getCheckpointContext,
       getPendingNativeUserInputs,
       getRuntimeRequest,
+      getPlan,
       getProviderControlContext,
       getRecoveryThreadIds,
       getUnreadableThreadIds,
@@ -4514,6 +4536,13 @@ export const layerMemory: Layer.Layer<ProjectionStoreV2> = 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);

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<OrchestrationV2PlanArtifact, { readonly kind: "todo_list" }>;
+      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",
     () =>

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<OrchestrationV2PlanArtifact, { readonly kind: "todo_list" }>;
+
+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") {

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"),

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"),