feat(routines): let a routine's result ask for someone, and stay quiet when there's nothing to report

packages/contracts/src/chats.ts

@@ -231,8 +231,9 @@ const activityPlaceFields = {
 
 /**
  * Something in one of the person's conversations worth their attention: a message
- * that mentions them or that they have not read, a routine run, or a
- * collaboration one bot opened with another.
+ * that mentions them or that they have not read, a routine run that did not
+ * end with nothing to report, or a collaboration one bot opened with another.
+ * A run is unread only once it posted a result or failed.
  */
 export const activityItemSchema = Schema.Union([
 	Schema.Struct({
@@ -256,6 +257,8 @@ export const activityItemSchema = Schema.Union([
 		agent: agentParticipantSchema,
 		/** The start of the run's last reply, or null while it has written nothing. */
 		preview: Schema.NullOr(Schema.String),
+		/** Whether the result it posted mentions the person, asking something of them. */
+		mentionsYou: Schema.Boolean,
 		...activityPlaceFields,
 	}),
 	Schema.Struct({

packages/contracts/src/routines.ts

@@ -482,6 +482,11 @@ export const routineExecutionSchema = Schema.Struct({
 	trigger: routineExecutionTriggerSchema,
 	state: routineExecutionStateSchema,
 	error: Schema.NullOr(Schema.String),
+	/**
+	 * Whether it posted a result for people to read. A run that completed
+	 * without posting one had nothing to report.
+	 */
+	posted: Schema.Boolean,
 	acceptedAt: isoTimestampSchema,
 	startedAt: Schema.NullOr(isoTimestampSchema),
 	finishedAt: Schema.NullOr(isoTimestampSchema),

packages/core/src/conversations/chats/chat-view.ts

@@ -74,7 +74,7 @@ import { isUuid } from "../../ids/ids.ts";
 import type { UserFacing } from "../../user-message.ts";
 import { crewAgentRow, toAgent } from "../../workspaces/agents/agent.ts";
 import { type CursorPoint, decodeCursor, earlierThan, encodeCursor } from "../cursor.ts";
-import { latestRunFailedUnseen, routineExecutionIdOf } from "../routines/routines.ts";
+import { latestRunFailedUnseen, resultPosted, routineExecutionIdOf } from "../routines/routines.ts";
 import {
 	agentColumns,
 	authorRow,
@@ -116,7 +116,8 @@ export interface Interface {
 	/**
 	 * What is new for the actor across the workspace's conversations, newest first:
 	 * messages that mention them, read or not, and those they have not read;
-	 * and every routine run and collaboration, read or not.
+	 * every routine run, read or not, but those that completed with nothing to
+	 * report; and every collaboration, read or not.
 	 */
 	readonly activity: (
 		workspace: string,
@@ -328,13 +329,18 @@ export const make = Effect.gen(function* () {
 					const { workspaceId } = yield* authorization.workspace(workspace, "workspace.read");
 					const reachesPod = yield* visibility.reachesPod;
 					const { userId } = yield* CurrentActor.Service;
+					const [person] = yield* query((db) =>
+						db.select({ name: user.name }).from(user).where(eq(user.id, userId)).limit(1),
+					);
+					if (!person) return { items: [] };
 					const since = DateTime.toDate(
 						DateTime.subtract(yield* DateTime.now, { days: ACTIVITY_WINDOW_DAYS }),
 					);
 					const scope = {
 						workspaceId,
 						reachesPod,
 						userId,
+						handle: handleFromName(person.name),
 						since,
 					};
 					const messages = yield* query((db) => activityRows(db, scope));
@@ -362,6 +368,7 @@ export const make = Effect.gen(function* () {
 								triggerKind: row.triggerKind,
 								agent: { kind: "agent", ...row.agent },
 								preview: row.lastReply && runReportPreview(row.lastReply),
+								mentionsYou: row.mentionsYou,
 								podId: row.podId,
 								conversationId: row.conversationId,
 								at: row.at.toISOString(),
@@ -753,13 +760,7 @@ const activityRows = Effect.fn("ChatView.activityRows")(function* (
 	db: Executor,
 	input: ActivityScope,
 ) {
-	const [person] = yield* db
-		.select({ name: user.name })
-		.from(user)
-		.where(eq(user.id, input.userId))
-		.limit(1);
-	if (!person) return [];
-	const mentionsYou = mentions(handleFromName(person.name));
+	const mentionsYou = mentions(input.handle);
 	const unread = unreadBy(db, input.userId, {
 		threadId: message.threadId,
 		at: message.createdAt,
@@ -812,6 +813,8 @@ interface ActivityScope {
 	workspaceId: string;
 	reachesPod: Visibility.ReachesPod;
 	userId: string;
+	/** The person's handle, which mentions of them use. */
+	handle: string;
 	since: Date;
 }
 
@@ -880,12 +883,40 @@ const failedRoutinesByPod = Effect.fn("ChatView.failedRoutinesByPod")(function*
 	return new Map(rows.flatMap((row) => (row.podId ? [[row.podId, row.count] as const] : [])));
 });
 
-/** The routine runs in the conversations `userId` reaches since `since`, newest first, with their last replies. */
+/**
+ * The routine runs in the conversations `userId` reaches since `since`, newest
+ * first, with their last replies, but not those that completed with nothing to
+ * report. A run is unread news only once it posted a result or failed, from
+ * when it ended. It mentions the person when its result, posted only in the
+ * routine's thread, does: one also in a pod's chat mentions them there.
+ */
 const routineRunRows = Effect.fn("ChatView.routineRunRows")(function* (
 	db: Executor,
 	input: ActivityScope,
 ) {
 	const conversation = alias(thread, "conversation");
+	const posted = resultPosted(routineExecution.id);
+	const elsewhere = alias(message, "result_elsewhere");
+	const resultMentioningYou = db
+		.select({ id: message.id })
+		.from(message)
+		.where(
+			and(
+				eq(message.routineExecutionId, routineExecution.id),
+				eq(message.threadId, conversation.id),
+				mentions(input.handle),
+				sql`not exists (${db
+					.select({ id: elsewhere.id })
+					.from(elsewhere)
+					.where(
+						and(
+							eq(elsewhere.routineExecutionId, routineExecution.id),
+							ne(elsewhere.threadId, conversation.id),
+						),
+					)})`,
+			),
+		);
+	const isNews = or(eq(routineExecution.state, "failed"), posted);
 	return yield* db
 		.select({
 			threadId: routineExecution.threadId,
@@ -903,11 +934,12 @@ const routineRunRows = Effect.fn("ChatView.routineRunRows")(function* (
 			},
 			podId: conversation.podId,
 			conversationId: conversation.id,
-			unread: itemUnreadBy(db, input.userId, {
+			unread: sql<boolean>`(${isNews} and ${itemUnreadBy(db, input.userId, {
 				conversationId: conversation.id,
 				itemThreadId: routineExecution.threadId,
-				at: routineExecution.acceptedAt,
-			}),
+				at: sql`coalesce(${routineExecution.finishedAt}, ${routineExecution.acceptedAt})`,
+			})})`,
+			mentionsYou: sql<boolean>`exists (${resultMentioningYou})`,
 			lastReply: lastReplyIn(db, routineExecution.threadId),
 		})
 		.from(routineExecution)
@@ -919,6 +951,7 @@ const routineRunRows = Effect.fn("ChatView.routineRunRows")(function* (
 				eq(conversation.workspaceId, input.workspaceId),
 				input.reachesPod(conversation.podId),
 				gt(routineExecution.acceptedAt, input.since),
+				or(ne(routineExecution.state, "completed"), posted),
 			),
 		)
 		.orderBy(desc(routineExecution.acceptedAt))
@@ -1254,7 +1287,10 @@ const loadMainPage = Effect.fn("ChatView.loadMainPage")(function* (
 			state: true,
 			error: true,
 		},
-		extras: { report: (row) => lastReplyIn(db, row.threadId) },
+		extras: {
+			report: (row) => lastReplyIn(db, row.threadId),
+			posted: (row) => resultPosted(sql`${row.id}`),
+		},
 	});
 	const [paused] = yield* db
 		.select({ routineId: routine.id, at: routine.pausedAfterFailuresAt })
@@ -1373,7 +1409,6 @@ function pausedCandidate(routineId: string, at: Date) {
 
 /** The sources merged newest first, cut to the page, and put back in reading order. */
 function toMainPage(row: MainPage, limit: number) {
-	const posted = new Set(row.messages.flatMap((stored) => stored.routineExecutionId ?? []));
 	const candidates = [
 		...row.messages.map((stored) => ({
 			id: stored.id,
@@ -1414,7 +1449,7 @@ function toMainPage(row: MainPage, limit: number) {
 		})),
 		// A run that posted its result is shown by the post, which opens the run.
 		...row.routineExecutions
-			.filter((execution) => !posted.has(execution.id))
+			.filter((execution) => !execution.posted)
 			.map((execution) => ({
 				id: execution.id,
 				createdAt: execution.acceptedAt,
@@ -1426,7 +1461,8 @@ function toMainPage(row: MainPage, limit: number) {
 					triggerKind: execution.triggerKind,
 					state: execution.state,
 					error: execution.error,
-					report: execution.report && spokenText(execution.report),
+					// Only what the bot said after its last tool call, as a posted result would be.
+					report: execution.report && (textWithoutNarration(execution.report.parts) || null),
 					createdAt: execution.acceptedAt.toISOString(),
 				}),
 			})),

packages/core/src/conversations/chats/chats.test.ts

@@ -2151,6 +2151,38 @@ describe.skipIf(!process.env.DATABASE_URL)("chats, against Postgres", async () =
 			return written;
 		};
 
+		/** The run completes, posting `content` as its result in the routine's thread. */
+		const runPosts = async (run: { executionId: string; threadId: string }, content: string) => {
+			await onDatabase((db) =>
+				db
+					.update(routineExecution)
+					.set({
+						state: "completed",
+						startedAt: sql`${routineExecution.acceptedAt}`,
+						finishedAt: sql`${routineExecution.acceptedAt} + interval '1 minute'`,
+					})
+					.where(eq(routineExecution.id, run.executionId)),
+			);
+			const [runThread] = await onDatabase((db) =>
+				db
+					.select({ conversationId: thread.conversationId })
+					.from(thread)
+					.where(eq(thread.id, run.threadId)),
+			);
+			if (!runThread?.conversationId) throw new Error("fixture");
+			await onDatabase((db) =>
+				db.insert(message).values({
+					threadId: runThread.conversationId as string,
+					authorAgentId: agentId,
+					routineExecutionId: run.executionId,
+					kind: "text",
+					status: "complete",
+					parts: [{ type: "text", text: content }],
+					content,
+				}),
+			);
+		};
+
 		/** A reply in the chat that called `tool` and stopped for someone to approve it. */
 		const askForApproval = async (threadId: string, tool = "linear__create_issue") => {
 			const reply = await botWrites(threadId, "", 1);
@@ -2414,6 +2446,7 @@ describe.skipIf(!process.env.DATABASE_URL)("chats, against Postgres", async () =
 				requestId: crypto.randomUUID(),
 			});
 			await botWrites(accepted.threadId, "Two PRs merged.\nThe release branch is green.", 1);
+			await runPosts(accepted, "Two PRs merged.\nThe release branch is green.");
 
 			expect((await view.activity(workspaceId)).items).toContainEqual(
 				expect.objectContaining({
@@ -2425,11 +2458,41 @@ describe.skipIf(!process.env.DATABASE_URL)("chats, against Postgres", async () =
 					preview: "Two PRs merged. The release branch is green.",
 					conversationId: (await view.list({ workspace: workspaceId, pod: podId }))?.routines[0]
 						?.conversation?.id,
+					mentionsYou: false,
 					unread: true,
 				}),
 			);
 		});
 
+		it("counts a routine run unread only once it posted a result or failed", async () => {
+			const created = await routines.create(
+				{ agentId },
+				{ name: "Overnight review", instructions: "Review it.", triggers: [{ kind: "webhook" }] },
+			);
+			const runRoutine = () =>
+				routines.run({ agentId, routineId: created.routine.id, requestId: crypto.randomUUID() });
+			const running = await runRoutine();
+			const quiet = await runRoutine();
+			await onDatabase((db) =>
+				db
+					.update(routineExecution)
+					.set({ state: "completed", startedAt: new Date(), finishedAt: new Date() })
+					.where(eq(routineExecution.id, quiet.executionId)),
+			);
+			const asking = await runRoutine();
+			await runPosts(asking, "@chat-member the deploy needs your sign-off.");
+
+			const runs = (await view.activity(workspaceId)).items.flatMap((item) =>
+				item.kind === "routine"
+					? [{ threadId: item.threadId, unread: item.unread, mentionsYou: item.mentionsYou }]
+					: [],
+			);
+			expect(runs).toEqual([
+				{ threadId: asking.threadId, unread: true, mentionsYou: true },
+				{ threadId: running.threadId, unread: false, mentionsYou: false },
+			]);
+		});
+
 		it("shows what a routine run said in the routine's thread", async () => {
 			const created = await routines.create(
 				{ agentId },
@@ -2453,6 +2516,62 @@ describe.skipIf(!process.env.DATABASE_URL)("chats, against Postgres", async () =
 			]);
 		});
 
+		it("makes a run opened while it ran news again once it posts its result", async () => {
+			const created = await routines.create(
+				{ agentId },
+				{ name: "Overnight review", instructions: "Review it.", triggers: [{ kind: "webhook" }] },
+			);
+			const run = await routines.run({
+				agentId,
+				routineId: created.routine.id,
+				requestId: crypto.randomUUID(),
+			});
+			await chats.markItemRead(run.threadId);
+
+			await runPosts(run, "@chat-member the deploy needs your sign-off.");
+
+			expect((await view.activity(workspaceId)).items).toContainEqual(
+				expect.objectContaining({ threadId: run.threadId, unread: true, mentionsYou: true }),
+			);
+		});
+
+		it("leaves a result posted in a pod's chat as well to mention the person there", async () => {
+			const created = await routines.create(
+				{ agentId },
+				{ name: "Overnight review", instructions: "Review it.", triggers: [{ kind: "webhook" }] },
+			);
+			const run = await routines.run({
+				agentId,
+				routineId: created.routine.id,
+				requestId: crypto.randomUUID(),
+			});
+			const content = "@chat-member the deploy needs your sign-off.";
+			await runPosts(run, content);
+			const podChat = await chats.openPodChat({ workspace: workspaceId, podId });
+			const [inChat] = await onDatabase((db) =>
+				db
+					.insert(message)
+					.values({
+						threadId: podChat.id,
+						authorAgentId: agentId,
+						routineExecutionId: run.executionId,
+						kind: "text",
+						status: "complete",
+						parts: [{ type: "text", text: content }],
+						content,
+					})
+					.returning(),
+			);
+
+			const { items } = await view.activity(workspaceId);
+			expect(items).toContainEqual(
+				expect.objectContaining({ kind: "mention", messageId: inChat?.id }),
+			);
+			expect(items).toContainEqual(
+				expect.objectContaining({ threadId: run.threadId, mentionsYou: false }),
+			);
+		});
+
 		it("lists a collaboration in the person's chats", async () => {
 			const current = await chats.openBotChannel({
 				workspace: workspaceId,
@@ -2510,6 +2629,7 @@ describe.skipIf(!process.env.DATABASE_URL)("chats, against Postgres", async () =
 			const earlier = await runRoutine();
 			const opened = await runRoutine();
 			const later = await runRoutine();
+			for (const run of [earlier, opened, later]) await runPosts(run, "All green.");
 
 			await chats.markItemRead(opened.threadId);
 
@@ -2581,6 +2701,7 @@ describe.skipIf(!process.env.DATABASE_URL)("chats, against Postgres", async () =
 				routineId: created.routine.id,
 				requestId: crypto.randomUUID(),
 			});
+			await runPosts(run, "All green.");
 			const childThreadId = await collaborate(current.id, current.id);
 			const unreadOf = async () =>
 				(await view.activity(workspaceId)).items.flatMap((item) =>