diff --git a/src/main/database/migrations/054-close-stale-wake-tasks.ts b/src/main/database/migrations/054-close-stale-wake-tasks.ts new file mode 100644 index 00000000..fbbf0245 --- /dev/null +++ b/src/main/database/migrations/054-close-stale-wake-tasks.ts @@ -0,0 +1,16 @@ +export const closeStaleWakeTasksSchema = ` + -- Chat-wake tasks are conversation vehicles: the reply run is their whole + -- deliverable. Runs used to complete without closing them, stranding every + -- "Reply to ..." row in the backlog forever (a task with a completed run is + -- never enqueued again). Close each wake task whose runs have all been + -- spent; run completion closes them from now on. In-flight wakes (a queued + -- or running run) keep working normally. + update tasks + set status = 'done', updated_at = datetime('now') + where status in ('backlog', 'in_progress') + and exists (select 1 from task_runs r where r.task_id = tasks.id and r.trigger = 'chat_mention') + and not exists (select 1 from task_runs r where r.task_id = tasks.id and r.state in ('queued', 'running')); + + insert into schema_migrations(version, name, applied_at) + values (54, 'close_stale_wake_tasks', datetime('now')); +`; diff --git a/src/main/database/schema.ts b/src/main/database/schema.ts index b1b175a0..393e02cd 100644 --- a/src/main/database/schema.ts +++ b/src/main/database/schema.ts @@ -52,6 +52,7 @@ import { financeSalesDepartmentsSchema } from "./migrations/050-finance-sales-de import { purgeRuntimeSnapshotBloatSchema } from "./migrations/051-purge-runtime-snapshot-bloat"; import { avatarImagesSchema } from "./migrations/052-avatar-images"; import { autonomousPauseSchema } from "./migrations/053-autonomous-pause"; +import { closeStaleWakeTasksSchema } from "./migrations/054-close-stale-wake-tasks"; interface Migration { version: number; @@ -113,6 +114,7 @@ const MIGRATIONS: Migration[] = [ { version: 51, name: "purge_runtime_snapshot_bloat", sql: purgeRuntimeSnapshotBloatSchema }, { version: 52, name: "avatar_images", sql: avatarImagesSchema }, { version: 53, name: "autonomous_pause", sql: autonomousPauseSchema }, + { version: 54, name: "close_stale_wake_tasks", sql: closeStaleWakeTasksSchema }, ]; const SCHEMA_VERSION = MIGRATIONS[MIGRATIONS.length - 1].version; diff --git a/src/main/services/chat-agent-wake-service.ts b/src/main/services/chat-agent-wake-service.ts index 31a243ab..fef9b709 100644 --- a/src/main/services/chat-agent-wake-service.ts +++ b/src/main/services/chat-agent-wake-service.ts @@ -37,6 +37,11 @@ export class ChatAgentWakeService { wakeFromMention(input: SendMessageInput, agentId: string, messageId: string, threadId: string, now: string): AgentWakeResult { const agent = this.requireRunnableAgent(input.orgId, agentId, "Agent mention"); + // One pending wake per (agent, thread): the queued run reads the thread's + // newest messages when it executes, so waking again before it starts would + // spend a whole run to answer the same conversation twice. + const pending = this.pendingWakeRun(input.orgId, agentId, threadId); + if (pending) return { targetId: agentId, taskId: pending.taskId, runId: pending.runId }; const workspace = this.workspaceForAgent(input.orgId, agent.workspace_id); const from = this.principalName(input.orgId, input.senderType, input.senderId); const space = this.spaceRow(input.orgId, input.spaceId); @@ -81,6 +86,16 @@ export class ChatAgentWakeService { return { targetId: agent.id, taskId, runId }; } + private pendingWakeRun(orgId: string, agentId: string, threadId: string): { taskId: string; runId: string } | null { + const row = this.db.prepare(` + select id, task_id from task_runs + where org_id = ? and agent_id = ? and state = 'queued' and trigger = 'chat_mention' + and json_extract(secret_scope_json, '$.sourceThreadId') = ? + limit 1 + `).get(orgId, agentId, threadId) as { id: string; task_id: string } | undefined; + return row ? { taskId: row.task_id, runId: row.id } : null; + } + 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."); diff --git a/src/main/services/task-execution-service.ts b/src/main/services/task-execution-service.ts index 7a912b1d..4a0de7bf 100644 --- a/src/main/services/task-execution-service.ts +++ b/src/main/services/task-execution-service.ts @@ -1,6 +1,6 @@ import type { DatabaseSync } from "node:sqlite"; import type { RunTaskInput } from "../../shared/commands"; -import { APPROVAL_STATE, RISK_LEVEL } from "../../shared/domain"; +import { APPROVAL_STATE, RISK_LEVEL, TASK_STATUS } from "../../shared/domain"; import { currentProviderUsageWindow, isTransientProviderFailure } from "../../shared/provider-runtime-domain"; import { createId } from "../../shared/ids"; import { AuditService } from "./audit-service"; @@ -524,6 +524,23 @@ export class TaskExecutionService { set state = ?, provider_request_id = ?, completed_at = ?, error_message = null where id = ? and org_id = ? `).run(TASK_RUN_STATE.COMPLETED, providerRequestId, new Date().toISOString(), runId, orgId); + this.closeChatWakeTask(orgId, runId); + } + + /** + * A chat-wake task exists only to carry its reply run — the reply is the + * whole deliverable. The run completing therefore closes the task; left + * open it would strand forever (a task with a completed run is never + * enqueued again) and pile zombie "Reply to ..." rows onto the board. + */ + private closeChatWakeTask(orgId: string, runId: string): void { + const run = this.db.prepare("select task_id, trigger from task_runs where id = ? and org_id = ?") + .get(runId, orgId) as { task_id: string; trigger: string } | undefined; + if (!run || run.trigger !== "chat_mention") return; + const task = this.db.prepare("select status from tasks where id = ? and org_id = ?") + .get(run.task_id, orgId) as { status: string } | undefined; + if (!task || (task.status !== TASK_STATUS.BACKLOG && task.status !== TASK_STATUS.IN_PROGRESS)) return; + this.organization.updateTaskStatus(orgId, run.task_id, TASK_STATUS.DONE); } private failRun(orgId: string, runId: string, providerRequestId: string | null, errorMessage: string): void { diff --git a/tests/integration/agent-reply.test.ts b/tests/integration/agent-reply.test.ts index 467ed14a..210851b6 100644 --- a/tests/integration/agent-reply.test.ts +++ b/tests/integration/agent-reply.test.ts @@ -98,6 +98,12 @@ describe("inter-agent reply", () => { // Execute Dana's woken run: she replies in the CEO's thread. await taskExecution.executeRun(orgId, wake!.id); + // The reply was the wake task's whole deliverable, so completing the run closes it. + 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" }); + const engineering = connection.db.prepare("select id from chat_spaces where org_id = ? and name = 'engineering'").get(orgId) as { id: string }; const messages = snapshot.load().chatMessages.filter((m) => m.orgId === orgId && m.spaceId === engineering.id); const senders = new Set(messages.map((m) => m.senderId)); diff --git a/tests/integration/thread-participant-wake.test.ts b/tests/integration/thread-participant-wake.test.ts index e0f0bfa1..cbe298c3 100644 --- a/tests/integration/thread-participant-wake.test.ts +++ b/tests/integration/thread-participant-wake.test.ts @@ -59,5 +59,26 @@ describe("thread participant wake", () => { const runsAfterReply = connection.db.prepare("select count(*) as c from task_runs where org_id = ? and trigger = 'chat_mention'") .get(orgId) as { c: number }; expect(runsAfterReply.c).toBeGreaterThan(runsBeforeReply.c); + + // A second human message in the same thread coalesces into the still-queued + // wake instead of spending another run on the same conversation. + connection.transaction(() => { + chat.sendMessage({ + orgId, + spaceId: (connection.db.prepare("select id from chat_spaces where org_id = ? and name = 'general'").get(orgId) as { id: string }).id, + threadId: threadId!, + senderType: "employee", + senderId: employeeId, + bodyMd: "One more thing: schedule the rollout for Friday.", + attachmentFileIds: [], + }); + }); + const runsAfterSecondReply = connection.db.prepare("select count(*) as c from task_runs where org_id = ? and trigger = 'chat_mention'") + .get(orgId) as { c: number }; + expect(runsAfterSecondReply.c).toBe(runsAfterReply.c); + const wakeTasks = connection.db.prepare( + "select count(distinct r.task_id) as c from task_runs r where r.org_id = ? and r.trigger = 'chat_mention'", + ).get(orgId) as { c: number }; + expect(wakeTasks.c).toBe(1); }); }); diff --git a/tests/unit/chat-governance.test.ts b/tests/unit/chat-governance.test.ts index 3684c4f5..65145a7d 100644 --- a/tests/unit/chat-governance.test.ts +++ b/tests/unit/chat-governance.test.ts @@ -402,12 +402,22 @@ describe("chat governance", () => { "select count(*) as n from task_runs where org_id = ? and agent_id = ? and trigger = 'chat_mention'", ).get(orgId, targetId) as { n: number }).n; - // Alice (an agent) keeps @mentioning Bob in one thread. Each post is an - // autonomous turn; only the first batch wakes Bob, then the budget is spent. + // Simulates the worker finishing Bob's pending reply run, the state a real + // exchange is in when the next message lands (a still-queued run coalesces). + const finishPendingWakeRuns = (): void => { + connection.db.prepare( + "update task_runs set state = 'completed', completed_at = datetime('now') where org_id = ? and agent_id = ? and state = 'queued'", + ).run(orgId, targetId); + }; + + // Alice (an agent) keeps @mentioning Bob in one thread, and Bob's reply run + // completes before each next post. Only the first batch of turns wakes Bob, + // then the autonomous budget is spent. connection.transaction(() => { const first = chat.sendMessage({ orgId, spaceId, threadId: null, senderType: "agent", senderId: speakerId, bodyMd: "@Bob turn 1", attachmentFileIds: [] }); threadId = (connection.db.prepare("select thread_id from chat_messages where id = ?").get(first) as { thread_id: string }).thread_id; for (let turn = 2; turn <= 40; turn += 1) { + finishPendingWakeRuns(); chat.sendMessage({ orgId, spaceId, threadId, senderType: "agent", senderId: speakerId, bodyMd: `@Bob turn ${turn}`, attachmentFileIds: [] }); } }); @@ -419,7 +429,9 @@ describe("chat governance", () => { // A human re-enters the thread: their message wakes Bob and grants the // agents a fresh budget, so the next agent turn wakes him again. connection.transaction(() => { + finishPendingWakeRuns(); chat.sendMessage({ orgId, spaceId, threadId, senderType: "employee", senderId: employeeId, bodyMd: "Keep going @Bob", attachmentFileIds: [] }); + finishPendingWakeRuns(); chat.sendMessage({ orgId, spaceId, threadId, senderType: "agent", senderId: speakerId, bodyMd: "@Bob resuming", attachmentFileIds: [] }); }); expect(wakeRunCount()).toBeGreaterThan(capped); diff --git a/tests/unit/close-stale-wake-tasks-migration.test.ts b/tests/unit/close-stale-wake-tasks-migration.test.ts new file mode 100644 index 00000000..aedab813 --- /dev/null +++ b/tests/unit/close-stale-wake-tasks-migration.test.ts @@ -0,0 +1,56 @@ +import { randomUUID } from "node:crypto"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { DatabaseSync } from "node:sqlite"; +import { describe, expect, test } from "vitest"; +import { closeStaleWakeTasksSchema } from "../../src/main/database/migrations/054-close-stale-wake-tasks"; + +function createDatabase(): DatabaseSync { + const db = new DatabaseSync(join(tmpdir(), `agentcompany-wake-migration-${randomUUID()}.db`)); + db.exec(` + create table schema_migrations(version integer primary key, name text not null, applied_at text not null); + create table tasks(id text primary key, status text not null, updated_at text not null); + create table task_runs(id text primary key, task_id text not null, trigger text not null, state text not null); + insert into schema_migrations(version, name, applied_at) values (53, 'autonomous_pause', datetime('now')); + `); + return db; +} + +function seedTask(db: DatabaseSync, taskId: string, status: string, runs: Array<{ trigger: string; state: string }>): void { + db.prepare("insert into tasks(id, status, updated_at) values (?, ?, datetime('now'))").run(taskId, status); + for (const run of runs) { + db.prepare("insert into task_runs(id, task_id, trigger, state) values (?, ?, ?, ?)") + .run(`run_${randomUUID()}`, taskId, run.trigger, run.state); + } +} + +describe("close stale wake tasks migration", () => { + test("closes spent chat-wake tasks and leaves everything else alone", () => { + const db = createDatabase(); + // Zombie: reply went out but the task was never closed. + seedTask(db, "tsk_replied", "backlog", [{ trigger: "chat_mention", state: "completed" }]); + // Zombie: every attempt failed on the provider limit; the conversation has moved on. + seedTask(db, "tsk_starved", "backlog", [ + { trigger: "chat_mention", state: "failed" }, + { trigger: "heartbeat", state: "failed" }, + { trigger: "heartbeat", state: "failed" }, + ]); + // In-flight wake: its queued run must keep working. + seedTask(db, "tsk_pending", "backlog", [{ trigger: "chat_mention", state: "queued" }]); + // Real work is not a wake vehicle and is never touched. + seedTask(db, "tsk_real", "backlog", [{ trigger: "heartbeat", state: "completed" }]); + + db.exec(closeStaleWakeTasksSchema); + + const statuses = Object.fromEntries( + (db.prepare("select id, status from tasks").all() as Array<{ id: string; status: string }>).map((row) => [row.id, row.status]), + ); + expect(statuses).toEqual({ + tsk_replied: "done", + tsk_starved: "done", + tsk_pending: "backlog", + tsk_real: "backlog", + }); + expect(db.prepare("select version from schema_migrations order by version desc limit 1").get()).toEqual({ version: 54 }); + }); +}); diff --git a/tests/unit/domain-services.test.ts b/tests/unit/domain-services.test.ts index 580be2ab..d74461e6 100644 --- a/tests/unit/domain-services.test.ts +++ b/tests/unit/domain-services.test.ts @@ -24,7 +24,7 @@ describe("domain services", () => { const connection = createConnection(); const rows = connection.db.prepare("select version from schema_migrations order by version asc").all() as Array<{ version: number }>; - expect(rows.map((row) => row.version)).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53]); + expect(rows.map((row) => row.version)).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54]); connection.close(); });