diff --git a/src/main/app/ipc-contracts/collaboration.ts b/src/main/app/ipc-contracts/collaboration.ts index a187869d..b3167174 100644 --- a/src/main/app/ipc-contracts/collaboration.ts +++ b/src/main/app/ipc-contracts/collaboration.ts @@ -52,5 +52,5 @@ export const collaborationContracts = { }) satisfies z.ZodType, editMessage: z.strictObject({ orgId: id, messageId: id, actorType: sender, actorId: nullableId, bodyMd: z.string() }) satisfies z.ZodType, deleteMessage: z.strictObject({ orgId: id, messageId: id, actorType: sender, actorId: nullableId }) satisfies z.ZodType, - markChatSpaceRead: z.strictObject({ orgId: id, employeeId: id, spaceId: id }) satisfies z.ZodType, + markChatSpaceRead: z.strictObject({ orgId: id, employeeId: id, spaceId: id, upToMessageId: id }) satisfies z.ZodType, }; diff --git a/src/main/services/chat-service.ts b/src/main/services/chat-service.ts index 2c764bc0..2ce28ed4 100644 --- a/src/main/services/chat-service.ts +++ b/src/main/services/chat-service.ts @@ -220,22 +220,19 @@ export class ChatService { const employee = this.db.prepare("select id from employees where id = ? and org_id = ?") .get(input.employeeId, input.orgId) as { id: string } | undefined; if (!employee) throw new Error("Employee was not found."); - // Read state is per-employee bookkeeping, not access control: the board - // observes every conversation in its org (agent↔agent directs included), - // so opening one clears its unread count even without membership. - const latest = this.db.prepare(` - select id from chat_messages - where space_id = ? and org_id = ? - order by created_at desc - limit 1 - `).get(input.spaceId, input.orgId) as { id: string } | undefined; - const now = new Date().toISOString(); + // Read state is per-employee bookkeeping, not access control: the board reads every conversation in its org. + const seen = this.db.prepare("select id, created_at from chat_messages where id = ? and space_id = ? and org_id = ?") + .get(input.upToMessageId, input.spaceId, input.orgId) as { id: string; created_at: string } | undefined; + if (!seen) throw new Error("The message read is not in this conversation."); + // Read up to what was seen, never past it; messages order by time then insertion, and a late mark never moves back. this.db.prepare(` insert into chat_read_state(org_id, employee_id, space_id, last_read_message_id, last_read_at) values (?, ?, ?, ?, ?) on conflict(employee_id, space_id) do update set last_read_message_id = excluded.last_read_message_id, last_read_at = excluded.last_read_at - `).run(input.orgId, input.employeeId, input.spaceId, latest?.id ?? null, now); + where (excluded.last_read_at, (select rowid from chat_messages where id = excluded.last_read_message_id)) + > (chat_read_state.last_read_at, coalesce((select rowid from chat_messages where id = chat_read_state.last_read_message_id), -1)) + `).run(input.orgId, input.employeeId, input.spaceId, seen.id, seen.created_at); } editMessage(input: EditMessageInput): void { diff --git a/src/main/services/snapshot-service.ts b/src/main/services/snapshot-service.ts index 1733f3ed..db6c0d97 100644 --- a/src/main/services/snapshot-service.ts +++ b/src/main/services/snapshot-service.ts @@ -96,11 +96,11 @@ export class SnapshotService { approvals: list(this.db, "approvals", "created_at desc"), chatSpaces: list(this.db, "chat_spaces", "created_at asc"), chatMemberships: list(this.db, "chat_memberships", "space_id asc, principal_type asc, principal_id asc"), - chatMessages: recentList(this.db, "chat_messages", "created_at desc", 500).reverse(), + chatMessages: recentList(this.db, "chat_messages", "created_at desc, rowid desc", 500).reverse(), // Every thread a loaded message belongs to, plus the most recently active ones, so no shown message loses its thread. chatThreads: all(this.db.prepare(` select * from chat_threads - where id in (select thread_id from (select thread_id from chat_messages order by created_at desc limit 500)) + where id in (select thread_id from (select thread_id from chat_messages order by created_at desc, rowid desc limit 500)) or id in (select id from chat_threads order by updated_at desc limit 200) order by updated_at desc `)).map((row) => camel(row) as unknown as AppSnapshot["chatThreads"][number]), diff --git a/src/renderer/src/pages/ChatPage.tsx b/src/renderer/src/pages/ChatPage.tsx index df292752..0f14cb25 100644 --- a/src/renderer/src/pages/ChatPage.tsx +++ b/src/renderer/src/pages/ChatPage.tsx @@ -1,4 +1,4 @@ -import { useCallback, useEffect, useMemo, useRef, useState, type FormEvent } from "react"; +import { useCallback, useMemo, useRef, useState, type FormEvent } from "react"; import { Field } from "../shared/ui"; import { Dialog } from "../shared/Dialog"; import { t } from "../i18n/i18n"; @@ -90,21 +90,10 @@ export function ChatPage({ snapshot, actions }: PageProps) { // whichever space just got a message and marking it read unseen. if (!selectedSpaceId && firstAvailableSpaceId) setSelectedSpaceId(firstAvailableSpaceId); const activeSpace = spaces.find((space) => space.id === activeSpaceId) ?? null; - const latestActiveMessageId = useMemo(() => { - if (!activeSpaceId) return null; - const inSpace = messagesAll.filter((message) => message.spaceId === activeSpaceId); - if (inSpace.length === 0) return null; - return inSpace[inSpace.length - 1].id; - }, [messagesAll, activeSpaceId]); - - useEffect(() => { - if (!activeSpaceId || !latestActiveMessageId) return; - void actions.run(() => window.agentCompany.markChatSpaceRead({ - orgId, - employeeId: employee.id, - spaceId: activeSpaceId, - })); - }, [activeSpaceId, latestActiveMessageId, orgId, employee.id, actions]); + // The conversation reports what the reader actually has in view; only that counts as read. + const markRead = useCallback((spaceId: string, upToMessageId: string) => { + void actions.run(() => window.agentCompany.markChatSpaceRead({ orgId, employeeId: employee.id, spaceId, upToMessageId })); + }, [actions, orgId, employee.id]); const messagesInSpace = useMemo( () => (activeSpace ? messagesAll.filter((m) => m.spaceId === activeSpace.id) : []), @@ -271,6 +260,7 @@ export function ChatPage({ snapshot, actions }: PageProps) { onDeleteMessage={deleteChatMessage} onOpenFile={openFile} onSubmit={sendInChannel} + onRead={markRead} onReplyInThread={replyInThread} send={(activeSpaceId ? sends[activeSpaceId] : undefined) ?? IDLE_SEND} composerRef={composerRef} diff --git a/src/renderer/src/pages/chat/ChatConversation.tsx b/src/renderer/src/pages/chat/ChatConversation.tsx index 8ff84478..30e9a059 100644 --- a/src/renderer/src/pages/chat/ChatConversation.tsx +++ b/src/renderer/src/pages/chat/ChatConversation.tsx @@ -1,4 +1,4 @@ -import { useLayoutEffect, useMemo, useRef, useState, type FormEvent, type MutableRefObject } from "react"; +import { useEffect, useLayoutEffect, useMemo, useRef, useState, type FormEvent, type MutableRefObject } from "react"; import { AvatarStack, Icon, IconButton } from "../../shared/ui"; import type { ApprovalRecord, @@ -40,6 +40,7 @@ export function ChatConversation({ onDeleteMessage, onOpenFile, onSubmit, + onRead, onReplyInThread, send, composerRef, @@ -66,6 +67,8 @@ export function ChatConversation({ onDeleteMessage: (messageId: string) => void; onOpenFile: (fileId: string) => void; onSubmit: (event: FormEvent) => void; + /** Marks the conversation read up to a message the reader has in view; must be stable. */ + onRead: (spaceId: string, upToMessageId: string) => void; onReplyInThread: (threadId: string, body: string) => Promise; send: SendState; composerRef: MutableRefObject; @@ -112,6 +115,20 @@ export function ChatConversation({ if (atBottomRef.current) jumpToLatest("auto"); }, [lastMessageId]); + // Read means seen: the newest message is in view, the window is visible, and no search hides messages. + const newestId = messagesInSpace.length > 0 ? messagesInSpace[messagesInSpace.length - 1].id : null; + const spaceId = space?.id ?? null; + const searching = trimmed.length > 0; + useEffect(() => { + const markIfSeen = () => { + // The DOM is measured, not the last scroll state: clearing a search can grow the list under a reader who was at the bottom. + if (spaceId && newestId && !searching && measureAtBottom() && document.visibilityState === "visible") onRead(spaceId, newestId); + }; + markIfSeen(); + document.addEventListener("visibilitychange", markIfSeen); + return () => document.removeEventListener("visibilitychange", markIfSeen); + }, [spaceId, newestId, atBottom, searching, onRead]); + return ( <>
diff --git a/src/renderer/src/pages/chat/chat-helpers.ts b/src/renderer/src/pages/chat/chat-helpers.ts index f8e95453..7c215864 100644 --- a/src/renderer/src/pages/chat/chat-helpers.ts +++ b/src/renderer/src/pages/chat/chat-helpers.ts @@ -115,6 +115,34 @@ export interface ConversationItem { unread: number; } +/** Unread messages per space for one reader, counted by position after the last read message (time then insertion order). */ +export function unreadBySpace(messages: ChatMessageRecord[], readStates: ChatReadStateRecord[], employeeId: string): Map { + const countsSince = new Map(); + const readStateBySpace = new Map(); + for (const state of readStates) { + if (state.employeeId === employeeId) readStateBySpace.set(state.spaceId, state); + } + const readPassed = new Set(); + const loadedIds = new Set(messages.map((message) => message.id)); + for (const message of messages) { + const state = readStateBySpace.get(message.spaceId); + const readId = state?.lastReadMessageId ?? null; + const loadedRead = readId !== null && loadedIds.has(readId); + if (message.id === readId) { + readPassed.add(message.spaceId); + continue; + } + if (message.senderType === "employee" && message.senderId === employeeId) continue; + // A read message older than the loaded window means every loaded message came after it. + const unread = !state ? true + : loadedRead ? readPassed.has(message.spaceId) + : readId !== null ? true + : message.createdAt > state.lastReadAt; + if (unread) countsSince.set(message.spaceId, (countsSince.get(message.spaceId) ?? 0) + 1); + } + return countsSince; +} + /** * Folds spaces, messages, memberships, and per-employee read state into the * conversation list: every channel and direct conversation as its own row, @@ -133,18 +161,8 @@ export function buildConversations(input: { ceoAgentId: string | null; }): ConversationItem[] { const lastBySpace = new Map(); - const countsSince = new Map(); - const lastReadBySpace = new Map(); - for (const state of input.readStates) { - if (state.employeeId === input.currentEmployeeId) lastReadBySpace.set(state.spaceId, state.lastReadAt); - } - for (const message of input.messages) { - lastBySpace.set(message.spaceId, message); - if (message.senderType === "employee" && message.senderId === input.currentEmployeeId) continue; - const lastRead = lastReadBySpace.get(message.spaceId); - if (lastRead && message.createdAt <= lastRead) continue; - countsSince.set(message.spaceId, (countsSince.get(message.spaceId) ?? 0) + 1); - } + for (const message of input.messages) lastBySpace.set(message.spaceId, message); + const countsSince = unreadBySpace(input.messages, input.readStates, input.currentEmployeeId); const membersBySpace = new Map(); for (const membership of input.memberships) { const group = membersBySpace.get(membership.spaceId); diff --git a/src/renderer/src/shared/Shell.tsx b/src/renderer/src/shared/Shell.tsx index 4442938a..9b373ffa 100644 --- a/src/renderer/src/shared/Shell.tsx +++ b/src/renderer/src/shared/Shell.tsx @@ -1,3 +1,4 @@ +import { unreadBySpace } from "../pages/chat/chat-helpers"; import { useEffect, useState, type FormEvent, type ReactNode } from "react"; import type { AppSnapshot, EmployeeRecord } from "../../../shared/domain"; import { Avatar, Field, Icon, IconButton } from "./ui"; @@ -225,19 +226,10 @@ function computeBadges(snapshot: AppSnapshot, orgId: string): Partial(); - for (const state of snapshot.chatReadStates) { - if (state.orgId !== orgId || state.employeeId !== employeeId) continue; - readByspace.set(state.spaceId, state.lastReadAt); - } + const messages = snapshot.chatMessages.filter((message) => message.orgId === orgId); + const readStates = snapshot.chatReadStates.filter((state) => state.orgId === orgId); let unread = 0; - for (const message of snapshot.chatMessages) { - if (message.orgId !== orgId) continue; - if (message.senderType === "employee" && message.senderId === employeeId) continue; - const lastRead = readByspace.get(message.spaceId); - if (lastRead && message.createdAt <= lastRead) continue; - unread += 1; - } + for (const count of unreadBySpace(messages, readStates, employeeId).values()) unread += count; return unread; } diff --git a/src/shared/commands/chat.ts b/src/shared/commands/chat.ts index e06136f1..181f4451 100644 --- a/src/shared/commands/chat.ts +++ b/src/shared/commands/chat.ts @@ -38,4 +38,6 @@ export interface MarkChatSpaceReadInput { orgId: string; employeeId: string; spaceId: string; + /** The newest message the reader actually saw; later arrivals stay unread. */ + upToMessageId: string; } diff --git a/tests/e2e/ceo-chat.spec.ts b/tests/e2e/ceo-chat.spec.ts index f0141ea4..d960c0ec 100644 --- a/tests/e2e/ceo-chat.spec.ts +++ b/tests/e2e/ceo-chat.spec.ts @@ -133,6 +133,39 @@ test("the pinned CEO conversation is the Messages front door", async () => { await expect(root).toContainText("1 reply"); await thread.getByRole("button", { name: "Close thread" }).click(); await expect(thread).toHaveCount(0); + + // A message that arrives while the reader is up in history stays unread until they scroll down to it. + await page.evaluate(async () => { + const api = (window as unknown as { agentCompany: { + loadSnapshot: () => Promise<{ ok: true; data: { activeOrgId: string; employees: Array<{ id: string; orgId: string }>; chatSpaces: Array<{ id: string; orgId: string; spaceType: string }> } }>; + sendMessage: (input: unknown) => Promise; + } }).agentCompany; + const snapshot = (await api.loadSnapshot()).data; + const employee = snapshot.employees.find((item) => item.orgId === snapshot.activeOrgId)!; + const ceoChat = snapshot.chatSpaces.find((item) => item.orgId === snapshot.activeOrgId && item.spaceType === "direct")!; + for (let index = 0; index < 40; index += 1) { + await api.sendMessage({ orgId: snapshot.activeOrgId, spaceId: ceoChat.id, threadId: null, senderType: "employee", senderId: employee.id, bodyMd: `History line ${index}`, attachmentFileIds: [] }); + } + }); + await expect(chatLog.getByText("History line 39")).toBeVisible(); + await chatLog.evaluate((element) => { + element.scrollTop = 0; + element.dispatchEvent(new Event("scroll")); + }); + await page.evaluate(async () => { + const api = (window as unknown as { agentCompany: { + loadSnapshot: () => Promise<{ ok: true; data: { activeOrgId: string; agents: Array<{ id: string; orgId: string; roleLevel: string }>; chatSpaces: Array<{ id: string; orgId: string; spaceType: string }> } }>; + sendMessage: (input: unknown) => Promise; + } }).agentCompany; + const snapshot = (await api.loadSnapshot()).data; + const ceo = snapshot.agents.find((agent) => agent.orgId === snapshot.activeOrgId && agent.roleLevel === "c_suite")!; + const ceoChat = snapshot.chatSpaces.find((item) => item.orgId === snapshot.activeOrgId && item.spaceType === "direct")!; + await api.sendMessage({ orgId: snapshot.activeOrgId, spaceId: ceoChat.id, threadId: null, senderType: "agent", senderId: ceo.id, bodyMd: "An update you have not seen yet.", attachmentFileIds: [] }); + }); + const ceoRow = page.locator(".convo-item").first(); + await expect(ceoRow.locator(".nav-badge")).toHaveText("1"); + await page.getByRole("button", { name: "Jump to latest" }).click(); + await expect(ceoRow.locator(".nav-badge")).toHaveCount(0); } finally { await app.close(); } diff --git a/tests/unit/chat-read-state.test.ts b/tests/unit/chat-read-state.test.ts new file mode 100644 index 00000000..916e70f6 --- /dev/null +++ b/tests/unit/chat-read-state.test.ts @@ -0,0 +1,47 @@ +import { randomUUID } from "node:crypto"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { describe, expect, test } from "vitest"; +import { DatabaseConnection } from "../../src/main/database/connection"; +import { OrganizationService } from "../../src/main/services/organization-service"; +import { ChatService } from "../../src/main/services/chat-service"; + +describe("read state", () => { + test("marks read up to the message the reader saw, never past it and never backwards", () => { + const connection = new DatabaseConnection(join(tmpdir(), `ac-read-state-${randomUUID()}.db`)); + const chat = new ChatService(connection.db); + let orgId = ""; + connection.transaction(() => { + orgId = new OrganizationService(connection.db).createOrganization({ organizationName: "Acme", employeeName: "Yang", mission: "m", goal: "g" }); + }); + const employeeId = (connection.db.prepare("select id from employees where org_id = ?").get(orgId) as { id: string }).id; + const space = (name: string) => (connection.db.prepare("select id from chat_spaces where org_id = ? and name = ?").get(orgId, name) as { id: string }).id; + const general = space("general"); + const send = (spaceId: string, bodyMd: string) => connection.transaction(() => chat.sendMessage({ + orgId, spaceId, threadId: null, senderType: "employee", senderId: employeeId, bodyMd, attachmentFileIds: [], + })); + const createdAt = (id: string) => (connection.db.prepare("select created_at from chat_messages where id = ?").get(id) as { created_at: string }).created_at; + const readState = () => connection.db.prepare("select last_read_message_id as id, last_read_at as at from chat_read_state where space_id = ?").get(general); + const seen = send(general, "Seen"); + const later = send(general, "Arrived while scrolled up"); + connection.db.prepare("update chat_messages set created_at = ? where id = ?").run("2099-01-01T00:00:00.000Z", later); + + chat.markSpaceRead({ orgId, employeeId, spaceId: general, upToMessageId: seen }); + expect(readState()).toEqual({ id: seen, at: createdAt(seen) }); + chat.markSpaceRead({ orgId, employeeId, spaceId: general, upToMessageId: later }); + chat.markSpaceRead({ orgId, employeeId, spaceId: general, upToMessageId: seen }); + expect(readState()).toEqual({ id: later, at: "2099-01-01T00:00:00.000Z" }); + // Two messages in the same millisecond still order by insertion, so reading the second advances. + const first = send(general, "Same instant one"); + const second = send(general, "Same instant two"); + connection.db.prepare("update chat_messages set created_at = '2099-06-01T00:00:00.000Z' where id in (?, ?)").run(first, second); + chat.markSpaceRead({ orgId, employeeId, spaceId: general, upToMessageId: first }); + chat.markSpaceRead({ orgId, employeeId, spaceId: general, upToMessageId: second }); + expect(readState()).toEqual({ id: second, at: "2099-06-01T00:00:00.000Z" }); + chat.markSpaceRead({ orgId, employeeId, spaceId: general, upToMessageId: first }); + expect(readState()).toEqual({ id: second, at: "2099-06-01T00:00:00.000Z" }); + expect(() => chat.markSpaceRead({ orgId, employeeId, spaceId: general, upToMessageId: send(space("engineering"), "Elsewhere") })) + .toThrow("The message read is not in this conversation."); + connection.close(); + }); +}); diff --git a/tests/unit/conversation-pinning.test.ts b/tests/unit/conversation-pinning.test.ts index ed55af5f..cd2585a2 100644 --- a/tests/unit/conversation-pinning.test.ts +++ b/tests/unit/conversation-pinning.test.ts @@ -4,6 +4,7 @@ import type { AgentRecord, ChatMembershipRecord, ChatMessageRecord, + ChatReadStateRecord, ChatSpaceRecord, EmployeeRecord, } from "../../src/shared/domain"; @@ -103,3 +104,18 @@ describe("conversation pinning", () => { expect(conversations[0].preview).toContain("CEO:"); }); }); + +describe("unread counts", () => { + test("count by position after the last read message, so a same-millisecond arrival is still unread", () => { + const { spaces, memberships } = fixtures(); + const at = "2026-02-01T00:00:00.000Z"; + const messages = [message("m_read", "spc_general", CEO, at), message("m_same_ms", "spc_general", ENGINEER, at)]; + const unread = (readStates: ChatReadStateRecord[]) => buildConversations({ spaces, memberships, messages, readStates, employees, agents, currentEmployeeId: FOUNDER, ceoAgentId: CEO }) + .find((item) => item.space.id === "spc_general")!.unread; + const state = (lastReadMessageId: string | null): ChatReadStateRecord => ({ orgId: ORG, employeeId: FOUNDER, spaceId: "spc_general", lastReadMessageId, lastReadAt: at }); + expect(unread([state("m_read")])).toBe(1); + expect(unread([state("m_same_ms")])).toBe(0); + // A read message older than the loaded window means every loaded message came after it. + expect(unread([state("m_older_than_window")])).toBe(2); + }); +});