fix(agent-to-agent): route A2A replies back to originating session (#2267)

Squash merge of PR #2267 by ddaniels.

When an agent group has more than one active session, A2A replies landed
in the newest session via findSessionByAgentGroup's ORDER BY created_at
DESC. The session that asked the question never saw the answer.

Adds origin-aware return-path routing with three layers:

1. Direct return-path: if the reply has in_reply_to, look up the
   triggering inbound row's source_session_id and route there.
2. Peer-affinity fallback: find the most recent A2A inbound from this
   peer and use its source_session_id.
3. Legacy fallback: newest active session (pre-migration compat).

Container-side: MCP send_message/send_file now thread the current
batch's in_reply_to through to outbound rows via current-batch.ts.

Also flips our A2A bug-documenting test (#2332) from asserting the
broken behavior to asserting the fixed behavior.

Co-Authored-By: Doug Daniels <ddaniels888@gmail.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

src/db/schema.ts

@@ -171,7 +171,13 @@ CREATE TABLE IF NOT EXISTS messages_in (
   platform_id    TEXT,
   channel_type   TEXT,
   thread_id      TEXT,
-  content        TEXT NOT NULL
+  content        TEXT NOT NULL,
+  -- For agent-to-agent inbound rows: the source session that emitted the
+  -- triggering outbound. Used as a return path when the target replies —
+  -- the reply routes back to this exact session, not to the source agent
+  -- group's "newest" session. NULL on channel-side inbound and on a2a rows
+  -- written before this column existed.
+  source_session_id TEXT
 );
 CREATE INDEX IF NOT EXISTS idx_messages_in_series ON messages_in(series_id);
 

src/db/session-db.test.ts

@@ -10,7 +10,7 @@ import fs from 'fs';
 import path from 'path';
 import { describe, it, expect, afterEach } from 'vitest';
 
-import { migrateMessagesInTable } from './session-db.js';
+import { getInboundSourceSessionId, migrateMessagesInTable } from './session-db.js';
 
 const TEST_DIR = '/tmp/nanoclaw-session-db-test';
 const DB_PATH = path.join(TEST_DIR, 'inbound.db');
@@ -55,4 +55,40 @@ describe('migrateMessagesInTable', () => {
     expect(row.series_id).toBe('legacy-1');
     db.close();
   });
+
+  it('adds source_session_id on a legacy DB, leaves existing rows NULL, is idempotent', () => {
+    if (fs.existsSync(TEST_DIR)) fs.rmSync(TEST_DIR, { recursive: true });
+    fs.mkdirSync(TEST_DIR, { recursive: true });
+
+    const db = new Database(DB_PATH);
+    db.exec(`
+      CREATE TABLE messages_in (
+        id             TEXT PRIMARY KEY,
+        seq            INTEGER UNIQUE,
+        kind           TEXT NOT NULL,
+        timestamp      TEXT NOT NULL,
+        status         TEXT DEFAULT 'pending',
+        process_after  TEXT,
+        recurrence     TEXT,
+        tries          INTEGER DEFAULT 0,
+        platform_id    TEXT,
+        channel_type   TEXT,
+        thread_id      TEXT,
+        content        TEXT NOT NULL
+      );
+    `);
+    db.prepare(
+      "INSERT INTO messages_in (id, seq, kind, timestamp, status, content) VALUES (?, ?, 'chat', datetime('now'), 'pending', '{}')",
+    ).run('legacy-2', 2);
+
+    migrateMessagesInTable(db);
+    migrateMessagesInTable(db); // idempotent
+
+    const cols = (db.prepare("PRAGMA table_info('messages_in')").all() as Array<{ name: string }>).map((c) => c.name);
+    expect(cols).toContain('source_session_id');
+
+    expect(getInboundSourceSessionId(db, 'legacy-2')).toBeNull();
+    expect(getInboundSourceSessionId(db, 'does-not-exist')).toBeNull();
+    db.close();
+  });
 });

src/db/session-db.ts

@@ -108,14 +108,21 @@ export function insertMessage(
      * Host countDueMessages gates on this; container reads everything.
      */
     trigger?: 0 | 1;
+    /**
+     * For agent-to-agent inbound: the source session id that emitted the
+     * outbound message which became this inbound row. Used as the return
+     * path for the target's reply. NULL on channel-side inbound.
+     */
+    sourceSessionId?: string | null;
   },
 ): void {
   db.prepare(
-    `INSERT INTO messages_in (id, seq, kind, timestamp, status, platform_id, channel_type, thread_id, content, process_after, recurrence, series_id, trigger)
-     VALUES (@id, @seq, @kind, @timestamp, 'pending', @platformId, @channelType, @threadId, @content, @processAfter, @recurrence, @id, @trigger)`,
+    `INSERT INTO messages_in (id, seq, kind, timestamp, status, platform_id, channel_type, thread_id, content, process_after, recurrence, series_id, trigger, source_session_id)
+     VALUES (@id, @seq, @kind, @timestamp, 'pending', @platformId, @channelType, @threadId, @content, @processAfter, @recurrence, @id, @trigger, @sourceSessionId)`,
   ).run({
     ...message,
     trigger: message.trigger ?? 1,
+    sourceSessionId: message.sourceSessionId ?? null,
     seq: nextEvenSeq(db),
   });
 }
@@ -239,6 +246,7 @@ export interface OutboundMessage {
   channel_type: string | null;
   thread_id: string | null;
   content: string;
+  in_reply_to: string | null;
 }
 
 export function getDueOutboundMessages(db: Database.Database): OutboundMessage[] {
@@ -305,4 +313,47 @@ export function migrateMessagesInTable(db: Database.Database): void {
     // the agent" semantics, so backfill 1 and default 1 for new inserts.
     db.prepare('ALTER TABLE messages_in ADD COLUMN trigger INTEGER NOT NULL DEFAULT 1').run();
   }
+  if (!cols.has('source_session_id')) {
+    // For agent-to-agent return-path routing. NULL on existing rows is fine —
+    // their replies fall back to the legacy "newest active session" lookup.
+    db.prepare('ALTER TABLE messages_in ADD COLUMN source_session_id TEXT').run();
+  }
+}
+
+/**
+ * Look up an inbound row's source_session_id by its message id. Returns null
+ * if the row doesn't exist or the column is NULL (channel inbound or
+ * pre-migration a2a inbound). Used by a2a routing to route replies back to
+ * the originating session.
+ */
+export function getInboundSourceSessionId(db: Database.Database, messageId: string): string | null {
+  const row = db.prepare('SELECT source_session_id FROM messages_in WHERE id = ?').get(messageId) as
+    | { source_session_id: string | null }
+    | undefined;
+  return row?.source_session_id ?? null;
+}
+
+/**
+ * Find the source_session_id of the most recent a2a inbound row from a
+ * specific peer (by agent group id). Used as a peer-affinity fallback in
+ * a2a routing when an outbound reply has no `in_reply_to` (e.g. the
+ * container's send_message MCP tool path didn't thread the batch's
+ * in_reply_to through).
+ *
+ * Heuristic: "the last time this peer talked to me, which session was it?"
+ * Returns null when no prior a2a inbound from that peer carries a
+ * non-null source_session_id (typical for pre-migration installs).
+ */
+export function getMostRecentPeerSourceSessionId(db: Database.Database, peerAgentGroupId: string): string | null {
+  const row = db
+    .prepare(
+      `SELECT source_session_id FROM messages_in
+        WHERE channel_type = 'agent'
+          AND platform_id = ?
+          AND source_session_id IS NOT NULL
+        ORDER BY seq DESC
+        LIMIT 1`,
+    )
+    .get(peerAgentGroupId) as { source_session_id: string | null } | undefined;
+  return row?.source_session_id ?? null;
 }