fix: track opencode subagent status per pane (#4357)

refs #1362

1 files changed, 326 insertions, 50 deletions

diff --git a/src/integration/assets/opencode/herdr-tui-session.js b/src/integration/assets/opencode/herdr-tui-session.js
index 4e5631ac..d21c1f7e 100644
--- a/src/integration/assets/opencode/herdr-tui-session.js
+++ b/src/integration/assets/opencode/herdr-tui-session.js
@@ -1,7 +1,7 @@
 // installed by herdr
 // managed by herdr; reinstalling or updating the integration overwrites this file.
 // HERDR_INTEGRATION_ID=opencode-tui
-// HERDR_INTEGRATION_VERSION=12
+// HERDR_INTEGRATION_VERSION=13
 
 import net from "node:net";
 
@@ -67,65 +67,341 @@ export default {
   // Keep this plain object dependency-free: V1 and V2 expose different SDK
   // packages, but both loaders accept their own lifecycle entry on this object.
   setup,
-  tui: async (api) => {
-    if (
-      process.env.HERDR_ENV !== "1" ||
-      !process.env.HERDR_SOCKET_PATH ||
-      !process.env.HERDR_PANE_ID
-    ) {
-      return;
+  tui,
+};
+
+async function tui(api) {
+  if (process.env.HERDR_ENV !== "1" || !process.env.HERDR_SOCKET_PATH || !process.env.HERDR_PANE_ID) return;
+
+  let disposed = false;
+  let context;
+  let sequence = Date.now() * 1000;
+  let chain = Promise.resolve();
+
+  function routeID() {
+    const route = api.route.current;
+    return route?.name === "session" ? route.params?.sessionID : undefined;
+  }
+
+  function current(ctx) {
+    return !disposed && context === ctx && routeID() === ctx.route;
+  }
+
+  async function read(ctx, request) {
+    const result = await request({
+      signal: AbortSignal.any([ctx.controller.signal, AbortSignal.timeout(5_000)]),
+      throwOnError: true,
+    });
+    if (!current(ctx) || result?.data === undefined) throw new Error("session data unavailable");
+    return result.data;
+  }
+
+  // The first selected session owns this pane's subtree. Browsing descendants
+  // keeps that boundary; directly attaching to a child does not claim siblings.
+  function root(ctx, id) {
+    const seen = new Set();
+    while (typeof id === "string" && !seen.has(id) && !ctx.deleted.has(id)) {
+      seen.add(id);
+      if (id === ctx.boundary) return id;
+      const session = ctx.sessions.get(id) ?? api.state.session.get(id);
+      if (!session) return;
+      if (!session.parentID) return id;
+      id = session.parentID;
     }
+  }
 
-    let selectedSessionID;
-    let retryIndex = 0;
-    let nextReportAt = 0;
-    let reportPending = false;
-    const syncSelectedSession = async () => {
-      const route = api.route.current;
-      const sessionID = route?.name === "session" ? route.params?.sessionID : undefined;
-      const session =
-        typeof sessionID === "string" && sessionID
-          ? api.state.session.get(sessionID)
-          : undefined;
-      if (!session || session.parentID) {
-        selectedSessionID = undefined;
-        retryIndex = 0;
-        nextReportAt = 0;
-        return;
+  async function resolveRoot(ctx, id) {
+    const seen = new Set();
+    while (typeof id === "string" && !seen.has(id) && !ctx.deleted.has(id)) {
+      seen.add(id);
+      if (id === ctx.boundary) return id;
+      let session = ctx.sessions.get(id) ?? api.state.session.get(id);
+      if (!session) {
+        if (!ctx.lookups.has(id)) {
+          const requestedID = id;
+          const lookup = read(ctx, (options) => api.client.session.get({ sessionID: requestedID }, options));
+          ctx.lookups.set(id, lookup);
+          lookup.finally(() => ctx.lookups.delete(requestedID)).catch(() => {});
+        }
+        session = await ctx.lookups.get(id);
+        if (ctx.deleted.has(id)) return;
+        session = ctx.sessions.get(id) ?? session;
+        if (session.id !== id) throw new Error("unexpected session identity");
+        ctx.sessions.set(id, session);
       }
-      if (sessionID !== selectedSessionID) {
-        selectedSessionID = sessionID;
-        retryIndex = 0;
-        nextReportAt = 0;
+      if (!session.parentID) return id;
+      id = session.parentID;
+    }
+  }
+
+  function owners(ctx) {
+    return new Set([...ctx.statuses.keys(), ...[...ctx.blockers.values()].map((r) => r.sessionID)]);
+  }
+
+  function state(ctx) {
+    if (ctx.settled) return "idle";
+    for (const request of ctx.blockers.values()) {
+      if (root(ctx, request.sessionID) === ctx.selected) return "blocked";
+    }
+    if (ctx.errors.has(ctx.selected)) return "blocked";
+    for (const id of ctx.statuses.keys()) {
+      if (root(ctx, id) === ctx.selected) return "working";
+    }
+    if (ctx.hydrated && [...owners(ctx)].every((id) => root(ctx, id) !== undefined)) return "idle";
+  }
+
+  function publish(ctx, selection = false) {
+    if (!current(ctx) || !ctx.selected) return;
+    ctx.selectionPending ||= selection;
+    if (ctx.queued) return;
+    ctx.queued = true;
+    chain = chain.then(async () => {
+      ctx.queued = false;
+      const selected = ctx.selected;
+      const isCurrent = () => current(ctx) && !!selected && ctx.selected === selected &&
+        (ctx.settled || root(ctx, ctx.route) === selected);
+      if (!isCurrent()) return;
+      if (ctx.selectionPending) {
+        if (!await requestOnce(selected, undefined, undefined, isCurrent)) {
+          ctx.retryAt = Date.now() + 500;
+          return;
+        }
+        if (!isCurrent()) return;
+        ctx.selectionPending = false;
+        ctx.lastState = undefined;
       }
-      if (reportPending || Date.now() < nextReportAt) {
-        return;
+      const value = state(ctx);
+      if (value === undefined || value === ctx.lastState) return;
+      const delivered = await requestOnce(selected, value, ++sequence, isCurrent);
+      if (!isCurrent()) return;
+      ctx.lastState = delivered ? value : undefined;
+      if (!delivered) ctx.retryAt = Date.now() + 500;
+    }).catch(() => {
+      if (current(ctx)) {
+        ctx.lastState = undefined;
+        ctx.retryAt = Date.now() + 500;
+      }
+    });
+  }
+
+  function retire(ctx, route) {
+    ctx?.controller?.abort();
+    context = ctx?.selected
+      ? { route, selected: ctx.selected, settled: true, retryAt: Infinity }
+      : undefined;
+    if (context) publish(context);
+  }
+
+  function reconcile(ctx) {
+    if (!current(ctx)) return;
+    publish(ctx);
+    if (ctx.resolving) {
+      ctx.resolveAgain = true;
+      return;
+    }
+    ctx.resolving = true;
+    void (async () => {
+      const ancestor = await resolveRoot(ctx, ctx.route);
+      if (!current(ctx) || !ancestor) return;
+      const selected = ancestor === ctx.boundary ? ctx.boundary : ctx.route;
+      if (selected !== ctx.selected) {
+        ctx.boundary = selected;
+        ctx.selected = selected;
+        publish(ctx, true);
       }
+      // Fetch ancestry only for active/request-bearing sessions, not historical trees.
+      await Promise.all([...owners(ctx)].map((id) => resolveRoot(ctx, id)));
+      if (current(ctx)) publish(ctx);
+    })().catch(() => {
+      if (current(ctx)) ctx.retryAt = Date.now() + 500;
+    }).finally(() => {
+      ctx.resolving = false;
+      if (ctx.resolveAgain) {
+        ctx.resolveAgain = false;
+        reconcile(ctx);
+      }
+    });
+  }
 
-      const reportingSessionID = sessionID;
-      reportPending = true;
-      try {
-        await requestOnce(reportingSessionID);
-      } catch {
-        // Best-effort reporting retries below while the selected route remains active.
-      } finally {
-        reportPending = false;
+  function clearRequests(ctx, matches) {
+    for (const [key, request] of ctx.blockers) if (matches(request)) ctx.blockers.delete(key);
+  }
+
+  function terminalTool(part) {
+    return part?.type === "tool" && ["completed", "error"].includes(part.state?.status);
+  }
+
+  function apply(ctx, event) {
+    const data = event.properties;
+    if (!data) return;
+    if (event.type === "message.part.updated") {
+      const part = data.part;
+      if (terminalTool(part)) {
+        clearRequests(ctx, (r) => r.sessionID === part.sessionID &&
+          r.tool?.messageID === part.messageID && r.tool?.callID === part.callID);
+      }
+      return;
+    }
+    const id = data.sessionID ?? data.info?.id;
+    if (typeof id !== "string" || ctx.deleted.has(id)) return;
+    switch (event.type) {
+      case "session.created":
+      case "session.updated":
+        if (data.info) ctx.sessions.set(id, data.info);
+        break;
+      case "session.deleted":
+        ctx.deleted.add(id);
+        ctx.sessions.delete(id);
+        ctx.statuses.delete(id);
+        ctx.errors.delete(id);
+        clearRequests(ctx, (r) => r.sessionID === id);
+        if (ctx.selected === id) retire(ctx, ctx.route);
+        break;
+      case "session.status":
+      case "session.idle": {
+        const status = event.type === "session.idle" ? "idle" : data.status?.type;
+        if (status === "busy" || status === "retry") {
+          ctx.statuses.set(id, status);
+          ctx.errors.delete(id);
+        } else if (status === "idle") {
+          ctx.statuses.delete(id);
+          ctx.errors.delete(id);
+        }
+        break;
       }
-      if (selectedSessionID !== reportingSessionID) {
-        retryIndex = 0;
-        nextReportAt = 0;
+      case "session.error":
+        if (data.error?.name !== "MessageAbortedError") ctx.errors.add(id);
+        break;
+      case "permission.asked":
+      case "question.asked":
+        if (typeof data.id === "string") ctx.blockers.set(`${event.type.split(".")[0]}:${data.id}`, data);
+        break;
+      case "permission.replied":
+      case "question.replied":
+      case "question.rejected":
+        ctx.blockers.delete(`${event.type.split(".")[0]}:${data.requestID}`);
+        break;
+    }
+  }
+
+  async function hydrate(ctx) {
+    if (!current(ctx) || ctx.loading) return;
+    ctx.loading = true;
+    try {
+      const [statuses, permissions, questions] = await Promise.all([
+        read(ctx, (options) => api.client.session.status(undefined, options)),
+        read(ctx, (options) => api.client.permission.list(undefined, options)),
+        read(ctx, (options) => api.client.question.list(undefined, options)),
+      ]);
+      if (!statuses || typeof statuses !== "object" || Array.isArray(statuses) ||
+          !Object.values(statuses).every((s) => ["busy", "retry", "idle"].includes(s?.type)) ||
+          !Array.isArray(permissions) || !Array.isArray(questions)) {
+        throw new Error("incomplete session snapshot");
+      }
+      const messages = new Map();
+      const blockers = new Map();
+      let validated = true;
+      await Promise.all([["permission", permissions], ["question", questions]].flatMap(([kind, requests]) =>
+        requests.map(async (request) => {
+          if (ctx.deleted.has(request.sessionID)) return;
+          if (request.tool) {
+            const { messageID, callID } = request.tool;
+            const key = `${request.sessionID}:${messageID}`;
+            if (!messages.has(key)) messages.set(key, read(ctx, (options) => api.client.session.message({
+              sessionID: request.sessionID, messageID,
+            }, options)));
+            try {
+              const message = await messages.get(key);
+              const part = message.parts?.find((p) => p.type === "tool" && p.callID === callID);
+              // OpenCode can retain pending requests after abort. Status alone is
+              // insufficient: the same session may already be running a new turn.
+              if (terminalTool(part)) return;
+            } catch {
+              validated = false;
+            }
+          }
+          if (!ctx.deleted.has(request.sessionID)) blockers.set(`${kind}:${request.id}`, request);
+        })));
+      if (!current(ctx)) return;
+      ctx.statuses = new Map(Object.entries(statuses)
+        .filter(([id, status]) => !ctx.deleted.has(id) && (status.type === "busy" || status.type === "retry")));
+      ctx.blockers = blockers;
+      // Preserve deltas until hydration succeeds, including retry backoff:
+      // a later snapshot must not revive a replied or cancelled request.
+      for (const event of ctx.events) apply(ctx, event);
+      ctx.hydrated = validated;
+      if (!validated) ctx.retryAt = Date.now() + 500;
+      reconcile(ctx);
+    } catch {
+      if (current(ctx)) ctx.retryAt = Date.now() + 500;
+    } finally {
+      ctx.loading = false;
+      if (ctx.hydrated) ctx.events = [];
+    }
+  }
+
+  function syncSelection(reset = false) {
+    if (disposed) return;
+    const id = routeID();
+    if (reset || id !== context?.route) {
+      if (typeof id !== "string" || !id) {
+        retire(context, id);
         return;
       }
-      const retryDelay = SELECTION_RETRY_DELAYS_MS[retryIndex];
-      retryIndex += 1;
-      nextReportAt = retryDelay === undefined ? Number.POSITIVE_INFINITY : Date.now() + retryDelay;
-    };
+      const boundary = context?.settled ? undefined : context?.selected;
+      context?.controller?.abort();
+      const ctx = {
+        route: id, boundary, controller: new AbortController(), sessions: new Map(), lookups: new Map(),
+        statuses: new Map(), blockers: new Map(), errors: new Set(), deleted: new Set(),
+        events: [], hydrated: false, loading: false, resolving: false,
+        retryAt: Infinity, selectionAt: 0, retryIndex: 0,
+      };
+      context = ctx;
+      reconcile(ctx);
+      void hydrate(ctx);
+    }
+    const ctx = context;
+    if (!ctx) return;
+    if (!ctx.settled && ctx.selected && Date.now() >= ctx.selectionAt) {
+      publish(ctx, true);
+      const delay = SELECTION_RETRY_DELAYS_MS[ctx.retryIndex++];
+      ctx.selectionAt = delay === undefined ? Infinity : Date.now() + delay;
+    }
+    if (Date.now() >= ctx.retryAt) {
+      ctx.retryAt = Infinity;
+      if (ctx.settled) publish(ctx);
+      else {
+        if (!ctx.hydrated) void hydrate(ctx);
+        reconcile(ctx);
+      }
+    }
+  }
 
-    await syncSelectedSession();
-    const routePoll = setInterval(() => void syncSelectedSession(), ROUTE_POLL_INTERVAL_MS);
-    api.lifecycle.onDispose(() => clearInterval(routePoll));
-  },
-};
+  const subscriptions = [
+    "session.created", "session.updated", "session.deleted", "session.status", "session.idle", "session.error",
+    "permission.asked", "permission.replied", "question.asked", "question.replied", "question.rejected",
+    "message.part.updated",
+  ].map((type) => api.event.on(type, (event) => {
+    if (type === "message.part.updated" && !terminalTool(event.properties?.part)) return;
+    syncSelection();
+    const ctx = context;
+    if (!ctx || ctx.settled) return;
+    if (ctx.loading || !ctx.hydrated) ctx.events.push(event);
+    apply(ctx, event);
+    reconcile(ctx);
+  }));
+  for (const type of ["server.connected", "server.instance.disposed", "global.disposed"]) {
+    subscriptions.push(api.event.on(type, () => syncSelection(true)));
+  }
+  const poll = setInterval(syncSelection, ROUTE_POLL_INTERVAL_MS);
+  api.lifecycle.onDispose(() => {
+    disposed = true;
+    context?.controller?.abort();
+    clearInterval(poll);
+    for (const unsubscribe of subscriptions) unsubscribe();
+  });
+  syncSelection();
+}
 
 function setup(api) {
   if (process.env.HERDR_ENV !== "1" || !process.env.HERDR_SOCKET_PATH || !process.env.HERDR_PANE_ID) return;