feat: better and more reliable ai agent pipeline

Author: Rieranthony

apps/api/package.json

@@ -29,8 +29,6 @@
     "@hono/zod-openapi": "^1.2.0",
     "@hono/zod-validator": "^0.7.6",
     "@openrouter/ai-sdk-provider": "^2.0.0",
-    "@polar-sh/better-auth": "^1.6.4",
-    "@polar-sh/hono": "^0.5.3",
     "@tinybirdco/sdk": "^0.0.33",
     "@trpc/server": "^11.8.1",
     "@types/node": "^24.10.9",

apps/api/src/ai-agent/AI-README.md

@@ -658,31 +658,36 @@ When adding/updating a tool:
 ### Response Timing
 
 Queue delay is disabled (0ms) so the AI responds as fast as possible.
-For visitor-trigger bursts, worker-side debounce (`AI_AGENT_VISITOR_DEBOUNCE_MS`, default 800ms) is applied before selecting an effective trigger.
+No visitor burst coalescing or debounce is applied.
 Natural typing delays between multi-part messages are still applied to keep the experience human.
 
 ### Queueing Model
 
 - Each conversation has a Redis sorted set queue ordered by `createdAt` (with `messageId` tiebreaker).
-- A BullMQ drain job processes messages sequentially and advances a DB cursor for recovery.
-- Visitor bursts are coalesced at queue head: contiguous visitor triggers are handled as one effective trigger (latest message in the burst).
+- Wake jobs are conversation-scoped (`ai-agent-{conversationId}`), with single-active semantics:
+  - `waiting`/`delayed`/`completed`/`failed` wake jobs are replaced
+  - `active` wake jobs are never replaced
+- A BullMQ drain job processes queued messages sequentially and advances a DB cursor for recovery.
 - BullMQ wake jobs remain signals only; Redis queue + DB cursor are authoritative state.
+- Conversations with queued items are tracked in Redis (`ai-agent:active-conversations`), and producer/worker recovery markers are tracked via `ai-agent:wake-needed:{conversationId}`.
+- A worker-side wake sweeper periodically repairs missing wakes for non-empty queues.
 
 ### Trigger-Level Reliability Rules
 
 1. **FIFO Trigger Processing**: Conversation triggers are processed in queue order using the Redis ZSET cursor model.
-2. **Burst Coalescing**: Contiguous visitor messages at queue head are coalesced and processed once using the latest coalesced trigger.
-3. **Continuation Gate**: If a queued visitor trigger already has a newer public AI reply, the pipeline runs `skip vs supplement` before generation.
-4. **Bias to Supplement on Uncertainty**: If continuation classification is uncertain (timeout/model error), fallback favors `supplement` (never silent miss).
-5. **No Full-Turn Retry After Visible Reply**: If a trigger already sent any public message, that trigger is marked `retryable=false` and dropped on subsequent pipeline error.
-6. **Retry Only Pre-Reply Failures**: If a trigger fails before any public send, it stays queued and is retried (with per-message failure threshold).
-7. **Typing Always Ends**: Typing is stopped before each visible send and force-stopped in final pipeline cleanup.
+2. **Strict Per-Conversation Serial Execution**: Redis lock (`ai-agent:lock:{conversationId}`) ensures only one worker processes a conversation at a time.
+3. **No Burst Coalescing**: Every queued message is processed in order; no contiguous visitor batching.
+4. **Reliable Producer Path**: Producer enqueues message (`ZADD NX`) then ensures wake with bounded retries; on exhaustion it marks `wake-needed` recovery.
+5. **Lock Miss/Loss Recovery**: Worker attempts continuation wake with jitter when lock cannot be acquired or is lost during processing.
+6. **End-of-Job Invariant**: If queue remains non-empty, worker must ensure a runnable wake exists or mark recovery.
+7. **Sweeper Reconciliation**: Periodic sweeper scans active + wake-needed conversations and recreates missing wakes.
+8. **Typing Always Ends**: Typing is stopped before each visible send and force-stopped in final pipeline cleanup.
 
 ### Failure Handling
 
-1. **`retryable=true` and below threshold**: Keep trigger message at queue head, schedule continuation drain.
-2. **`retryable=false`**: Advance cursor to effective trigger and remove processed/coalesced queue items immediately.
-3. **Threshold reached**: Drop trigger/coalesced batch, advance cursor, continue draining.
+1. **`retryable=true` and below threshold**: Keep trigger message at queue head for retry.
+2. **`retryable=false`**: Advance cursor and remove the failed message immediately.
+3. **Threshold reached**: Drop failed message, advance cursor, continue draining.
 4. **Stalled jobs**: BullMQ stalled-job recovery still applies at worker level.
 5. **Error events**: `aiAgentProcessingCompleted` with `status: "error"` is still emitted for dashboard observability.
 

apps/api/src/ai-agent/actions/send-message.ts

@@ -105,7 +105,6 @@ export async function sendMessage(
 		userId: null,
 		visitorId: null,
 		createdAt,
-		triggerNotificationWorkflow: false,
 	});
 
 	try {

apps/api/src/ai-agent/capabilities-studio.test.ts

@@ -68,6 +68,10 @@ function createDocument(
 }
 
 describe("buildCapabilitiesStudioResponse", () => {
+	it("keeps wait.md in dropped skill template names", () => {
+		expect(AI_AGENT_DROPPED_SKILL_TEMPLATE_NAMES).toContain("wait.md");
+	});
+
 	it("maps behavior settings to runtime tool enabled state", () => {
 		const response = buildCapabilitiesStudioResponse({
 			aiAgent: createAgent({

apps/api/src/ai-agent/context/state.test.ts

@@ -0,0 +1,82 @@
+import { describe, expect, it } from "bun:test";
+import { getConversationState } from "./state";
+
+describe("getConversationState", () => {
+	it("starts assignee and participant queries in parallel", async () => {
+		let selectCallCount = 0;
+		let resolveAssignees!: (value: Array<{ userId: string }>) => void;
+		let resolveParticipants!: (value: Array<{ userId: string }>) => void;
+
+		const assigneesPromise = new Promise<Array<{ userId: string }>>(
+			(resolve) => {
+				resolveAssignees = (value) => resolve(value);
+			}
+		);
+		const participantsPromise = new Promise<Array<{ userId: string }>>(
+			(resolve) => {
+				resolveParticipants = (value) => resolve(value);
+			}
+		);
+
+		const db = {
+			select: () => {
+				selectCallCount++;
+				const pending =
+					selectCallCount === 1 ? assigneesPromise : participantsPromise;
+				return {
+					from: () => ({
+						where: () => pending,
+					}),
+				};
+			},
+		};
+
+		const statePromise = getConversationState(
+			db as never,
+			{ conversationId: "conv-1", organizationId: "org-1" },
+			{
+				escalatedAt: "2025-01-01T00:00:00.000Z",
+				escalationHandledAt: null,
+				escalationReason: "needs specialist",
+			} as never
+		);
+
+		await Promise.resolve();
+		expect(selectCallCount).toBe(2);
+
+		resolveAssignees([{ userId: "user-1" }]);
+		resolveParticipants([{ userId: "user-2" }]);
+
+		const result = await statePromise;
+		expect(result).toEqual({
+			hasHumanAssignee: true,
+			assigneeIds: ["user-1"],
+			participantIds: ["user-2"],
+			isEscalated: true,
+			escalationReason: "needs specialist",
+		});
+	});
+
+	it("keeps escalation false when escalation is handled", async () => {
+		const db = {
+			select: () => ({
+				from: () => ({
+					where: async () => [],
+				}),
+			}),
+		};
+
+		const result = await getConversationState(
+			db as never,
+			{ conversationId: "conv-1", organizationId: "org-1" },
+			{
+				escalatedAt: "2025-01-01T00:00:00.000Z",
+				escalationHandledAt: "2025-01-01T00:01:00.000Z",
+				escalationReason: null,
+			} as never
+		);
+
+		expect(result.isEscalated).toBe(false);
+		expect(result.escalationReason).toBeNull();
+	});
+});

apps/api/src/ai-agent/context/state.ts

@@ -37,29 +37,28 @@ export async function getConversationState(
 	params: GetStateParams,
 	conversation: ConversationSelect
 ): Promise<ConversationState> {
-	// Get active assignees
-	const assignees = await db
-		.select({ userId: conversationAssignee.userId })
-		.from(conversationAssignee)
-		.where(
-			and(
-				eq(conversationAssignee.conversationId, params.conversationId),
-				eq(conversationAssignee.organizationId, params.organizationId),
-				isNull(conversationAssignee.unassignedAt)
-			)
-		);
-
-	// Get active participants
-	const participants = await db
-		.select({ userId: conversationParticipant.userId })
-		.from(conversationParticipant)
-		.where(
-			and(
-				eq(conversationParticipant.conversationId, params.conversationId),
-				eq(conversationParticipant.organizationId, params.organizationId),
-				isNull(conversationParticipant.leftAt)
-			)
-		);
+	const [assignees, participants] = await Promise.all([
+		db
+			.select({ userId: conversationAssignee.userId })
+			.from(conversationAssignee)
+			.where(
+				and(
+					eq(conversationAssignee.conversationId, params.conversationId),
+					eq(conversationAssignee.organizationId, params.organizationId),
+					isNull(conversationAssignee.unassignedAt)
+				)
+			),
+		db
+			.select({ userId: conversationParticipant.userId })
+			.from(conversationParticipant)
+			.where(
+				and(
+					eq(conversationParticipant.conversationId, params.conversationId),
+					eq(conversationParticipant.organizationId, params.organizationId),
+					isNull(conversationParticipant.leftAt)
+				)
+			),
+	]);
 
 	const assigneeIds = assignees.map((a) => a.userId);
 	const participantIds = participants.map((p) => p.userId);

apps/api/src/ai-agent/output/parser.ts

@@ -49,7 +49,6 @@ export function validateDecisionForExecution(decision: AiDecision): {
 		case "resolve":
 		case "mark_spam":
 		case "skip":
-		case "wait":
 			// No special validation needed - messages are sent via tools
 			break;
 

apps/api/src/ai-agent/output/schemas.ts

@@ -41,7 +41,6 @@ export const aiDecisionSchema = z.object({
 			"resolve", // Mark conversation as resolved
 			"mark_spam", // Mark as spam
 			"skip", // No action needed
-			"wait", // Defer briefly, then re-evaluate from decision stage
 		])
 		.describe("The action to take after sending messages"),