diff --git a/apps/hub/src/config/resume.ts b/apps/hub/src/config/resume.ts index cb2475d1..f11eeee1 100644 --- a/apps/hub/src/config/resume.ts +++ b/apps/hub/src/config/resume.ts @@ -1,4 +1,5 @@ import type { ConversationTextTurn } from "@corbits-builder/schemas/conversation"; +import type { ApprovalDecision } from "@intx/types/signals"; import { IMPORTED_PROJECT_LEAD } from "./import.ts"; import { SPECIALIST_ENDED_LEAD } from "./specialist-ended.ts"; @@ -60,6 +61,18 @@ export function resumeContext(state: { export const resumeMessage = (context: string | undefined, message: string) => context === undefined ? message : context + RESUME_CONTEXT_END + message; +/** The person's approval decision as a resumed coordinator reads it. */ +export function approvalDecisionLine({ + outcome, + message, +}: ApprovalDecision): string { + const decided = outcome === "approved" ? "Approved" : "Rejected"; + // A blank note is no note: the line ends at the decision, not at a dangling colon. + const note = message?.trim(); + if (note === undefined || note === "") return `${decided} a request.`; + return `${decided} a request: ${note}`; +} + /** A person-sent turn without the held resume context, or undefined for a hub-written opening. */ export function withoutResumeContext(text: string): string | undefined { if (text.startsWith(LEGACY_RESUME_OPENING)) return undefined; diff --git a/apps/hub/src/mail/conversation-turns.test.ts b/apps/hub/src/mail/conversation-turns.test.ts new file mode 100644 index 00000000..c8d9ad2a --- /dev/null +++ b/apps/hub/src/mail/conversation-turns.test.ts @@ -0,0 +1,99 @@ +import { afterEach, expect, test } from "bun:test"; +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { $ } from "bun"; +import { readDurableConversationTurns } from "./conversation-turns.ts"; + +const person = "b65b0e2f4c4a444db11b7321dc413964@sunny-harbor-panda.localhost"; +const dirs: string[] = []; + +afterEach(async () => { + await Promise.all( + dirs.splice(0).map((dir) => rm(dir, { recursive: true, force: true })), + ); +}); + +// A run repo whose committed coordinator mirror holds these inbound messages, as Interchange stores them. +async function runRepo(inbound: { from: string; text: string }[]) { + const dir = await mkdtemp(join(tmpdir(), "sb-conversation-turns-")); + dirs.push(dir); + const wal = join(dir, "agent-state", "coordinate", "wal", "0"); + await mkdir(wal, { recursive: true }); + const turns = inbound.map(({ from, text }, at) => ({ + role: "user", + content: [{ type: "text", text: `[From: ${from}]\n\n${text}` }], + timestamp: at + 1, + })); + const metadata = { + pendingOperations: [], + tokenUsage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + thinking: 0, + }, + connectorState: null, + }; + await writeFile( + join(wal, "0.json"), + JSON.stringify({ seq: 0, turns, metadata }), + ); + const env = { + ...process.env, + GIT_AUTHOR_NAME: "test", + GIT_AUTHOR_EMAIL: "test@example.com", + GIT_COMMITTER_NAME: "test", + GIT_COMMITTER_EMAIL: "test@example.com", + }; + await $`git init -q -b main ${dir}`.env(env); + await $`git -C ${dir} add -A`.env(env); + await $`git -C ${dir} -c commit.gpgsign=false commit -q -m mirror`.env(env); + return dir; +} + +const decisions = () => + runRepo([ + { from: person, text: "Sure" }, + { from: "signal@local", text: JSON.stringify({ outcome: "approved" }) }, + { from: "signal@local", text: JSON.stringify({ outcome: "rejected" }) }, + { + from: "signal@local", + text: JSON.stringify({ outcome: "rejected", message: "Keep it free" }), + }, + { + from: "signal@local", + text: JSON.stringify({ outcome: "approved", message: " " }), + }, + ]); + +const personTexts = async ( + dir: string, + approvalDecisions: "note" | "decision", +) => + ( + await readDurableConversationTurns({ + repoStore: { getRepoDir: () => dir }, + address: + "run_04ba732b8686649f77a2797393bafbc5@sunny-harbor-panda.localhost", + approvalDecisions, + }) + ).map(({ role, text }) => ({ role, text })); + +test("page turns keep only the note sent with an approval decision", async () => { + expect(await personTexts(await decisions(), "note")).toEqual([ + { role: "person", text: "Sure" }, + { role: "person", text: "Keep it free" }, + ]); +}); + +test("resume turns read each approval decision as a plain line", async () => { + expect(await personTexts(await decisions(), "decision")).toEqual([ + { role: "person", text: "Sure" }, + { role: "person", text: "Approved a request." }, + { role: "person", text: "Rejected a request." }, + { role: "person", text: "Rejected a request: Keep it free" }, + { role: "person", text: "Approved a request." }, + ]); +}); diff --git a/apps/hub/src/mail/conversation-turns.ts b/apps/hub/src/mail/conversation-turns.ts index 30c1e911..f0ef44dc 100644 --- a/apps/hub/src/mail/conversation-turns.ts +++ b/apps/hub/src/mail/conversation-turns.ts @@ -6,13 +6,18 @@ import { reconstructDurableConversation } from "@intx/workflow-host"; import { deriveWorkflowRunRepoId } from "@intx/workflow-deploy"; import { extractAddrSpec } from "@intx/mime"; import { isRunAddress } from "@intx/types/agent-address"; +import { ApprovalDecision } from "@intx/types/signals"; +import { type } from "arktype"; import { maxConversationTurns, type ConversationTextTurn, type ConversationThread, } from "@corbits-builder/schemas/conversation"; -import { withoutResumeContext } from "../config/resume.ts"; +import { + approvalDecisionLine, + withoutResumeContext, +} from "../config/resume.ts"; import { personTurnText } from "../config/specialist-ended.ts"; import { maxTurnText, @@ -22,6 +27,7 @@ import { withCommittedAgentState } from "./committed-agent-state.ts"; import type { CoordinatorRepoDirSource } from "./coordinator-question.ts"; import { getLogger } from "@intx/log"; import { errorCode, requestErrorMetadata } from "../request-error.ts"; +import { parseJson } from "../lib/json.ts"; const log = getLogger(["corbits-builder", "hub", "conversation-turns"]); @@ -59,8 +65,33 @@ const inboundEnvelope = // thread headers; the mirror's routing line names only its sender. const runSent = (from: string): boolean => isRunAddress(extractAddrSpec(from)); +// Interchange delivers the person's approval decision to the waiting tool from this address, as +// JSON. The page already shows the decision on its approval card; a resumed coordinator has no card. +const APPROVAL_SIGNAL_SENDER = "signal@local"; + +/** + * 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 + * coordinator. + */ +export type ApprovalDecisionTurns = "note" | "decision"; + +function approvalText( + body: string, + decisions: ApprovalDecisionTurns, +): string | undefined { + const decision = ApprovalDecision(parseJson(body)); + if (decision instanceof type.errors) return undefined; + return decisions === "note" + ? decision.message + : approvalDecisionLine(decision); +} + // A missing or corrupt turn list is empty; it does not throw. -function mapMirrorConversationTurns(turns: unknown): ConversationTextTurn[] { +function mapMirrorConversationTurns( + turns: unknown, + decisions: ApprovalDecisionTurns, +): ConversationTextTurn[] { if (!Array.isArray(turns)) return []; const mapped: ConversationTextTurn[] = []; for (const turn of turns) { @@ -77,6 +108,15 @@ function mapMirrorConversationTurns(turns: unknown): ConversationTextTurn[] { } const from = inboundEnvelope.exec(raw)?.[1]; if (from !== undefined && runSent(from)) continue; + if ( + from !== undefined && + extractAddrSpec(from) === APPROVAL_SIGNAL_SENDER + ) { + const text = approvalText(raw.replace(inboundEnvelope, ""), decisions); + if (text !== undefined && text.trim() !== "") + mapped.push({ role: "person", text: clipTurn(text), at }); + continue; + } const text = withoutResumeContext(raw.replace(inboundEnvelope, "")); if (text === undefined || text.trim() === "") continue; mapped.push({ role: "person", text: clipTurn(personTurnText(text)), at }); @@ -95,6 +135,7 @@ function mapMirrorConversationTurns(turns: unknown): ConversationTextTurn[] { export async function readDurableConversationTurns(input: { repoStore: CoordinatorRepoDirSource; address: string; + approvalDecisions: ApprovalDecisionTurns; }): Promise { try { const repoDir = input.repoStore.getRepoDir({ @@ -128,7 +169,8 @@ export async function readDurableConversationTurns(input: { return collected; }, ); - return mapMirrorConversationTurns(turns ?? []); + if (turns === undefined) return []; + return mapMirrorConversationTurns(turns, input.approvalDecisions); } catch (error) { log.warn("Durable conversation unreadable", { ...errorCode(error), @@ -151,7 +193,10 @@ export async function projectRunTurns(input: { runId: string; address: string; }): Promise<{ turns: ConversationTextTurn[]; threads?: ConversationThread[] }> { - const turns = await readDurableConversationTurns(input); + const turns = await readDurableConversationTurns({ + ...input, + approvalDecisions: "note", + }); if (input.mailDirectory === undefined) return { turns }; const mail = await readCoordinatorMailThreads({ ...input, diff --git a/apps/hub/src/projects/coordinator-resume.ts b/apps/hub/src/projects/coordinator-resume.ts index 6cc4ad79..232b0b7e 100644 --- a/apps/hub/src/projects/coordinator-resume.ts +++ b/apps/hub/src/projects/coordinator-resume.ts @@ -77,6 +77,7 @@ export function createCoordinatorResumeContext(deps: { ...(await readDurableConversationTurns({ repoStore: deps.repoStore, address, + approvalDecisions: "decision", })), ); turns.sort((left, right) => left.at - right.at);