From 2a89d1e8831d86f9462004efbce3ab320e12f9ab Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Fri, 9 Oct 2026 00:08:32 -0400 Subject: [PATCH] Q144: each message says where the reply it asked for stands Co-Authored-By: Claude Opus 5.5 --- .../migrations/100-chat-reply-requests.ts | 19 +++++ src/main/database/schema.ts | 2 + src/main/services/chat-agent-wake-service.ts | 12 ++- src/main/services/chat-history.ts | 3 + src/main/services/chat-reply-status.ts | 32 ++++++++ src/main/services/snapshot-service.ts | 5 +- src/renderer/src/i18n/zh.ts | 7 ++ src/renderer/src/pages/ChatPage.tsx | 1 + .../src/pages/chat/ChatConversation.tsx | 9 ++- .../src/pages/chat/ChatMessageRow.tsx | 23 ++++++ src/renderer/src/pages/chat/ChatTimeline.tsx | 5 ++ src/shared/chat-domain.ts | 9 +++ src/shared/domain.ts | 2 +- src/shared/snapshot-domain.ts | 2 + tests/e2e/ceo-chat.spec.ts | 2 + tests/integration/chat-reply-status.test.ts | 80 +++++++++++++++++++ tests/unit/chat-thread-history.test.ts | 1 + tests/unit/domain-services.test.ts | 2 +- 18 files changed, 211 insertions(+), 5 deletions(-) create mode 100644 src/main/database/migrations/100-chat-reply-requests.ts create mode 100644 src/main/services/chat-reply-status.ts create mode 100644 tests/integration/chat-reply-status.test.ts diff --git a/src/main/database/migrations/100-chat-reply-requests.ts b/src/main/database/migrations/100-chat-reply-requests.ts new file mode 100644 index 00000000..c0e52102 --- /dev/null +++ b/src/main/database/migrations/100-chat-reply-requests.ts @@ -0,0 +1,19 @@ +// Every message that asked an agent for a reply, and the wake task answering it; a message folded into a queued wake shares its task. +export const chatReplyRequestsSchema = ` + create table if not exists chat_reply_requests ( + message_id text not null references chat_messages(id) on delete cascade, + agent_id text not null references agents(id) on delete cascade, + task_id text not null references tasks(id) on delete cascade, + created_at text not null, + primary key (message_id, agent_id) + ); + + insert or ignore into chat_reply_requests(message_id, agent_id, task_id, created_at) + select json_extract(r.secret_scope_json, '$.sourceMessageId'), r.agent_id, r.task_id, r.created_at + from task_runs r + where r.trigger = 'chat_mention' + and exists (select 1 from chat_messages m where m.id = json_extract(r.secret_scope_json, '$.sourceMessageId')); + + insert into schema_migrations(version, name, applied_at) + values (100, 'chat_reply_requests', datetime('now')); +`; diff --git a/src/main/database/schema.ts b/src/main/database/schema.ts index cd483d14..2e126d74 100644 --- a/src/main/database/schema.ts +++ b/src/main/database/schema.ts @@ -98,6 +98,7 @@ import { dropChaosDrillsSchema } from "./migrations/096-drop-chaos-drills"; import { dropDrillOnlySloSchema } from "./migrations/097-drop-drill-only-slo"; import { dropAuditReplaysSchema } from "./migrations/098-drop-audit-replays"; import { dropRollbackRecordsSchema } from "./migrations/099-drop-rollback-records"; +import { chatReplyRequestsSchema } from "./migrations/100-chat-reply-requests"; interface Migration { version: number; @@ -205,6 +206,7 @@ const MIGRATIONS: Migration[] = [ { version: 97, name: "drop_drill_only_slo", sql: dropDrillOnlySloSchema }, { version: 98, name: "drop_audit_replays", sql: dropAuditReplaysSchema }, { version: 99, name: "drop_rollback_records", sql: dropRollbackRecordsSchema }, + { version: 100, name: "chat_reply_requests", sql: chatReplyRequestsSchema }, ]; const SCHEMA_VERSION = MIGRATIONS[MIGRATIONS.length - 1].version; diff --git a/src/main/services/chat-agent-wake-service.ts b/src/main/services/chat-agent-wake-service.ts index 711ba2b8..e1a1aa2f 100644 --- a/src/main/services/chat-agent-wake-service.ts +++ b/src/main/services/chat-agent-wake-service.ts @@ -122,7 +122,10 @@ export class ChatAgentWakeService { // newest messages when it executes, so waking again before it starts would // spend a whole run to answer the same conversation twice. const pending = this.pendingWakeRun(input.orgId, agentId, threadId); - if (pending) return { targetId: agentId, taskId: pending.taskId, runId: pending.runId }; + if (pending) { + this.recordReplyRequest(messageId, agentId, pending.taskId, now); + return { targetId: agentId, taskId: pending.taskId, runId: pending.runId }; + } const workspace = this.workspaceForAgent(input.orgId, agent.workspace_id); const from = this.principalName(input.orgId, input.senderType, input.senderId); const space = this.spaceRow(input.orgId, input.spaceId); @@ -142,6 +145,7 @@ export class ChatAgentWakeService { origin: "chat_wake", }); const runId = this.createQueuedRun({ orgId: input.orgId, taskId, agent, workspace, messageId, threadId, senderType: input.senderType, now }); + this.recordReplyRequest(messageId, agent.id, taskId, now); this.recordPresence(agent.id, { command: "mention", args: "" }, now); return { targetId: agent.id, taskId, runId }; } @@ -168,10 +172,16 @@ export class ChatAgentWakeService { origin: "chat_wake", }); const runId = this.createQueuedRun({ orgId: input.orgId, taskId, agent, workspace, messageId, threadId, senderType: input.senderType, now }); + this.recordReplyRequest(messageId, agent.id, taskId, now); this.recordPresence(agent.id, slash, now); return { targetId: agent.id, taskId, runId }; } + /** Links the message to the wake task that answers it, so the message can show where its reply stands. */ + private recordReplyRequest(messageId: string, agentId: string, taskId: string, now: string): void { + this.db.prepare("insert or ignore into chat_reply_requests(message_id, agent_id, task_id, created_at) values (?, ?, ?, ?)").run(messageId, agentId, taskId, now); + } + private pendingWakeRun(orgId: string, agentId: string, threadId: string): { taskId: string; runId: string } | null { const row = this.db.prepare(` select id, task_id from task_runs diff --git a/src/main/services/chat-history.ts b/src/main/services/chat-history.ts index 446dd83d..3dac7b8c 100644 --- a/src/main/services/chat-history.ts +++ b/src/main/services/chat-history.ts @@ -1,6 +1,8 @@ import type { DatabaseSync } from "node:sqlite"; import type { SpaceHistoryPage, ThreadHistory } from "../../shared/chat-domain"; import { camel } from "./snapshot-service"; +import { chatReplyStatuses } from "./chat-reply-status"; +import { taskHolds } from "./task-holds"; /** Every message of one thread with its attachments, mentions and commands, however far back it started. */ export function threadHistory(db: DatabaseSync, orgId: string, threadId: string): ThreadHistory { @@ -55,6 +57,7 @@ export function spaceHistory(db: DatabaseSync, orgId: string, spaceId: string, b 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(?)))"), + replyStatuses: chatReplyStatuses(db, page.map((row) => row.id as string), taskHolds(db, new Date())), hasMore, }; } diff --git a/src/main/services/chat-reply-status.ts b/src/main/services/chat-reply-status.ts new file mode 100644 index 00000000..e09706b3 --- /dev/null +++ b/src/main/services/chat-reply-status.ts @@ -0,0 +1,32 @@ +import type { DatabaseSync } from "node:sqlite"; +import type { ChatReplyStatus } from "../../shared/chat-domain"; +import type { TaskHold } from "../../shared/domain"; + +const EXPIRED_PREFIX = "Expired:"; + +/** Where each agent reply a loaded message asked for stands: its wake task, through every run (checkpoints and continuations too). */ +export function chatReplyStatuses(db: DatabaseSync, messageIds: string[], holds: Map): ChatReplyStatus[] { + const requests = db.prepare(` + select q.message_id, q.agent_id, q.task_id, t.status as task_status, r.state as run_state, r.error_message + from chat_reply_requests q + join tasks t on t.id = q.task_id + left join task_runs r on r.id = (select id from task_runs where task_id = q.task_id order by created_at desc, rowid desc limit 1) + where q.message_id in (select value from json_each(?)) + order by q.created_at, q.rowid + `).all(JSON.stringify(messageIds)) as Array<{ + message_id: string; agent_id: string; task_id: string; task_status: string; run_state: string | null; error_message: string | null; + }>; + return requests.map((request): ChatReplyStatus => { + const base = { messageId: request.message_id, agentId: request.agent_id }; + if (request.run_state === "running") return { ...base, state: "running", detail: null, resumesAt: null }; + if (request.run_state === "failed") { + if (request.error_message === null) throw new Error(`The reply to message ${request.message_id} failed without a recorded reason.`); + return { ...base, state: request.error_message.startsWith(EXPIRED_PREFIX) ? "expired" : "failed", detail: request.error_message, resumesAt: null }; + } + if (request.run_state === "stopped" || request.run_state === "superseded") return { ...base, state: "stopped", detail: request.error_message, resumesAt: null }; + // A completed run replied only if it closed the wake task; a step-limit checkpoint leaves it open for the next session. + if (request.task_status === "done") return { ...base, state: "replied", detail: null, resumesAt: null }; + const hold = holds.get(request.task_id); + return hold ? { ...base, state: "waiting", detail: hold.reason, resumesAt: hold.resumesAt } : { ...base, state: "queued", detail: null, resumesAt: null }; + }); +} diff --git a/src/main/services/snapshot-service.ts b/src/main/services/snapshot-service.ts index db6c0d97..8d8bb23a 100644 --- a/src/main/services/snapshot-service.ts +++ b/src/main/services/snapshot-service.ts @@ -5,6 +5,7 @@ import { IntegrationKeyStore } from "./integration-key-store"; import { currentProviderUsageWindow } from "../../shared/provider-runtime-domain"; import { buildUsageStats } from "./usage-stats"; import { taskHolds } from "./task-holds"; +import { chatReplyStatuses } from "./chat-reply-status"; type Row = Record; type ProviderUsageWindowRow = Pick< @@ -51,6 +52,7 @@ export class SnapshotService { const agents = list(this.db, "agents", "created_at asc") .map((agent) => ({ ...agent, heartbeatEnabled: toBool(agent.heartbeatEnabled) })); const holds = taskHolds(this.db, new Date()); + const chatMessages = recentList(this.db, "chat_messages", "created_at desc, rowid desc", 500).reverse(); const tasks = list(this.db, "tasks", "created_at asc") .map((task) => ({ ...task, requiresFilesystem: toBool(task.requiresFilesystem), hold: holds.get(task.id) ?? null })); const workspaces = list(this.db, "workspaces", "created_at asc") @@ -96,7 +98,8 @@ 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, rowid desc", 500).reverse(), + chatMessages, + chatReplyStatuses: chatReplyStatuses(this.db, chatMessages.map((message) => message.id), holds), // 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 diff --git a/src/renderer/src/i18n/zh.ts b/src/renderer/src/i18n/zh.ts index 677c5399..1ba4821d 100644 --- a/src/renderer/src/i18n/zh.ts +++ b/src/renderer/src/i18n/zh.ts @@ -50,6 +50,13 @@ export const zh: Record = { "Send reply": "发送回复", "Earlier messages could not load: {reason}": "更早的消息未能加载：{reason}", "This is the start of the conversation.": "这是会话的开头。", + "{agent} will reply; queued.": "{agent} 将回复，已排队。", + "{agent} is replying…": "{agent} 正在回复…", + "{agent} is waiting: {reason} Resumes {time}.": "{agent} 正在等待：{reason} 将于 {time} 恢复。", + "{agent} is waiting: {reason}": "{agent} 正在等待：{reason}", + "{agent} did not reply: {reason}": "{agent} 未回复：{reason}", + "{agent}'s reply failed: {reason}": "{agent} 的回复失败：{reason}", + "{agent}'s reply was stopped.": "{agent} 的回复已停止。", "Resume": "恢复", "Missing this run": "本次运行缺少的能力", "Disconnect": "断开连接", diff --git a/src/renderer/src/pages/ChatPage.tsx b/src/renderer/src/pages/ChatPage.tsx index 0f14cb25..494753f5 100644 --- a/src/renderer/src/pages/ChatPage.tsx +++ b/src/renderer/src/pages/ChatPage.tsx @@ -238,6 +238,7 @@ export function ChatPage({ snapshot, actions }: PageProps) { participants={participants} messagesInSpace={messagesInSpace} threads={snapshot.chatThreads} + replyStatuses={snapshot.chatReplyStatuses} pendingApprovals={pendingApprovalsInSpace} attachments={snapshot.chatAttachments} mentions={mentionsAll} diff --git a/src/renderer/src/pages/chat/ChatConversation.tsx b/src/renderer/src/pages/chat/ChatConversation.tsx index ba2e6f6d..8d669809 100644 --- a/src/renderer/src/pages/chat/ChatConversation.tsx +++ b/src/renderer/src/pages/chat/ChatConversation.tsx @@ -14,7 +14,7 @@ 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"; +import type { ChatReplyStatus, SpaceHistoryPage } from "../../../../shared/chat-domain"; /** Older messages fetched per "Show earlier messages". */ const HISTORY_PAGE = 150; @@ -26,6 +26,7 @@ export function ChatConversation({ participants, messagesInSpace, threads, + replyStatuses, pendingApprovals, attachments, mentions, @@ -53,6 +54,7 @@ export function ChatConversation({ participants: string[]; messagesInSpace: ChatMessageRecord[]; threads: ChatThreadRecord[]; + replyStatuses: ChatReplyStatus[]; pendingApprovals: ApprovalRecord[]; attachments: ChatAttachmentRecord[]; mentions: ChatMentionRecord[]; @@ -130,6 +132,10 @@ export function ChatConversation({ 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 allReplyStatuses = useMemo( + () => [...(older?.replyStatuses ?? []), ...replyStatuses.filter((item) => !pagedIds.has(item.messageId))], + [older, replyStatuses, pagedIds], + ); const allSlashInvocations = useMemo( () => [...(older?.slashInvocations ?? []), ...slashInvocations.filter((item) => !pagedIds.has(item.messageId))], [older, slashInvocations, pagedIds], @@ -214,6 +220,7 @@ export function ChatConversation({ historyError={historyError} onLoadOlder={loadOlder} threadLinks={threadLinks} + replyStatuses={allReplyStatuses} onOpenThread={setOpenThreadId} onApprove={onApprove} onDecline={onDecline} diff --git a/src/renderer/src/pages/chat/ChatMessageRow.tsx b/src/renderer/src/pages/chat/ChatMessageRow.tsx index 388f30bb..df114bcf 100644 --- a/src/renderer/src/pages/chat/ChatMessageRow.tsx +++ b/src/renderer/src/pages/chat/ChatMessageRow.tsx @@ -10,6 +10,7 @@ import type { import { formatTime } from "../types"; import { messageAttachments, senderName, type MessageContext } from "./chat-helpers"; import type { ThreadLink } from "./thread-links"; +import type { ChatReplyStatus } from "../../../../shared/chat-domain"; import { t } from "../../i18n/i18n"; export function ChatMessageRow({ @@ -21,6 +22,7 @@ export function ChatMessageRow({ context, canModify = false, thread, + replyStatuses = [], onOpenThread, onEdit, onDelete, @@ -35,6 +37,8 @@ export function ChatMessageRow({ canModify?: boolean; /** Where the message sits in its thread; absent inside the thread view itself. */ thread?: ThreadLink; + /** Where each agent's reply to this message stands; a finished reply shows nothing. */ + replyStatuses?: ChatReplyStatus[]; onOpenThread?: () => void; onEdit?: (bodyMd: string) => void; onDelete?: () => void; @@ -124,6 +128,11 @@ export function ChatMessageRow({ {thread.replyCount === 1 ? t("1 reply") : t("{count} replies", { count: String(thread.replyCount) })} )} + {replyStatuses.filter((status) => status.state !== "replied").map((status) => ( +
+ {replyStatusText(status, context)} +
+ ))} {readers.length > 0 && (
{t("Read by {names}", { names: readers.join(", ") })} @@ -134,6 +143,20 @@ export function ChatMessageRow({ ); } +function replyStatusText(status: ChatReplyStatus, context: MessageContext): string { + const agent = context.agents.find((item) => item.id === status.agentId)?.name ?? t("Former member"); + if (status.state === "queued") return t("{agent} will reply; queued.", { agent }); + if (status.state === "running") return t("{agent} is replying…", { agent }); + if (status.state === "waiting") { + return status.resumesAt + ? t("{agent} is waiting: {reason} Resumes {time}.", { agent, reason: status.detail, time: formatTime(status.resumesAt) }) + : t("{agent} is waiting: {reason}", { agent, reason: status.detail }); + } + if (status.state === "expired") return t("{agent} did not reply: {reason}", { agent, reason: status.detail }); + if (status.state === "failed") return t("{agent}'s reply failed: {reason}", { agent, reason: status.detail }); + return t("{agent}'s reply was stopped.", { agent }); +} + function renderBody(body: string, mentions: ChatMentionRecord[], context: MessageContext): ReactNode { if (mentions.length === 0) return formatInline(body, "b"); const parts: ReactNode[] = []; diff --git a/src/renderer/src/pages/chat/ChatTimeline.tsx b/src/renderer/src/pages/chat/ChatTimeline.tsx index fad9135e..6b3fbe59 100644 --- a/src/renderer/src/pages/chat/ChatTimeline.tsx +++ b/src/renderer/src/pages/chat/ChatTimeline.tsx @@ -10,6 +10,7 @@ import type { } from "../../../../shared/domain"; import { ChatMessageRow } from "./ChatMessageRow"; import type { ThreadLink } from "./thread-links"; +import type { ChatReplyStatus } from "../../../../shared/chat-domain"; import { formatDay, type MessageContext } from "./chat-helpers"; import { t } from "../../i18n/i18n"; @@ -42,6 +43,7 @@ export function ChatTimeline({ historyError, onLoadOlder, threadLinks, + replyStatuses, onOpenThread, onApprove, onDecline, @@ -62,6 +64,7 @@ export function ChatTimeline({ historyError: string | null; onLoadOlder: () => void; threadLinks: Map; + replyStatuses: ChatReplyStatus[]; onOpenThread: (threadId: string) => void; onApprove: (approvalId: string) => void; onDecline: (approvalId: string) => void; @@ -72,6 +75,7 @@ export function ChatTimeline({ const [visibleCount, setVisibleCount] = useState(WINDOW_SIZE); const attachmentsByMessage = groupBy(attachments, (item) => item.messageId); + const repliesByMessage = groupBy(replyStatuses, (item) => item.messageId); const mentionsByMessage = groupBy(mentions, (item) => item.messageId); const invocationsByMessage = groupBy(slashInvocations, (item) => item.messageId); @@ -113,6 +117,7 @@ export function ChatTimeline({ context={context} canModify={message.senderType === "employee" && message.senderId === currentEmployeeId && !message.deletedAt} thread={threadLinks.get(message.id)} + replyStatuses={repliesByMessage.get(message.id)} onOpenThread={threadId ? () => onOpenThread(threadId) : undefined} onEdit={(bodyMd) => onEditMessage(message.id, bodyMd)} onDelete={() => onDeleteMessage(message.id)} diff --git a/src/shared/chat-domain.ts b/src/shared/chat-domain.ts index 5d950f88..9e2db632 100644 --- a/src/shared/chat-domain.ts +++ b/src/shared/chat-domain.ts @@ -102,5 +102,14 @@ export interface ThreadHistory { export interface SpaceHistoryPage extends ThreadHistory { /** The threads these messages belong to, which the snapshot may no longer hold. */ threads: ChatThreadRecord[]; + replyStatuses: ChatReplyStatus[]; hasMore: boolean; } + +/** Where one agent's reply to a message stands: waiting carries the hold's reason and resume time, expired and failed say why. */ +export type ChatReplyStatus = { messageId: string; agentId: string } & ( + | { state: "queued" | "running" | "replied"; detail: null; resumesAt: null } + | { state: "waiting"; detail: string; resumesAt: string | null } + | { state: "expired" | "failed"; detail: string; resumesAt: null } + | { state: "stopped"; detail: string | null; resumesAt: null } +); diff --git a/src/shared/domain.ts b/src/shared/domain.ts index b2a7dd15..175f3dba 100644 --- a/src/shared/domain.ts +++ b/src/shared/domain.ts @@ -1,6 +1,6 @@ import type { ProviderDataPolicy } from "./provider-policy-domain"; export { BOARD_INTERVENTION_ACTION, BOARD_INTERVENTION_MODE, CHAT_THREAD_BOARD_STATE, type BoardInterventionAction, type BoardInterventionMode, type BoardInterventionRecord, type ChatThreadBoardState } from "./board-domain"; -export type { ChatSpaceRecord, ChatMessageRecord, ChatMembershipRecord, ChatThreadRecord, ChatAttachmentRecord, ChatMentionRecord, ChatSlashInvocationRecord, ChatPresenceRecord, ChatReadStateRecord } from "./chat-domain"; +export type { ChatReplyStatus, ChatSpaceRecord, ChatMessageRecord, ChatMembershipRecord, ChatThreadRecord, ChatAttachmentRecord, ChatMentionRecord, ChatSlashInvocationRecord, ChatPresenceRecord, ChatReadStateRecord } from "./chat-domain"; export { COMPLIANCE_EXPORT_STATE, COMPLIANCE_EXPORT_TYPE, type ComplianceExportRecord, type ComplianceExportState, type ComplianceExportType, type ComplianceSubjectType, type LegalHoldEntityType, type LegalHoldRecord, type RetentionFreezeRecord, type RetentionFreezeScopeType } from "./compliance-domain"; export { ACCESS_DECISION, IDENTITY_SOURCE, ROLE_KEY, type AccessDecision, type AccessDecisionRecord, type EmployeeIdentityRecord, type IdentitySource, type PrincipalType, type RoleGrantRecord, type RoleKey, type ServiceIdentityRecord } from "./identity-domain"; export { ROUTE_TO_TYPE, type ApprovalAssignmentRecord, type ApprovalRoutingRuleRecord, type HiringPolicyRecord, type OrgDepartmentRecord, type RouteToType } from "./org-governance-domain"; diff --git a/src/shared/snapshot-domain.ts b/src/shared/snapshot-domain.ts index ec063602..6ecaf141 100644 --- a/src/shared/snapshot-domain.ts +++ b/src/shared/snapshot-domain.ts @@ -11,6 +11,7 @@ import type { ChatSlashInvocationRecord, ChatSpaceRecord, ChatThreadRecord, + ChatReplyStatus, } from "./chat-domain"; import type { ComplianceExportRecord, LegalHoldRecord, RetentionFreezeRecord } from "./compliance-domain"; import type { @@ -84,6 +85,7 @@ export interface AppSnapshot { chatMemberships: ChatMembershipRecord[]; chatMessages: ChatMessageRecord[]; chatThreads: ChatThreadRecord[]; + chatReplyStatuses: ChatReplyStatus[]; chatAttachments: ChatAttachmentRecord[]; chatMentions: ChatMentionRecord[]; chatSlashInvocations: ChatSlashInvocationRecord[]; diff --git a/tests/e2e/ceo-chat.spec.ts b/tests/e2e/ceo-chat.spec.ts index 7f7fa4a7..456ef585 100644 --- a/tests/e2e/ceo-chat.spec.ts +++ b/tests/e2e/ceo-chat.spec.ts @@ -69,6 +69,8 @@ test("the pinned CEO conversation is the Messages front door", async () => { return snapshot.data.taskRuns.filter((run) => run.agentId === ceo.id && run.trigger === "chat_mention").map((run) => run.state); }); expect(wake).toEqual(["queued"]); + // The message itself says where the reply stands. + await expect(chatLog.locator(".chat-message", { hasText: "How are we tracking against the goal?" }).first()).toContainText("CEO will reply; queued."); // Recency pressure: newer traffic in a channel must not displace the pin. await page.evaluate(async () => { diff --git a/tests/integration/chat-reply-status.test.ts b/tests/integration/chat-reply-status.test.ts new file mode 100644 index 00000000..9359bd1d --- /dev/null +++ b/tests/integration/chat-reply-status.test.ts @@ -0,0 +1,80 @@ +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 { OnboardingService } from "../../src/main/services/onboarding-service"; +import { ProviderService } from "../../src/main/services/provider-service"; +import { providerAdapters } from "../../src/main/services/provider-adapters"; +import { ChatService } from "../../src/main/services/chat-service"; +import { SnapshotService } from "../../src/main/services/snapshot-service"; +import { expireStaleChatWakes } from "../../src/main/services/chat-agent-wake-service"; +import { spaceHistory } from "../../src/main/services/chat-history"; + +const testVault = { + encryptSecret: (secret: string) => Buffer.from(secret, "utf8"), + decryptSecret: (ciphertext: Buffer) => ciphertext.toString("utf8"), +}; + +describe("a message's reply status", () => { + test("says whether the reply is queued, waiting and until when, expired, or failed and why", () => { + const connection = new DatabaseConnection(join(tmpdir(), `ac-reply-status-${randomUUID()}.db`)); + const providers = new ProviderService(connection.db, providerAdapters, testVault); + const onboarding = new OnboardingService({ db: connection.db, saveProviderCredential: (input) => providers.saveCredential(input) }); + let orgId = ""; + let ceoId = ""; + connection.transaction(() => { + const summary = onboarding.bootstrap({ + companyName: "FlowPilot", companyDescription: "CSM analytics.", employeeName: "Yang", workspacePath: "", + executionMode: "ai_ceo", autoApproveHires: true, reviewDeliverables: false, + goalTitle: "Ship the MVP.", goalDescription: "A working product.", + providerKey: "openai", providerSecret: "sk-test-key", providerModel: "gpt-5.4", + }); + orgId = summary.orgId; + ceoId = summary.ceoAgentId; + }); + const employeeId = (connection.db.prepare("select id from employees where org_id = ?").get(orgId) as { id: string }).id; + const dm = (connection.db.prepare("select id from chat_spaces where org_id = ? and space_type = 'direct'").get(orgId) as { id: string }).id; + const messageId = connection.transaction(() => new ChatService(connection.db).sendMessage({ + orgId, spaceId: dm, threadId: null, senderType: "employee", senderId: employeeId, bodyMd: "Status?", attachmentFileIds: [], + })); + const status = () => new SnapshotService(connection.db).load().chatReplyStatuses.filter((item) => item.messageId === messageId); + + expect(status()).toEqual([{ messageId, agentId: ceoId, state: "queued", detail: null, resumesAt: null }]); + // A second message folded into the same queued reply gets the same status. + const follow = connection.transaction(() => new ChatService(connection.db).sendMessage({ + orgId, spaceId: dm, threadId: (connection.db.prepare("select thread_id from chat_messages where id = ?").get(messageId) as { thread_id: string }).thread_id, + senderType: "employee", senderId: employeeId, bodyMd: "Also, the budget?", attachmentFileIds: [], + })); + expect(new SnapshotService(connection.db).load().chatReplyStatuses.filter((item) => item.messageId === follow).map((item) => item.state)).toEqual(["queued"]); + // A run that stopped at its step limit has not replied; the task continues. + const wakeRun = connection.db.prepare("select id, task_id from task_runs where trigger = 'chat_mention'").get() as { id: string; task_id: string }; + connection.db.prepare("update task_runs set state = 'completed', completed_at = ? where id = ?").run(new Date().toISOString(), wakeRun.id); + expect(status().map((item) => item.state)).toEqual(["queued"]); + connection.db.prepare("insert into task_runs(id, org_id, task_id, agent_id, trigger, state, created_at) values ('run_continue', ?, ?, ?, 'heartbeat', 'running', ?)") + .run(orgId, wakeRun.task_id, ceoId, new Date().toISOString()); + expect(status().map((item) => item.state)).toEqual(["running"]); + connection.db.prepare("delete from task_runs where id = 'run_continue'").run(); + connection.db.prepare("update task_runs set state = 'queued', completed_at = null where id = ?").run(wakeRun.id); + + const providerId = (connection.db.prepare("select model_provider_id as id from agents where id = ?").get(ceoId) as { id: string }).id; + const until = Date.now() + 120_000; + connection.db.prepare("insert into provider_cooldowns(provider_id, label, until_ms, strikes) values (?, 'OpenAI', ?, 1)").run(providerId, until); + expect(status()).toEqual([{ + messageId, agentId: ceoId, state: "waiting", + detail: "OpenAI reported a passing failure; its work waits out a cooldown.", resumesAt: new Date(until).toISOString(), + }]); + // A message paged back from history carries its status too. + expect(spaceHistory(connection.db, orgId, dm, null, { limit: 10 }).replyStatuses.filter((item) => item.messageId === messageId).map((item) => item.state)).toEqual(["waiting"]); + + connection.transaction(() => expireStaleChatWakes(connection.db, new Date(Date.now() + 2 * 60 * 60_000).toISOString())); + expect(status()).toEqual([{ + messageId, agentId: ceoId, state: "expired", + detail: "Expired: the conversation moved on before this reply could start. A newer message re-wakes the agent.", resumesAt: null, + }]); + + connection.db.prepare("update task_runs set error_message = 'Provider refused the request' where trigger = 'chat_mention'").run(); + expect(status()).toEqual([{ messageId, agentId: ceoId, state: "failed", detail: "Provider refused the request", resumesAt: null }]); + connection.close(); + }); +}); diff --git a/tests/unit/chat-thread-history.test.ts b/tests/unit/chat-thread-history.test.ts index cfb3b9fe..2dfbc5bd 100644 --- a/tests/unit/chat-thread-history.test.ts +++ b/tests/unit/chat-thread-history.test.ts @@ -70,6 +70,7 @@ describe("a conversation's history", () => { 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()); + expect(refreshed.replyStatuses).toEqual([]); // 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."); diff --git a/tests/unit/domain-services.test.ts b/tests/unit/domain-services.test.ts index 61d79f34..226cc7ac 100644 --- a/tests/unit/domain-services.test.ts +++ b/tests/unit/domain-services.test.ts @@ -35,7 +35,7 @@ describe("domain services", () => { const connection = createConnection(); const rows = connection.db.prepare("select version from schema_migrations order by version asc").all() as Array<{ version: number }>; - expect(rows.map((row) => row.version)).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79, 80, 81, 82, 83, 84, 85, 86, 87, 88, 89, 90, 91, 92, 93, 94, 95, 96, 97, 98, 99]); + expect(rows.map((row) => row.version)).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79, 80, 81, 82, 83, 84, 85, 86, 87, 88, 89, 90, 91, 92, 93, 94, 95, 96, 97, 98, 99, 100]); connection.close(); });