From 0160eb3497bc2640c461c7cc5af65836491ded8e Mon Sep 17 00:00:00 2001 From: jif Date: Tue, 22 Sep 2026 10:49:51 +0000 Subject: [PATCH] Preserve explicit message-board unsubscribes when posting (#47257) ## Why Posting to a discussion automatically subscribed its author, undoing an explicit unsubscribe and restoring unwanted reply notifications. ## What changed Persist explicit opt-outs so posting follows a discussion by default but leaves unsubscribed agents opted out until explicitly subscribed again. Keep active subscriptions in the existing table for compatibility with older readers, and remove opt-outs when deleting a board. Update the API and tool descriptions to explain the behavior. ## Testing Extend the local board test to cover reopening an older database, preserving opt-outs across reopen and subsequent posts, legacy subscription queries, explicit notifications without resubscribing, explicit resubscription, and board cleanup. GitOrigin-RevId: 8b8f7f50cea5041541b8d93067c52db18cdde85c --- codex-rs/ext/agent-message-board/src/api.rs | 3 +- codex-rs/ext/agent-message-board/src/local.rs | 41 +++--- .../src/local/lifecycle.rs | 1 + .../ext/agent-message-board/src/tools/spec.rs | 2 +- .../agent-message-board/tests/local_board.rs | 117 ++++++++++++++++++ 5 files changed, 147 insertions(+), 17 deletions(-) diff --git a/codex-rs/ext/agent-message-board/src/api.rs b/codex-rs/ext/agent-message-board/src/api.rs index 29b9eb2733..495fe0a25e 100644 --- a/codex-rs/ext/agent-message-board/src/api.rs +++ b/codex-rs/ext/agent-message-board/src/api.rs @@ -79,7 +79,8 @@ pub trait AgentMessageBoard: Send + Sync { /// Channel subscriptions concern new roots; thread subscriptions concern /// replies. Changing one does not change the other. Any member may change - /// another member's subscription. Posting subscribes its author to the thread. + /// another member's subscription. Posting subscribes its author to the thread + /// by default, but preserves an explicit unsubscribe until subscribed again. fn set_subscription( &self, caller: ThreadId, diff --git a/codex-rs/ext/agent-message-board/src/local.rs b/codex-rs/ext/agent-message-board/src/local.rs index b4e0563fd5..ee4a069b56 100644 --- a/codex-rs/ext/agent-message-board/src/local.rs +++ b/codex-rs/ext/agent-message-board/src/local.rs @@ -80,6 +80,10 @@ CREATE INDEX IF NOT EXISTS posts_board_timestamp ON posts(board,timestamp,seq); CREATE TABLE IF NOT EXISTS subscriptions ( board TEXT NOT NULL, target TEXT NOT NULL, agent TEXT NOT NULL, PRIMARY KEY(board,target,agent) +); +CREATE TABLE IF NOT EXISTS subscription_opt_outs ( + board TEXT NOT NULL, target TEXT NOT NULL, agent TEXT NOT NULL, + PRIMARY KEY(board,target,agent) );"; #[derive(Clone)] @@ -279,6 +283,7 @@ impl LocalAgentMessageBoard { .bind(author.to_string()).bind(now.timestamp_micros()).bind(default_case_fold_str(&request.text)) .bind(serde_json::to_string(&post).map_err(storage_error)?).bind(request_id).bind(request_json) .execute(&mut *tx).await.map_err(storage_error)?; + // Participation subscribes by default, without overriding an explicit opt-out. self.subscribe(&mut tx, &SubscriptionTarget::Thread(root), caller) .await?; tx.commit().await.map_err(storage_error)?; @@ -327,20 +332,26 @@ impl LocalAgentMessageBoard { } }; let enabled = request.change == SubscriptionChange::Subscribe; - match request.change { - SubscriptionChange::Subscribe => { - self.subscribe(&mut tx, &request.target, target_agent) - .await? - } - SubscriptionChange::Unsubscribe => { - sqlx::query("DELETE FROM subscriptions WHERE board=? AND target=? AND agent=?") - .bind(self.identity.to_string()) - .bind(target_key(&request.target)?) - .bind(target_agent.to_string()) - .execute(&mut *tx) - .await - .map_err(storage_error)?; - } + // Keep active subscriptions readable by older binaries. Opt-outs only + // prevent implicit subscription when this agent participates again. + let statements = match request.change { + SubscriptionChange::Subscribe => [ + "DELETE FROM subscription_opt_outs WHERE board=? AND target=? AND agent=?", + "INSERT OR IGNORE INTO subscriptions(board,target,agent) VALUES(?,?,?)", + ], + SubscriptionChange::Unsubscribe => [ + "DELETE FROM subscriptions WHERE board=? AND target=? AND agent=?", + "INSERT OR IGNORE INTO subscription_opt_outs(board,target,agent) VALUES(?,?,?)", + ], + }; + for statement in statements { + sqlx::query(statement) + .bind(self.identity.to_string()) + .bind(target_key(&request.target)?) + .bind(target_agent.to_string()) + .execute(&mut *tx) + .await + .map_err(storage_error)?; } tx.commit().await.map_err(storage_error)?; Ok(SubscriptionState { @@ -399,7 +410,7 @@ impl LocalAgentMessageBoard { target: &SubscriptionTarget, agent: ThreadId, ) -> Result<()> { - sqlx::query("INSERT OR IGNORE INTO subscriptions(board,target,agent) VALUES(?,?,?)") + sqlx::query("INSERT OR IGNORE INTO subscriptions(board,target,agent) SELECT ?1,?2,?3 WHERE NOT EXISTS(SELECT 1 FROM subscription_opt_outs WHERE board=?1 AND target=?2 AND agent=?3)") .bind(self.identity.to_string()) .bind(target_key(target)?) .bind(agent.to_string()) diff --git a/codex-rs/ext/agent-message-board/src/local/lifecycle.rs b/codex-rs/ext/agent-message-board/src/local/lifecycle.rs index d7f332fa5e..787d02bf77 100644 --- a/codex-rs/ext/agent-message-board/src/local/lifecycle.rs +++ b/codex-rs/ext/agent-message-board/src/local/lifecycle.rs @@ -75,6 +75,7 @@ impl LocalAgentMessageBoard { .map_err(storage_error)?; for statement in [ "DELETE FROM subscriptions WHERE board=?", + "DELETE FROM subscription_opt_outs WHERE board=?", "DELETE FROM posts WHERE board=?", "DELETE FROM channels WHERE board=?", ] { diff --git a/codex-rs/ext/agent-message-board/src/tools/spec.rs b/codex-rs/ext/agent-message-board/src/tools/spec.rs index 1cfdd55025..134ec17473 100644 --- a/codex-rs/ext/agent-message-board/src/tools/spec.rs +++ b/codex-rs/ext/agent-message-board/src/tools/spec.rs @@ -72,7 +72,7 @@ pub(super) fn tool(name: &str, namespace: Option<&str>, namespace_description: & &[], ), "post" => ( - "Post to exactly one existing channel, new channel, or discussion thread. Posting subscribes you to that discussion. Explicit agent recipients get a notice without subscribing. Notices never start idle agents. Returns metadata, not the post body.", + "Post to exactly one existing channel, new channel, or discussion thread. Posting follows the discussion unless you unsubscribed. Explicit agent recipients get a notice without subscribing. Notices never start idle agents. Returns metadata, not the post body.", &[ "text", "channel_name", diff --git a/codex-rs/ext/agent-message-board/tests/local_board.rs b/codex-rs/ext/agent-message-board/tests/local_board.rs index 6fae5a40e3..8ed21b1e70 100644 --- a/codex-rs/ext/agent-message-board/tests/local_board.rs +++ b/codex-rs/ext/agent-message-board/tests/local_board.rs @@ -133,6 +133,17 @@ async fn shared_handles_resume_posts_and_preserve_subscription_rules() { ); drop(first); + // Reopen an older database, preserving its existing subscriptions. + let legacy_pool = sqlite + .open_read_write_pool(&dir.path().join("agent_message_board_1.sqlite")) + .await + .unwrap(); + sqlx::query("DROP TABLE subscription_opt_outs") + .execute(&legacy_pool) + .await + .unwrap(); + legacy_pool.close().await; + let resumed = LocalAgentMessageBoard::open(&sqlite, tree, host.clone()) .await .unwrap(); @@ -232,6 +243,99 @@ async fn shared_handles_resume_posts_and_preserve_subscription_rules() { expected.sort_by_key(|(id, post)| (id.to_string(), post.message_id)); assert_eq!(received, expected); + resumed + .set_subscription( + root, + SubscriptionRequest { + target: SubscriptionTarget::Thread(metadata.thread_id), + target_agent: None, + change: SubscriptionChange::Unsubscribe, + }, + ) + .await + .unwrap(); + drop(resumed); + let resumed = LocalAgentMessageBoard::open(&sqlite, tree, host.clone()) + .await + .unwrap(); + resumed + .post( + root, + PostRequest { + request_id: "still-opted-out".into(), + destination: PostDestination::Thread(metadata.thread_id), + text: "one more reply without resubscribing".into(), + agents_to_notify: Vec::new(), + }, + ) + .await + .unwrap(); + // The old reader's query must also see the opt-out after participation. + let legacy_pool = sqlite + .open_read_write_pool(&dir.path().join("agent_message_board_1.sqlite")) + .await + .unwrap(); + let legacy_subscribers: Vec = + sqlx::query_scalar("SELECT agent FROM subscriptions WHERE board=? AND target=?") + .bind(tree.to_string()) + .bind(serde_json::to_string(&SubscriptionTarget::Thread(metadata.thread_id)).unwrap()) + .fetch_all(&legacy_pool) + .await + .unwrap(); + assert_eq!(legacy_subscribers, vec![child.to_string()]); + legacy_pool.close().await; + host.notifications.lock().unwrap().clear(); + for (index, agents_to_notify) in [Vec::new(), vec![AgentPath::root()], Vec::new()] + .into_iter() + .enumerate() + { + let reply = resumed + .post( + child, + PostRequest { + request_id: format!("after-unsubscribe-{index}"), + destination: PostDestination::Thread(metadata.thread_id), + text: "reply after opt-out".into(), + agents_to_notify, + }, + ) + .await + .unwrap(); + let expected = if index == 1 { + vec![(root, reply)] + } else { + Vec::new() + }; + assert_eq!( + std::mem::take(&mut *host.notifications.lock().unwrap()), + expected + ); + } + resumed + .set_subscription( + root, + SubscriptionRequest { + target: SubscriptionTarget::Thread(metadata.thread_id), + target_agent: None, + change: SubscriptionChange::Subscribe, + }, + ) + .await + .unwrap(); + let reply = resumed + .post( + child, + PostRequest { + request_id: "resubscribed".into(), + destination: PostDestination::Thread(metadata.thread_id), + text: "notifications restored".into(), + agents_to_notify: Vec::new(), + }, + ) + .await + .unwrap(); + assert_eq!(*host.notifications.lock().unwrap(), vec![(root, reply)]); + let other = LocalAgentMessageBoard::open(&sqlite, SessionId::new(), host.clone()) .await .unwrap(); @@ -248,6 +352,19 @@ async fn shared_handles_resume_posts_and_preserve_subscription_rules() { .await .is_err() ); + LocalAgentMessageBoard::delete_boards(&sqlite, &[tree]) + .await + .unwrap(); + let pool = sqlite + .open_read_write_pool(&dir.path().join("agent_message_board_1.sqlite")) + .await + .unwrap(); + let remaining_opt_outs: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM subscription_opt_outs") + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(remaining_opt_outs, 0); + pool.close().await; } #[tokio::test]