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,148 @@ 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");
+
+      // Phones key alerts by phase and time, so a second wait needs a new time.
+      yield* setShell(
+        waitingChild({
+          pendingRuntimeRequest: {
+            id: RuntimeRequestId.make("child-request-2"),
+            kind: "command",
+            createdAt: DateTime.makeUnsafe("2026-09-04T12:05:00.000Z"),
+          },
+        }),
+      );
+      yield* relay.publishThread(CHILD_THREAD_ID);
+      assert.equal(publications.length, 4);
+      assert.equal(publications[3]?.state?.phase, "waiting_for_approval");
+      assert.equal(publications[3]?.state?.updatedAt, "2026-09-04T12:05:00.000Z");
+      assert.notEqual(publications[3]?.state?.updatedAt, publications[1]?.state?.updatedAt);
+    }),
+  );
+
+  it.effect(
+    "retries a parent publish for a new subagent request after earlier retries ran out",
+    () =>
+      Effect.gen(function* () {
+        const shells = yield* Ref.make(
+          new Map<ThreadId, OrchestrationV2ThreadShell>([
+            [THREAD_ID, shell()],
+            [CHILD_THREAD_ID, waitingChild({ pendingRuntimeRequest: null })],
+          ]),
+        );
+        const { relay, publications } = yield* makeTestRelay({
+          readShell: (threadId) =>
+            Ref.get(shells).pipe(Effect.map((current) => current.get(threadId) ?? null)),
+          respond: (attempt) =>
+            attempt <= 7
+              ? new Response("unavailable", { status: 503 })
+              : Response.json({ ok: true, deliveries: [] }),
+        });
+        yield* relay.publishThread(THREAD_ID);
+        for (const delay of [
+          "1 second",
+          "2 seconds",
+          "4 seconds",
+          "8 seconds",
+          "16 seconds",
+        ] as const) {
+          yield* TestClock.adjust(delay);
+          yield* relay.drain;
+        }
+        assert.equal(publications.length, 6);
+
+        yield* Ref.update(shells, (current) =>
+          new Map(current).set(CHILD_THREAD_ID, waitingChild()),
+        );
+        yield* relay.publishThread(CHILD_THREAD_ID);
+        assert.equal(publications.length, 7);
+        yield* TestClock.adjust("1 second");
+        yield* relay.drain;
+        assert.equal(publications.length, 8);
+        assert.equal(publications[7]?.state?.phase, "waiting_for_approval");
+      }),
+  );
+
+  it.effect("keeps a subagent request that started while publishing was off", () =>
+    Effect.gen(function* () {
+      const shells = yield* Ref.make(
+        new Map<ThreadId, OrchestrationV2ThreadShell>([
+          [THREAD_ID, shell()],
+          [CHILD_THREAD_ID, waitingChild({ pendingRuntimeRequest: null })],
+        ]),
+      );
+      const { relay, secrets, publications } = yield* makeTestRelay({
+        readShell: (threadId) =>
+          Ref.get(shells).pipe(Effect.map((current) => current.get(threadId) ?? null)),
+      });
+      yield* relay.publishThread(THREAD_ID);
+      assert.equal(publications[0]?.state?.phase, "running");
+
+      yield* secrets.set(PUBLISH_AGENT_ACTIVITY_SECRET, new TextEncoder().encode("false"));
+      yield* Ref.update(shells, (current) => new Map(current).set(CHILD_THREAD_ID, waitingChild()));
+      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, "waiting_for_approval");
+    }),
+  );
+
+  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,12 @@ import {
   type RelayAgentActivityPublishProofPayload,
   type RelayAgentActivityState,
 } from "@t3tools/contracts/relay";
-import { projectThreadAwarenessV2 } from "@t3tools/shared/agentAwareness";
+import {
+  type PendingSubagentRequest,
+  type SubagentRequestAwareness,
+  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 +292,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: SubagentRequestAwareness | null;
 }): {
   readonly projectId: string | null;
   readonly state: RelayAgentActivityState | null;
@@ -318,6 +329,7 @@ function resolveAgentAwarenessRelayPublishSnapshot(input: {
         environmentId: input.environmentId,
         project: input.project.value,
         thread: input.thread.value,
+        pendingSubagentRequest: input.pendingSubagentRequest,
       }),
     ),
     reason: "snapshot",
@@ -340,6 +352,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 +386,9 @@ 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>();
+  // Environments that never publish skip the extra shell read while paused.
+  let hasPublished = false;
 
   const readSecretString = (name: string) =>
     secrets
@@ -427,6 +444,7 @@ export const make = Effect.gen(function* () {
     Effect.forEach([...publishRetries.keys()], cancelPublishRetry, { discard: true }),
   );
   let schedulePublishRetry: (threadId: ThreadId) => Effect.Effect<void> = () => Effect.void;
+  let enqueueTopLevelPublish: (threadId: ThreadId) => Effect.Effect<void> = () => Effect.void;
   const resetPublishedConnection = Effect.gen(function* () {
     publishedRelayConfig = undefined;
     cachedRelayClient = undefined;
@@ -439,6 +457,7 @@ export const make = Effect.gen(function* () {
     if (!publishAgentActivity) {
       yield* cancelPublishRetries;
       yield* resetPublishedConnection;
+      yield* trackSubagentWhilePaused(threadId);
       yield* Effect.logDebug("agent activity publish skipped; publication disabled", {
         threadId,
       });
@@ -448,6 +467,7 @@ export const make = Effect.gen(function* () {
     if (!relayConfig) {
       yield* cancelPublishRetries;
       yield* resetPublishedConnection;
+      yield* trackSubagentWhilePaused(threadId);
       yield* Effect.logDebug("agent activity publish skipped; relay link credentials unavailable", {
         threadId,
       });
@@ -515,6 +535,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 +557,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);
@@ -615,6 +637,7 @@ export const make = Effect.gen(function* () {
       state: snapshot.state,
       reason: snapshot.reason,
     }).pipe(withRelayClientTracing);
+    hasPublished = true;
     publishConfirmDeadlines.delete(threadId);
     yield* Ref.update(publishedStateByThreadRef, (publishedStates) => {
       const nextPublishedStates = new Map(publishedStates);
@@ -627,6 +650,71 @@ 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* enqueueTopLevelPublish(previous.topLevelThreadId);
+      return;
+    }
+    const topLevelThreadId =
+      previous?.topLevelThreadId ?? (yield* findTopLevelThreadId(threadShell));
+    if (topLevelThreadId === null) return;
+    subagentRequests.set(threadId, { topLevelThreadId, request });
+    yield* enqueueTopLevelPublish(topLevelThreadId);
+  });
+
+  // A child that starts waiting while publishing is paused must still count
+  // when its parent is next published.
+  const trackSubagentWhilePaused = Effect.fnUntraced(function* (threadId: ThreadId) {
+    if (!hasPublished) return;
+    yield* trackSubagentRequest(threadId, yield* threads.getThreadShell(threadId));
+  });
+
+  // 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: SubagentRequestAwareness | 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);
+      const requestedAt = child?.pendingRuntimeRequest?.createdAt;
+      if (request === null || requestedAt === undefined) {
+        subagentRequests.delete(childId);
+        continue;
+      }
+      if (request !== entry.request) subagentRequests.set(childId, { ...entry, request });
+      if (request === "approval") return { kind: request, requestedAt };
+      pending = { kind: request, requestedAt };
+    }
+    return pending;
+  });
+
   const processThreadPublish = Effect.fnUntraced(function* (threadId: ThreadId) {
     const retry = publishRetries.get(threadId);
     if (retry?.timer !== undefined) {
@@ -650,6 +738,9 @@ export const make = Effect.gen(function* () {
   const worker = yield* makeAgentAwarenessPublishWorker(processThreadPublish);
   const enqueueThreadPublish = (threadId: ThreadId) =>
     cancelPublishRetry(threadId).pipe(Effect.andThen(worker.enqueue(threadId)));
+  // A child's new request is fresh activity for its parent, so it gets a fresh
+  // retry budget like any other event.
+  enqueueTopLevelPublish = enqueueThreadPublish;
   const publishThread: AgentAwarenessRelay["Service"]["publishThread"] = (threadId) =>
     enqueueThreadPublish(threadId).pipe(Effect.andThen(worker.drain));