diff --git a/src/main/services/chat-service-internal.ts b/src/main/services/chat-service-internal.ts index 20442d53..15ddcb5f 100644 --- a/src/main/services/chat-service-internal.ts +++ b/src/main/services/chat-service-internal.ts @@ -30,6 +30,7 @@ export interface FileAttachmentRow { export interface SpaceRow { id: string; org_id: string; + name: string; visibility: string; space_type: string; } diff --git a/src/main/services/chat-service.ts b/src/main/services/chat-service.ts index bbb6e577..d6c33071 100644 --- a/src/main/services/chat-service.ts +++ b/src/main/services/chat-service.ts @@ -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(); 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): 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): 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): 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; diff --git a/tests/integration/channel-anchor.test.ts b/tests/integration/channel-anchor.test.ts new file mode 100644 index 00000000..a12c300d --- /dev/null +++ b/tests/integration/channel-anchor.test.ts @@ -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): ToolCall { + return { id: `call_${randomUUID()}`, name, arguments: args }; +} + +/** CEO answers in the thread the wake prompt names, then finishes. */ +function scriptedCeoReply(): (input: ProviderToolTurnInput) => Promise { + let turn = 0; + return async (input: ProviderToolTurnInput): Promise => { + 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(); + }); +});