fix(session): drain poison outbox entries and accept steering into delivery turns

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

@@ -93,17 +93,31 @@ export class PersistenceOutbox {
           turnId: current.turnId,
         });
       } catch (error) {
-        if (!isDuplicateMessageIdError(error)) {
+        // A duplicate message id means the host already has the row; drop it
+        // and keep draining (D318/#560).
+        if (isDuplicateMessageIdError(error)) {
+          this.logger("warn", "session persistence flush skipped duplicate message id", {
+            key: current.key,
+            data: String(error),
+          });
+        } else if (isPoisonMessageError(error)) {
+          // The host will reject this row forever (for example provenance
+          // check: a steering message written into the wrong session). Drop
+          // only this entry and keep draining so one poisoned head cannot
+          // starve every later message out of the transcript.
+          this.logger("warn", "session persistence flush dropped poisoned message", {
+            key: current.key,
+            data: String(error),
+          });
+        } else {
+          // Transient failure (host busy/overloaded/pipe dead). Keep the head
+          // and retry on the next enqueue.
           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),
-        });
       }
       // A newer snapshot may have replaced this key while the host wrote it.
       // Only remove the exact entry acknowledged by that write.
@@ -155,3 +169,13 @@ export class PersistenceOutbox {
 function isDuplicateMessageIdError(error: unknown): boolean {
   return /UNIQUE constraint failed: messages\.id/i.test(String(error));
 }
+
+/**
+ * The host will reject this message on every attempt, no matter how many times
+ * it is retried. These are permanent, message-level errors (provenance /
+ * validation / permission), not transient host failures. Dropping the row is
+ * the only way to keep the FIFO outbox from starving every message behind it.
+ */
+function isPoisonMessageError(error: unknown): boolean {
+  return /PERMISSION_DENIED|INVALID_(ARGUMENT|PARAMS)|NOT_FOUND: session/i.test(String(error));
+}

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

@@ -122,3 +122,42 @@ test("non-unique flush errors still pause the outbox", async () => {
   assert.equal(outbox.size(), 2);
 });
 
+
+test("poisoned provenance message does not stall later outbox entries (D597)", 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 === "steering-poison") {
+      throw new Error("PERMISSION_DENIED: transcript input does not match its session delivery");
+    }
+  });
+  const getHost = () => host;
+  await outbox.enqueue(
+    {
+      key: "message:s1:steering-poison",
+      sessionId: "s1",
+      message: { id: "steering-poison", role: "user", steering: true },
+    },
+    getHost,
+  );
+  await outbox.enqueue(
+    {
+      key: "message:s2:assistant-1",
+      sessionId: "s2",
+      message: { id: "assistant-1", role: "assistant" },
+    },
+    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 dropped poisoned message"),
+  );
+});

crates/host-core/src/session_collaboration/provenance.rs

@@ -24,6 +24,18 @@ pub fn prepare_append(
     let mut message = input.clone();
     if message.role == "user" && message.parent_tool_call_id.is_none() {
         if let Some(delivery) = delivery {
+            // A steering input is additional human input to an already-claimed
+            // delivery turn, not the delivery itself: it must land in the same
+            // session but is exempt from the delivery's content/attachment
+            // contract and must not inherit the delivery's agent origin.
+            if message.steering == Some(true) {
+                if delivery.target_session_id != session_id {
+                    return Err(anyhow!(
+                        "PERMISSION_DENIED: steering input does not target its delivery session"
+                    ));
+                }
+                return Ok(message);
+            }
             if delivery.target_session_id != session_id
                 || delivery.content != message.content
                 || message

crates/host-core/src/session_collaboration/tests.rs

@@ -238,6 +238,40 @@ fn provenance_cannot_be_forged_or_stripped_and_permissions_are_rechecked() {
     assert!(sessions::replace_messages(&db, &child, &detail.messages).is_err());
 }
 
+#[test]
+fn steering_input_persists_without_inheriting_delivery_origin() {
+    let dir = tempfile::tempdir().unwrap();
+    let db = Database::open(&dir.path().join("pi.sqlite")).unwrap();
+    let parent = session(&db, "Parent");
+    let child = session(&db, "Child");
+    let message = send(&db, &parent, &child, "one");
+    let turn = begin_turn(&db, &child, &message.id, None, None).unwrap();
+
+    // Additional human input into the delivery turn keeps its human origin: it
+    // must not be rejected as a delivery mismatch, and it must not be stamped
+    // with the delivery's agent origin.
+    let mut steering = ui("user", "also update the docs");
+    steering.steering = Some(true);
+    sessions::append_message(&db, &child, &steering, Some(&turn)).unwrap();
+
+    let detail = sessions::get_session(&db, &child).unwrap().unwrap();
+    let appended = detail
+        .messages
+        .iter()
+        .find(|m| m.id == steering.id)
+        .expect("steering message persisted");
+    assert_eq!(appended.content, "also update the docs");
+    assert!(appended.session_message.is_none());
+
+    // Steering must still target the delivery's session.
+    let mut wrong = ui("user", "steer into another session");
+    wrong.steering = Some(true);
+    assert!(sessions::append_message(&db, &parent, &wrong, Some(&turn))
+        .unwrap_err()
+        .to_string()
+        .starts_with("PERMISSION_DENIED"));
+}
+
 #[test]
 fn queued_cancellation_keeps_the_session_and_notifies_once() {
     let dir = tempfile::tempdir().unwrap();