feat(routines): report runs in source conversations

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<string> => {
   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}`);

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<Message["routineRun"]> {
+  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<Message["routineRun"]> = {
+    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<Message["routineRun"]>): 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<Message, "id" | "at"> = {
+      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<RoutineManager["listRoutines"]>[number]) => {
+const routineTimestamp = (value: number | undefined) =>
+  value !== undefined && Number.isFinite(value) ? new Date(value).toISOString() : null;
+const agentRoutine = (
+  routine: ReturnType<RoutineManager["listRoutines"]>[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<RoutineManager["listRoutines"]>[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<string, RoutineRun>();
+        // 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") {