From e732b4e7fba627d19cd187bd0c3f2566d5ec3a47 Mon Sep 17 00:00:00 2001 From: gitcommit90 Date: Thu, 23 Jul 2026 23:58:04 +0000 Subject: [PATCH] Fix Photon conversation persistence --- CHANGELOG.md | 15 +++++++- README.md | 2 +- package-lock.json | 4 +- package.json | 2 +- src/server/channel-computers.ts | 2 +- src/server/db.ts | 35 ++++++++++++++++- src/server/photon.ts | 64 ++++++++++++++++++++++++++++---- test/channel-computers.mjs | 2 +- test/photon.mjs | 66 +++++++++++++++++++++++++++++++-- 9 files changed, 173 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 27b9293..8892bf6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.0.2] - 2026-07-23 + +### Fixed + +- Photon now keeps every message from the same mapped sender in one durable + 1Helm thread across connector restarts and Photon conversation-ID changes. + Sending `/new` closes that conversation without invoking the resident; the + sender's next text starts a new thread. +- Existing Photon installations carry their latest sender thread forward on + upgrade, and completed resident replies remain deduplicated on the return + path. + ## [0.0.1] - 2026-07-23 ### Initial release @@ -37,5 +49,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 notarization, stapled tickets, Gatekeeper verification, persistent Application Support, and isolated Apple container machines. -[Unreleased]: https://github.com/gitcommit90/1Helm/compare/v0.0.1...HEAD +[Unreleased]: https://github.com/gitcommit90/1Helm/compare/v0.0.2...HEAD +[0.0.2]: https://github.com/gitcommit90/1Helm/releases/tag/v0.0.2 [0.0.1]: https://github.com/gitcommit90/1Helm/releases/tag/v0.0.1 diff --git a/README.md b/README.md index ea7bd53..b9c227e 100644 --- a/README.md +++ b/README.md @@ -208,7 +208,7 @@ A fresh data directory opens first-run setup. The source runtime defaults to | `PORT` | `8123` | HTTP/WebSocket control-plane port. | | `CTRL_DATA_DIR` | `./data` | Databases, routing state, uploads, and narrow workspace mirrors. | | `HELM_CHANNEL_COMPUTER_BACKEND` | `apple` on macOS, `native` elsewhere | Explicit development/test backend override. | -| `HELM_CHANNEL_MACHINE_IMAGE` | `local/1helm-channel-machine:0.0.1` | Versioned Apple channel-machine image. | +| `HELM_CHANNEL_MACHINE_IMAGE` | `local/1helm-channel-machine:0.0.2` | Versioned Apple channel-machine image. | ### Agent-first JSON CLI diff --git a/package-lock.json b/package-lock.json index b1f3d2a..3e58348 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "1helm", - "version": "0.0.1", + "version": "0.0.2", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "1helm", - "version": "0.0.1", + "version": "0.0.2", "hasInstallScript": true, "dependencies": { "@gitcommit90/rerouted": "https://github.com/gitcommit90/rerouted/releases/download/v0.5.7/ReRouted-0.5.7-linux-node.tgz", diff --git a/package.json b/package.json index 61aace8..2dc3871 100644 --- a/package.json +++ b/package.json @@ -1,7 +1,7 @@ { "name": "1helm", "productName": "1Helm", - "version": "0.0.1", + "version": "0.0.2", "private": true, "type": "module", "description": "1Helm is the self-hosted home for durable AI employees: one resident, one private computer, compounding memory and skills, and Skipper for every boundary.", diff --git a/src/server/channel-computers.ts b/src/server/channel-computers.ts index c96ba5b..f27a5bd 100644 --- a/src/server/channel-computers.ts +++ b/src/server/channel-computers.ts @@ -67,7 +67,7 @@ const APPLE_RUNTIME_VERSION = "1.1.0"; export const APPLE_RUNTIME_PACKAGE = `container-${APPLE_RUNTIME_VERSION}-installer-signed.pkg`; export const APPLE_RUNTIME_URL = `https://github.com/apple/container/releases/download/${APPLE_RUNTIME_VERSION}/${APPLE_RUNTIME_PACKAGE}`; export const APPLE_RUNTIME_SHA256 = "0ca1c42a2269c2557efb1d82b1b38ac553e6a3a3da1b1179c439bcee1e7d6714"; -export const DEFAULT_CHANNEL_IMAGE = process.env.HELM_CHANNEL_MACHINE_IMAGE || "local/1helm-channel-machine:0.0.1"; +export const DEFAULT_CHANNEL_IMAGE = process.env.HELM_CHANNEL_MACHINE_IMAGE || "local/1helm-channel-machine:0.0.2"; const CONTAINER_CANDIDATES = [process.env.HELM_CONTAINER_CLI, "/usr/local/bin/container", "/opt/homebrew/bin/container", "container"].filter(Boolean) as string[]; const COMMAND_TIMEOUT_MS = Math.max(5_000, Number(process.env.HELM_MACHINE_COMMAND_TIMEOUT_MS || 120_000)); const IDLE_AFTER_MS = Math.max(60_000, Number(process.env.HELM_MACHINE_IDLE_MS || 15 * 60_000)); diff --git a/src/server/db.ts b/src/server/db.ts index d4b94dc..3981b22 100644 --- a/src/server/db.ts +++ b/src/server/db.ts @@ -340,6 +340,39 @@ export function migrate(): void { ); CREATE UNIQUE INDEX IF NOT EXISTS idx_photon_external ON photon_messages(external_id) WHERE external_id<>''; CREATE INDEX IF NOT EXISTS idx_photon_channel_time ON photon_messages(channel_id,received_at DESC); + CREATE TABLE IF NOT EXISTS photon_conversations ( + id INTEGER PRIMARY KEY, + channel_id INTEGER NOT NULL REFERENCES channels(id) ON DELETE CASCADE, + sender TEXT NOT NULL, + space_id TEXT NOT NULL, + root_message_id INTEGER NOT NULL REFERENCES messages(id) ON DELETE CASCADE, + thread_id INTEGER NOT NULL REFERENCES threads(id) ON DELETE CASCADE, + active INTEGER NOT NULL DEFAULT 1 CHECK (active IN (0,1)), + started INTEGER NOT NULL, + updated INTEGER NOT NULL, + closed INTEGER + ); + CREATE UNIQUE INDEX IF NOT EXISTS idx_photon_conversation_active + ON photon_conversations(channel_id,sender) WHERE active=1; + CREATE INDEX IF NOT EXISTS idx_photon_conversation_history + ON photon_conversations(channel_id,sender,started DESC); + INSERT INTO photon_conversations + (channel_id,sender,space_id,root_message_id,thread_id,active,started,updated) + SELECT pm.channel_id,pm.sender,pm.space_id,COALESCE(m.parent_id,m.id),t.id,1,m.created,pm.received_at + FROM photon_messages pm + JOIN messages m ON m.id=pm.message_id AND m.channel_id=pm.channel_id + JOIN threads t ON t.root_message_id=COALESCE(m.parent_id,m.id) AND t.channel_id=pm.channel_id + WHERE pm.direction='inbound' + AND lower(trim(pm.body))<>'/new' + AND NOT EXISTS ( + SELECT 1 FROM photon_conversations pc + WHERE pc.channel_id=pm.channel_id AND pc.sender=pm.sender + ) + AND NOT EXISTS ( + SELECT 1 FROM photon_messages newer + WHERE newer.channel_id=pm.channel_id AND newer.sender=pm.sender AND newer.direction='inbound' + AND (newer.received_at>pm.received_at OR (newer.received_at=pm.received_at AND newer.id>pm.id)) + ); CREATE TABLE IF NOT EXISTS connector_deliveries ( id INTEGER PRIMARY KEY, connector TEXT NOT NULL, @@ -746,7 +779,7 @@ export function migrate(): void { // developer deliberately opts into the native compatibility backend. const configuredBackend = String(process.env.HELM_CHANNEL_COMPUTER_BACKEND || (process.platform === "darwin" ? "apple" : "native")); const backend = ["apple", "native", "mock"].includes(configuredBackend) ? configuredBackend : "native"; - const image = String(process.env.HELM_CHANNEL_MACHINE_IMAGE || "local/1helm-channel-machine:0.0.1"); + const image = String(process.env.HELM_CHANNEL_MACHINE_IMAGE || "local/1helm-channel-machine:0.0.2"); for (const channel of q(`SELECT c.id FROM channels c JOIN agent_channels ac ON ac.channel_id=c.id WHERE c.kind='channel' AND c.status<>'deleted'`)) { const channelId = Number(channel.id); diff --git a/src/server/photon.ts b/src/server/photon.ts index ab5c84a..b31b786 100644 --- a/src/server/photon.ts +++ b/src/server/photon.ts @@ -157,12 +157,37 @@ function recoverInterruptedPhotonDeliveries(): void { } } -function queuePhotonDelivery(channelId: number, destination: string, body: string, sourceMessageId: number, key: string): void { +function queuePhotonDelivery(channelId: number, destination: string, body: string, sourceMessageId: number | null, key: string): void { run(`INSERT OR IGNORE INTO connector_deliveries (connector,idempotency_key,channel_id,destination,body,source_message_id,state,created,updated) VALUES ('photon',?,?,?,?,?,'pending',?,?)`, key, channelId, destination, body.slice(0, 50_000), sourceMessageId, now(), now()); } +function activePhotonConversation(channelId: number, sender: string): Row | undefined { + return q1(`SELECT pc.* FROM photon_conversations pc + JOIN messages root ON root.id=pc.root_message_id AND root.channel_id=pc.channel_id AND root.parent_id IS NULL + JOIN threads t ON t.id=pc.thread_id AND t.root_message_id=pc.root_message_id AND t.channel_id=pc.channel_id + WHERE pc.channel_id=? AND pc.sender=? AND pc.active=1 + ORDER BY pc.updated DESC,pc.id DESC LIMIT 1`, channelId, sender); +} + +function closePhotonConversation(channelId: number, sender: string, spaceId: string, event: PhotonEvent): number | null { + const conversation = activePhotonConversation(channelId, sender); + const timestamp = Date.parse(event.timestamp) || now(); + run("INSERT INTO photon_messages (channel_id,external_id,space_id,sender,direction,body,received_at,message_id) VALUES (?,?,?,?, 'inbound',?,?,NULL)", + channelId, event.id, spaceId, sender, event.text.slice(0, 50_000), timestamp); + if (!conversation) return null; + run("UPDATE photon_conversations SET active=0,space_id=?,updated=?,closed=? WHERE id=? AND active=1", spaceId, timestamp, timestamp, conversation.id); + run("UPDATE threads SET status='resolved',updated_at=? WHERE id=?", timestamp, conversation.thread_id); + const noteId = createMessage({ channelId, parentId: Number(conversation.root_message_id), botId: null, body: "Photon conversation closed with /new. The next text starts a new thread." }); + run("UPDATE messages SET system_message=1 WHERE id=?", noteId); + run("INSERT INTO channel_activity (channel_id,thread_id,kind,summary,actor_type,created) VALUES (?,?,'connector',?,'system',?)", + channelId, conversation.thread_id, `Photon closed the active conversation for ${sender}; the next text will start a new thread.`, timestamp); + broadcastToChannel(channelId, { type: "message", message: serializeMessage(noteId), parent: serializeMessage(Number(conversation.root_message_id)) }); + refreshThreadSummary(Number(conversation.root_message_id)); + return noteId; +} + /** Drain durable reply obligations. The attempt is persisted before crossing * the external boundary. A crash after that point is reported as uncertain on * restart and is never silently replayed into a duplicate iMessage. */ @@ -198,21 +223,44 @@ export async function deliverPhotonEvent(event: PhotonEvent): Promise { const channelId = Number(mapping.channel_id); const resident = agentForChannel(channelId); if (!resident?.bot_id) return false; - const body = `[Photon iMessage from ${event.sender}; conversation ${event.space_id}]\n${event.text}`; - const messageId = createMessage({ channelId, parentId: null, botId: null, body }); + const timestamp = Date.parse(event.timestamp) || now(); + if (event.text.trim().toLowerCase() === "/new") { + const noteId = closePhotonConversation(channelId, event.sender, event.space_id, event); + queuePhotonDelivery(channelId, event.space_id, + noteId ? "Started fresh. Your next text will open a new 1Helm thread." : "No conversation was open. Your next text will start a new 1Helm thread.", + noteId, `photon:event:${event.id}:new`); + await drainPhotonDeliveries(); + return true; + } + let conversation = activePhotonConversation(channelId, event.sender); + let rootMessageId = Number(conversation?.root_message_id || 0); + const body = `[Photon iMessage from ${event.sender}]\n${event.text}`; + const messageId = createMessage({ channelId, parentId: rootMessageId || null, botId: null, body }); run("UPDATE messages SET system_message=1 WHERE id=?", messageId); - const threadId = ensureThread(messageId, channelId); + if (!rootMessageId) { + rootMessageId = messageId; + const threadId = ensureThread(rootMessageId, channelId); + const conversationId = run(`INSERT INTO photon_conversations + (channel_id,sender,space_id,root_message_id,thread_id,active,started,updated) + VALUES (?,?,?,?,?,1,?,?)`, channelId, event.sender, event.space_id, rootMessageId, threadId, timestamp, timestamp).lastInsertRowid; + conversation = q1("SELECT * FROM photon_conversations WHERE id=?", conversationId); + } else { + run("UPDATE photon_conversations SET space_id=?,updated=? WHERE id=? AND active=1", event.space_id, timestamp, conversation!.id); + } + const threadId = Number(conversation!.thread_id); + run("UPDATE threads SET status='open',updated_at=? WHERE id=?", timestamp, threadId); run("INSERT INTO photon_messages (channel_id,external_id,space_id,sender,direction,body,received_at,message_id) VALUES (?,?,?,?, 'inbound',?,?,?)", channelId, event.id, event.space_id, event.sender, event.text.slice(0, 50_000), Date.parse(event.timestamp) || now(), messageId); run("INSERT INTO channel_activity (channel_id,thread_id,kind,summary,actor_type,created) VALUES (?,?,'connector',?,'system',?)", channelId, threadId, `Photon delivered an iMessage from ${event.sender}.`, now()); - broadcastToChannel(channelId, { type: "message", message: serializeMessage(messageId) }); - refreshThreadSummary(messageId); + broadcastToChannel(channelId, { type: "message", message: serializeMessage(messageId), parent: rootMessageId === messageId ? null : serializeMessage(rootMessageId) }); + refreshThreadSummary(rootMessageId); const bot = q1("SELECT * FROM bots WHERE id=?", resident.bot_id); if (bot && dispatchInbound) { try { const outboundBefore = Number(q1("SELECT COALESCE(MAX(id),0) id FROM photon_messages WHERE channel_id=? AND space_id=? AND direction='outbound'", channelId, event.space_id)?.id || 0); - await dispatchInbound(bot, channelId, messageId, messageId); + const replyBefore = Number(q1("SELECT COALESCE(MAX(id),0) id FROM messages WHERE channel_id=?", channelId)?.id || 0); + await dispatchInbound(bot, channelId, messageId, rootMessageId); const reply = q1(`SELECT id,body FROM messages WHERE channel_id=? AND parent_id=? AND bot_id=? - AND trim(body)<>'' AND body<>'_Working…_' ORDER BY id DESC LIMIT 1`, channelId, messageId, bot.id); + AND id>? AND trim(body)<>'' AND body<>'_Working…_' ORDER BY id DESC LIMIT 1`, channelId, rootMessageId, bot.id, replyBefore); const alreadyReturned = q1("SELECT 1 FROM photon_messages WHERE channel_id=? AND space_id=? AND direction='outbound' AND id>? LIMIT 1", channelId, event.space_id, outboundBefore); const alreadyQueued = reply?.id ? q1("SELECT 1 FROM connector_deliveries WHERE connector='photon' AND source_message_id=? LIMIT 1", reply.id) : undefined; if (reply?.body && !alreadyReturned && !alreadyQueued) queuePhotonDelivery(channelId, event.space_id, String(reply.body), Number(reply.id), `photon:event:${event.id}:final`); diff --git a/test/channel-computers.mjs b/test/channel-computers.mjs index 45897ff..846c980 100644 --- a/test/channel-computers.mjs +++ b/test/channel-computers.mjs @@ -154,7 +154,7 @@ test("Apple channel-computer contract preserves isolation, files, wakes, archive test("runtime digest and packaged image recipe stay pinned", async () => { assert.equal(computers.APPLE_RUNTIME_SHA256, "0ca1c42a2269c2557efb1d82b1b38ac553e6a3a3da1b1179c439bcee1e7d6714"); assert.match(computers.APPLE_RUNTIME_URL, /\/1\.1\.0\/container-1\.1\.0-installer-signed\.pkg$/); - assert.equal(computers.DEFAULT_CHANNEL_IMAGE, "local/1helm-channel-machine:0.0.1"); + assert.equal(computers.DEFAULT_CHANNEL_IMAGE, "local/1helm-channel-machine:0.0.2"); const packaging = await readFile(join(root, "scripts", "package-mac-dmg.cjs"), "utf8"); assert.match(packaging, /container\(\?:\$\|\\\/\)/, "release packaging includes container/ image assets"); const image = await readFile(join(root, "container", "Containerfile"), "utf8"); diff --git a/test/photon.mjs b/test/photon.mjs index de04e1b..38f48dd 100644 --- a/test/photon.mjs +++ b/test/photon.mjs @@ -36,7 +36,7 @@ server.listen(Number(process.env.PHOTON_SIDECAR_PORT),"127.0.0.1"); process.stdin.resume(); process.stdin.on("end",()=>process.exit(0)); `); -const { db, now, q, q1, run, seed } = await import("../src/server/db.ts"); +const { db, migrate, now, q, q1, run, seed } = await import("../src/server/db.ts"); const photon = await import("../src/server/photon.ts"); const photonAuth = await import("../src/server/photon-auth.ts"); const { createPhotonInboundQueue } = await import("../src/server/photon-queue.mjs"); @@ -88,16 +88,76 @@ test("Photon inbound delivery is allowlisted, deduplicated, and invokes the mapp assert.equal(dispatched[0].channelId, channel.id); }); +test("Photon keeps one durable sender thread until exact /new", async () => { + const channel = q1("SELECT id FROM channels WHERE name='messages'"); + const dispatched = []; + photon.registerPhotonDispatcher((_bot, channelId, triggerId, rootId) => dispatched.push({ channelId, triggerId, rootId })); + const sender = "+15551234567"; + await photon.deliverPhotonEvent({ id: "same-thread-reset", space_id: "space-persist-reset", space_type: "dm", sender, text: "/new", timestamp: new Date().toISOString() }); + assert.equal(await photon.deliverPhotonEvent({ id: "same-thread-1", space_id: "space-persist-a", space_type: "dm", sender, text: "first", timestamp: new Date().toISOString() }), true); + assert.equal(await photon.deliverPhotonEvent({ id: "same-thread-2", space_id: "space-persist-b", space_type: "dm", sender, text: "second", timestamp: new Date().toISOString() }), true); + const first = q1("SELECT message_id FROM photon_messages WHERE external_id='same-thread-1'"); + const second = q1("SELECT message_id FROM photon_messages WHERE external_id='same-thread-2'"); + assert.equal(q1("SELECT parent_id FROM messages WHERE id=?", second.message_id).parent_id, first.message_id, "later texts append beneath the first root even if Photon changes space ids"); + assert.equal(dispatched.at(-1).rootId, first.message_id); + const activeBefore = q1("SELECT * FROM photon_conversations WHERE channel_id=? AND sender=? AND active=1", channel.id, sender); + assert.equal(activeBefore.root_message_id, first.message_id); + + await photon.restartPhotonConnector(); + assert.equal(await photon.deliverPhotonEvent({ id: "same-thread-after-restart", space_id: "space-persist-restart", space_type: "dm", sender, text: "after restart", timestamp: new Date().toISOString() }), true); + const afterRestart = q1("SELECT message_id FROM photon_messages WHERE external_id='same-thread-after-restart'"); + assert.equal(q1("SELECT parent_id FROM messages WHERE id=?", afterRestart.message_id).parent_id, first.message_id, "connector restarts retain the same active thread"); + + const dispatchCount = dispatched.length; + assert.equal(await photon.deliverPhotonEvent({ id: "same-thread-new", space_id: "space-persist-b", space_type: "dm", sender, text: " /new ", timestamp: new Date().toISOString() }), true); + assert.equal(dispatched.length, dispatchCount, "/new is a connector control and never invokes the resident"); + assert.equal(q1("SELECT COUNT(*) n FROM photon_conversations WHERE channel_id=? AND sender=? AND active=1", channel.id, sender).n, 0); + + assert.equal(await photon.deliverPhotonEvent({ id: "same-thread-3", space_id: "space-persist-c", space_type: "dm", sender, text: "fresh", timestamp: new Date().toISOString() }), true); + const third = q1("SELECT message_id FROM photon_messages WHERE external_id='same-thread-3'"); + assert.equal(q1("SELECT parent_id FROM messages WHERE id=?", third.message_id).parent_id, null); + assert.notEqual(third.message_id, first.message_id); + assert.equal(q1("SELECT COUNT(*) n FROM photon_conversations WHERE channel_id=? AND sender=?", channel.id, sender).n >= 2, true, "conversation history survives the reset"); +}); + +test("Photon accepts /new when no conversation has ever been opened", async () => { + const sender = "+15559876543"; + const channel = q1("SELECT id FROM channels WHERE name='messages'"); + photon.mapPhotonChannel(channel.id, ["+15551234567", sender]); + let dispatched = false; + photon.registerPhotonDispatcher(() => { dispatched = true; }); + assert.equal(await photon.deliverPhotonEvent({ id: "first-ever-new", space_id: "space-first-new", space_type: "dm", sender, text: "/NEW", timestamp: new Date().toISOString() }), true); + assert.equal(dispatched, false); + assert.equal(q1("SELECT COUNT(*) n FROM photon_conversations WHERE channel_id=? AND sender=?", channel.id, sender).n, 0); + assert.equal(q1("SELECT source_message_id FROM connector_deliveries WHERE idempotency_key='photon:event:first-ever-new:new'").source_message_id, null); +}); + +test("Photon upgrades the most recent existing sender thread without splitting it", () => { + const sender = "+15551112222"; + const channel = q1("SELECT id FROM channels WHERE name='messages'"); + const stamp = now(); + const rootId = run("INSERT INTO messages (channel_id,parent_id,body,system_message,created) VALUES (?,NULL,?,1,?)", channel.id, `[Photon iMessage from ${sender}]\nexisting conversation`, stamp).lastInsertRowid; + const threadId = run("INSERT INTO threads (root_message_id,channel_id,status,title,summary,opened_at,updated_at) VALUES (?,?,'open','Existing Photon conversation','',?,?)", rootId, channel.id, stamp, stamp).lastInsertRowid; + run("INSERT INTO photon_messages (channel_id,external_id,space_id,sender,direction,body,received_at,message_id) VALUES (?,?,?,?, 'inbound',?,?,?)", channel.id, "pre-upgrade", "space-pre-upgrade", sender, "existing conversation", stamp, rootId); + + migrate(); + + const conversation = q1("SELECT * FROM photon_conversations WHERE channel_id=? AND sender=? AND active=1", channel.id, sender); + assert.equal(conversation.root_message_id, rootId); + assert.equal(conversation.thread_id, threadId); +}); + test("Photon returns the completed 1Helm reply exactly once to the inbound conversation", async () => { const channel = q1("SELECT id FROM channels WHERE name='messages'"); const bot = q1("SELECT b.id FROM bots b JOIN agents a ON a.bot_id=b.id JOIN agent_channels ac ON ac.agent_id=a.id WHERE ac.channel_id=?", channel.id); photon.registerPhotonDispatcher(async (_bot, channelId, _triggerId, rootId) => { run("INSERT INTO messages (channel_id,parent_id,bot_id,body,created) VALUES (?,?,?,?,?)", channelId, rootId, bot.id, "Automatic Photon answer", now()); }); + await photon.deliverPhotonEvent({ id: "reply-new-1", space_id: "space-reply", space_type: "dm", sender: "+15551234567", text: "/new", timestamp: new Date().toISOString() }); const event = { id: "reply-1", space_id: "space-reply", space_type: "dm", sender: "+15551234567", text: "please reply", timestamp: new Date().toISOString() }; assert.equal(await photon.deliverPhotonEvent(event), true); - assert.equal(q1("SELECT COUNT(*) n FROM messages WHERE channel_id=? AND parent_id=(SELECT message_id FROM photon_messages WHERE external_id='reply-1') AND bot_id=?", channel.id, bot.id).n, 1, "the agent answer is retained in its 1Helm thread"); - assert.equal(q1("SELECT COUNT(*) n FROM photon_messages WHERE channel_id=? AND space_id='space-reply' AND direction='outbound'", channel.id).n, 1, "the completed answer is sent back once"); + assert.equal(q1("SELECT COUNT(*) n FROM messages WHERE channel_id=? AND parent_id=(SELECT root_message_id FROM photon_conversations WHERE channel_id=? AND sender=? AND active=1) AND bot_id=?", channel.id, channel.id, "+15551234567", bot.id).n, 1, "the agent answer is retained in its 1Helm thread"); + assert.equal(q1("SELECT COUNT(*) n FROM photon_messages WHERE channel_id=? AND space_id='space-reply' AND direction='outbound' AND body='Automatic Photon answer'", channel.id).n, 1, "the completed answer is sent back once"); photon.registerPhotonDispatcher(async (_bot, channelId, _triggerId, rootId) => { run("INSERT INTO messages (channel_id,parent_id,bot_id,body,created) VALUES (?,?,?,?,?)", channelId, rootId, bot.id, "Tool-sent Photon answer", now());