Answer plain user messages in channels via a channel-anchor wake
A human message in a channel woke agents only through @mentions, DM counterparts, or prior thread participants - and the flat composer always sends threadId=null, so typing in #general summoned nobody and the room stayed silent. - Track which agents each wake path summoned in one woken set (also prevents latent double-wakes between the paths) - When a non-slash human message in a channel wakes nobody, wake the channel's most senior active agent member with an enabled provider (c_suite > director > lead > staff, then hire date); they can pull teammates in with @mentions - Log when a channel has no runnable agent so silence is diagnosable - Integration test: plain #general message wakes the CEO who replies; an explicit mention still wakes exactly one agent
This commit is contained in:
1 parent
349547a0de
commit
55c043f8b4
3 files changed
+161
-13
No files matched your search
@@ -30,6 +30,7 @@ export interface FileAttachmentRow {
|
||||
export interface SpaceRow {
|
||||
id: string;
|
||||
org_id: string;
|
||||
name: string;
|
||||
visibility: string;
|
||||
space_type: string;
|
||||
}
|
||||
|
||||
@@ -166,12 +166,19 @@ export class ChatService {
|
||||
recordChatMentions(this.db, id, mentions);
|
||||
const slashResultRef = slash ? this.slashRouter.execute(input, space, sender, mentions, attachments, slash, id, threadId, now) : null;
|
||||
if (slash) this.slashRouter.recordInvocation(id, slash, slashResultRef, now);
|
||||
const woken = new Set<string>();
|
||||
if (!slash || slash.command !== "agent") {
|
||||
this.wakeAgentMentions(input, mentions, id, threadId, now);
|
||||
if (space.space_type === "direct") this.wakeDirectCounterparts(input, mentions, id, threadId, now);
|
||||
this.wakeAgentMentions(input, mentions, id, threadId, now, woken);
|
||||
if (space.space_type === "direct") this.wakeDirectCounterparts(input, id, threadId, now, woken);
|
||||
}
|
||||
if (input.senderType === "employee" || input.senderType === "board") {
|
||||
this.wakeThreadParticipants(input, mentions, id, threadId, now);
|
||||
this.wakeThreadParticipants(input, id, threadId, now, woken);
|
||||
// A human speaking in a channel expects an answer. If neither mentions
|
||||
// nor thread history summoned anyone, the channel's most senior agent
|
||||
// member picks it up (and can @mention teammates to pull them in).
|
||||
if (!slash && space.space_type !== "direct" && woken.size === 0) {
|
||||
this.wakeChannelAnchor(input, space, id, threadId, now);
|
||||
}
|
||||
}
|
||||
this.handoffs.record({
|
||||
orgId: input.orgId,
|
||||
@@ -238,7 +245,7 @@ export class ChatService {
|
||||
this.messageMutations.deleteMessage(input);
|
||||
}
|
||||
|
||||
private wakeAgentMentions(input: SendMessageInput, mentions: ResolvedMention[], messageId: string, threadId: string, now: string): void {
|
||||
private wakeAgentMentions(input: SendMessageInput, mentions: ResolvedMention[], messageId: string, threadId: string, now: string, woken: Set<string>): void {
|
||||
const agentMentions = mentions.filter(
|
||||
(m) => m.target.type === "agent" && !(input.senderType === "agent" && input.senderId === m.target.id),
|
||||
);
|
||||
@@ -247,6 +254,7 @@ export class ChatService {
|
||||
for (const mention of agentMentions) {
|
||||
try {
|
||||
this.agentWake.wakeFromMention(input, mention.target.id, messageId, threadId, now);
|
||||
woken.add(mention.target.id);
|
||||
this.db.prepare(
|
||||
"update chat_mentions set wake_state = 'woken' where message_id = ? and target_type = 'agent' and target_id = ?",
|
||||
).run(messageId, mention.target.id);
|
||||
@@ -258,20 +266,18 @@ export class ChatService {
|
||||
}
|
||||
}
|
||||
|
||||
private wakeThreadParticipants(input: SendMessageInput, alreadyMentioned: ResolvedMention[], messageId: string, threadId: string, now: string): void {
|
||||
private wakeThreadParticipants(input: SendMessageInput, messageId: string, threadId: string, now: string, woken: Set<string>): void {
|
||||
if (!input.threadId) return;
|
||||
const mentionedAgentIds = new Set(
|
||||
alreadyMentioned.filter((m) => m.target.type === "agent").map((m) => m.target.id),
|
||||
);
|
||||
const participants = this.db.prepare(`
|
||||
select distinct m.sender_id as agent_id
|
||||
from chat_messages m
|
||||
where m.thread_id = ? and m.org_id = ? and m.sender_type = 'agent' and m.deleted_at is null
|
||||
`).all(input.threadId, input.orgId) as Array<{ agent_id: string }>;
|
||||
for (const row of participants) {
|
||||
if (mentionedAgentIds.has(row.agent_id)) continue;
|
||||
if (woken.has(row.agent_id)) continue;
|
||||
try {
|
||||
this.agentWake.wakeFromMention(input, row.agent_id, messageId, threadId, now);
|
||||
woken.add(row.agent_id);
|
||||
} catch {
|
||||
// Agent may not have an enabled provider.
|
||||
}
|
||||
@@ -284,25 +290,54 @@ export class ChatService {
|
||||
* — the same social contract as any messenger. Mentioned agents were
|
||||
* already woken; the autonomous turn budget still bounds agent↔agent runs.
|
||||
*/
|
||||
private wakeDirectCounterparts(input: SendMessageInput, mentions: ResolvedMention[], messageId: string, threadId: string, now: string): void {
|
||||
private wakeDirectCounterparts(input: SendMessageInput, messageId: string, threadId: string, now: string, woken: Set<string>): void {
|
||||
if (input.senderType === "agent" && this.autonomousThreadTurnsExhausted(input.orgId, threadId)) return;
|
||||
const mentionedAgentIds = new Set(mentions.filter((m) => m.target.type === "agent").map((m) => m.target.id));
|
||||
const members = this.db.prepare(`
|
||||
select principal_id as agent_id
|
||||
from chat_memberships
|
||||
where space_id = ? and principal_type = 'agent'
|
||||
`).all(input.spaceId) as Array<{ agent_id: string }>;
|
||||
for (const member of members) {
|
||||
if (mentionedAgentIds.has(member.agent_id)) continue;
|
||||
if (woken.has(member.agent_id)) continue;
|
||||
if (input.senderType === "agent" && input.senderId === member.agent_id) continue;
|
||||
try {
|
||||
this.agentWake.wakeFromMention(input, member.agent_id, messageId, threadId, now);
|
||||
woken.add(member.agent_id);
|
||||
} catch {
|
||||
// Agent may not have an enabled provider.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Fallback responder for a human message in a channel that named no one:
|
||||
* the most senior active agent member with an enabled provider answers,
|
||||
* exactly like a room going quiet until the highest-ranking person speaks.
|
||||
*/
|
||||
private wakeChannelAnchor(input: SendMessageInput, space: SpaceRow, messageId: string, threadId: string, now: string): void {
|
||||
const anchor = this.db.prepare(`
|
||||
select a.id as agent_id
|
||||
from chat_memberships cm
|
||||
join agents a on a.id = cm.principal_id and a.org_id = ?
|
||||
join providers p on p.id = a.model_provider_id and p.org_id = a.org_id
|
||||
where cm.space_id = ? and cm.principal_type = 'agent'
|
||||
and a.lifecycle = 'active'
|
||||
and p.enabled = 1
|
||||
order by case a.role_level when 'c_suite' then 0 when 'director' then 1 when 'lead' then 2 else 3 end,
|
||||
a.created_at asc
|
||||
limit 1
|
||||
`).get(input.orgId, input.spaceId) as { agent_id: string } | undefined;
|
||||
if (!anchor) {
|
||||
console.info(`[chat] no runnable agent member in #${space.name}; the message stays unanswered until one is hired or a provider is enabled.`);
|
||||
return;
|
||||
}
|
||||
try {
|
||||
this.agentWake.wakeFromMention(input, anchor.agent_id, messageId, threadId, now);
|
||||
} catch (error) {
|
||||
console.warn(`[chat] waking channel anchor for #${space.name} failed: ${error instanceof Error ? error.message : String(error)}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* True once a thread holds MAX_AUTONOMOUS_THREAD_AGENT_TURNS agent messages
|
||||
* since the last human or board message — the point past which agent↔agent
|
||||
@@ -337,7 +372,7 @@ export class ChatService {
|
||||
}
|
||||
|
||||
private requireSpace(orgId: string, spaceId: string): SpaceRow {
|
||||
const space = this.db.prepare("select id, org_id, visibility, space_type from chat_spaces where id = ? and org_id = ?")
|
||||
const space = this.db.prepare("select id, org_id, name, visibility, space_type from chat_spaces where id = ? and org_id = ?")
|
||||
.get(spaceId, orgId) as SpaceRow | undefined;
|
||||
if (!space) throw new Error("Chat space was not found.");
|
||||
return space;
|
||||
|
||||
@@ -0,0 +1,112 @@
|
||||
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 { ChatService } from "../../src/main/services/chat-service";
|
||||
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 { canonicalMentionHandle } from "../../src/shared/mention-handles";
|
||||
import type { AgentTurnResult, ProviderToolTurnInput, ToolCall } from "../../src/main/services/provider-tool-turn";
|
||||
|
||||
const testVault = {
|
||||
encryptSecret: (secret: string) => Buffer.from(secret, "utf8"),
|
||||
decryptSecret: (ciphertext: Buffer) => ciphertext.toString("utf8"),
|
||||
};
|
||||
|
||||
function call(name: string, args: Record<string, unknown>): ToolCall {
|
||||
return { id: `call_${randomUUID()}`, name, arguments: args };
|
||||
}
|
||||
|
||||
/** CEO answers in the thread the wake prompt names, then finishes. */
|
||||
function scriptedCeoReply(): (input: ProviderToolTurnInput) => Promise<AgentTurnResult> {
|
||||
let turn = 0;
|
||||
return async (input: ProviderToolTurnInput): Promise<AgentTurnResult> => {
|
||||
const firstUser = input.messages.find((message) => message.role === "user");
|
||||
const threadId = firstUser ? /thread_id="([^"]+)"/.exec(firstUser.text)?.[1] : undefined;
|
||||
const scripted = turn === 0
|
||||
? { text: "Answering.", toolCalls: [call("send_message", { channel: "general", body: "Here is where we stand: kickoff is planned.", ...(threadId ? { thread_id: threadId } : {}) })] }
|
||||
: { text: "Done.", toolCalls: [] };
|
||||
turn += 1;
|
||||
return { text: scripted.text, toolCalls: scripted.toolCalls, rawResponseJson: "{}" };
|
||||
};
|
||||
}
|
||||
|
||||
describe("channel anchor wake", () => {
|
||||
test("a plain user message in a channel wakes the most senior member, who replies", async () => {
|
||||
const connection = new DatabaseConnection(join(tmpdir(), `ac-anchor-${randomUUID()}.db`));
|
||||
const providers = new ProviderService(connection.db, providerAdapters, testVault, undefined, scriptedCeoReply());
|
||||
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 = "";
|
||||
connection.transaction(() => {
|
||||
orgId = onboarding.bootstrap({
|
||||
companyName: "OrbitOps AI",
|
||||
companyDescription: "AI-run operations company.",
|
||||
employeeName: "Yang",
|
||||
workspacePath: "",
|
||||
executionMode: "ai_ceo",
|
||||
autoApproveHires: true,
|
||||
reviewDeliverables: false,
|
||||
goalTitle: "Stand up a service business.",
|
||||
goalDescription: "Win the first customers.",
|
||||
providerKey: "openai",
|
||||
providerSecret: "sk-test-key",
|
||||
providerModel: "gpt-5.4",
|
||||
}).orgId;
|
||||
});
|
||||
|
||||
const employee = connection.db.prepare("select id from employees where org_id = ?").get(orgId) as { id: string };
|
||||
const general = connection.db.prepare("select id from chat_spaces where org_id = ? and name = 'general'").get(orgId) as { id: string };
|
||||
const ceo = connection.db.prepare("select id from agents where org_id = ? and role_level = 'c_suite'").get(orgId) as { id: string };
|
||||
|
||||
// A plain message: no @mention, no thread. Somebody must still answer.
|
||||
connection.transaction(() => {
|
||||
chat.sendMessage({
|
||||
orgId,
|
||||
spaceId: general.id,
|
||||
threadId: null,
|
||||
senderType: "employee",
|
||||
senderId: employee.id,
|
||||
bodyMd: "Where do we stand on the goal?",
|
||||
attachmentFileIds: [],
|
||||
});
|
||||
});
|
||||
|
||||
const wake = connection.db.prepare(
|
||||
"select id, agent_id, state, trigger from task_runs where org_id = ? and trigger = 'chat_mention'",
|
||||
).all(orgId) as Array<{ id: string; agent_id: string; state: string; trigger: string }>;
|
||||
expect(wake).toHaveLength(1);
|
||||
expect(wake[0]).toMatchObject({ agent_id: ceo.id, state: "queued" });
|
||||
|
||||
// The woken CEO answers in #general.
|
||||
await taskExecution.executeRun(orgId, wake[0].id);
|
||||
const reply = connection.db.prepare(
|
||||
"select body_md from chat_messages where org_id = ? and space_id = ? and sender_type = 'agent' and sender_id = ?",
|
||||
).get(orgId, general.id, ceo.id) as { body_md: string } | undefined;
|
||||
expect(reply?.body_md).toContain("Here is where we stand");
|
||||
|
||||
// An explicit @mention wakes only the mentioned agent - no anchor duplicate.
|
||||
connection.transaction(() => {
|
||||
chat.sendMessage({
|
||||
orgId,
|
||||
spaceId: general.id,
|
||||
threadId: null,
|
||||
senderType: "employee",
|
||||
senderId: employee.id,
|
||||
bodyMd: `@${canonicalMentionHandle("CEO")} thanks, keep me posted.`,
|
||||
attachmentFileIds: [],
|
||||
});
|
||||
});
|
||||
const runsAfterMention = connection.db.prepare(
|
||||
"select count(*) as count from task_runs where org_id = ? and trigger = 'chat_mention'",
|
||||
).get(orgId) as { count: number };
|
||||
expect(runsAfterMention.count).toBe(2);
|
||||
|
||||
connection.close();
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user