From 66b680b431e7ffba95c274e4893f94762c9cb76f Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Tue, 3 Nov 2026 17:56:06 -0800 Subject: [PATCH 1/2] fix(web): subscribe to a project's runs under the project's tenant Each project is its own tenant beneath the workspace, and its runs live there. The project page subscribed to the coordinator run's live stream under the workspace's tenant, so the hub found no run and answered 404 on every attempt: no reply streamed in, and the page never saw a turn end. The page subscribes under the tenant of the project it already reads for its lifecycle. --- apps/web/src/pages/workspace/index.tsx | 76 ++++++++++---------------- apps/web/src/queries/keys.ts | 7 ++- 2 files changed, 34 insertions(+), 49 deletions(-) diff --git a/apps/web/src/pages/workspace/index.tsx b/apps/web/src/pages/workspace/index.tsx index 25169095..e1a5db3a 100644 --- a/apps/web/src/pages/workspace/index.tsx +++ b/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, @@ -308,26 +307,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 +340,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 +481,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. @@ -1013,7 +997,7 @@ export function StageWorkspace({ setRestoreFault(null); try { await restoreProject(projectId); - await lifecycle.refetch(); + await projectRead.refetch(); onChanged(); } catch (cause) { setRestoreFault( @@ -1228,8 +1212,8 @@ export function StageWorkspace({ paneDeliverables.find((item) => item.kind === shownId) ?? null; const selectedState = selected === null ? undefined : documentState(selected); - if (lifecycle.isError && status === undefined) - return void lifecycle.refetch()} />; + if (projectRead.isError && status === undefined) + return void projectRead.refetch()} />; if (!roomSnapshotReady(projectId, snapshot)) { if (room.isError) return void room.refetch()} />; return ; diff --git a/apps/web/src/queries/keys.ts b/apps/web/src/queries/keys.ts index 2b2b4b48..2e271785 100644 --- a/apps/web/src/queries/keys.ts +++ b/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. */ From 67681a46fca5e1566f9bcea59c4f971cc0a67d2c Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Tue, 3 Nov 2026 21:02:24 -0800 Subject: [PATCH 2/2] fix(web): stop showing a turn as working while it waits on an approval MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A run held on the person's approval keeps its message run open, and with the live stream connected the page kept the coordinator's turn marked as streaming: its header read "Working…" above the approval card. A live turn held on an approval the coordinator raised is not streaming, so its header shows the finished summary. --- apps/web/src/pages/workspace/index.tsx | 48 ++++++++++--------- apps/web/src/pages/workspace/live-run.ts | 38 ++++++++++++--- .../src/pages/workspace/transcript.test.tsx | 32 +++++++++++++ 3 files changed, 90 insertions(+), 28 deletions(-) diff --git a/apps/web/src/pages/workspace/index.tsx b/apps/web/src/pages/workspace/index.tsx index e1a5db3a..c6604870 100644 --- a/apps/web/src/pages/workspace/index.tsx +++ b/apps/web/src/pages/workspace/index.tsx @@ -94,6 +94,8 @@ import { foldLiveRun, latestAssistantAt, pendingLiveTurn, + liveTurnStreaming, + parkedOnApproval, waitingOnPerson, type LiveRun, } from "./live-run.ts"; @@ -589,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 @@ -781,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 => { @@ -1007,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 = diff --git a/apps/web/src/pages/workspace/live-run.ts b/apps/web/src/pages/workspace/live-run.ts index ae0cc6b3..2189c51c 100644 --- a/apps/web/src/pages/workspace/live-run.ts +++ b/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[], + openApprovals: ReadonlySet, + 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, + ) ); } diff --git a/apps/web/src/pages/workspace/transcript.test.tsx b/apps/web/src/pages/workspace/transcript.test.tsx index d62c89d2..217837d7 100644 --- a/apps/web/src/pages/workspace/transcript.test.tsx +++ b/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.";