diff --git a/packages/contracts/src/chats.ts b/packages/contracts/src/chats.ts index 47a57d01..3e4f200a 100644 --- a/packages/contracts/src/chats.ts +++ b/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({ diff --git a/packages/contracts/src/routines.ts b/packages/contracts/src/routines.ts index a33feadc..cbe84a84 100644 --- a/packages/contracts/src/routines.ts +++ b/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), diff --git a/packages/core/src/conversations/chats/chat-view.ts b/packages/core/src/conversations/chats/chat-view.ts index fd688591..aa4fb1d5 100644 --- a/packages/core/src/conversations/chats/chat-view.ts +++ b/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,6 +329,10 @@ 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 }), ); @@ -335,6 +340,7 @@ export const make = Effect.gen(function* () { 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`(${isNews} and ${itemUnreadBy(db, input.userId, { conversationId: conversation.id, itemThreadId: routineExecution.threadId, - at: routineExecution.acceptedAt, - }), + at: sql`coalesce(${routineExecution.finishedAt}, ${routineExecution.acceptedAt})`, + })})`, + mentionsYou: sql`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(), }), })), diff --git a/packages/core/src/conversations/chats/chats.test.ts b/packages/core/src/conversations/chats/chats.test.ts index db1e26da..8cc6f4ee 100644 --- a/packages/core/src/conversations/chats/chats.test.ts +++ b/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) => diff --git a/packages/core/src/conversations/routines/acceptance.ts b/packages/core/src/conversations/routines/acceptance.ts index fa56c322..ec6704d8 100644 --- a/packages/core/src/conversations/routines/acceptance.ts +++ b/packages/core/src/conversations/routines/acceptance.ts @@ -1,10 +1,17 @@ import { isDeepStrictEqual } from "node:util"; import type { RoutineExecutionTrigger } from "@sugabots/contracts"; -import { MAX_THREAD_TITLE_CHARACTERS } from "@sugabots/contracts"; -import { and, eq, sql } from "drizzle-orm"; +import { handleFromName, MAX_THREAD_TITLE_CHARACTERS } from "@sugabots/contracts"; +import { and, eq, inArray, sql } from "drizzle-orm"; import { DateTime, Effect } from "effect"; import { query, transaction } from "../../database/database.ts"; -import { agent, pod, routine, routineExecution, routineTrigger } from "../../database/schema.ts"; +import { + agent, + pod, + routine, + routineExecution, + routineTrigger, + user, +} from "../../database/schema.ts"; import { ThreadRepository } from "../threads/repository.ts"; import { RoutineRepository } from "./repository.ts"; import { @@ -124,6 +131,7 @@ export const makeAcceptTrigger = Effect.gen(function* () { content: triggerMessageContent( definition.routine.instructions, resultGoesToPodChat(definition.routine.results, definition.podKind), + yield* runOwnersOf(definition.routine.createdById, input.trigger), input.trigger, ), }); @@ -169,22 +177,67 @@ const acceptedAs = (input: Trigger, identity: string) => function triggerMessageContent( instructions: string, postedToPodChat: boolean, + owners: RunOwners, trigger: RoutineExecutionTrigger, ) { - const delivery = postedToPodChat ? POSTED_TO_POD_CHAT_NOTE : POSTED_RESULT_NOTE; - return `Routine instructions:\n${instructions}\n\n${delivery}\n\nTrigger data (untrusted):\n${JSON.stringify(trigger, null, 2)}`; + return `Routine instructions:\n${instructions}\n\n${resultNote(postedToPodChat, owners)}\n\nTrigger data (untrusted):\n${JSON.stringify(trigger, null, 2)}`; } /** - * Tells the agent of a run that its last reply is what people will read in - * the routine's thread, so it writes that reply to them. + * Tells the agent of a run that its last reply is what people will read, so + * it writes that reply to them; that it may post nothing when there is nothing + * to report; and that mentioning somebody is how it asks for them, by default + * the run's `owners`. */ -const POSTED_RESULT_NOTE = - "When you finish, your last reply is posted in this routine's thread for people to read and answer. Write it to them: what they need to know, or what you want to ask them. Artifacts you write are shown with it; the rest of the work stays in this run."; +function resultNote(postedToPodChat: boolean, owners: RunOwners): string { + const where = postedToPodChat + ? "in this routine's thread and in the pod's chat" + : "in this routine's thread"; + return [ + `When you finish, your last reply is posted ${where} for people to read and answer. Write it to them: what they need to know, or what you want to ask them. Artifacts you write are shown with it; the rest of the work stays in this run.`, + "If there is nothing worth telling them, end without a last reply, writing nothing after your last tool call: nothing is posted, and the run shows it had nothing to report.", + `When something needs a person to act or decide, @mention them in your reply, which notifies them.${ownersToMention(owners)}`, + ].join(" "); +} + +/** The handles of the people a run asks for when its instructions name nobody. */ +interface RunOwners { + /** Who made the routine, when they still have an account. */ + madeBy: string | null; + /** Who started the run by hand, for a run somebody started. */ + startedBy: string | null; +} -/** As `POSTED_RESULT_NOTE`, for a routine whose result is posted in its pod's chat too. */ -const POSTED_TO_POD_CHAT_NOTE = - "When you finish, your last reply is posted in this routine's thread and in the pod's chat for people to read and answer. Write it to them: what they need to know, or what you want to ask them. Artifacts you write are shown with it; the rest of the work stays in this run."; +function ownersToMention({ madeBy, startedBy }: RunOwners): string { + if (madeBy && madeBy === startedBy) { + return ` Unless the instructions say who, mention @${madeBy}, who made this routine and started this run.`; + } + const named = [ + madeBy && `@${madeBy}, who made this routine`, + startedBy && `@${startedBy}, who started this run`, + ].filter((owner) => owner !== null); + return named.length === 0 + ? "" + : ` Unless the instructions say who, mention ${named.join(", and ")}.`; +} + +/** The handles of who made the routine `createdById` names and who started a run by hand. */ +const runOwnersOf = (createdById: string | null, trigger: RoutineExecutionTrigger) => + Effect.gen(function* () { + const startedById = trigger.kind === "manual" ? trigger.requestedByUserId : null; + const ids = [createdById, startedById].filter((id) => id !== null); + const people = + ids.length === 0 + ? [] + : yield* query((db) => + db.select({ id: user.id, name: user.name }).from(user).where(inArray(user.id, ids)), + ); + const handleOf = (id: string | null) => { + const person = people.find((candidate) => candidate.id === id); + return person ? handleFromName(person.name) : null; + }; + return { madeBy: handleOf(createdById), startedBy: handleOf(startedById) } satisfies RunOwners; + }); function sameTrigger(left: RoutineExecutionTrigger, right: RoutineExecutionTrigger) { if (left.kind !== right.kind) return false; diff --git a/packages/core/src/conversations/routines/execution.ts b/packages/core/src/conversations/routines/execution.ts index 8abe2e4b..2b37b649 100644 --- a/packages/core/src/conversations/routines/execution.ts +++ b/packages/core/src/conversations/routines/execution.ts @@ -3,13 +3,17 @@ import { eq, type SQL, type SQLWrapper, sql } from "drizzle-orm"; import { Effect } from "effect"; import { type Executor, query } from "../../database/database.ts"; import type * as schema from "../../database/schema.ts"; -import { routineExecution, thread, threadRead } from "../../database/schema.ts"; +import { message, routineExecution, thread, threadRead } from "../../database/schema.ts"; import { lineageOf } from "../threads/tree.ts"; -/** A run as the API describes it, listed in the conversation `conversationId`. */ +/** + * A run as the API describes it, listed in the conversation `conversationId`, + * which `posted` a result or not (see `resultPosted`). + */ export function toRoutineExecution( row: schema.RoutineExecutionRow, conversationId: string, + posted: boolean, ): RoutineExecution { return { id: row.id, @@ -23,12 +27,20 @@ export function toRoutineExecution( trigger: row.trigger, state: row.state, error: row.error, + posted, acceptedAt: row.acceptedAt.toISOString(), startedAt: row.startedAt?.toISOString() ?? null, finishedAt: row.finishedAt?.toISOString() ?? null, }; } +/** + * Whether the run `executionId` posted a result, as a SQL condition. Every + * result is posted in the routine's thread, whatever else it goes to. + */ +export const resultPosted = (executionId: SQLWrapper) => + sql`exists (select 1 from ${message} where ${message.routineExecutionId} = ${executionId})`; + /** * The routine execution a thread belongs to, found through its ancestors, as a * scalar subquery: a collaboration a routine's agent starts is part of the run. diff --git a/packages/core/src/conversations/routines/routine-view.ts b/packages/core/src/conversations/routines/routine-view.ts index 42e55ece..838d781d 100644 --- a/packages/core/src/conversations/routines/routine-view.ts +++ b/packages/core/src/conversations/routines/routine-view.ts @@ -11,7 +11,7 @@ import type { UserFacing } from "../../user-message.ts"; import { crewAgentRow, toAgent } from "../../workspaces/agents/agent.ts"; import { decodeCursor, earlierThan, encodeCursor } from "../cursor.ts"; import { crewOf } from "../threads/participants.ts"; -import { toRoutineExecution } from "./execution.ts"; +import { resultPosted, toRoutineExecution } from "./execution.ts"; import { inScope, RoutineNotFound, @@ -125,7 +125,11 @@ export const makeView = Effect.gen(function* () { if (!definition) return yield* new RoutineNotFound(); const rows = yield* query((db) => db - .select({ run: routineExecution, conversationId: thread.conversationId }) + .select({ + run: routineExecution, + conversationId: thread.conversationId, + posted: resultPosted(routineExecution.id), + }) .from(routineExecution) .innerJoin(thread, eq(thread.id, routineExecution.threadId)) .where( @@ -142,9 +146,9 @@ export const makeView = Effect.gen(function* () { const items = rows.slice(0, page.limit); const oldest = items.at(-1)?.run; return { - items: items.map(({ run, conversationId }) => { + items: items.map(({ run, conversationId, posted }) => { if (!conversationId) throw new Error(`Routine run ${run.id} is in no conversation`); - return toRoutineExecution(run, conversationId); + return toRoutineExecution(run, conversationId, posted); }), nextCursor: rows.length > page.limit && oldest diff --git a/packages/core/src/conversations/routines/routines.ts b/packages/core/src/conversations/routines/routines.ts index 8296e388..b0e79d1f 100644 --- a/packages/core/src/conversations/routines/routines.ts +++ b/packages/core/src/conversations/routines/routines.ts @@ -626,6 +626,7 @@ export const connectionEventsLayer = Layer.effect(ConnectionEvents, makeConnecti export { latestRunFailed, latestRunFailedUnseen, + resultPosted, routineExecutionIdOf, toRoutineExecution, } from "./execution.ts"; diff --git a/packages/core/src/conversations/thread-view.ts b/packages/core/src/conversations/thread-view.ts index 396652eb..0ec8ec80 100644 --- a/packages/core/src/conversations/thread-view.ts +++ b/packages/core/src/conversations/thread-view.ts @@ -29,7 +29,12 @@ import { isUuid } from "../ids/ids.ts"; import type { UserFacing } from "../user-message.ts"; import { compactionLineTokens } from "./compaction/window.ts"; import { type CursorPoint, decodeCursor, earlierThan, encodeCursor } from "./cursor.ts"; -import { latestRunFailed, routineExecutionIdOf, toRoutineExecution } from "./routines/routines.ts"; +import { + latestRunFailed, + resultPosted, + routineExecutionIdOf, + toRoutineExecution, +} from "./routines/routines.ts"; import { agentColumns, authorRow, @@ -217,7 +222,7 @@ const loadConversation = Effect.fn("ThreadView.loadConversation")(function* ( columns: { readThrough: true, updatedAt: true }, with: { user: personColumns }, }, - routineExecution: true, + routineExecution: { extras: { posted: (row) => resultPosted(sql`${row.id}`) } }, origin: { with: messageRelations }, messages: { // One more than the page, so an older page is known to exist without a count. @@ -262,7 +267,7 @@ function toThreadDetails( // A run's thread is always opened in a conversation. routineExecution: row.routineExecution && row.conversationId - ? toRoutineExecution(row.routineExecution, row.conversationId) + ? toRoutineExecution(row.routineExecution, row.conversationId, row.routineExecution.posted) : null, startedFrom: row.origin && messageFromRelations(row.origin), participants: row.participants.map(({ user: person, agent: participant }) => diff --git a/packages/core/src/conversations/threads/repository.ts b/packages/core/src/conversations/threads/repository.ts index f02f96ff..2c758d71 100644 --- a/packages/core/src/conversations/threads/repository.ts +++ b/packages/core/src/conversations/threads/repository.ts @@ -232,7 +232,8 @@ export interface Interface { readonly markRead: (userId: string, threadId: string) => Effect.Effect; /** * Records that `userId` has read a routine run or collaboration as an item, - * by its own thread `itemThreadId`: that thread up to when the item began. + * by its own thread `itemThreadId`: that thread up to when the item began, + * or up to when a run ended once it has. * The conversation it is in, and what was written in the item since, stay * as they were. Tells the item's watchers when that moved them on. Never * moves anyone back, and does nothing for any other thread. @@ -1133,8 +1134,11 @@ export const make = Effect.gen(function* () { Effect.gen(function* () { // As text, which keeps the microseconds a `Date` would drop. const [item] = yield* query((db) => { - const runStart = db - .select({ at: sql`${routineExecution.acceptedAt}` }) + // Through when a run ended, so one opened while it ran is news again once it ends. + const runEnd = db + .select({ + at: sql`coalesce(${routineExecution.finishedAt}, ${routineExecution.acceptedAt})`, + }) .from(routineExecution) .where(eq(routineExecution.threadId, itemThreadId)); const collaborationStart = db @@ -1143,7 +1147,7 @@ export const make = Effect.gen(function* () { .where(eq(collaboration.childThreadId, itemThreadId)); return db .select({ - at: sql`coalesce((${runStart}), (${collaborationStart}))::text`, + at: sql`coalesce((${runEnd}), (${collaborationStart}))::text`, }) .from(thread) .where(eq(thread.id, itemThreadId)) diff --git a/packages/core/src/notifications/notifications.test.ts b/packages/core/src/notifications/notifications.test.ts index 453b44ec..fa1d9f14 100644 --- a/packages/core/src/notifications/notifications.test.ts +++ b/packages/core/src/notifications/notifications.test.ts @@ -1,5 +1,6 @@ import { type Channel, + conversationNotificationTag, handleFromName, type Message, type NotificationPreferences, @@ -7,6 +8,7 @@ import { type StreamEvent, } from "@sugabots/contracts"; import { type UserText, userText } from "@sugabots/errors"; +import { eq } from "drizzle-orm"; import { Effect, Exit, Layer, Scope } from "effect"; import { afterAll, beforeEach, describe, expect, it, vi } from "vitest"; import { CurrentActor } from "../authorization/current-actor.ts"; @@ -746,6 +748,95 @@ describe.skipIf(!process.env.DATABASE_URL)("notifications, against Postgres", () expect(sentTo(people.ada.id)).toEqual([]); }); + it("tells those a routine's result mentions in its routine's thread, unless it went to a pod's chat as well", async () => { + const runThread = await routineRunElsewhere({ state: "completed" }, people.bob.id); + const [run] = await onDatabase((db) => + db + .select({ id: routineExecution.id, routineId: routineExecution.routineId }) + .from(routineExecution) + .where(eq(routineExecution.threadId, runThread)), + ); + if (!run) throw new Error("fixture"); + const feed = crypto.randomUUID(); + await onDatabase((db) => + db.insert(thread).values({ + id: feed, + workspaceId, + podId, + hostAgentId: botId, + routineId: run.routineId, + type: "routine_feed", + title: "Digest", + }), + ); + await chooseLevel(people.cy.id, "all"); + const content = `@${handleFromName(people.ada.name)} the calendar needs reconnecting.`; + const resultPosted = async (threadId: string) => { + const id = crypto.randomUUID(); + await onDatabase((db) => + db.insert(message).values({ + id, + threadId, + authorAgentId: botId, + routineExecutionId: run.id, + kind: "text", + status: "complete", + parts: [{ type: "text", text: content }], + content, + }), + ); + return ConversationEvent.MessagePosted({ + threadId, + workspaceId, + podId, + message: { + id, + content, + parts: [{ type: "text", text: content }], + author: { + kind: "agent", + id: botId, + name: "Scout", + handle: "scout", + color: "green", + face: "pill", + shape: "circle", + }, + routineResultOf: { + executionId: run.id, + threadId: runThread, + routineId: run.routineId, + routineName: "Digest", + }, + } as Message, + }); + }; + + const inFeed = await resultPosted(feed); + await emit(inFeed); + + expect(await deliveredTo(people.ada.id)).toEqual([ + { + title: "Scout mentioned you in Digest", + body: content, + tag: conversationNotificationTag(feed), + path: `/${workspaceSlug}/activity/${podSlug}/${feed}?threadPage=${runThread}`, + conversationId: feed, + }, + ]); + // Everything is what is said in chats; a routine's thread speaks up only for those it mentions. + expect(sentTo(people.cy.id)).toEqual([]); + + published.length = 0; + const chat = await podChat(); + const inChat = await resultPosted(chat); + await emit(inFeed, inChat); + + expect(await deliveredTo(people.ada.id)).toEqual([ + expect.objectContaining({ tag: conversationNotificationTag(chat) }), + ]); + }); + it("tells the person a bot was answering that its reply failed, unless they turned that off", async () => { const conversation = await conversationWithBot([people.ada.id, people.bob.id]); const failedTurn = async (asker: { id: string }) => { diff --git a/packages/core/src/notifications/notifications.ts b/packages/core/src/notifications/notifications.ts index 69e7a349..b0fb29a3 100644 --- a/packages/core/src/notifications/notifications.ts +++ b/packages/core/src/notifications/notifications.ts @@ -323,6 +323,7 @@ const notificationsFor = Effect.fn("Notifications.notificationsFor")(function* ( event.message.replyTo?.author.kind === "person" ? event.message.replyTo.author.id : null, + resultOf: event.message.routineResultOf ?? null, }, now, ); diff --git a/packages/core/src/notifications/recipients.ts b/packages/core/src/notifications/recipients.ts index 03e1168f..497cf0ff 100644 --- a/packages/core/src/notifications/recipients.ts +++ b/packages/core/src/notifications/recipients.ts @@ -10,7 +10,7 @@ import { type RoutineRunNotifications, } from "@sugabots/contracts"; import type { UserText } from "@sugabots/errors"; -import { type AnyColumn, and, eq, type SQL, sql } from "drizzle-orm"; +import { type AnyColumn, and, eq, ne, type SQL, sql } from "drizzle-orm"; import { alias } from "drizzle-orm/pg-core"; import { Effect } from "effect"; import { mayDecideApprovals, podStandingFor, reachesPodFor } from "../authorization/access.ts"; @@ -118,12 +118,17 @@ export interface WrittenMessage { artifactTitles: readonly string[]; /** The person whose message it answers, who hears of it as of a reply in Slack. */ repliesTo: string | null; + /** The routine run it is the result of, and that run's own thread, when it is one. */ + resultOf: { executionId: string; threadId: string } | null; } /** * The notifications a finished message sends. Only conversations, pods' * chats and bots' channels notify: routine runs, collaborations and system agents' threads are - * bots' work, which reaches people through what they post in those. + * bots' work, which reaches people through what they post in those. A + * routine's result in its routine's thread notifies those it mentions, which is + * how its bot asks for somebody, unless the result went to a pod's chat as + * well, which notifies instead. */ export const forMessage = Effect.fn("Notifications.forMessage")(function* ( db: Executor, @@ -136,7 +141,12 @@ export const forMessage = Effect.fn("Notifications.forMessage")(function* ( sharedPreview(written.artifactTitles, "artifacts"); if (!preview) return []; const place = yield* placeOf(db, written.threadId); - if (!place || !NOTIFYING_THREAD_TYPES.has(place.type)) return []; + if (!place) return []; + const routineResult = place.type === "routine_feed" ? written.resultOf : null; + if (!routineResult && !NOTIFYING_THREAD_TYPES.has(place.type)) return []; + if (routineResult && (yield* postedElsewhere(db, routineResult.executionId, written.threadId))) { + return []; + } const mentioned = new Set(mentionedHandles(written.content)); const { author, authorName } = written; // Only a person's `@here` or `@channel`: a bot's would let whatever it read or was asked to say page the whole pod. @@ -154,6 +164,7 @@ export const forMessage = Effect.fn("Notifications.forMessage")(function* ( : person.participant && place.type === "conversation" ? "participant" : "activity"; + if (routineResult && reason !== "mention") return []; if (!wants(person, reason, now)) return []; // A bot's channel is named for it, so what the bot says there needs no "in". const where = @@ -168,7 +179,11 @@ export const forMessage = Effect.fn("Notifications.forMessage")(function* ( reason === "mention" ? `${authorName} mentioned you${where}` : `${authorName}${where}`, body: preview, tag: conversationNotificationTag(place.conversationId), - path: activityPath(place, place.conversationId, { message: written.messageId }), + path: activityPath( + place, + place.conversationId, + routineResult ? { threadPage: routineResult.threadId } : { message: written.messageId }, + ), conversationId: place.conversationId, }, }, @@ -176,6 +191,20 @@ export const forMessage = Effect.fn("Notifications.forMessage")(function* ( }); }); +/** Whether the run `executionId` posted its result in a thread besides `threadId`. */ +const postedElsewhere = Effect.fn("Notifications.postedElsewhere")(function* ( + db: Executor, + executionId: string, + threadId: string, +) { + const [elsewhere] = yield* db + .select({ id: message.id }) + .from(message) + .where(and(eq(message.routineExecutionId, executionId), ne(message.threadId, threadId))) + .limit(1); + return elsewhere !== undefined; +}); + /** The notifications a bot's finished reply sends, as a reply to the person whose message it answers. */ export const forReply = Effect.fn("Notifications.forReply")(function* ( db: Executor, @@ -195,6 +224,7 @@ export const forReply = Effect.fn("Notifications.forReply")(function* ( fileNames: [], artifactTitles: [], repliesTo: answering.userId, + resultOf: null, }, now, ); diff --git a/packages/web/src/lib/chats.ts b/packages/web/src/lib/chats.ts index 6e7089c1..1bc8bbeb 100644 --- a/packages/web/src/lib/chats.ts +++ b/packages/web/src/lib/chats.ts @@ -1,4 +1,5 @@ import { + type ActivityItem, type Agent, type ChatMessageItem, type FilePart, @@ -117,7 +118,7 @@ export function useApprovalInbox() { /** How much in the workspace is waiting for the person. */ export type AttentionCounts = { - /** Messages that mention the person and that they have not read. */ + /** Messages and routine results that mention the person and that they have not read. */ unreadMentions: number; /** Everything in Inbox the person has not read: mentions, messages, runs and collaborations. */ unreadInInbox: number; @@ -134,13 +135,17 @@ export function useAttentionCounts(): AttentionCounts { const { data: approvals } = useApprovalInbox(); const { data: activity } = useActivityFeed(); return { - unreadMentions: - activity?.items.filter((item) => item.kind === "mention" && item.unread).length ?? 0, + unreadMentions: activity?.items.filter((item) => item.unread && mentionsYou(item)).length ?? 0, unreadInInbox: activity?.items.filter((item) => item.unread).length ?? 0, waitingApprovals: approvals?.waiting.length ?? 0, }; } +/** Whether an Inbox item names the person: a mention, or a routine run whose result asks for them. */ +function mentionsYou(item: ActivityItem): boolean { + return item.kind === "mention" || (item.kind === "routine" && item.mentionsYou); +} + /** * Records that the person has read the conversation up to its newest message, and * clears its dot and its count in the sidebar. diff --git a/packages/web/src/screens/ActivityPane.stories.tsx b/packages/web/src/screens/ActivityPane.stories.tsx index ad405717..d48ecbcb 100644 --- a/packages/web/src/screens/ActivityPane.stories.tsx +++ b/packages/web/src/screens/ActivityPane.stories.tsx @@ -97,6 +97,7 @@ const run: ThreadDetails = { }, state: "completed", error: null, + posted: true, acceptedAt: "2026-09-18T06:00:00.000Z", startedAt: "2026-09-18T06:00:01.000Z", finishedAt: "2026-09-18T06:02:00.000Z", diff --git a/packages/web/src/screens/ChatThreadPanel.stories.tsx b/packages/web/src/screens/ChatThreadPanel.stories.tsx index 991d2a55..2a549364 100644 --- a/packages/web/src/screens/ChatThreadPanel.stories.tsx +++ b/packages/web/src/screens/ChatThreadPanel.stories.tsx @@ -139,6 +139,7 @@ function routineRun(failed: boolean): ThreadDetails { }, state: failed ? "failed" : "completed", error: failed ? "HubSpot did not answer within 30 seconds." : null, + posted: !failed, acceptedAt: "2026-09-18T06:00:00.000Z", startedAt: "2026-09-18T06:00:01.000Z", finishedAt: "2026-09-18T06:02:00.000Z", diff --git a/packages/web/src/screens/RoutineRun.tsx b/packages/web/src/screens/RoutineRun.tsx index 784ffe72..a2b4b764 100644 --- a/packages/web/src/screens/RoutineRun.tsx +++ b/packages/web/src/screens/RoutineRun.tsx @@ -29,7 +29,8 @@ type RunItem = Extract; /** How a run went, or that its routine is paused, as the dot beside it says it. */ export type RunOutcome = - | "passed" + | "posted" + | "nothingToReport" | "failed" | "running" | "cancelled" @@ -37,7 +38,8 @@ export type RunOutcome = | "pausedAfterFailures"; const OUTCOME_LABEL: Record = { - passed: "Passed", + posted: "Posted", + nothingToReport: "Nothing to report", failed: "Failed", running: "Running", cancelled: "Cancelled", @@ -46,7 +48,8 @@ const OUTCOME_LABEL: Record = { }; const OUTCOME_DOT: Record = { - passed: "bg-success", + posted: "bg-success", + nothingToReport: "bg-disabled-foreground", failed: "bg-destructive-text", running: "bg-subtle-foreground", cancelled: "bg-disabled-foreground", @@ -54,11 +57,15 @@ const OUTCOME_DOT: Record = { pausedAfterFailures: "bg-destructive-text", }; -/** A run's state, as the dot beside it says it. */ +/** + * How a run in `state` went, as the dot beside it says it. One that completed + * without having `posted` a result had nothing to report. + */ export function outcomeOf( state: RunItem["state"], + posted: boolean, ): Exclude { - if (state === "completed") return "passed"; + if (state === "completed") return posted ? "posted" : "nothingToReport"; if (state === "failed") return "failed"; if (state === "cancelled") return "cancelled"; return "running"; @@ -118,9 +125,10 @@ function Divider({ children }: { children: string }) { /** * A run that posted no result, drawn as its bot's message: what the bot last * said in it, or, when it said nothing, how it went and, when it failed, why. - * A run that passed and said something needs no word on how it went. The - * latest failure, while the routine is still failing, offers to run it again - * or to see to the connection it needs. One still running says so with dots. + * A run that completed and said something needs no word on how it went, and + * one that completed saying nothing had nothing to report. The latest + * failure, while the routine is still failing, offers to run it again or to + * see to the connection it needs. One still running says so with dots. */ export function RunMessage({ item, @@ -142,8 +150,8 @@ export function RunMessage({ mentionable: ThreadParticipant[]; }) { const controls = useRoutineControls(conversationId, routine); - const outcome = outcomeOf(item.state); - const report = outcome === "passed" ? item.report : null; + const outcome = outcomeOf(item.state, false); + const report = item.state === "completed" ? item.report : null; return (
- ) : outcome === "running" ? ( + ) : outcome === "nothingToReport" ? null : outcome === "running" ? ( diff --git a/packages/web/src/screens/agent-chat.test.tsx b/packages/web/src/screens/agent-chat.test.tsx index 19410efa..df83dc10 100644 --- a/packages/web/src/screens/agent-chat.test.tsx +++ b/packages/web/src/screens/agent-chat.test.tsx @@ -69,6 +69,7 @@ const routineExecution: RoutineExecution = { }, state: "completed", error: null, + posted: true, acceptedAt: "2026-09-18T08:00:00.000Z", startedAt: "2026-09-18T08:00:01.000Z", finishedAt: "2026-09-18T08:30:00.000Z", diff --git a/packages/web/src/shell/Inbox.stories.tsx b/packages/web/src/shell/Inbox.stories.tsx index 3b9a12a6..78128cdc 100644 --- a/packages/web/src/shell/Inbox.stories.tsx +++ b/packages/web/src/shell/Inbox.stories.tsx @@ -57,6 +57,7 @@ const feed: ActivityFeed = { triggerKind: "cron", preview: "Queued 38 leads overnight and six have already replied. The two Northwind accounts both asked for pricing.", + mentionsYou: false, agent: { kind: "agent", id: growthDesk.id, diff --git a/packages/web/src/shell/Inbox.tsx b/packages/web/src/shell/Inbox.tsx index 90474769..e301f432 100644 --- a/packages/web/src/shell/Inbox.tsx +++ b/packages/web/src/shell/Inbox.tsx @@ -302,7 +302,9 @@ function wordsOf(item: ActivityItem): { title: string; detail: string } { }; case "routine": return { - title: routineText(item.routineName, runState(item.state)), + title: item.mentionsYou + ? `${item.routineName} needs you` + : routineText(item.routineName, runState(item.state)), detail: item.preview ?? item.agent.name, }; case "collaboration": diff --git a/packages/web/src/shell/Routines.stories.tsx b/packages/web/src/shell/Routines.stories.tsx index 6aa70de1..9a82990f 100644 --- a/packages/web/src/shell/Routines.stories.tsx +++ b/packages/web/src/shell/Routines.stories.tsx @@ -131,6 +131,7 @@ const outboundFailure: RoutineExecution = { finishedAt: "2026-09-18T06:00:12.000Z", state: "failed", error: "Gmail refused the connection: the token has expired.", + posted: false, }; const chatList: ChatList = { @@ -234,7 +235,7 @@ function routineHandlers() { triggerKind: "cron", state: "completed", error: null, - report: "Queued 38 leads. Nobody replied overnight.", + report: null, createdAt: "2026-09-16T06:00:00.000Z", }, posted( @@ -303,9 +304,8 @@ export const Default = meta.story({ await userEvent.click(await canvas.findByRole("button", { name: "Details" })); await expect(await canvas.findByRole("button", { name: "Run now" })).toBeVisible(); const log = await canvas.findByRole("log", { name: "Chat messages" }); - const quiet = await within(log).findByRole("article", { name: /: Passed$/ }); - await expect(quiet).toHaveTextContent("Queued 38 leads. Nobody replied overnight."); - await expect(quiet).not.toHaveTextContent("Passed"); + const quiet = await within(log).findByRole("article", { name: /: Nothing to report$/ }); + await expect(quiet).toHaveTextContent("Nothing to report"); await expect(within(log).getByRole("article", { name: /: Failed$/ })).toHaveTextContent( "Gmail refused the connection", ); @@ -451,7 +451,7 @@ export const BesideTheBotsChat = meta.story({ expect( canvas .getAllByRole("log", { name: "Chat messages" }) - .some((log) => log.textContent?.includes("Queued 38 leads. Nobody replied overnight.")), + .some((log) => log.textContent?.includes("Queued 41 leads. Three replied overnight")), ).toBe(true), ); await userEvent.click( diff --git a/packages/web/src/shell/Routines.tsx b/packages/web/src/shell/Routines.tsx index 85107499..705e68ad 100644 --- a/packages/web/src/shell/Routines.tsx +++ b/packages/web/src/shell/Routines.tsx @@ -213,7 +213,7 @@ export function RoutineRow({ ? row.routine.pausedAfterFailures ? "pausedAfterFailures" : "paused" - : lastRun && outcomeOf(lastRun.state); + : lastRun && outcomeOf(lastRun.state, lastRun.posted); const schedule = routine && scheduleLabel(routine); return useRender({ render: render ?? ,