fix(orchestration): settle subagent statuses and activation timers

apps/mobile/src/features/threads/ThreadAgentsSheet.tsx

@@ -1,7 +1,12 @@
 import { useAtomValue } from "@effect/atom-react";
 import type { ThreadTurnSubagents } from "@t3tools/client-runtime/state/thread-subagents";
-import type { EnvironmentId, OrchestrationV2Subagent, ThreadId } from "@t3tools/contracts";
-import { formatDuration } from "@t3tools/shared/orchestrationTiming";
+import {
+  isOrchestrationV2WorkActive,
+  type EnvironmentId,
+  type OrchestrationV2Subagent,
+  type ThreadId,
+} from "@t3tools/contracts";
+import { deriveSubagentElapsedMs, formatDuration } from "@t3tools/shared/orchestrationTiming";
 import { StackActions, useNavigation, type StaticScreenProps } from "@react-navigation/native";
 import * as DateTime from "effect/DateTime";
 import * as Haptics from "expo-haptics";
@@ -175,19 +180,23 @@ function AgentRow(props: {
  * shared second tick, so a settled sheet never repaints.
  */
 function useSubagentElapsed(
-  subagent: Pick<OrchestrationV2Subagent, "startedAt" | "completedAt">,
+  subagent: Pick<OrchestrationV2Subagent, "status" | "startedAt" | "completedAt">,
   tickSeconds: boolean,
 ): string | null {
   const [nowMs, setNowMs] = useState(() => Date.now());
-  const running = subagent.completedAt === null;
+  const running = isOrchestrationV2WorkActive(subagent.status);
   useEffect(() => {
     if (!tickSeconds || !running) return;
     const intervalId = setInterval(() => setNowMs(Date.now()), 1_000);
     return () => clearInterval(intervalId);
   }, [running, tickSeconds]);
-  if (subagent.startedAt === null) return null;
-  const startedAtMs = DateTime.toEpochMillis(subagent.startedAt);
-  const endMs =
-    subagent.completedAt === null ? nowMs : DateTime.toEpochMillis(subagent.completedAt);
-  return endMs <= startedAtMs ? null : formatDuration(endMs - startedAtMs);
+  const elapsedMs = deriveSubagentElapsedMs(
+    {
+      status: subagent.status,
+      startedAt: subagent.startedAt === null ? null : DateTime.formatIso(subagent.startedAt),
+      completedAt: subagent.completedAt === null ? null : DateTime.formatIso(subagent.completedAt),
+    },
+    nowMs,
+  );
+  return elapsedMs === null || elapsedMs === 0 ? null : formatDuration(elapsedMs);
 }

apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts

@@ -29,6 +29,7 @@ import {
 import { assert, describe, it } from "@effect/vitest";
 import * as Context from "effect/Context";
 import * as DateTime from "effect/DateTime";
+import { TestClock } from "effect/testing";
 import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
 import * as Exit from "effect/Exit";
@@ -4861,6 +4862,9 @@ describe("ClaudeAdapterV2 background wake turns", () => {
         assert.equal(subagentEvents().at(-1)?.subagent.result, FIRST_SUMMARY);
         assert.isFalse(yield* harness.hasPendingBackgroundWork);
 
+        const firstStartedAt = subagentEvents().at(-1)?.subagent.startedAt;
+        yield* TestClock.adjust("30 seconds");
+
         // A user turn nudges the completed subagent via SendMessage; the
         // resume task_started re-opens the row across turn contexts (the new
         // turn's maps are empty, so this exercises the session registry).
@@ -4930,6 +4934,11 @@ describe("ClaudeAdapterV2 background wake turns", () => {
         );
         const reopened = subagentEvents().at(-1)?.subagent;
         assert.isNull(reopened?.result);
+        assert.isNull(reopened?.completedAt);
+        assert.equal(
+          DateTime.toEpochMillis(reopened!.startedAt!) - DateTime.toEpochMillis(firstStartedAt!),
+          30_000,
+        );
         // The reopen re-attributes the subagent to the resuming run:
         // RunExecutionService routes parent-thread events by runId, and the
         // launch run's ingestion fiber stops once its child subagents

apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts

@@ -3408,7 +3408,8 @@ export function makeClaudeAdapterV2(
             ...(input.model === undefined ? {} : { model: input.model }),
             ...(input.progress === undefined ? {} : { progress: input.progress }),
             ...(input.result === undefined ? {} : { result: input.result }),
-            completedAt: input.status === "running" ? null : now,
+            ...(isReopen ? { startedAt: now } : {}),
+            completedAt: input.status === "running" ? null : (priorTask?.completedAt ?? now),
             updatedAt: now,
           } satisfies OrchestrationV2Subagent;
           const subagent = {
@@ -3505,7 +3506,7 @@ export function makeClaudeAdapterV2(
                 runtimeRequestId: null,
                 checkpointScopeId: null,
                 startedAt: task.startedAt,
-                completedAt: input.status === "running" ? null : now,
+                completedAt: task.completedAt,
               },
             });
             yield* emitProviderEvent({
@@ -3526,7 +3527,7 @@ export function makeClaudeAdapterV2(
                 runtimeRequestId: null,
                 checkpointScopeId: null,
                 startedAt: task.startedAt,
-                completedAt: input.status === "running" ? null : now,
+                completedAt: task.completedAt,
               },
             });
           }
@@ -4389,6 +4390,7 @@ export function makeClaudeAdapterV2(
                   ...priorTask,
                   status: "running",
                   result: null,
+                  startedAt: now,
                   completedAt: null,
                   updatedAt: now,
                 },

apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts

@@ -5487,7 +5487,10 @@ describe("CodexAdapterV2 post-settle continuation", () => {
       method: "turn/started",
       params: {
         threadId: RESUME_CHILD_THREAD,
-        turn: makeCodexReplayTurn({ id: turnId, status: "inProgress" }),
+        turn: {
+          ...makeCodexReplayTurn({ id: turnId, status: "inProgress" }),
+          startedAt: turnId === RESUME_CHILD_TURN_2 ? 1782622470 : 1782622440,
+        },
       },
     },
   });
@@ -5639,6 +5642,8 @@ describe("CodexAdapterV2 post-settle continuation", () => {
         );
         const reopened = harness.subagentUpdates()[settledUpdateCount];
         assert.equal(reopened?.subagent.status, "running");
+        assert.equal(DateTime.toEpochMillis(reopened!.subagent.startedAt!), 1782622470000);
+        assert.isNull(reopened!.subagent.completedAt);
         assert.isTrue(yield* harness.hasPendingBackgroundWork);
 
         yield* TestClock.adjust("30 seconds");
@@ -5658,6 +5663,248 @@ describe("CodexAdapterV2 post-settle continuation", () => {
     ),
   );
 
+  it.effect("rejects duplicate child starts across parent runs", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const firstDone = yield* Deferred.make<void>();
+        const resumed = yield* Deferred.make<void>();
+        const secondTurn = "native-parent-resume-turn";
+        const secondPrompt = "Resume the child.";
+        const entries = [...resumeSubagentTranscript.entries];
+        const resumeIndex = entries.findIndex(
+          (e) => e.type === "emit_inbound" && e.label === `turn/started/${RESUME_CHILD_TURN_2}`,
+        );
+        const suffix = entries.splice(resumeIndex);
+        for (const entry of codexReplayPreamble({
+          nativeThreadId: RESUME_NATIVE_THREAD,
+          nativeTurnId: secondTurn,
+          prompt: secondPrompt,
+        }).slice(-3)) {
+          entries.push(
+            entry.type === "expect_outbound" || entry.type === "emit_inbound"
+              ? {
+                  ...entry,
+                  frame:
+                    Predicate.isObject(entry.frame) && "id" in entry.frame
+                      ? { ...entry.frame, id: 4 }
+                      : entry.frame,
+                }
+              : entry,
+          );
+        }
+        entries.push(childTurnStarted(RESUME_CHILD_TURN_1));
+        entries.push(...suffix);
+        const harness = yield* makeCodexReplayHarness(
+          makeCodexReplayTranscript({
+            scenario: "codex-cross-run-resume",
+            entries,
+          }),
+          (event) =>
+            event.type === "turn.terminal"
+              ? Deferred.succeed(firstDone, undefined)
+              : event.type === "subagent.updated" && event.subagent.runId === "run-cross-run-second"
+                ? Deferred.succeed(resumed, undefined)
+                : Effect.void,
+        );
+        const now = yield* DateTime.now;
+        yield* harness.runtime.startTurn(
+          makeCodexTestTurnInput({
+            threadId: harness.threadId,
+            providerThread: harness.providerThread,
+            now,
+            attemptId: RunAttemptId.make("cross-run-first"),
+            text: RESUME_PROMPT,
+          }),
+        );
+        yield* TestClock.adjust("100 millis");
+        yield* Deferred.await(firstDone);
+        yield* harness.runtime.startTurn({
+          ...makeCodexTestTurnInput({
+            threadId: harness.threadId,
+            providerThread: harness.providerThread,
+            now,
+            attemptId: RunAttemptId.make("cross-run-second"),
+            text: secondPrompt,
+          }),
+          runOrdinal: 2,
+          providerTurnOrdinal: 2,
+        });
+        yield* TestClock.adjust("30 seconds");
+        yield* Deferred.await(resumed);
+        const row = harness
+          .subagentUpdates()
+          .find((e) => e.subagent.runId === "run-cross-run-second")?.subagent;
+        assert.equal(row?.status, "running");
+        assert.equal(row?.parentNodeId, "node-cross-run-second");
+        assert.isNull(row?.completedAt);
+        assert.isNotNull(row?.startedAt);
+        assert.equal(DateTime.toEpochMillis(row!.startedAt!), 1782622470000);
+      }).pipe(Effect.provide(Layer.merge(idAllocatorLayer, NodeServices.layer))),
+    ),
+  );
+
+  for (const [nativeStatus, expectedStatus] of [
+    ["pendingInit", "pending"],
+    ["running", "running"],
+    ["interrupted", "interrupted"],
+    ["shutdown", "cancelled"],
+    ["notFound", "failed"],
+    ["errored", "failed"],
+    ["completed", "completed"],
+    ["activity-completed", "completed"],
+    ["late-activity-completed", "completed"],
+    ["stale-running", "completed"],
+    ["duplicate-completed", "completed"],
+  ] as const) {
+    it.effect(`normalizes subagent ${nativeStatus} without losing its lifecycle`, () =>
+      Effect.scoped(
+        Effect.gen(function* () {
+          const marker = yield* Deferred.make<void>();
+          const firstCompletion = yield* Deferred.make<void>();
+          const stateEntry = (
+            status: string,
+            id: string,
+          ): Extract<CodexReplay.CodexAppServerReplayEntry, { type: "emit_inbound" }> => ({
+            type: "emit_inbound",
+            label: id,
+            frame: {
+              method: "item/completed",
+              params: {
+                threadId: RESUME_NATIVE_THREAD,
+                turnId: RESUME_NATIVE_TURN,
+                item: {
+                  type: "collabAgentToolCall",
+                  id,
+                  tool: "listAgents",
+                  status: "completed",
+                  senderThreadId: RESUME_NATIVE_THREAD,
+                  receiverThreadIds: [RESUME_CHILD_THREAD],
+                  agentsStates: { [RESUME_CHILD_THREAD]: { status, message: null } },
+                },
+              },
+            },
+          });
+          const entries: Array<CodexReplay.CodexAppServerReplayEntry> = [
+            ...codexReplayPreamble({
+              nativeThreadId: RESUME_NATIVE_THREAD,
+              nativeTurnId: RESUME_NATIVE_TURN,
+              prompt: RESUME_PROMPT,
+            }),
+            resumeSubagentTranscript.entries.find(
+              (e) =>
+                e.type === "emit_inbound" && e.label === "item/completed/subAgentActivity-started",
+            )!,
+          ];
+          if (nativeStatus === "late-activity-completed") {
+            entries.push(
+              resumeSubagentTranscript.entries.find(
+                (e) => e.type === "emit_inbound" && e.label === "turn/completed/root",
+              )!,
+            );
+          }
+          if (nativeStatus === "activity-completed" || nativeStatus === "late-activity-completed") {
+            entries.push({
+              type: "emit_inbound",
+              label: "activity-done",
+              frame: {
+                method: "item/completed",
+                params: {
+                  threadId: RESUME_NATIVE_THREAD,
+                  turnId: RESUME_NATIVE_TURN,
+                  item: {
+                    type: "subAgentActivity",
+                    id: "activity-done",
+                    kind: "completed",
+                    agentThreadId: RESUME_CHILD_THREAD,
+                    agentPath: "/root/resume_agent",
+                  },
+                },
+              },
+            });
+          } else if (nativeStatus === "stale-running" || nativeStatus === "duplicate-completed") {
+            entries.push(stateEntry("completed", "child-completed"), {
+              ...stateEntry(
+                nativeStatus === "stale-running" ? "running" : "completed",
+                "trailing-snapshot",
+              ),
+              afterMs: 100,
+            });
+          } else {
+            entries.push(stateEntry(nativeStatus, "status-update"));
+          }
+          // A known child's turn provides a receipt even after the parent context is released.
+          if (nativeStatus === "late-activity-completed") {
+            entries.push({
+              type: "emit_inbound",
+              label: "late-marker",
+              frame: {
+                method: "turn/started",
+                params: {
+                  threadId: RESUME_CHILD_THREAD,
+                  turn: makeCodexReplayTurn({ id: RESUME_CHILD_TURN_1, status: "inProgress" }),
+                },
+              },
+            });
+          } else {
+            entries.push({
+              type: "emit_inbound",
+              label: "marker",
+              frame: {
+                method: "item/completed",
+                params: {
+                  threadId: RESUME_NATIVE_THREAD,
+                  turnId: RESUME_NATIVE_TURN,
+                  item: {
+                    type: "agentMessage",
+                    id: "marker",
+                    text: "LIFECYCLE_MARKER",
+                    phase: "final_answer",
+                    memoryCitation: null,
+                  },
+                },
+              },
+            });
+          }
+          const harness = yield* makeCodexReplayHarness(
+            makeCodexReplayTranscript({ scenario: `subagent-${nativeStatus}`, entries }),
+            (event) =>
+              (event.type === "message.updated" && event.message.text === "LIFECYCLE_MARKER") ||
+              (nativeStatus === "late-activity-completed" &&
+                event.type === "provider_turn.updated" &&
+                event.providerTurn.nativeTurnRef?.nativeId === RESUME_CHILD_TURN_1)
+                ? Deferred.succeed(marker, undefined)
+                : event.type === "subagent.updated" && event.subagent.status === "completed"
+                  ? Deferred.succeed(firstCompletion, undefined)
+                  : Effect.void,
+          );
+          yield* harness.runtime.startTurn(
+            makeCodexTestTurnInput({
+              threadId: harness.threadId,
+              providerThread: harness.providerThread,
+              now: yield* DateTime.now,
+              attemptId: RunAttemptId.make(`subagent-${nativeStatus}`),
+              text: RESUME_PROMPT,
+            }),
+          );
+          if (nativeStatus === "stale-running" || nativeStatus === "duplicate-completed") {
+            yield* Deferred.await(firstCompletion);
+            yield* TestClock.adjust("100 millis");
+          }
+          yield* Deferred.await(marker);
+          const latest = harness.subagentUpdates().at(-1)!.subagent;
+          assert.equal(latest.status, expectedStatus);
+          if (nativeStatus === "duplicate-completed") {
+            const first = harness.subagentUpdates().find((e) => e.subagent.status === "completed")!;
+            assert.equal(
+              DateTime.toEpochMillis(latest.completedAt!),
+              DateTime.toEpochMillis(first.subagent.completedAt!),
+            );
+          }
+        }).pipe(Effect.provide(Layer.merge(idAllocatorLayer, NodeServices.layer))),
+      ),
+    );
+  }
+
   const codexReplayThreadResult = (input: {
     readonly nativeThreadId: string;
     readonly forkedFromId: string | null;