diff --git a/src/db/schema.ts b/src/db/schema.ts index 8433035be..48d9ce397 100644 --- a/src/db/schema.ts +++ b/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); diff --git a/src/db/session-db.test.ts b/src/db/session-db.test.ts index 530790094..a20210050 100644 --- a/src/db/session-db.test.ts +++ b/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(); + }); }); diff --git a/src/db/session-db.ts b/src/db/session-db.ts index addc39dbf..d4e8663d7 100644 --- a/src/db/session-db.ts +++ b/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,22 @@ 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; } diff --git a/src/delivery.ts b/src/delivery.ts index 036153a8b..a47fec270 100644 --- a/src/delivery.ts +++ b/src/delivery.ts @@ -239,6 +239,7 @@ async function deliverMessage( channel_type: string | null; thread_id: string | null; content: string; + in_reply_to: string | null; }, session: Session, inDb: Database.Database, diff --git a/src/modules/agent-to-agent/agent-route.test.ts b/src/modules/agent-to-agent/agent-route.test.ts index 4d48f6f7b..9b36503e1 100644 --- a/src/modules/agent-to-agent/agent-route.test.ts +++ b/src/modules/agent-to-agent/agent-route.test.ts @@ -1,20 +1,53 @@ -import { describe, expect, it } from 'vitest'; +import Database from 'better-sqlite3'; +import fs from 'fs'; +import { describe, expect, it, beforeEach, afterEach, vi } from 'vitest'; -import { isSafeAttachmentName } from './agent-route.js'; +import { isSafeAttachmentName, routeAgentMessage } from './agent-route.js'; +import { createDestination } from './db/agent-destinations.js'; +import { initTestDb, closeDb, runMigrations, createAgentGroup } from '../../db/index.js'; +import { createSession } from '../../db/sessions.js'; +import { initSessionFolder, inboundDbPath } from '../../session-manager.js'; +import type { Session } from '../../types.js'; + +vi.mock('../../container-runner.js', () => ({ + wakeContainer: vi.fn().mockResolvedValue(undefined), + isContainerRunning: vi.fn().mockReturnValue(false), + getActiveContainerCount: vi.fn().mockReturnValue(0), + killContainer: vi.fn(), +})); + +vi.mock('../../config.js', async () => { + const actual = await vi.importActual('../../config.js'); + return { ...actual, DATA_DIR: '/tmp/nanoclaw-test-a2a-route' }; +}); + +const TEST_DIR = '/tmp/nanoclaw-test-a2a-route'; + +function now(): string { + return new Date().toISOString(); +} + +function readInbound(agentGroupId: string, sessionId: string) { + const db = new Database(inboundDbPath(agentGroupId, sessionId), { readonly: true }); + const rows = db + .prepare('SELECT id, platform_id, channel_type, content, source_session_id FROM messages_in ORDER BY seq') + .all() as Array<{ + id: string; + platform_id: string | null; + channel_type: string | null; + content: string; + source_session_id: string | null; + }>; + db.close(); + return rows; +} -/** - * `forwardAttachedFiles` has a filesystem side that's awkward to unit-test - * without mocking DATA_DIR. The guarantee worth pinning is that the - * filename validator rejects everything that could escape the inbox dir — - * `forwardAttachedFiles` runs this guard before any I/O, so traversal is - * impossible as long as this matrix holds. - */ describe('isSafeAttachmentName', () => { it('accepts plain filenames', () => { expect(isSafeAttachmentName('baby-duck.png')).toBe(true); expect(isSafeAttachmentName('file with spaces.pdf')).toBe(true); expect(isSafeAttachmentName('report.v2.docx')).toBe(true); - expect(isSafeAttachmentName('.hidden')).toBe(true); // leading dot is fine, just not `.` / `..` + expect(isSafeAttachmentName('.hidden')).toBe(true); }); it('rejects empty / sentinel values', () => { @@ -44,3 +77,165 @@ describe('isSafeAttachmentName', () => { expect(isSafeAttachmentName(undefined as unknown as string)).toBe(false); }); }); + +/** + * Return-path routing: when an a2a reply targets an agent group with multiple + * sessions, it must land in the *originating* session — not the newest one. + * + * Setup: agent A has two active sessions S1 (older) + S2 (newer). + * Agent B is the peer A talks to. Bidirectional destinations wired. + */ +describe('routeAgentMessage return-path', () => { + const A = 'ag-A'; + const B = 'ag-B'; + let S1: Session; + let S2: Session; + let SB: Session; + + beforeEach(() => { + if (fs.existsSync(TEST_DIR)) fs.rmSync(TEST_DIR, { recursive: true }); + fs.mkdirSync(TEST_DIR, { recursive: true }); + + const db = initTestDb(); + runMigrations(db); + + createAgentGroup({ id: A, name: 'A', folder: 'a', agent_provider: null, created_at: now() }); + createAgentGroup({ id: B, name: 'B', folder: 'b', agent_provider: null, created_at: now() }); + + // S1 (older), S2 (newer) — both active sessions on A. + S1 = { + id: 'sess-A-old', + agent_group_id: A, + messaging_group_id: null, + thread_id: null, + agent_provider: null, + status: 'active', + container_status: 'stopped', + last_active: null, + created_at: '2026-01-01T00:00:00.000Z', + }; + S2 = { + id: 'sess-A-new', + agent_group_id: A, + messaging_group_id: null, + thread_id: null, + agent_provider: null, + status: 'active', + container_status: 'stopped', + last_active: null, + created_at: '2026-02-01T00:00:00.000Z', + }; + SB = { + id: 'sess-B', + agent_group_id: B, + messaging_group_id: null, + thread_id: null, + agent_provider: null, + status: 'active', + container_status: 'stopped', + last_active: null, + created_at: '2026-01-15T00:00:00.000Z', + }; + createSession(S1); + createSession(S2); + createSession(SB); + initSessionFolder(A, S1.id); + initSessionFolder(A, S2.id); + initSessionFolder(B, SB.id); + + createDestination({ + agent_group_id: A, + local_name: 'b', + target_type: 'agent', + target_id: B, + created_at: now(), + }); + createDestination({ + agent_group_id: B, + local_name: 'a', + target_type: 'agent', + target_id: A, + created_at: now(), + }); + }); + + afterEach(() => { + closeDb(); + if (fs.existsSync(TEST_DIR)) fs.rmSync(TEST_DIR, { recursive: true }); + }); + + it('forward direction: stamps source_session_id on the target inbound row', async () => { + // A.S1 emits an outbound a2a to B. + await routeAgentMessage( + { + id: 'msg-from-A-S1', + platform_id: B, + content: JSON.stringify({ text: 'hello B' }), + in_reply_to: null, + }, + S1, + ); + + const bRows = readInbound(B, SB.id); + expect(bRows).toHaveLength(1); + expect(bRows[0].platform_id).toBe(A); + expect(bRows[0].source_session_id).toBe(S1.id); // <- the return address + }); + + it('reply direction: routes back to the originating session, not the newest', async () => { + // A.S1 sends to B. + await routeAgentMessage( + { + id: 'msg-from-A-S1', + platform_id: B, + content: JSON.stringify({ text: 'ping' }), + in_reply_to: null, + }, + S1, + ); + + // Capture the synthetic id the host stamped on B's inbound — that's what + // B's container would reference as `in_reply_to` when replying. + const bRows = readInbound(B, SB.id); + const yId = bRows[0].id; + + // B replies to that message. + await routeAgentMessage( + { + id: 'msg-from-B', + platform_id: A, + content: JSON.stringify({ text: 'pong' }), + in_reply_to: yId, + }, + SB, + ); + + const s1Rows = readInbound(A, S1.id); + const s2Rows = readInbound(A, S2.id); + + // The reply lands in S1 (originator) even though S2 is newer. + expect(s1Rows).toHaveLength(1); + expect(s1Rows[0].platform_id).toBe(B); + expect(JSON.parse(s1Rows[0].content).text).toBe('pong'); + expect(s2Rows).toHaveLength(0); + }); + + it('fallback: a2a with no in_reply_to falls through to newest-session lookup', async () => { + // No prior conversation. B initiates an a2a to A out of the blue. + await routeAgentMessage( + { + id: 'msg-from-B-fresh', + platform_id: A, + content: JSON.stringify({ text: 'unsolicited' }), + in_reply_to: null, + }, + SB, + ); + + // Newest session wins (current heuristic, preserved). + const s1Rows = readInbound(A, S1.id); + const s2Rows = readInbound(A, S2.id); + expect(s1Rows).toHaveLength(0); + expect(s2Rows).toHaveLength(1); + }); +}); diff --git a/src/modules/agent-to-agent/agent-route.ts b/src/modules/agent-to-agent/agent-route.ts index 613a1edfb..691f30a55 100644 --- a/src/modules/agent-to-agent/agent-route.ts +++ b/src/modules/agent-to-agent/agent-route.ts @@ -23,10 +23,11 @@ import path from 'path'; import { isSafeAttachmentName } from '../../attachment-safety.js'; import { getAgentGroup } from '../../db/agent-groups.js'; +import { getInboundSourceSessionId } from '../../db/session-db.js'; import { getSession } from '../../db/sessions.js'; import { wakeContainer } from '../../container-runner.js'; import { log } from '../../log.js'; -import { resolveSession, sessionDir, writeSessionMessage } from '../../session-manager.js'; +import { openInboundDb, resolveSession, sessionDir, writeSessionMessage } from '../../session-manager.js'; import type { Session } from '../../types.js'; import { hasDestination } from './db/agent-destinations.js'; @@ -101,6 +102,49 @@ export interface RoutableAgentMessage { id: string; platform_id: string | null; content: string; + /** + * For replies, the id of the inbound message being replied to. The + * container's formatter sets this from the first inbound in the batch + * (`container/agent-runner/src/formatter.ts`). Used here to route the + * reply back to the originating session — see `resolveTargetSession`. + */ + in_reply_to: string | null; +} + +/** + * Pick which session of `targetAgentGroupId` should receive this a2a message. + * + * Return-path lookup: if the message is a reply (`in_reply_to` set), open the + * source agent's inbound DB and read the original triggering row's + * `source_session_id`. That column was stamped when the original outbound was + * routed — it's the session that started the conversation, and replies should + * land there even when the target agent group has multiple active sessions. + * + * Falls back to `resolveSession(..., 'agent-shared')` (which selects the + * newest active session) when: + * - the message has no `in_reply_to` (fresh-initiated a2a), OR + * - the referenced row isn't in source's inbound (cross-batch reference), OR + * - the referenced row's source_session_id is NULL (channel inbound or + * pre-migration row), OR + * - the recovered session no longer exists / belongs to a different agent. + */ +function resolveTargetSession(msg: RoutableAgentMessage, sourceSession: Session, targetAgentGroupId: string): Session { + if (msg.in_reply_to) { + const srcDb = openInboundDb(sourceSession.agent_group_id, sourceSession.id); + let originSessionId: string | null; + try { + originSessionId = getInboundSourceSessionId(srcDb, msg.in_reply_to); + } finally { + srcDb.close(); + } + if (originSessionId) { + const candidate = getSession(originSessionId); + if (candidate && candidate.agent_group_id === targetAgentGroupId && candidate.status === 'active') { + return candidate; + } + } + } + return resolveSession(targetAgentGroupId, null, null, 'agent-shared').session; } export async function routeAgentMessage(msg: RoutableAgentMessage, session: Session): Promise { @@ -119,7 +163,7 @@ export async function routeAgentMessage(msg: RoutableAgentMessage, session: Sess if (!getAgentGroup(targetAgentGroupId)) { throw new Error(`target agent group ${targetAgentGroupId} not found for message ${msg.id}`); } - const { session: targetSession } = resolveSession(targetAgentGroupId, null, null, 'agent-shared'); + const targetSession = resolveTargetSession(msg, session, targetAgentGroupId); const a2aMsgId = `a2a-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`; // If the source message references files (via `send_file`), forward the @@ -137,6 +181,7 @@ export async function routeAgentMessage(msg: RoutableAgentMessage, session: Sess channelType: 'agent', threadId: null, content: forwardedContent, + sourceSessionId: session.id, }); log.info('Agent message routed', { from: session.agent_group_id, diff --git a/src/session-manager.ts b/src/session-manager.ts index e3f3f7a0c..5c423eaeb 100644 --- a/src/session-manager.ts +++ b/src/session-manager.ts @@ -210,6 +210,12 @@ export function writeSessionMessage( * a trigger-1 message does arrive. */ 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 so the target's reply routes back to that exact session. + */ + sourceSessionId?: string | null; }, ): void { // Extract base64 attachment data, save to inbox, replace with file paths @@ -228,6 +234,7 @@ export function writeSessionMessage( processAfter: message.processAfter ?? null, recurrence: message.recurrence ?? null, trigger: message.trigger ?? 1, + sourceSessionId: message.sourceSessionId ?? null, }); } finally { db.close();