From 7de378b41cb31f19c4ad1a357ee8433fe9907c23 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Thu, 16 Jul 2026 11:46:53 -0400 Subject: [PATCH] A chat-wake reply is delivered by the system, not by prompt compliance MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A wake run whose model answered in plain final text (the coding-tuned Kimi family does this instead of calling send_message) completed 'successfully', closed its vehicle task, and left the founder staring at a silent conversation — the reply lived only in provider_requests. deliverChatReplyIfMissing now runs before a chat-wake run may complete: when the run never posted into the source thread (run-start watermark, so a send_message it did make suppresses the fallback), its final text is delivered as the agent's reply. A delivery failure fails the run loudly instead of closing it on silence. --- src/main/services/chat-agent-wake-service.ts | 46 ++++++++++ src/main/services/task-execution-service.ts | 23 +++++ src/main/services/task-prompts.ts | 2 + tests/integration/ceo-founder-chat.test.ts | 90 ++++++++++++++++++++ 4 files changed, 161 insertions(+) diff --git a/src/main/services/chat-agent-wake-service.ts b/src/main/services/chat-agent-wake-service.ts index ff5cad2f..01f4fd4f 100644 --- a/src/main/services/chat-agent-wake-service.ts +++ b/src/main/services/chat-agent-wake-service.ts @@ -4,6 +4,7 @@ import { RISK_LEVEL, TASK_PRIORITY, TASK_STATUS } from "../../shared/domain"; import { createId } from "../../shared/ids"; import type { ResolvedMention } from "./chat-mentions"; import type { SlashInvocation } from "./chat-service-internal"; +import type { ChatService } from "./chat-service"; import { OrganizationService } from "./organization-service"; interface RunnableAgentRow { @@ -63,6 +64,51 @@ export function expireStaleChatWakes(db: DatabaseSync, now: string): number { return stale.length; } +/** + * A chat wake's deliverable is a reply VISIBLE in the source thread. The model + * is asked to post it with send_message, but a run that answers in plain final + * text has still produced the reply — delivery is the system's contract, not + * the model's prompt compliance. When the run never posted into the thread + * (checked by the run's start watermark, so any send_message it did make + * counts), its final text is delivered as the agent's message; otherwise the + * asker watches the run "complete" while the conversation stays silent. + * Returns true when a fallback message was posted. A delivery failure throws, + * so the run fails loudly instead of completing with nothing said. + */ +export function deliverChatReplyIfMissing( + db: DatabaseSync, + chat: ChatService, + input: { + orgId: string; + agentId: string; + spaceId: string; + threadId: string; + finalText: string; + /** ISO timestamp captured before the run's first provider turn. */ + runStartedAt: string; + }, +): boolean { + const body = input.finalText.trim(); + if (!body) return false; + const posted = db.prepare(` + select 1 from chat_messages + where org_id = ? and thread_id = ? and sender_type = 'agent' and sender_id = ? and created_at >= ? + limit 1 + `).get(input.orgId, input.threadId, input.agentId, input.runStartedAt); + if (posted) return false; + console.info(`[chat] the reply run never posted into thread ${input.threadId}; delivering its final text as the reply.`); + chat.sendMessage({ + orgId: input.orgId, + spaceId: input.spaceId, + threadId: input.threadId, + senderType: "agent", + senderId: input.agentId, + bodyMd: body, + attachmentFileIds: [], + }); + return true; +} + export class ChatAgentWakeService { private readonly organization: OrganizationService; diff --git a/src/main/services/task-execution-service.ts b/src/main/services/task-execution-service.ts index 9f87d49f..088d160f 100644 --- a/src/main/services/task-execution-service.ts +++ b/src/main/services/task-execution-service.ts @@ -7,6 +7,7 @@ import { AuditService } from "./audit-service"; import { ProviderService } from "./provider-service"; import { OrganizationService } from "./organization-service"; import { ChatService } from "./chat-service"; +import { deliverChatReplyIfMissing } from "./chat-agent-wake-service"; import { SkillService } from "./skill-service"; import { runAgentLoop } from "./agent-run-loop"; import type { AgentToolContext } from "./agent-tools"; @@ -151,6 +152,9 @@ export class TaskExecutionService { .map(connectorDisplayName); const language = this.appLanguage(); const externalTools = await loadComposioTools(this.db, orgId, agent.id); + // Watermark for the reply-delivery check below: anything the agent posts + // into the source thread from here on counts as its reply. + const runStartedAt = new Date().toISOString(); const result = await runAgentLoop( { providers: this.providers, @@ -183,6 +187,21 @@ export class TaskExecutionService { }, ); providerRequestId = result.lastRequestId; + // A chat wake's deliverable is the visible reply. If the model answered + // in plain text instead of calling send_message, deliver that text into + // the source thread BEFORE the run may complete — a delivery failure + // fails the run rather than closing it on silence. + if (chatContext) { + const delivered = deliverChatReplyIfMissing(this.db, this.chat, { + orgId, + agentId: agent.id, + spaceId: chatContext.spaceId, + threadId: chatContext.threadId, + finalText: result.finalText, + runStartedAt, + }); + if (delivered) this.notifyDataChanged(); + } this.completeRun(orgId, runId, result.lastRequestId); this.clearProviderCooldown(provider); this.audit.record({ @@ -511,6 +530,9 @@ export class TaskExecutionService { 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 @@ -528,6 +550,7 @@ export class TaskExecutionService { 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"), }; } diff --git a/src/main/services/task-prompts.ts b/src/main/services/task-prompts.ts index 2d123086..e00329ac 100644 --- a/src/main/services/task-prompts.ts +++ b/src/main/services/task-prompts.ts @@ -68,6 +68,8 @@ export interface SkillRuntimeSnapshotRow { export interface ChatContextBlock { threadId: string; + /** The space the thread lives in — where the run's reply must land. */ + spaceId: string; transcript: string; } diff --git a/tests/integration/ceo-founder-chat.test.ts b/tests/integration/ceo-founder-chat.test.ts index ed7ca0b3..ac15b2ea 100644 --- a/tests/integration/ceo-founder-chat.test.ts +++ b/tests/integration/ceo-founder-chat.test.ts @@ -21,6 +21,23 @@ function call(name: string, args: Record): ToolCall { return { id: `call_${randomUUID()}`, name, arguments: args }; } +/** + * Scripts a CEO who never calls send_message: the chat-wake run is answered + * with plain final text only — the way a direct-answering model (e.g. a + * coding-tuned one) responds when it treats the wake as a question, not a + * conversation to post into. + */ +function plainTextCeo(): (input: ProviderToolTurnInput) => Promise { + return async (input: ProviderToolTurnInput): Promise => { + const firstUser = input.messages.find((message) => message.role === "user"); + const prompt = firstUser?.text ?? ""; + if (!/thread_id="/.test(prompt)) { + return { text: "Kickoff noted.", toolCalls: [], rawResponseJson: "{}" }; + } + return { text: "We are on track — the kickoff wave is planned.", toolCalls: [], rawResponseJson: "{}" }; + }; +} + /** * Scripts the CEO: the kickoff run ends quietly; the chat-wake run (its * prompt carries the thread transcript) is captured and answered with a @@ -123,4 +140,77 @@ describe("founder⇄CEO chat", () => { connection.close(); }); + + test("a CEO who answers in plain text still has the reply delivered into the founder's thread", async () => { + const connection = new DatabaseConnection(join(tmpdir(), `ac-ceo-plain-${randomUUID()}.db`)); + const providers = new ProviderService(connection.db, providerAdapters, testVault, undefined, plainTextCeo()); + const onboarding = new OnboardingService({ db: connection.db, saveProviderCredential: (input) => providers.saveCredential(input) }); + const taskExecution = 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 for SMBs.", + 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 founder = connection.db.prepare("select id from employees where org_id = ?").get(orgId) as { id: string }; + const pinnedSpaceId = ensureCeoDirectSpace(connection.db, orgId); + + // Drain the kickoff run so the next queued run is the chat wake. + const kickoff = claimNextQueuedRun(connection); + await taskExecution.executeRun(kickoff!.org_id, kickoff!.id); + + const founderMessageId = chat.sendMessage({ + orgId, + spaceId: pinnedSpaceId!, + threadId: null, + senderType: "employee", + senderId: founder.id, + bodyMd: "How are we tracking against the MVP?", + attachmentFileIds: [], + }); + const founderThreadId = (connection.db.prepare("select thread_id from chat_messages where id = ?") + .get(founderMessageId) as { thread_id: string }).thread_id; + + 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 taskExecution.executeRun(orgId, wake.id); + + // The model answered in its final text and never called send_message; the + // reply must still land in the founder's thread — exactly once. + const replies = connection.db.prepare( + "select space_id, thread_id, body_md from chat_messages where org_id = ? and sender_type = 'agent' and sender_id = ?", + ).all(orgId, ceoId) as Array<{ space_id: string; thread_id: string; body_md: string }>; + expect(replies).toHaveLength(1); + expect(replies[0].space_id).toBe(pinnedSpaceId); + expect(replies[0].thread_id).toBe(founderThreadId); + expect(replies[0].body_md).toContain("on track"); + + // The run completed and closed its wake task, not failed on a delivery it made. + const wakeRun = connection.db.prepare("select state, error_message from task_runs where id = ?") + .get(wake.id) as { state: string; error_message: string | null }; + expect(wakeRun).toMatchObject({ state: "completed", error_message: null }); + const wakeTask = connection.db.prepare( + "select t.status from tasks t join task_runs r on r.task_id = t.id where r.id = ?", + ).get(wake.id) as { status: string }; + expect(wakeTask).toEqual({ status: "done" }); + + connection.close(); + }); });