diff --git a/src/main/services/chat-wake-context.ts b/src/main/services/chat-wake-context.ts new file mode 100644 index 00000000..18978ba7 --- /dev/null +++ b/src/main/services/chat-wake-context.ts @@ -0,0 +1,54 @@ +import type { DatabaseSync } from "node:sqlite"; +import type { ChatContextBlock } from "./task-prompts"; + +interface ThreadMessageRow { + body_md: string; + created_at: string; + edited_at: string | null; + sender_type: string; + sender_id: string | null; + sender_name: string; + sender_role: string | null; +} + +/** The thread a chat-wake run answers, split at the agent's last reply. */ +export function loadChatWakeContext(db: DatabaseSync, orgId: string, agentId: string, secretScopeJson: string): ChatContextBlock | null { + const scope = JSON.parse(secretScopeJson) as { sourceThreadId?: string }; + if (!scope.sourceThreadId) return null; + const thread = db.prepare("select space_id from chat_threads where id = ? and org_id = ?") + .get(scope.sourceThreadId, orgId) as { space_id: string } | undefined; + if (!thread) throw new Error(`Chat wake thread ${scope.sourceThreadId} no longer exists.`); + const messages = db.prepare(` + select m.body_md, m.created_at, m.edited_at, m.sender_type, m.sender_id, + coalesce(a.name, e.display_name, m.sender_type) as sender_name, + coalesce(a.role, e.role) as sender_role + from chat_messages m + left join agents a on a.id = m.sender_id and m.sender_type = 'agent' + left join employees e on e.id = m.sender_id and m.sender_type = 'employee' + where m.thread_id = ? and m.org_id = ? + order by m.created_at asc + limit 20 + `).all(scope.sourceThreadId, orgId) as unknown as ThreadMessageRow[]; + if (messages.length === 0) return null; + const isOwn = (m: ThreadMessageRow) => m.sender_type === "agent" && m.sender_id === agentId; + const ownReplies = messages.filter(isOwn); + const lastReplyAt = ownReplies.length > 0 ? ownReplies[ownReplies.length - 1].created_at : ""; + const isFresh = (m: ThreadMessageRow) => + !isOwn(m) && (m.created_at > lastReplyAt || (m.edited_at !== null && m.edited_at > lastReplyAt)); + return { + threadId: scope.sourceThreadId, + spaceId: thread.space_id, + history: messages.filter((m) => !isFresh(m)).map(formatMessage).join("\n"), + fresh: messages.filter(isFresh).map(formatMessage).join("\n"), + }; +} + +function formatMessage(m: ThreadMessageRow): string { + const who = m.sender_role ? `${m.sender_name} (${m.sender_type}, ${m.sender_role})` : `${m.sender_name} (${m.sender_type})`; + const edited = m.edited_at ? ` (edited ${utcMinute(m.edited_at)})` : ""; + return `[${utcMinute(m.created_at)}] ${who}${edited}: ${m.body_md}`; +} + +function utcMinute(iso: string): string { + return `${iso.slice(0, 10)} ${iso.slice(11, 16)} UTC`; +} diff --git a/src/main/services/task-execution-service.ts b/src/main/services/task-execution-service.ts index 28271085..c016545d 100644 --- a/src/main/services/task-execution-service.ts +++ b/src/main/services/task-execution-service.ts @@ -14,7 +14,6 @@ import type { AgentToolContext } from "./agent-tools"; import { buildAgentInstructions, buildTaskPrompt, - type ChatContextBlock, type GoalRow, type ProjectRow, type SkillRuntimeSnapshotRow, @@ -25,6 +24,7 @@ import { type WorkspaceRow, } from "./task-prompts"; import { renderTaskHandoff } from "./agent-task-handoff"; +import { loadChatWakeContext } from "./chat-wake-context"; import { loadComposioTools } from "./composio/composio-tool-bridge"; import { connectorDisplayName } from "./composio/connector-brands"; import { ConnectorGrantsService } from "./composio/connector-grants"; @@ -134,7 +134,7 @@ export class TaskExecutionService { let skillRuntimeSnapshot: SkillRuntimeSnapshotRow | null = null; try { skillRuntimeSnapshot = this.resolveSkillRuntimeSnapshot(orgId, agent.id, runId); - const chatContext = this.loadChatContext(orgId, isolation); + const chatContext = loadChatWakeContext(this.db, orgId, agent.id, isolation.secretScopeJson); const toolContext: AgentToolContext = { orgId, agent: { id: agent.id, modelProviderId: agent.model_provider_id, modelId: agent.model_id }, @@ -523,39 +523,6 @@ export class TaskExecutionService { return runId; } - private loadChatContext(orgId: string, isolation: TaskRunIsolation): ChatContextBlock | null { - let scope: { sourceMessageId?: string; sourceThreadId?: string }; - try { - scope = JSON.parse(isolation.secretScopeJson); - } catch { - return null; - } - if (!scope.sourceThreadId) return null; - const thread = this.db.prepare("select space_id from chat_threads where id = ? and org_id = ?") - .get(scope.sourceThreadId, orgId) as { space_id: string } | undefined; - if (!thread) return null; - const messages = this.db.prepare(` - select m.body_md, m.sender_type, m.sender_id, - coalesce(a.name, e.display_name, m.sender_type) as sender_name - from chat_messages m - left join agents a on a.id = m.sender_id and m.sender_type = 'agent' - left join employees e on e.id = m.sender_id and m.sender_type = 'employee' - where m.thread_id = ? and m.org_id = ? - order by m.created_at asc - limit 20 - `).all(scope.sourceThreadId, orgId) as Array<{ body_md: string; sender_type: string; sender_name: string }>; - if (messages.length === 0) return null; - // Human principals are marked as the founder: an agent replying to its - // boss must know it is not talking to another agent. - const label = (m: { sender_type: string; sender_name: string }) => - m.sender_type === "employee" || m.sender_type === "board" ? `${m.sender_name} (founder)` : m.sender_name; - return { - threadId: scope.sourceThreadId, - spaceId: thread.space_id, - transcript: messages.map((m) => `[${label(m)}]: ${m.body_md}`).join("\n"), - }; - } - private completedRunCount(taskId: string, excludingRunId: string): number { const row = this.db.prepare("select count(*) as n from task_runs where task_id = ? and state = 'completed' and id <> ?") .get(taskId, excludingRunId) as { n: number }; diff --git a/src/main/services/task-prompts.ts b/src/main/services/task-prompts.ts index 328b9671..22330a38 100644 --- a/src/main/services/task-prompts.ts +++ b/src/main/services/task-prompts.ts @@ -71,7 +71,10 @@ export interface ChatContextBlock { threadId: string; /** The space the thread lives in — where the run's reply must land. */ spaceId: string; - transcript: string; + /** Messages up to and including the agent's last reply. */ + history: string; + /** Messages after it: what this run answers. */ + fresh: string; } export interface AgentInstructionContext { @@ -211,8 +214,11 @@ export function buildTaskPrompt(input: { if (input.chatContext) { lines.push( "", - "The conversation in this thread so far:", - input.chatContext.transcript, + "Earlier in this thread (context only — newer messages override it):", + input.chatContext.history || "(nothing before)", + "", + "New since your last reply in this thread — this is what you are answering:", + input.chatContext.fresh || "(nothing new since your last reply)", "", `You have been brought into this thread. Reply to the people here in your own voice — answer what they asked, give your honest take, push back if you disagree, or hand work over. Stay in the thread by passing thread_id="${input.chatContext.threadId}" to send_message. This is a conversation between colleagues, not a status report; only change a task's status if the work itself actually moved.`, ); diff --git a/tests/integration/ceo-founder-chat.test.ts b/tests/integration/ceo-founder-chat.test.ts index ac15b2ea..a0497408 100644 --- a/tests/integration/ceo-founder-chat.test.ts +++ b/tests/integration/ceo-founder-chat.test.ts @@ -121,7 +121,7 @@ describe("founder⇄CEO chat", () => { // The CEO's prompt names the human principal, not just another teammate. expect(capturedWakePrompts).toHaveLength(1); - expect(capturedWakePrompts[0]).toContain("[Yang (founder)]: How are we tracking against the MVP?"); + expect(capturedWakePrompts[0]).toMatch(/\] Yang \(employee, Board Chair\): How are we tracking against the MVP\?/); // The reply landed in the SAME pinned conversation and thread — to="Yang" // resolves the employee and ensureDirectSpace reuses the founder's line. diff --git a/tests/integration/chat-context.test.ts b/tests/integration/chat-context.test.ts new file mode 100644 index 00000000..446d73a9 --- /dev/null +++ b/tests/integration/chat-context.test.ts @@ -0,0 +1,102 @@ +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 { TaskExecutionService } from "../../src/main/services/task-execution-service"; +import { ChatService } from "../../src/main/services/chat-service"; +import { ensureCeoDirectSpace } from "../../src/main/services/ceo-direct-space"; +import { claimNextQueuedRun } from "../../src/main/app/task-queue-worker"; +import type { AgentTurnResult, ProviderToolTurnInput } from "../../src/main/services/provider-tool-turn"; + +const testVault = { + encryptSecret: (secret: string) => Buffer.from(secret, "utf8"), + decryptSecret: (ciphertext: Buffer) => ciphertext.toString("utf8"), +}; + +async function founderLine() { + const wakePrompts: string[] = []; + const turn = async (input: ProviderToolTurnInput): Promise => { + const first = input.messages.find((message) => message.role === "user"); + const prompt = first && first.role === "user" ? first.text : ""; + if (/thread_id="/.test(prompt) && input.messages.length === 1) wakePrompts.push(prompt); + return { text: `Reply ${wakePrompts.length}.`, toolCalls: [], rawResponseJson: "{}" }; + }; + const connection = new DatabaseConnection(join(tmpdir(), `ac-chatctx-${randomUUID()}.db`)); + const providers = new ProviderService(connection.db, providerAdapters, testVault, undefined, turn); + const onboarding = new OnboardingService({ db: connection.db, saveProviderCredential: (input) => providers.saveCredential(input) }); + const execution = new TaskExecutionService(connection.db, providers); + const chat = new ChatService(connection.db); + 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 founderId = (connection.db.prepare("select id from employees where org_id = ?").get(orgId) as { id: string }).id; + const spaceId = ensureCeoDirectSpace(connection.db, orgId)!; + const kickoff = claimNextQueuedRun(connection); + await execution.executeRun(kickoff!.org_id, kickoff!.id); + const say = (bodyMd: string, threadId: string | null) => { + const id = chat.sendMessage({ orgId, spaceId, threadId, senderType: "employee", senderId: founderId, bodyMd, attachmentFileIds: [] }); + return (connection.db.prepare("select thread_id from chat_messages where id = ?").get(id) as { thread_id: string }).thread_id; + }; + const runWake = async () => { + const wake = connection.db.prepare( + "select id from task_runs where org_id = ? and agent_id = ? and trigger = 'chat_mention' and state = 'queued'", + ).get(orgId, ceoId) as { id: string }; + await execution.executeRun(orgId, wake.id); + }; + const edit = (bodyMd: string, newBody: string) => { + const row = connection.db.prepare("select id from chat_messages where body_md = ?").get(bodyMd) as { id: string }; + chat.editMessage({ orgId, messageId: row.id, bodyMd: newBody, actorType: "employee", actorId: founderId }); + }; + return { connection, wakePrompts, say, runWake, edit }; +} + +describe("chat wake context", () => { + test("separates what the agent already answered from what is new, with times and senders", async () => { + const h = await founderLine(); + const threadId = h.say("How are we tracking?", null); + await h.runWake(); + h.say("Also, who owns billing?", threadId); + h.say("And ship date?", threadId); + await h.runWake(); + + const prompt = h.wakePrompts[1]; + const earlier = prompt.slice(prompt.indexOf("Earlier in this thread"), prompt.indexOf("New since your last reply")); + const fresh = prompt.slice(prompt.indexOf("New since your last reply")); + expect(earlier).toContain("How are we tracking?"); + expect(earlier).toContain("Reply 1."); + expect(fresh).toContain("Also, who owns billing?"); + expect(fresh).toContain("And ship date?"); + expect(fresh).not.toContain("How are we tracking?"); + expect(fresh).toMatch(/\[\d{4}-\d{2}-\d{2} \d{2}:\d{2} UTC\] Yang \(employee, Board Chair\)/); + expect(earlier).toMatch(/\] CEO \(agent, [^)]+\): Reply 1\./); + h.connection.close(); + }); + + test("a message corrected after the agent's reply counts as new", async () => { + const h = await founderLine(); + const threadId = h.say("Ship on Monday.", null); + await h.runWake(); + h.edit("Ship on Monday.", "Ship on Friday."); + h.say("See the correction above.", threadId); + await h.runWake(); + + const prompt = h.wakePrompts[1]; + const fresh = prompt.slice(prompt.indexOf("New since your last reply")); + expect(fresh).toMatch(/Yang \(employee, Board Chair\) \(edited \d{4}-\d{2}-\d{2} \d{2}:\d{2} UTC\): Ship on Friday\./); + expect(fresh).not.toContain("Reply 1."); + h.connection.close(); + }); +});