From a4754b65b010e970b79e433f8f9170caa2b520ba Mon Sep 17 00:00:00 2001 From: Gabriel Brown Date: Sun, 1 Nov 2026 20:15:22 -0500 Subject: [PATCH] Gateway: show subagent requests on the parent's card, and settle threads --- docs/api.md | 22 +++++++- gateway/src/api.ts | 10 +++- gateway/src/contract.ts | 5 ++ gateway/src/notify.ts | 7 ++- gateway/src/t3/types.ts | 1 + gateway/src/views.ts | 105 +++++++++++++++++++++++++++++++------ gateway/test/views.test.ts | 54 ++++++++++++++++++- 7 files changed, 183 insertions(+), 21 deletions(-) diff --git a/docs/api.md b/docs/api.md index dd3a655..b376bcd 100644 --- a/docs/api.md +++ b/docs/api.md @@ -75,10 +75,17 @@ FeedItem { "phase": "waiting_for_approval", "updatedAt": "2026-11-01T09:41:00.000Z", "lastMessage": "Parser and tests are written. I want to run the suite.", - "request": PendingRequest | null + "request": PendingRequest | null, + "via": { "hostId": "...", "threadId": "...", "title": "Check Panama PATH" } | null } ``` +Threads an agent started for another thread (subagents and `delegate_task` children) do not get +their own cards. When one waits on you, its request moves onto the card of the thread that started +it. `via` names the subagent, `request` and `phase` are the subagent's, and the answer goes to the +subagent: `POST /v1/threads///respond`. A subagent keeps its own card +when the gateway cannot find its parent, or the parent already shows a request. + `phase` is one of `starting`, `running`, `waiting_for_approval`, `waiting_for_input`, `completed`, or `failed`. `lastMessage` is the agent's latest message trimmed to about 280 characters. It can be null for a moment after a thread changes. @@ -133,6 +140,16 @@ POST /v1/threads/:hostId/:threadId/stop 202 ``` +``` +POST /v1/threads/:hostId/:threadId/settle +202 +``` + +Settles a thread, as in T3. A settled thread that finished or failed leaves the feed. To settle all +of Recent, settle each item. + +Thread detail fills `request` and `via` the same way as the feed when a subagent waits on you. + ## Summary The counts behind the watch face complication. It reads no thread pages, so it is cheap to call. @@ -170,7 +187,8 @@ The new thread uses the project's default model, or the host's, and the host's d The gateway pushes when a thread starts waiting for you, finishes, or fails. Each push has an `aps.category` and the custom keys `hostId`, `threadId` (absent on `host` pushes), and, for -approvals and questions, `requestId`. A `test` push carries no custom keys. +approvals and questions, `requestId`. A `test` push carries no custom keys. A subagent's push is titled with the thread that started it, +its body starts with "Subagent :", and its `hostId` and `threadId` are the subagent's. | Category | When | Actions the watch registers | | --- | --- | --- | diff --git a/gateway/src/api.ts b/gateway/src/api.ts index a4a2b28..d534e12 100644 --- a/gateway/src/api.ts +++ b/gateway/src/api.ts @@ -232,7 +232,7 @@ export function createRoutes({ db, hub, cache, pusher, startedAt, log = () => {} "/v1/threads/:hostId/:threadId": { GET: authed<"/v1/threads/:hostId/:threadId">(async (req) => { const { host, thread } = findThread(req.params.hostId, req.params.threadId); - return json(await buildThreadDetail(host, thread, cache)); + return json(await buildThreadDetail(host, thread, cache, hub.list())); }), }, @@ -277,6 +277,14 @@ export function createRoutes({ db, hub, cache, pusher, startedAt, log = () => {} }), }, + "/v1/threads/:hostId/:threadId/settle": { + POST: authed<"/v1/threads/:hostId/:threadId/settle">(async (req) => { + const { host, thread } = findThread(req.params.hostId, req.params.threadId); + await host.dispatch({ type: "thread.settle", commandId: crypto.randomUUID(), threadId: thread.id }); + return accepted(); + }), + }, + "/v1/threads/:hostId/:threadId/stop": { POST: authed<"/v1/threads/:hostId/:threadId/stop">(async (req) => { const { host, thread } = findThread(req.params.hostId, req.params.threadId); diff --git a/gateway/src/contract.ts b/gateway/src/contract.ts index b66184f..0d11680 100644 --- a/gateway/src/contract.ts +++ b/gateway/src/contract.ts @@ -105,6 +105,11 @@ export interface FeedItem { /** The agent's latest message, trimmed for the watch. Null until the gateway has read it. */ lastMessage: string | null; request: PendingRequest | null; + /** + * Set when `request` comes from a subagent this thread delegated to, rather than the thread + * itself. Answer the request on that thread. `phase` is then the subagent's. + */ + via: { hostId: string; threadId: string; title: string } | null; } /** Counts for the watch face complication. Cheap: no thread pages are read. */ diff --git a/gateway/src/notify.ts b/gateway/src/notify.ts index ca9304f..b5f5120 100644 --- a/gateway/src/notify.ts +++ b/gateway/src/notify.ts @@ -91,10 +91,13 @@ export class Notifier { ]); const project = host.projects.get(thread.projectId)?.title; const requestId = thread.pendingRuntimeRequest?.id; + // A subagent's push is about the thread that delegated to it, which is the one the user knows. + const parent = thread.delegatedFrom?.title; + const text = body(phase, thread.lastError ?? null, summary); await this.send({ - title: thread.title, + title: parent ?? thread.title, subtitle: project ? `${host.label} ยท ${project}` : host.label, - body: body(phase, thread.lastError ?? null, summary), + body: parent ? `Subagent ${thread.title}: ${text}` : text, category: category(phase), data: { hostId: host.id, threadId: thread.id, ...(requestId ? { requestId } : {}) }, collapseId: `${host.id}:${thread.id}`.slice(0, 64), diff --git a/gateway/src/t3/types.ts b/gateway/src/t3/types.ts index 4259958..1e83f03 100644 --- a/gateway/src/t3/types.ts +++ b/gateway/src/t3/types.ts @@ -182,6 +182,7 @@ export type Command = answers?: Record<string, unknown>; } | { type: "thread.stop"; commandId: string; threadId: string } + | { type: "thread.settle"; commandId: string; threadId: string } | { type: "run.interrupt"; commandId: string; threadId: string; runId: string }; export interface LaunchThreadInput { diff --git a/gateway/src/views.ts b/gateway/src/views.ts index 191133e..6d38deb 100644 --- a/gateway/src/views.ts +++ b/gateway/src/views.ts @@ -180,7 +180,56 @@ export function routeKind(url: string): RouteKind { return hostname.includes("vpn") ? "vpn" : "direct"; } -function feedItem(host: T3Host, thread: ThreadShell, phase: Phase, summary: ThreadSummary | undefined): FeedItem { +/** A thread on the feed. With `via`, the card shows that subagent's request on this thread. */ +interface Candidate { + host: T3Host; + thread: ThreadShell; + phase: Phase; + via?: { host: T3Host; thread: ThreadShell; phase: Phase }; +} + +const isWaiting = (phase: Phase | null) => phase === "waiting_for_approval" || phase === "waiting_for_input"; + +/** The thread that delegated `child`, if the gateway is connected to its host. */ +function delegator(hosts: T3Host[], child: ThreadShell): { host: T3Host; thread: ThreadShell } | null { + const from = child.delegatedFrom; + if (!from) return null; + const host = hosts.find((candidate) => candidate.config.environmentId === from.environmentId); + const thread = host?.threads.get(from.threadId); + return host && thread ? { host, thread } : null; +} + +/** + * Moves each waiting subagent's request onto the card of the thread that delegated it, the oldest + * request first. A subagent keeps its own card when its parent is unknown, hidden, archived, or + * already showing a request. + */ +function attachDelegates(candidates: Candidate[], hosts: T3Host[]): Candidate[] { + const result = [...candidates]; + const delegates = candidates + .filter(({ thread, phase }) => thread.delegatedFrom && isWaiting(phase)) + .sort((a, b) => requestAge(a.thread).localeCompare(requestAge(b.thread))); + for (const child of delegates) { + const parent = delegator(hosts, child.thread); + const index = parent ? result.findIndex(({ host, thread }) => host === parent.host && thread.id === parent.thread.id) : -1; + const card = result[index]; + const own = result.indexOf(child); + if (!card || own === -1 || card.via || isWaiting(card.phase)) continue; + result[index] = { ...card, phase: child.phase, via: { host: child.host, thread: child.thread, phase: child.phase } }; + result.splice(own, 1); + } + return result; +} + +const requestAge = (thread: ThreadShell) => thread.pendingRuntimeRequest?.createdAt ?? ""; + +function feedItem( + host: T3Host, + thread: ThreadShell, + phase: Phase, + summary: ThreadSummary | undefined, + via?: { host: T3Host; thread: ThreadShell; summary: ThreadSummary | undefined }, +): FeedItem { return { hostId: host.id, hostLabel: host.label, @@ -191,10 +240,26 @@ function feedItem(host: T3Host, thread: ThreadShell, phase: Phase, summary: Thre phase, updatedAt: thread.updatedAt, lastMessage: summary?.lastAssistantMessage ?? null, - request: summary?.request ?? null, + request: (via ? via.summary : summary)?.request ?? null, + via: via ? { hostId: via.host.id, threadId: via.thread.id, title: via.thread.title } : null, }; } +/** The longest waiting subagent that `parent` delegated to, if any. */ +function waitingDelegate(hosts: T3Host[], parentHost: T3Host, parent: ThreadShell): Candidate | undefined { + let oldest: Candidate | undefined; + for (const host of hosts) { + for (const thread of host.threads.values()) { + const from = thread.delegatedFrom; + if (from?.threadId !== parent.id || from.environmentId !== parentHost.config.environmentId) continue; + const phase = threadPhase(thread); + if (!isWaiting(phase) || thread.archivedAt || thread.deletedAt) continue; + if (!oldest || requestAge(thread) < requestAge(oldest.thread)) oldest = { host, thread, phase: phase! }; + } + } + return oldest; +} + /** Counts for the complication, from thread shells alone. */ export function buildSummary(hosts: T3Host[]): Summary { let needsYou = 0; @@ -221,37 +286,37 @@ export function buildSummary(hosts: T3Host[]): Summary { /** Builds the feed across every host, waiting briefly for summaries it has not read yet. */ export async function buildFeed(hosts: T3Host[], cache: SummaryCache, now = Date.now()): Promise<Feed> { - const candidates: Array<{ host: T3Host; thread: ThreadShell; phase: Phase }> = []; + const found: Candidate[] = []; for (const host of hosts) { for (const thread of host.threads.values()) { const phase = threadPhase(thread); if (phase === null || thread.archivedAt || thread.deletedAt) continue; if ((phase === "completed" || phase === "failed") && now - Date.parse(thread.updatedAt) > RECENT_WINDOW_MS) continue; - candidates.push({ host, thread, phase }); + found.push({ host, thread, phase }); } } + const candidates = attachDelegates(found, hosts); const byNewest = (a: { thread: ThreadShell }, b: { thread: ThreadShell }) => b.thread.updatedAt.localeCompare(a.thread.updatedAt); + const asker = (candidate: Candidate) => candidate.via?.thread ?? candidate.thread; const needsYou = candidates - .filter(({ phase }) => phase === "waiting_for_approval" || phase === "waiting_for_input") - .sort((a, b) => - (a.thread.pendingRuntimeRequest?.createdAt ?? "").localeCompare(b.thread.pendingRuntimeRequest?.createdAt ?? ""), - ); + .filter(({ phase }) => isWaiting(phase)) + .sort((a, b) => requestAge(asker(a)).localeCompare(requestAge(asker(b)))); const working = candidates.filter(({ phase }) => phase === "starting" || phase === "running").sort(byNewest); const recent = candidates .filter(({ phase }) => phase === "completed" || phase === "failed") .sort(byNewest) .slice(0, RECENT_LIMIT); - const shown = [...needsYou, ...working, ...recent]; + const shown = [...needsYou, ...working, ...recent].flatMap((candidate) => (candidate.via ? [candidate, candidate.via] : [candidate])); const missing = shown .filter(({ host, thread }) => !cache.peek(host, thread)) .map(({ host, thread }) => cache.get(host, thread).catch(() => undefined)); if (missing.length > 0) await Promise.race([Promise.all(missing), Bun.sleep(SUMMARY_WAIT_MS)]); - const toItem = ({ host, thread, phase }: (typeof candidates)[number]) => - feedItem(host, thread, phase, cache.peek(host, thread)); + const toItem = ({ host, thread, phase, via }: Candidate) => + feedItem(host, thread, phase, cache.peek(host, thread), via && { ...via, summary: cache.peek(via.host, via.thread) }); return { needsYou: needsYou.map(toItem), working: working.map(toItem), @@ -260,12 +325,22 @@ export async function buildFeed(hosts: T3Host[], cache: SummaryCache, now = Date }; } -/** One thread with its recent messages. Fetches the summary if it is not cached. */ -export async function buildThreadDetail(host: T3Host, thread: ThreadShell, cache: SummaryCache): Promise<ThreadDetail> { +/** + * One thread with its recent messages. Fetches the summary if it is not cached. When the thread + * has no request of its own but a subagent it delegated to is waiting, shows that request. + */ +export async function buildThreadDetail( + host: T3Host, + thread: ThreadShell, + cache: SummaryCache, + hosts: T3Host[], +): Promise<ThreadDetail> { const summary = await cache.get(host, thread); - const phase = threadPhase(thread) ?? (thread.status === "failed" ? "failed" : "completed"); + const own = threadPhase(thread) ?? (thread.status === "failed" ? "failed" : "completed"); + const child = isWaiting(own) ? undefined : waitingDelegate(hosts, host, thread); + const via = child && { host: child.host, thread: child.thread, summary: await cache.get(child.host, child.thread).catch(() => undefined) }; return { - item: feedItem(host, thread, phase, summary), + item: feedItem(host, thread, child?.phase ?? own, summary, via), messages: summary.messages, canStop: thread.activeRunId !== null, }; diff --git a/gateway/test/views.test.ts b/gateway/test/views.test.ts index da5b1ef..0cc792a 100644 --- a/gateway/test/views.test.ts +++ b/gateway/test/views.test.ts @@ -2,7 +2,7 @@ import { describe, expect, it } from "bun:test"; import type { BoundedThread, ThreadShell } from "../src/t3/types.ts"; import type { T3Host } from "../src/t3/host.ts"; -import { buildSummary, routeKind, summarize, trimText } from "../src/views.ts"; +import { buildFeed, buildSummary, routeKind, summarize, SummaryCache, trimText } from "../src/views.ts"; const thread = (overrides: Partial<ThreadShell> = {}): ThreadShell => ({ id: "thread-1", @@ -138,3 +138,55 @@ describe("routeKind", () => { expect(routeKind("https://t3laptopvpn.gbrown.org")).toBe("public"); }); }); + +describe("buildFeed", () => { + const now = Date.parse("2026-11-01T09:30:00.000Z"); + const host = (label: string, threads: ThreadShell[]) => + ({ + id: label, + label, + status: "online", + config: { environmentId: `env-${label}` }, + projects: new Map(), + threads: new Map(threads.map((t) => [t.id, t])), + bounded: () => Promise.reject(new Error("not needed")), + }) as unknown as T3Host; + const ask = (id: string, minute: number) => ({ id, kind: "command", createdAt: `2026-11-01T09:${minute}:00.000Z` }); + const delegatedFrom = { environmentId: "env-Desktop", threadId: "parent", title: "HomeLab Stats" }; + + it("shows a waiting subagent's request on the card of the thread that delegated it", async () => { + const feed = await buildFeed( + [ + host("Desktop", [thread({ id: "parent", title: "HomeLab Stats" })]), + host("Mac mini", [ + thread({ id: "child", title: "Check PATH", delegatedFrom, pendingRuntimeRequest: ask("r1", 10) }), + thread({ id: "busy", title: "Survey setup", delegatedFrom }), + ]), + ], + new SummaryCache(), + now, + ); + expect(feed.working).toEqual([]); + expect(feed.needsYou.map((item) => [item.threadId, item.phase, item.via])).toEqual([ + ["parent", "waiting_for_approval", { hostId: "Mac mini", threadId: "child", title: "Check PATH" }], + ]); + }); + + it("keeps a second waiting subagent on its own card", async () => { + const feed = await buildFeed( + [ + host("Desktop", [thread({ id: "parent", title: "HomeLab Stats" })]), + host("Mac mini", [ + thread({ id: "later", delegatedFrom, pendingRuntimeRequest: ask("r2", 20) }), + thread({ id: "first", delegatedFrom, pendingRuntimeRequest: ask("r1", 10) }), + ]), + ], + new SummaryCache(), + now, + ); + expect(feed.needsYou.map((item) => [item.threadId, item.via?.threadId ?? null])).toEqual([ + ["parent", "first"], + ["later", null], + ]); + }); +});