Fix Photon conversation persistence

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

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
 

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.",

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));

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);

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<boolean> {
   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`);

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");