mark unread messages in Chat, count them in the sidebar and notify on the desktop

packages/api/src/harness/fake-store.ts

@@ -26,6 +26,7 @@ export function fakeStore(): FakeStore {
 		getAgentTask: mock<HarnessStore['getAgentTask']>(),
 		createAgentTask: mock<HarnessStore['createAgentTask']>(),
 		listPersonas: mock<HarnessStore['listPersonas']>(),
+		listAnswers: mock<HarnessStore['listAnswers']>(),
 		ensureFirstPersona: mock<HarnessStore['ensureFirstPersona']>(),
 		getPersona: mock<HarnessStore['getPersona']>(),
 		createPersona: mock<HarnessStore['createPersona']>(),
@@ -36,6 +37,7 @@ export function fakeStore(): FakeStore {
 		listShownEntries: mock<HarnessStore['listShownEntries']>(),
 		listOpenJobs: mock<HarnessStore['listOpenJobs']>(),
 		appendPersonMessage: mock<HarnessStore['appendPersonMessage']>(),
+		markThreadRead: mock<HarnessStore['markThreadRead']>(),
 		completeSetAlert: mock<HarnessStore['completeSetAlert']>(),
 		deleteAlert: mock<HarnessStore['deleteAlert']>(),
 		listThreadAlerts: mock<HarnessStore['listThreadAlerts']>(),

packages/api/src/harness/route.test.ts

@@ -76,10 +76,11 @@ beforeEach(async () => {
 
 describe('the chat routes', () => {
 	test('are refused without the experiment', async () => {
-		const {handleListPersonasRoute, handlePostMessageRoute} = routes;
+		const {handleListPersonasRoute, handleListAnswersRoute, handlePostMessageRoute} = routes;
 		requireViewer.mockResolvedValue(Ok({...viewer, experiments: []}));
 
 		expect((await handleListPersonasRoute(request())).unwrapErr().kind).toBe('forbidden');
+		expect((await handleListAnswersRoute(request())).unwrapErr().kind).toBe('forbidden');
 		expect((await handlePostMessageRoute(request({personaId: 'persona-1'}, {text: 'Hi'}))).unwrapErr().kind).toBe('forbidden');
 		expect(store.appendPersonMessage).not.toHaveBeenCalled();
 	});
@@ -110,6 +111,17 @@ describe('the chat routes', () => {
 		});
 	});
 
+	test('list the persona’s last answer in each of the viewer’s conversations, on one line, and whether it is unread, making nothing', async () => {
+		const {handleListAnswersRoute} = routes;
+		store.listAnswers.mockResolvedValue(Ok([{personaId: 'persona-1', personaName: 'Finance', answer: {id: 'e9', text: 'Your vault\n has  one entry.'}, unread: true}]));
+
+		const listed = (await handleListAnswersRoute(request())).unwrap();
+
+		expect(store.listAnswers).toHaveBeenCalledWith('org-1', 'user-1');
+		expect(store.ensureFirstPersona).not.toHaveBeenCalled();
+		expect(listed).toEqual({answers: [{personaId: 'persona-1', name: 'Finance', answer: {id: 'e9', text: 'Your vault has one entry.'}, unread: true}]});
+	});
+
 	test('an organization without a persona gets a General one, nobody’s, on its first listing', async () => {
 		const {handleListPersonasRoute} = routes;
 		store.listPersonas.mockResolvedValue(Ok([]));
@@ -234,4 +246,26 @@ describe('the chat routes', () => {
 		expect((await handlePostMessageRoute(request({personaId: 'persona-1'}, {text: '   '}))).unwrapErr().kind).toBe('parse_error');
 		expect((await handlePostMessageRoute(request({personaId: 'persona-1'}, {text: 'x'.repeat(20_001)}))).unwrapErr().kind).toBe('parse_error');
 	});
+
+	test('mark the viewer’s conversation with the persona read up to the answer shown, beginning none, and say where the answer, the conversation or the persona is gone', async () => {
+		const {handleMarkReadRoute} = routes;
+		store.markThreadRead.mockResolvedValue(Ok(true));
+
+		expect((await handleMarkReadRoute(request({personaId: 'persona-1'}, {entryId: 'e9'}))).isOk()).toBe(true);
+		expect(store.markThreadRead).toHaveBeenCalledWith('thread-1', 'e9');
+
+		expect((await handleMarkReadRoute(request({personaId: 'persona-1'}, {}))).unwrapErr().kind).toBe('parse_error');
+
+		store.markThreadRead.mockResolvedValue(Ok(false));
+		expect((await handleMarkReadRoute(request({personaId: 'persona-1'}, {entryId: 'elsewhere'}))).unwrapErr().kind).toBe('not_found');
+
+		store.getMemberThread.mockResolvedValue(Ok(null));
+		expect((await handleMarkReadRoute(request({personaId: 'persona-1'}, {entryId: 'e9'}))).unwrapErr().kind).toBe('not_found');
+		expect(store.ensureMemberThread).not.toHaveBeenCalled();
+		expect(ensureBuiltInAgent).not.toHaveBeenCalled();
+
+		store.getPersona.mockResolvedValue(Ok(null));
+		expect((await handleMarkReadRoute(request({personaId: 'persona-9'}, {entryId: 'e9'}))).unwrapErr().kind).toBe('not_found');
+		expect(store.markThreadRead).toHaveBeenCalledTimes(2);
+	});
 });

packages/api/src/harness/route.ts

@@ -6,7 +6,9 @@ import {preview, threadState, toAvatarColor} from '@proxy/utils/harness';
 import type {EntryRow, LastMessage, Persona, PersonaRow, ThreadRow} from '@proxy/db/harness';
 import {
 	type ChatEntry,
+	type chatAnswersSchema,
 	type ChatPersona,
+	markReadBodySchema,
 	personaBodySchema,
 	type personaAnswerSchema,
 	type personaListSchema,
@@ -32,6 +34,7 @@ import type {HarnessStore} from './store';
 export type ChatStore = Pick<
 	HarnessStore,
 	| 'listPersonas'
+	| 'listAnswers'
 	| 'ensureFirstPersona'
 	| 'getPersona'
 	| 'createPersona'
@@ -42,6 +45,7 @@ export type ChatStore = Pick<
 	| 'listShownEntries'
 	| 'listOpenJobs'
 	| 'appendPersonMessage'
+	| 'markThreadRead'
 >;
 
 // Whether the viewer may rename, reface, switch the model of or delete the persona: whoever made it, or the organization's owner (utils/manages.ts).
@@ -133,6 +137,22 @@ function handleListPersonasRoute(store: ChatStore, request: AuthenticatedRequest
 	});
 }
 
+/** `GET /api/orgs/:org/chat/answers`: the persona's last answer in each of the viewer's conversations that has one, and whether they haven't seen it; read only, for every page to ask every so often. */
+function handleListAnswersRoute(store: ChatStore, request: AuthenticatedRequest): Promise<Result<z.infer<typeof chatAnswersSchema>, ApiError>> {
+	return Do(async ($) => {
+		const viewer = $(await requireChat(request));
+		const answers = $(await store.listAnswers(viewer.orgId, viewer.userId));
+		return {
+			answers: answers.map(({personaId, personaName, answer, unread}) => ({
+				personaId,
+				name: personaName,
+				answer: {id: answer.id, text: preview(answer.text)},
+				unread,
+			})),
+		};
+	});
+}
+
 /** `POST /api/orgs/:org/chat/personas` with `{name, look, color, chatModel?}`: a new persona, every member's, and the organization's built-in login with it where there is none yet. */
 function handleCreatePersonaRoute(store: ChatStore, request: AuthenticatedRequest): Promise<Result<z.infer<typeof personaAnswerSchema>, ApiError>> {
 	return Do(async ($) => {
@@ -198,14 +218,30 @@ function handlePostMessageRoute(store: ChatStore, request: AuthenticatedRequest)
 	});
 }
 
+/** `POST /api/orgs/:org/chat/personas/:personaId/read` with `{entryId}`: the viewer was shown their conversation with the persona up to that answer; what the persona said after it stays unread. */
+function handleMarkReadRoute(store: ChatStore, request: AuthenticatedRequest): Promise<Result<void, ApiError>> {
+	return Do(async ($) => {
+		const viewer = $(await requireChat(request));
+		const {entryId} = $(parseSchema(markReadBodySchema, request.body));
+		const persona = $(await requirePersona(store, viewer, request));
+		// Read, never begun: a viewer without a conversation has no answer in it to have seen.
+		const thread = $(await store.getMemberThread(persona.id, viewer.userId));
+		if (!thread || !$(await store.markThreadRead(thread.id, entryId))) {
+			return $(Err(ApiErr.notFound('entry', entryId)));
+		}
+	});
+}
+
 /** The Chat page's routes, on the store. */
 export function chatRoutes(store: ChatStore) {
 	return {
 		handleListPersonasRoute: (request: AuthenticatedRequest) => handleListPersonasRoute(store, request),
+		handleListAnswersRoute: (request: AuthenticatedRequest) => handleListAnswersRoute(store, request),
 		handleCreatePersonaRoute: (request: AuthenticatedRequest) => handleCreatePersonaRoute(store, request),
 		handleUpdatePersonaRoute: (request: AuthenticatedRequest) => handleUpdatePersonaRoute(store, request),
 		handleDeletePersonaRoute: (request: AuthenticatedRequest) => handleDeletePersonaRoute(store, request),
 		handleGetThreadRoute: (request: AuthenticatedRequest) => handleGetThreadRoute(store, request),
 		handlePostMessageRoute: (request: AuthenticatedRequest) => handlePostMessageRoute(store, request),
+		handleMarkReadRoute: (request: AuthenticatedRequest) => handleMarkReadRoute(store, request),
 	};
 }

packages/api/src/harness/store.ts

@@ -6,7 +6,7 @@ import * as alerts from '@proxy/db/alerts';
 import type {AlertRow, NewAlert, PollRow, WatchMatch} from '@proxy/db/alerts';
 import {type ConnectionRow, getConnectionWithCredentialById, listConnectionsByIds} from '@proxy/db/connection';
 import * as harness from '@proxy/db/harness';
-import type {AfterAnswer, AssistantTurn, EntryRow, FailureNote, JobRow, LastMessage, Persona, PersonaRow, SubAgentRow, ThreadRow} from '@proxy/db/harness';
+import type {AfterAnswer, AssistantTurn, ConversationAnswer, EntryRow, FailureNote, JobRow, LastMessage, Persona, PersonaRow, SubAgentRow, ThreadRow} from '@proxy/db/harness';
 import {type AgentTaskRow, createAgentTask, getAgentTask, type NewAgentTask} from '@proxy/db/task';
 import type {SignedInAgent} from '@proxy/db/agent';
 
@@ -40,6 +40,7 @@ export type HarnessStore = {
 	createAgentTask: (task: NewAgentTask) => Promise<Result<AgentTaskRow | null, ApiError>>;
 
 	listPersonas: (orgId: string, userId: string) => Promise<Result<Array<PersonaRow & {lastMessage: LastMessage | null}>, ApiError>>;
+	listAnswers: (orgId: string, userId: string) => Promise<Result<ConversationAnswer[], ApiError>>;
 	ensureFirstPersona: (orgId: string) => Promise<Result<PersonaRow, ApiError>>;
 	getPersona: (orgId: string, personaId: string) => Promise<Result<PersonaRow | null, ApiError>>;
 	createPersona: (orgId: string, createdById: string, persona: Persona) => Promise<Result<PersonaRow, ApiError>>;
@@ -50,6 +51,7 @@ export type HarnessStore = {
 	listShownEntries: (threadId: string) => Promise<Result<EntryRow[], ApiError>>;
 	listOpenJobs: (threadId: string) => Promise<Result<JobRow[], ApiError>>;
 	appendPersonMessage: (threadId: string, text: string) => Promise<Result<EntryRow | null, ApiError>>;
+	markThreadRead: (threadId: string, entryId: string) => Promise<Result<boolean, ApiError>>;
 
 	completeSetAlert: (job: {jobId: string; workerId: string; threadId: string; callId: string}, alert: NewAlert, answer: (alert: AlertRow) => string) => Promise<Result<boolean, ApiError>>;
 	deleteAlert: (alertId: string, owner: {threadId: string} | {orgId: string; userId: string}) => Promise<Result<boolean, ApiError>>;
@@ -86,6 +88,7 @@ export const harnessStore: HarnessStore = {
 	createAgentTask,
 
 	listPersonas: harness.listPersonas,
+	listAnswers: harness.listAnswers,
 	ensureFirstPersona: harness.ensureFirstPersona,
 	getPersona: harness.getPersona,
 	createPersona: harness.createPersona,
@@ -96,6 +99,7 @@ export const harnessStore: HarnessStore = {
 	listShownEntries: harness.listShownEntries,
 	listOpenJobs: harness.listOpenJobs,
 	appendPersonMessage: (threadId, text) => harness.appendPersonMessage(threadId, text, readyForInference),
+	markThreadRead: harness.markThreadRead,
 
 	completeSetAlert: (job, alert, answer) => alerts.completeSetAlert(job, alert, answer, readyForInference),
 	deleteAlert: alerts.deleteAlert,

packages/api/src/server/create-api-router.ts

@@ -214,11 +214,13 @@ export function createApiRouter(): express.Router {
 	// message queues its work.
 	const chat = chatRoutes(harnessStore);
 	org.get('/chat/personas', withAuthResult('List personas route', chat.handleListPersonasRoute));
+	org.get('/chat/answers', withAuthResult('List answers route', chat.handleListAnswersRoute));
 	org.post('/chat/personas', withAuthResult('Create persona route', chat.handleCreatePersonaRoute));
 	org.put('/chat/personas/:personaId', withAuthResult('Update persona route', chat.handleUpdatePersonaRoute));
 	org.delete('/chat/personas/:personaId', withAuthResult('Delete persona route', chat.handleDeletePersonaRoute));
 	org.get('/chat/personas/:personaId/thread', withAuthResult('Get chat thread route', chat.handleGetThreadRoute));
 	org.post('/chat/personas/:personaId/messages', withAuthResult('Post chat message route', chat.handlePostMessageRoute));
+	org.post('/chat/personas/:personaId/read', withAuthResult('Mark chat read route', chat.handleMarkReadRoute));
 	// The Alerts page: what the agents set in the member's conversations, to be woken outside them.
 	const alerts = alertRoutes(harnessStore);
 	org.get('/alerts', withAuthResult('List alerts route', alerts.handleListAlertsRoute));

packages/contracts/src/harness.ts

@@ -56,6 +56,26 @@ export type ChatPersona = z.infer<typeof chatPersonaSchema>;
 /** `GET chat/personas`: the organization's personas, in the order they were made. */
 export const personaListSchema = z.object({personas: z.array(chatPersonaSchema)});
 
+/**
+ * The persona's last answer in one of the person's conversations, cut to a line, which may come
+ * before what the person said last, and whether they haven't seen it: the page hasn't shown it to
+ * them since it came (`POST chat/personas/:personaId/read`).
+ */
+export const chatAnswerSchema = z.object({
+	personaId: z.string(),
+	name: z.string(),
+	answer: z.object({id: z.string(), text: z.string()}),
+	unread: z.boolean(),
+});
+
+export type ChatAnswer = z.infer<typeof chatAnswerSchema>;
+
+/**
+ * `GET chat/answers`: the last answer in each of the person's conversations that has one, as
+ * every page asks every so often; it makes nothing, where `GET chat/personas` makes General.
+ */
+export const chatAnswersSchema = z.object({answers: z.array(chatAnswerSchema)});
+
 /** The persona made or changed. */
 export const personaAnswerSchema = z.object({persona: chatPersonaSchema});
 
@@ -94,5 +114,11 @@ const MAX_MESSAGE = 20_000;
 /** `POST chat/personas/:personaId/messages`: what the person says to the persona; it is woken. */
 export const postMessageBodySchema = z.object({text: z.string().trim().min(1).max(MAX_MESSAGE)});
 
+/**
+ * `POST chat/personas/:personaId/read`: the persona's answer the page showed the person, which
+ * the conversation is read up to.
+ */
+export const markReadBodySchema = z.object({entryId: z.string().min(1)});
+
 /** What the person said, as a turn of the conversation. */
 export const postedMessageSchema = z.object({entry: chatEntrySchema});

packages/db/prisma/migrations/v046_add_thread_read_cursor/migration.sql

@@ -0,0 +1,9 @@
+-- How far each member has read their conversation: the `seq` of the last entry they have seen.
+-- Every conversation there is counts as read up to its last entry, so nothing said before this
+-- turns unread.
+
+-- AlterTable
+ALTER TABLE "AgentThread" ADD COLUMN     "readSeq" INTEGER NOT NULL DEFAULT 0;
+
+UPDATE "AgentThread" AS t
+SET "readSeq" = COALESCE((SELECT MAX(e."seq") FROM "AgentEntry" AS e WHERE e."threadId" = t."id"), 0);

packages/db/prisma/schema.prisma

@@ -665,6 +665,10 @@ model AgentThread {
   // and made anew when it is taken up again; a conversation has none: its agent makes no calls
   // but to its sub-agents.
   taskId  String?
+  // How far the member has read their conversation: the `seq` of the last answer the Chat page
+  // showed them while they looked at it, never back. What the persona said past it is unread. A
+  // sub-agent's thread, which nobody reads, keeps 0.
+  readSeq Int     @default(0)
 
   createdAt DateTime @default(now())
   updatedAt DateTime @default(now()) @updatedAt

packages/db/src/harness.test.ts

@@ -18,7 +18,9 @@ import {
 	listEntries,
 	listJobEvents,
 	listOpenJobs,
+	listAnswers,
 	listSubAgents,
+	markThreadRead,
 	requeueJob,
 	type JobRow,
 	type NewEntry,
@@ -268,6 +270,59 @@ describe.skipIf(!databaseUrl)('the harness, against the database', () => {
 		]);
 	});
 
+	test('the persona’s last answer is unread until the thread is read up to it, never back, and what the person says after it reads nothing', async () => {
+		const persona = unwrap(await ensureFirstPersona(orgId));
+		const lastAnswer = async () =>
+			unwrap(await listAnswers(orgId, userId)).find(
+				(candidate) => candidate.personaId === persona.id,
+			);
+		const answer = async (workerId: string, text: string) => {
+			const job = unwrap(await claimJob(workerId, 60_000));
+			if (!job) {
+				throw new Error('nothing to claim');
+			}
+			unwrap(
+				await completeInference(
+					job.id,
+					workerId,
+					thread.id,
+					subTurn(text),
+					{plans: {}},
+					readyForInference,
+				),
+			);
+			const last = await lastAnswer();
+			if (last?.answer.text !== text) {
+				throw new Error(`expected the answer ${text}, got ${JSON.stringify(last)}`);
+			}
+			return last.answer.id;
+		};
+
+		unwrap(await appendPersonMessage(thread.id, 'Anything new?', readyForInference));
+		const first = await answer('worker-r', 'Nothing new.');
+		expect((await lastAnswer())?.unread).toBe(true);
+
+		unwrap(await appendPersonMessage(thread.id, 'And now?', readyForInference));
+		expect(await lastAnswer()).toEqual({
+			personaId: persona.id,
+			personaName: persona.name,
+			answer: {id: first, text: 'Nothing new.'},
+			unread: true,
+		});
+
+		const second = await answer('worker-r2', 'Still nothing.');
+		expect(unwrap(await markThreadRead(thread.id, first))).toBe(true);
+		expect((await lastAnswer())?.unread).toBe(true);
+
+		expect(unwrap(await markThreadRead(thread.id, second))).toBe(true);
+		expect((await lastAnswer())?.unread).toBe(false);
+		expect(unwrap(await markThreadRead(thread.id, first))).toBe(true);
+		expect((await lastAnswer())?.unread).toBe(false);
+
+		expect(unwrap(await markThreadRead(thread.id, 'no-such-entry'))).toBe(false);
+		expect(unwrap(await markThreadRead('no-such-thread', second))).toBe(false);
+	});
+
 	test('a finish with nothing written is a done event too', async () => {
 		unwrap(await appendPersonMessage(thread.id, 'Try again', readyForInference));
 		const job = unwrap(await claimJob('worker-e', 60_000));