From bce7ce59c2781a05062989c110802998f07fb566 Mon Sep 17 00:00:00 2001 From: milind-soni Date: Sat, 29 Aug 2026 18:56:51 +0530 Subject: [PATCH] feat(routines): report runs in source conversations --- server/index.test.ts | 151 +++++++++++++++++++++++++- server/index.ts | 138 +++++++++++++++++++++-- server/routine-requests.test.ts | 7 +- server/routines.test.ts | 143 +++++++++++++++++++++++- server/routines.ts | 68 +++++++++++- server/store.test.ts | 17 +++ server/store.ts | 13 ++- shared/routine-run.ts | 16 +++ src/components/ChatView.tsx | 18 +++ src/components/GroupView.tsx | 16 +++ src/components/RoutineRunCard.test.ts | 98 +++++++++++++++++ src/components/RoutineRunCard.tsx | 134 +++++++++++++++++++++++ src/lib/notify.test.ts | 20 +++- src/lib/notify.ts | 9 +- src/lib/routines.ts | 2 + src/lib/taskTimeline.ts | 2 +- src/state/store.test.ts | 22 ++++ src/state/store.tsx | 20 +++- 18 files changed, 866 insertions(+), 28 deletions(-) create mode 100644 shared/routine-run.ts create mode 100644 src/components/RoutineRunCard.test.ts create mode 100644 src/components/RoutineRunCard.tsx diff --git a/server/index.test.ts b/server/index.test.ts index cb75a4f79..ed63ea58c 100644 --- a/server/index.test.ts +++ b/server/index.test.ts @@ -9,6 +9,7 @@ import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, statSync, wri import { tmpdir } from "node:os"; import { dirname, join } from "node:path"; import { fileURLToPath } from "node:url"; +import { DatabaseSync } from "node:sqlite"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { z } from "zod"; @@ -42,6 +43,18 @@ const api = async (method: string, path: string, body?: unknown): Promise<{ stat return { status: res.status, body: await res.json() }; }; +const storedMessageCount = (threadId: string): number => { + const db = new DatabaseSync(join(home, ".openmausbot", "messages.db"), { readOnly: true }); + try { + const row = z.object({ count: z.number() }).parse( + db.prepare("SELECT COUNT(*) AS count FROM messages WHERE thread_id = ?").get(threadId), + ); + return row.count; + } finally { + db.close(); + } +}; + const uploadAvatar = async (mime = "image/png"): Promise => { const response = await fetch(`${BASE}/api/attachments`, { method: "POST", @@ -2560,6 +2573,7 @@ describe("harness HTTP API", () => { it("keeps chat-created routines inert until their durable card is confirmed", async () => { const bot = (await api("POST", "/api/bots", {})).body.bot; let routineId = ""; + let orphanRoutineId = ""; let legacyRoutineId = ""; try { const selected = await api("PATCH", `/api/bots/${bot.id}`, { @@ -2655,7 +2669,11 @@ describe("harness HTTP API", () => { }).toEqual(["card-shown:routine", "user-approved:user"]); const after = await api("GET", "/api/routines"); - expect(after.body.routines.filter((routine: { botId: string }) => routine.botId === bot.id)).toHaveLength(1); + const confirmedRoutine = after.body.routines.find((routine: { id: string }) => routine.id === routineId); + expect(confirmedRoutine).toMatchObject({ + botId: bot.id, + sourceThreadId: bot.threadId, + }); const duplicate = await api("POST", `/api/threads/${bot.threadId}/respond`, { requestId: proposal.requestId, behavior: "allow", @@ -2664,6 +2682,136 @@ describe("harness HTTP API", () => { expect((await api("GET", "/api/routines")).body.routines .filter((routine: { botId: string }) => routine.botId === bot.id)).toHaveLength(1); + // The initial fixture turn is deliberately hung. Once it is stopped, + // force a deterministic dispatch failure by choosing the configured + // but unavailable ghost provider. The execution stays detached, while one source card is + // appended then patched through queued → running → failed. + expect((await api("POST", `/api/bots/${bot.id}/interrupt`)).status).toBe(200); + await expect.poll(async () => { + const current = (await api("GET", "/api/bots?messages=0")).body.bots + .find((candidate: { id: string }) => candidate.id === bot.id); + return Boolean(current?.busy); + }, { timeout: 5_000 }).toBe(false); + expect((await api("PATCH", `/api/bots/${bot.id}`, { + modelSelection: { instanceId: "ghost", model: "unavailable-fixture" }, + })).status).toBe(200); + + const routineEvents = await openSse(`${BASE}/api/events`); + try { + const queued = await api("POST", `/api/routines/${routineId}/run`); + expect(queued.status).toBe(201); + const failedNotice = await routineEvents.until( + (frame) => + frame.kind === "notify" && + frame.notification?.kind === "routine-failed" && + frame.notification?.botId === bot.id, + 5_000, + ); + expect(failedNotice.notification.threadId).toBe(bot.threadId); + + await expect.poll(async () => { + const current = (await api("GET", "/api/bots")).body.bots + .find((candidate: { id: string }) => candidate.id === bot.id); + return current?.messages.filter( + (message: { kind?: string; routineRun?: { runId?: string } }) => + message.kind === "routine.run" && message.routineRun?.runId === queued.body.run.id, + ) ?? []; + }, { timeout: 5_000 }).toHaveLength(1); + const current = (await api("GET", "/api/bots")).body.bots + .find((candidate: { id: string }) => candidate.id === bot.id); + const runCards = current.messages.filter( + (message: { kind?: string; routineRun?: { runId?: string } }) => + message.kind === "routine.run" && message.routineRun?.runId === queued.body.run.id, + ); + expect(runCards).toHaveLength(1); + expect(runCards[0].routineRun).toMatchObject({ + runId: queued.body.run.id, + routineId, + routineName: "Weekday brief", + status: "failed", + }); + expect(runCards[0].routineRun.executionThreadId).not.toBe(bot.threadId); + + // Reading the source and then marking the failure seen in Routines + // must not make the original conversation unread again. markSeen + // re-emits the receipt without changing its lifecycle status. + expect((await api("POST", `/api/bots/${bot.id}/read`)).status).toBe(200); + expect((await api("POST", `/api/routine-runs/${queued.body.run.id}/seen`)).status).toBe(200); + const afterSeen = (await api("GET", "/api/bots?messages=0")).body.bots + .find((candidate: { id: string }) => candidate.id === bot.id); + expect(afterSeen.unread).toBe(false); + + const grounded = await fetch( + `${BASE}/api/internal/routines?fromBotId=${encodeURIComponent(bot.id)}&fromThreadId=${encodeURIComponent(bot.threadId)}`, + { headers: internalHeaders }, + ); + const groundedBody = z.object({ + routines: z.array(z.object({ + id: z.string(), + latestRun: z.object({ + status: z.string(), + scheduledFor: z.string().nullable(), + startedAt: z.string().nullable(), + finishedAt: z.string().nullable(), + output: z.string().nullable(), + error: z.string().nullable(), + executionThreadId: z.string().nullable(), + }).nullable(), + }).passthrough()), + }).parse(await grounded.json()); + expect(groundedBody.routines.find((routine) => routine.id === routineId)?.latestRun).toMatchObject({ + status: "failed", + startedAt: expect.any(String), + finishedAt: expect.any(String), + error: expect.stringMatching(/provider instance "ghost" is unavailable/i), + executionThreadId: runCards[0].routineRun.executionThreadId, + }); + } finally { + routineEvents.close(); + } + + // A deleted source conversation is a safe fallback, not an instruction + // to recreate its transcript. The run still gets its detached receipt + // and failure, but no lifecycle message is written to the orphan id. + const orphanSource = await api("POST", `/api/bots/${bot.id}/tasks`, { title: "Temporary routine source" }); + expect(orphanSource.status).toBe(201); + const orphanThreadId = z.object({ + task: z.object({ threadId: z.string() }), + }).parse(orphanSource.body).task.threadId; + const orphanProposalResponse = await fetch(`${BASE}/api/internal/routine-requests`, { + method: "POST", + headers: internalHeaders, + body: JSON.stringify({ + fromBotId: bot.id, + fromThreadId: orphanThreadId, + action: "create", + routine: { + name: "Orphan-safe brief", + instructions: "Summarize without recreating the deleted source.", + schedule: { type: "weekly", time: "09:00", weekdays: ["monday"] }, + runOn: "maus", + }, + }), + }); + expect(orphanProposalResponse.status).toBe(201); + const orphanProposal = z.object({ requestId: z.string() }).parse(await orphanProposalResponse.json()); + const orphanConfirmed = await api("POST", `/api/threads/${orphanThreadId}/respond`, { + requestId: orphanProposal.requestId, + behavior: "allow", + }); + expect(orphanConfirmed.status).toBe(200); + orphanRoutineId = orphanConfirmed.body.resultId; + expect((await api("DELETE", `/api/bots/${bot.id}/tasks/${orphanThreadId}`)).status).toBe(200); + expect(storedMessageCount(orphanThreadId)).toBe(0); + + const orphanRun = await api("POST", `/api/routines/${orphanRoutineId}/run`); + expect(orphanRun.status).toBe(201); + await expect.poll(async () => { + const runs = (await api("GET", "/api/routines")).body.runs; + return runs.find((run: { id: string }) => run.id === orphanRun.body.run.id)?.status; + }, { timeout: 5_000 }).toBe("failed"); + expect(storedMessageCount(orphanThreadId)).toBe(0); + // Calendar-created routines may predate chat-card redaction. Listing // them to a model must redact the whole prompt before returning its // bounded preview, and tell the model when that preview is incomplete. @@ -2709,6 +2857,7 @@ describe("harness HTTP API", () => { expect(wrongThread.status).toBe(403); } finally { if (legacyRoutineId) await api("DELETE", `/api/routines/${legacyRoutineId}`); + if (orphanRoutineId) await api("DELETE", `/api/routines/${orphanRoutineId}`); if (routineId) await api("DELETE", `/api/routines/${routineId}`); await api("POST", `/api/bots/${bot.id}/interrupt`); await api("DELETE", `/api/bots/${bot.id}`); diff --git a/server/index.ts b/server/index.ts index 269fbc61e..8500b2660 100644 --- a/server/index.ts +++ b/server/index.ts @@ -151,7 +151,7 @@ import { LocalVmLease, LocalVmLeasePool } from "./local-vm-lease.ts"; import { RepeatDetector, callKey } from "./repeat-detector.ts"; import { redactSecretsInText } from "./redact.ts"; import * as vps from "./vps-computer.ts"; -import { RoutineManager, type RoutineRunOn, type RoutineRunTrigger } from "./routines.ts"; +import { RoutineManager, type RoutineRun, type RoutineRunOn, type RoutineRunTrigger } from "./routines.ts"; import { RoutineRequestService } from "./routine-requests.ts"; import { fetchBotDirectory, matchDirectoryBots, type MatchedDirectoryBot } from "./bot-directory.ts"; import { scoutProject, suggestTeam } from "./project-scout.ts"; @@ -1279,7 +1279,12 @@ bus.subscribe((event: RuntimeEvent) => { if (!card || card.answered) return; // the bot is not working now — it is waiting on a person if (asker.busy) store.setActivity(asker.id, "waiting-on-you"); - notify(buildNotification(permission ? "approval" : "question", asker, event.threadId, event.summary)); + notify(buildNotification( + permission ? "approval" : "question", + asker, + (routineRun && routineSourceThread(routineRun)) || event.threadId, + event.summary, + )); }; if (reviewTask && reviewMode === "enforce") { // Avoid buzzing the owner for a card the reviewer is about to answer. @@ -1366,11 +1371,18 @@ bus.subscribe((event: RuntimeEvent) => { }); // settled → idle; a setup failure already marked it dead, keep that if (store.bot(bot.id)?.activity !== "dead") store.setActivity(bot.id, "idle"); - store.patchBot(bot.id, { unread: true }); + const routineReportThread = routineRun ? routineSourceThread(routineRun) : null; + const routineReportGroup = routineReportThread ? store.groupByThread(routineReportThread) : undefined; + // Group-origin routines belong to that channel's unread state. Their + // hidden execution task should not light up the bot's 1:1 sidebar too. + if (!routineReportGroup) store.patchBot(bot.id, { unread: true }); if (routineRun?.status !== "failed") { // the frame carries the bot's avatar so every desktop client can // show the notification under that bot's own face - notify(buildNotification("done", bot, event.threadId, reply, { avatarUrl: bot.avatarUrl })); + const completionDetail = routineRun + ? reply || routineRun.output || routineRun.routineName + : reply; + notify(buildNotification("done", bot, routineReportThread ?? event.threadId, completionDetail, { avatarUrl: bot.avatarUrl })); } if (screenPollers.has(bot.id)) { // the last live frame becomes a settled inline screen message — @@ -2156,6 +2168,91 @@ async function startTurn( // ── routines: persisted definitions → detached bot tasks ─────────────── // The scheduler owns timing and receipts; the existing harness remains the // only owner of provider sessions, approvals, tools, computers and messages. +function routineSourceOwner(run: RoutineRun) { + const threadId = run.sourceThreadId?.trim(); + if (!threadId) return null; + // Validate before messagesFor(): Store lazily opens transcript storage, so + // reading an orphan id first would recreate a deleted conversation. + const bot = store.bot(run.botId); + if (!bot) return null; + if (store.taskByThread(bot.id, threadId)) return { bot, group: undefined, threadId }; + const group = store.groupByThread(threadId); + return group?.memberIds.includes(bot.id) ? { bot, group, threadId } : null; +} + +function routineSourceThread(run: RoutineRun): string | null { + return routineSourceOwner(run)?.threadId ?? null; +} + +function routineRunCard(run: RoutineRun): NonNullable { + const visibleSummary = run.status === "waiting" ? run.attention : run.output; + const summary = visibleSummary ? redactSecretsInText(visibleSummary).slice(0, 2_000) : undefined; + const error = run.error ? redactSecretsInText(run.error).slice(0, 500) : undefined; + const card: NonNullable = { + runId: run.id, + routineId: run.routineId, + routineName: redactSecretsInText(run.routineName), + status: run.status, + }; + if (run.threadId) card.executionThreadId = run.threadId; + if (summary) card.summary = summary; + if (error) card.error = error; + return card; +} + +function routineRunFallbackText(card: NonNullable): string { + const state = + card.status === "waiting" + ? "needs your attention" + : card.status === "completed" + ? "completed" + : card.status === "failed" + ? "failed" + : card.status === "cancelled" + ? "was cancelled" + : card.status === "missed" + ? "was missed" + : card.status; + return `Routine “${card.routineName}” ${state}`; +} + +/** Upsert one durable lifecycle card per run. Replaying the same transition, + * including restart recovery, patches the existing run id instead of adding + * another chat message. */ +function syncRoutineRunToSource(run: RoutineRun): string | null { + const source = routineSourceOwner(run); + if (!source) return null; + const sourceThreadId = source.threadId; + const card = routineRunCard(run); + const text = routineRunFallbackText(card); + const existing = store.messagesFor(sourceThreadId).find( + (message) => message.kind === "routine.run" && message.routineRun?.runId === run.id, + ); + const statusChanged = existing?.routineRun?.status !== run.status; + if (existing) { + store.patchMessage(sourceThreadId, existing.id, { text, routineRun: card }); + } else { + const message: Omit = { + role: "bot", + kind: "routine.run", + text, + routineRun: card, + }; + if (source.group) { + message.from = { botId: source.bot.id, name: source.bot.name, color: source.bot.color }; + } + store.appendMessage(sourceThreadId, message); + } + + // Merely queueing/running is ambient progress. Attention and terminal + // states become unread in the conversation where the user asked for them. + if (statusChanged && ["waiting", "completed", "failed", "missed"].includes(run.status)) { + if (source.group) store.patchGroup(source.group.id, { unread: true }); + else store.patchBot(source.bot.id, { unread: true }); + } + return sourceThreadId; +} + routines = new RoutineManager({ emit: broadcast, botState: (botId) => { @@ -2180,11 +2277,12 @@ routines = new RoutineManager({ : null; await instance?.adapter.interruptTurn(threadId); }, + onRunChanged: syncRoutineRunToSource, onRunFailed: (run) => { const bot = store.bot(run.botId); if (!bot) return; const detail = run.error ? `${run.routineName}: ${run.error}` : run.routineName; - notify(buildNotification("routine-failed", bot, run.threadId ?? bot.threadId, detail)); + notify(buildNotification("routine-failed", bot, routineSourceThread(run) ?? run.threadId ?? bot.threadId, detail)); }, }); const recoveryOwners = routines.routineRequestReceiptOwners(); @@ -2240,7 +2338,12 @@ const routineRequests = new RoutineRequestService({ }); const ROUTINE_WEEKDAY_NAMES = ["sunday", "monday", "tuesday", "wednesday", "thursday", "friday", "saturday"] as const; const routineTimeZone = () => Intl.DateTimeFormat().resolvedOptions().timeZone || "UTC"; -const agentRoutine = (routine: ReturnType[number]) => { +const routineTimestamp = (value: number | undefined) => + value !== undefined && Number.isFinite(value) ? new Date(value).toISOString() : null; +const agentRoutine = ( + routine: ReturnType[number], + latestRun?: RoutineRun, +) => { // Routines created in the calendar predate chat-card redaction and may // contain a credential in their instructions. The list result is handed // back to the model, so scrub the complete value before taking its preview. @@ -2262,6 +2365,20 @@ const agentRoutine = (routine: ReturnType[number weekdays: routine.schedule.weekdays.map((day) => ROUTINE_WEEKDAY_NAMES[day]), }, nextRunAt: routine.nextRunAt === null ? null : new Date(routine.nextRunAt).toISOString(), + latestRun: latestRun + ? { + id: latestRun.id, + status: latestRun.status, + triggerSource: latestRun.triggerSource ?? (latestRun.manual ? "manual" : "schedule"), + scheduledFor: routineTimestamp(latestRun.scheduledFor), + startedAt: routineTimestamp(latestRun.startedAt), + finishedAt: routineTimestamp(latestRun.finishedAt), + attention: latestRun.attention ? redactSecretsInText(latestRun.attention).slice(0, 500) : null, + output: latestRun.output ? redactSecretsInText(latestRun.output).slice(0, 1_000) : null, + error: latestRun.error ? redactSecretsInText(latestRun.error).slice(0, 500) : null, + executionThreadId: latestRun.threadId ?? null, + } + : null, }; }; function sendRoutineResolution( @@ -3312,13 +3429,20 @@ const server = createServer(async (req, res) => { if (!connectorThread(from.id, fromThreadId)) { return json(res, 403, { error: "source conversation does not belong to sender" }); } + const latestRuns = new Map(); + // listRuns is newest-first. Keep the first receipt per definition so + // the agent can answer "did it run?" from scheduler truth rather + // than guessing from conversation history. + for (const run of routines!.listRuns()) { + if (run.botId === from.id && !latestRuns.has(run.routineId)) latestRuns.set(run.routineId, run); + } return json(res, 200, { now: new Date().toISOString(), timeZone: routineTimeZone(), routines: routines!.listRoutines() .filter((routine) => routine.botId === from.id) .slice(0, 100) - .map(agentRoutine), + .map((routine) => agentRoutine(routine, latestRuns.get(routine.id))), }); } if (method === "POST" && path === "/api/internal/routine-requests") { diff --git a/server/routine-requests.test.ts b/server/routine-requests.test.ts index 3f972810c..7e4bc3ba2 100644 --- a/server/routine-requests.test.ts +++ b/server/routine-requests.test.ts @@ -454,7 +454,12 @@ describe("RoutineRequestService", () => { }); expect(first).toMatchObject({ claimed: true, state: "applied", action: "create" }); if (first.state !== "applied") throw new Error("Expected the routine to be applied"); - expect(routines.listRoutines()).toMatchObject([{ botId: "bot-a", name: "Morning brief", enabled: true }]); + expect(routines.listRoutines()).toMatchObject([{ + botId: "bot-a", + name: "Morning brief", + enabled: true, + sourceThreadId: "thread-a", + }]); expect(routines.routineRequestReceipt(proposal.requestId)).toBeNull(); expect(store.messagesFor("thread-a")[0]!.card).toMatchObject({ held: undefined, diff --git a/server/routines.test.ts b/server/routines.test.ts index 5b57a278b..9bd6f8ed5 100644 --- a/server/routines.test.ts +++ b/server/routines.test.ts @@ -1,9 +1,14 @@ -import { mkdirSync, mkdtempSync, readFileSync, rmSync } from "node:fs"; +import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { afterEach, describe, expect, it, vi } from "vitest"; -import { nextOccurrence, RoutineManager, type RoutineManagerOptions } from "./routines.ts"; +import { + nextOccurrence, + RoutineManager, + type RoutineManagerOptions, + type RoutineSchedule, +} from "./routines.ts"; const dirs: string[] = []; @@ -22,6 +27,7 @@ function harness(start = new Date(2026, 7, 17, 8, 0, 0).getTime()) { const triggerSources: string[] = []; const taskActivations: boolean[] = []; const emitted: any[] = []; + const changed: any[] = []; const failed: any[] = []; const options: RoutineManagerOptions = { file: tempFile(), @@ -37,6 +43,7 @@ function harness(start = new Date(2026, 7, 17, 8, 0, 0).getTime()) { runOns.push(runOn); triggerSources.push(triggerSource); }, + onRunChanged: (run) => changed.push(run), onRunFailed: (run) => failed.push(run), }; const manager = new RoutineManager(options); @@ -48,6 +55,7 @@ function harness(start = new Date(2026, 7, 17, 8, 0, 0).getTime()) { runOns, triggerSources, taskActivations, + changed, failed, setNow: (value: number) => (now = value), setBot: (value: typeof bot) => (bot = value), @@ -153,6 +161,120 @@ describe("RoutineManager", () => { expect(new RoutineManager(h.options).routineRequestReceipt(request.requestId)).toBeNull(); }); + it("persists trusted chat provenance and snapshots it onto detached runs", async () => { + const h = harness(); + const request = { + requestId: "request-source-thread", + messageId: "message-source-thread", + botId: "maus-1", + threadId: "conversation-that-created-it", + action: "create" as const, + fingerprintVersion: 1 as const, + fingerprint: "e".repeat(64), + }; + const routine = h.manager.create({ + name: "Source report", + prompt: "Summarize the queue", + botId: "maus-1", + schedule: { type: "once", at: new Date(2026, 7, 17, 8, 5).getTime() }, + }, request); + + expect(routine.sourceThreadId).toBe(request.threadId); + const reloaded = new RoutineManager(h.options); + expect(reloaded.listRoutines()[0]?.sourceThreadId).toBe(request.threadId); + + h.setNow(routine.nextRunAt!); + await reloaded.tick(); + const run = reloaded.listRuns()[0]!; + expect(run).toMatchObject({ + sourceThreadId: request.threadId, + threadId: "thread-1", + status: "running", + }); + expect(run.threadId).not.toBe(run.sourceThreadId); + expect(h.changed.map(({ status, sourceThreadId, threadId }) => ({ status, sourceThreadId, threadId }))) + .toEqual([ + { status: "queued", sourceThreadId: request.threadId, threadId: undefined }, + { status: "running", sourceThreadId: request.threadId, threadId: "thread-1" }, + ]); + }); + + it("does not trust a calendar payload to choose another conversation", () => { + const h = harness(); + const schedule: RoutineSchedule = { type: "daily", time: "09:00", weekdays: [1] }; + const calendarPayload = { + name: "Calendar-owned", + prompt: "Run without a chat source", + botId: "maus-1", + schedule, + sourceThreadId: "forged-thread", + }; + const routine = h.manager.create(calendarPayload); + expect(routine.sourceThreadId).toBeUndefined(); + }); + + it("keeps routine history when a persisted source thread is malformed", () => { + const h = harness(); + const request = { + requestId: "request-malformed-source", + messageId: "message-malformed-source", + botId: "maus-1", + threadId: "trusted-source", + action: "create" as const, + fingerprintVersion: 1 as const, + fingerprint: "2".repeat(64), + }; + h.manager.create({ + name: "Survives malformed provenance", + prompt: "Keep this routine", + botId: "maus-1", + schedule: { type: "daily", time: "09:00", weekdays: [1] }, + }, request); + const file = h.options.file; + if (!file) throw new Error("test harness did not configure routine persistence"); + const stored = readFileSync(file, "utf8"); + const malformed = stored.replace(`"sourceThreadId": "${request.threadId}"`, '"sourceThreadId": 42'); + expect(malformed).not.toBe(stored); + writeFileSync(file, malformed); + + const reloaded = new RoutineManager(h.options); + expect(reloaded.listRoutines()).toMatchObject([{ + name: "Survives malformed provenance", + sourceThreadId: undefined, + }]); + }); + + it("reports a chat-confirmed run-now to its invoking thread without rebinding the routine", () => { + const h = harness(); + const createRequest = { + requestId: "request-create-origin", + messageId: "message-create-origin", + botId: "maus-1", + threadId: "original-thread", + action: "create" as const, + fingerprintVersion: 1 as const, + fingerprint: "f".repeat(64), + }; + const routine = h.manager.create({ + name: "Daily source", + prompt: "Review it", + botId: "maus-1", + schedule: { type: "daily", time: "09:00", weekdays: [1] }, + }, createRequest); + const runRequest = { + ...createRequest, + requestId: "request-run-now-elsewhere", + messageId: "message-run-now-elsewhere", + threadId: "invoking-thread", + action: "run_now" as const, + fingerprint: "1".repeat(64), + }; + + const run = h.manager.runNow(routine.id, runRequest)!; + expect(run.sourceThreadId).toBe("invoking-thread"); + expect(h.manager.listRoutines()[0]?.sourceThreadId).toBe("original-thread"); + }); + it("removes unreachable recovery receipts when their conversation is deleted", () => { const h = harness(); const routine = h.manager.create({ @@ -347,9 +469,21 @@ describe("RoutineManager", () => { threadId: "thread-1", createdAt: new Date(h.manager.listRuns()[0]!.startedAt!).toISOString(), }; - h.manager.handleRuntimeEvent({ ...base, type: "request.opened", requestType: "question", tool: "ask", summary: "Need a date" }); - expect(h.manager.listRuns()[0]!.status).toBe("waiting"); + const secret = `sk-ant-api03-${"abcdefghijklmnopqrstuvwxyz0123456789"}`; + h.manager.handleRuntimeEvent({ + ...base, + type: "request.opened", + requestType: "question", + tool: "ask", + summary: `Choose the two actions before using ${secret}`, + }); + expect(h.manager.listRuns()[0]).toMatchObject({ + status: "waiting", + attention: expect.stringContaining("Choose the two actions"), + }); + expect(h.manager.listRuns()[0]!.attention).not.toContain(secret); h.manager.handleRuntimeEvent({ ...base, type: "request.resolved", behavior: "answer", source: "user" }); + expect(h.manager.listRuns()[0]!.attention).toBeUndefined(); h.manager.handleRuntimeEvent({ ...base, type: "item.completed", itemType: "assistant_text", text: "Report shipped." }); h.manager.handleRuntimeEvent({ ...base, type: "turn.completed", ok: true, cost: 0.02 }); @@ -431,6 +565,7 @@ describe("RoutineManager", () => { await h.manager.tick(); expect(h.manager.listRuns()[0]).toMatchObject({ status: "missed" }); expect(h.started).toHaveLength(0); + expect(h.failed).toMatchObject([{ id: h.manager.listRuns()[0]!.id, status: "missed" }]); }); it("records a missed receipt for a once routine created with a long-past time", async () => { diff --git a/server/routines.ts b/server/routines.ts index 2e9f28fed..a55856003 100644 --- a/server/routines.ts +++ b/server/routines.ts @@ -1,9 +1,11 @@ import { randomUUID } from "node:crypto"; import { mkdirSync, readFileSync, renameSync, writeFileSync } from "node:fs"; import { dirname, join } from "node:path"; +import { z } from "zod"; import { DATA_DIR } from "./config.ts"; import type { RuntimeEvent } from "./contracts.ts"; +import { redactSecretsInText } from "./redact.ts"; import type { RoutineRequestOperation } from "../shared/routine-request.ts"; export type RoutineSchedule = @@ -15,6 +17,8 @@ export type RoutineSchedule = * computer tools, if any. */ export type RoutineRunOn = "maus" | "cloud"; +const persistedSourceThreadId = z.string().trim().min(1).optional().catch(undefined); + export type RoutineRunTrigger = "schedule" | "manual" | "webhook"; export type RoutineRunStatus = @@ -35,6 +39,9 @@ export interface Routine { enabled: boolean; schedule: RoutineSchedule; durationMinutes: number; + /** Conversation that created this routine in chat. Calendar/import-created + * routines intentionally have no source, and older files migrate in place. */ + sourceThreadId?: string; nextRunAt: number | null; createdAt: number; updatedAt: number; @@ -56,10 +63,15 @@ export interface RoutineRun { triggerSource?: RoutineRunTrigger; webhookId?: string; deliveryId?: string; + /** Snapshot the routine's reporting destination. Execution remains on the + * separate `threadId` so recurring work never contaminates chat context. */ + sourceThreadId?: string; threadId?: string; startedAt?: number; finishedAt?: number; output?: string; + /** Human-readable reason the detached execution is waiting. */ + attention?: string; error?: string; cost?: number | null; denials?: string[]; @@ -134,6 +146,8 @@ export interface RoutineManagerOptions { onDispatchError: (message: string) => void, ) => Promise; interruptTurn?: (botId: string, threadId: string, runOn: RoutineRunOn) => Promise; + /** Projects every durable transition into the source conversation. */ + onRunChanged?: (run: RoutineRun) => void; onRunFailed?: (run: RoutineRun) => void; } @@ -224,10 +238,18 @@ export class RoutineManager { try { const disk = JSON.parse(readFileSync(this.file, "utf8")) as Partial; this.routines = Array.isArray(disk.routines) - ? disk.routines.map((routine) => ({ ...routine, runOn: routine.runOn ?? "maus" })) + ? disk.routines.map((routine) => ({ + ...routine, + runOn: routine.runOn ?? "maus", + sourceThreadId: persistedSourceThreadId.parse(routine.sourceThreadId), + })) : []; this.runs = Array.isArray(disk.runs) - ? disk.runs.map((run) => ({ ...run, runOn: run.runOn ?? "maus" })) + ? disk.runs.map((run) => ({ + ...run, + runOn: run.runOn ?? "maus", + sourceThreadId: persistedSourceThreadId.parse(run.sourceThreadId), + })) : []; this.routineRequestReceipts = Array.isArray(disk.routineRequestReceipts) ? disk.routineRequestReceipts.filter((receipt): receipt is RoutineRequestReceipt => @@ -253,13 +275,17 @@ export class RoutineManager { if (run.status === "running" || run.status === "waiting") { run.status = "failed"; run.error = "OpenMausBot restarted while this routine was running"; + run.attention = undefined; run.finishedAt = this.now(); recovered.push({ ...run }); } } if (recovered.length > 0) { this.save(); - for (const run of recovered) this.options.onRunFailed?.(run); + for (const run of recovered) { + this.notifyRunChanged(run); + this.options.onRunFailed?.(run); + } } } @@ -356,6 +382,9 @@ export class RoutineManager { const routine: Routine = { id: randomUUID(), ...clean, + // Only a confirmed chat card supplies `request`; the public calendar + // API cannot choose an arbitrary transcript as a reporting target. + sourceThreadId: request?.threadId, nextRunAt: clean.enabled ? this.initialOccurrence(clean.schedule, at) : null, createdAt: at, updatedAt: at, @@ -405,6 +434,7 @@ export class RoutineManager { for (const run of this.runs) { if (run.routineId !== routine.id || run.status !== "queued") continue; run.status = "cancelled"; + run.attention = undefined; run.finishedAt = this.now(); run.error = "The routine was paused before this run started"; cancelledRuns.push(run); @@ -432,6 +462,7 @@ export class RoutineManager { for (const run of this.runs) { if (run.routineId !== id || run.status !== "queued") continue; run.status = "cancelled"; + run.attention = undefined; run.finishedAt = this.now(); cancelledRuns.push(run); } @@ -455,6 +486,7 @@ export class RoutineManager { for (const run of this.runs) { if (run.botId !== botId || !["queued", "running", "waiting"].includes(run.status)) continue; run.status = "cancelled"; + run.attention = undefined; run.finishedAt = this.now(); run.error = "The assigned bot was deleted"; this.emitRun(run); @@ -477,6 +509,9 @@ export class RoutineManager { let run!: RoutineRun; this.commitMutation(() => { run = this.newRun(routine, this.now(), true); + // A chat-confirmed "run now" reports back to the conversation that + // invoked this one run. It must not silently rebind future schedules. + if (request) run.sourceThreadId = request.threadId; if (request) this.rememberRoutineRequest(request, run.id, this.now()); }); this.emitRun(run); @@ -534,6 +569,7 @@ export class RoutineManager { for (const run of this.runs) { if (run.webhookId !== webhookId || run.status !== "queued") continue; run.status = "cancelled"; + run.attention = undefined; run.finishedAt = this.now(); run.error = message.slice(0, 500); this.emitRun(run); @@ -546,6 +582,7 @@ export class RoutineManager { const run = this.runs.find((r) => r.id === id); if (!run || !["queued", "running", "waiting"].includes(run.status)) return null; run.status = "cancelled"; + run.attention = undefined; run.finishedAt = this.now(); this.save(); this.emitRun(run); @@ -583,6 +620,7 @@ export class RoutineManager { try { const now = this.now(); let changed = false; + const missedRuns: RoutineRun[] = []; for (const routine of this.routines) { if (!routine.enabled || routine.nextRunAt == null || routine.nextRunAt > now) continue; const scheduledFor = routine.nextRunAt; @@ -593,6 +631,7 @@ export class RoutineManager { missed.finishedAt = now; missed.error = "This computer was offline for more than 12 hours after the scheduled time"; this.emitRun(missed); + missedRuns.push({ ...missed }); } else { const run = this.newRun(routine, scheduledFor, false); this.emitRun(run); @@ -605,6 +644,7 @@ export class RoutineManager { changed = true; } if (changed) this.save(); + for (const missed of missedRuns) this.options.onRunFailed?.(missed); for (const run of [...this.runs].reverse()) { if (run.status !== "queued") continue; @@ -655,12 +695,14 @@ export class RoutineManager { if (!run) return null; if (event.type === "request.opened") { run.status = "waiting"; + run.attention = redactSecretsInText(event.summary).trim().slice(0, 500) || undefined; } else if (event.type === "request.resolved") { run.status = "running"; + run.attention = undefined; } else if (event.type === "item.completed" && event.itemType === "assistant_text") { - run.output = event.text.trim().slice(0, 2_000); + run.output = redactSecretsInText(event.text).trim().slice(0, 2_000); } else if (event.type === "runtime.error") { - run.error = event.message.slice(0, 500); + run.error = redactSecretsInText(event.message).slice(0, 500); } else if (event.type === "turn.retrying") { // the driver will relaunch this same run; a transient blip is not a // receipt-worthy failure, so keep the run running and stay quiet @@ -674,6 +716,7 @@ export class RoutineManager { return { ...run }; } run.status = "completed"; + run.attention = undefined; run.finishedAt = this.now(); run.error = undefined; } else { @@ -694,7 +737,8 @@ export class RoutineManager { private failRun(run: RoutineRun, message: string) { run.status = "failed"; - run.error = message.slice(0, 500); + run.attention = undefined; + run.error = redactSecretsInText(message).slice(0, 500); run.finishedAt = this.now(); this.save(); this.emitRun(run); @@ -725,6 +769,7 @@ export class RoutineManager { status: "queued", manual, triggerSource: manual ? "manual" : "schedule", + sourceThreadId: routine.sourceThreadId, createdAt: this.now(), }; this.runs.push(run); @@ -738,6 +783,17 @@ export class RoutineManager { private emitRun(run: RoutineRun) { this.options.emit?.({ kind: "routine.run", run: { ...run } }); + this.notifyRunChanged(run); + } + + private notifyRunChanged(run: RoutineRun) { + try { + this.options.onRunChanged?.({ ...run }); + } catch (error) { + // Reporting is secondary to scheduler truth. A transcript write must + // never strand the run in memory or prevent the next tick. + console.error("routine: source-thread lifecycle update failed", error); + } } private matchingRoutineRequestReceipt(request: RoutineRequestCommit): RoutineRequestReceipt | null { diff --git a/server/store.test.ts b/server/store.test.ts index 6e2799049..0206a7e2c 100644 --- a/server/store.test.ts +++ b/server/store.test.ts @@ -664,6 +664,23 @@ describe("Store redacts bot-authored secrets on write", () => { if (routineCard.card?.routineRequest?.operation.action !== "create") throw new Error("missing routine payload"); expect(routineCard.card.routineRequest.operation.routine.name).not.toContain(key); expect(routineCard.card.routineRequest.operation.routine.instructions).not.toContain(key); + const runCard = store.appendMessage(bot.threadId, { + role: "bot", + kind: "routine.run", + text: `Routine ${key} completed`, + routineRun: { + runId: "run-1", + routineId: "routine-1", + routineName: `Report ${key}`, + status: "completed", + executionThreadId: "execution-1", + summary: `Finished with ${key}`, + error: `Ignored ${key}`, + }, + }); + expect(runCard.routineRun?.routineName).not.toContain(key); + expect(runCard.routineRun?.summary).not.toContain(key); + expect(runCard.routineRun?.error).not.toContain(key); const secretCard = store.appendMessage(bot.threadId, { role: "bot", kind: "secret", diff --git a/server/store.ts b/server/store.ts index 4cef9dd53..50d205e0b 100644 --- a/server/store.ts +++ b/server/store.ts @@ -15,6 +15,7 @@ import { pickBotName } from "./names.ts"; import { redactSecretsInText } from "./redact.ts"; import { botAvatarProfile, type BotAvatarCrop } from "../shared/bot-avatar.ts"; import type { RoutineRequestCardData } from "../shared/routine-request.ts"; +import type { RoutineRunCardData } from "../shared/routine-run.ts"; export type MausColor = | "green" @@ -87,11 +88,14 @@ export interface SecretRequestCardData { export interface Message { id: string; role: "bot" | "user"; - kind: "text" | "options" | "activity" | "screen" | "connector" | "secret"; + kind: "text" | "options" | "activity" | "screen" | "connector" | "secret" | "routine.run"; text?: string; card?: OptionCardData; connector?: ConnectorCardData; secret?: SecretRequestCardData; + /** One idempotently updated status card in the conversation that created a + * routine. The actual provider turn remains in its isolated task. */ + routineRun?: RoutineRunCardData; /** activity messages: tool name + outcome. `spoken` is the same chip as * a phrase a voice can read ("reading a file") — computed once here so * call mode never has to re-derive it from the raw tool name, and absent @@ -242,6 +246,13 @@ function redactBotAuthored & { at?: number const out = { ...message }; if (typeof out.text === "string") out.text = redactSecretsInText(out.text); if (out.tool?.name) out.tool = { ...out.tool, name: redactSecretsInText(out.tool.name) }; + if (out.routineRun) { + const routineRun = { ...out.routineRun }; + routineRun.routineName = redactSecretsInText(routineRun.routineName); + if (routineRun.summary) routineRun.summary = redactSecretsInText(routineRun.summary); + if (routineRun.error) routineRun.error = redactSecretsInText(routineRun.error); + out.routineRun = routineRun; + } if (out.card) { const card = { ...out.card } as OptionCardData & { summary?: string }; card.title = redactSecretsInText(card.title); diff --git a/shared/routine-run.ts b/shared/routine-run.ts new file mode 100644 index 000000000..8ec38a486 --- /dev/null +++ b/shared/routine-run.ts @@ -0,0 +1,16 @@ +/** Durable, non-actionable projection of one background routine run. + * + * The provider still runs in its isolated execution task. This small card is + * upserted into the trusted conversation that created the routine so the user + * can see progress, results, and where to review an approval without hunting + * through the routines calendar. + */ +export interface RoutineRunCardData { + runId: string; + routineId: string; + routineName: string; + status: "queued" | "running" | "waiting" | "completed" | "failed" | "cancelled" | "missed"; + executionThreadId?: string; + summary?: string; + error?: string; +} diff --git a/src/components/ChatView.tsx b/src/components/ChatView.tsx index 29da9fa24..2dc71e2b0 100644 --- a/src/components/ChatView.tsx +++ b/src/components/ChatView.tsx @@ -30,6 +30,7 @@ import { useStreaming, formatTime, messageVersions, + openNotificationTarget, visibleMessages, type Bot, type InstanceInfo, @@ -50,6 +51,7 @@ import { ChatFindBar } from "./ChatFindBar"; import { ReplyQuote } from "./ReplyQuote"; import { ConnectorCard } from "./ConnectorCard"; import { SecretRequestCard } from "./SecretRequestCard"; +import { hasRoutineExecutionTask, RoutineRunCard } from "./RoutineRunCard"; import { AttachedImageGallery } from "./AttachmentPreview"; import { ModelPicker } from "./ModelPicker"; import { RenameTitle } from "./RenameTitle"; @@ -704,6 +706,22 @@ const MessagesList = memo(function MessagesList({ } if (shouldHideOnboardingCard(m, transcript)) return null; return ; + case "routine.run": { + const executionThreadId = m.routineRun?.executionThreadId; + const canOpen = hasRoutineExecutionTask(bot.tasks, executionThreadId); + return ( + openNotificationTarget( + dispatch, + { botId: bot.id, threadId: executionThreadId }, + state, + ) + : undefined} + /> + ); + } case "activity": { // a failed turn is an error, not a tool run — render it as one. // bot⇄bot comm chips stay because they link to another conversation. diff --git a/src/components/GroupView.tsx b/src/components/GroupView.tsx index 7d7c4bf76..50da0a075 100644 --- a/src/components/GroupView.tsx +++ b/src/components/GroupView.tsx @@ -9,6 +9,7 @@ import { useStore, useStreaming, formatTime, + openNotificationTarget, type Bot, type Group, type GroupDefaultResponder, @@ -26,6 +27,7 @@ import { GroupTaskPicker } from "./TaskPicker"; import { ReplyQuote } from "./ReplyQuote"; import { ConnectorCard } from "./ConnectorCard"; import { SecretRequestCard } from "./SecretRequestCard"; +import { hasRoutineExecutionTask, RoutineRunCard } from "./RoutineRunCard"; import { AttachedImageGallery } from "./AttachmentPreview"; import { GroupCallButton, GroupCallOverlay } from "./GroupCallView"; import { ReactionBar, ReactionChips } from "./Reactions"; @@ -179,6 +181,11 @@ const Transcript = memo(function Transcript({ const user = m.role === "user"; const attachedImages = user && m.text ? splitAttachedImages(m.text) : null; const newCluster = !prev || prev.role !== m.role || prev.from?.botId !== m.from?.botId || newDay; + const routineOwner = m.kind === "routine.run" ? memberOf(m.from?.botId) : undefined; + const routineExecutionThreadId = m.routineRun?.executionThreadId; + const routineTarget = routineOwner && hasRoutineExecutionTask(routineOwner.tasks, routineExecutionThreadId) + ? { botId: routineOwner.id, threadId: routineExecutionThreadId } + : undefined; const row = // a member can hit a permission ask mid-turn; without this the // card never rendered here and the bot waited out its timeout. @@ -193,6 +200,15 @@ const Transcript = memo(function Transcript({
+ ) : m.kind === "routine.run" ? ( +
+ openNotificationTarget(dispatch, routineTarget, state) + : undefined} + /> +
) : m.kind === "activity" && m.tool ? ( m.tool.ok === false || m.tool.name.startsWith("error:") || showToolCalls ? ( diff --git a/src/components/RoutineRunCard.test.ts b/src/components/RoutineRunCard.test.ts new file mode 100644 index 000000000..21e00ffbe --- /dev/null +++ b/src/components/RoutineRunCard.test.ts @@ -0,0 +1,98 @@ +import { createElement } from "react"; +import { renderToStaticMarkup } from "react-dom/server"; +import { describe, expect, it, vi } from "vitest"; + +import type { RoutineRunCardData } from "../../shared/routine-run"; +import type { Message } from "@/state/store"; +import { hasRoutineExecutionTask, RoutineRunCard } from "./RoutineRunCard"; + +function message( + status: RoutineRunCardData["status"], + patch: Partial = {}, +): Message { + return { + id: "routine-run-card", + role: "bot", + kind: "routine.run", + at: 1, + text: "Morning brief routine update", + routineRun: { + runId: "run-1", + routineId: "routine-1", + routineName: "Morning brief", + status, + executionThreadId: "execution-thread", + ...patch, + }, + }; +} + +describe("RoutineRunCard", () => { + it("shows a compact completion receipt and a path to the isolated run", () => { + const markup = renderToStaticMarkup(createElement(RoutineRunCard, { + message: message("completed", { summary: "The brief is ready with three follow-ups." }), + onOpen: vi.fn(), + })); + + expect(markup).toContain("Morning brief"); + expect(markup).toContain("Completed"); + expect(markup).toContain("The brief is ready with three follow-ups."); + expect(markup).toContain("Open run"); + expect(markup).toContain('aria-label="Morning brief routine run: Completed"'); + }); + + it("makes a waiting question or approval an explicit Review action", () => { + const markup = renderToStaticMarkup(createElement(RoutineRunCard, { + message: message("waiting", { summary: "The routine needs an answer before it can continue." }), + onOpen: vi.fn(), + })); + + expect(markup).toContain("Needs your input"); + expect(markup).toContain("Review"); + expect(markup).toContain('aria-label="Review for Morning brief"'); + }); + + it("shows a concise error without dumping a verbose run log into chat", () => { + const verbose = `Provider failed ${"trace-line ".repeat(100)}`; + const markup = renderToStaticMarkup(createElement(RoutineRunCard, { + message: message("failed", { error: verbose }), + onOpen: vi.fn(), + })); + + expect(markup).toContain("Failed"); + expect(markup).toContain("Provider failed"); + expect(markup).toContain("…"); + expect(markup).not.toContain(verbose); + }); + + it("keeps a concise text fallback visible for an incomplete or newer payload", () => { + const legacy: Message = { + id: "legacy-run", + role: "bot", + kind: "routine.run", + at: 1, + text: "Morning brief completed.", + }; + + const markup = renderToStaticMarkup(createElement(RoutineRunCard, { message: legacy })); + expect(markup).toContain("Morning brief completed."); + }); + + it("does not offer a dead navigation action when the execution task is unavailable", () => { + const markup = renderToStaticMarkup(createElement(RoutineRunCard, { + message: message("missed", { executionThreadId: undefined, error: "Computer was offline." }), + onOpen: vi.fn(), + })); + + expect(markup).toContain("Missed"); + expect(markup).not.toContain("Open run"); + }); + + it("recognizes only an execution thread still present in the owning bot's tasks", () => { + const tasks = [{ threadId: "source-thread" }, { threadId: "execution-thread" }]; + + expect(hasRoutineExecutionTask(tasks, "execution-thread")).toBe(true); + expect(hasRoutineExecutionTask(tasks, "deleted-thread")).toBe(false); + expect(hasRoutineExecutionTask(undefined, "execution-thread")).toBe(false); + }); +}); diff --git a/src/components/RoutineRunCard.tsx b/src/components/RoutineRunCard.tsx new file mode 100644 index 000000000..50d35ceef --- /dev/null +++ b/src/components/RoutineRunCard.tsx @@ -0,0 +1,134 @@ +import { + CalendarClock, + CheckCircle2, + CircleAlert, + ExternalLink, + Loader2, + ShieldAlert, + XCircle, +} from "lucide-react"; + +import { cn } from "@/lib/cn"; +import type { RoutineRunCardData } from "../../shared/routine-run"; +import type { Message } from "@/state/store"; + +const DETAIL_LIMIT = 280; + +const COPY = { + queued: { label: "Queued", tone: "text-ink-secondary", border: "border-hairline/45" }, + running: { label: "Running", tone: "text-accent", border: "border-accent/30" }, + waiting: { label: "Needs your input", tone: "text-warning", border: "border-warning/35" }, + completed: { label: "Completed", tone: "text-success", border: "border-success/30" }, + failed: { label: "Failed", tone: "text-danger", border: "border-danger/35" }, + cancelled: { label: "Cancelled", tone: "text-ink-secondary", border: "border-hairline/45" }, + missed: { label: "Missed", tone: "text-danger", border: "border-danger/35" }, +} satisfies Record< + RoutineRunCardData["status"], + { label: string; tone: string; border: string } +>; + +function compactDetail(value: string | undefined): string { + const clean = value?.replace(/\s+/g, " ").trim() ?? ""; + return clean.length > DETAIL_LIMIT ? `${clean.slice(0, DETAIL_LIMIT - 1).trimEnd()}…` : clean; +} + +/** A lifecycle receipt can outlive its isolated execution task. Only offer + * navigation while the task is still present in the owning bot's task list. */ +export function hasRoutineExecutionTask( + tasks: ReadonlyArray<{ threadId: string }> | undefined, + executionThreadId: string | undefined, +): executionThreadId is string { + return Boolean( + executionThreadId && tasks?.some((task) => task.threadId === executionThreadId), + ); +} + +function StatusIcon({ status }: { status: RoutineRunCardData["status"] }) { + const className = "size-4 shrink-0"; + switch (status) { + case "running": + return