Gateway: show subagent requests on the parent's card, and settle threads
Gateway / check (push) Successful in 8s
Gateway / image (push) Successful in 21s

This commit is contained in:
gib committed 2026-11-01 20:15:22 -05:00
1 parent a8b6375afe
commit a4754b65b0
7 files changed
+183 -21

No files matched your search

+9 -1
View File
@@ -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);
+5
View File
@@ -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. */
+5 -2
View File
@@ -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),
+1
View File
@@ -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 {
+90 -15
View File
@@ -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,
};