Gateway: activity line, thread history paging, a project's threads, and a thread's subagents

gateway/src/views.ts

@@ -3,8 +3,9 @@
 // are large (hundreds of KB), so they are reduced to a small summary and cached per thread until
 // the thread's `updatedAt` moves.
 
-import { threadPhase, type Phase } from "./awareness.ts";
+import { resolvePhase, threadPhase, type Phase } from "./awareness.ts";
 import type {
+  Activity,
   AgentProvider,
   Feed,
   FeedItem,
@@ -16,18 +17,26 @@ import type {
   RouteKind,
   Summary,
   ThreadDetail,
+  ThreadRow,
+  HistoryResponse,
 } from "./contract.ts";
 import type { T3Host } from "./t3/host.ts";
 import {
   isApprovalRequest,
   isUserInputRequest,
   type BoundedThread,
+  type ConversationMessage,
   type ThreadShell,
+  type TurnItem,
 } from "./t3/types.ts";
 
 const LAST_MESSAGE_CHARS = 280;
 const MESSAGE_CHARS = 2000;
 const MESSAGES_PER_THREAD = 20;
+const ACTIVITY_CHARS = 80;
+/** How long a running thread's summary stays fresh. Its steps change without moving `updatedAt`. */
+const RUNNING_SUMMARY_MS = 5_000;
+const THREAD_LIST_LIMIT = 50;
 const RECENT_LIMIT = 10;
 /** How long a feed request waits for summaries it does not have yet. */
 const SUMMARY_WAIT_MS = 2_000;
@@ -36,6 +45,9 @@ export interface ThreadSummary {
   messages: Message[];
   lastAssistantMessage: string | null;
   request: PendingRequest | null;
+  activity: Activity | null;
+  /** The history route's cursor for messages before `messages`. */
+  earlierCursor: string | null;
 }
 
 /** Trims text for a small screen: collapses whitespace and cuts at a word near `max`. */
@@ -50,22 +62,101 @@ export function trimText(text: string, max: number): string {
 /** Reduces a bounded thread snapshot to what the watch shows. */
 export function summarize(bounded: BoundedThread, thread: ThreadShell): ThreadSummary {
   const { messages, turnItems } = bounded.projection;
-  const visible = messages.filter(
-    (message) => message.text.trim().length > 0 && (message.role === "user" || message.role === "assistant"),
-  );
+  const visible = visibleMessages(messages);
   const lastAssistant = visible.findLast((message) => message.role === "assistant");
+  const start = Math.max(0, visible.length - MESSAGES_PER_THREAD);
   return {
-    messages: visible.slice(-MESSAGES_PER_THREAD).map((message) => ({
-      id: message.id,
-      role: message.role as Message["role"],
-      text: message.text.length > MESSAGE_CHARS ? `${message.text.slice(0, MESSAGE_CHARS)}…` : message.text,
-      createdAt: message.createdAt,
-    })),
+    messages: visible.slice(start).map(toMessage),
     lastAssistantMessage: lastAssistant ? trimText(lastAssistant.text, LAST_MESSAGE_CHARS) : null,
     request: pendingRequest(thread, turnItems),
+    activity: currentActivity(turnItems),
+    earlierCursor: windowCursor(start, bounded),
   };
 }
 
+const visibleMessages = (messages: ConversationMessage[]) =>
+  messages.filter((message) => message.text.trim().length > 0 && (message.role === "user" || message.role === "assistant"));
+
+const toMessage = (message: Pick<ConversationMessage, "id" | "role" | "text" | "createdAt">): Message => ({
+  id: message.id,
+  role: message.role as Message["role"],
+  text: message.text.length > MESSAGE_CHARS ? `${message.text.slice(0, MESSAGE_CHARS)}…` : message.text,
+  createdAt: message.createdAt,
+});
+
+// History cursors are the gateway's own. `w<n>` means the bounded window's messages before index n,
+// which the thread detail leaves out past its 20. `h<cursor>` is T3's cursor for older pages.
+
+/** The cursor for what comes before the window's message `start`. */
+function windowCursor(start: number, bounded: BoundedThread): string | null {
+  if (start > 0) return `w${start}`;
+  return bounded.hasMoreHistory && bounded.historyCursor ? `h${bounded.historyCursor}` : null;
+}
+
+/** Up to 20 messages before `cursor`. T3 pages without messages are skipped, three at most. */
+export async function loadHistory(host: T3Host, thread: ThreadShell, cursor: string): Promise<HistoryResponse> {
+  if (cursor.startsWith("w")) {
+    const end = Number(cursor.slice(1));
+    const bounded = await host.bounded(thread.id);
+    const visible = visibleMessages(bounded.projection.messages);
+    const start = Math.max(0, Math.min(end, visible.length) - MESSAGES_PER_THREAD);
+    return { messages: visible.slice(start, end).map(toMessage), earlierCursor: windowCursor(start, bounded) };
+  }
+  let next: string | null = cursor.slice(1);
+  for (let pages = 0; next && pages < 3; pages++) {
+    const page = await host.history(thread.id, next);
+    next = page.hasMoreHistory ? page.nextCursor : null;
+    const messages = page.items
+      .map(({ item }) => item)
+      .filter(
+        (item) =>
+          (item.type === "user_message" || item.type === "assistant_message") && (item.text ?? "").trim().length > 0,
+      )
+      .map((item) =>
+        toMessage({
+          id: item.messageId ?? "",
+          role: item.type === "user_message" ? "user" : "assistant",
+          text: item.text ?? "",
+          createdAt: item.startedAt ?? item.updatedAt ?? "",
+        }),
+      );
+    if (messages.length > 0) return { messages: messages.slice(-MESSAGES_PER_THREAD), earlierCursor: next && `h${next}` };
+  }
+  return { messages: [], earlierCursor: next && `h${next}` };
+}
+
+/** 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--) {
+    const item = turnItems[i]!;
+    if (item.status !== "running" && item.status !== "pending") continue;
+    const step = describeStep(item);
+    if (step) return { ...step, text: trimText(step.text, ACTIVITY_CHARS), startedAt: item.startedAt ?? null };
+  }
+  return null;
+}
+
+function describeStep(item: TurnItem): Omit<Activity, "startedAt"> | null {
+  switch (item.type) {
+    case "command_execution":
+      return item.monitor || typeof item.input !== "string" ? null : { kind: "command", text: item.input };
+    case "file_change":
+      return item.fileName ? { kind: "edit", text: item.fileName.split("/").pop()! } : null;
+    case "file_search":
+      return { kind: "search", text: item.pattern ?? item.title ?? "files" };
+    case "web_search":
+      return { kind: "web", text: item.title ?? "the web" };
+    case "subagent":
+      return { kind: "subagent", text: item.title ?? "Subagent" };
+    case "reasoning":
+      return { kind: "thinking", text: "Thinking" };
+    case "dynamic_tool":
+      return item.toolName || item.title ? { kind: "tool", text: (item.toolName ?? item.title)! } : null;
+    default:
+      return null;
+  }
+}
+
 function pendingRequest(thread: ThreadShell, turnItems: BoundedThread["projection"]["turnItems"]): PendingRequest | null {
   const pending = thread.pendingRuntimeRequest;
   if (!pending || pending.kind === "auth_refresh") return null;
@@ -109,28 +200,31 @@ function pendingRequest(thread: ThreadShell, turnItems: BoundedThread["projectio
 
 /** Per-thread summaries, refetched when a thread's `updatedAt` changes. */
 export class SummaryCache {
-  private readonly entries = new Map<string, { updatedAt: string; summary: ThreadSummary }>();
+  private readonly entries = new Map<string, { updatedAt: string; fetchedAt: number; summary: ThreadSummary }>();
   private readonly inflight = new Map<string, Promise<ThreadSummary>>();
 
   constructor(private readonly log: (message: string) => void = () => {}) {}
 
-  /** The cached summary, if it matches the thread's current version. */
+  /** The cached summary, if it matches the thread's current version and, while running, is recent. */
   peek(host: T3Host, thread: ThreadShell): ThreadSummary | undefined {
     const entry = this.entries.get(key(host, thread.id));
-    return entry?.updatedAt === thread.updatedAt ? entry.summary : undefined;
+    if (entry?.updatedAt !== thread.updatedAt) return undefined;
+    if (thread.activeRunId !== null && Date.now() - entry.fetchedAt > RUNNING_SUMMARY_MS) return undefined;
+    return entry.summary;
   }
 
   get(host: T3Host, thread: ThreadShell): Promise<ThreadSummary> {
     const cached = this.peek(host, thread);
     if (cached) return Promise.resolve(cached);
     const k = `${key(host, thread.id)}@${thread.updatedAt}`;
+    const fetchedAt = Date.now();
     const running = this.inflight.get(k);
     if (running) return running;
     const promise = host
       .bounded(thread.id)
       .then((bounded) => {
         const summary = summarize(bounded, thread);
-        this.entries.set(key(host, thread.id), { updatedAt: thread.updatedAt, summary });
+        this.entries.set(key(host, thread.id), { updatedAt: thread.updatedAt, fetchedAt, summary });
         return summary;
       })
       .finally(() => this.inflight.delete(k));
@@ -241,6 +335,7 @@ function feedItem(
     phase,
     updatedAt: thread.updatedAt,
     lastMessage: summary?.lastAssistantMessage ?? null,
+    activity: summary?.activity ?? null,
     request: (via ? via.summary : summary)?.request ?? null,
     via: via ? { hostId: via.host.id, threadId: via.thread.id, title: via.thread.title } : null,
   };
@@ -362,9 +457,56 @@ export async function buildThreadDetail(
     messages: summary.messages,
     canStop: thread.activeRunId !== null,
     askFirst: thread.runtimeMode === ASK_FIRST,
+    earlierCursor: summary.earlierCursor,
+    subagents: subagents(hosts, host, thread),
   };
 }
 
+const threadRow = (host: T3Host, thread: ThreadShell): ThreadRow => ({
+  hostId: host.id,
+  threadId: thread.id,
+  title: thread.title,
+  phase: resolvePhase(thread),
+  settled: thread.settledOverride === "settled",
+  updatedAt: thread.updatedAt,
+});
+
+const newestFirst = (a: ThreadRow, b: ThreadRow) => b.updatedAt.localeCompare(a.updatedAt);
+const isLive = (thread: ThreadShell) => !thread.archivedAt && !thread.deletedAt;
+
+/** The subagents `parent` started, its own and the ones it delegated to on any host. */
+function subagents(hosts: T3Host[], parentHost: T3Host, parent: ThreadShell): ThreadRow[] {
+  const rows: ThreadRow[] = [];
+  for (const host of hosts) {
+    for (const thread of host.threads.values()) {
+      if (!isLive(thread)) continue;
+      const native =
+        host === parentHost &&
+        thread.lineage.relationshipToParent === "subagent" &&
+        thread.lineage.parentThreadId === parent.id;
+      const from = thread.delegatedFrom;
+      const delegated = from?.threadId === parent.id && from.environmentId === parentHost.config.environmentId;
+      if (native || delegated) rows.push(threadRow(host, thread));
+    }
+  }
+  return rows.sort(newestFirst);
+}
+
+/** 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()]
+    .filter(
+      (thread) =>
+        thread.projectId === projectId &&
+        isLive(thread) &&
+        thread.lineage.relationshipToParent !== "subagent" &&
+        !thread.delegatedFrom,
+    )
+    .map((thread) => threadRow(host, thread))
+    .sort(newestFirst)
+    .slice(0, THREAD_LIST_LIMIT);
+}
+
 /** T3's runtime mode that asks before each action. */
 export const ASK_FIRST = "approval-required";