From 8c1266cab7258480e80089cb6e4d35dfdfabdff0 Mon Sep 17 00:00:00 2001 From: Vishnu Jayavel Date: Fri, 10 Jul 2026 04:48:35 -0700 Subject: [PATCH] fix(agent-runner): bound hasIdenticalSend to the turn in flight MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit hasIdenticalSend had no seq/time bound, so it scanned all of messages_out history. A recurring task's fixed-text reminder delivers on its first fire, then every later fire's turn-final block matches yesterday's identical row and gets silently dropped as a stale-history "already sent" echo — zero delivery from the second fire onward. Add a seq floor (getMaxSeq(), captured once at the top of each processQuery turn) so the lookup only considers rows written during the current turn. The original #2981 guard (dropping the echo of an MCP send_message made earlier in the same turn) still works, since that row lands above the floor. Closes #2997 --- container/agent-runner/src/db/messages-out.ts | 34 +++++++--- container/agent-runner/src/poll-loop.test.ts | 63 +++++++++++++++++++ container/agent-runner/src/poll-loop.ts | 26 ++++++-- 3 files changed, 109 insertions(+), 14 deletions(-) diff --git a/container/agent-runner/src/db/messages-out.ts b/container/agent-runner/src/db/messages-out.ts index 6f834fbee36..53603fcb6c6 100644 --- a/container/agent-runner/src/db/messages-out.ts +++ b/container/agent-runner/src/db/messages-out.ts @@ -32,6 +32,22 @@ export interface WriteMessageOut { content: string; } +/** + * Current maximum seq across both DBs. Used both to assign the next seq + * (writeMessageOut below) and as a floor for scoping a lookup to writes + * made after a given point in time (hasIdenticalSend's sinceSeq). + */ +export function getMaxSeq(): number { + const outbound = getOutboundDb(); + const inbound = getInboundDb(); + + // Read max seq from both DBs to maintain global ordering. + // Safe: each side only reads the other DB, never writes to it. + const maxOut = (outbound.prepare('SELECT COALESCE(MAX(seq), 0) AS m FROM messages_out').get() as { m: number }).m; + const maxIn = (inbound.prepare('SELECT COALESCE(MAX(seq), 0) AS m FROM messages_in').get() as { m: number }).m; + return Math.max(maxOut, maxIn); +} + /** * Write a new outbound message, auto-assigning an odd seq number. * Container uses odd seq (1, 3, 5...), host uses even (2, 4, 6...). @@ -44,13 +60,8 @@ export interface WriteMessageOut { */ export function writeMessageOut(msg: WriteMessageOut): number { const outbound = getOutboundDb(); - const inbound = getInboundDb(); - // Read max seq from both DBs to maintain global ordering. - // Safe: each side only reads the other DB, never writes to it. - const maxOut = (outbound.prepare('SELECT COALESCE(MAX(seq), 0) AS m FROM messages_out').get() as { m: number }).m; - const maxIn = (inbound.prepare('SELECT COALESCE(MAX(seq), 0) AS m FROM messages_in').get() as { m: number }).m; - const max = Math.max(maxOut, maxIn); + const max = getMaxSeq(); const nextSeq = max % 2 === 0 ? max + 1 : max + 2; // next odd // bun:sqlite requires named parameters to be passed with the prefix character @@ -148,16 +159,23 @@ export function getUndeliveredMessages(): MessageOutRow[] { * (an MCP send_message row from the current turn). Used by the task-fire * final-text dispatcher to drop the turn-final echo of a send the * agent already made — the dedup happens where the duplication originates. + * + * `sinceSeq` bounds the lookup to the turn in flight (rows with seq > + * sinceSeq). Without it, a fixed-text recurring reminder matches its OWN + * send from a previous fire and never gets delivered again (#2997) — the + * turn-final echo this guards against always lands above the floor + * captured at the start of the current turn, so that case still matches. */ -export function hasIdenticalSend(platformId: string, channelType: string, text: string): boolean { +export function hasIdenticalSend(platformId: string, channelType: string, text: string, sinceSeq: number): boolean { const row = getOutboundDb() .prepare( `SELECT 1 FROM messages_out WHERE platform_id = $platform_id AND channel_type = $channel_type AND (in_reply_to IS NULL OR in_reply_to = '') AND json_extract(content, '$.text') = $text + AND seq > $since_seq LIMIT 1`, ) - .get({ $platform_id: platformId, $channel_type: channelType, $text: text }); + .get({ $platform_id: platformId, $channel_type: channelType, $text: text, $since_seq: sinceSeq }); return row != null; } diff --git a/container/agent-runner/src/poll-loop.test.ts b/container/agent-runner/src/poll-loop.test.ts index 023c59e23af..05f74f926ff 100644 --- a/container/agent-runner/src/poll-loop.test.ts +++ b/container/agent-runner/src/poll-loop.test.ts @@ -411,6 +411,69 @@ const ERR_ROUTING = { inReplyTo: 'm1', }; +describe('hasIdenticalSend seq floor (#2997)', () => { + it('delivers a recurring task reminder even when an identical row exists from a previous fire', async () => { + getInboundDb() + .prepare( + `INSERT INTO destinations (name, display_name, type, channel_type, platform_id) VALUES (?, ?, ?, ?, ?)`, + ) + .run('discord-test', 'Discord', 'channel', 'discord', 'chan-1'); + + const { writeMessageOut } = await import('./db/messages-out.js'); + + // Yesterday's fire: identical text, in_reply_to null, written before this turn starts. + writeMessageOut({ + id: 'previous-fire', + kind: 'chat', + platform_id: 'chan-1', + channel_type: 'discord', + thread_id: null, + content: JSON.stringify({ text: 'Water the plants!' }), + }); + + const { query } = makeResultQuery({ + type: 'result', + text: 'Water the plants!', + }); + + const taskRouting = { + platformId: 'chan-1', + channelType: 'discord', + threadId: null, + inReplyTo: 'm1', + taskFire: true, + }; + + await processQuery(query, taskRouting, ['m1'], 'claude', undefined, 'prompt', undefined); + + // Both rows survive: the previous fire's row, and today's reminder. + // Before the seq floor, today's send matched the previous-fire row and + // was dropped as a turn-final echo — only 1 row. + expect(getUndeliveredMessages()).toHaveLength(2); + }); + + it('still drops the turn-final echo of an MCP send made within the same turn', async () => { + const { getMaxSeq, hasIdenticalSend, writeMessageOut } = await import('./db/messages-out.js'); + + // Floor captured at the top of this (simulated) turn, before the MCP + // send_message call below — this is the case hasIdenticalSend was + // introduced (#2981) to guard against, and it must still match because + // the row lands above the floor. + const floorSeq = getMaxSeq(); + + writeMessageOut({ + id: 'mcp-send-this-turn', + kind: 'chat', + platform_id: 'chan-1', + channel_type: 'discord', + thread_id: null, + content: JSON.stringify({ text: 'Water the plants!' }), + }); + + expect(hasIdenticalSend('chan-1', 'discord', 'Water the plants!', floorSeq)).toBe(true); + }); +}); + describe('error result with no envelope', () => { it('delivers a budget/billing error to the triggering channel and does not nudge', async () => { const budgetText = 'Spending limit reached. Add your own key at https://example.com/keys'; diff --git a/container/agent-runner/src/poll-loop.ts b/container/agent-runner/src/poll-loop.ts index 347ad8f3c42..b77ba3519bd 100644 --- a/container/agent-runner/src/poll-loop.ts +++ b/container/agent-runner/src/poll-loop.ts @@ -1,6 +1,6 @@ import { findByName, getAllDestinations, type DestinationEntry } from './destinations.js'; import { getPendingMessages, markProcessing, markCompleted, markScriptSkipped, type MessageInRow } from './db/messages-in.js'; -import { hasIdenticalSend, writeMessageOut } from './db/messages-out.js'; +import { getMaxSeq, hasIdenticalSend, writeMessageOut } from './db/messages-out.js'; import { getInboundDb, touchHeartbeat, clearStaleProcessingAcks } from './db/connection.js'; import { clearContinuation, @@ -340,6 +340,11 @@ export async function processQuery( let queryContinuation: string | undefined; let done = false; let unwrappedNudged = false; + // Floor for hasIdenticalSend: only rows written from this point on count as + // "this turn". Captured once, before any turn-final dispatch, so + // a fixed-text recurring reminder never matches its own send from a + // previous fire (#2997). + const sendFloorSeq = getMaxSeq(); // Prompt queue for the exchange hook — each result event consumes the // oldest unanswered prompt, except a wrapping-retry result, which answers // the same prompt again. Unused (and unmaintained) when the provider @@ -487,7 +492,7 @@ export async function processQuery( // at all — either way the turn is finished. markCompleted(initialBatchIds); if (event.text) { - const { sent, hasUnwrapped } = dispatchResultText(event.text, routing); + const { sent, hasUnwrapped } = dispatchResultText(event.text, routing, sendFloorSeq); if (sent === 0 && event.isError === true) { // Non-retryable error turn (e.g. a 403 billing_error) with no // envelope: deliver the notice instead of dropping it as @@ -605,7 +610,11 @@ function deliverErrorResult(text: string, routing: RoutingContext): void { * The agent must always wrap output in ... * blocks, even with a single destination. Bare text is scratchpad only. */ -function dispatchResultText(text: string, routing: RoutingContext): { sent: number; hasUnwrapped: boolean } { +function dispatchResultText( + text: string, + routing: RoutingContext, + sendFloorSeq: number, +): { sent: number; hasUnwrapped: boolean } { const MESSAGE_RE = /([\s\S]*?)<\/message>/g; let match: RegExpExecArray | null; @@ -627,7 +636,7 @@ function dispatchResultText(text: string, routing: RoutingContext): { sent: numb scratchpadParts.push(`[dropped: unknown destination "${toName}"] ${body}`); continue; } - sendToDestination(dest, body, routing); + sendToDestination(dest, body, routing, sendFloorSeq); sent++; } if (lastIndex < text.length) { @@ -647,7 +656,12 @@ function dispatchResultText(text: string, routing: RoutingContext): { sent: numb return { sent, hasUnwrapped }; } -function sendToDestination(dest: DestinationEntry, body: string, routing: RoutingContext): void { +function sendToDestination( + dest: DestinationEntry, + body: string, + routing: RoutingContext, + sendFloorSeq: number, +): void { const platformId = dest.type === 'channel' ? dest.platformId! : dest.agentGroupId!; const channelType = dest.type === 'channel' ? dest.channelType! : 'agent'; // Task fires: an explicitly-addressed final-text block is either the echo of @@ -655,7 +669,7 @@ function sendToDestination(dest: DestinationEntry, body: string, routing: Routin // duplication originates) or the agent's only deliberate send (write it // in_reply_to-null like the MCP path, or the host's task-fire suppression // would discard it — zero delivery). - if (routing.taskFire && hasIdenticalSend(platformId, channelType, body)) { + if (routing.taskFire && hasIdenticalSend(platformId, channelType, body, sendFloorSeq)) { log(`Dropping turn-final echo of an already-sent task message to ${dest.name}`); return; }