diff --git a/apps/desktop/electron/main/persistence-outbox.ts b/apps/desktop/electron/main/persistence-outbox.ts index d9a6572853..08c60d8309 100644 --- a/apps/desktop/electron/main/persistence-outbox.ts +++ b/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)); +} diff --git a/apps/desktop/test/persistence-outbox.test.mjs b/apps/desktop/test/persistence-outbox.test.mjs index c416e2c4e0..612a1bdfc2 100644 --- a/apps/desktop/test/persistence-outbox.test.mjs +++ b/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"), + ); +}); diff --git a/crates/host-core/src/session_collaboration/provenance.rs b/crates/host-core/src/session_collaboration/provenance.rs index 60ceef67b4..16a41cd505 100644 --- a/crates/host-core/src/session_collaboration/provenance.rs +++ b/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 diff --git a/crates/host-core/src/session_collaboration/tests.rs b/crates/host-core/src/session_collaboration/tests.rs index 563d164d43..bc82106ae1 100644 --- a/crates/host-core/src/session_collaboration/tests.rs +++ b/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();