fix(web): subscribe to a project's runs under the project's tenant

apps/web/src/pages/workspace/index.tsx

@@ -56,7 +56,6 @@ import type { InFlightRun } from "@corbits-builder/schemas/artifact";
 import { toast } from "sonner";
 import { sendMessage, type ConversationTurn } from "../../api/conversation.ts";
 import { subscribeWorkflowRun } from "../../api/run-stream.ts";
-import { workspaceTenantId } from "../../api/auth.ts";
 import {
   acceptedMaterials,
   attachMaterials,
@@ -95,6 +94,8 @@ import {
   foldLiveRun,
   latestAssistantAt,
   pendingLiveTurn,
+  liveTurnStreaming,
+  parkedOnApproval,
   waitingOnPerson,
   type LiveRun,
 } from "./live-run.ts";
@@ -308,26 +309,25 @@ export function StageWorkspace({
   // for a read that has not happened.
   const queryClient = useQueryClient();
   // An archived project has no coordinator or conversation to read, so the room waits on its lifecycle.
-  const lifecycle = useQuery({
-    queryKey: keys.projectLifecycle.of(projectId),
+  const projectRead = useQuery({
+    queryKey: keys.project.of(projectId),
     retry: false,
     refetchOnMount: "always",
     refetchOnWindowFocus: false,
-    queryFn: async () => (await getProject(projectId)).status,
+    queryFn: () => getProject(projectId),
   });
   // A status restored from the saved cache may predate an archive, so only a read made since
   // opening counts. The project list learns of an archive first, so it wins while the read catches up.
-  const status = listedArchived
-    ? "archived"
-    : lifecycle.isFetchedAfterMount
-      ? lifecycle.data
-      : undefined;
+  const project = projectRead.isFetchedAfterMount
+    ? projectRead.data
+    : undefined;
+  const status = listedArchived ? "archived" : project?.status;
   const archived = status === "archived";
   const listedBefore = useRef(listedArchived);
   useEffect(() => {
     if (listedBefore.current === listedArchived) return;
     listedBefore.current = listedArchived;
-    void lifecycle.refetch();
+    void projectRead.refetch();
   }, [listedArchived]);
   const roomKey = keys.projectRoom.of(projectId);
   const room = useQuery({
@@ -342,6 +342,8 @@ export function StageWorkspace({
   const events = snapshot?.conversation.events ?? null;
   const mirror = snapshot?.conversation.turns ?? [];
   const runId = archived ? null : (snapshot?.conversation.runId ?? null);
+  // Each project is its own tenant beneath the workspace, and its runs live there.
+  const runTenantId = project?.tenantId;
   // The coder's filed app source is the resume point for a later coder, not a document to read.
   const artifacts = (snapshot?.artifacts ?? [])
     .filter((item) => item.kind !== "app")
@@ -481,40 +483,24 @@ export function StageWorkspace({
   useEffect(() => {
     liveRunRef.current = emptyLiveRun;
     setLiveRun(emptyLiveRun);
-    if (runId === null) return;
-    let unsubscribe = () => {};
-    let cancelled = false;
+    if (runId === null || runTenantId === undefined) return;
     const reload = () => reloadRef.current();
-    void workspaceTenantId()
-      .then((tenantId) => {
-        if (cancelled) return;
-        unsubscribe = subscribeWorkflowRun(tenantId, runId, {
-          onEvent: (event) => {
-            const next = foldLiveRun(
-              liveRunRef.current,
-              event,
-              committedAtRef.current,
-            );
-            liveRunRef.current = next.run;
-            setLiveRun(next.run);
-            if (next.reload) reload();
-          },
-          onChange: reload,
-          onReconnect: reload,
-          onError: reload,
-        }).unsubscribe;
-      })
-      .catch((cause: unknown) => {
-        if (!cancelled)
-          toast.error(humanFault(cause, "Live updates are not available."), {
-            id: "live-subscription",
-          });
-      });
-    return () => {
-      cancelled = true;
-      unsubscribe();
-    };
-  }, [projectId, runId]);
+    return subscribeWorkflowRun(runTenantId, runId, {
+      onEvent: (event) => {
+        const next = foldLiveRun(
+          liveRunRef.current,
+          event,
+          committedAtRef.current,
+        );
+        liveRunRef.current = next.run;
+        setLiveRun(next.run);
+        if (next.reload) reload();
+      },
+      onChange: reload,
+      onReconnect: reload,
+      onError: reload,
+    }).unsubscribe;
+  }, [projectId, runId, runTenantId]);
 
   // The runs held on an approval, from the polled run states: a coder's package request raises no
   // room event, so the room alone would not say the person's turn came until something else moved.
@@ -605,8 +591,30 @@ export function StageWorkspace({
   );
   const latestAssistant =
     [...mirror].reverse().find((turn) => turn.role === "assistant")?.text ?? "";
+  const documentApprovals = documentApprovalsFrom(waits, artifacts);
+  const { settled: settledApprovals, settle } = useSettledApprovals(
+    projectId,
+    documentApprovals,
+    artifacts,
+  );
+  // A card decided here is already its settled line; the pending read lags behind it.
+  const pendingFrom = (list: readonly Wait[]): OpenApproval[] =>
+    documentApprovalsFrom(list, artifacts).filter(
+      (approval) =>
+        !settledApprovals.some(
+          (item) => item.approvalId === approval.approvalId,
+        ),
+    );
+  const pendingApprovals = pendingFrom(waits);
+  const openApprovalIds = new Set(
+    pendingApprovals.map((approval) => approval.approvalId),
+  );
   const liveTurn = pendingLiveTurn(liveRun, mirror);
-  const streaming = liveTurn !== null && !liveRun.ended;
+  const streaming = liveTurnStreaming(
+    liveRun,
+    mirror,
+    parkedOnApproval(waits, openApprovalIds, runId),
+  );
   const liveText = liveTurn?.text ?? "";
   const shownTurns =
     liveTurn === null
@@ -797,21 +805,6 @@ export function StageWorkspace({
   const addToPrompt = (passage: PassageQuote) =>
     addToComposer(formatPassageQuote(passage));
 
-  const documentApprovals = documentApprovalsFrom(waits, artifacts);
-  const { settled: settledApprovals, settle } = useSettledApprovals(
-    projectId,
-    documentApprovals,
-    artifacts,
-  );
-  // A card decided here is already its settled line; the pending read lags behind it.
-  const pendingFrom = (list: readonly Wait[]): OpenApproval[] =>
-    documentApprovalsFrom(list, artifacts).filter(
-      (approval) =>
-        !settledApprovals.some(
-          (item) => item.approvalId === approval.approvalId,
-        ),
-    );
-  const pendingApprovals = pendingFrom(waits);
   // A park created moments after `inference.done` can already stand on the hub before the next
   // room reload lands; read once on demand when the cached list shows no open ask.
   const readPendingApprovals = async (): Promise<readonly OpenApproval[]> => {
@@ -1013,7 +1006,7 @@ export function StageWorkspace({
     setRestoreFault(null);
     try {
       await restoreProject(projectId);
-      await lifecycle.refetch();
+      await projectRead.refetch();
       onChanged();
     } catch (cause) {
       setRestoreFault(
@@ -1023,12 +1016,7 @@ export function StageWorkspace({
       setRestoring(false);
     }
   };
-  const held = waitingOnPerson(
-    waits,
-    new Set(pendingApprovals.map((approval) => approval.approvalId)),
-    runId,
-    liveRun.running,
-  );
+  const held = waitingOnPerson(waits, openApprovalIds, runId, liveRun.running);
   // A message sent here waits on the coordinator from the moment it leaves, before the stream
   // reports its run: until a reply follows it or a run ends after it.
   const awaitingReply =
@@ -1228,8 +1216,8 @@ export function StageWorkspace({
     paneDeliverables.find((item) => item.kind === shownId) ?? null;
   const selectedState = selected === null ? undefined : documentState(selected);
 
-  if (lifecycle.isError && status === undefined)
-    return <RoomFault onRetry={() => void lifecycle.refetch()} />;
+  if (projectRead.isError && status === undefined)
+    return <RoomFault onRetry={() => void projectRead.refetch()} />;
   if (!roomSnapshotReady(projectId, snapshot)) {
     if (room.isError) return <RoomFault onRetry={() => void room.refetch()} />;
     return <RoomSkeleton />;

apps/web/src/pages/workspace/live-run.ts

@@ -252,6 +252,29 @@ export function awaitingCommit(
   return !hasContent(run.current) && pending.parts.length > 0;
 }
 
+/** An approval the coordinator's own run raised is still open: the run waits on the person. */
+export function parkedOnApproval(
+  waits: readonly Pick<Wait, "runId" | "approvalId">[],
+  openApprovals: ReadonlySet<string>,
+  coordinatorRunId: string | null,
+): boolean {
+  return waits.some(
+    (wait) =>
+      wait.runId === coordinatorRunId &&
+      wait.approvalId !== undefined &&
+      openApprovals.has(wait.approvalId),
+  );
+}
+
+/** A run held on an approval keeps its message run open with nothing being written. */
+export function liveTurnStreaming(
+  run: LiveRun,
+  mirror: readonly ConversationTurn[],
+  heldOnApproval: boolean,
+): boolean {
+  return pendingLiveTurn(run, mirror) !== null && !run.ended && !heldOnApproval;
+}
+
 /**
  * The coordinator is held on the person: an approval its own run raised is still open, or its
  * question stands and no message run has started since. A held tool call keeps the native message
@@ -263,11 +286,14 @@ export function waitingOnPerson(
   coordinatorRunId: string | null,
   running: boolean,
 ): boolean {
-  return waits.some(
-    (wait) =>
-      wait.runId === coordinatorRunId &&
-      (wait.approvalId === undefined
-        ? wait.park !== undefined && !running
-        : openApprovals.has(wait.approvalId)),
+  return (
+    parkedOnApproval(waits, openApprovals, coordinatorRunId) ||
+    waits.some(
+      (wait) =>
+        wait.runId === coordinatorRunId &&
+        wait.approvalId === undefined &&
+        wait.park !== undefined &&
+        !running,
+    )
   );
 }

apps/web/src/pages/workspace/transcript.test.tsx

@@ -7,6 +7,8 @@ import {
   awaitingCommit,
   emptyLiveRun,
   foldLiveRun,
+  liveTurnStreaming,
+  parkedOnApproval,
   waitingOnPerson,
 } from "./live-run.ts";
 import { RoomThread } from "./transcript.tsx";
@@ -188,6 +190,36 @@ describe("conversation transcript", () => {
     expect(awaitingCommit(done, [])).toBe(true);
   });
 
+  test("a turn held on the person's approval is no longer being written", () => {
+    const held = [
+      { type: "message.run.started" },
+      { type: "inference.start" },
+      {
+        type: "inference.tool_call.end",
+        data: {
+          callId: "call-1",
+          name: "request_approval",
+          arguments: { kinds: ["brief"] },
+        },
+      },
+      { type: "inference.done" },
+    ].reduce((run, event) => foldLiveRun(run, event, null).run, emptyLiveRun);
+    const waits = [
+      { runId: "run-c", approvalId: "appr-1" },
+      { runId: "run-coder", approvalId: "appr-2" },
+    ];
+    const streaming = (open: string[]) =>
+      liveTurnStreaming(
+        held,
+        [],
+        parkedOnApproval(waits, new Set(open), "run-c"),
+      );
+    expect(held.ended).toBe(false);
+    expect(streaming(["appr-1"])).toBe(false);
+    expect(streaming(["appr-2"])).toBe(true);
+    expect(streaming([])).toBe(true);
+  });
+
   test("a specialist's report joins its hand-off as one message and a long run of lone ones folds", () => {
     const summary =
       "Wrote the limits for your weather app, covering offline use and units.";

apps/web/src/queries/keys.ts

@@ -65,9 +65,10 @@ export const keys = {
     all: ["projectRoom"] as const,
     of: (projectId: string) => ["projectRoom", projectId] as const,
   },
-  projectLifecycle: {
-    all: ["projectLifecycle"] as const,
-    of: (projectId: string) => ["projectLifecycle", projectId] as const,
+  /** A project: its lifecycle status and the tenant its runs live under. */
+  project: {
+    all: ["project"] as const,
+    of: (projectId: string) => ["project", projectId] as const,
   },
   modelCatalog: ["modelCatalog"] as const,
   /** The person's own model choice. */