From 8ea8f17b2e272f79edaa376e661e5e47130c6069 Mon Sep 17 00:00:00 2001 From: pow1035 <71381697+pow1035@users.noreply.github.com> Date: Sat, 19 Sep 2026 00:29:54 +0800 Subject: [PATCH] fix(session): drain poison outbox entries and accept steering into delivery turns Two defects compound to silently drop assistant/tool transcripts: 1. The persistence outbox treats every append failure as transient and re-queues the head forever. A permanently-rejected entry (for example a steering message the host provenance check refuses) stalls the whole FIFO queue; once it fills to MAX_ENTRIES, every later message is discarded. 2. Steering (Alt+Enter) into a session-collaboration delivery turn builds a user message whose turn_id already owns a claimed delivery. prepare_append then rejects it as "transcript input does not match its session delivery", producing exactly the poison entry above. Fix the outbox to drop permanently-rejected (poison) messages and keep draining, while still pausing on transient failures. Fix provenance to accept steering input into its owning delivery turn without stamping it with the delivery's agent origin, and to still reject steering that targets another session. --- .../electron/main/persistence-outbox.ts | 34 +++++++++++++--- apps/desktop/test/persistence-outbox.test.mjs | 39 +++++++++++++++++++ .../src/session_collaboration/provenance.rs | 12 ++++++ .../src/session_collaboration/tests.rs | 34 ++++++++++++++++ 4 files changed, 114 insertions(+), 5 deletions(-) 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();