Queue agent runs from chat mentions

This commit is contained in:
ZacharyZhang-NY committed 2026-06-07 13:34:27 -04:00
1 parent ab546f822d
commit 8eec454927
3 files changed
+242 -22

No files matched your search

@@ -0,0 +1,184 @@
import type { DatabaseSync } from "node:sqlite";
import type { SendMessageInput } from "../../shared/commands";
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 { OrganizationService } from "./organization-service";
interface RunnableAgentRow {
id: string;
name: string;
model_provider_id: string | null;
workspace_id: string | null;
}
interface WorkspaceContext {
id: string;
name: string;
local_path: string;
repo_ref: string;
project_id: string | null;
goal_id: string | null;
}
export interface AgentWakeResult {
targetId: string;
taskId: string;
runId: string;
}
export class ChatAgentWakeService {
private readonly organization: OrganizationService;
constructor(private readonly db: DatabaseSync) {
this.organization = new OrganizationService(db);
}
wakeFromSlash(input: SendMessageInput, mentions: ResolvedMention[], slash: SlashInvocation, messageId: string, threadId: string, now: string): AgentWakeResult {
const targetId = this.requireAgentMention(mentions);
const agent = this.requireRunnableAgent(input.orgId, targetId);
const workspace = this.workspaceForAgent(input.orgId, agent.workspace_id);
const taskId = this.organization.createTask({
orgId: input.orgId,
title: this.taskTitle(agent.name, slash),
description: this.taskDescription(input, slash, messageId, threadId),
status: TASK_STATUS.BACKLOG,
assigneeAgentId: agent.id,
risk: RISK_LEVEL.LOW,
projectId: workspace?.project_id ?? null,
goalId: workspace?.goal_id ?? null,
workspaceId: workspace?.id ?? null,
priority: TASK_PRIORITY.HIGH,
requiresFilesystem: false,
});
const runId = this.createQueuedRun({ orgId: input.orgId, taskId, agent, workspace, messageId, threadId, now });
this.recordPresence(agent.id, slash, now);
return { targetId: agent.id, taskId, runId };
}
private requireAgentMention(mentions: ResolvedMention[]): string {
const target = mentions.find((mention) => mention.target.type === "agent")?.target;
if (!target || target.type !== "agent") throw new Error("Slash command /agent requires an agent mention.");
return target.id;
}
private requireRunnableAgent(orgId: string, agentId: string): RunnableAgentRow {
const agent = this.db.prepare(`
select a.id, a.name, a.model_provider_id, a.workspace_id
from agents a
join providers p on p.id = a.model_provider_id and p.org_id = a.org_id
where a.id = ?
and a.org_id = ?
and a.lifecycle = 'active'
and p.enabled = 1
limit 1
`).get(agentId, orgId) as RunnableAgentRow | undefined;
if (!agent) throw new Error("Slash command /agent requires an active agent with an enabled provider.");
return agent;
}
private workspaceForAgent(orgId: string, workspaceId: string | null): WorkspaceContext | null {
if (!workspaceId) return null;
const workspace = this.workspaceById(orgId, workspaceId);
if (!workspace) throw new Error("Agent workspace was not found.");
return workspace;
}
private workspaceById(orgId: string, workspaceId: string): WorkspaceContext | null {
return (this.db.prepare(`
select w.id, w.name, w.local_path, w.repo_ref, w.project_id, p.goal_id
from workspaces w
left join projects p on p.id = w.project_id and p.org_id = w.org_id
where w.org_id = ? and w.id = ?
limit 1
`).get(orgId, workspaceId) as WorkspaceContext | undefined) ?? null;
}
private createQueuedRun(input: {
orgId: string;
taskId: string;
agent: RunnableAgentRow;
workspace: WorkspaceContext | null;
messageId: string;
threadId: string;
now: string;
}): string {
const providerId = input.agent.model_provider_id;
if (!providerId) throw new Error("Slash command /agent requires an enabled provider.");
const serviceIdentityId = this.requireServiceIdentity(input.orgId, input.agent.id);
const runId = createId("run");
const executionContextId = createId("ctx");
const workspaceMountJson = JSON.stringify(input.workspace ? [{
workspaceId: input.workspace.id,
name: input.workspace.name,
localPath: input.workspace.local_path,
repoRef: input.workspace.repo_ref,
mode: "read_only",
}] : []);
const secretScopeJson = JSON.stringify({
serviceIdentityId,
providerIds: [providerId],
scope: "chat_mention",
credentialAccess: "provider_runtime_only",
sourceMessageId: input.messageId,
sourceThreadId: input.threadId,
});
this.db.prepare(`
insert into task_runs(
id, org_id, task_id, agent_id, provider_id, provider_request_id,
execution_context_id, service_identity_id, workspace_mount_json, secret_scope_json,
trigger, state, created_at, started_at, completed_at, error_message
)
values (?, ?, ?, ?, ?, null, ?, ?, ?, ?, ?, ?, ?, null, null, null)
`).run(
runId,
input.orgId,
input.taskId,
input.agent.id,
providerId,
executionContextId,
serviceIdentityId,
workspaceMountJson,
secretScopeJson,
"chat_mention",
"queued",
input.now,
);
return runId;
}
private requireServiceIdentity(orgId: string, agentId: string): string {
const row = this.db.prepare("select id from service_identities where org_id = ? and agent_id = ?")
.get(orgId, agentId) as { id: string } | undefined;
if (!row) throw new Error("Agent service identity was not found.");
return row.id;
}
private recordPresence(agentId: string, slash: SlashInvocation, now: string): void {
this.db.prepare(`
insert into chat_presence(principal_type, principal_id, status, status_text, last_seen_at)
values (?, ?, ?, ?, ?)
on conflict(principal_type, principal_id)
do update set status = excluded.status, status_text = excluded.status_text, last_seen_at = excluded.last_seen_at
`).run("agent", agentId, "thinking", slash.args.trim().slice(0, 160), now);
}
private taskTitle(agentName: string, slash: SlashInvocation): string {
const details = slash.args.replace(/^@[^\s]+\s*/, "").trim();
const title = details || `reply as ${agentName}`;
return `Chat mention: ${title.length > 64 ? `${title.slice(0, 61)}...` : title}`;
}
private taskDescription(input: SendMessageInput, slash: SlashInvocation, messageId: string, threadId: string): string {
return [
`Chat command: /agent ${slash.args.trim()}`,
`Source message: ${messageId}`,
`Source thread: ${threadId}`,
`Source space: ${input.spaceId}`,
`Sender: ${input.senderType}${input.senderId ? `:${input.senderId}` : ""}`,
"",
"Reply in the source thread with send_message and pass the thread_id above.",
].join("\n");
}
}
@@ -32,16 +32,9 @@ import { buildChatMessageStructure } from "./chat-message-structure";
import { ChatMessageMutationService } from "./chat-message-mutation-service";
import { recordChatMentions, resolveChatMentions, type ResolvedMention } from "./chat-mentions";
import { FileHandoffService, type FileHandoffAttachment } from "./file-handoff-service";
import { ChatAgentWakeService } from "./chat-agent-wake-service";
const SPACE_TYPES = new Set<CreateChatSpaceInput["spaceType"]>([
"company",
"division",
"department",
"project",
"pod",
"ad_hoc",
"direct",
]);
const SPACE_TYPES = new Set<CreateChatSpaceInput["spaceType"]>(["company", "division", "department", "project", "pod", "ad_hoc", "direct"]);
const SPACE_VISIBILITIES = new Set<CreateChatSpaceInput["visibility"]>(["workspace", "restricted"]);
export class ChatService {
@@ -51,6 +44,7 @@ export class ChatService {
private readonly handoffs: FileHandoffService;
private readonly messageMutations: ChatMessageMutationService;
private readonly identity: IdentityAccessService;
private readonly agentWake: ChatAgentWakeService;
constructor(private readonly db: DatabaseSync) {
this.audit = new AuditService(db);
@@ -59,6 +53,7 @@ export class ChatService {
this.handoffs = new FileHandoffService(db);
this.messageMutations = new ChatMessageMutationService(db);
this.identity = new IdentityAccessService(db);
this.agentWake = new ChatAgentWakeService(db);
}
createSpace(input: CreateChatSpaceInput): string {
@@ -232,7 +227,7 @@ export class ChatService {
if (slash.command === "approve" || slash.command === "deny") {
return this.decideApprovalFromSlash(input, sender, slash);
}
if (slash.command === "agent") return this.routeAgentSlash(input, mentions, slash, messageId, now);
if (slash.command === "agent") return this.routeAgentSlash(input, mentions, slash, messageId, threadId, now);
if (slash.command === "skill") return this.routeSkillSlash(input, slash, messageId);
if (slash.command === "file") return this.routeFileSlash(input, attachments, slash, messageId);
if (slash.command === "runbook") return this.routeRunbookSlash(input, attachments, slash, messageId);
@@ -255,18 +250,17 @@ export class ChatService {
mentions: ResolvedMention[],
slash: SlashInvocation,
messageId: string,
threadId: string,
now: string,
): string {
const target = mentions.find((mention) => mention.target.type === "agent")?.target;
if (!target || target.type !== "agent") throw new Error("Slash command /agent requires an agent mention.");
this.db.prepare(`
insert into chat_presence(principal_type, principal_id, status, status_text, last_seen_at)
values (?, ?, ?, ?, ?)
on conflict(principal_type, principal_id)
do update set status = excluded.status, status_text = excluded.status_text, last_seen_at = excluded.last_seen_at
`).run("agent", target.id, "thinking", slash.args.trim().slice(0, 160), now);
this.recordSlashRoute(input, messageId, slash.command, target.id, { targetType: "agent", targetId: target.id });
return target.id;
const wake = this.agentWake.wakeFromSlash(input, mentions, slash, messageId, threadId, now);
this.recordSlashRoute(input, messageId, slash.command, wake.targetId, {
targetType: "agent",
targetId: wake.targetId,
taskId: wake.taskId,
runId: wake.runId,
});
return wake.targetId;
}
private routeSkillSlash(input: SendMessageInput, slash: SlashInvocation, messageId: string): string {
@@ -6,6 +6,8 @@ import { DatabaseConnection } from "../../src/main/database/connection";
import { ApprovalService } from "../../src/main/services/approval-service";
import { ChatService } from "../../src/main/services/chat-service";
import { OrganizationService } from "../../src/main/services/organization-service";
import { ProviderService } from "../../src/main/services/provider-service";
import { providerAdapters } from "../../src/main/services/provider-adapters";
import { SnapshotService } from "../../src/main/services/snapshot-service";
import { RISK_LEVEL } from "../../src/shared/domain";
@@ -13,6 +15,11 @@ function createConnection(): DatabaseConnection {
return new DatabaseConnection(join(tmpdir(), `agentcompany-new-chat-${randomUUID()}.db`));
}
const testVault = {
encryptSecret: (secret: string) => Buffer.from(secret, "utf8"),
decryptSecret: (ciphertext: Buffer) => ciphertext.toString("utf8"),
};
describe("chat governance", () => {
test("persists memberships, retention policy, and structured delivery metadata", () => {
const connection = createConnection();
@@ -218,11 +225,13 @@ describe("chat governance", () => {
test("routes collaboration slash commands to concrete domain targets", () => {
const connection = createConnection();
const organization = new OrganizationService(connection.db);
const providers = new ProviderService(connection.db, providerAdapters, testVault);
const chat = new ChatService(connection.db);
const snapshot = new SnapshotService(connection.db);
let orgId = "";
let employeeId = "";
let agentId = "";
let providerId = "";
let spaceId = "";
let skillId = "";
let fileId = "";
@@ -240,13 +249,16 @@ describe("chat governance", () => {
goal: "Ship a verified control plane.",
});
employeeId = (connection.db.prepare("select id from employees where org_id = ?").get(orgId) as { id: string }).id;
providerId = (connection.db.prepare("select id from providers where org_id = ? and provider_key = 'openai'")
.get(orgId) as { id: string }).id;
providers.saveCredential({ orgId, providerId, secret: "sk-live-provider", model: "gpt-5.4" });
agentId = organization.createAgent({
orgId,
name: "Ada",
role: "Release reviewer",
department: "Engineering",
modelProviderId: null,
modelId: null,
modelProviderId: providerId,
modelId: "gpt-5.4",
budgetCents: 10000,
managerAgentId: null,
});
@@ -316,6 +328,36 @@ describe("chat governance", () => {
expect(slashByMessage.get(fileMessageId)).toMatchObject({ command: "file", resultRef: fileId });
expect(slashByMessage.get(runbookMessageId)).toMatchObject({ command: "runbook", resultRef: runbookMessageId });
expect(slashByMessage.get(handoffMessageId)).toMatchObject({ command: "handoff", resultRef: fileId });
const agentTask = state.tasks.find((item) => item.assigneeAgentId === agentId && item.title === "Chat mention: review the gate");
expect(agentTask).toMatchObject({
orgId,
assigneeAgentId: agentId,
status: "backlog",
risk: "low",
priority: "high",
requiresFilesystem: false,
});
expect(agentTask?.description).toContain(`Source message: ${agentMessageId}`);
const agentRun = state.taskRuns.find((item) => item.taskId === agentTask?.id);
expect(agentRun).toMatchObject({
orgId,
agentId,
providerId,
trigger: "chat_mention",
state: "queued",
startedAt: null,
completedAt: null,
errorMessage: null,
});
expect(agentRun?.executionContextId).toMatch(/^ctx_/);
expect(agentRun?.serviceIdentityId).toMatch(/^sid_/);
expect(JSON.parse(agentRun?.secretScopeJson ?? "{}")).toMatchObject({
serviceIdentityId: agentRun?.serviceIdentityId,
providerIds: [providerId],
scope: "chat_mention",
credentialAccess: "provider_runtime_only",
sourceMessageId: agentMessageId,
});
expect(state.chatPresence).toContainEqual(expect.objectContaining({ principalType: "agent", principalId: agentId, status: "thinking" }));
expect(state.fileHandoffRecords).toContainEqual(expect.objectContaining({
messageId: handoffMessageId,