diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index ba87697d60f03..3d33f76b7a494 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1012,6 +1012,7 @@ export type AgentAwarenessThread = ProjectThreadAwarenessV2Input["thread"] & Pick< OrchestrationV2ThreadShell, | "projectId" + | "creationSource" | "latestRunCompletedAt" | "archivedAt" | "activeRunId" @@ -7173,6 +7174,7 @@ export const layer: Layer.Layer = Effect.map((state): AgentAwarenessThread => ({ id: state.thread.id, projectId: state.thread.projectId, + creationSource: state.thread.creationSource, archivedAt: state.thread.archivedAt, activeRunId: state.activeRunId, latestRunId: state.latestRunId, diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index ae17bd7583bc3..ccd28c165f14e 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -11,6 +11,7 @@ import { ProjectId, ProviderInstanceId, RunId, + RuntimeRequestId, ThreadId, TurnItemId, } from "@t3tools/contracts"; @@ -93,6 +94,31 @@ function shell(overrides: Partial = {}): Orchestrati }; } +const CHILD_THREAD_ID = ThreadId.make("relay-child-thread"); + +function waitingChild( + overrides: Partial = {}, +): 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"); @@ -305,6 +331,112 @@ 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([ + [THREAD_ID, shell({ status: "running" })], + [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* 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"); + + // A rename keeps the same request's alert time, so the phone doesn't re-alert. + yield* setShell( + shell({ + status: "completed", + title: "Renamed thread", + updatedAt: DateTime.makeUnsafe("2026-09-04T12:01:00.000Z"), + }), + ); + yield* relay.publishThread(THREAD_ID); + assert.equal(publications.length, 3); + assert.equal(publications[2]?.state?.threadTitle, "Renamed thread"); + assert.equal(publications[2]?.state?.updatedAt, publications[1]?.state?.updatedAt); + + yield* setShell(waitingChild({ pendingRuntimeRequest: null })); + yield* relay.publishThread(CHILD_THREAD_ID); + yield* TestClock.adjust("5 seconds"); + yield* relay.drain; + assert.equal(publications.length, 4); + assert.equal(publications[3]?.state?.phase, "completed"); + }), + ); + + it.effect("republishes a completed parent for its waiting child after a relink", () => + Effect.gen(function* () { + const parent = shell({ status: "completed" }); + const child = waitingChild(); + const { relay, secrets, publications } = yield* makeTestRelay({ + catchUpThreads: [parent, child], + readShell: (id) => Effect.succeed(id === child.id ? child : parent), + }); + yield* relay.start(); + yield* TestClock.adjust("1 second"); + yield* relay.drain; + assert.equal(publications.length, 1); + assert.equal(publications[0]?.state?.phase, "waiting_for_approval"); + + yield* secrets.set( + RELAY_ENVIRONMENT_CREDENTIAL_SECRET, + new TextEncoder().encode("credential-2"), + ); + yield* relay.requestCatchUp(); + yield* TestClock.adjust("1 second"); + yield* relay.drain; + assert.equal(publications.length, 2); + assert.equal(publications[1]?.state?.phase, "waiting_for_approval"); + }).pipe(Effect.scoped), + ); + + it.effect("keeps a child's alert time when the parent's own request comes and goes", () => + Effect.gen(function* () { + const shells = yield* Ref.make( + new Map([ + [THREAD_ID, shell()], + [CHILD_THREAD_ID, waitingChild()], + ]), + ); + const setParent = (next: OrchestrationV2ThreadShell) => + Ref.update(shells, (current) => new Map(current).set(THREAD_ID, next)); + const { relay, publications } = yield* makeTestRelay({ + readShell: (threadId) => + Ref.get(shells).pipe(Effect.map((current) => current.get(threadId) ?? null)), + }); + yield* relay.publishThread(CHILD_THREAD_ID); + yield* setParent( + shell({ + updatedAt: DateTime.makeUnsafe("2026-09-04T12:01:00.000Z"), + pendingRuntimeRequest: { + id: RuntimeRequestId.make("parent-request"), + kind: "command", + createdAt: DateTime.makeUnsafe("2026-09-04T12:01:00.000Z"), + }, + }), + ); + yield* relay.publishThread(THREAD_ID); + yield* setParent(shell({ updatedAt: DateTime.makeUnsafe("2026-09-04T12:02:00.000Z") })); + yield* relay.publishThread(THREAD_ID); + assert.deepStrictEqual( + publications.map(({ state }) => state?.phase), + ["waiting_for_approval", "waiting_for_approval", "waiting_for_approval"], + ); + assert.equal(publications[2]?.state?.updatedAt, publications[0]?.state?.updatedAt); + }), + ); + it.effect("coalesces queued updates and reruns a thread dirtied during publishing", () => Effect.gen(function* () { const started = yield* Deferred.make(); diff --git a/apps/server/src/relay/AgentAwarenessRelay.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index 2523fea9f2cf0..c14558762183a 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -5,6 +5,7 @@ import { type OrchestrationV2DomainEvent, type OrchestrationV2TurnItem, type Project, + type RuntimeRequestId, type ThreadId, } from "@t3tools/contracts"; import { @@ -12,7 +13,11 @@ import { type RelayAgentActivityPublishProofPayload, type RelayAgentActivityState, } from "@t3tools/contracts/relay"; -import { projectThreadAwarenessV2 } from "@t3tools/shared/agentAwareness"; +import { + type SubagentRequestAwareness, + projectThreadAwarenessV2, + subagentAwaitsApproval, +} from "@t3tools/shared/agentAwareness"; import { turnItemUpdateCanEndBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; import { withRelayClientTracing } from "@t3tools/shared/relayTracing"; @@ -153,12 +158,19 @@ export const makeAgentAwarenessPublishWorker = Effect.fnUntraced(function* ( return { enqueue, drain: worker.drain }; }); -function agentAwarenessPublishIdentity(state: RelayAgentActivityState | null): string { +function agentAwarenessPublishIdentity( + state: RelayAgentActivityState | null, + subagentRequest: SubagentRequestAwareness | null = null, +): string { if (state === null) { return "null"; } const { updatedAt: _updatedAt, ...meaningfulState } = state; - return JSON.stringify(meaningfulState); + // A child's next request reads the same as its last one, but it is a new + // wait and must reach the phone. + return JSON.stringify( + subagentRequest === null ? meaningfulState : [meaningfulState, subagentRequest.requestId], + ); } const CLEARED_PUBLISH_IDENTITY = agentAwarenessPublishIdentity(null); @@ -285,19 +297,38 @@ function describeThreadShellForAwareness( }; } +interface SubagentRequest { + readonly topLevelThreadId: ThreadId; + readonly requestId: RuntimeRequestId; +} + +interface SubagentAlert { + readonly requestId: RuntimeRequestId; + readonly updatedAt: string; +} + +const waitingApprovalId = (thread: AgentAwarenessThread | null) => + thread !== null && thread.archivedAt === null && subagentAwaitsApproval(thread) + ? thread.pendingRuntimeRequest?.id + : undefined; + function resolveAgentAwarenessRelayPublishSnapshot(input: { readonly environmentId: EnvironmentId; readonly thread: Option.Option; readonly project: Option.Option; + readonly pendingSubagentRequest: SubagentRequestAwareness | null; }): { readonly projectId: string | null; readonly state: RelayAgentActivityState | null; + /** The child request behind the state's wait, when the parent isn't waiting itself. */ + readonly subagentWait: SubagentRequestAwareness | null; readonly reason: "snapshot" | "thread-not-found" | "project-not-found"; } { if (Option.isNone(input.thread)) { return { projectId: null, state: null, + subagentWait: null, reason: "thread-not-found", }; } @@ -305,18 +336,26 @@ function resolveAgentAwarenessRelayPublishSnapshot(input: { return { projectId: input.thread.value.projectId, state: null, + subagentWait: null, reason: "project-not-found", }; } + const awarenessInput = { + environmentId: input.environmentId, + project: input.project.value, + thread: input.thread.value, + }; + const state = projectThreadAwarenessV2({ + ...awarenessInput, + pendingSubagentRequest: input.pendingSubagentRequest, + }); + const childDrivesWait = + input.pendingSubagentRequest !== null && + state?.phase !== projectThreadAwarenessV2(awarenessInput)?.phase; return { projectId: input.thread.value.projectId, - state: sanitizeRelayAgentActivityState( - projectThreadAwarenessV2({ - environmentId: input.environmentId, - project: input.project.value, - thread: input.thread.value, - }), - ), + state: sanitizeRelayAgentActivityState(state), + subagentWait: childDrivesWait ? input.pendingSubagentRequest : null, reason: "snapshot", }; } @@ -337,6 +376,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 (subagentAwaitsApproval(thread)) return true; const project = projectById.get(thread.projectId); if (!project) { return false; @@ -368,6 +409,10 @@ export const make = Effect.gen(function* () { const startedAt = (yield* DateTime.now).epochMilliseconds; const catchUpRequests = yield* Queue.dropping(1); const publishedStateByThreadRef = yield* Ref.make(new Map()); + // Android keys alerts by phase and time, so a child's unchanged request keeps + // its first alert time through parent renames, relinks and the parent's own requests. + const childAlertByThread = new Map(); + const subagentRequests = new Map(); const readSecretString = (name: string) => secrets @@ -423,6 +468,7 @@ export const make = Effect.gen(function* () { Effect.forEach([...publishRetries.keys()], cancelPublishRetry, { discard: true }), ); let schedulePublishRetry: (threadId: ThreadId) => Effect.Effect = () => Effect.void; + let enqueueTopLevelPublish: (threadId: ThreadId) => Effect.Effect = () => Effect.void; const resetPublishedConnection = Effect.gen(function* () { publishedRelayConfig = undefined; cachedRelayClient = undefined; @@ -508,7 +554,8 @@ export const make = Effect.gen(function* () { }); // Read only this thread; full shell reads would make each activity event scale with all threads. - const threadShell = (yield* threads.getAgentAwarenessSnapshot(threadId)).threads[0] ?? null; + const threadShell = yield* readAwarenessThread(threadId); + yield* trackSubagentRequest(threadId, threadShell); if ( threadShell !== null && !isListedThreadLineage(threadShell.lineage) && @@ -528,8 +575,9 @@ export const make = Effect.gen(function* () { environmentId, thread, project, + pendingSubagentRequest: yield* currentSubagentRequestFor(threadId), }); - const publishIdentity = agentAwarenessPublishIdentity(snapshot.state); + const publishIdentity = agentAwarenessPublishIdentity(snapshot.state, snapshot.subagentWait); const publishedStateByThread = yield* Ref.get(publishedStateByThreadRef); if ( (snapshot.state?.phase === "completed" || snapshot.state?.phase === "failed") && @@ -594,6 +642,7 @@ export const make = Effect.gen(function* () { environmentId, thread: Option.some({ ...thread.value, settledOverride: null }), project, + pendingSubagentRequest: null, }) : null; if (unsettled?.state?.phase === "completed") { @@ -629,11 +678,22 @@ export const make = Effect.gen(function* () { }); } + const childRequestId = snapshot.subagentWait?.requestId; + const childAlert = childAlertByThread.get(threadId); + const state = + snapshot.state !== null && + childRequestId !== undefined && + childAlert?.requestId === childRequestId + ? { ...snapshot.state, updatedAt: childAlert.updatedAt } + : snapshot.state; yield* publishState({ projectId: snapshot.projectId, - state: snapshot.state, + state, reason: snapshot.reason, }).pipe(withRelayClientTracing); + if (state !== null && childRequestId !== undefined) { + childAlertByThread.set(threadId, { requestId: childRequestId, updatedAt: state.updatedAt }); + } publishConfirmDeadlines.delete(threadId); heldCompletions.delete(threadId); yield* Ref.update(publishedStateByThreadRef, (publishedStates) => @@ -641,6 +701,69 @@ export const make = Effect.gen(function* () { ); }); + const readAwarenessThread = (threadId: ThreadId) => + threads + .getAgentAwarenessSnapshot(threadId) + .pipe(Effect.map((snapshot) => snapshot.threads[0] ?? null)); + + // The subagent helpers below are untraced: they run per event inside + // publishThreadUnsafe, whose span already covers them. + const findTopLevelThreadId = Effect.fnUntraced(function* (child: AgentAwarenessThread) { + const seen = new Set([child.id]); + let current: AgentAwarenessThread | null = child; + while ( + current?.lineage.relationshipToParent === "subagent" || + current?.lineage.sideChat === true + ) { + const parentId: ThreadId | null = current.lineage.parentThreadId; + if (parentId === null || seen.has(parentId)) return null; + seen.add(parentId); + current = yield* readAwarenessThread(parentId); + } + return current?.id ?? null; + }); + + const trackSubagentRequest = Effect.fnUntraced(function* ( + threadId: ThreadId, + threadShell: AgentAwarenessThread | null, + ) { + const requestId = waitingApprovalId(threadShell); + const previous = subagentRequests.get(threadId); + // An unchanged wait still republishes its parent: a relink catch-up reaches it only here. + if (threadShell === null || requestId === undefined) { + 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, requestId }); + yield* enqueueTopLevelPublish(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: SubagentRequestAwareness | null = null; + for (const [childId, entry] of subagentRequests) { + if (entry.topLevelThreadId !== topLevelThreadId) continue; + const child = yield* readAwarenessThread(childId); + const requestId = waitingApprovalId(child); + if (child === null || requestId === undefined) { + subagentRequests.delete(childId); + continue; + } + 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 }; + } + } + return pending; + }); + const processThreadPublish = Effect.fnUntraced(function* (threadId: ThreadId) { const retry = publishRetries.get(threadId); if (retry?.timer !== undefined) { @@ -664,6 +787,7 @@ export const make = Effect.gen(function* () { const worker = yield* makeAgentAwarenessPublishWorker(processThreadPublish); const enqueueThreadPublish = (threadId: ThreadId) => cancelPublishRetry(threadId).pipe(Effect.andThen(worker.enqueue(threadId))); + enqueueTopLevelPublish = enqueueThreadPublish; const publishThread: AgentAwarenessRelay["Service"]["publishThread"] = (threadId) => enqueueThreadPublish(threadId).pipe(Effect.andThen(worker.drain)); @@ -720,11 +844,18 @@ export const make = Effect.gen(function* () { const environmentId = yield* serverEnvironment.getEnvironmentId; const projectSnapshot = yield* projects.snapshot; const activitySnapshot = yield* threads.getAgentAwarenessSnapshot(); + // Live children aren't listed, but a waiting one makes its top-level thread wait. + const liveChildren = yield* Effect.forEach( + (yield* threads.listActiveShellIndex()).filter( + (thread) => thread.lineage.relationshipToParent === "subagent" && thread.mayBeLive, + ), + (thread) => readAwarenessThread(thread.id), + ); const activeThreadIds = resolveAgentAwarenessRelayActiveThreadIds({ environmentId, startedAt, projects: projectSnapshot.projects, - threads: activitySnapshot.threads, + threads: [...activitySnapshot.threads, ...liveChildren.filter((thread) => thread !== null)], }); if (activeThreadIds.length === 0) { yield* Effect.logDebug("agent activity snapshot has no publishable threads"); diff --git a/packages/shared/src/agentAwareness.ts b/packages/shared/src/agentAwareness.ts index c7bf09fe0dd0b..e6b377c5b0ddf 100644 --- a/packages/shared/src/agentAwareness.ts +++ b/packages/shared/src/agentAwareness.ts @@ -1,8 +1,10 @@ import { isListedThreadLineage, type EnvironmentId, + isProviderNativeSubagentThread, type OrchestrationV2ThreadShell, type Project, + type RuntimeRequestId, type ThreadId, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; @@ -18,6 +20,31 @@ export type AgentAwarenessPhase = | "failed" | "stale"; +export interface SubagentRequestAwareness { + readonly requestId: RuntimeRequestId; + /** The child's own update time, which moves whenever its request changes. */ + readonly updatedAt: DateTime.Utc; +} + +/** Provider-native subagents ask on their parent; the coordinator answers child questions. */ +export function subagentAwaitsApproval( + thread: Pick< + OrchestrationV2ThreadShell, + "activityRunStatus" | "creationSource" | "lineage" | "pendingRuntimeRequest" + >, +): boolean { + const kind = thread.pendingRuntimeRequest?.kind; + return ( + kind !== undefined && + kind !== "auth_refresh" && + kind !== "user_input" && + thread.lineage.relationshipToParent === "subagent" && + !isProviderNativeSubagentThread(thread) && + // The activity run stays set when a follow-up is queued behind a blocked run. + thread.activityRunStatus != null + ); +} + export interface AgentAwarenessState { readonly environmentId: EnvironmentId; readonly threadId: ThreadId; @@ -55,6 +82,8 @@ export interface ProjectThreadAwarenessV2Input { | "title" | "updatedAt" >; + /** The approval a live subagent below this thread is waiting on. */ + readonly pendingSubagentRequest?: SubagentRequestAwareness | null; } export function projectThreadAwarenessV2( @@ -62,7 +91,8 @@ export function projectThreadAwarenessV2( ): AgentAwarenessState | null { const { environmentId, project, thread } = input; if (!isListedThreadLineage(thread.lineage)) return null; - const phase = resolveThreadAwarenessPhaseV2(thread); + const subagentRequest = input.pendingSubagentRequest ?? null; + const phase = resolveThreadAwarenessPhaseV2(thread, subagentRequest !== null); if (phase === null) { return null; } @@ -90,13 +120,20 @@ export function projectThreadAwarenessV2( headline: headlineForPhase(phase), ...(detail === undefined ? {} : { detail }), modelTitle: thread.modelSelection.model, - updatedAt: DateTime.formatIso(thread.updatedAt), + // A child's request leaves the parent's own timestamp alone. Alerts are + // keyed by phase and time, so a second wait needs the child's time. + updatedAt: DateTime.formatIso( + subagentRequest === null + ? thread.updatedAt + : DateTime.max(thread.updatedAt, subagentRequest.updatedAt), + ), deepLink: buildAgentAwarenessDeepLink({ environmentId, threadId: thread.id }), }; } function resolveThreadAwarenessPhaseV2( thread: ProjectThreadAwarenessV2Input["thread"], + childAwaitsApproval: boolean, ): AgentAwarenessPhase | null { if (thread.pendingRuntimeRequest?.kind === "user_input") return "waiting_for_input"; if ( @@ -105,6 +142,9 @@ function resolveThreadAwarenessPhaseV2( ) { return "waiting_for_approval"; } + // 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"; switch (thread.activityRunStatus ?? thread.status) { case "preparing": case "starting":