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
This commit is contained in:
jif
2026-09-22 11:07:21 +00:00
committed by copyberry
parent fdd89e78ac
commit 0160eb3497
5 changed files with 147 additions and 17 deletions
+2 -1
View File
@@ -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,
+26 -15
View File
@@ -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())
@@ -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=?",
] {
@@ -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",
@@ -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<String> =
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]