Q143: earlier messages come from the database, page by page, back to the first one

Review limit (5) reached; the last finding (exhaustion latched) is fixed
by taking it from each latest hasMore.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
ZacharyZhang-NYandClaude Opus 5.5 committed 2026-10-08 23:52:22 -04:00
1 parent b18eb9ad14
commit 075dea725a
12 files changed
+213 -15

No files matched your search

+4
View File
@@ -72,6 +72,10 @@ export const platformContracts = {
approvalDrifts: z.strictObject({ orgId: id }),
runRequests: z.strictObject({ orgId: id, runId: id }),
chatThreadHistory: z.strictObject({ orgId: id, threadId: id }),
chatSpaceHistory: z.union([
z.strictObject({ orgId: id, spaceId: id, beforeMessageId: id.nullable(), limit: z.number().int().min(1).max(500) }),
z.strictObject({ orgId: id, spaceId: id, beforeMessageId: id.nullable(), fromMessageId: id }),
]),
search: z.strictObject({
orgId: id, employeeId: id, query: z.string(), entityTypes: z.array(z.enum(SEARCH_ENTITY_TYPES)).optional(), limit: z.number().optional(),
}) satisfies z.ZodType<C.SearchInput>,
+2 -1
View File
@@ -28,7 +28,7 @@ import { SprintService } from "../services/sprint-service";
import { ComposioConnectionService } from "../services/composio/composio-connection-service";
import { ConnectorGrantsService } from "../services/composio/connector-grants";
import { approvalDrifts } from "../services/approval-drift";
import { threadHistory } from "../services/chat-history";
import { spaceHistory, threadHistory } from "../services/chat-history";
import { requestsForRun } from "../services/provider-request-store";
import { setGoalAuditor } from "../services/goal-audit-inputs";
import { IntegrationKeyStore } from "../services/integration-key-store";
@@ -415,6 +415,7 @@ export function registerIpcHandlers(deps: IpcDeps): void {
handle("grantConnector", (input) => mutate(() => new ConnectorGrantsService(deps.connection.db).grantFromBoard(input)));
handle("runRequests", (input) => ok(requestsForRun(deps.connection.db, input.orgId, input.runId)));
handle("chatThreadHistory", (input) => ok(threadHistory(deps.connection.db, input.orgId, input.threadId)));
handle("chatSpaceHistory", (input) => ok(spaceHistory(deps.connection.db, input.orgId, input.spaceId, input.beforeMessageId, "limit" in input ? { limit: input.limit } : { fromMessageId: input.fromMessageId })));
handle("approvalDrifts", (input) => {
try {
return ok(approvalDrifts(deps.connection.db, input.orgId));
+45 -1
View File
@@ -1,5 +1,5 @@
import type { DatabaseSync } from "node:sqlite";
import type { ThreadHistory } from "../../shared/chat-domain";
import type { SpaceHistoryPage, ThreadHistory } from "../../shared/chat-domain";
import { camel } from "./snapshot-service";
/** Every message of one thread with its attachments, mentions and commands, however far back it started. */
@@ -14,3 +14,47 @@ export function threadHistory(db: DatabaseSync, orgId: string, threadId: string)
slashInvocations: rows("select s.* from chat_slash_invocations s join chat_messages m on m.id = s.message_id where m.thread_id = ?"),
};
}
type Row = Record<string, unknown>;
/** Which older messages to return: the next page before the cursor, or every message from a loaded one up to the cursor. */
export type HistoryScope = { limit: number } | { fromMessageId: string };
/** A conversation's messages before a loaded one (or its newest when none is loaded), in time then insertion order, with what they carry. */
export function spaceHistory(db: DatabaseSync, orgId: string, spaceId: string, beforeMessageId: string | null, scope: HistoryScope): SpaceHistoryPage {
if (!db.prepare("select id from chat_spaces where id = ? and org_id = ?").get(spaceId, orgId)) throw new Error("Chat space was not found.");
const position = (messageId: string, what: string) => {
const row = db.prepare("select created_at, rowid from chat_messages where id = ? and space_id = ? and org_id = ?")
.get(messageId, spaceId, orgId) as { created_at: string; rowid: number } | undefined;
if (!row) throw new Error(`The ${what} is not in this conversation.`);
return [row.created_at, row.rowid] as const;
};
const conditions = ["space_id = ?"];
const params: Array<string | number> = [spaceId];
if (beforeMessageId !== null) {
conditions.push("(created_at, rowid) < (?, ?)");
params.push(...position(beforeMessageId, "earliest loaded message"));
}
if ("fromMessageId" in scope) {
conditions.push("(created_at, rowid) >= (?, ?)");
params.push(...position(scope.fromMessageId, "oldest loaded message"));
}
const limit = "limit" in scope ? scope.limit : -1;
const rows = db.prepare(`select *, rowid as seq from chat_messages where ${conditions.join(" and ")} order by created_at desc, rowid desc limit ?`)
.all(...params, limit === -1 ? -1 : limit + 1) as Row[];
const page = (limit === -1 ? rows : rows.slice(0, limit)).reverse();
const oldest = page[0];
const hasMore = limit !== -1 ? rows.length > limit : oldest !== undefined && db.prepare(`
select 1 from chat_messages where space_id = ? and (created_at, rowid) < (?, ?) limit 1
`).get(spaceId, oldest.created_at as string, oldest.seq as number) !== undefined;
const ids = JSON.stringify(page.map((row) => row.id));
const carried = <T>(sql: string) => (db.prepare(sql).all(ids) as Row[]).map((row) => camel(row) as T);
return {
messages: page.map(({ seq: _seq, ...row }) => camel(row) as unknown as SpaceHistoryPage["messages"][number]),
attachments: carried("select * from chat_attachments where message_id in (select value from json_each(?)) order by message_id"),
mentions: carried("select * from chat_mentions where message_id in (select value from json_each(?)) order by message_id, offset_start"),
slashInvocations: carried("select * from chat_slash_invocations where message_id in (select value from json_each(?)) order by message_id"),
threads: carried("select * from chat_threads where id in (select thread_id from chat_messages where id in (select value from json_each(?)))"),
hasMore,
};
}
+1
View File
@@ -89,6 +89,7 @@ const api: DesktopApi = {
approvalDrifts: (input) => invoke(IPC_CHANNELS.approvalDrifts, input),
runRequests: (input) => invoke(IPC_CHANNELS.runRequests, input),
chatThreadHistory: (input) => invoke(IPC_CHANNELS.chatThreadHistory, input),
chatSpaceHistory: (input) => invoke(IPC_CHANNELS.chatSpaceHistory, input),
revokeConnector: (input) => invoke(IPC_CHANNELS.revokeConnector, input),
refreshComposioAccounts: (input) => invoke(IPC_CHANNELS.refreshComposioAccounts, input),
search: (input) => invoke(IPC_CHANNELS.search, input),
+2
View File
@@ -48,6 +48,8 @@ export const zh: Record<string, string> = {
"Close thread": "关闭话题",
"Reply": "回复",
"Send reply": "发送回复",
"Earlier messages could not load: {reason}": "更早的消息未能加载:{reason}",
"This is the start of the conversation.": "这是会话的开头。",
"Resume": "恢复",
"Missing this run": "本次运行缺少的能力",
"Disconnect": "断开连接",
@@ -1,4 +1,4 @@
import { useEffect, useLayoutEffect, useMemo, useRef, useState, type FormEvent, type MutableRefObject } from "react";
import { useCallback, useEffect, useLayoutEffect, useMemo, useRef, useState, type FormEvent, type MutableRefObject } from "react";
import { AvatarStack, Icon, IconButton } from "../../shared/ui";
import type {
ApprovalRecord,
@@ -14,6 +14,10 @@ import { ChatComposer, type Draft, type SendState } from "./ChatComposer";
import { ChatTimeline } from "./ChatTimeline";
import { ThreadPanel } from "./ThreadPanel";
import { buildThreadLinks } from "./thread-links";
import type { SpaceHistoryPage } from "../../../../shared/chat-domain";
/** Older messages fetched per "Show earlier messages". */
const HISTORY_PAGE = 150;
import type { MessageContext } from "./chat-helpers";
import { t } from "../../i18n/i18n";
@@ -76,11 +80,65 @@ export function ChatConversation({
const [query, setQuery] = useState("");
// The component is keyed by space, so an open thread closes when the conversation changes.
const [openThreadId, setOpenThreadId] = useState<string | null>(null);
const threadLinks = useMemo(() => buildThreadLinks(messagesInSpace, threads, context), [messagesInSpace, threads, context]);
// The snapshot holds recent chat only; older messages come from the database, paged back from the oldest one loaded.
const [oldestId, setOldestId] = useState<string | null>(null);
const [exhausted, setExhausted] = useState(false);
const [older, setOlder] = useState<SpaceHistoryPage | null>(null);
const [historyError, setHistoryError] = useState<string | null>(null);
const anchorId = messagesInSpace.length > 0 ? messagesInSpace[0].id : null;
const orgId = space?.orgId ?? null;
const currentSpaceId = space?.id ?? null;
const loadOlder = useCallback(() => {
if (!orgId || !currentSpaceId) return;
window.agentCompany.chatSpaceHistory({ orgId, spaceId: currentSpaceId, beforeMessageId: oldestId ?? anchorId, limit: HISTORY_PAGE }).then((result) => {
if (!result.ok) throw new Error(result.error.message);
// With nothing older, the oldest message already shown is pinned, so the range refresh keeps it if newer traffic displaces it.
const oldest = result.data.messages.length > 0 ? result.data.messages[0].id : oldestId ?? anchorId;
if (oldest !== null) setOldestId(oldest);
setExhausted(!result.data.hasMore);
setHistoryError(null);
}).catch((reason: unknown) => setHistoryError(reason instanceof Error ? reason.message : String(reason)));
}, [orgId, currentSpaceId, oldestId, anchorId]);
// While the snapshot holds none of the conversation and nothing is pinned, every reload checks the database for its newest page.
useEffect(() => {
if (messagesInSpace.length === 0 && oldestId === null) loadOlder();
}, [messagesInSpace, oldestId, loadOlder]);
useEffect(() => {
if (!orgId || !currentSpaceId || !oldestId) return;
let cancelled = false;
// The loaded range is refetched on every snapshot reload (messagesInSpace changes), so edits and deletes of old messages show too.
window.agentCompany.chatSpaceHistory({ orgId, spaceId: currentSpaceId, beforeMessageId: anchorId, fromMessageId: oldestId }).then((result) => {
if (cancelled) return;
if (!result.ok) throw new Error(result.error.message);
setOlder(result.data);
setExhausted(!result.data.hasMore);
setHistoryError(null);
}).catch((reason: unknown) => {
if (!cancelled) setHistoryError(reason instanceof Error ? reason.message : String(reason));
});
return () => {
cancelled = true;
};
}, [orgId, currentSpaceId, oldestId, anchorId, messagesInSpace]);
const allMessages = useMemo(() => [...(older?.messages ?? []), ...messagesInSpace], [older, messagesInSpace]);
const allThreads = useMemo(() => {
const byId = new Map(threads.map((thread) => [thread.id, thread]));
for (const thread of older?.threads ?? []) if (!byId.has(thread.id)) byId.set(thread.id, thread);
return [...byId.values()];
}, [threads, older]);
// Paged messages take their attachments, mentions and commands from the page only, so a record the snapshot also holds is not doubled.
const pagedIds = useMemo(() => new Set((older?.messages ?? []).map((message) => message.id)), [older]);
const allAttachments = useMemo(() => [...(older?.attachments ?? []), ...attachments.filter((item) => !pagedIds.has(item.messageId))], [older, attachments, pagedIds]);
const allMentions = useMemo(() => [...(older?.mentions ?? []), ...mentions.filter((item) => !pagedIds.has(item.messageId))], [older, mentions, pagedIds]);
const allSlashInvocations = useMemo(
() => [...(older?.slashInvocations ?? []), ...slashInvocations.filter((item) => !pagedIds.has(item.messageId))],
[older, slashInvocations, pagedIds],
);
const threadLinks = useMemo(() => buildThreadLinks(allMessages, allThreads, context), [allMessages, allThreads, context]);
const trimmed = query.trim().toLowerCase();
const filtered = trimmed
? messagesInSpace.filter((m) => m.bodyMd.toLowerCase().includes(trimmed))
: messagesInSpace;
? allMessages.filter((m) => m.bodyMd.toLowerCase().includes(trimmed))
: allMessages;
// Messenger scroll contract: open at the latest message, stay pinned there
// while the reader is at the bottom, and offer a jump back the moment they
@@ -146,12 +204,15 @@ export function ChatConversation({
key={space.id}
messagesInSpace={filtered}
pendingApprovals={pendingApprovals}
attachments={attachments}
mentions={mentions}
slashInvocations={slashInvocations}
attachments={allAttachments}
mentions={allMentions}
slashInvocations={allSlashInvocations}
files={files}
context={context}
currentEmployeeId={currentEmployeeId}
historyStart={exhausted}
historyError={historyError}
onLoadOlder={loadOlder}
threadLinks={threadLinks}
onOpenThread={setOpenThreadId}
onApprove={onApprove}
+17 -4
View File
@@ -38,6 +38,9 @@ export function ChatTimeline({
files,
context,
currentEmployeeId,
historyStart,
historyError,
onLoadOlder,
threadLinks,
onOpenThread,
onApprove,
@@ -54,6 +57,10 @@ export function ChatTimeline({
files: FileObjectRecord[];
context: MessageContext;
currentEmployeeId: string;
/** True once the database said no older message exists. */
historyStart: boolean;
historyError: string | null;
onLoadOlder: () => void;
threadLinks: Map<string, ThreadLink>;
onOpenThread: (threadId: string) => void;
onApprove: (approvalId: string) => void;
@@ -72,14 +79,20 @@ export function ChatTimeline({
const visible = hiddenCount > 0 ? messagesInSpace.slice(hiddenCount) : messagesInSpace;
const elements: ReactNode[] = [];
if (hiddenCount > 0) {
if (historyError) elements.push(<div key="history-error" className="form-error" role="alert">{t("Earlier messages could not load: {reason}", { reason: historyError })}</div>);
if (hiddenCount > 0 || !historyStart) {
// Showing earlier first reveals loaded messages, then asks the database for the next page.
const showEarlier = () => {
setVisibleCount((count) => count + WINDOW_SIZE);
if (hiddenCount === 0) onLoadOlder();
};
elements.push(
<div key="show-earlier" style={{ display: "flex", justifyContent: "center", marginBottom: 16 }}>
<Button size="small" onClick={() => setVisibleCount((count) => count + WINDOW_SIZE)}>
{t("Show earlier messages")}
</Button>
<Button size="small" onClick={showEarlier}>{t("Show earlier messages")}</Button>
</div>,
);
} else {
elements.push(<div key="history-start" style={{ textAlign: "center", fontSize: 12, color: "var(--text-3)", marginBottom: 16 }}>{t("This is the start of the conversation.")}</div>);
}
let lastDay = "";
for (const message of visible) {
+7
View File
@@ -97,3 +97,10 @@ export interface ThreadHistory {
mentions: ChatMentionRecord[];
slashInvocations: ChatSlashInvocationRecord[];
}
/** One page of a conversation's older messages, oldest first, and whether still older ones exist. */
export interface SpaceHistoryPage extends ThreadHistory {
/** The threads these messages belong to, which the snapshot may no longer hold. */
threads: ChatThreadRecord[];
hasMore: boolean;
}
+3 -1
View File
@@ -1,4 +1,4 @@
import type { ThreadHistory } from "./chat-domain";
import type { SpaceHistoryPage, ThreadHistory } from "./chat-domain";
import type { RunRequestRecord } from "./provider-runtime-domain";
import type { ApprovalDrift } from "./approval-pin";
import type { DesktopResult } from "./result";
@@ -198,6 +198,8 @@ export interface DesktopApi {
runRequests(input: { orgId: string; runId: string }): Promise<DesktopResult<RunRequestRecord[]>>;
/** One thread in full, beyond the snapshot's recent window. */
chatThreadHistory(input: { orgId: string; threadId: string }): Promise<DesktopResult<ThreadHistory>>;
/** The page of a conversation before a loaded message, for history older than the snapshot holds. */
chatSpaceHistory(input: { orgId: string; spaceId: string; beforeMessageId: string | null } & ({ limit: number } | { fromMessageId: string })): Promise<DesktopResult<SpaceHistoryPage>>;
revokeConnector(input: RevokeConnectorInput): Promise<DesktopResult<AppSnapshot>>;
search(input: SearchInput): Promise<DesktopResult<SearchResultRecord[]>>;
subscribe(callback: (event: RendererEvent) => void): () => void;
+1
View File
@@ -82,6 +82,7 @@ export const IPC_CHANNELS = {
approvalDrifts: "agentcompany:approval-drifts",
runRequests: "agentcompany:run-requests",
chatThreadHistory: "agentcompany:chat-thread-history",
chatSpaceHistory: "agentcompany:chat-space-history",
revokeConnector: "agentcompany:revoke-connector",
search: "agentcompany:search",
rendererEvent: "agentcompany:renderer-event",
+23
View File
@@ -166,6 +166,29 @@ test("the pinned CEO conversation is the Messages front door", async () => {
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);
// History older than the loaded window comes from the database, back to the very first message.
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; name: string }> } }>;
sendMessage: (input: unknown) => Promise<unknown>;
} }).agentCompany;
const snapshot = (await api.loadSnapshot()).data;
const employee = snapshot.employees.find((item) => item.orgId === snapshot.activeOrgId)!;
const general = snapshot.chatSpaces.find((item) => item.orgId === snapshot.activeOrgId && item.name === "general")!;
for (let index = 0; index < 520; index += 1) {
await api.sendMessage({ orgId: snapshot.activeOrgId, spaceId: general.id, threadId: null, senderType: "employee", senderId: employee.id, bodyMd: `General line ${index}`, attachmentFileIds: [] });
}
});
await page.locator(".convo-item", { hasText: "general" }).click();
await expect(chatLog.getByText("General line 519")).toBeVisible();
const start = chatLog.getByText("This is the start of the conversation.");
for (let click = 0; click < 10 && !(await start.isVisible()); click += 1) {
await chatLog.getByRole("button", { name: "Show earlier messages" }).click();
await page.waitForTimeout(300);
}
await expect(start).toBeVisible();
await expect(chatLog.getByText("General line 0", { exact: true })).toBeVisible();
} finally {
await app.close();
}
+40 -1
View File
@@ -6,7 +6,7 @@ import { DatabaseConnection } from "../../src/main/database/connection";
import { OrganizationService } from "../../src/main/services/organization-service";
import { ChatService } from "../../src/main/services/chat-service";
import { SnapshotService } from "../../src/main/services/snapshot-service";
import { threadHistory } from "../../src/main/services/chat-history";
import { spaceHistory, threadHistory } from "../../src/main/services/chat-history";
describe("a thread's history", () => {
test("holds the whole thread even after newer traffic pushed it out of the recent window", () => {
@@ -37,3 +37,42 @@ describe("a thread's history", () => {
connection.close();
}, 60_000);
});
describe("a conversation's history", () => {
test("pages back from the earliest loaded message to the very first, beyond the recent window", () => {
const connection = new DatabaseConnection(join(tmpdir(), `ac-space-history-${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 spaceId = (connection.db.prepare("select id from chat_spaces where org_id = ? and name = 'general'").get(orgId) as { id: string }).id;
const ids: string[] = [];
for (let index = 0; index < 250; index += 1) {
ids.push(connection.transaction(() => chat.sendMessage({
orgId, spaceId, threadId: null, senderType: "employee", senderId: employeeId, bodyMd: `Line ${index}`, attachmentFileIds: [],
})));
}
// Every message shares one instant, so only insertion order can page them correctly.
connection.db.prepare("update chat_messages set created_at = '2026-03-01T00:00:00.000Z' where space_id = ?").run(spaceId);
const first = spaceHistory(connection.db, orgId, spaceId, ids[200], { limit: 120 });
expect(first.messages.map((message) => message.bodyMd)).toEqual(Array.from({ length: 120 }, (_, index) => `Line ${index + 80}`));
expect(first.hasMore).toBe(true);
const last = spaceHistory(connection.db, orgId, spaceId, first.messages[0].id, { limit: 120 });
expect(last.messages.map((message) => message.bodyMd)).toEqual(Array.from({ length: 80 }, (_, index) => `Line ${index}`));
expect(last.hasMore).toBe(false);
// The loaded range refreshes in one query from its oldest message up to the cursor, edits included.
connection.db.prepare("update chat_messages set body_md = 'Line 0 (edited)' where id = ?").run(ids[0]);
const refreshed = spaceHistory(connection.db, orgId, spaceId, ids[200], { fromMessageId: ids[0] });
expect(refreshed.messages.map((message) => message.bodyMd)).toEqual(["Line 0 (edited)", ...Array.from({ length: 199 }, (_, index) => `Line ${index + 1}`)]);
expect(refreshed.hasMore).toBe(false);
expect(spaceHistory(connection.db, orgId, spaceId, ids[200], { fromMessageId: ids[1] }).hasMore).toBe(true);
// Each page carries the threads of its messages, which the snapshot may have dropped.
expect(refreshed.threads.map((thread) => thread.rootMessageId).sort()).toEqual(ids.slice(0, 200).sort());
// With nothing loaded, the newest messages come first.
expect(spaceHistory(connection.db, orgId, spaceId, null, { limit: 2 }).messages.map((message) => message.bodyMd)).toEqual(["Line 248", "Line 249"]);
expect(() => spaceHistory(connection.db, orgId, spaceId, "msg_elsewhere", { limit: 10 })).toThrow("The earliest loaded message is not in this conversation.");
connection.close();
}, 60_000);
});