fix(agent-runner): bound hasIdenticalSend to the turn in flight

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 <message> 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;
 }

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: '<message to="discord-test">Water the plants!</message>',
+    });
+
+    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 <message> 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';

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 <message> 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
             // <message> 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 <message to="name">...</message>
  * 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 = /<message\s+to="([^"]+)"\s*>([\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,15 +656,20 @@ 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
   // an MCP send the agent already made this turn (drop it HERE, where the
   // 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;
   }