From f581f5a0643b11f91f8e20f2fcd41a601dd541b2 Mon Sep 17 00:00:00 2001 From: Stanislav Zmiev Date: Tue, 10 Nov 2026 01:29:42 +0400 Subject: [PATCH 1/5] fix: notify when a delegated agent waits on a secret card A child started with delegate_task that calls request_secret raised a card no one was told about: no desktop notification, no mobile push, and the parent row showed Done. The card expired unanswered. The projection reports a waiting secret or link card as a user_input request, and both roll-ups drop a child's user_input because the coordinating agent answers child questions. It can't answer a secret, and it isn't told about one. Mark secret and link cards as user-only in the pending-request summary, and let user-only requests through the desktop roll-up and the relay roll-up. The relay publishes the parent as waiting for input. Plain child questions are unchanged. --- .../orchestration-v2/ProjectionStore.test.ts | 5 ++++ .../src/orchestration-v2/ProjectionStore.ts | 1 + .../src/relay/AgentAwarenessRelay.test.ts | 22 ++++++++++++++ apps/server/src/relay/AgentAwarenessRelay.ts | 12 +++++--- .../src/state/threadShell.test.ts | 30 +++++++++++++++++++ .../client-runtime/src/state/threadShell.ts | 15 ++++++---- packages/contracts/src/orchestrationV2.ts | 2 ++ packages/shared/src/agentAwareness.ts | 21 ++++++++----- 8 files changed, 92 insertions(+), 16 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 1ca7eb67036fa..16e9506f7463c 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -6664,6 +6664,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { ]) { assert.equal(shell?.pendingRuntimeRequest?.id, "secret:b"); assert.isTrue(shell?.pendingRuntimeRequest?.blocking); + assert.isTrue(shell?.pendingRuntimeRequest?.userOnly); } const requestId = RuntimeRequestId.make(`request:${suffix}:async`); @@ -6755,6 +6756,10 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { [thread?.pendingRuntimeRequest?.blocking, memoryShell.pendingRuntimeRequest?.blocking], terminal ? [false, false] : [true, true], ); + assert.deepEqual( + [thread?.pendingRuntimeRequest?.userOnly, memoryShell.pendingRuntimeRequest?.userOnly], + terminal ? [undefined, undefined] : [true, true], + ); } }), ); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 3d33f76b7a494..3ce18601c8f0c 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1515,6 +1515,7 @@ function secretRequestAsPendingInput( kind: "user_input", createdAt: item.startedAt ?? item.updatedAt, blocking: true, + userOnly: true, }; } diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index ccd28c165f14e..8ef5df3d1b9c6 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -375,6 +375,28 @@ describe("AgentAwarenessRelay", () => { }), ); + it.effect("publishes the parent as needing input while its subagent waits on a secret card", () => + Effect.gen(function* () { + const child = waitingChild({ + pendingRuntimeRequest: { + id: RuntimeRequestId.make("child-secret-card"), + kind: "user_input", + createdAt: DateTime.makeUnsafe(NOW), + blocking: true, + userOnly: true, + }, + }); + const parent = shell({ status: "completed" }); + const { relay, publications } = yield* makeTestRelay({ + readShell: (threadId) => Effect.succeed(threadId === child.id ? child : parent), + }); + yield* relay.publishThread(CHILD_THREAD_ID); + assert.equal(publications.length, 1); + assert.isTrue(publications[0]?.url.includes(`/threads/${THREAD_ID}/`) ?? false); + assert.equal(publications[0]?.state?.phase, "waiting_for_input"); + }), + ); + it.effect("republishes a completed parent for its waiting child after a relink", () => Effect.gen(function* () { const parent = shell({ status: "completed" }); diff --git a/apps/server/src/relay/AgentAwarenessRelay.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index c14558762183a..3ab86280ca33c 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -16,7 +16,7 @@ import { import { type SubagentRequestAwareness, projectThreadAwarenessV2, - subagentAwaitsApproval, + subagentAwaitsUser, } from "@t3tools/shared/agentAwareness"; import { turnItemUpdateCanEndBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; @@ -308,7 +308,7 @@ interface SubagentAlert { } const waitingApprovalId = (thread: AgentAwarenessThread | null) => - thread !== null && thread.archivedAt === null && subagentAwaitsApproval(thread) + thread !== null && thread.archivedAt === null && subagentAwaitsUser(thread) ? thread.pendingRuntimeRequest?.id : undefined; @@ -377,7 +377,7 @@ export function resolveAgentAwarenessRelayActiveThreadIds(input: { return input.threads .filter((thread) => { // Publishing a waiting subagent republishes its top-level thread. - if (subagentAwaitsApproval(thread)) return true; + if (subagentAwaitsUser(thread)) return true; const project = projectById.get(thread.projectId); if (!project) { return false; @@ -758,7 +758,11 @@ export const make = Effect.gen(function* () { if (requestId !== entry.requestId) subagentRequests.set(childId, { ...entry, requestId }); // The newest wait is the one to announce. if (pending === null || DateTime.isGreaterThan(child.updatedAt, pending.updatedAt)) { - pending = { requestId, updatedAt: child.updatedAt }; + pending = { + requestId, + updatedAt: child.updatedAt, + waitsForInput: child.pendingRuntimeRequest?.kind === "user_input", + }; } } return pending; diff --git a/packages/client-runtime/src/state/threadShell.test.ts b/packages/client-runtime/src/state/threadShell.test.ts index 39607575627dd..0b6aa7eb0d0b9 100644 --- a/packages/client-runtime/src/state/threadShell.test.ts +++ b/packages/client-runtime/src/state/threadShell.test.ts @@ -160,6 +160,36 @@ describe("v2 thread shell lists", () => { registry.dispose(); }); + it("flags the parent while a live subagent waits on a secret card", () => { + const { registry, threads, snapshotAtom } = makeHarness(); + const root = v2ThreadShell; + const child = { + ...root, + id: ThreadId.make("child"), + creationSource: "mcp" as const, + status: "running" as const, + activityRunStatus: "running" as const, + lineage: { + ...root.lineage, + parentThreadId: root.id, + relationshipToParent: "subagent" as const, + }, + pendingRuntimeRequest: { + id: RuntimeRequestId.make("secret-card"), + kind: "user_input" as const, + createdAt: v2Now, + blocking: true, + userOnly: true, + }, + }; + registry.set(snapshotAtom(environmentId), { ...v2ShellSnapshot, threads: [root, child] }); + expect( + registry.get(threads.threadShellAtom({ environmentId, threadId: root.id })) + ?.pendingSubagentRequest, + ).toBe("user_input"); + registry.dispose(); + }); + it("rolls a side chat's questions and approvals up to its parent row", () => { const { registry, threads, snapshotAtom } = makeHarness(); const root = v2ThreadShell; diff --git a/packages/client-runtime/src/state/threadShell.ts b/packages/client-runtime/src/state/threadShell.ts index f5ced761c5665..e0e78f92f92a1 100644 --- a/packages/client-runtime/src/state/threadShell.ts +++ b/packages/client-runtime/src/state/threadShell.ts @@ -105,6 +105,7 @@ function pendingSubagentRequestsByThread( delegated: ReadonlyArray<{ readonly parentThreadId: ThreadId; readonly request: PendingSubagentRequest; + readonly userOnly?: boolean; }> = [], ): ReadonlyMap { const requests = new Map(); @@ -113,9 +114,10 @@ function pendingSubagentRequestsByThread( firstParentId: ThreadId | null, request: PendingSubagentRequest, seen: Set, + userOnly = false, ) => { - // The coordinator answers child questions; only approvals need user attention. - if (request === "user_input") return; + // The coordinator answers child questions; approvals and user-only cards need the user. + if (request === "user_input" && !userOnly) return; byId ??= new Map(threads.map((thread) => [thread.id, thread] as const)); let parentId = firstParentId; while (parentId !== null && !seen.has(parentId)) { @@ -139,9 +141,11 @@ function pendingSubagentRequestsByThread( } if (child.lineage.relationshipToParent !== "subagent") continue; const request = subagentRequest(child); - if (request !== null) flag(parentId, request, new Set([child.id])); + if (request !== null) { + flag(parentId, request, new Set([child.id]), child.pendingRuntimeRequest?.userOnly === true); + } } - for (const task of delegated) flag(task.parentThreadId, task.request, new Set()); + for (const task of delegated) flag(task.parentThreadId, task.request, new Set(), task.userOnly); return requests.size === 0 ? EMPTY_SUBAGENT_REQUESTS : requests; } @@ -316,7 +320,8 @@ export function createEnvironmentThreadShellAtoms(input: { const own = subagentRequest(thread); const nested = requestsOf(runsIn).get(thread.id); const request = own === "approval" ? own : (nested ?? own); - return request === null ? [] : [{ parentThreadId, request }]; + const userOnly = request === own && thread.pendingRuntimeRequest?.userOnly === true; + return request === null ? [] : [{ parentThreadId, request, userOnly }]; }); if (delegated.length === 0) continue; const requests = pendingSubagentRequestsByThread( diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index ca72b708ad6e4..ddc42190c4b37 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -2080,6 +2080,8 @@ export const OrchestrationV2PendingRuntimeRequestSummary = Schema.Struct({ createdAt: Schema.DateTimeUtc, /** False for async questions the run keeps working past. Absent from older servers. */ blocking: Schema.optional(Schema.Boolean), + /** Secret and link cards: only the user can answer them, never a coordinating agent. */ + userOnly: Schema.optional(Schema.Boolean), }); export type OrchestrationV2PendingRuntimeRequestSummary = typeof OrchestrationV2PendingRuntimeRequestSummary.Type; diff --git a/packages/shared/src/agentAwareness.ts b/packages/shared/src/agentAwareness.ts index e6b377c5b0ddf..5587613884c1b 100644 --- a/packages/shared/src/agentAwareness.ts +++ b/packages/shared/src/agentAwareness.ts @@ -24,10 +24,15 @@ export interface SubagentRequestAwareness { readonly requestId: RuntimeRequestId; /** The child's own update time, which moves whenever its request changes. */ readonly updatedAt: DateTime.Utc; + /** A card only the user can answer, rather than an approval. */ + readonly waitsForInput?: boolean; } -/** Provider-native subagents ask on their parent; the coordinator answers child questions. */ -export function subagentAwaitsApproval( +/** + * Provider-native subagents ask on their parent. The coordinator answers child questions, + * but not a card only the user can answer, such as a secret. + */ +export function subagentAwaitsUser( thread: Pick< OrchestrationV2ThreadShell, "activityRunStatus" | "creationSource" | "lineage" | "pendingRuntimeRequest" @@ -37,7 +42,7 @@ export function subagentAwaitsApproval( return ( kind !== undefined && kind !== "auth_refresh" && - kind !== "user_input" && + (kind !== "user_input" || thread.pendingRuntimeRequest?.userOnly === true) && thread.lineage.relationshipToParent === "subagent" && !isProviderNativeSubagentThread(thread) && // The activity run stays set when a follow-up is queued behind a blocked run. @@ -82,7 +87,7 @@ export interface ProjectThreadAwarenessV2Input { | "title" | "updatedAt" >; - /** The approval a live subagent below this thread is waiting on. */ + /** What a live subagent below this thread is waiting on the user for. */ readonly pendingSubagentRequest?: SubagentRequestAwareness | null; } @@ -92,7 +97,7 @@ export function projectThreadAwarenessV2( const { environmentId, project, thread } = input; if (!isListedThreadLineage(thread.lineage)) return null; const subagentRequest = input.pendingSubagentRequest ?? null; - const phase = resolveThreadAwarenessPhaseV2(thread, subagentRequest !== null); + const phase = resolveThreadAwarenessPhaseV2(thread, subagentRequest); if (phase === null) { return null; } @@ -133,7 +138,7 @@ export function projectThreadAwarenessV2( function resolveThreadAwarenessPhaseV2( thread: ProjectThreadAwarenessV2Input["thread"], - childAwaitsApproval: boolean, + subagentRequest: SubagentRequestAwareness | null, ): AgentAwarenessPhase | null { if (thread.pendingRuntimeRequest?.kind === "user_input") return "waiting_for_input"; if ( @@ -144,7 +149,9 @@ function resolveThreadAwarenessPhaseV2( } // The parent can finish its own turn while a delegated child keeps working, // so a child's request outranks the parent's run status. - if (childAwaitsApproval) return "waiting_for_approval"; + if (subagentRequest !== null) { + return subagentRequest.waitsForInput === true ? "waiting_for_input" : "waiting_for_approval"; + } switch (thread.activityRunStatus ?? thread.status) { case "preparing": case "starting": From c0f9d7c807a239d616b3c1e5b20afbaf18c01d2f Mon Sep 17 00:00:00 2001 From: Stanislav Zmiev Date: Tue, 10 Nov 2026 02:33:40 +0400 Subject: [PATCH 2/5] fix: keep secret cards user-only through nested task roll-ups A secret card raised by a subagent of a task delegated to a linked environment was dropped on its way to the original parent: the roll-up checked only the intermediate task's own request for the user-only mark. Keep the mark inside the roll-up maps as an internal value, and report it to clients as user_input. Plain child questions and side-chat questions roll up as before. --- .../src/state/threadShell.test.ts | 15 +++++ .../client-runtime/src/state/threadShell.ts | 58 ++++++++++--------- 2 files changed, 47 insertions(+), 26 deletions(-) diff --git a/packages/client-runtime/src/state/threadShell.test.ts b/packages/client-runtime/src/state/threadShell.test.ts index 0b6aa7eb0d0b9..d18a3e3adae1c 100644 --- a/packages/client-runtime/src/state/threadShell.test.ts +++ b/packages/client-runtime/src/state/threadShell.test.ts @@ -410,6 +410,21 @@ describe("v2 thread shell lists", () => { expect(registry.get(threads.threadShellAtom(parentRef))?.pendingSubagentRequest).toBeNull(); expect(registry.get(inputs).map((thread) => thread.id)).toEqual([helper.id]); + // A secret card needs the user, so it still reaches the parent here. + registry.set(snapshotAtom(remoteEnvironmentId), { + ...v2ShellSnapshot, + threads: [ + task, + { + ...helper, + pendingRuntimeRequest: { ...approval, kind: "user_input" as const, userOnly: true }, + }, + ], + }); + expect(registry.get(threads.threadShellAtom(parentRef))?.pendingSubagentRequest).toBe( + "user_input", + ); + // A task whose parent is gone there is listed on its own, as is one whose // parent's environment is switched off. registry.set(snapshotAtom(environmentId), { ...v2ShellSnapshot, threads: [] }); diff --git a/packages/client-runtime/src/state/threadShell.ts b/packages/client-runtime/src/state/threadShell.ts index e0e78f92f92a1..3aaaab30720ad 100644 --- a/packages/client-runtime/src/state/threadShell.ts +++ b/packages/client-runtime/src/state/threadShell.ts @@ -30,7 +30,7 @@ import { const EMPTY_THREADS: ReadonlyArray = Object.freeze([]); const EMPTY_SCOPED_THREAD_REFS: ReadonlyArray = Object.freeze([]); const EMPTY_THREAD_INDEX: ReadonlyMap = new Map(); -const EMPTY_SUBAGENT_REQUESTS: ReadonlyMap = new Map(); +const EMPTY_SUBAGENT_REQUESTS: ReadonlyMap = new Map(); const EMPTY_SIDE_CHATS_BY_PARENT: ReadonlyMap< ThreadId, ReadonlyArray @@ -85,6 +85,17 @@ function subagentRequest(thread: OrchestrationV2ThreadShell): PendingSubagentReq return kind === "user_input" ? "user_input" : "approval"; } +/** A user-only card keeps its mark through roll-ups; clients see it as user_input. */ +type RolledUpRequest = PendingSubagentRequest | "user_only_input"; + +const rolledUpRequest = ( + thread: OrchestrationV2ThreadShell, + request: PendingSubagentRequest | null, +): RolledUpRequest | null => + request === "user_input" && thread.pendingRuntimeRequest?.userOnly === true + ? "user_only_input" + : request; + /** A side chat asks its user directly, so questions roll up to the parent as well as approvals. */ function sideChatRequest(thread: OrchestrationV2ThreadShell): PendingSubagentRequest | null { const kind = thread.pendingRuntimeRequest?.kind; @@ -104,20 +115,14 @@ function pendingSubagentRequestsByThread( threads: ReadonlyArray, delegated: ReadonlyArray<{ readonly parentThreadId: ThreadId; - readonly request: PendingSubagentRequest; - readonly userOnly?: boolean; + readonly request: RolledUpRequest; }> = [], -): ReadonlyMap { - const requests = new Map(); +): ReadonlyMap { + const requests = new Map(); let byId: ReadonlyMap | undefined; - const flag = ( - firstParentId: ThreadId | null, - request: PendingSubagentRequest, - seen: Set, - userOnly = false, - ) => { + const flag = (firstParentId: ThreadId | null, request: RolledUpRequest, seen: Set) => { // The coordinator answers child questions; approvals and user-only cards need the user. - if (request === "user_input" && !userOnly) return; + if (request === "user_input") return; byId ??= new Map(threads.map((thread) => [thread.id, thread] as const)); let parentId = firstParentId; while (parentId !== null && !seen.has(parentId)) { @@ -140,18 +145,16 @@ function pendingSubagentRequestsByThread( continue; } if (child.lineage.relationshipToParent !== "subagent") continue; - const request = subagentRequest(child); - if (request !== null) { - flag(parentId, request, new Set([child.id]), child.pendingRuntimeRequest?.userOnly === true); - } + const request = rolledUpRequest(child, subagentRequest(child)); + if (request !== null) flag(parentId, request, new Set([child.id])); } - for (const task of delegated) flag(task.parentThreadId, task.request, new Set(), task.userOnly); + for (const task of delegated) flag(task.parentThreadId, task.request, new Set()); return requests.size === 0 ? EMPTY_SUBAGENT_REQUESTS : requests; } const sameRequests = ( - left: ReadonlyMap, - right: ReadonlyMap, + left: ReadonlyMap, + right: ReadonlyMap, ) => left.size === right.size && [...right].every(([threadId, request]) => left.get(threadId) === request); @@ -278,7 +281,7 @@ export function createEnvironmentThreadShellAtoms(input: { // environments run for it, so two environments delegating to each other // never read each other's totals. const localSubagentRequestsAtom = Atom.family((environmentId: EnvironmentId) => { - let previous = EMPTY_SUBAGENT_REQUESTS; + let previous: ReadonlyMap = EMPTY_SUBAGENT_REQUESTS; return Atom.make((get) => { const next = pendingSubagentRequestsByThread(get(environmentThreadsAtom(environmentId))); if (sameRequests(previous, next)) return previous; @@ -309,7 +312,7 @@ export function createEnvironmentThreadShellAtoms(input: { else tasks.push(task); } } - let flags = new Map>(); + let flags = new Map>(); const requestsOf = (environmentId: EnvironmentId) => flags.get(environmentId) ?? get(localSubagentRequestsAtom(environmentId)); for (let changed = tasksByDelegator.size > 0; changed;) { @@ -317,11 +320,10 @@ export function createEnvironmentThreadShellAtoms(input: { const next = new Map(flags); for (const [environmentId, tasks] of tasksByDelegator) { const delegated = tasks.flatMap(({ runsIn, thread, parentThreadId }) => { - const own = subagentRequest(thread); + const own = rolledUpRequest(thread, subagentRequest(thread)); const nested = requestsOf(runsIn).get(thread.id); const request = own === "approval" ? own : (nested ?? own); - const userOnly = request === own && thread.pendingRuntimeRequest?.userOnly === true; - return request === null ? [] : [{ parentThreadId, request, userOnly }]; + return request === null ? [] : [{ parentThreadId, request }]; }); if (delegated.length === 0) continue; const requests = pendingSubagentRequestsByThread( @@ -338,11 +340,15 @@ export function createEnvironmentThreadShellAtoms(input: { }).pipe(Atom.withLabel("environment-delegated-subagent-requests")); const pendingSubagentRequestsAtom = Atom.family((environmentId: EnvironmentId) => { - let previous = EMPTY_SUBAGENT_REQUESTS; + let previous: ReadonlyMap = EMPTY_SUBAGENT_REQUESTS; return Atom.make((get) => { - const next = + const rolledUp = get(delegatedSubagentRequestsAtom).get(environmentId) ?? get(localSubagentRequestsAtom(environmentId)); + const next = new Map(); + for (const [threadId, request] of rolledUp) { + next.set(threadId, request === "user_only_input" ? "user_input" : request); + } if (sameRequests(previous, next)) return previous; previous = next; return next; From f201a2ee52de859df099bebcd854e4b8f367c9bb Mon Sep 17 00:00:00 2001 From: Stanislav Zmiev Date: Tue, 10 Nov 2026 02:45:45 +0400 Subject: [PATCH 3/5] fix(relay): wake the relay when a subagent raises a secret card Recording a secret or link card emits only a turn-item update, which the relay ignored, so it never re-read the waiting child and never published its parent as waiting for input. Let those card updates through the relay's event filter. The relay test now drives the real event instead of calling publishThread for the child. --- .../src/relay/AgentAwarenessRelay.test.ts | 46 +++++++++++++++++-- apps/server/src/relay/AgentAwarenessRelay.ts | 8 +++- 2 files changed, 49 insertions(+), 5 deletions(-) diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index 8ef5df3d1b9c6..7a51c4dc4801c 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -375,26 +375,64 @@ describe("AgentAwarenessRelay", () => { }), ); - it.effect("publishes the parent as needing input while its subagent waits on a secret card", () => + it.effect("publishes the parent as needing input when its subagent raises a secret card", () => Effect.gen(function* () { + const now = DateTime.makeUnsafe(NOW); const child = waitingChild({ pendingRuntimeRequest: { id: RuntimeRequestId.make("child-secret-card"), kind: "user_input", - createdAt: DateTime.makeUnsafe(NOW), + createdAt: now, blocking: true, userOnly: true, }, }); const parent = shell({ status: "completed" }); + const events = yield* Queue.unbounded(); + const pulls = yield* Queue.unbounded(); const { relay, publications } = yield* makeTestRelay({ readShell: (threadId) => Effect.succeed(threadId === child.id ? child : parent), + domainEvents: Stream.fromEffectRepeat( + Queue.offer(pulls, undefined).pipe(Effect.andThen(Queue.take(events))), + ), }); - yield* relay.publishThread(CHILD_THREAD_ID); + yield* relay.start(); + yield* Queue.take(pulls); + const runId = RunId.make("run:child-secret-card"); + // Recording a secret card emits only this event for the child. + yield* Queue.offer(events, { + id: EventId.make("event:child-secret-card"), + type: "turn-item.updated", + threadId: CHILD_THREAD_ID, + runId, + occurredAt: now, + payload: { + id: TurnItemId.make("child-secret-card"), + threadId: CHILD_THREAD_ID, + runId, + nodeId: NodeId.make("node:child-secret-card"), + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "waiting", + title: "SMS code", + startedAt: now, + completedAt: null, + updatedAt: now, + type: "secret_request", + label: "SMS code", + reason: "The agent needs the code.", + secretStatus: "pending", + }, + }); + yield* Queue.take(pulls); + yield* relay.drain; assert.equal(publications.length, 1); assert.isTrue(publications[0]?.url.includes(`/threads/${THREAD_ID}/`) ?? false); assert.equal(publications[0]?.state?.phase, "waiting_for_input"); - }), + }).pipe(Effect.scoped), ); it.effect("republishes a completed parent for its waiting child after a relink", () => diff --git a/apps/server/src/relay/AgentAwarenessRelay.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index 3ab86280ca33c..5dacb301e21cf 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -128,7 +128,13 @@ function shouldPublishAgentAwarenessEvent( case "provider-session.detached": case "provider-turn.updated": case "turn-item.updated": - return isTurnItemPayload(event.payload) && turnItemUpdateCanEndBackgroundWork(event.payload); + return ( + isTurnItemPayload(event.payload) && + (turnItemUpdateCanEndBackgroundWork(event.payload) || + // A subagent's secret or link card waits on the user through its parent. + event.payload.type === "secret_request" || + event.payload.type === "link_request") + ); case "message.updated": case "plan.updated": case "checkpoint-scope.created": From 4ffe6bccad53cdc7a16e9884fee1d1a40eb0f89d Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 10 Nov 2026 13:47:35 +0000 Subject: [PATCH 4/5] fix(server): publish delegated secret cards to background clients --- .../src/orchestration-v2/ShellStream.test.ts | 20 ++++++++++++++++--- .../src/orchestration-v2/ShellStream.ts | 6 ++++++ 2 files changed, 23 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/ShellStream.test.ts b/apps/server/src/orchestration-v2/ShellStream.test.ts index 336f9b12f96bf..7d1a1cfaba0a1 100644 --- a/apps/server/src/orchestration-v2/ShellStream.test.ts +++ b/apps/server/src/orchestration-v2/ShellStream.test.ts @@ -38,8 +38,8 @@ const emptyShellSnapshot = { } as OrchestrationV2ShellSnapshot; describe("subagent shell events", () => { - const stored = (threadId: string, type: string) => - ({ event: { threadId, type } }) as ApplicationStoredEvent; + const stored = (threadId: string, type: string, itemType = "command_execution") => + ({ event: { threadId, type, payload: { type: itemType } } }) as ApplicationStoredEvent; it("keeps regular threads and subagent state changes while omitting subagent transcript updates", () => { const subagents = new Set(["child"]); @@ -49,6 +49,14 @@ describe("subagent shell events", () => { expect( shouldPublishShellApplicationEvent(stored("child", "turn-item.updated"), subagents), ).toBe(false); + for (const itemType of ["secret_request", "link_request"]) { + expect( + shouldPublishShellApplicationEvent( + stored("child", "turn-item.updated", itemType), + subagents, + ), + ).toBe(true); + } expect(shouldPublishShellApplicationEvent(stored("child", "node.updated"), subagents)).toBe( false, ); @@ -198,7 +206,13 @@ describe("omitIdleSubagentShells", () => { expect(omitted.threads.map((thread) => thread.id)).toEqual(["root"]); expect( shouldPublishShellApplicationEvent( - { event: { threadId: "idle", type: "turn-item.updated" } } as ApplicationStoredEvent, + { + event: { + threadId: "idle", + type: "turn-item.updated", + payload: { type: "command_execution" }, + }, + } as ApplicationStoredEvent, omitted.subagentThreadIds, ), ).toBe(false); diff --git a/apps/server/src/orchestration-v2/ShellStream.ts b/apps/server/src/orchestration-v2/ShellStream.ts index 0fa5251a7ef73..41c7bb21746ee 100644 --- a/apps/server/src/orchestration-v2/ShellStream.ts +++ b/apps/server/src/orchestration-v2/ShellStream.ts @@ -428,6 +428,12 @@ export function shouldPublishShellApplicationEvent( // Replay and the live tail overlap; keep known IDs so late events stay filtered. return true; } + if ( + event.type === "turn-item.updated" && + (event.payload.type === "secret_request" || event.payload.type === "link_request") + ) { + return true; + } return !( subagentThreadIds.has(event.threadId) && (event.type === "node.updated" || From 7e3922a1a8cb257cab9d7bb082b4fa915e9432a7 Mon Sep 17 00:00:00 2001 From: Stanislav Zmiev Date: Fri, 13 Nov 2026 00:57:10 +0400 Subject: [PATCH 5/5] fix: clear a subagent's link-card wait when the user steers past it The shell stream and relay dropped a subagent's user_message turn item, so after the user steered a running child past its link card the parent kept showing Input needed until a later event. --- .../src/orchestration-v2/ShellStream.test.ts | 2 +- .../src/orchestration-v2/ShellStream.ts | 6 +- .../src/relay/AgentAwarenessRelay.test.ts | 94 +++++++++++++++++++ apps/server/src/relay/AgentAwarenessRelay.ts | 5 +- packages/shared/src/agentAwareness.ts | 8 ++ 5 files changed, 107 insertions(+), 8 deletions(-) diff --git a/apps/server/src/orchestration-v2/ShellStream.test.ts b/apps/server/src/orchestration-v2/ShellStream.test.ts index 7d1a1cfaba0a1..d97635d9cf8ad 100644 --- a/apps/server/src/orchestration-v2/ShellStream.test.ts +++ b/apps/server/src/orchestration-v2/ShellStream.test.ts @@ -49,7 +49,7 @@ describe("subagent shell events", () => { expect( shouldPublishShellApplicationEvent(stored("child", "turn-item.updated"), subagents), ).toBe(false); - for (const itemType of ["secret_request", "link_request"]) { + for (const itemType of ["secret_request", "link_request", "user_message"]) { expect( shouldPublishShellApplicationEvent( stored("child", "turn-item.updated", itemType), diff --git a/apps/server/src/orchestration-v2/ShellStream.ts b/apps/server/src/orchestration-v2/ShellStream.ts index b1c1bd665b403..5ee1005b16830 100644 --- a/apps/server/src/orchestration-v2/ShellStream.ts +++ b/apps/server/src/orchestration-v2/ShellStream.ts @@ -18,6 +18,7 @@ import { OrchestrationProjectShell as ProjectShellSchema, OrchestrationV2ThreadShell as ThreadShellSchema, } from "@t3tools/contracts"; +import { turnItemCanChangeUserWait } from "@t3tools/shared/agentAwareness"; import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; // @effect-diagnostics-next-line nodeBuiltinImport:off - a synchronous digest of a few project rows. import * as NodeCrypto from "node:crypto"; @@ -547,10 +548,7 @@ export function shouldPublishShellApplicationEvent( // Replay and the live tail overlap; keep known IDs so late events stay filtered. return true; } - if ( - event.type === "turn-item.updated" && - (event.payload.type === "secret_request" || event.payload.type === "link_request") - ) { + if (event.type === "turn-item.updated" && turnItemCanChangeUserWait(event.payload)) { return true; } return !( diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index 969e4ebc8a11a..dd8d608deeb83 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -437,6 +437,100 @@ describe("AgentAwarenessRelay", () => { }).pipe(Effect.scoped), ); + it.effect( + "clears the parent's input wait when the user steers its subagent past a link card", + () => + Effect.gen(function* () { + const now = DateTime.makeUnsafe(NOW); + let child = waitingChild({ + pendingRuntimeRequest: { + id: RuntimeRequestId.make("child-link-card"), + kind: "user_input", + createdAt: now, + blocking: true, + userOnly: true, + }, + }); + const parent = shell({ status: "completed" }); + const events = yield* Queue.unbounded(); + const pulls = yield* Queue.unbounded(); + const { relay, publications } = yield* makeTestRelay({ + readShell: (threadId) => Effect.succeed(threadId === child.id ? child : parent), + domainEvents: Stream.fromEffectRepeat( + Queue.offer(pulls, undefined).pipe(Effect.andThen(Queue.take(events))), + ), + }); + yield* relay.start(); + yield* Queue.take(pulls); + const runId = RunId.make("run:child-link-card"); + const item = { + threadId: CHILD_THREAD_ID, + runId, + nodeId: NodeId.make("node:child-link-card"), + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + title: null, + startedAt: now, + updatedAt: now, + } as const; + yield* Queue.offer(events, { + id: EventId.make("event:child-link-card"), + type: "turn-item.updated", + threadId: CHILD_THREAD_ID, + runId, + occurredAt: now, + payload: { + ...item, + id: TurnItemId.make("child-link-card"), + ordinal: 1, + status: "waiting", + completedAt: null, + type: "link_request", + url: "https://example.com/login", + label: "Sign in", + reason: "The agent needs you to sign in.", + linkStatus: "pending", + }, + }); + yield* Queue.take(pulls); + yield* relay.drain; + assert.equal(publications[0]?.state?.phase, "waiting_for_input"); + + // Steering a running child records only these events for it; the + // projection then stops counting the link card as pending. + child = waitingChild({ pendingRuntimeRequest: null }); + yield* Queue.offer(events, { + id: EventId.make("event:child-steer"), + type: "turn-item.updated", + threadId: CHILD_THREAD_ID, + runId, + occurredAt: now, + payload: { + ...item, + id: TurnItemId.make("child-steer"), + ordinal: 2, + status: "completed", + completedAt: now, + createdBy: "user", + creationSource: "web", + type: "user_message", + messageId: MessageId.make("message:child-steer"), + inputIntent: "steer", + text: "Skip the login.", + attachments: [], + }, + }); + yield* Queue.take(pulls); + // The relay confirms a move into completed before publishing it. + yield* TestClock.adjust("5 seconds"); + yield* relay.drain; + assert.equal(publications.length, 2); + assert.equal(publications[1]?.state?.phase, "completed"); + }).pipe(Effect.scoped), + ); + it.effect("republishes a completed parent for its waiting child after a relink", () => Effect.gen(function* () { const parent = shell({ status: "completed" }); diff --git a/apps/server/src/relay/AgentAwarenessRelay.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index 5dacb301e21cf..73b52571e7416 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -17,6 +17,7 @@ import { type SubagentRequestAwareness, projectThreadAwarenessV2, subagentAwaitsUser, + turnItemCanChangeUserWait, } from "@t3tools/shared/agentAwareness"; import { turnItemUpdateCanEndBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; @@ -131,9 +132,7 @@ function shouldPublishAgentAwarenessEvent( return ( isTurnItemPayload(event.payload) && (turnItemUpdateCanEndBackgroundWork(event.payload) || - // A subagent's secret or link card waits on the user through its parent. - event.payload.type === "secret_request" || - event.payload.type === "link_request") + turnItemCanChangeUserWait(event.payload)) ); case "message.updated": case "plan.updated": diff --git a/packages/shared/src/agentAwareness.ts b/packages/shared/src/agentAwareness.ts index 1a30dd42b4d63..4e68951308c51 100644 --- a/packages/shared/src/agentAwareness.ts +++ b/packages/shared/src/agentAwareness.ts @@ -31,6 +31,14 @@ export interface SubagentRequestAwareness { readonly waitsForInput?: boolean; } +/** + * Turn items that can start or end a subagent's user-only wait: its secret or link card, + * or a user message that moves past a link card. Filters that drop subagent transcript + * updates must keep these, or the parent's "Input needed" never appears or never clears. + */ +export const turnItemCanChangeUserWait = (item: { readonly type: string }) => + item.type === "secret_request" || item.type === "link_request" || item.type === "user_message"; + /** * Provider-native subagents ask on their parent. The coordinator answers child questions, * but not a card only the user can answer, such as a secret.