diff --git a/apps/hub/src/mail/conversation-routes.ts b/apps/hub/src/mail/conversation-routes.ts index f62fcd55..4a68760d 100644 --- a/apps/hub/src/mail/conversation-routes.ts +++ b/apps/hub/src/mail/conversation-routes.ts @@ -63,7 +63,7 @@ export function createConversationRoutes( operationId: "readProjectConversationEvents", tags: ["Conversation"], description: - "Read the current registered coordinator's committed native runtime events. While no coordinator is registered (one is starting or replacing another), runId is null, events is empty and turns still carry the conversation of the project's earlier coordinator runs. These events are not a rendered conversation transcript by themselves. When present, turns are the person's thread across every coordinator run of the project, in timestamp order: person and coordinator text from the committed durable conversation mirror, plus participant mail from other workflow runs on a retained mail thread that includes a person. threads are the retained native mail threads among workflow runs only, keyed by References root. A still-open input park also carries the coordinator's asking prose when the run's committed durable conversation mirror provides it (body.question, sourced from the coordinator's turn); parks without real prose stay verbatim. Native workflow read authority is required in addition to project read.", + "Read the current registered coordinator's committed native runtime events. While no coordinator is registered (one is starting or replacing another), runId is null, events is empty and turns still carry the conversation of the project's earlier coordinator runs. These events are not a rendered conversation transcript by themselves. When present, turns are the person's thread across every coordinator run of the project, in timestamp order: person and coordinator text from the committed durable conversation mirror, plus participant mail from other workflow runs on a retained mail thread that includes a person. threads are the retained native mail threads among workflow runs only, keyed by References root. A still-open input park also carries the coordinator's asking prose when the run's committed durable conversation mirror provides it (body.question, sourced from the coordinator's turn), with the ask_questions call it waits on since the person's last message (body.question.ask); parks without real prose stay verbatim. Native workflow read authority is required in addition to project read.", security: [{ sessionCookie: [] }, { deviceBearer: [] }], parameters: ["tenantId", "projectId"].map((name) => ({ name, diff --git a/apps/hub/src/mail/conversation-turns.ts b/apps/hub/src/mail/conversation-turns.ts index 2518f27f..555ecb15 100644 --- a/apps/hub/src/mail/conversation-turns.ts +++ b/apps/hub/src/mail/conversation-turns.ts @@ -67,6 +67,13 @@ const runSent = (from: string): boolean => isRunAddress(extractAddrSpec(from)); // JSON. The page already shows the decision on its approval card; a resumed coordinator has no card. const APPROVAL_SIGNAL_SENDER = "signal@local"; +export function isApprovalDecision(turn: Record): boolean { + const raw = textFromContent(turn["content"]); + if (raw === undefined) return false; + const from = inboundEnvelope.exec(raw)?.[1]; + return from !== undefined && extractAddrSpec(from) === APPROVAL_SIGNAL_SENDER; +} + /** * How the person's approval decisions read: "note" keeps only the note they sent with a * decision, for the page; "decision" reads each decision as a plain line, for a resumed @@ -96,7 +103,7 @@ function personText(body: string): string | undefined { } // A missing or corrupt turn list is empty; it does not throw. -function mapMirrorConversationTurns( +export function mapMirrorConversationTurns( turns: unknown, decisions: ApprovalDecisionTurns, ): ConversationTextTurn[] { diff --git a/apps/hub/src/mail/coordinator-question.test.ts b/apps/hub/src/mail/coordinator-question.test.ts new file mode 100644 index 00000000..a0ef8f12 --- /dev/null +++ b/apps/hub/src/mail/coordinator-question.test.ts @@ -0,0 +1,49 @@ +import { expect, test } from "bun:test"; +import type { AskQuestions } from "@corbits-builder/schemas/questions"; +import { lastOpenAsk } from "./coordinator-question.ts"; + +const person = "b65b0e2f4c4a444db11b7321dc413964@sunny-harbor-panda.localhost"; +const run = "run_04ba732b8686649f77a2797393bafbc5@sunny-harbor-panda.localhost"; +const questions: AskQuestions["questions"] = [ + { + id: "audience", + prompt: "Who is this for?", + kind: "single", + options: [{ value: "me", label: "Just me" }], + }, +]; + +const inbound = (from: string, text: string, timestamp: number) => ({ + role: "user", + content: [{ type: "text", text: `[From: ${from}]\n\n${text}` }], + timestamp, +}); +const ask = (id: string, args: unknown, timestamp: number) => ({ + role: "assistant", + content: [{ type: "tool_call", id, name: "ask_questions", arguments: args }], + timestamp, +}); +const asked = [ + inbound(person, "A habit tracker", 1), + ask("call_1", { questions }, 2), +]; +const open = { callId: "call_1", arguments: { questions } }; + +test("the person's next turn answers the ask", () => { + expect( + lastOpenAsk([...asked, inbound(person, "Just me", 3)], undefined), + ).toBeUndefined(); +}); + +test("a workflow run's mail leaves the ask open", () => { + expect( + lastOpenAsk([...asked, inbound(run, "The brief is filed.", 3)], undefined), + ).toEqual(open); +}); + +test("an approval decision's note leaves the ask open", () => { + const declined = JSON.stringify({ outcome: "rejected", message: "Not yet" }); + expect( + lastOpenAsk([...asked, inbound("signal@local", declined, 3)], undefined), + ).toEqual(open); +}); diff --git a/apps/hub/src/mail/coordinator-question.ts b/apps/hub/src/mail/coordinator-question.ts index 85797dcf..bda150f2 100644 --- a/apps/hub/src/mail/coordinator-question.ts +++ b/apps/hub/src/mail/coordinator-question.ts @@ -4,9 +4,19 @@ import { join } from "node:path"; import { getLogger } from "@intx/log"; import { reconstructDurableConversation } from "@intx/workflow-host"; import { deriveWorkflowRunRepoId } from "@intx/workflow-deploy"; +import { + AskQuestions, + askQuestionsIssue, + type OpenAsk, +} from "@corbits-builder/schemas/questions"; import { errorCode, requestErrorMetadata } from "../request-error.ts"; import { withCommittedAgentState } from "./committed-agent-state.ts"; +import { + isApprovalDecision, + isRecord, + mapMirrorConversationTurns, +} from "./conversation-turns.ts"; const log = getLogger(["corbits-builder", "hub", "coordinator-question"]); @@ -23,8 +33,16 @@ export type CoordinatorRepoDirSource = { type CoordinatorQuestionProse = Readonly<{ prose: string; source: Readonly<{ kind: "coordinator-turn"; agentKey: string }>; + ask?: OpenAsk; }>; +type Asking = { + prose: string; + agentKey: string; + timestamp: number; + ask?: OpenAsk | undefined; +}; + export function createCoordinatorQuestionProseReader(options: { repoStore: CoordinatorRepoDirSource; }) { @@ -43,18 +61,14 @@ export function createCoordinatorQuestionProseReader(options: { typeof parkedAt === "string" && Number.isFinite(Date.parse(parkedAt)) ? Date.parse(parkedAt) : undefined; - let best: - | { prose: string; agentKey: string; timestamp: number } - | undefined; + let best: Asking | undefined; try { best = await withCommittedAgentState(repoDir, async (agentStateRoot) => { const entries = await readdir(agentStateRoot, { withFileTypes: true }); const keys = entries .filter((entry) => entry.isDirectory()) .map((entry) => entry.name); - let found: - | { prose: string; agentKey: string; timestamp: number } - | undefined; + let found: Asking | undefined; for (const agentKey of keys) { let turns: unknown[]; try { @@ -88,6 +102,8 @@ export function createCoordinatorQuestionProseReader(options: { timestamp: candidate.timestamp, }; } + if (found?.agentKey === agentKey) + found.ask = lastOpenAsk(turns, deadline); } return found; }); @@ -102,6 +118,7 @@ export function createCoordinatorQuestionProseReader(options: { return { prose: best.prose, source: { kind: "coordinator-turn", agentKey: best.agentKey }, + ...(best.ask === undefined ? {} : { ask: best.ask }), }; }; } @@ -130,3 +147,49 @@ function assistantText( if (text.trim() === "") return undefined; return { text, timestamp: record.timestamp }; } + +/** + * The last ask_questions call the tool accepted after the person's last message; that message answers + * it. An approval decision answers its own card, not the questions. + */ +export function lastOpenAsk( + turns: unknown[], + deadline: number | undefined, +): OpenAsk | undefined { + const asked = turns.filter( + (turn): turn is Record & { timestamp: number } => + isRecord(turn) && + typeof turn["timestamp"] === "number" && + (deadline === undefined || turn["timestamp"] <= deadline), + ); + const answer = mapMirrorConversationTurns( + asked.filter((turn) => !isApprovalDecision(turn)), + "note", + ).findLast((turn) => turn.role === "person"); + let found: OpenAsk | undefined; + for (const turn of asked) { + if (answer !== undefined && turn.timestamp <= answer.at) continue; + found = acceptedAsk(turn) ?? found; + } + return found; +} + +/** The turn's last ask_questions call that passes the same checks the tool makes. */ +function acceptedAsk(turn: Record): OpenAsk | undefined { + if (turn["role"] !== "assistant" || !Array.isArray(turn["content"])) + return undefined; + let found: OpenAsk | undefined; + for (const block of turn["content"]) { + if ( + !isRecord(block) || + block["type"] !== "tool_call" || + block["name"] !== "ask_questions" || + typeof block["id"] !== "string" + ) + continue; + const input = block["arguments"]; + if (AskQuestions.allows(input) && askQuestionsIssue(input) === undefined) + found = { callId: block["id"], arguments: input }; + } + return found; +} diff --git a/apps/web/src/api/conversation.ts b/apps/web/src/api/conversation.ts index b3f1f7f4..182f85f7 100644 --- a/apps/web/src/api/conversation.ts +++ b/apps/web/src/api/conversation.ts @@ -9,6 +9,8 @@ * and its releasing signal) from the events. */ import type { ConversationEvents } from "@corbits-builder/schemas/conversation"; +import { OpenAsk } from "@corbits-builder/schemas/questions"; +import { type } from "arktype"; import { wording } from "../config.ts"; import { productClient } from "./transport.ts"; import { @@ -171,3 +173,11 @@ export function latestQuestion( ...(correlationId === undefined ? {} : { correlationId }), }; } + +const AskingQuestion = type({ ask: OpenAsk }); + +/** The questions the coordinator waits on, which the hub attaches only to a still-open input park. */ +export function openAsk(events: ConversationEvent[]): OpenAsk | undefined { + const question = events.findLast(isInputPark)?.body["question"]; + return AskingQuestion.allows(question) ? question.ask : undefined; +} diff --git a/apps/web/src/pages/workspace/index.tsx b/apps/web/src/pages/workspace/index.tsx index 05210787..bedf6aa9 100644 --- a/apps/web/src/pages/workspace/index.tsx +++ b/apps/web/src/pages/workspace/index.tsx @@ -63,7 +63,7 @@ import "./persistent-workspace.css"; import { ProductApiError, type ProjectArtifact } from "@corbits-builder/client"; import type { InFlightRun } from "@corbits-builder/schemas/artifact"; import { toast } from "sonner"; -import { sendMessage } from "../../api/conversation.ts"; +import { openAsk, sendMessage } from "../../api/conversation.ts"; import { subscribeWorkflowRun } from "../../api/run-stream.ts"; import { refusalBody } from "../../api/transport.ts"; import { @@ -82,7 +82,13 @@ import type { ChatMessage, Wait } from "../../api/types.ts"; import { deliveredText } from "@corbits-builder/schemas/conversation"; import { keys } from "../../queries/keys.ts"; import { COMPOSER_BOX_CLASS, CONV_CLASS, PANES_CLASS } from "./pane-classes.ts"; -import { questionsOpen, RoomThread, turnItemId } from "./transcript.tsx"; +import { + questionsOpen, + RoomThread, + toolsOf, + turnItemId, +} from "./transcript.tsx"; +import { ASK_QUESTIONS } from "./questions-card.tsx"; import { cardDocuments } from "./collaboration.ts"; import { useFiledVersions } from "./filed-versions.tsx"; import { threadUpdates } from "./specialist-updates.tsx"; @@ -668,6 +674,35 @@ function ReadyRoom({ for (const [at, parts] of committedSteps(liveRun, mirror)) stepsByAt.current.set(at, parts); + // A session that did not watch the coordinator ask its waiting questions draws their card from the + // hub's copy, on the coordinator's latest reply. A reply before the person's last turn did not ask + // them, and the card would read as answered. + const ask = events === null ? undefined : openAsk(events); + const lastTurn = mirror.findLast((turn) => turn.role !== "participant"); + const askedAt = lastTurn?.role === "assistant" ? lastTurn.at : null; + const liveTurn = pendingLiveTurn(liveRun, mirror); + const drawn = [...stepsByAt.current.values()]; + if (liveTurn !== null) drawn.push(liveTurn.parts); + const shown = + ask !== undefined && + drawn.some((parts) => + toolsOf(parts).some((tool) => tool.toolCallId === ask.callId), + ); + const restored = ask !== undefined && askedAt !== null && !shown; + if (restored) { + const card: ConversationPart = { + kind: "tool-trace", + toolCallId: ask.callId, + name: ASK_QUESTIONS, + status: "output-available", + input: ask.arguments, + }; + const steps = stepsByAt.current.get(askedAt); + stepsByAt.current.set( + askedAt, + steps === undefined ? [card] : [...steps, card], + ); + } const turns = useMemo( () => withPendingSends(mirror, sent).map((turn) => { @@ -683,8 +718,9 @@ function ReadyRoom({ parts: [...steps, { kind: "text" as const, text: turn.text }], }; }), - // The steps map only grows as the live run commits, which re-renders with a new mirror. - [mirror, sent], + // The steps map only grows as the live run commits, which re-renders with a new mirror, or as + // a waiting ask's card is restored. + [mirror, sent, restored], ); const mailThreads = snapshot.conversation.threads; const specialistUpdates = useMemo( @@ -719,7 +755,6 @@ function ReadyRoom({ // Until the open decisions are read, no approval is known to hold the coordinator. const parked = waits !== undefined && parkedOnApproval(waits, openApprovalIds, runId); - const liveTurn = pendingLiveTurn(liveRun, mirror); const streaming = liveTurnStreaming(liveRun, mirror, parked); const liveText = liveTurn?.text ?? ""; const shownTurns = diff --git a/apps/web/src/pages/workspace/transcript.tsx b/apps/web/src/pages/workspace/transcript.tsx index 856519db..d932df8e 100644 --- a/apps/web/src/pages/workspace/transcript.tsx +++ b/apps/web/src/pages/workspace/transcript.tsx @@ -353,7 +353,9 @@ type Section = cards: Card[]; }; -function toolsOf(parts: readonly ConversationPart[]): readonly PartTool[] { +export function toolsOf( + parts: readonly ConversationPart[], +): readonly PartTool[] { return parts.flatMap((part) => part.kind === "tool-trace" ? [part] diff --git a/packages/schemas/src/questions.ts b/packages/schemas/src/questions.ts index 95c4c2b8..94e3addf 100644 --- a/packages/schemas/src/questions.ts +++ b/packages/schemas/src/questions.ts @@ -23,6 +23,15 @@ export const AskQuestions = type({ }); export type AskQuestions = typeof AskQuestions.infer; +// The ask_questions call the coordinator waits on, which the conversation read attaches to its open +// input park so a page opened later can still show the card. +export const OpenAsk = type({ + "+": "reject", + callId: "string > 0", + arguments: AskQuestions, +}); +export type OpenAsk = typeof OpenAsk.infer; + export const duplicate = (values: readonly string[]) => values.find((value, index) => values.indexOf(value) !== index);