fix(session): keep history when tool ids collide across sessions

apps/desktop/electron/main/persistence-outbox.ts

@@ -41,13 +41,17 @@ export class PersistenceOutbox {
     await this.loaded;
     const existing = this.entries.findIndex((item) => item.key === entry.key);
     if (existing >= 0) this.entries[existing] = entry;
-    else if (this.entries.length >= MAX_ENTRIES) {
-      this.logger("error", "session persistence outbox is full", {
-        size: this.entries.length,
-        max: MAX_ENTRIES,
-      });
-      return;
-    } else this.entries.push(entry);
+    else {
+      if (this.entries.length >= MAX_ENTRIES) await this.flush(getHost);
+      if (this.entries.length >= MAX_ENTRIES) {
+        this.logger("error", "session persistence outbox is full", {
+          size: this.entries.length,
+          max: MAX_ENTRIES,
+        });
+        return;
+      }
+      this.entries.push(entry);
+    }
     await this.persist();
     void this.flush(getHost);
   }
@@ -89,11 +93,17 @@ export class PersistenceOutbox {
           turnId: current.turnId,
         });
       } catch (error) {
-        this.logger("warn", "session persistence flush paused", {
+        if (!isDuplicateMessageIdError(error)) {
+          this.logger("warn", "session persistence flush paused", {
+            key: current.key,
+            data: String(error),
+          });
+          return;
+        }
+        this.logger("warn", "session persistence flush skipped duplicate message id", {
           key: current.key,
           data: String(error),
         });
-        return;
       }
       // A newer snapshot may have replaced this key while the host wrote it.
       // Only remove the exact entry acknowledged by that write.
@@ -141,3 +151,7 @@ export class PersistenceOutbox {
     await write;
   }
 }
+
+function isDuplicateMessageIdError(error: unknown): boolean {
+  return /UNIQUE constraint failed: messages\.id/i.test(String(error));
+}

apps/desktop/test/persistence-outbox.test.mjs

@@ -56,3 +56,69 @@ test("deleting a session drops its queued outbox entries (D318)", async () => {
     ["keep"],
   );
 });
+
+function mockHost(handler) {
+  return {
+    isAvailable: () => true,
+    call: async (method, params) => handler(method, params),
+  };
+}
+
+test("duplicate message id does not stall later outbox entries (D444)", async () => {
+  const dir = await mkdtemp(join(tmpdir(), "pi-outbox-"));
+  const logs = [];
+  const outbox = new PersistenceOutbox(dir, (level, message, data) => {
+    logs.push({ level, message, data });
+  });
+  const calls = [];
+  const host = mockHost(async (_method, params) => {
+    calls.push(params);
+    if (params.message.id === "call_421522") {
+      throw new Error("UNIQUE constraint failed: messages.id");
+    }
+  });
+  const getHost = () => host;
+  await outbox.enqueue(
+    {
+      key: "message:s1:call_421522",
+      sessionId: "s1",
+      message: { id: "call_421522" },
+    },
+    getHost,
+  );
+  await outbox.enqueue(
+    {
+      key: "message:s2:assistant-1",
+      sessionId: "s2",
+      message: { id: "assistant-1" },
+    },
+    getHost,
+  );
+  await outbox.flush(getHost);
+  assert.equal(outbox.size(), 0);
+  assert.equal(calls.length, 2);
+  assert.equal(calls[1].message.id, "assistant-1");
+  assert.ok(
+    logs.some((row) => row.message === "session persistence flush skipped duplicate message id"),
+  );
+});
+
+test("non-unique flush errors still pause the outbox", async () => {
+  const dir = await mkdtemp(join(tmpdir(), "pi-outbox-"));
+  const outbox = new PersistenceOutbox(dir, silent);
+  const host = mockHost(async () => {
+    throw new Error("session not found");
+  });
+  const getHost = () => host;
+  await outbox.enqueue(
+    { key: "message:s1:a", sessionId: "s1", message: { id: "a" } },
+    getHost,
+  );
+  await outbox.enqueue(
+    { key: "message:s2:b", sessionId: "s2", message: { id: "b" } },
+    getHost,
+  );
+  await outbox.flush(getHost);
+  assert.equal(outbox.size(), 2);
+});
+

crates/host-core/src/sessions.rs

@@ -964,7 +964,31 @@ pub fn restore_orphaned_session(db: &Database, session_id: &str) -> Result<bool>
     if !path.exists() {
         return Ok(false);
     }
-    let records = dedupe_records(transcripts::read_transcript(db.data_dir(), session_id)?);
+    let read = transcripts::read_transcript_with_compactions(db.data_dir(), session_id)?;
+    let mut records = dedupe_records(read.messages);
+    let mut remapped = false;
+    for record in &mut records {
+        if let Some(owner) = message_owner(db, &record.id)? {
+            if owner != session_id {
+                record.id = namespaced_message_id(session_id, &record.id);
+                remapped = true;
+            }
+        }
+    }
+    if remapped {
+        let header = records
+            .first()
+            .map(|record| record.created_at.clone())
+            .unwrap_or_else(|| ms_to_ts(now_ms()));
+        transcripts::write_transcript_with_compactions(
+            db.data_dir(),
+            session_id,
+            &header,
+            &records,
+            &read.compactions,
+        )?;
+        invalidate_transcript_layout(session_id);
+    }
     let created_at = records
         .first()
         .map(|record| ts_to_ms(&record.created_at))
@@ -1718,9 +1742,11 @@ pub fn append_message(
 ) -> Result<()> {
     let message = crate::session_collaboration::prepare_append(db, session_id, message, turn_id)?;
     let session_created = ensure_session_for_append(db, session_id)?;
-    let (record, text) = ui_to_record(&message);
+    let (mut record, text) = ui_to_record(&message);
     // Electron may replay an outbox entry after a host restart. Message ids
-    // are globally unique, so an existing row is already the durable result.
+    // are globally unique, so an existing row in this session is already the
+    // durable result. Provider toolCallIds are not globally unique: a collision
+    // with another session is remapped before any transcript write (D444).
     // A steering input reserves its preceding streaming assistant's position;
     // only a terminal assistant snapshot may replace that provisional row.
     if message_indexed(db, session_id, &record.id)? {
@@ -1742,6 +1768,22 @@ pub fn append_message(
             return Ok(());
         }
     } else {
+        if let Some(owner) = message_owner(db, &record.id)? {
+            if owner != session_id {
+                let original_id = record.id.clone();
+                record.id = namespaced_message_id(session_id, &record.id);
+                if message_indexed(db, session_id, &record.id)? {
+                    return Ok(());
+                }
+                // Old hosts wrote JSONL then failed UNIQUE. Replaying that
+                // leftover must not append a second remapped line (D444).
+                if transcript_contains_id(db, session_id, &original_id)? {
+                    return Ok(());
+                }
+            } else {
+                return Ok(());
+            }
+        }
         append_record(
             db,
             session_id,
@@ -1798,6 +1840,29 @@ fn message_indexed(db: &Database, session_id: &str, message_id: &str) -> Result<
     Ok(existing.is_some())
 }
 
+/// Session that currently owns this globally unique message id, if any.
+fn message_owner(db: &Database, message_id: &str) -> Result<Option<String>> {
+    let owner: Option<String> = db
+        .conn()
+        .query_row(
+            "SELECT session_id FROM messages WHERE id = ?1",
+            params![message_id],
+            |row| row.get(0),
+        )
+        .optional()?;
+    Ok(owner)
+}
+
+fn namespaced_message_id(session_id: &str, message_id: &str) -> String {
+    format!("{session_id}:{message_id}")
+}
+
+fn transcript_contains_id(db: &Database, session_id: &str, message_id: &str) -> Result<bool> {
+    Ok(transcripts::read_transcript(db.data_dir(), session_id)?
+        .iter()
+        .any(|record| record.id == message_id))
+}
+
 fn streaming_assistant_indexed(db: &Database, session_id: &str, message_id: &str) -> Result<bool> {
     let layout = session_layout(db, session_id)?;
     let mut end = layout.message_count();
@@ -3942,6 +4007,94 @@ mod tests {
         assert_eq!(last_seq, 2);
     }
 
+    #[test]
+    fn append_message_remaps_ids_owned_by_another_session() {
+        let db = test_db();
+        let first = create_session(&db, None, None, None, None, None).unwrap();
+        let second = create_session(&db, None, None, None, None, None).unwrap();
+        let colliding = user_msg("call_421522", "first", "2026-09-16T15:42:41Z");
+        append_message(&db, &first.id, &colliding, None).unwrap();
+        append_message(&db, &second.id, &colliding, None).unwrap();
+
+        let first_detail = get_session(&db, &first.id).unwrap().unwrap();
+        let second_detail = get_session(&db, &second.id).unwrap().unwrap();
+        assert_eq!(first_detail.messages[0].id, "call_421522");
+        assert_eq!(
+            second_detail.messages[0].id,
+            format!("{}:call_421522", second.id)
+        );
+        assert_eq!(second_detail.messages[0].content, "first");
+
+        append_message(&db, &second.id, &colliding, None).unwrap();
+        assert_eq!(
+            get_session(&db, &second.id)
+                .unwrap()
+                .unwrap()
+                .messages
+                .len(),
+            1
+        );
+        assert_eq!(
+            transcripts::read_transcript(db.data_dir(), &second.id)
+                .unwrap()
+                .len(),
+            1
+        );
+    }
+
+    #[test]
+    fn append_message_does_not_duplicate_a_unique_failed_jsonl_leftover() {
+        let db = test_db();
+        let first = create_session(&db, None, None, None, None, None).unwrap();
+        let second = create_session(&db, None, None, None, None, None).unwrap();
+        let colliding = user_msg("call_421522", "leftover", "2026-09-16T15:42:41Z");
+        append_message(&db, &first.id, &colliding, None).unwrap();
+        let (record, _) = ui_to_record(&colliding);
+        transcripts::append_message(db.data_dir(), &second.id, &second.created_at, &record)
+            .unwrap();
+        append_message(&db, &second.id, &colliding, None).unwrap();
+        assert_eq!(
+            transcripts::read_transcript(db.data_dir(), &second.id)
+                .unwrap()
+                .len(),
+            1
+        );
+        assert_eq!(
+            get_session(&db, &second.id)
+                .unwrap()
+                .unwrap()
+                .messages
+                .len(),
+            1
+        );
+    }
+
+    #[test]
+    fn restore_orphaned_session_remaps_ids_owned_by_another_session() {
+        let db = test_db();
+        let first = create_session(&db, None, None, None, None, None).unwrap();
+        let second = create_session(&db, None, None, None, None, None).unwrap();
+        let colliding = user_msg("call_421522", "orphan", "2026-09-16T15:42:41Z");
+        append_message(&db, &first.id, &colliding, None).unwrap();
+        let (record, _) = ui_to_record(&colliding);
+        transcripts::append_message(db.data_dir(), &second.id, &second.created_at, &record)
+            .unwrap();
+        db.conn()
+            .execute("DELETE FROM sessions WHERE id = ?1", params![second.id])
+            .unwrap();
+        assert!(restore_orphaned_session(&db, &second.id).unwrap());
+        let restored = get_session(&db, &second.id).unwrap().unwrap();
+        assert_eq!(restored.messages.len(), 1);
+        assert_eq!(
+            restored.messages[0].id,
+            format!("{}:call_421522", second.id)
+        );
+        assert_eq!(
+            get_session(&db, &first.id).unwrap().unwrap().messages[0].id,
+            "call_421522"
+        );
+    }
+
     #[test]
     fn user_attachments_roundtrip_as_canonical_blocks() {
         let db = test_db();

docs/adr/0041-bounded-host-runtime-and-persistence-outbox.md

@@ -25,7 +25,10 @@ tool message appends pass through an Electron-main-owned, file-backed outbox
 and are flushed sequentially after a successful host handshake. The handshake
 **awaits** that drain before the host is advertised ready, so a cold
 `session.get` cannot race a queued assistant/tool row (D327). Host-side
-message append is idempotent by message id.
+message append is idempotent by message id. A colliding id that already belongs
+to another session is remapped to `{sessionId}:{id}` before the JSONL write
+(D444). The outbox treats `UNIQUE constraint failed: messages.id` as an ack,
+not a pause.
 
 ## Consequences
 

docs/spec/03-runtime/04-data-storage.md

@@ -711,7 +711,7 @@ stream (as today) with `text = NULL`.
 ```sql
 CREATE TABLE messages (
   mid          INTEGER PRIMARY KEY,             -- stable rowid: FTS anchor, VACUUM-safe
-  id           TEXT NOT NULL UNIQUE,            -- caller-facing uuid (optimistic UI)
+  id           TEXT NOT NULL UNIQUE,            -- caller-facing uuid (optimistic UI); colliding provider toolCallIds remap to {sessionId}:{id} (D444)
   session_id   TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
   turn_id      TEXT REFERENCES turns(id) ON DELETE SET NULL,
   seq          INTEGER NOT NULL,                -- per-session ordinal
@@ -1418,7 +1418,11 @@ line and search text, retaining sequence, owning turn and every other row.
 Late partial snapshots and duplicate terminal snapshots cannot overwrite the
 settled result. Recovery promotes the latest checkpoint in that same position.
 The outbox likewise keeps a newer snapshot that replaces an append while its
-host call is still pending. No schema migration is required.
+host call is still pending. If `messages.id` already belongs to another
+session, the host remaps to `{sessionId}:{id}` before any JSONL write; a
+replay of the original id is a no-op against that remapped row. The outbox
+treats `UNIQUE constraint failed: messages.id` as an ack and keeps draining
+(D444). No schema migration is required.
 
 ## 12. Native Pi session authority (ADR 0254)
 

docs/spec/03-runtime/06-host-rpc-protocol.md

@@ -560,14 +560,17 @@ resource exhaustion (`EAGAIN` / `WouldBlock`) with bounded backoff, never
 retries a command after it has started, and reaps timed-out children before
 releasing the execution slot.
 
-`session.appendMessage` is idempotent by message id. Electron main may keep
+`session.appendMessage` is idempotent by message id. An id already indexed in
+another session is remapped to `{sessionId}:{id}` before the JSONL write, and
+a later replay of the original id is a no-op (D444). Electron main may keep
 message appends in its application-owned outbox while host-core is restarting;
-the outbox flushes in order after a successful handshake. A missing sessions
-row is restored from the live JSONL (or created as a stub under the same id
-when the file is gone) so a queued outbox can drain (D318). `session.delete`
-drops that session's outbox entries. In-flight checkpoints never go through
-the outbox: a checkpoint is only meaningful against a live host, and replaying
-one after the final row would be wrong.
+the outbox flushes in order after a successful handshake and treats
+`UNIQUE constraint failed: messages.id` as an ack rather than pausing the
+queue. A missing sessions row is restored from the live JSONL (or created as a
+stub under the same id when the file is gone) so a queued outbox can drain
+(D318). `session.delete` drops that session's outbox entries. In-flight
+checkpoints never go through the outbox: a checkpoint is only meaningful
+against a live host, and replaying one after the final row would be wrong.
 
 ### Permissions
 - `permissions.evaluate`