Gateway: changed files, search across hosts, a richer complication summary, and question options in pushes

gateway/src/api.ts

@@ -37,6 +37,7 @@ import {
   hostSummary,
   loadHistory,
   projectThreads,
+  searchThreads,
   type SummaryCache,
 } from "./views.ts";
 
@@ -259,7 +260,7 @@ export function createRoutes({ db, hub, cache, pusher, startedAt, log = () => {}
     },
 
     "/v1/summary": {
-      GET: authed(() => json(buildSummary(hub.list()))),
+      GET: authed(async () => json(await buildSummary(hub.list(), cache))),
     },
 
     "/v1/projects": {
@@ -280,6 +281,14 @@ export function createRoutes({ db, hub, cache, pusher, startedAt, log = () => {}
       }),
     },
 
+    "/v1/search": {
+      GET: authed(async (req) => {
+        const query = (new URL(req.url).searchParams.get("q") ?? "").trim().slice(0, 200);
+        if (query.length < 2) throw new HttpError(400, "Search for at least two letters.");
+        return json(await searchThreads(hub.list(), query));
+      }),
+    },
+
     "/v1/projects/:hostId/:projectId/threads": {
       GET: authed<"/v1/projects/:hostId/:projectId/threads">((req) => {
         const host = findHost(req.params.hostId);

gateway/src/contract.ts

@@ -118,6 +118,8 @@ export interface FeedItem {
   lastMessage: string | null;
   /** What a working thread is doing right now. Null when nothing is in progress or not read yet. */
   activity: Activity | null;
+  /** What the latest run changed, in total. Null when it changed no files or is not read yet. */
+  changes: ChangeTotals | null;
   request: PendingRequest | null;
   /**
    * Set when `request` comes from a subagent this thread delegated to, rather than the thread
@@ -135,6 +137,35 @@ export interface Activity {
   startedAt: string | null;
 }
 
+export interface ChangeTotals {
+  files: number;
+  additions: number;
+  deletions: number;
+}
+
+/** A file the latest run changed. `name` is the last part of `path`. */
+export interface ChangedFile {
+  path: string;
+  name: string;
+  additions: number;
+  deletions: number;
+}
+
+/** A thread search hit: the thread, and where the words matched when it was inside a message. */
+export interface SearchHit {
+  thread: ThreadRow;
+  projectTitle: string;
+  hostLabel: string;
+  /** Text around the match in a message, or null when only the title matched. */
+  snippet: string | null;
+}
+
+export interface SearchResponse {
+  hits: SearchHit[];
+  /** Labels of hosts that were offline or failed, so their threads were not searched. */
+  unsearched: string[];
+}
+
 /** A thread in a list: a project's threads, or a thread's subagents. */
 export interface ThreadRow {
   hostId: string;
@@ -162,7 +193,16 @@ export interface Summary {
   needsYou: number;
   working: number;
   /** The thread that has waited longest for you, if any. */
-  next: { hostId: string; threadId: string; title: string; hostLabel: string } | null;
+  next: {
+    hostId: string;
+    threadId: string;
+    title: string;
+    hostLabel: string;
+    /** The command it wants to run or its first question, trimmed. Null until the gateway has read it. */
+    request: string | null;
+  } | null;
+  /** The working thread updated most recently, with what it is doing. Null when nothing is working. */
+  current: { hostId: string; threadId: string; title: string; activity: Activity | null } | null;
   hostsOffline: number;
 }
 
@@ -208,6 +248,8 @@ export interface ThreadDetail {
   earlierCursor: string | null;
   /** Subagents and delegated threads this thread started, newest first. */
   subagents: ThreadRow[];
+  /** The files the latest run changed, most lines changed first. */
+  changedFiles: ChangedFile[];
 }
 
 export interface ThreadModeRequest {
@@ -290,6 +332,11 @@ export interface PushPayload {
   /** Absent on `host` pushes, which are about a host rather than a thread. */
   threadId?: string;
   requestId?: string;
+  /**
+   * On `input` pushes that ask one single-choice question: its options, so the notification can
+   * offer each as a button. `allowCustomAnswer` adds a spoken answer.
+   */
+  question?: { id: string; options: Array<{ label: string; value: string }>; allowCustomAnswer: boolean };
 }
 
 /** Notification categories the watch registers actions for. */

gateway/src/notify.ts

@@ -102,12 +102,30 @@ export class Notifier {
       body: parent ? `Subagent ${thread.title}: ${text}` : text,
       category: category(phase),
       fromSubagent: parent !== undefined,
-      data: { hostId: host.id, threadId: thread.id, ...(requestId ? { requestId } : {}) },
+      data: { hostId: host.id, threadId: thread.id, ...(requestId ? { requestId } : {}), ...pushQuestion(phase, summary) },
       collapseId: `${host.id}:${thread.id}`.slice(0, 64),
     });
   }
 }
 
+/**
+ * A lone single-choice question's options, so the notification can offer them as buttons. Labels
+ * are trimmed and the list capped to keep the payload under APNs' 4 KB.
+ */
+function pushQuestion(phase: Phase, summary: ThreadSummary | undefined): Pick<PushPayload, "question"> {
+  const questions = summary?.request?.questions ?? [];
+  const question = questions[0];
+  if (phase !== "waiting_for_input" || questions.length !== 1 || !question || question.multiSelect) return {};
+  if (question.options.length === 0 && !question.allowCustomAnswer) return {};
+  return {
+    question: {
+      id: question.id,
+      options: question.options.slice(0, 4).map(({ label, value }) => ({ label: trimText(label, 40), value: value.slice(0, 200) })),
+      allowCustomAnswer: question.allowCustomAnswer,
+    },
+  };
+}
+
 function category(phase: Phase): PushCategory {
   switch (phase) {
     case "waiting_for_approval":

gateway/src/t3/host.ts

@@ -8,6 +8,7 @@ import {
   type Command,
   type EnvironmentDescriptor,
   type HistoryPage,
+  type SearchMatch,
   type LaunchThreadInput,
   type Project,
   type ServerConfig,
@@ -176,6 +177,17 @@ export class T3Host {
     return this.http<BoundedThread>(`/api/orchestration/threads/${encodeURIComponent(threadId)}/bounded`);
   }
 
+  /** Threads whose titles or messages match `query`, best first. Fails after 5 s. */
+  async search(query: string): Promise<SearchMatch[]> {
+    const { matches } = await Promise.race([
+      this.requireSocket().call<{ matches: SearchMatch[] }>("orchestration.searchThreads", { query, limit: 20 }),
+      Bun.sleep(5_000).then(() => {
+        throw new HostError(`${this.label} took too long to search`, 504);
+      }),
+    ]);
+    return matches;
+  }
+
   /** One page of conversation history older than `cursor`, from `historyCursor` or a page's `nextCursor`. */
   history(threadId: string, cursor: string): Promise<HistoryPage> {
     const query = new URLSearchParams({ cursor, view: "conversation" });

gateway/src/t3/types.ts

@@ -123,8 +123,11 @@ export interface TurnItem {
   input?: unknown;
   /** Set on `command_execution` items that are monitors rather than commands. */
   monitor?: boolean;
-  /** The file, on `file_change`. */
+  runId?: string | null;
+  /** The file, on `file_change`, with its line counts. */
   fileName?: string;
+  additions?: number;
+  deletions?: number;
   /** The search, on `file_search`. */
   pattern?: string;
   /** The tool, on `dynamic_tool`. */
@@ -167,6 +170,15 @@ export interface BoundedThread {
   };
 }
 
+/** One hit from `orchestration.searchThreads`, best first. */
+export interface SearchMatch {
+  threadId: string;
+  projectId: string;
+  source: Open<"user" | "assistant">;
+  snippet: string;
+  messageCreatedAt: string | null;
+}
+
 /** One page of `GET /api/orchestration/threads/:threadId/history`, older than the cursor. */
 export interface HistoryPage {
   items: Array<{ item: TurnItem }>;

gateway/src/views.ts

@@ -7,6 +7,7 @@ import { resolvePhase, threadPhase, type Phase } from "./awareness.ts";
 import type {
   Activity,
   AgentProvider,
+  ChangedFile,
   Feed,
   FeedItem,
   HostStatusDetail,
@@ -19,6 +20,7 @@ import type {
   ThreadDetail,
   ThreadRow,
   HistoryResponse,
+  SearchResponse,
 } from "./contract.ts";
 import type { T3Host } from "./t3/host.ts";
 import {
@@ -46,6 +48,8 @@ export interface ThreadSummary {
   lastAssistantMessage: string | null;
   request: PendingRequest | null;
   activity: Activity | null;
+  /** The files the latest run in the window changed. */
+  changedFiles: ChangedFile[];
   /** The history route's cursor for messages before `messages`. */
   earlierCursor: string | null;
 }
@@ -70,6 +74,7 @@ export function summarize(bounded: BoundedThread, thread: ThreadShell): ThreadSu
     lastAssistantMessage: lastAssistant ? trimText(lastAssistant.text, LAST_MESSAGE_CHARS) : null,
     request: pendingRequest(thread, turnItems),
     activity: currentActivity(turnItems),
+    changedFiles: latestChanges(turnItems),
     earlierCursor: windowCursor(start, bounded),
   };
 }
@@ -128,6 +133,21 @@ export async function loadHistory(host: T3Host, thread: ThreadShell, cursor: str
   return { messages: [], earlierCursor: next && `h${next}` };
 }
 
+/** The files the latest run changed, summed per file, most lines changed first. */
+export function latestChanges(turnItems: TurnItem[]): ChangedFile[] {
+  const runId = turnItems.findLast((item) => item.runId)?.runId;
+  if (!runId) return [];
+  const files = new Map<string, ChangedFile>();
+  for (const item of turnItems) {
+    if (item.type !== "file_change" || item.runId !== runId || !item.fileName) continue;
+    const file = files.get(item.fileName) ?? { path: item.fileName, name: item.fileName.split("/").pop()!, additions: 0, deletions: 0 };
+    file.additions += item.additions ?? 0;
+    file.deletions += item.deletions ?? 0;
+    files.set(item.fileName, file);
+  }
+  return [...files.values()].sort((a, b) => b.additions + b.deletions - (a.additions + a.deletions));
+}
+
 /** The newest step still in progress, said briefly. Monitors are left out, as they run for hours. */
 export function currentActivity(turnItems: TurnItem[]): Activity | null {
   for (let i = turnItems.length - 1; i >= 0; i--) {
@@ -341,6 +361,13 @@ function feedItem(
     updatedAt: thread.updatedAt,
     lastMessage: summary?.lastAssistantMessage ?? null,
     activity: summary?.activity ?? null,
+    changes: summary?.changedFiles.length
+      ? {
+          files: summary.changedFiles.length,
+          additions: summary.changedFiles.reduce((sum, file) => sum + file.additions, 0),
+          deletions: summary.changedFiles.reduce((sum, file) => sum + file.deletions, 0),
+        }
+      : null,
     request: (via ? via.summary : summary)?.request ?? null,
     via: via ? { hostId: via.host.id, threadId: via.thread.id, title: via.thread.title } : null,
   };
@@ -362,25 +389,55 @@ function waitingDelegate(hosts: T3Host[], parentHost: T3Host, parent: ThreadShel
 }
 
 /** Counts for the complication, from thread shells alone. */
-export function buildSummary(hosts: T3Host[]): Summary {
+/**
+ * Counts for the complication, plus the longest waiting request and the busiest working thread.
+ * Reads at most those two threads, waiting briefly as the feed does.
+ */
+export async function buildSummary(hosts: T3Host[], cache: SummaryCache): Promise<Summary> {
   let needsYou = 0;
   let working = 0;
   let next: { host: T3Host; thread: ThreadShell } | null = null;
+  let current: { host: T3Host; thread: ThreadShell } | null = null;
   for (const host of hosts) {
     for (const thread of host.threads.values()) {
       if (thread.archivedAt || thread.deletedAt) continue;
       const phase = threadPhase(thread);
-      if (phase === "starting" || phase === "running") working++;
+      if (phase === "starting" || phase === "running") {
+        working++;
+        if (!current || thread.updatedAt > current.thread.updatedAt) current = { host, thread };
+      }
       if (phase !== "waiting_for_approval" && phase !== "waiting_for_input") continue;
       needsYou++;
       const waitingSince = thread.pendingRuntimeRequest?.createdAt ?? "";
       if (!next || waitingSince < (next.thread.pendingRuntimeRequest?.createdAt ?? "")) next = { host, thread };
     }
   }
+  const read = [next, current].filter((pick) => pick !== null);
+  const missing = read.filter(({ host, thread }) => !cache.peek(host, thread));
+  if (missing.length > 0) {
+    await Promise.race([
+      Promise.all(missing.map(({ host, thread }) => cache.get(host, thread).catch(() => undefined))),
+      Bun.sleep(SUMMARY_WAIT_MS),
+    ]);
+  }
+  const request = next && cache.peek(next.host, next.thread)?.request;
+  const asked = request?.prompt ?? request?.questions[0]?.question;
   return {
     needsYou,
     working,
-    next: next && { hostId: next.host.id, threadId: next.thread.id, title: next.thread.title, hostLabel: next.host.label },
+    next: next && {
+      hostId: next.host.id,
+      threadId: next.thread.id,
+      title: next.thread.title,
+      hostLabel: next.host.label,
+      request: asked ? trimText(asked, ACTIVITY_CHARS) : null,
+    },
+    current: current && {
+      hostId: current.host.id,
+      threadId: current.thread.id,
+      title: current.thread.title,
+      activity: cache.peek(current.host, current.thread)?.activity ?? null,
+    },
     hostsOffline: hosts.filter((host) => host.status !== "online").length,
   };
 }
@@ -464,6 +521,7 @@ export async function buildThreadDetail(
     askFirst: thread.runtimeMode === ASK_FIRST,
     earlierCursor: summary.earlierCursor,
     subagents: subagents(hosts, host, thread),
+    changedFiles: summary.changedFiles,
   };
 }
 
@@ -497,6 +555,43 @@ function subagents(hosts: T3Host[], parentHost: T3Host, parent: ThreadShell): Th
   return rows.sort(newestFirst);
 }
 
+/**
+ * Searches every online host at once and merges the hits, newest match first. Hosts that are
+ * offline or fail are named in `unsearched`.
+ */
+export async function searchThreads(hosts: T3Host[], query: string): Promise<SearchResponse> {
+  const results = await Promise.all(
+    hosts.map(async (host) => {
+      if (host.status !== "online") return null;
+      try {
+        return { host, matches: await host.search(query) };
+      } catch {
+        return null;
+      }
+    }),
+  );
+  const hits = results.flatMap((result) =>
+    (result?.matches ?? []).flatMap((match) => {
+      const thread = result!.host.threads.get(match.threadId);
+      if (!thread || !isLive(thread)) return [];
+      const snippet = match.snippet.trim();
+      return [
+        {
+          thread: threadRow(result!.host, thread),
+          projectTitle: result!.host.projects.get(thread.projectId)?.title ?? "",
+          hostLabel: result!.host.label,
+          snippet: snippet && snippet !== thread.title ? trimText(snippet, 160) : null,
+          at: match.messageCreatedAt ?? thread.updatedAt,
+        },
+      ];
+    }),
+  );
+  return {
+    hits: hits.sort((a, b) => b.at.localeCompare(a.at)).map(({ at: _, ...hit }) => hit),
+    unsearched: hosts.filter((host, i) => results[i] === null).map((host) => host.label),
+  };
+}
+
 /** A project's newest 50 threads, settled ones included. Subagents are listed under their parents. */
 export function projectThreads(host: T3Host, projectId: string): ThreadRow[] {
   return [...host.threads.values()]