Stale chat wakes expire instead of resuming days later
Live incident: the founder DMed the CEO and got silence — the reply run sat 13th in a FIFO queue behind chat wakes from four days ago, which the reopened app was faithfully executing (necro-replies that woke yet more agents). A wake run is a conversation reply; past one hour the moment is gone. expireStaleChatWakes (chat-agent-wake-service) fails queued chat_mention runs older than the TTL and closes their vehicle tasks; the worker tick runs it before claiming, with one log line. Real work (heartbeat/kickoff triggers) never expires. Red-green tested.
This commit is contained in:
1 parent
c7eafb2836
commit
39aa292d09
3 files changed
+143
No files matched your search
@@ -1,6 +1,7 @@
|
|||||||
import type { BrowserWindow } from "electron";
|
import type { BrowserWindow } from "electron";
|
||||||
import type { DatabaseConnection } from "../database/connection";
|
import type { DatabaseConnection } from "../database/connection";
|
||||||
import { IPC_CHANNELS } from "../../shared/ipc";
|
import { IPC_CHANNELS } from "../../shared/ipc";
|
||||||
|
import { expireStaleChatWakes } from "../services/chat-agent-wake-service";
|
||||||
import { OrganizationService } from "../services/organization-service";
|
import { OrganizationService } from "../services/organization-service";
|
||||||
import { spawnGoalReviews } from "../services/goal-review-service";
|
import { spawnGoalReviews } from "../services/goal-review-service";
|
||||||
import { TaskExecutionService } from "../services/task-execution-service";
|
import { TaskExecutionService } from "../services/task-execution-service";
|
||||||
@@ -77,6 +78,14 @@ export function startTaskQueueWorker(input: TaskQueueWorkerInput): () => void {
|
|||||||
wasPaused = paused;
|
wasPaused = paused;
|
||||||
}
|
}
|
||||||
if (paused) return;
|
if (paused) return;
|
||||||
|
// Stale conversation wakes die before anything is claimed: replying to
|
||||||
|
// a thread hours later is noise, and a backlog of them would starve
|
||||||
|
// fresh messages (the founder's included) for the whole grind.
|
||||||
|
const expired = input.connection.transaction(() => expireStaleChatWakes(input.connection.db, new Date().toISOString()));
|
||||||
|
if (expired > 0) {
|
||||||
|
console.info(`[task-queue] expired ${expired} stale chat wake run(s); newer messages re-wake their agents.`);
|
||||||
|
notifyDataChanged();
|
||||||
|
}
|
||||||
// Goal checkpoints first, so a review task created this tick is
|
// Goal checkpoints first, so a review task created this tick is
|
||||||
// eligible for the same tick's heartbeat enqueue below.
|
// eligible for the same tick's heartbeat enqueue below.
|
||||||
const reviews = input.connection.transaction(() => spawnGoalReviews(input.connection.db, organization));
|
const reviews = input.connection.transaction(() => spawnGoalReviews(input.connection.db, organization));
|
||||||
|
|||||||
@@ -28,6 +28,41 @@ export interface AgentWakeResult {
|
|||||||
runId: string;
|
runId: string;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* How long a queued chat wake stays claimable. A wake run is a conversation
|
||||||
|
* reply; once the moment has passed — the app sat closed overnight, or the
|
||||||
|
* queue spent an hour behind other work — answering the stale thread reads
|
||||||
|
* as noise, burns provider budget, and its reply wakes yet more agents off a
|
||||||
|
* dead conversation. Expiring it costs nothing: any newer message re-wakes
|
||||||
|
* the same agent fresh.
|
||||||
|
*/
|
||||||
|
export const CHAT_WAKE_TTL_MS = 60 * 60_000;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Fails every queued chat-mention run older than the TTL and closes its
|
||||||
|
* vehicle task (the reply was the task's whole deliverable — see
|
||||||
|
* closeChatWakeTask). Returns how many wakes were expired.
|
||||||
|
*/
|
||||||
|
export function expireStaleChatWakes(db: DatabaseSync, now: string): number {
|
||||||
|
const cutoff = new Date(new Date(now).getTime() - CHAT_WAKE_TTL_MS).toISOString();
|
||||||
|
const stale = db.prepare(`
|
||||||
|
select id, org_id, task_id from task_runs
|
||||||
|
where state = 'queued' and trigger = 'chat_mention' and created_at < ?
|
||||||
|
`).all(cutoff) as Array<{ id: string; org_id: string; task_id: string }>;
|
||||||
|
if (stale.length === 0) return 0;
|
||||||
|
const organization = new OrganizationService(db);
|
||||||
|
for (const run of stale) {
|
||||||
|
db.prepare("update task_runs set state = 'failed', completed_at = ?, error_message = ? where id = ?")
|
||||||
|
.run(now, "Expired: the conversation moved on before this reply could start. A newer message re-wakes the agent.", run.id);
|
||||||
|
const task = db.prepare("select status from tasks where id = ? and org_id = ?")
|
||||||
|
.get(run.task_id, run.org_id) as { status: string } | undefined;
|
||||||
|
if (task && (task.status === TASK_STATUS.BACKLOG || task.status === TASK_STATUS.IN_PROGRESS)) {
|
||||||
|
organization.updateTaskStatus(run.org_id, run.task_id, TASK_STATUS.DONE);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return stale.length;
|
||||||
|
}
|
||||||
|
|
||||||
export class ChatAgentWakeService {
|
export class ChatAgentWakeService {
|
||||||
private readonly organization: OrganizationService;
|
private readonly organization: OrganizationService;
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,99 @@
|
|||||||
|
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 { OrganizationService } from "../../src/main/services/organization-service";
|
||||||
|
import { ChatService } from "../../src/main/services/chat-service";
|
||||||
|
import { CHAT_WAKE_TTL_MS, expireStaleChatWakes } from "../../src/main/services/chat-agent-wake-service";
|
||||||
|
|
||||||
|
interface Harness {
|
||||||
|
connection: DatabaseConnection;
|
||||||
|
chat: ChatService;
|
||||||
|
orgId: string;
|
||||||
|
employeeId: string;
|
||||||
|
ceoId: string;
|
||||||
|
danaId: string;
|
||||||
|
}
|
||||||
|
|
||||||
|
function harness(): Harness {
|
||||||
|
const connection = new DatabaseConnection(join(tmpdir(), `ac-wake-expiry-${randomUUID()}.db`));
|
||||||
|
const organization = new OrganizationService(connection.db);
|
||||||
|
let orgId = "";
|
||||||
|
let ceoId = "";
|
||||||
|
let danaId = "";
|
||||||
|
connection.transaction(() => {
|
||||||
|
orgId = organization.createOrganization({ organizationName: "OrbitOps", employeeName: "Yang", mission: "m", goal: "g" });
|
||||||
|
const providerId = (connection.db.prepare("select id from providers where org_id = ? and provider_key = 'openai'").get(orgId) as { id: string }).id;
|
||||||
|
connection.db.prepare("update providers set enabled = 1 where id = ? and org_id = ?").run(providerId, orgId);
|
||||||
|
ceoId = organization.createAgent({
|
||||||
|
orgId, name: "CEO", role: "CEO", department: "Executive",
|
||||||
|
modelProviderId: providerId, modelId: "gpt-5.4", budgetCents: 25000, managerAgentId: null, roleLevel: "c_suite",
|
||||||
|
});
|
||||||
|
danaId = organization.createAgent({
|
||||||
|
orgId, name: "Dana", role: "Engineer", department: "Engineering",
|
||||||
|
modelProviderId: providerId, modelId: "gpt-5.4", budgetCents: 1000, managerAgentId: ceoId,
|
||||||
|
});
|
||||||
|
});
|
||||||
|
const employeeId = (connection.db.prepare("select id from employees where org_id = ?").get(orgId) as { id: string }).id;
|
||||||
|
return { connection, chat: new ChatService(connection.db), orgId, employeeId, ceoId, danaId };
|
||||||
|
}
|
||||||
|
|
||||||
|
/** DM an agent as the founder, returning the woken run id. */
|
||||||
|
function wake(h: Harness, agentId: string): string {
|
||||||
|
const spaceId = h.chat.ensureDirectSpace(h.orgId, { type: "employee", id: h.employeeId }, { type: "agent", id: agentId });
|
||||||
|
h.chat.sendMessage({
|
||||||
|
orgId: h.orgId, spaceId, threadId: null, senderType: "employee", senderId: h.employeeId,
|
||||||
|
bodyMd: "ping", attachmentFileIds: [],
|
||||||
|
});
|
||||||
|
const run = h.connection.db.prepare(
|
||||||
|
"select id from task_runs where org_id = ? and agent_id = ? and trigger = 'chat_mention' and state = 'queued'",
|
||||||
|
).get(h.orgId, agentId) as { id: string };
|
||||||
|
return run.id;
|
||||||
|
}
|
||||||
|
|
||||||
|
function runRow(h: Harness, runId: string): { state: string; error_message: string | null; task_status: string } {
|
||||||
|
return h.connection.db.prepare(`
|
||||||
|
select r.state, r.error_message, t.status as task_status
|
||||||
|
from task_runs r join tasks t on t.id = r.task_id
|
||||||
|
where r.id = ?
|
||||||
|
`).get(runId) as { state: string; error_message: string | null; task_status: string };
|
||||||
|
}
|
||||||
|
|
||||||
|
describe("stale chat wakes expire", () => {
|
||||||
|
test("a wake queued past the TTL is expired with its vehicle task; fresh wakes survive", () => {
|
||||||
|
const h = harness();
|
||||||
|
const staleRunId = wake(h, h.danaId);
|
||||||
|
const freshRunId = wake(h, h.ceoId);
|
||||||
|
const now = new Date();
|
||||||
|
const beyondTtl = new Date(now.getTime() - CHAT_WAKE_TTL_MS - 60_000).toISOString();
|
||||||
|
h.connection.db.prepare("update task_runs set created_at = ? where id = ?").run(beyondTtl, staleRunId);
|
||||||
|
|
||||||
|
const expired = h.connection.transaction(() => expireStaleChatWakes(h.connection.db, now.toISOString()));
|
||||||
|
expect(expired).toBe(1);
|
||||||
|
|
||||||
|
const stale = runRow(h, staleRunId);
|
||||||
|
expect(stale.state).toBe("failed");
|
||||||
|
expect(stale.error_message).toContain("Expired");
|
||||||
|
expect(stale.task_status).toBe("done");
|
||||||
|
|
||||||
|
const fresh = runRow(h, freshRunId);
|
||||||
|
expect(fresh.state).toBe("queued");
|
||||||
|
expect(fresh.task_status).toBe("backlog");
|
||||||
|
|
||||||
|
// Idempotent: nothing left to expire.
|
||||||
|
expect(h.connection.transaction(() => expireStaleChatWakes(h.connection.db, now.toISOString()))).toBe(0);
|
||||||
|
h.connection.close();
|
||||||
|
});
|
||||||
|
|
||||||
|
test("real work is never expired: heartbeat and kickoff runs outlive any queue age", () => {
|
||||||
|
const h = harness();
|
||||||
|
// The bootstrap-style run: a plain queued run with a non-chat trigger.
|
||||||
|
const staleRunId = wake(h, h.danaId);
|
||||||
|
h.connection.db.prepare("update task_runs set trigger = 'heartbeat', created_at = '2026-01-01T00:00:00Z' where id = ?").run(staleRunId);
|
||||||
|
|
||||||
|
expect(h.connection.transaction(() => expireStaleChatWakes(h.connection.db, new Date().toISOString()))).toBe(0);
|
||||||
|
expect((h.connection.db.prepare("select state from task_runs where id = ?").get(staleRunId) as { state: string }).state).toBe("queued");
|
||||||
|
h.connection.close();
|
||||||
|
});
|
||||||
|
});
|
||||||
Reference in new issue
Block a user