fix(server): phones get the approval push when a subagent waits

apps/server/src/relay/AgentAwarenessRelay.test.ts

@@ -11,6 +11,7 @@ import {
   ProjectId,
   ProviderInstanceId,
   RunId,
+  RuntimeRequestId,
   ThreadId,
   TurnItemId,
 } from "@t3tools/contracts";
@@ -90,6 +91,31 @@ function shell(overrides: Partial<OrchestrationV2ThreadShell> = {}): Orchestrati
   };
 }
 
+const CHILD_THREAD_ID = ThreadId.make("relay-child-thread");
+
+function waitingChild(
+  overrides: Partial<OrchestrationV2ThreadShell> = {},
+): OrchestrationV2ThreadShell {
+  return shell({
+    id: CHILD_THREAD_ID,
+    title: "Child",
+    lineage: {
+      rootThreadId: THREAD_ID,
+      parentThreadId: THREAD_ID,
+      relationshipToParent: "subagent",
+    },
+    createdBy: "agent",
+    creationSource: "mcp",
+    activityRunStatus: "running",
+    pendingRuntimeRequest: {
+      id: RuntimeRequestId.make("child-request"),
+      kind: "command",
+      createdAt: DateTime.makeUnsafe(NOW),
+    },
+    ...overrides,
+  });
+}
+
 const PublishPayload = Schema.Struct({ state: Schema.NullOr(RelayAgentActivityState) });
 const decodePublishPayload = Schema.decodeUnknownSync(Schema.fromJsonString(PublishPayload));
 const unused = () => Effect.die("Unexpected test dependency call");
@@ -133,6 +159,16 @@ describe("startup agent activity", () => {
     });
     assert.deepStrictEqual(ids, [THREAD_ID, newCompleted, newFailed]);
   });
+
+  it("publishes a subagent waiting on the user so its parent is republished", () => {
+    const ids = AgentAwarenessRelay.resolveAgentAwarenessRelayActiveThreadIds({
+      environmentId: EnvironmentId.make("relay-env"),
+      startedAt: DateTime.toEpochMillis(DateTime.makeUnsafe(NOW)),
+      projects: [{ id: PROJECT_ID, title: "Project" }],
+      threads: [shell({ status: "idle" }), waitingChild()],
+    });
+    assert.deepStrictEqual(ids, [CHILD_THREAD_ID]);
+  });
 });
 
 const makeTestRelay = Effect.fnUntraced(function* (
@@ -288,6 +324,64 @@ const makeTestRelay = Effect.fnUntraced(function* (
 });
 
 describe("AgentAwarenessRelay", () => {
+  it.effect("publishes the parent as needing approval while its subagent waits", () =>
+    Effect.gen(function* () {
+      const shells = yield* Ref.make(
+        new Map<ThreadId, OrchestrationV2ThreadShell>([
+          [THREAD_ID, shell({ status: "completed" })],
+          [CHILD_THREAD_ID, waitingChild({ pendingRuntimeRequest: null })],
+        ]),
+      );
+      const setShell = (next: OrchestrationV2ThreadShell) =>
+        Ref.update(shells, (current) => new Map(current).set(next.id, next));
+      const { relay, publications } = yield* makeTestRelay({
+        readShell: (threadId) =>
+          Ref.get(shells).pipe(Effect.map((current) => current.get(threadId) ?? null)),
+      });
+      yield* setShell(shell({ status: "running" }));
+      yield* relay.publishThread(THREAD_ID);
+      yield* setShell(shell({ status: "completed" }));
+      yield* setShell(waitingChild());
+      yield* relay.publishThread(CHILD_THREAD_ID);
+      assert.equal(publications.length, 2);
+      assert.isTrue(publications[1]?.url.includes(`/threads/${THREAD_ID}/`) ?? false);
+      assert.equal(publications[1]?.state?.phase, "waiting_for_approval");
+      assert.equal(publications[1]?.state?.threadId, THREAD_ID);
+
+      yield* setShell(waitingChild({ pendingRuntimeRequest: null }));
+      yield* relay.publishThread(CHILD_THREAD_ID);
+      assert.equal(publications.length, 3);
+      assert.equal(publications[2]?.state?.phase, "completed");
+    }),
+  );
+
+  it.effect("drops a subagent request that cleared while publishing was off", () =>
+    Effect.gen(function* () {
+      const shells = yield* Ref.make(
+        new Map<ThreadId, OrchestrationV2ThreadShell>([
+          [THREAD_ID, shell()],
+          [CHILD_THREAD_ID, waitingChild()],
+        ]),
+      );
+      const { relay, secrets, publications } = yield* makeTestRelay({
+        readShell: (threadId) =>
+          Ref.get(shells).pipe(Effect.map((current) => current.get(threadId) ?? null)),
+      });
+      yield* relay.publishThread(CHILD_THREAD_ID);
+      assert.equal(publications[0]?.state?.phase, "waiting_for_approval");
+
+      yield* secrets.set(PUBLISH_AGENT_ACTIVITY_SECRET, new TextEncoder().encode("false"));
+      yield* Ref.update(shells, (current) =>
+        new Map(current).set(CHILD_THREAD_ID, waitingChild({ pendingRuntimeRequest: null })),
+      );
+      yield* relay.publishThread(CHILD_THREAD_ID);
+      yield* secrets.set(PUBLISH_AGENT_ACTIVITY_SECRET, new TextEncoder().encode("true"));
+      yield* relay.publishThread(THREAD_ID);
+      assert.equal(publications.length, 2);
+      assert.equal(publications[1]?.state?.phase, "running");
+    }),
+  );
+
   it.effect("coalesces queued updates and reruns a thread dirtied during publishing", () =>
     Effect.gen(function* () {
       const started = yield* Deferred.make<void>();

apps/server/src/relay/AgentAwarenessRelay.ts

@@ -11,7 +11,11 @@ import {
   type RelayAgentActivityPublishProofPayload,
   type RelayAgentActivityState,
 } from "@t3tools/contracts/relay";
-import { projectThreadAwarenessV2 } from "@t3tools/shared/agentAwareness";
+import {
+  type PendingSubagentRequest,
+  projectThreadAwarenessV2,
+  subagentPendingRequest,
+} from "@t3tools/shared/agentAwareness";
 import { turnItemUpdateCanEndBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork";
 import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";
 import { withRelayClientTracing } from "@t3tools/shared/relayTracing";
@@ -287,11 +291,17 @@ function describeThreadShellForAwareness(
   };
 }
 
+interface SubagentRequest {
+  readonly topLevelThreadId: ThreadId;
+  readonly request: PendingSubagentRequest;
+}
+
 function resolveAgentAwarenessRelayPublishSnapshot(input: {
   readonly environmentId: EnvironmentId;
   readonly threadId: ThreadId;
   readonly thread: Option.Option<OrchestrationV2ThreadShell>;
   readonly project: Option.Option<Project>;
+  readonly pendingSubagentRequest: PendingSubagentRequest | null;
 }): {
   readonly projectId: string | null;
   readonly state: RelayAgentActivityState | null;
@@ -318,6 +328,7 @@ function resolveAgentAwarenessRelayPublishSnapshot(input: {
         environmentId: input.environmentId,
         project: input.project.value,
         thread: input.thread.value,
+        pendingSubagentRequest: input.pendingSubagentRequest,
       }),
     ),
     reason: "snapshot",
@@ -340,6 +351,8 @@ export function resolveAgentAwarenessRelayActiveThreadIds(input: {
   const projectById = new Map(input.projects.map((project) => [project.id, project]));
   return input.threads
     .filter((thread) => {
+      // Publishing a waiting subagent republishes its top-level thread.
+      if (subagentPendingRequest(thread) !== null) return true;
       const project = projectById.get(thread.projectId);
       if (!project) {
         return false;
@@ -372,6 +385,7 @@ export const make = Effect.gen(function* () {
   const activeSnapshotPublishedRef = yield* Ref.make(false);
   const catchUpRequests = yield* Queue.dropping<void>(1);
   const publishedStateByThreadRef = yield* Ref.make(new Map<ThreadId, string>());
+  const subagentRequests = new Map<ThreadId, SubagentRequest>();
 
   const readSecretString = (name: string) =>
     secrets
@@ -427,6 +441,7 @@ export const make = Effect.gen(function* () {
     Effect.forEach([...publishRetries.keys()], cancelPublishRetry, { discard: true }),
   );
   let schedulePublishRetry: (threadId: ThreadId) => Effect.Effect<void> = () => Effect.void;
+  let enqueueWorkerPublish: (threadId: ThreadId) => Effect.Effect<void> = () => Effect.void;
   const resetPublishedConnection = Effect.gen(function* () {
     publishedRelayConfig = undefined;
     cachedRelayClient = undefined;
@@ -515,6 +530,7 @@ export const make = Effect.gen(function* () {
     // domain event, so materializing the full shell here would make the cost
     // of one thread's activity proportional to how many threads exist.
     const threadShell = yield* threads.getThreadShell(threadId);
+    yield* trackSubagentRequest(threadId, threadShell);
     if (
       threadShell?.lineage.relationshipToParent === "subagent" &&
       !(yield* Ref.get(publishedStateByThreadRef)).has(threadId)
@@ -536,6 +552,7 @@ export const make = Effect.gen(function* () {
       threadId,
       thread,
       project,
+      pendingSubagentRequest: yield* currentSubagentRequestFor(threadId),
     });
     const publishIdentity = agentAwarenessPublishIdentity(snapshot.state);
     const publishedStateByThread = yield* Ref.get(publishedStateByThreadRef);
@@ -627,6 +644,63 @@ export const make = Effect.gen(function* () {
     });
   });
 
+  const findTopLevelThreadId = Effect.fnUntraced(function* (child: OrchestrationV2ThreadShell) {
+    const seen = new Set<ThreadId>([child.id]);
+    let current: OrchestrationV2ThreadShell | null = child;
+    while (current?.lineage.relationshipToParent === "subagent") {
+      const parentId: ThreadId | null = current.lineage.parentThreadId;
+      if (parentId === null || seen.has(parentId)) return null;
+      seen.add(parentId);
+      current = yield* threads.getThreadShell(parentId);
+    }
+    return current?.id ?? null;
+  });
+
+  // A subagent's request changes its top-level thread's row, so that row is
+  // republished whenever the child's request appears, changes or clears.
+  const trackSubagentRequest = Effect.fnUntraced(function* (
+    threadId: ThreadId,
+    threadShell: OrchestrationV2ThreadShell | null,
+  ) {
+    const request =
+      threadShell === null || threadShell.archivedAt !== null
+        ? null
+        : subagentPendingRequest(threadShell);
+    const previous = subagentRequests.get(threadId);
+    if (previous?.request === (request ?? undefined)) return;
+    if (request === null || threadShell === null) {
+      if (previous === undefined) return;
+      subagentRequests.delete(threadId);
+      yield* enqueueWorkerPublish(previous.topLevelThreadId);
+      return;
+    }
+    const topLevelThreadId =
+      previous?.topLevelThreadId ?? (yield* findTopLevelThreadId(threadShell));
+    if (topLevelThreadId === null) return;
+    subagentRequests.set(threadId, { topLevelThreadId, request });
+    yield* enqueueWorkerPublish(topLevelThreadId);
+  });
+
+  // The map can miss a child's update while publishing is off or the relay is
+  // unlinked, so each waiting child is re-read before its parent is published.
+  const currentSubagentRequestFor = Effect.fnUntraced(function* (topLevelThreadId: ThreadId) {
+    let pending: PendingSubagentRequest | null = null;
+    for (const [childId, entry] of [...subagentRequests]) {
+      if (entry.topLevelThreadId !== topLevelThreadId) continue;
+      const child = yield* threads.getThreadShell(childId);
+      const request =
+        child === null || child.archivedAt !== null ? null : subagentPendingRequest(child);
+      if (request === null) {
+        subagentRequests.delete(childId);
+        continue;
+      }
+      if (request !== entry.request) subagentRequests.set(childId, { ...entry, request });
+      if (request === "approval") return request;
+      pending = request;
+    }
+    return pending;
+  });
+
   const processThreadPublish = Effect.fnUntraced(function* (threadId: ThreadId) {
     const retry = publishRetries.get(threadId);
     if (retry?.timer !== undefined) {
@@ -648,6 +722,7 @@ export const make = Effect.gen(function* () {
   });
 
   const worker = yield* makeAgentAwarenessPublishWorker(processThreadPublish);
+  enqueueWorkerPublish = worker.enqueue;
   const enqueueThreadPublish = (threadId: ThreadId) =>
     cancelPublishRetry(threadId).pipe(Effect.andThen(worker.enqueue(threadId)));
   const publishThread: AgentAwarenessRelay["Service"]["publishThread"] = (threadId) =>

packages/client-runtime/src/state/models.ts

@@ -1,3 +1,4 @@
+import type { PendingSubagentRequest } from "@t3tools/shared/agentAwareness";
 import { backgroundWorkHoldsCompletion } from "@t3tools/shared/orchestrationV2PendingBackgroundWork";
 import { threadPullRequestsOf } from "@t3tools/shared/threadPullRequests";
 import type {
@@ -245,7 +246,7 @@ export function scopeProject(
   return { ...project, environmentId };
 }
 
-export type PendingSubagentRequest = "approval" | "user_input";
+export type { PendingSubagentRequest } from "@t3tools/shared/agentAwareness";
 
 export function presentThreadShell(
   environmentId: EnvironmentId,

packages/client-runtime/src/state/threadShell.ts

@@ -6,7 +6,8 @@ import type {
   ScopedThreadRef,
   ThreadId,
 } from "@t3tools/contracts";
-import { isProviderNativeSubagentThread, OrchestrationV2ThreadShell } from "@t3tools/contracts";
+import { OrchestrationV2ThreadShell } from "@t3tools/contracts";
+import { subagentPendingRequest } from "@t3tools/shared/agentAwareness";
 import * as Schema from "effect/Schema";
 import { Atom } from "effect/reactivity";
 
@@ -31,43 +32,13 @@ const EMPTY_THREAD_REFS_BY_PROJECT: ReadonlyMap<
   ReadonlyArray<ScopedThreadRef>
 > = new Map();
 
-// An async question outlives the subagent's run, but once that run ends the
-// parent is no longer waiting on the answer, so it stops counting as input.
-// The activity run stays set when a follow-up is queued behind a blocked run;
-// older servers omit it, so fall back to the latest run's status.
-const isChildRunLive = (
-  thread: Pick<OrchestrationV2ThreadShell, "activityRunStatus" | "status">,
-) =>
-  thread.activityRunStatus !== undefined
-    ? thread.activityRunStatus !== null
-    : thread.status === "preparing" ||
-      thread.status === "starting" ||
-      thread.status === "running" ||
-      thread.status === "waiting";
-
-/** Provider-native
-    subagents ask on their parent, so they are skipped. */
-function subagentRequest(thread: OrchestrationV2ThreadShell): PendingSubagentRequest | null {
-  const kind = thread.pendingRuntimeRequest?.kind;
-  if (
-    kind === undefined ||
-    kind === "auth_refresh" ||
-    thread.lineage.relationshipToParent !== "subagent" ||
-    isProviderNativeSubagentThread(thread) ||
-    !isChildRunLive(thread)
-  ) {
-    return null;
-  }
-  return kind === "user_input" ? "user_input" : "approval";
-}
-
 export function pendingSubagentRequestsByThread(
   threads: ReadonlyArray<OrchestrationV2ThreadShell>,
 ): ReadonlyMap<ThreadId, PendingSubagentRequest> {
   const requests = new Map<ThreadId, PendingSubagentRequest>();
   let byId: ReadonlyMap<ThreadId, OrchestrationV2ThreadShell> | undefined;
   for (const child of threads) {
-    const request = subagentRequest(child);
+    const request = subagentPendingRequest(child);
     if (request === null) continue;
     byId ??= new Map(threads.map((thread) => [thread.id, thread] as const));
     const seen = new Set<ThreadId>([child.id]);
@@ -302,7 +273,7 @@ export function createEnvironmentThreadShellAtoms(input: {
           if (seen.has(child.id)) continue;
           seen.add(child.id);
           pending.push(child.id);
-          if (subagentRequest(child) !== null) {
+          if (subagentPendingRequest(child) !== null) {
             next.push(scopedThread(ref.environmentId, child, subagentRequests));
           }
         }