A chat-wake reply is delivered by the system, not by prompt compliance
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.
This commit is contained in:
1 parent
75c73ce6f0
commit
7de378b41c
4 files changed
+161
No files matched your search
@@ -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;
|
||||
|
||||
|
||||
@@ -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"),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
|
||||
@@ -21,6 +21,23 @@ function call(name: string, args: Record<string, unknown>): 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<AgentTurnResult> {
|
||||
return async (input: ProviderToolTurnInput): Promise<AgentTurnResult> => {
|
||||
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();
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user