hugo/RuDS
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Enforce unique MCP proposal identities

Browse Source
This commit is contained in:
Chili Palmer
2026-07-22 18:25:19 +02:00
parent 2dcaf7f7f7
commit 70503c8bc2
12 changed files with 353 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
crates/bds-core/migrations/2026-07-22-161429-0000_add_mcp_proposal_entity_status_unique_index/down.sql Normal file
View File

@@ -0,0 +1,38 @@
DROP INDEX mcp_proposals_entity_idx;
CREATE TABLE mcp_proposals_old (
id TEXT PRIMARY KEY NOT NULL,
project_id TEXT NOT NULL REFERENCES projects(id) ON DELETE CASCADE,
kind TEXT NOT NULL CHECK (kind IN (
'draft_post',
'propose_script',
'propose_template',
'propose_media_translation',
'propose_media_metadata',
'propose_post_metadata'
)),
status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN (
'pending', 'executing', 'accepted', 'rejected', 'expired'
)),
entity_id TEXT,
data TEXT NOT NULL,
result TEXT,
created_at BIGINT NOT NULL,
expires_at BIGINT NOT NULL,
resolved_at BIGINT
);
INSERT INTO mcp_proposals_old (
id, project_id, kind, status, entity_id, data, result,
created_at, expires_at, resolved_at
)
SELECT
id, project_id, kind, status, entity_id, data, result,
created_at, expires_at, resolved_at
FROM mcp_proposals;
DROP TABLE mcp_proposals;
ALTER TABLE mcp_proposals_old RENAME TO mcp_proposals;
CREATE INDEX mcp_proposals_project_status_idx
ON mcp_proposals(project_id, status, created_at);

55
crates/bds-core/migrations/2026-07-22-161429-0000_add_mcp_proposal_entity_status_unique_index/up.sql Normal file
View File

@@ -0,0 +1,55 @@
CREATE TABLE mcp_proposals_new (
id TEXT PRIMARY KEY NOT NULL,
project_id TEXT NOT NULL REFERENCES projects(id) ON DELETE CASCADE,
kind TEXT NOT NULL CHECK (kind IN (
'draft_post',
'propose_script',
'propose_template',
'propose_media_translation',
'propose_media_metadata',
'propose_post_metadata'
)),
status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN (
'pending', 'executing', 'accepted', 'rejected', 'expired'
)),
entity_id TEXT NOT NULL,
data TEXT NOT NULL,
result TEXT,
created_at BIGINT NOT NULL,
expires_at BIGINT NOT NULL,
resolved_at BIGINT
);
INSERT INTO mcp_proposals_new (
id, project_id, kind, status, entity_id, data, result,
created_at, expires_at, resolved_at
)
SELECT
id, project_id, kind, status, COALESCE(entity_id, id), data, result,
created_at, expires_at, resolved_at
FROM mcp_proposals;
DROP TABLE mcp_proposals;
ALTER TABLE mcp_proposals_new RENAME TO mcp_proposals;
-- Keep the newest record if a pre-migration database already contains an
-- invalid duplicate. NULL-backed proposals were assigned their own IDs above
-- and therefore remain distinct.
DELETE FROM mcp_proposals AS duplicate
WHERE EXISTS (
SELECT 1
FROM mcp_proposals AS keeper
WHERE keeper.kind = duplicate.kind
AND keeper.entity_id = duplicate.entity_id
AND keeper.status = duplicate.status
AND (
keeper.created_at > duplicate.created_at
OR (keeper.created_at = duplicate.created_at AND keeper.id > duplicate.id)
)
);
CREATE INDEX mcp_proposals_project_status_idx
ON mcp_proposals(project_id, status, created_at);
CREATE UNIQUE INDEX mcp_proposals_entity_idx
ON mcp_proposals(kind, entity_id, status);

9
crates/bds-core/src/db/connection.rs
View File

@@ -62,6 +62,15 @@ impl DbConnection {
)
}
#[cfg(test)]
pub(crate) fn insert_legacy_null_mcp_proposal_for_test(&self) -> diesel::QueryResult<()> {
self.0.borrow_mut().batch_execute(
"INSERT INTO mcp_proposals \
(id, project_id, kind, status, entity_id, data, created_at, expires_at) \
VALUES ('null-entity', 'p1', 'draft_post', 'pending', NULL, '{}', 3, 99)",
)
}
/// Filesystem database path for sibling surfaces that must open their own
/// short-lived connection (gallery workers, preview servers, Lua hosts).
pub fn database_path(&self) -> diesel::QueryResult<std::path::PathBuf> {

55
crates/bds-core/src/db/migrations.rs
View File

@@ -36,7 +36,60 @@ mod tests {
let applied = db
.conn()
.with_migrations(|conn| conn.applied_migrations().unwrap().len());
assert_eq!(applied, 8);
assert_eq!(applied, 9);
}
#[test]
fn mcp_proposal_upgrade_normalizes_entities_and_enforces_uniqueness() {
use crate::db::queries::{mcp_proposal, project};
use crate::model::{McpProposal, ProposalKind, ProposalStatus};
let db = Database::open_in_memory().unwrap();
db.conn().with_migrations(|conn| {
for _ in 0..8 {
conn.run_next_migration(MIGRATIONS).unwrap();
}
});
project::insert_project(db.conn(), &project::make_test_project("p1", "blog")).unwrap();
let targeted = |id: &str, created_at| McpProposal {
id: id.into(),
project_id: "p1".into(),
kind: ProposalKind::ProposePostMetadata,
status: ProposalStatus::Pending,
entity_id: "post-1".into(),
data: "{}".into(),
result: None,
created_at,
expires_at: 99,
resolved_at: None,
};
mcp_proposal::insert_proposal(db.conn(), &targeted("older", 1)).unwrap();
mcp_proposal::insert_proposal(db.conn(), &targeted("newer", 2)).unwrap();
db.conn()
.insert_legacy_null_mcp_proposal_for_test()
.unwrap();
run_migrations(db.conn()).unwrap();
let proposals = mcp_proposal::list_proposals(db.conn(), "p1").unwrap();
assert_eq!(proposals.len(), 2);
assert!(proposals.iter().any(|proposal| proposal.id == "newer"));
assert_eq!(
proposals
.iter()
.find(|proposal| proposal.id == "null-entity")
.unwrap()
.entity_id,
"null-entity"
);
assert!(matches!(
mcp_proposal::insert_proposal(db.conn(), &targeted("duplicate", 4)),
Err(diesel::result::Error::DatabaseError(
diesel::result::DatabaseErrorKind::UniqueViolation,
_
))
));
}
#[test]

34
crates/bds-core/src/db/queries/mcp_proposal.rs
View File

@@ -48,6 +48,21 @@ pub fn list_pending_proposals(
pub fn expire_pending(conn: &DbConnection, now: i64) -> QueryResult<usize> {
conn.with(|connection| {
let expiring = mcp_proposals::table
.filter(mcp_proposals::status.eq(ProposalStatus::Pending))
.filter(mcp_proposals::expires_at.le(now))
.select(McpProposal::as_select())
.load::<McpProposal>(connection)?;
for proposal in &expiring {
diesel::delete(
mcp_proposals::table
.filter(mcp_proposals::id.ne(&proposal.id))
.filter(mcp_proposals::kind.eq(proposal.kind))
.filter(mcp_proposals::entity_id.eq(&proposal.entity_id))
.filter(mcp_proposals::status.eq(ProposalStatus::Expired)),
)
.execute(connection)?;
}
diesel::update(
mcp_proposals::table
.filter(mcp_proposals::status.eq(ProposalStatus::Pending))
@@ -62,6 +77,23 @@ pub fn expire_pending(conn: &DbConnection, now: i64) -> QueryResult<usize> {
})
}
pub fn delete_status_collision(
conn: &DbConnection,
proposal: &McpProposal,
status: ProposalStatus,
) -> QueryResult<usize> {
conn.with(|connection| {
diesel::delete(
mcp_proposals::table
.filter(mcp_proposals::id.ne(&proposal.id))
.filter(mcp_proposals::kind.eq(proposal.kind))
.filter(mcp_proposals::entity_id.eq(&proposal.entity_id))
.filter(mcp_proposals::status.eq(status)),
)
.execute(connection)
})
}
pub fn claim_pending(conn: &DbConnection, id: &str, now: i64) -> QueryResult<bool> {
conn.with(|connection| {
diesel::update(
@@ -123,7 +155,7 @@ mod tests {
project_id: "p1".into(),
kind: ProposalKind::DraftPost,
status: ProposalStatus::Pending,
entity_id: None,
entity_id: id.into(),
data: "{}".into(),
result: None,
created_at: 1,

2
crates/bds-core/src/db/schema.rs
View File

@@ -161,7 +161,7 @@ diesel::table! {
project_id -> Text,
kind -> Text,
status -> Text,
entity_id -> Nullable<Text>,
entity_id -> Text,
data -> Text,
result -> Nullable<Text>,
created_at -> BigInt,

35
crates/bds-core/src/engine/mcp/mod.rs
View File

@@ -98,7 +98,17 @@ pub fn list_pending_proposals(
}
pub fn expire_proposals(conn: &DbConnection) -> EngineResult<usize> {
let expired = proposal_q::expire_pending(conn, now_unix_ms())?;
conn.begin_savepoint()?;
let expired = match proposal_q::expire_pending(conn, now_unix_ms()) {
Ok(expired) => {
conn.release_savepoint()?;
expired
}
Err(error) => {
let _ = conn.rollback_savepoint();
return Err(error.into());
}
};
if expired > 0 {
notify_proposals_changed();
}
@@ -114,19 +124,37 @@ pub(crate) fn create_proposal(
) -> EngineResult<McpProposal> {
expire_proposals(conn)?;
let now = now_unix_ms();
let entity_id = entity_id
.map(str::to_string)
.unwrap_or_else(|| Uuid::new_v4().to_string());
let proposal = McpProposal {
id: Uuid::new_v4().to_string(),
project_id: project_id.to_string(),
kind,
status: ProposalStatus::Pending,
entity_id: entity_id.map(str::to_string),
entity_id,
data: serde_json::to_string(data)?,
result: None,
created_at: now,
expires_at: now + PROPOSAL_TTL_MS,
resolved_at: None,
};
proposal_q::insert_proposal(conn, &proposal)?;
if let Err(error) = proposal_q::insert_proposal(conn, &proposal) {
if matches!(
error,
diesel::result::Error::DatabaseError(
diesel::result::DatabaseErrorKind::UniqueViolation,
_
)
) {
return Err(EngineError::Conflict(format!(
"proposal already pending for {} entity {}",
proposal.kind.as_str(),
proposal.entity_id
)));
}
return Err(error.into());
}
cli_sync::record_cli_event(
conn,
&DomainEvent::SettingsChanged {
@@ -182,6 +210,7 @@ fn resolve_proposal(
} else {
ProposalStatus::Rejected
};
proposal_q::delete_status_collision(conn, &proposal, status)?;
if !proposal_q::resolve_claimed(
conn,
proposal_id,

2
crates/bds-core/src/model/mcp.rs
View File

@@ -119,7 +119,7 @@ pub struct McpProposal {
pub project_id: String,
pub kind: ProposalKind,
pub status: ProposalStatus,
pub entity_id: Option<String>,
pub entity_id: String,
pub data: String,
pub result: Option<String>,
pub created_at: i64,

128
crates/bds-core/tests/mcp.rs
View File

@@ -2,7 +2,7 @@ use std::net::{IpAddr, Ipv4Addr};
use std::sync::{Arc, Barrier};
use bds_core::db::Database;
use bds_core::engine::{mcp, project};
use bds_core::engine::{EngineError, mcp, project};
use bds_core::model::{DomainEntity, DomainEvent, McpProposal, ProposalKind, ProposalStatus};
use serde_json::json;
@@ -523,7 +523,7 @@ fn expiry_invalid_ids_unavailable_projects_and_concurrent_acceptance_are_safe()
project_id: fixture.project_id.clone(),
kind: ProposalKind::DraftPost,
status: ProposalStatus::Pending,
entity_id: None,
entity_id: "expired-entity".into(),
data: json!({"title":"Expired","content":"Body"}).to_string(),
result: None,
created_at: 1,
@@ -587,6 +587,130 @@ fn expiry_invalid_ids_unavailable_projects_and_concurrent_acceptance_are_safe()
);
}
#[test]
fn duplicate_pending_entity_proposals_are_rejected_cleanly() {
let fixture = Fixture::new();
let post = fixture.post("One target", "Body");
let context = fixture.context();
let arguments = json!({"postId": post.id, "title": "First change"});
context
.call_tool("propose_post_metadata", arguments.clone())
.unwrap();
let error = context
.call_tool("propose_post_metadata", arguments)
.unwrap_err();
assert!(matches!(error, EngineError::Conflict(_)));
assert_eq!(
error.to_string(),
format!(
"conflict: proposal already pending for propose_post_metadata entity {}",
post.id
)
);
let db = fixture.database();
assert_eq!(
mcp::list_pending_proposals(db.conn(), &fixture.project_id)
.unwrap()
.len(),
1
);
}
#[test]
fn untargeted_proposals_receive_distinct_entity_ids() {
let fixture = Fixture::new();
let context = fixture.context();
context
.call_tool("draft_post", json!({"title": "First", "content": "Body"}))
.unwrap();
context
.call_tool("draft_post", json!({"title": "Second", "content": "Body"}))
.unwrap();
let db = fixture.database();
let proposals = mcp::list_pending_proposals(db.conn(), &fixture.project_id).unwrap();
assert_eq!(proposals.len(), 2);
assert_ne!(proposals[0].entity_id, proposals[1].entity_id);
}
#[test]
fn newer_terminal_proposals_replace_older_status_history() {
let fixture = Fixture::new();
let post = fixture.post("Original", "Body");
let context = fixture.context();
let first = context
.call_tool(
"propose_post_metadata",
json!({"postId": post.id, "title": "First change"}),
)
.unwrap();
let db = fixture.database();
mcp::accept_proposal(
db.conn(),
&fixture.data_dir,
first["proposalId"].as_str().unwrap(),
)
.unwrap();
let second = context
.call_tool(
"propose_post_metadata",
json!({"postId": post.id, "title": "Second change"}),
)
.unwrap();
mcp::accept_proposal(
db.conn(),
&fixture.data_dir,
second["proposalId"].as_str().unwrap(),
)
.unwrap();
let accepted = mcp::list_proposals(db.conn(), &fixture.project_id)
.unwrap()
.into_iter()
.filter(|proposal| proposal.status == ProposalStatus::Accepted)
.collect::<Vec<_>>();
assert_eq!(accepted.len(), 1);
assert_eq!(accepted[0].id, second["proposalId"]);
}
#[test]
fn newer_expired_proposals_replace_older_expired_history() {
let fixture = Fixture::new();
let db = fixture.database();
for (id, status, expires_at) in [
("old-expired", ProposalStatus::Expired, 1),
("new-expiring", ProposalStatus::Pending, 2),
] {
bds_core::db::queries::mcp_proposal::insert_proposal(
db.conn(),
&McpProposal {
id: id.into(),
project_id: fixture.project_id.clone(),
kind: ProposalKind::ProposePostMetadata,
status,
entity_id: "same-post".into(),
data: "{}".into(),
result: None,
created_at: expires_at,
expires_at,
resolved_at: (status == ProposalStatus::Expired).then_some(1),
},
)
.unwrap();
}
assert_eq!(mcp::expire_proposals(db.conn()).unwrap(), 1);
let proposals = mcp::list_proposals(db.conn(), &fixture.project_id).unwrap();
assert_eq!(proposals.len(), 1);
assert_eq!(proposals[0].id, "new-expiring");
assert_eq!(proposals[0].status, ProposalStatus::Expired);
}
#[test]
fn protocol_routes_every_family_and_http_enforces_loopback_origin_and_cors() {
let fixture = Fixture::new();