diff --git a/apps/web/src/pages/workspace/index.tsx b/apps/web/src/pages/workspace/index.tsx index 25169095..c6604870 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, @@ -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 => { @@ -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 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/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."; 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. */