merge: integrate selector-thread merge request authority
This commit is contained in:
@@ -5,10 +5,12 @@ edition.workspace = true
|
||||
license.workspace = true
|
||||
|
||||
[dependencies]
|
||||
chrono = { version = "0.4", default-features = false, features = ["clock", "serde"] }
|
||||
rusqlite.workspace = true
|
||||
serde = { workspace = true, features = ["derive"] }
|
||||
sha2.workspace = true
|
||||
serde_json.workspace = true
|
||||
thiserror.workspace = true
|
||||
uuid = { workspace = true, features = ["v7"] }
|
||||
|
||||
[dev-dependencies]
|
||||
tempfile.workspace = true
|
||||
|
||||
+1275
-1878
File diff suppressed because it is too large
Load Diff
+371
-283
@@ -1,298 +1,386 @@
|
||||
use chrono::{TimeZone, Utc};
|
||||
use merge_request::*;
|
||||
use rusqlite::{Connection, params};
|
||||
use std::sync::{Arc, Barrier};
|
||||
use std::thread;
|
||||
use tempfile::TempDir;
|
||||
|
||||
fn setup() -> (TempDir, SqliteMergeRequestStore) {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let path = dir.path().join("server.db");
|
||||
let conn = Connection::open(&path).unwrap();
|
||||
conn.execute_batch(r#"
|
||||
PRAGMA foreign_keys=ON;
|
||||
CREATE TABLE repositories(workspace_id TEXT NOT NULL,repository_id TEXT NOT NULL,PRIMARY KEY(workspace_id,repository_id));
|
||||
CREATE TABLE typed_tickets(workspace_id TEXT NOT NULL,ticket_id TEXT NOT NULL,workflow_state TEXT NOT NULL,workflow_state_explicit INTEGER NOT NULL DEFAULT 1,updated_at TEXT NOT NULL,PRIMARY KEY(workspace_id,ticket_id));
|
||||
CREATE TABLE typed_ticket_events(workspace_id TEXT NOT NULL,ticket_id TEXT NOT NULL,event_index INTEGER NOT NULL,kind TEXT NOT NULL,author TEXT,at TEXT,status TEXT,from_state TEXT,to_state TEXT,heading TEXT,body TEXT,PRIMARY KEY(workspace_id,ticket_id,event_index));
|
||||
CREATE TABLE typed_ticket_event_attributes(workspace_id TEXT NOT NULL,ticket_id TEXT NOT NULL,event_index INTEGER NOT NULL,key TEXT NOT NULL,value TEXT NOT NULL,PRIMARY KEY(workspace_id,ticket_id,event_index,key));
|
||||
CREATE TABLE ticket_worker_assignments(workspace_id TEXT NOT NULL,ticket_id TEXT NOT NULL,assignment_id TEXT NOT NULL,runtime_id TEXT NOT NULL,worker_id TEXT NOT NULL,PRIMARY KEY(workspace_id,ticket_id,assignment_id));
|
||||
CREATE TABLE ticket_current_worker_assignments(workspace_id TEXT NOT NULL,ticket_id TEXT NOT NULL,assignment_id TEXT NOT NULL,runtime_id TEXT NOT NULL,worker_id TEXT NOT NULL,PRIMARY KEY(workspace_id,ticket_id));
|
||||
"#).unwrap();
|
||||
for ws in ["ws-a", "ws-b"] {
|
||||
conn.execute("INSERT INTO repositories VALUES(?1,'repo')", params![ws])
|
||||
.unwrap();
|
||||
conn.execute(
|
||||
"INSERT INTO typed_tickets VALUES(?1,'T1','inprogress',1,'t0')",
|
||||
params![ws],
|
||||
)
|
||||
.unwrap();
|
||||
conn.execute(
|
||||
"INSERT INTO ticket_worker_assignments VALUES(?1,'T1','A1','R1','W1')",
|
||||
params![ws],
|
||||
)
|
||||
.unwrap();
|
||||
conn.execute(
|
||||
"INSERT INTO ticket_current_worker_assignments VALUES(?1,'T1','A1','R1','W1')",
|
||||
params![ws],
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
drop(conn);
|
||||
let store = SqliteMergeRequestStore::open(&path, "ws-a").unwrap();
|
||||
(dir, store)
|
||||
}
|
||||
|
||||
fn revision(id: &str, ordinal: u64, head: &str) -> MergeRequestRevision {
|
||||
MergeRequestRevision {
|
||||
revision_id: id.into(),
|
||||
ordinal,
|
||||
base_commit: "base".into(),
|
||||
head_commit: head.into(),
|
||||
diff_digest: format!("sha256:diff-{head}"),
|
||||
changed_paths: vec!["src/lib.rs".into()],
|
||||
summary: format!("revision {id}"),
|
||||
assignment_id: "A1".into(),
|
||||
created_at: format!("t{ordinal}"),
|
||||
use rusqlite::Connection;
|
||||
use std::sync::{Arc, Mutex};
|
||||
#[derive(Clone)]
|
||||
struct Assignments(Arc<Mutex<CurrentAssignment>>);
|
||||
impl AssignmentSource for Assignments {
|
||||
fn current_assignment(&self, _: &str, _: &str) -> Result<Option<CurrentAssignment>, String> {
|
||||
Ok(Some(self.0.lock().unwrap().clone()))
|
||||
}
|
||||
}
|
||||
|
||||
fn open(store: &SqliteMergeRequestStore) {
|
||||
store
|
||||
.open_merge_request(OpenMergeRequest {
|
||||
merge_request_id: "MR1".into(),
|
||||
ticket_id: "T1".into(),
|
||||
repository_id: "repo".into(),
|
||||
target_ref_selector: "refs/heads/develop".into(),
|
||||
revision: revision("V1", 1, "head"),
|
||||
authenticated_runtime_id: "R1".into(),
|
||||
authenticated_worker_id: "W1".into(),
|
||||
now: "t1".into(),
|
||||
})
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
fn attempt(store: &SqliteMergeRequestStore, revision: &str, token: &str) {
|
||||
let child = format!("child-{revision}");
|
||||
store
|
||||
.register_reviewer_child_session(RegisterReviewerChildSession {
|
||||
parent_runtime_id: "R1".into(),
|
||||
parent_worker_id: "W1".into(),
|
||||
child_session_id: child.clone(),
|
||||
now: "t2".into(),
|
||||
})
|
||||
.unwrap();
|
||||
store
|
||||
.register_review_attempt(RegisterReviewAttempt {
|
||||
attempt_id: format!("attempt-{revision}"),
|
||||
ticket_id: "T1".into(),
|
||||
revision_id: revision.into(),
|
||||
parent_assignment_id: "A1".into(),
|
||||
parent_runtime_id: "R1".into(),
|
||||
parent_worker_id: "W1".into(),
|
||||
child_session_id: child,
|
||||
capability_token: token.into(),
|
||||
now: "t2".into(),
|
||||
})
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
fn approve(store: &SqliteMergeRequestStore, revision: &str, token: &str) {
|
||||
attempt(store, revision, token);
|
||||
store
|
||||
.submit_review(SubmitReview {
|
||||
ticket_id: "T1".into(),
|
||||
revision_id: revision.into(),
|
||||
capability_token: token.into(),
|
||||
decision: ReviewDecision::Approve,
|
||||
body: "approved".into(),
|
||||
findings: vec![],
|
||||
now: "t3".into(),
|
||||
})
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
fn completion(operation_id: &str) -> CompleteMergeRequest {
|
||||
CompleteMergeRequest {
|
||||
operation_id: operation_id.into(),
|
||||
ticket_id: "T1".into(),
|
||||
expected_revision_id: "V1".into(),
|
||||
target_commit: "base".into(),
|
||||
source_commit: "head".into(),
|
||||
result_commit: "head".into(),
|
||||
strategy: MergeStrategy::FastForward,
|
||||
resolution: MergeResolution::None,
|
||||
implementation_assignment_id: "A1".into(),
|
||||
completion_actor_runtime_id: "OR".into(),
|
||||
completion_actor_worker_id: "OW".into(),
|
||||
now: "t4".into(),
|
||||
struct Repositories;
|
||||
impl RepositorySource for Repositories {
|
||||
fn repository_belongs_to_workspace(&self, w: &str, r: &str) -> Result<bool, String> {
|
||||
Ok(w == "W" && r == "R")
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn target_movement_does_not_invalidate_source_revision_approval() {
|
||||
let (_dir, store) = setup();
|
||||
open(&store);
|
||||
approve(&store, "V1", "token-v1");
|
||||
for target in ["base", "advanced-target"] {
|
||||
let readiness = store
|
||||
.readiness_for_ticket_with_target("T1", Some(target))
|
||||
.unwrap();
|
||||
assert!(
|
||||
readiness.ready,
|
||||
"target movement must not invalidate source approval"
|
||||
);
|
||||
assert_eq!(readiness.review_status, ReviewStatus::Approved);
|
||||
assert_eq!(readiness.observed_target_commit.as_deref(), Some(target));
|
||||
fn at(s: u32) -> chrono::DateTime<Utc> {
|
||||
Utc.with_ymd_and_hms(2026, 7, 26, 12, 0, s)
|
||||
.single()
|
||||
.unwrap()
|
||||
}
|
||||
fn auth() -> MergeRequestAuth {
|
||||
MergeRequestAuth {
|
||||
workspace_id: "W".into(),
|
||||
repository_id: "R".into(),
|
||||
runtime_id: "runtime".into(),
|
||||
worker_id: "coder".into(),
|
||||
assignment_id: "A".into(),
|
||||
}
|
||||
store
|
||||
.add_revision(AddRevision {
|
||||
ticket_id: "T1".into(),
|
||||
expected_current_revision_id: "V1".into(),
|
||||
revision: revision("V2", 2, "head2"),
|
||||
authenticated_runtime_id: "R1".into(),
|
||||
authenticated_worker_id: "W1".into(),
|
||||
now: "t5".into(),
|
||||
})
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
store.readiness_for_ticket("T1").unwrap().review_status,
|
||||
ReviewStatus::Pending
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn completion_records_one_final_merge_outcome_and_replays_idempotently() {
|
||||
let (_dir, store) = setup();
|
||||
open(&store);
|
||||
approve(&store, "V1", "token-v1");
|
||||
let first = store.complete(completion("OP1")).unwrap();
|
||||
assert!(!first.replayed);
|
||||
assert_eq!(first.ticket_state, "done");
|
||||
let merged = store.show_for_ticket("T1").unwrap().unwrap();
|
||||
assert_eq!(merged.state, MergeRequestState::Merged);
|
||||
assert_eq!(merged.merged_revision_id.as_deref(), Some("V1"));
|
||||
assert_eq!(merged.merged_target_commit.as_deref(), Some("base"));
|
||||
assert_eq!(merged.merged_result_commit.as_deref(), Some("head"));
|
||||
assert_eq!(merged.merge_strategy, Some(MergeStrategy::FastForward));
|
||||
assert_eq!(merged.merge_resolution, Some(MergeResolution::None));
|
||||
assert_eq!(merged.merged_by_runtime_id.as_deref(), Some("OR"));
|
||||
assert_eq!(merged.merged_by_worker_id.as_deref(), Some("OW"));
|
||||
assert!(store.complete(completion("OP1")).unwrap().replayed);
|
||||
let mut conflicting = completion("OP1");
|
||||
conflicting.target_commit = "other".into();
|
||||
assert!(matches!(
|
||||
store.complete(conflicting),
|
||||
Err(MergeRequestError::OperationConflict)
|
||||
));
|
||||
fn fixture() -> (tempfile::TempDir, MergeRequestStore) {
|
||||
let d = tempfile::tempdir().unwrap();
|
||||
let p = d.path().join("db");
|
||||
let c = Connection::open(&p).unwrap();
|
||||
c.execute_batch("CREATE TABLE workspaces(workspace_id TEXT PRIMARY KEY);CREATE TABLE repositories(workspace_id TEXT,repository_id TEXT,PRIMARY KEY(workspace_id,repository_id));CREATE TABLE ticket_current_worker_assignments(workspace_id TEXT,ticket_id TEXT,assignment_id TEXT,runtime_id TEXT,worker_id TEXT,updated_at TEXT,PRIMARY KEY(workspace_id,ticket_id));CREATE TABLE typed_tickets(workspace_id TEXT,ticket_id TEXT,workflow_state TEXT,workflow_state_explicit INTEGER,updated_at TEXT,PRIMARY KEY(workspace_id,ticket_id));CREATE TABLE typed_ticket_events(workspace_id TEXT,ticket_id TEXT,event_index INTEGER,kind TEXT,author TEXT,at TEXT,from_state TEXT,to_state TEXT,heading TEXT,body TEXT,PRIMARY KEY(workspace_id,ticket_id,event_index));CREATE TABLE typed_ticket_event_attributes(workspace_id TEXT,ticket_id TEXT,event_index INTEGER,key TEXT,value TEXT,PRIMARY KEY(workspace_id,ticket_id,event_index,key));INSERT INTO workspaces VALUES('W');INSERT INTO repositories VALUES('W','R');INSERT INTO ticket_current_worker_assignments VALUES('W','T','A','runtime','coder','t');INSERT INTO typed_tickets VALUES('W','T','inprogress',1,'t');").unwrap();
|
||||
drop(c);
|
||||
let a = Assignments(Arc::new(Mutex::new(CurrentAssignment {
|
||||
assignment_id: "A".into(),
|
||||
ticket_id: "T".into(),
|
||||
runtime_id: "runtime".into(),
|
||||
worker_id: "coder".into(),
|
||||
})));
|
||||
let s = MergeRequestStore::open(&p, Arc::new(a), Arc::new(Repositories)).unwrap();
|
||||
(d, s)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn completion_rejects_invalid_or_non_current_source_outcomes_without_side_effects() {
|
||||
let (_dir, store) = setup();
|
||||
open(&store);
|
||||
approve(&store, "V1", "token-v1");
|
||||
let mut invalid_ff = completion("bad-ff");
|
||||
invalid_ff.result_commit = "different".into();
|
||||
assert!(matches!(
|
||||
store.complete(invalid_ff),
|
||||
Err(MergeRequestError::InvalidMergeOutcome(_))
|
||||
));
|
||||
let mut invalid_merge = completion("bad-merge");
|
||||
invalid_merge.strategy = MergeStrategy::Merge;
|
||||
assert!(matches!(
|
||||
store.complete(invalid_merge),
|
||||
Err(MergeRequestError::InvalidMergeOutcome(_))
|
||||
));
|
||||
let mut wrong_source = completion("wrong-source");
|
||||
wrong_source.source_commit = "not-approved".into();
|
||||
wrong_source.result_commit = "not-approved".into();
|
||||
assert!(matches!(
|
||||
store.complete(wrong_source),
|
||||
Err(MergeRequestError::InvalidMergeOutcome(_))
|
||||
));
|
||||
assert_eq!(
|
||||
store.show_for_ticket("T1").unwrap().unwrap().state,
|
||||
MergeRequestState::Open
|
||||
);
|
||||
let conn = Connection::open(store.db_path()).unwrap();
|
||||
assert_eq!(
|
||||
conn.query_row(
|
||||
"SELECT workflow_state FROM typed_tickets WHERE workspace_id='ws-a' AND ticket_id='T1'",
|
||||
[],
|
||||
|row| row.get::<_, String>(0)
|
||||
)
|
||||
.unwrap(),
|
||||
"inprogress"
|
||||
);
|
||||
fn open(s: &MergeRequestStore) {
|
||||
s.open_merge_request(OpenMergeRequest {
|
||||
merge_request_id: "MR".into(),
|
||||
ticket_id: "T".into(),
|
||||
repository_id: "R".into(),
|
||||
selector_from: "work/t".into(),
|
||||
selector_to: "develop".into(),
|
||||
summary: "summary".into(),
|
||||
auth: auth(),
|
||||
now: at(1),
|
||||
})
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn concurrent_completion_converges_on_one_operation() {
|
||||
let (_dir, store) = setup();
|
||||
open(&store);
|
||||
approve(&store, "V1", "token-v1");
|
||||
let path = store.db_path().to_path_buf();
|
||||
let barrier = Arc::new(Barrier::new(3));
|
||||
let mut handles = Vec::new();
|
||||
for _ in 0..2 {
|
||||
let path = path.clone();
|
||||
let barrier = barrier.clone();
|
||||
handles.push(thread::spawn(move || {
|
||||
let store = SqliteMergeRequestStore::open_verified(path, "ws-a").unwrap();
|
||||
barrier.wait();
|
||||
store.complete(completion("OP-concurrent"))
|
||||
}));
|
||||
}
|
||||
barrier.wait();
|
||||
let outcomes: Vec<_> = handles
|
||||
.into_iter()
|
||||
.map(|handle| handle.join().unwrap().unwrap())
|
||||
.collect();
|
||||
assert_eq!(
|
||||
outcomes.iter().filter(|outcome| !outcome.replayed).count(),
|
||||
1
|
||||
);
|
||||
assert_eq!(
|
||||
outcomes.iter().filter(|outcome| outcome.replayed).count(),
|
||||
1
|
||||
);
|
||||
fn request(s: &MergeRequestStore, subject: &str, token: &str) -> ReviewRequestedEvent {
|
||||
s.register_reviewer_child_session(RegisterReviewerChildSession {
|
||||
workspace_id: "W".into(),
|
||||
parent_runtime_id: "runtime".into(),
|
||||
parent_worker_id: "coder".into(),
|
||||
child_session_id: format!("child-{token}"),
|
||||
reviewer_profile: "builtin:reviewer".into(),
|
||||
now: at(2),
|
||||
})
|
||||
.unwrap();
|
||||
s.request_review(RequestMergeRequestReview {
|
||||
ticket_id: "T".into(),
|
||||
subject_ref: subject.into(),
|
||||
child_session_id: format!("child-{token}"),
|
||||
capability_token: token.into(),
|
||||
auth: auth(),
|
||||
now: at(3),
|
||||
})
|
||||
.unwrap()
|
||||
.request_event
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reviewer_attempt_is_bound_to_direct_child_and_current_assignment() {
|
||||
let (_dir, store) = setup();
|
||||
open(&store);
|
||||
store
|
||||
.register_reviewer_child_session(RegisterReviewerChildSession {
|
||||
parent_runtime_id: "R1".into(),
|
||||
parent_worker_id: "W1".into(),
|
||||
child_session_id: "child".into(),
|
||||
now: "t2".into(),
|
||||
})
|
||||
.unwrap();
|
||||
store
|
||||
.register_review_attempt(RegisterReviewAttempt {
|
||||
attempt_id: "attempt".into(),
|
||||
ticket_id: "T1".into(),
|
||||
revision_id: "V1".into(),
|
||||
parent_assignment_id: "A1".into(),
|
||||
parent_runtime_id: "R1".into(),
|
||||
parent_worker_id: "W1".into(),
|
||||
child_session_id: "child".into(),
|
||||
capability_token: "token".into(),
|
||||
now: "t2".into(),
|
||||
})
|
||||
.unwrap();
|
||||
let wrong_token = store.submit_review(SubmitReview {
|
||||
ticket_id: "T1".into(),
|
||||
revision_id: "V1".into(),
|
||||
capability_token: "wrong".into(),
|
||||
fn approve(s: &MergeRequestStore, subject: &str, token: &str) -> ReviewEvent {
|
||||
request(s, subject, token);
|
||||
s.submit_review(SubmitMergeRequestReview {
|
||||
ticket_id: "T".into(),
|
||||
current_subject_ref: subject.into(),
|
||||
capability_token: token.into(),
|
||||
decision: ReviewDecision::Approve,
|
||||
body: "approved".into(),
|
||||
findings: vec![],
|
||||
now: "t3".into(),
|
||||
now: at(4),
|
||||
})
|
||||
.unwrap()
|
||||
}
|
||||
#[test]
|
||||
fn selectors_thread_and_completion_have_no_revision_or_commit_api() {
|
||||
let (_d, s) = fixture();
|
||||
open(&s);
|
||||
let review = approve(&s, "opaque-source-ref", "token");
|
||||
let ready = s
|
||||
.readiness(ReadinessCheck {
|
||||
ticket_id: "T".into(),
|
||||
current_subject_ref: Some("opaque-source-ref".into()),
|
||||
auth: auth(),
|
||||
})
|
||||
.unwrap();
|
||||
assert!(ready.ready);
|
||||
let merged = s
|
||||
.complete(CompleteMergeRequest {
|
||||
ticket_id: "T".into(),
|
||||
operation_id: "op".into(),
|
||||
approval_event_id: review.event_id,
|
||||
current_subject_ref: "opaque-source-ref".into(),
|
||||
target_ref_before: "old-target-ref".into(),
|
||||
target_ref_after: "new-target-ref".into(),
|
||||
strategy: MergeStrategy::FastForward,
|
||||
resolution: ConflictResolution::None,
|
||||
auth: auth(),
|
||||
now: at(5),
|
||||
})
|
||||
.unwrap();
|
||||
assert_eq!(merged.approved_source_ref, "opaque-source-ref");
|
||||
let mr = s.get("W", "T").unwrap();
|
||||
assert_eq!(mr.selector_from.as_deref(), Some("work/t"));
|
||||
assert_eq!(mr.state, MergeRequestState::Merged);
|
||||
let json = serde_json::to_string(&mr).unwrap();
|
||||
for banned in [
|
||||
"revision_id",
|
||||
"attempt_id",
|
||||
"base_commit",
|
||||
"head_commit",
|
||||
"source_commit",
|
||||
"result_commit",
|
||||
"current_revision",
|
||||
] {
|
||||
assert!(!json.contains(banned), "{banned} in {json}")
|
||||
}
|
||||
}
|
||||
#[test]
|
||||
fn source_move_cancels_submission_and_old_approval_is_reusable_when_source_returns() {
|
||||
let (_d, s) = fixture();
|
||||
open(&s);
|
||||
let approved = approve(&s, "source-a", "one");
|
||||
request(&s, "source-b", "two");
|
||||
assert!(
|
||||
s.submit_review(SubmitMergeRequestReview {
|
||||
ticket_id: "T".into(),
|
||||
current_subject_ref: "source-c".into(),
|
||||
capability_token: "two".into(),
|
||||
decision: ReviewDecision::Approve,
|
||||
body: "stale".into(),
|
||||
findings: vec![],
|
||||
now: at(6)
|
||||
})
|
||||
.is_err()
|
||||
);
|
||||
let mr = s.get("W", "T").unwrap();
|
||||
assert!(
|
||||
mr.thread
|
||||
.iter()
|
||||
.any(|e| matches!(e, MergeRequestThreadEvent::ReviewCancelled(_)))
|
||||
);
|
||||
assert_eq!(
|
||||
mr.effective_review("source-a").map(|r| &r.event_id),
|
||||
Some(&approved.event_id)
|
||||
);
|
||||
}
|
||||
#[test]
|
||||
fn review_revocation_invalidates_readiness() {
|
||||
let (_d, s) = fixture();
|
||||
open(&s);
|
||||
let review = approve(&s, "source", "one");
|
||||
s.revoke_review(RevokeMergeRequestReview {
|
||||
ticket_id: "T".into(),
|
||||
review_event_id: review.event_id,
|
||||
reason: "bad evidence".into(),
|
||||
auth: auth(),
|
||||
now: at(7),
|
||||
})
|
||||
.unwrap();
|
||||
let r = s
|
||||
.readiness(ReadinessCheck {
|
||||
ticket_id: "T".into(),
|
||||
current_subject_ref: Some("source".into()),
|
||||
auth: auth(),
|
||||
})
|
||||
.unwrap();
|
||||
assert!(!r.ready);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn v11_migration_preserves_review_events_and_requires_selector_repair() {
|
||||
let c = Connection::open_in_memory().unwrap();
|
||||
c.execute_batch("CREATE TABLE repositories(workspace_id TEXT,repository_id TEXT,PRIMARY KEY(workspace_id,repository_id));CREATE TABLE typed_tickets(workspace_id TEXT,ticket_id TEXT,PRIMARY KEY(workspace_id,ticket_id));INSERT INTO repositories VALUES('W','R');INSERT INTO typed_tickets VALUES('W','T');CREATE TABLE merge_request_schema(singleton INTEGER PRIMARY KEY,version INTEGER);INSERT INTO merge_request_schema VALUES(1,11);CREATE TABLE merge_requests(workspace_id TEXT,merge_request_id TEXT,repository_id TEXT,state TEXT,target_ref_selector TEXT,current_revision_ordinal INTEGER,current_revision_id TEXT,created_at TEXT,updated_at TEXT,merged_revision_id TEXT,merged_at TEXT);CREATE TABLE merge_request_ticket_relations(workspace_id TEXT,merge_request_id TEXT,ticket_id TEXT,relation_kind TEXT,created_at TEXT);CREATE TABLE merge_request_revisions(workspace_id TEXT,merge_request_id TEXT,revision_id TEXT,ordinal INTEGER,base_commit TEXT,head_commit TEXT,diff_digest TEXT,summary TEXT,assignment_id TEXT,created_at TEXT);CREATE TABLE merge_request_revision_paths(workspace_id TEXT,merge_request_id TEXT,revision_id TEXT,ordinal INTEGER,path TEXT);CREATE TABLE merge_request_reviewer_child_sessions(workspace_id TEXT,child_session_id TEXT,parent_runtime_id TEXT,parent_worker_id TEXT,reviewer_profile TEXT,registered_at TEXT);CREATE TABLE merge_request_review_attempts(workspace_id TEXT,attempt_id TEXT,merge_request_id TEXT,ticket_id TEXT,revision_id TEXT,revision_ordinal INTEGER,parent_assignment_id TEXT,parent_runtime_id TEXT,parent_worker_id TEXT,child_session_id TEXT,reviewer_effective_profile TEXT,capability_token TEXT,status TEXT,created_at TEXT,consumed_at TEXT);CREATE TABLE merge_request_reviews(workspace_id TEXT,attempt_id TEXT,merge_request_id TEXT,revision_id TEXT,decision TEXT,body TEXT,submitted_at TEXT);CREATE TABLE merge_request_review_findings(workspace_id TEXT,attempt_id TEXT,ordinal INTEGER,severity TEXT,code TEXT,path TEXT,line INTEGER,body TEXT);CREATE TABLE merge_request_completion_operations(workspace_id TEXT,operation_id TEXT,ticket_id TEXT,revision_id TEXT,authority_kind TEXT,implementation_assignment_id TEXT,completion_actor_runtime_id TEXT,completion_actor_worker_id TEXT,target_commit TEXT,source_commit TEXT,result_commit TEXT,strategy TEXT,resolution TEXT,fingerprint TEXT,status TEXT,result_ticket_state TEXT,created_at TEXT,updated_at TEXT);INSERT INTO merge_requests VALUES('W','MR','R','open','develop',1,'V','2026-07-26T12:00:00Z','2026-07-26T12:00:00Z',NULL,NULL);INSERT INTO merge_request_ticket_relations VALUES('W','MR','T','implements','2026-07-26T12:00:00Z');INSERT INTO merge_request_revisions VALUES('W','MR','V',1,'base','subject','digest','summary','A','2026-07-26T12:00:00Z');INSERT INTO merge_request_review_attempts VALUES('W','AT','MR','T','V',1,'A','runtime','coder','child','builtin:reviewer','token','submitted','2026-07-26T12:00:00Z','2026-07-26T12:00:01Z');INSERT INTO merge_request_reviews VALUES('W','AT','MR','V','approve','approved','2026-07-26T12:00:01Z');INSERT INTO merge_request_review_attempts VALUES('W','PENDING','MR','T','V',1,'A','runtime','coder','pending-child','builtin:reviewer','pending-token','registered','2026-07-26T12:00:02Z',NULL);").unwrap();
|
||||
merge_request::migrate(&c).unwrap();
|
||||
let selector: Option<String> = c
|
||||
.query_row("SELECT selector_from FROM merge_requests", [], |r| r.get(0))
|
||||
.unwrap();
|
||||
assert!(selector.is_none());
|
||||
let kinds: String = c
|
||||
.query_row(
|
||||
"SELECT group_concat(kind,',') FROM merge_request_thread_events ORDER BY sequence",
|
||||
[],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
kinds,
|
||||
"review_requested,review,review_requested,review_cancelled"
|
||||
);
|
||||
let old: bool = c
|
||||
.query_row(
|
||||
"SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE name='merge_request_revisions')",
|
||||
[],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(!old);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn authority_reads_full_thread_while_public_pages_remain_bounded() {
|
||||
let (_d, store) = fixture();
|
||||
open(&store);
|
||||
for index in 0..55 {
|
||||
approve(&store, "same-subject", &format!("token-{index}"));
|
||||
}
|
||||
let mr = store.get("W", "T").unwrap();
|
||||
assert!(mr.thread.len() > 100);
|
||||
assert_eq!(
|
||||
mr.effective_review("same-subject").unwrap().decision,
|
||||
ReviewDecision::Approve
|
||||
);
|
||||
assert_eq!(store.thread_page("W", "T", None, 20).unwrap().len(), 20);
|
||||
assert_eq!(
|
||||
store.thread_page("W", "T", Some(100), 20).unwrap().len(),
|
||||
11
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn completion_rejects_superseded_approval_for_same_subject() {
|
||||
let (_d, store) = fixture();
|
||||
open(&store);
|
||||
let old_approval = approve(&store, "subject", "approval");
|
||||
request(&store, "subject", "changes");
|
||||
store
|
||||
.submit_review(SubmitMergeRequestReview {
|
||||
ticket_id: "T".into(),
|
||||
current_subject_ref: "subject".into(),
|
||||
capability_token: "changes".into(),
|
||||
decision: ReviewDecision::RequestChanges,
|
||||
body: "changes required".into(),
|
||||
findings: vec![],
|
||||
now: at(5),
|
||||
})
|
||||
.unwrap();
|
||||
let result = store.complete(CompleteMergeRequest {
|
||||
ticket_id: "T".into(),
|
||||
operation_id: "op".into(),
|
||||
approval_event_id: old_approval.event_id,
|
||||
current_subject_ref: "subject".into(),
|
||||
target_ref_before: "before".into(),
|
||||
target_ref_after: "after".into(),
|
||||
strategy: MergeStrategy::FastForward,
|
||||
resolution: ConflictResolution::None,
|
||||
auth: auth(),
|
||||
now: at(6),
|
||||
});
|
||||
assert!(matches!(
|
||||
wrong_token,
|
||||
Err(MergeRequestError::InvalidReviewAttempt)
|
||||
));
|
||||
assert!(matches!(result, Err(MergeRequestError::NotReady(_))));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn completion_cancels_outstanding_grants_and_late_submit_fails() {
|
||||
let (_d, store) = fixture();
|
||||
open(&store);
|
||||
let approval = approve(&store, "subject", "approval");
|
||||
request(&store, "other-subject", "pending");
|
||||
store
|
||||
.complete(CompleteMergeRequest {
|
||||
ticket_id: "T".into(),
|
||||
operation_id: "op".into(),
|
||||
approval_event_id: approval.event_id,
|
||||
current_subject_ref: "subject".into(),
|
||||
target_ref_before: "before".into(),
|
||||
target_ref_after: "after".into(),
|
||||
strategy: MergeStrategy::FastForward,
|
||||
resolution: ConflictResolution::None,
|
||||
auth: auth(),
|
||||
now: at(6),
|
||||
})
|
||||
.unwrap();
|
||||
let late = store.submit_review(SubmitMergeRequestReview {
|
||||
ticket_id: "T".into(),
|
||||
current_subject_ref: "other-subject".into(),
|
||||
capability_token: "pending".into(),
|
||||
decision: ReviewDecision::Approve,
|
||||
body: "too late".into(),
|
||||
findings: vec![],
|
||||
now: at(7),
|
||||
});
|
||||
assert!(matches!(late, Err(MergeRequestError::Unauthorized(_))));
|
||||
let mr = store.get("W", "T").unwrap();
|
||||
assert!(mr.thread.iter().any(|event| matches!(event,
|
||||
MergeRequestThreadEvent::ReviewCancelled(value)
|
||||
if value.reason.contains("completed before review submission"))));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn selector_repair_requires_and_accepts_an_approved_resolved_subject() {
|
||||
let (dir, store) = fixture();
|
||||
open(&store);
|
||||
approve(&store, "approved-subject", "approval");
|
||||
Connection::open(dir.path().join("db")).unwrap()
|
||||
.execute("UPDATE merge_requests SET selector_from=NULL WHERE workspace_id='W' AND merge_request_id='MR'", [])
|
||||
.unwrap();
|
||||
let repaired = store
|
||||
.repair_selector_from(RepairSelectorFrom {
|
||||
workspace_id: "W".into(),
|
||||
ticket_id: "T".into(),
|
||||
selector_from: "restored-work".into(),
|
||||
resolved_subject_ref: "approved-subject".into(),
|
||||
repaired_by: WorkerIdentity {
|
||||
runtime_id: "browser".into(),
|
||||
worker_id: "user".into(),
|
||||
},
|
||||
reason: "confirmed migrated source".into(),
|
||||
now: at(8),
|
||||
})
|
||||
.unwrap();
|
||||
assert_eq!(repaired.selector_from.as_deref(), Some("restored-work"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn selector_repair_rejects_unapproved_resolved_subject() {
|
||||
let (dir, store) = fixture();
|
||||
open(&store);
|
||||
approve(&store, "approved-subject", "approval");
|
||||
Connection::open(dir.path().join("db")).unwrap()
|
||||
.execute("UPDATE merge_requests SET selector_from=NULL WHERE workspace_id='W' AND merge_request_id='MR'", [])
|
||||
.unwrap();
|
||||
let result = store.repair_selector_from(RepairSelectorFrom {
|
||||
workspace_id: "W".into(),
|
||||
ticket_id: "T".into(),
|
||||
selector_from: "wrong-work".into(),
|
||||
resolved_subject_ref: "different-subject".into(),
|
||||
repaired_by: WorkerIdentity {
|
||||
runtime_id: "browser".into(),
|
||||
worker_id: "user".into(),
|
||||
},
|
||||
reason: "wrong candidate".into(),
|
||||
now: at(8),
|
||||
});
|
||||
assert!(matches!(result, Err(MergeRequestError::NotReady(_))));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn transactional_completion_rejects_assignment_changed_in_control_plane_db() {
|
||||
let (dir, store) = fixture();
|
||||
open(&store);
|
||||
let approval = approve(&store, "subject", "approval");
|
||||
Connection::open(dir.path().join("db")).unwrap()
|
||||
.execute("UPDATE ticket_current_worker_assignments SET assignment_id='B' WHERE workspace_id='W' AND ticket_id='T'", [])
|
||||
.unwrap();
|
||||
let result = store.complete(CompleteMergeRequest {
|
||||
ticket_id: "T".into(),
|
||||
operation_id: "op".into(),
|
||||
approval_event_id: approval.event_id,
|
||||
current_subject_ref: "subject".into(),
|
||||
target_ref_before: "before".into(),
|
||||
target_ref_after: "after".into(),
|
||||
strategy: MergeStrategy::FastForward,
|
||||
resolution: ConflictResolution::None,
|
||||
auth: auth(),
|
||||
now: at(9),
|
||||
});
|
||||
assert!(matches!(result, Err(MergeRequestError::Unauthorized(_))));
|
||||
let state: String = Connection::open(dir.path().join("db"))
|
||||
.unwrap()
|
||||
.query_row(
|
||||
"SELECT workflow_state FROM typed_tickets WHERE workspace_id='W' AND ticket_id='T'",
|
||||
[],
|
||||
|row| row.get(0),
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(state, "inprogress");
|
||||
}
|
||||
|
||||
@@ -435,12 +435,6 @@ impl FeatureModule for ManageWorkerFeature {
|
||||
WorkerOperation::Remove => {
|
||||
definition::<WorkerRemoveInput>(operation, self.control.clone())
|
||||
}
|
||||
WorkerOperation::Share | WorkerOperation::Transfer => {
|
||||
definition::<WorkerDelegateInput>(operation, self.control.clone())
|
||||
}
|
||||
WorkerOperation::Revoke => {
|
||||
definition::<WorkerRevokeInput>(operation, self.control.clone())
|
||||
}
|
||||
};
|
||||
context
|
||||
.tools()
|
||||
@@ -540,19 +534,6 @@ struct WorkerRemoveInput {
|
||||
reason: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize, JsonSchema)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
struct WorkerRevokeInput {
|
||||
grant_id: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize, JsonSchema)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
struct WorkerDelegateInput {
|
||||
grant_id: String,
|
||||
target_controller: WorkerSubjectInput,
|
||||
}
|
||||
|
||||
struct WorkspaceWorkerTool {
|
||||
operation: WorkerOperation,
|
||||
control: Arc<dyn WorkerControlService>,
|
||||
@@ -568,13 +549,10 @@ enum WorkerOperation {
|
||||
Stop,
|
||||
Restore,
|
||||
Remove,
|
||||
Share,
|
||||
Transfer,
|
||||
Revoke,
|
||||
}
|
||||
|
||||
impl WorkerOperation {
|
||||
const ALL: [Self; 11] = [
|
||||
const ALL: [Self; 8] = [
|
||||
Self::List,
|
||||
Self::Spawn,
|
||||
Self::SendInput,
|
||||
@@ -583,9 +561,6 @@ impl WorkerOperation {
|
||||
Self::Stop,
|
||||
Self::Restore,
|
||||
Self::Remove,
|
||||
Self::Share,
|
||||
Self::Transfer,
|
||||
Self::Revoke,
|
||||
];
|
||||
|
||||
fn tool_name(self) -> &'static str {
|
||||
@@ -598,9 +573,6 @@ impl WorkerOperation {
|
||||
Self::Stop => "WorkerStop",
|
||||
Self::Restore => "WorkerRestore",
|
||||
Self::Remove => "WorkerRemove",
|
||||
Self::Share => "WorkerShare",
|
||||
Self::Transfer => "WorkerTransfer",
|
||||
Self::Revoke => "WorkerRevoke",
|
||||
}
|
||||
}
|
||||
|
||||
@@ -622,11 +594,6 @@ impl WorkerOperation {
|
||||
Self::Remove => {
|
||||
"Remove an eligible stopped, unassigned, non-internal Worker. Supply the current Worker revision and a bounded reason; Backend validation and retention are authoritative."
|
||||
}
|
||||
Self::Share => "Share one controlled Runtime Worker with another known Runtime Worker.",
|
||||
Self::Transfer => {
|
||||
"Transfer one controlled Runtime Worker to another known Runtime Worker."
|
||||
}
|
||||
Self::Revoke => "Revoke one durable Runtime Worker control grant owned by this Worker.",
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -792,59 +759,6 @@ impl Tool for WorkspaceWorkerTool {
|
||||
)
|
||||
.map_err(control_tool_error)?
|
||||
}
|
||||
WorkerOperation::Share | WorkerOperation::Transfer => {
|
||||
let input = parse::<WorkerDelegateInput>(input_json, self.operation.tool_name())?;
|
||||
let grant_id = authority_id(&input.grant_id, "grant_id")?;
|
||||
let (runtime_id, worker_id) =
|
||||
runtime_subject_ids(&input.target_controller, self.operation)?;
|
||||
let operation_id = format!(
|
||||
"worker-control-{}:{}",
|
||||
if self.operation == WorkerOperation::Transfer {
|
||||
"transfer"
|
||||
} else {
|
||||
"share"
|
||||
},
|
||||
non_empty(ctx.call_id.clone(), "tool call_id")?
|
||||
);
|
||||
let action = if self.operation == WorkerOperation::Transfer {
|
||||
"transfer"
|
||||
} else {
|
||||
"share"
|
||||
};
|
||||
self.control
|
||||
.execute_runtime(WorkspaceRequest::json(
|
||||
WorkspaceRequestMethod::Post,
|
||||
format!(
|
||||
"/api/w/{}/worker-control/grants/{grant_id}/{action}",
|
||||
self.control.workspace_id()
|
||||
),
|
||||
serde_json::json!({
|
||||
"target_controller": {
|
||||
"runtime_id": runtime_id,
|
||||
"worker_id": worker_id,
|
||||
},
|
||||
"operation_id": operation_id,
|
||||
})
|
||||
.to_string(),
|
||||
))
|
||||
.await
|
||||
.map_err(control_tool_error)?
|
||||
}
|
||||
WorkerOperation::Revoke => {
|
||||
let input = parse::<WorkerRevokeInput>(input_json, "WorkerRevoke")?;
|
||||
let grant_id = authority_id(&input.grant_id, "grant_id")?;
|
||||
self.control
|
||||
.execute_runtime(WorkspaceRequest::json(
|
||||
WorkspaceRequestMethod::Post,
|
||||
format!(
|
||||
"/api/w/{}/worker-control/grants/{grant_id}/revoke",
|
||||
self.control.workspace_id()
|
||||
),
|
||||
"{}",
|
||||
))
|
||||
.await
|
||||
.map_err(control_tool_error)?
|
||||
}
|
||||
};
|
||||
tool_output(self.operation, response)
|
||||
}
|
||||
@@ -1131,9 +1045,6 @@ mod tests {
|
||||
"WorkerStop",
|
||||
"WorkerRestore",
|
||||
"WorkerRemove",
|
||||
"WorkerShare",
|
||||
"WorkerTransfer",
|
||||
"WorkerRevoke",
|
||||
]
|
||||
);
|
||||
}
|
||||
@@ -1248,45 +1159,6 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn worker_share_and_transfer_use_typed_runtime_subjects_and_operation_ids() {
|
||||
let client = Arc::new(RecordingWorkspaceClient::default());
|
||||
for (operation, action) in [
|
||||
(WorkerOperation::Share, "share"),
|
||||
(WorkerOperation::Transfer, "transfer"),
|
||||
] {
|
||||
WorkspaceWorkerTool {
|
||||
operation,
|
||||
control: test_control(client.clone()),
|
||||
}
|
||||
.execute(
|
||||
&serde_json::json!({
|
||||
"grant_id": "grant-1",
|
||||
"target_controller": {
|
||||
"kind": "runtime_worker",
|
||||
"runtime_id": "runtime-2",
|
||||
"worker_id": "worker-9",
|
||||
},
|
||||
})
|
||||
.to_string(),
|
||||
ToolExecutionContext::new("call-delegate", "batch-delegate", 0),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let request = client.requests.lock().unwrap().last().cloned().unwrap();
|
||||
assert!(request.path.ends_with(&format!("/grant-1/{action}")));
|
||||
let body: serde_json::Value =
|
||||
serde_json::from_str(request.body.as_deref().unwrap()).unwrap();
|
||||
assert_eq!(body["target_controller"]["runtime_id"], "runtime-2");
|
||||
assert!(
|
||||
body["operation_id"]
|
||||
.as_str()
|
||||
.unwrap()
|
||||
.contains("call-delegate")
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn worker_remove_forwards_only_target_revision_and_bounded_reason() {
|
||||
let client = Arc::new(RecordingWorkspaceClient::default());
|
||||
|
||||
@@ -11,7 +11,6 @@ pub const MERGE_REQUEST_COMMON_TOOL_NAMES: &[&str] = &[
|
||||
"MergeRequestShow",
|
||||
"MergeRequestReadinessCheck",
|
||||
"MergeRequestOpen",
|
||||
"MergeRequestAddRevision",
|
||||
"MergeRequestComplete",
|
||||
];
|
||||
pub const MERGE_REQUEST_REVIEW_TOOL_NAME: &str = "MergeRequestReviewSubmit";
|
||||
@@ -20,7 +19,6 @@ enum Kind {
|
||||
Show,
|
||||
Readiness,
|
||||
Open,
|
||||
AddRevision,
|
||||
Complete,
|
||||
Review,
|
||||
}
|
||||
@@ -29,7 +27,6 @@ struct MergeRequestTool {
|
||||
client: Arc<dyn WorkspaceClient>,
|
||||
kind: Kind,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize, JsonSchema)]
|
||||
struct ShowInput {
|
||||
ticket: String,
|
||||
@@ -38,25 +35,8 @@ struct ShowInput {
|
||||
struct OpenInput {
|
||||
ticket: String,
|
||||
repository_id: String,
|
||||
revision_id: String,
|
||||
base_commit: String,
|
||||
head_commit: String,
|
||||
diff_digest: String,
|
||||
#[serde(default)]
|
||||
changed_paths: Vec<String>,
|
||||
#[serde(default)]
|
||||
summary: String,
|
||||
}
|
||||
#[derive(Debug, Deserialize, JsonSchema)]
|
||||
struct AddRevisionInput {
|
||||
ticket: String,
|
||||
expected_current_revision_id: String,
|
||||
revision_id: String,
|
||||
base_commit: String,
|
||||
head_commit: String,
|
||||
diff_digest: String,
|
||||
#[serde(default)]
|
||||
changed_paths: Vec<String>,
|
||||
selector_from: String,
|
||||
selector_to: String,
|
||||
#[serde(default)]
|
||||
summary: String,
|
||||
}
|
||||
@@ -64,10 +44,9 @@ struct AddRevisionInput {
|
||||
struct CompleteInput {
|
||||
ticket: String,
|
||||
operation_id: String,
|
||||
expected_revision_id: String,
|
||||
target_commit: String,
|
||||
source_commit: String,
|
||||
result_commit: String,
|
||||
approval_event_id: String,
|
||||
target_ref_before: String,
|
||||
target_ref_after: String,
|
||||
strategy: MergeStrategyInput,
|
||||
resolution: MergeResolutionInput,
|
||||
}
|
||||
@@ -106,63 +85,48 @@ struct ReviewFindingInput {
|
||||
#[serde(default)]
|
||||
path: Option<String>,
|
||||
#[serde(default)]
|
||||
line: Option<u64>,
|
||||
line: Option<u32>,
|
||||
body: String,
|
||||
}
|
||||
|
||||
impl Kind {
|
||||
fn name(self) -> &'static str {
|
||||
match self {
|
||||
Self::Show => "MergeRequestShow",
|
||||
Self::Readiness => "MergeRequestReadinessCheck",
|
||||
Self::Open => "MergeRequestOpen",
|
||||
Self::AddRevision => "MergeRequestAddRevision",
|
||||
Self::Complete => "MergeRequestComplete",
|
||||
Self::Review => "MergeRequestReviewSubmit",
|
||||
}
|
||||
}
|
||||
fn description(self) -> &'static str {
|
||||
description(self.name()).unwrap_or("Merge Request operation.")
|
||||
}
|
||||
fn schema(self) -> serde_json::Value {
|
||||
match self {
|
||||
Self::Show | Self::Readiness => json!(schemars::schema_for!(ShowInput)),
|
||||
Self::Open => json!(schemars::schema_for!(OpenInput)),
|
||||
Self::AddRevision => json!(schemars::schema_for!(AddRevisionInput)),
|
||||
Self::Complete => json!(schemars::schema_for!(CompleteInput)),
|
||||
Self::Review => json!(schemars::schema_for!(ReviewInput)),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Tool for MergeRequestTool {
|
||||
async fn execute(
|
||||
&self,
|
||||
input: &str,
|
||||
_context: ToolExecutionContext,
|
||||
) -> Result<ToolOutput, ToolError> {
|
||||
let workspace_id = self.client.workspace_id().ok_or_else(|| {
|
||||
async fn execute(&self, input: &str, _: ToolExecutionContext) -> Result<ToolOutput, ToolError> {
|
||||
let ws = self.client.workspace_id().ok_or_else(|| {
|
||||
ToolError::ExecutionFailed("Merge Request tools require Workspace identity".into())
|
||||
})?;
|
||||
let (method, path, body) = match self.kind {
|
||||
Kind::Show => {
|
||||
let v: ShowInput = parse(input)?;
|
||||
nonempty(&v.ticket)?;
|
||||
(
|
||||
WorkspaceRequestMethod::Get,
|
||||
format!("/api/w/{workspace_id}/tickets/{}/merge-request", v.ticket),
|
||||
None,
|
||||
)
|
||||
}
|
||||
Kind::Readiness => {
|
||||
Kind::Show | Kind::Readiness => {
|
||||
let v: ShowInput = parse(input)?;
|
||||
nonempty(&v.ticket)?;
|
||||
(
|
||||
WorkspaceRequestMethod::Get,
|
||||
format!(
|
||||
"/api/w/{workspace_id}/tickets/{}/merge-request/readiness",
|
||||
v.ticket
|
||||
"/api/w/{ws}/tickets/{}/merge-request{}",
|
||||
v.ticket,
|
||||
if matches!(self.kind, Kind::Readiness) {
|
||||
"/readiness"
|
||||
} else {
|
||||
""
|
||||
}
|
||||
),
|
||||
None,
|
||||
)
|
||||
@@ -172,62 +136,35 @@ impl Tool for MergeRequestTool {
|
||||
nonempty(&v.ticket)?;
|
||||
(
|
||||
WorkspaceRequestMethod::Post,
|
||||
format!("/api/w/{workspace_id}/tickets/{}/merge-request", v.ticket),
|
||||
format!("/api/w/{ws}/tickets/{}/merge-request", v.ticket),
|
||||
Some(
|
||||
json!({"repository_id":v.repository_id,"revision_id":v.revision_id,"base_commit":v.base_commit,"head_commit":v.head_commit,"diff_digest":v.diff_digest,"changed_paths":v.changed_paths,"summary":v.summary}),
|
||||
),
|
||||
)
|
||||
}
|
||||
Kind::AddRevision => {
|
||||
let v: AddRevisionInput = parse(input)?;
|
||||
nonempty(&v.ticket)?;
|
||||
(
|
||||
WorkspaceRequestMethod::Post,
|
||||
format!(
|
||||
"/api/w/{workspace_id}/tickets/{}/merge-request/revisions",
|
||||
v.ticket
|
||||
),
|
||||
Some(
|
||||
json!({"expected_current_revision_id":v.expected_current_revision_id,"revision_id":v.revision_id,"base_commit":v.base_commit,"head_commit":v.head_commit,"diff_digest":v.diff_digest,"changed_paths":v.changed_paths,"summary":v.summary}),
|
||||
json!({"repository_id":v.repository_id,"selector_from":v.selector_from,"selector_to":v.selector_to,"summary":v.summary}),
|
||||
),
|
||||
)
|
||||
}
|
||||
Kind::Complete => {
|
||||
let v: CompleteInput = parse(input)?;
|
||||
nonempty(&v.ticket)?;
|
||||
let strategy = match v.strategy {
|
||||
MergeStrategyInput::FastForward => "fast_forward",
|
||||
MergeStrategyInput::Merge => "merge",
|
||||
};
|
||||
let resolution = match v.resolution {
|
||||
MergeResolutionInput::None => "none",
|
||||
MergeResolutionInput::Clean => "clean",
|
||||
MergeResolutionInput::ConflictsResolved => "conflicts_resolved",
|
||||
};
|
||||
(
|
||||
WorkspaceRequestMethod::Post,
|
||||
format!(
|
||||
"/api/w/{workspace_id}/tickets/{}/merge-request/complete",
|
||||
v.ticket
|
||||
),
|
||||
format!("/api/w/{ws}/tickets/{}/merge-request/complete", v.ticket),
|
||||
Some(
|
||||
json!({"operation_id":v.operation_id,"expected_revision_id":v.expected_revision_id,"target_commit":v.target_commit,"source_commit":v.source_commit,"result_commit":v.result_commit,"strategy":strategy,"resolution":resolution}),
|
||||
json!({"operation_id":v.operation_id,"approval_event_id":v.approval_event_id,"target_ref_before":v.target_ref_before,"target_ref_after":v.target_ref_after,"strategy":match v.strategy{MergeStrategyInput::FastForward=>"fast_forward",MergeStrategyInput::Merge=>"merge"},"resolution":match v.resolution{MergeResolutionInput::None=>"none",MergeResolutionInput::Clean=>"clean",MergeResolutionInput::ConflictsResolved=>"conflicts_resolved"}}),
|
||||
),
|
||||
)
|
||||
}
|
||||
Kind::Review => {
|
||||
let v: ReviewInput = parse(input)?;
|
||||
let context = self.client.reviewer_attempt_context().ok_or_else(|| {
|
||||
let ctx = self.client.reviewer_context().ok_or_else(|| {
|
||||
ToolError::ExecutionFailed(
|
||||
"MergeRequestReviewSubmit is available only to an attested Reviewer child"
|
||||
.into(),
|
||||
"Review submit requires injected Reviewer capability".into(),
|
||||
)
|
||||
})?;
|
||||
(
|
||||
WorkspaceRequestMethod::Post,
|
||||
format!(
|
||||
"/api/w/{workspace_id}/tickets/{}/merge-request/reviews",
|
||||
context.ticket_id
|
||||
"/api/w/{ws}/tickets/{}/merge-request/reviews",
|
||||
ctx.ticket_id
|
||||
),
|
||||
Some(
|
||||
json!({"decision":match v.decision{ReviewDecisionInput::Approve=>"approve",ReviewDecisionInput::RequestChanges=>"request_changes"},"body":v.body,"findings":v.findings.into_iter().map(|f|json!({"severity":f.severity,"code":f.code,"path":f.path,"line":f.line,"body":f.body})).collect::<Vec<_>>() }),
|
||||
@@ -235,32 +172,32 @@ impl Tool for MergeRequestTool {
|
||||
)
|
||||
}
|
||||
};
|
||||
let request = match body {
|
||||
Some(body) => WorkspaceRequest::json(method, path, body.to_string()),
|
||||
let req = match body {
|
||||
Some(v) => WorkspaceRequest::json(method, path, v.to_string()),
|
||||
None => WorkspaceRequest::get(path),
|
||||
};
|
||||
let response = self
|
||||
let res = self
|
||||
.client
|
||||
.execute(request)
|
||||
.execute(req)
|
||||
.map_err(|e| ToolError::ExecutionFailed(e.to_string()))?;
|
||||
if !response.is_success() {
|
||||
if !res.is_success() {
|
||||
return Err(ToolError::ExecutionFailed(format!(
|
||||
"Merge Request API returned HTTP {}: {}",
|
||||
response.status, response.body
|
||||
res.status, res.body
|
||||
)));
|
||||
}
|
||||
Ok(ToolOutput {
|
||||
summary: self.kind.name().to_string(),
|
||||
content: Some(response.body),
|
||||
attachments: Vec::new(),
|
||||
summary: self.kind.name().into(),
|
||||
content: Some(res.body),
|
||||
attachments: vec![],
|
||||
})
|
||||
}
|
||||
}
|
||||
fn parse<T: serde::de::DeserializeOwned>(value: &str) -> Result<T, ToolError> {
|
||||
serde_json::from_str(value).map_err(|e| ToolError::InvalidArgument(e.to_string()))
|
||||
fn parse<T: serde::de::DeserializeOwned>(v: &str) -> Result<T, ToolError> {
|
||||
serde_json::from_str(v).map_err(|e| ToolError::InvalidArgument(e.to_string()))
|
||||
}
|
||||
fn nonempty(value: &str) -> Result<(), ToolError> {
|
||||
if value.trim().is_empty() {
|
||||
fn nonempty(v: &str) -> Result<(), ToolError> {
|
||||
if v.trim().is_empty() {
|
||||
Err(ToolError::InvalidArgument(
|
||||
"ticket must not be empty".into(),
|
||||
))
|
||||
@@ -270,72 +207,75 @@ fn nonempty(value: &str) -> Result<(), ToolError> {
|
||||
}
|
||||
fn definition(client: Arc<dyn WorkspaceClient>, kind: Kind) -> ToolDefinition {
|
||||
Arc::new(move || {
|
||||
let meta = ToolMeta::new(kind.name())
|
||||
.description(kind.description())
|
||||
.input_schema(kind.schema());
|
||||
let tool: Arc<dyn Tool> = Arc::new(MergeRequestTool {
|
||||
client: client.clone(),
|
||||
kind,
|
||||
});
|
||||
(meta, tool)
|
||||
(
|
||||
ToolMeta::new(kind.name())
|
||||
.description(description(kind.name()).unwrap_or("Merge Request operation."))
|
||||
.input_schema(kind.schema()),
|
||||
Arc::new(MergeRequestTool {
|
||||
client: client.clone(),
|
||||
kind,
|
||||
}) as Arc<dyn Tool>,
|
||||
)
|
||||
})
|
||||
}
|
||||
pub fn common_tools(client: Arc<dyn WorkspaceClient>) -> Vec<ToolDefinition> {
|
||||
pub fn common_tools(c: Arc<dyn WorkspaceClient>) -> Vec<ToolDefinition> {
|
||||
vec![
|
||||
definition(client.clone(), Kind::Show),
|
||||
definition(client.clone(), Kind::Readiness),
|
||||
definition(client.clone(), Kind::Open),
|
||||
definition(client.clone(), Kind::AddRevision),
|
||||
definition(client, Kind::Complete),
|
||||
definition(c.clone(), Kind::Show),
|
||||
definition(c.clone(), Kind::Readiness),
|
||||
definition(c.clone(), Kind::Open),
|
||||
definition(c, Kind::Complete),
|
||||
]
|
||||
}
|
||||
pub fn reviewer_tools(client: Arc<dyn WorkspaceClient>) -> Vec<ToolDefinition> {
|
||||
if client.reviewer_attempt_context().is_some() {
|
||||
pub fn reviewer_tools(c: Arc<dyn WorkspaceClient>) -> Vec<ToolDefinition> {
|
||||
if c.reviewer_context().is_some() {
|
||||
vec![
|
||||
definition(client.clone(), Kind::Show),
|
||||
definition(client, Kind::Review),
|
||||
definition(c.clone(), Kind::Show),
|
||||
definition(c, Kind::Review),
|
||||
]
|
||||
} else {
|
||||
Vec::new()
|
||||
vec![]
|
||||
}
|
||||
}
|
||||
pub fn description(name: &str) -> Option<&'static str> {
|
||||
match name {
|
||||
"MergeRequestShow" => Some(
|
||||
"Read the authoritative Merge Request, immutable current revision, and structured review status.",
|
||||
),
|
||||
pub fn description(n: &str) -> Option<&'static str> {
|
||||
match n {
|
||||
"MergeRequestShow" => Some("Read the selector-based Merge Request and append-only thread."),
|
||||
"MergeRequestReadinessCheck" => {
|
||||
Some("Check derived merge readiness for the current immutable revision.")
|
||||
Some("Resolve current provider refs and derive readiness from valid review events.")
|
||||
}
|
||||
"MergeRequestOpen" => {
|
||||
Some("Open an immutable Merge Request revision for the current assigned Coder.")
|
||||
}
|
||||
"MergeRequestAddRevision" => {
|
||||
Some("Append an immutable revision; prior approval cannot carry to the new revision.")
|
||||
Some("Open a Merge Request with immutable source and target selectors.")
|
||||
}
|
||||
"MergeRequestComplete" => {
|
||||
Some("CAS-complete an approved revision with operation-id replay and crash fencing.")
|
||||
Some("Complete using an approved review event and final target-ref evidence.")
|
||||
}
|
||||
"MergeRequestReviewSubmit" => {
|
||||
Some("Submit the injected Reviewer capability result for its captured subject ref.")
|
||||
}
|
||||
"MergeRequestReviewSubmit" => Some(
|
||||
"Submit the attested direct-child Reviewer result bound to its immutable revision.",
|
||||
),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn merge_request_tool_contract_omits_tree_hashes_and_candidate_result_tool() {
|
||||
let open = serde_json::to_string(&schemars::schema_for!(OpenInput)).unwrap();
|
||||
let add = serde_json::to_string(&schemars::schema_for!(AddRevisionInput)).unwrap();
|
||||
let complete = serde_json::to_string(&schemars::schema_for!(CompleteInput)).unwrap();
|
||||
assert!(!open.contains("head_tree"));
|
||||
assert!(!add.contains("head_tree"));
|
||||
assert!(complete.contains("result_commit"));
|
||||
assert!(complete.contains("conflicts_resolved"));
|
||||
assert!(!MERGE_REQUEST_COMMON_TOOL_NAMES.contains(&"MergeRequestRecordMergeResult"));
|
||||
fn schemas_hide_revision_and_commit_authority() {
|
||||
let schemas = [
|
||||
schemars::schema_for!(OpenInput),
|
||||
schemars::schema_for!(CompleteInput),
|
||||
];
|
||||
for s in schemas {
|
||||
let j = serde_json::to_string(&s).unwrap();
|
||||
for banned in [
|
||||
"revision_id",
|
||||
"attempt_id",
|
||||
"base_commit",
|
||||
"head_commit",
|
||||
"source_commit",
|
||||
"result_commit",
|
||||
] {
|
||||
assert!(!j.contains(banned), "{banned} in {j}")
|
||||
}
|
||||
}
|
||||
assert!(!MERGE_REQUEST_COMMON_TOOL_NAMES.contains(&"MergeRequestRequestReview"));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -589,7 +589,7 @@ impl FeatureModule for TicketFeature {
|
||||
));
|
||||
}
|
||||
if let TicketFeatureBackend::WorkspaceClient(client) = &self.backend {
|
||||
let names: Vec<&str> = if client.reviewer_attempt_context().is_some() {
|
||||
let names: Vec<&str> = if client.reviewer_context().is_some() {
|
||||
vec![
|
||||
"MergeRequestShow",
|
||||
merge_request::MERGE_REQUEST_REVIEW_TOOL_NAME,
|
||||
@@ -662,7 +662,7 @@ impl FeatureModule for TicketFeature {
|
||||
tools.register(ToolContribution::new(name, definition))?;
|
||||
}
|
||||
if let TicketFeatureBackend::WorkspaceClient(client) = &self.backend {
|
||||
let definitions = if client.reviewer_attempt_context().is_some() {
|
||||
let definitions = if client.reviewer_context().is_some() {
|
||||
merge_request::reviewer_tools(client.clone())
|
||||
} else {
|
||||
merge_request::common_tools(client.clone())
|
||||
|
||||
@@ -527,6 +527,31 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn commit_capable_roles_classify_commits_by_change_type() {
|
||||
let catalog = PromptCatalog::builtins_only().unwrap();
|
||||
assert!(catalog.projection.templates.contains_key("common.git"));
|
||||
let context = Value::from_serialize(serde_json::json!({
|
||||
"cwd": "/workspace",
|
||||
"date": "2026-08-16",
|
||||
"language": "match the user's language",
|
||||
"tool_capabilities": {
|
||||
"memory_any": false,
|
||||
"memory_mutation": false,
|
||||
"memory_query": false,
|
||||
"memory_read_document": false
|
||||
}
|
||||
}));
|
||||
|
||||
for prompt in ["default", "role.coder", "role.orchestrator"] {
|
||||
let rendered = catalog.render_name(prompt, context.clone()).unwrap();
|
||||
assert!(rendered.contains("use the change type as the subject prefix"));
|
||||
assert!(rendered.contains("A change made because review"));
|
||||
assert!(rendered.contains("Do not keep reusing a domain prefix"));
|
||||
assert!(rendered.contains("fix: scope merge request foreign key checks"));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn builtin_render_resolves_catalog_root_dotted_includes() {
|
||||
let catalog = PromptCatalog::builtins_only().unwrap();
|
||||
|
||||
@@ -28,7 +28,7 @@ use crate::internal_worker::{
|
||||
use crate::prompt::catalog::PromptCatalog;
|
||||
use crate::spawn::registry::SpawnedWorkerRegistry;
|
||||
use crate::worker::{
|
||||
ReviewerAttemptContext, ReviewerChildWorkspaceClient, Worker, WorkerFilesystemAuthority,
|
||||
ReviewerChildWorkspaceClient, ReviewerContext, Worker, WorkerFilesystemAuthority,
|
||||
WorkspaceRequest, WorkspaceRequestMethod,
|
||||
};
|
||||
use protocol::Method;
|
||||
@@ -58,8 +58,8 @@ struct SubWorkerSpawnInput {
|
||||
/// spawner's explicit delegation authority; direct tool scope alone is not
|
||||
/// sufficient. Omit `recursive` for normal workspace/worktree delegation; it defaults to true.
|
||||
scope: Vec<ScopeRuleInput>,
|
||||
/// Binds an actual read-only builtin Reviewer child to an immutable Merge Request revision.
|
||||
/// Review attempt identity and capability material are generated by the trusted spawn layer.
|
||||
/// Binds an actual read-only builtin Reviewer child to the current Merge Request candidate.
|
||||
/// Review capability material is generated by the trusted spawn layer.
|
||||
#[serde(default)]
|
||||
review: Option<ReviewerHandoffInput>,
|
||||
}
|
||||
@@ -67,7 +67,6 @@ struct SubWorkerSpawnInput {
|
||||
#[derive(Debug, Deserialize, schemars::JsonSchema)]
|
||||
struct ReviewerHandoffInput {
|
||||
ticket_id: String,
|
||||
revision_id: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize, schemars::JsonSchema)]
|
||||
@@ -337,9 +336,9 @@ fn validate_reviewer_handoff(input: &SubWorkerSpawnInput) -> Result<(), ToolErro
|
||||
let Some(review) = &input.review else {
|
||||
return Ok(());
|
||||
};
|
||||
if review.ticket_id.trim().is_empty() || review.revision_id.trim().is_empty() {
|
||||
if review.ticket_id.trim().is_empty() {
|
||||
return Err(ToolError::InvalidArgument(
|
||||
"reviewer handoff requires non-empty ticket_id and revision_id".to_string(),
|
||||
"reviewer handoff requires non-empty ticket_id".to_string(),
|
||||
));
|
||||
}
|
||||
if input.profile.as_deref() != Some("builtin:reviewer") {
|
||||
@@ -418,11 +417,9 @@ impl Tool for SubWorkerSpawnTool {
|
||||
.map_err(|error| {
|
||||
ToolError::ExecutionFailed(format!("resolve child manifest: {error}"))
|
||||
})?;
|
||||
let reviewer_attempt = input.review.as_ref().map(|review| {
|
||||
let reviewer_capability = input.review.as_ref().map(|review| {
|
||||
(
|
||||
review.ticket_id.clone(),
|
||||
review.revision_id.clone(),
|
||||
uuid::Uuid::now_v7().to_string(),
|
||||
format!(
|
||||
"{}{}",
|
||||
uuid::Uuid::now_v7().simple(),
|
||||
@@ -431,7 +428,7 @@ impl Tool for SubWorkerSpawnTool {
|
||||
)
|
||||
});
|
||||
let child_workspace_context =
|
||||
if let Some((ticket_id, revision_id, _, capability_token)) = &reviewer_attempt {
|
||||
if let Some((ticket_id, capability_token)) = &reviewer_capability {
|
||||
let workspace_id =
|
||||
self.workspace_context
|
||||
.workspace_id()
|
||||
@@ -450,9 +447,8 @@ impl Tool for SubWorkerSpawnTool {
|
||||
let child_client: Arc<dyn crate::worker::WorkspaceClient> =
|
||||
Arc::new(ReviewerChildWorkspaceClient::new(
|
||||
parent_client.clone(),
|
||||
ReviewerAttemptContext {
|
||||
ReviewerContext {
|
||||
ticket_id: ticket_id.clone(),
|
||||
revision_id: revision_id.clone(),
|
||||
},
|
||||
capability_token.clone(),
|
||||
));
|
||||
@@ -547,9 +543,9 @@ impl Tool for SubWorkerSpawnTool {
|
||||
}
|
||||
};
|
||||
|
||||
if let Some((ticket_id, revision_id, attempt_id, capability_token)) = &reviewer_attempt {
|
||||
if let Some((ticket_id, capability_token)) = &reviewer_capability {
|
||||
let workspace_id = self.workspace_context.workspace_id().ok_or_else(|| {
|
||||
ToolError::ExecutionFailed("reviewer attempt lost Workspace identity".to_string())
|
||||
ToolError::ExecutionFailed("review capability lost Workspace identity".to_string())
|
||||
})?;
|
||||
let child_session_id = session.session_id_string();
|
||||
let child_registration = WorkspaceRequest::json(
|
||||
@@ -577,15 +573,13 @@ impl Tool for SubWorkerSpawnTool {
|
||||
)));
|
||||
}
|
||||
let body = serde_json::json!({
|
||||
"attempt_id": attempt_id,
|
||||
"revision_id": revision_id,
|
||||
"child_session_id": child_session_id,
|
||||
"capability_token": capability_token,
|
||||
});
|
||||
let request = WorkspaceRequest::json(
|
||||
WorkspaceRequestMethod::Post,
|
||||
format!(
|
||||
"/api/w/{}/tickets/{}/merge-request/review-attempts",
|
||||
"/api/w/{}/tickets/{}/merge-request/review-capabilities",
|
||||
workspace_id.as_str(),
|
||||
ticket_id
|
||||
),
|
||||
@@ -596,12 +590,12 @@ impl Tool for SubWorkerSpawnTool {
|
||||
.client()
|
||||
.execute(request)
|
||||
.map_err(|error| {
|
||||
ToolError::ExecutionFailed(format!("register reviewer attempt: {error}"))
|
||||
ToolError::ExecutionFailed(format!("register review capability: {error}"))
|
||||
})?;
|
||||
if !response.is_success() {
|
||||
let _ = session.stop().await;
|
||||
return Err(ToolError::ExecutionFailed(format!(
|
||||
"register reviewer attempt failed with status {}: {}",
|
||||
"register review capability failed with status {}: {}",
|
||||
response.status, response.body
|
||||
)));
|
||||
}
|
||||
@@ -1044,21 +1038,21 @@ mod tests {
|
||||
let valid: SubWorkerSpawnInput = serde_json::from_value(serde_json::json!({
|
||||
"name":"reviewer","task":"review","profile":"builtin:reviewer",
|
||||
"scope":[{"target":"/tmp/work","permission":"read"}],
|
||||
"review":{"ticket_id":"T1","revision_id":"V1"}
|
||||
"review":{"ticket_id":"T1"}
|
||||
}))
|
||||
.unwrap();
|
||||
assert!(validate_reviewer_handoff(&valid).is_ok());
|
||||
let wrong_profile: SubWorkerSpawnInput = serde_json::from_value(serde_json::json!({
|
||||
"name":"reviewer","task":"review","profile":"builtin:coder",
|
||||
"scope":[{"target":"/tmp/work","permission":"read"}],
|
||||
"review":{"ticket_id":"T1","revision_id":"V1"}
|
||||
"review":{"ticket_id":"T1"}
|
||||
}))
|
||||
.unwrap();
|
||||
assert!(validate_reviewer_handoff(&wrong_profile).is_err());
|
||||
let writable: SubWorkerSpawnInput = serde_json::from_value(serde_json::json!({
|
||||
"name":"reviewer","task":"review","profile":"builtin:reviewer",
|
||||
"scope":[{"target":"/tmp/work","permission":"write"}],
|
||||
"review":{"ticket_id":"T1","revision_id":"V1"}
|
||||
"review":{"ticket_id":"T1"}
|
||||
}))
|
||||
.unwrap();
|
||||
assert!(validate_reviewer_handoff(&writable).is_err());
|
||||
|
||||
@@ -238,30 +238,29 @@ pub trait WorkspaceClient: std::fmt::Debug + Send + Sync {
|
||||
))
|
||||
}
|
||||
|
||||
/// Trusted review-attempt context is injected by the Internal SubWorker spawn layer.
|
||||
/// Trusted review capability context is injected by the Internal SubWorker spawn layer.
|
||||
/// It is never accepted from a model-visible tool argument.
|
||||
fn reviewer_attempt_context(&self) -> Option<&ReviewerAttemptContext> {
|
||||
fn reviewer_context(&self) -> Option<&ReviewerContext> {
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct ReviewerAttemptContext {
|
||||
pub struct ReviewerContext {
|
||||
pub ticket_id: String,
|
||||
pub revision_id: String,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct ReviewerChildWorkspaceClient {
|
||||
inner: Arc<dyn WorkspaceClient>,
|
||||
context: ReviewerAttemptContext,
|
||||
context: ReviewerContext,
|
||||
capability_token: String,
|
||||
}
|
||||
|
||||
impl ReviewerChildWorkspaceClient {
|
||||
pub fn new(
|
||||
inner: Arc<dyn WorkspaceClient>,
|
||||
context: ReviewerAttemptContext,
|
||||
context: ReviewerContext,
|
||||
capability_token: String,
|
||||
) -> Self {
|
||||
Self {
|
||||
@@ -282,7 +281,7 @@ impl WorkspaceClient for ReviewerChildWorkspaceClient {
|
||||
fn is_available(&self) -> bool {
|
||||
self.inner.is_available()
|
||||
}
|
||||
fn reviewer_attempt_context(&self) -> Option<&ReviewerAttemptContext> {
|
||||
fn reviewer_context(&self) -> Option<&ReviewerContext> {
|
||||
Some(&self.context)
|
||||
}
|
||||
|
||||
@@ -306,10 +305,6 @@ impl WorkspaceClient for ReviewerChildWorkspaceClient {
|
||||
"review submission body must be an object".to_string(),
|
||||
)
|
||||
})?;
|
||||
object.insert(
|
||||
"revision_id".to_string(),
|
||||
serde_json::Value::String(self.context.revision_id.clone()),
|
||||
);
|
||||
object.insert(
|
||||
"capability_token".to_string(),
|
||||
serde_json::Value::String(self.capability_token.clone()),
|
||||
@@ -450,9 +445,8 @@ mod reviewer_client_tests {
|
||||
});
|
||||
let client = ReviewerChildWorkspaceClient::new(
|
||||
inner,
|
||||
ReviewerAttemptContext {
|
||||
ReviewerContext {
|
||||
ticket_id: "T1".into(),
|
||||
revision_id: "V1".into(),
|
||||
},
|
||||
"secret".into(),
|
||||
);
|
||||
|
||||
@@ -1,7 +1,10 @@
|
||||
use std::path::PathBuf;
|
||||
use std::{path::PathBuf, sync::Arc};
|
||||
|
||||
use chrono::Utc;
|
||||
use merge_request::{MergeRequest, ReviewStatus, SqliteMergeRequestStore};
|
||||
use merge_request::{
|
||||
MergeRequest, MergeRequestError, MergeRequestState, MergeRequestStore, MergeRequestThreadEvent,
|
||||
ReviewDecision,
|
||||
};
|
||||
use project_record::{allocate_record_id, unix_epoch_millis_now};
|
||||
|
||||
use ticket::{
|
||||
@@ -126,27 +129,75 @@ pub struct MemoryStagingResolution {
|
||||
pub record_source: String,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct AuthorityMergeRequestSource {
|
||||
store: SqliteWorkspaceStore,
|
||||
}
|
||||
|
||||
impl merge_request::AssignmentSource for AuthorityMergeRequestSource {
|
||||
fn current_assignment(
|
||||
&self,
|
||||
workspace_id: &str,
|
||||
ticket_id: &str,
|
||||
) -> std::result::Result<Option<merge_request::CurrentAssignment>, String> {
|
||||
self.store
|
||||
.get_current_ticket_worker_assignment(workspace_id, ticket_id)
|
||||
.map(|assignment| {
|
||||
assignment.map(|assignment| merge_request::CurrentAssignment {
|
||||
assignment_id: assignment.assignment_id,
|
||||
ticket_id: ticket_id.to_string(),
|
||||
runtime_id: assignment.worker.runtime_id,
|
||||
worker_id: assignment.worker.worker_id,
|
||||
})
|
||||
})
|
||||
.map_err(|error| error.to_string())
|
||||
}
|
||||
}
|
||||
|
||||
impl merge_request::RepositorySource for AuthorityMergeRequestSource {
|
||||
fn repository_belongs_to_workspace(
|
||||
&self,
|
||||
workspace_id: &str,
|
||||
repository_id: &str,
|
||||
) -> std::result::Result<bool, String> {
|
||||
self.store
|
||||
.get_repository(workspace_id, repository_id)
|
||||
.map(|repository| repository.is_some())
|
||||
.map_err(|error| error.to_string())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct SqliteWorkspaceAuthority {
|
||||
workspace_id: String,
|
||||
store: SqliteWorkspaceStore,
|
||||
ticket_backend: SqliteTicketBackend,
|
||||
merge_request_store: SqliteMergeRequestStore,
|
||||
merge_request_store: Arc<MergeRequestStore>,
|
||||
}
|
||||
|
||||
impl SqliteWorkspaceAuthority {
|
||||
pub fn new(database_path: impl Into<PathBuf>, workspace_id: impl Into<String>) -> Result<Self> {
|
||||
let database_path = database_path.into();
|
||||
let workspace_id = workspace_id.into();
|
||||
let store = SqliteWorkspaceStore::open(&database_path)?;
|
||||
let merge_request_source = Arc::new(AuthorityMergeRequestSource {
|
||||
store: store.clone(),
|
||||
});
|
||||
Ok(Self {
|
||||
workspace_id: workspace_id.clone(),
|
||||
store: SqliteWorkspaceStore::open(&database_path)?,
|
||||
store,
|
||||
ticket_backend: SqliteTicketBackend::open_verified(
|
||||
database_path.clone(),
|
||||
workspace_id.clone(),
|
||||
workspace_id,
|
||||
)?,
|
||||
merge_request_store: SqliteMergeRequestStore::open(database_path, workspace_id)
|
||||
merge_request_store: Arc::new(
|
||||
MergeRequestStore::open(
|
||||
database_path,
|
||||
merge_request_source.clone(),
|
||||
merge_request_source,
|
||||
)
|
||||
.map_err(|error| Error::Store(error.to_string()))?,
|
||||
),
|
||||
})
|
||||
}
|
||||
|
||||
@@ -315,11 +366,11 @@ impl SqliteWorkspaceAuthority {
|
||||
runtime_id: assignment.worker.runtime_id,
|
||||
worker_id: assignment.worker.worker_id,
|
||||
});
|
||||
let merge_request = self
|
||||
.merge_request_store
|
||||
.show_for_ticket(id)
|
||||
.map_err(|error| Error::Store(error.to_string()))?
|
||||
.map(merge_request_summary);
|
||||
let merge_request = match self.merge_request_store.get(&self.workspace_id, id) {
|
||||
Ok(request) => Some(merge_request_summary(request)),
|
||||
Err(MergeRequestError::NotFound) => None,
|
||||
Err(error) => return Err(Error::Store(error.to_string())),
|
||||
};
|
||||
let evidence = ticket_evidence_summary(&ticket.events, merge_request.as_ref());
|
||||
let item_revision = ticket
|
||||
.events
|
||||
@@ -973,31 +1024,35 @@ fn ticket_evidence_event(sequence: usize, event: &TicketEvent) -> TicketEvidence
|
||||
}
|
||||
|
||||
fn merge_request_summary(request: MergeRequest) -> TicketMergeRequestSummary {
|
||||
let review_status = match request.review_status {
|
||||
ReviewStatus::Pending => "pending",
|
||||
ReviewStatus::Approved => "approved",
|
||||
ReviewStatus::ChangesRequested => "changes_requested",
|
||||
};
|
||||
let review_subject_ref = request.thread.iter().rev().find_map(|event| match event {
|
||||
MergeRequestThreadEvent::ReviewRequested(review) => Some(review.subject_ref.as_str()),
|
||||
_ => None,
|
||||
});
|
||||
let current_review =
|
||||
review_subject_ref.and_then(|subject_ref| request.effective_review(subject_ref));
|
||||
let review_status = match current_review.map(|review| &review.decision) {
|
||||
Some(ReviewDecision::Approve) => "approved",
|
||||
Some(ReviewDecision::RequestChanges) => "changes_requested",
|
||||
None => "pending",
|
||||
}
|
||||
.to_string();
|
||||
let state = match request.state {
|
||||
MergeRequestState::Open => "open",
|
||||
MergeRequestState::Merged => "merged",
|
||||
MergeRequestState::Closed => "closed",
|
||||
}
|
||||
.to_string();
|
||||
|
||||
TicketMergeRequestSummary {
|
||||
merge_request_id: request.merge_request_id,
|
||||
state: serde_json::to_value(request.state)
|
||||
.ok()
|
||||
.and_then(|value| value.as_str().map(str::to_string))
|
||||
.unwrap_or_else(|| "open".to_string()),
|
||||
review_status: review_status.to_string(),
|
||||
revision_id: request.current_revision.revision_id,
|
||||
base_commit: request.current_revision.base_commit,
|
||||
head_commit: request.current_revision.head_commit,
|
||||
changed_paths: request.current_revision.changed_paths,
|
||||
updated_at: request.updated_at,
|
||||
review_submitted_at: request
|
||||
.current_review
|
||||
.as_ref()
|
||||
.map(|review| review.submitted_at.clone()),
|
||||
review_excerpt: request
|
||||
.current_review
|
||||
.as_ref()
|
||||
.map(|review| truncate_body(&review.body, 512).0),
|
||||
merge_request_id: request.merge_request_id.clone(),
|
||||
state,
|
||||
review_status,
|
||||
selector_from: request.selector_from.clone(),
|
||||
selector_to: request.selector_to.clone(),
|
||||
updated_at: request.updated_at.to_rfc3339(),
|
||||
review_subject_ref: review_subject_ref.map(str::to_string),
|
||||
review_submitted_at: current_review.map(|review| review.created_at.to_rfc3339()),
|
||||
review_excerpt: current_review.map(|review| truncate_body(&review.body, 240).0),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1019,10 +1074,14 @@ fn ticket_evidence_summary(
|
||||
.map(|(sequence, _)| sequence);
|
||||
let report_after_rescope =
|
||||
latest_report.is_some_and(|report| latest_rescope.is_none_or(|rescope| report > rescope));
|
||||
let has_commit = merge_request.is_some_and(|request| !request.head_commit.is_empty())
|
||||
|| events.iter().any(|event| {
|
||||
event.attributes.contains_key("commit") || event.attributes.contains_key("head_commit")
|
||||
});
|
||||
let has_commit = merge_request.is_some_and(|request| {
|
||||
request
|
||||
.review_subject_ref
|
||||
.as_ref()
|
||||
.is_some_and(|subject_ref| !subject_ref.is_empty())
|
||||
}) || events.iter().any(|event| {
|
||||
event.attributes.contains_key("commit") || event.attributes.contains_key("head_commit")
|
||||
});
|
||||
let review_status = merge_request.map(|request| request.review_status.clone());
|
||||
let approved = review_status.as_deref() == Some("approved");
|
||||
let unresolved_request_changes = review_status.as_deref() == Some("changes_requested");
|
||||
@@ -1759,6 +1818,79 @@ mod tests {
|
||||
use super::*;
|
||||
use crate::store::{ObjectiveRecord, ObjectiveTicketLinkRecord, WorkspaceRecord};
|
||||
|
||||
#[test]
|
||||
fn merge_request_summary_tracks_the_latest_requested_subject() {
|
||||
let actor = merge_request::WorkerIdentity {
|
||||
runtime_id: "runtime-1".to_string(),
|
||||
worker_id: "worker-1".to_string(),
|
||||
};
|
||||
let at = Utc::now();
|
||||
let first_review = merge_request::ReviewEvent {
|
||||
event_id: "review-1".to_string(),
|
||||
sequence: 2,
|
||||
request_event_id: "request-1".to_string(),
|
||||
subject_ref: "commit-1".to_string(),
|
||||
decision: ReviewDecision::Approve,
|
||||
body: "approved".to_string(),
|
||||
findings: Vec::new(),
|
||||
reviewer: actor.clone(),
|
||||
created_at: at,
|
||||
};
|
||||
let mut request = MergeRequest {
|
||||
workspace_id: "workspace-1".to_string(),
|
||||
merge_request_id: "mr-1".to_string(),
|
||||
repository_id: "main".to_string(),
|
||||
state: MergeRequestState::Open,
|
||||
selector_from: Some("work/ticket-1".to_string()),
|
||||
selector_to: "orchestration".to_string(),
|
||||
ticket_ids: vec!["ticket-1".to_string()],
|
||||
thread: vec![
|
||||
MergeRequestThreadEvent::ReviewRequested(merge_request::ReviewRequestedEvent {
|
||||
event_id: "request-1".to_string(),
|
||||
sequence: 1,
|
||||
subject_ref: "commit-1".to_string(),
|
||||
requested_by: actor.clone(),
|
||||
reviewer: actor.clone(),
|
||||
created_at: at,
|
||||
}),
|
||||
MergeRequestThreadEvent::Review(first_review),
|
||||
MergeRequestThreadEvent::ReviewRequested(merge_request::ReviewRequestedEvent {
|
||||
event_id: "request-2".to_string(),
|
||||
sequence: 3,
|
||||
subject_ref: "commit-2".to_string(),
|
||||
requested_by: actor.clone(),
|
||||
reviewer: actor.clone(),
|
||||
created_at: at,
|
||||
}),
|
||||
],
|
||||
created_at: at,
|
||||
updated_at: at,
|
||||
};
|
||||
|
||||
let pending = merge_request_summary(request.clone());
|
||||
assert_eq!(pending.review_status, "pending");
|
||||
assert_eq!(pending.review_subject_ref.as_deref(), Some("commit-2"));
|
||||
assert_eq!(pending.review_submitted_at, None);
|
||||
|
||||
request.thread.push(MergeRequestThreadEvent::Review(
|
||||
merge_request::ReviewEvent {
|
||||
event_id: "review-2".to_string(),
|
||||
sequence: 4,
|
||||
request_event_id: "request-2".to_string(),
|
||||
subject_ref: "commit-2".to_string(),
|
||||
decision: ReviewDecision::Approve,
|
||||
body: "current approval".to_string(),
|
||||
findings: Vec::new(),
|
||||
reviewer: actor,
|
||||
created_at: at,
|
||||
},
|
||||
));
|
||||
let approved = merge_request_summary(request);
|
||||
assert_eq!(approved.review_status, "approved");
|
||||
assert_eq!(approved.review_subject_ref.as_deref(), Some("commit-2"));
|
||||
assert_eq!(approved.review_excerpt.as_deref(), Some("current approval"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ticket_evidence_summary_requires_report_after_latest_rescope_and_approved_revision() {
|
||||
let event = |kind: &str| TicketEvent {
|
||||
@@ -1779,11 +1911,10 @@ mod tests {
|
||||
merge_request_id: "mr-1".to_string(),
|
||||
state: "open".to_string(),
|
||||
review_status: "approved".to_string(),
|
||||
revision_id: "revision-1".to_string(),
|
||||
base_commit: "base".to_string(),
|
||||
head_commit: "head".to_string(),
|
||||
changed_paths: vec!["src/lib.rs".to_string()],
|
||||
selector_from: Some("work/ticket".to_string()),
|
||||
selector_to: "orchestration".to_string(),
|
||||
updated_at: "2026-01-01T00:00:00Z".to_string(),
|
||||
review_subject_ref: Some("head".to_string()),
|
||||
review_submitted_at: Some("2026-01-01T00:00:00Z".to_string()),
|
||||
review_excerpt: Some("approved".to_string()),
|
||||
};
|
||||
|
||||
@@ -241,11 +241,10 @@ pub struct TicketMergeRequestSummary {
|
||||
pub merge_request_id: String,
|
||||
pub state: String,
|
||||
pub review_status: String,
|
||||
pub revision_id: String,
|
||||
pub base_commit: String,
|
||||
pub head_commit: String,
|
||||
pub changed_paths: Vec<String>,
|
||||
pub selector_from: Option<String>,
|
||||
pub selector_to: String,
|
||||
pub updated_at: String,
|
||||
pub review_subject_ref: Option<String>,
|
||||
pub review_submitted_at: Option<String>,
|
||||
pub review_excerpt: Option<String>,
|
||||
}
|
||||
|
||||
+383
-1197
File diff suppressed because it is too large
Load Diff
@@ -191,6 +191,11 @@ const MIGRATIONS: &[Migration] = &[
|
||||
name: "create Worker control delegation operation authority",
|
||||
apply: create_worker_control_delegation_operation_authority,
|
||||
},
|
||||
Migration {
|
||||
version: 35,
|
||||
name: "remove Worker control delegation authority",
|
||||
apply: remove_worker_control_delegation_authority,
|
||||
},
|
||||
];
|
||||
|
||||
struct Migration {
|
||||
@@ -352,18 +357,6 @@ pub struct WorkerControlGrantRecord {
|
||||
pub revoked_at: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||
pub struct WorkerControlDelegationOperationRecord {
|
||||
pub workspace_id: String,
|
||||
pub source_controller: RuntimeWorkerRef,
|
||||
pub source_grant_id: String,
|
||||
pub operation_id: String,
|
||||
pub input_fingerprint: String,
|
||||
pub delegated_grant_id: Option<String>,
|
||||
pub created_at: String,
|
||||
pub completed_at: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||
pub struct TicketWorkerAssignmentRecord {
|
||||
pub workspace_id: String,
|
||||
@@ -810,19 +803,6 @@ pub trait ControlPlaneStore: Send + Sync {
|
||||
grant_id: &str,
|
||||
revoked_at: &str,
|
||||
) -> Result<bool>;
|
||||
fn reserve_worker_control_delegation_operation(
|
||||
&self,
|
||||
record: &WorkerControlDelegationOperationRecord,
|
||||
) -> Result<WorkerControlDelegationOperationRecord>;
|
||||
fn complete_worker_control_delegation_operation(
|
||||
&self,
|
||||
workspace_id: &str,
|
||||
source_controller: &RuntimeWorkerRef,
|
||||
operation_id: &str,
|
||||
delegated_grant_id: &str,
|
||||
completed_at: &str,
|
||||
) -> Result<WorkerControlDelegationOperationRecord>;
|
||||
|
||||
fn get_ticket_assignment_operation(
|
||||
&self,
|
||||
workspace_id: &str,
|
||||
@@ -2611,95 +2591,6 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
|
||||
})
|
||||
}
|
||||
|
||||
fn reserve_worker_control_delegation_operation(
|
||||
&self,
|
||||
record: &WorkerControlDelegationOperationRecord,
|
||||
) -> Result<WorkerControlDelegationOperationRecord> {
|
||||
self.with_conn(|conn| {
|
||||
conn.execute(
|
||||
r#"INSERT INTO worker_control_delegation_operations (
|
||||
workspace_id, source_controller_runtime_id, source_controller_worker_id,
|
||||
source_grant_id, operation_id, input_fingerprint,
|
||||
delegated_grant_id, created_at, completed_at
|
||||
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
|
||||
ON CONFLICT (
|
||||
workspace_id, source_controller_runtime_id,
|
||||
source_controller_worker_id, operation_id
|
||||
) DO NOTHING"#,
|
||||
params![
|
||||
record.workspace_id,
|
||||
record.source_controller.runtime_id,
|
||||
record.source_controller.worker_id,
|
||||
record.source_grant_id,
|
||||
record.operation_id,
|
||||
record.input_fingerprint,
|
||||
record.delegated_grant_id,
|
||||
record.created_at,
|
||||
record.completed_at,
|
||||
],
|
||||
)?;
|
||||
let persisted = read_worker_control_delegation_operation_by_key(
|
||||
conn,
|
||||
&record.workspace_id,
|
||||
&record.source_controller,
|
||||
&record.operation_id,
|
||||
)?
|
||||
.ok_or_else(|| {
|
||||
Error::Store("worker control delegation operation was not persisted".to_string())
|
||||
})?;
|
||||
if persisted.source_grant_id != record.source_grant_id
|
||||
|| persisted.input_fingerprint != record.input_fingerprint
|
||||
{
|
||||
return Err(Error::InvalidInput(format!(
|
||||
"worker control delegation operation `{}` was already used with different input",
|
||||
record.operation_id
|
||||
)));
|
||||
}
|
||||
Ok(persisted)
|
||||
})
|
||||
}
|
||||
|
||||
fn complete_worker_control_delegation_operation(
|
||||
&self,
|
||||
workspace_id: &str,
|
||||
source_controller: &RuntimeWorkerRef,
|
||||
operation_id: &str,
|
||||
delegated_grant_id: &str,
|
||||
completed_at: &str,
|
||||
) -> Result<WorkerControlDelegationOperationRecord> {
|
||||
self.with_conn(|conn| {
|
||||
conn.execute(
|
||||
r#"UPDATE worker_control_delegation_operations
|
||||
SET delegated_grant_id = ?5, completed_at = ?6
|
||||
WHERE workspace_id = ?1
|
||||
AND source_controller_runtime_id = ?2
|
||||
AND source_controller_worker_id = ?3
|
||||
AND operation_id = ?4
|
||||
AND (delegated_grant_id IS NULL OR delegated_grant_id = ?5)"#,
|
||||
params![
|
||||
workspace_id,
|
||||
source_controller.runtime_id,
|
||||
source_controller.worker_id,
|
||||
operation_id,
|
||||
delegated_grant_id,
|
||||
completed_at,
|
||||
],
|
||||
)?;
|
||||
read_worker_control_delegation_operation_by_key(
|
||||
conn,
|
||||
workspace_id,
|
||||
source_controller,
|
||||
operation_id,
|
||||
)?
|
||||
.filter(|record| record.delegated_grant_id.as_deref() == Some(delegated_grant_id))
|
||||
.ok_or_else(|| {
|
||||
Error::InvalidInput(format!(
|
||||
"worker control delegation operation `{operation_id}` completed with a different grant"
|
||||
))
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
fn get_ticket_assignment_operation(
|
||||
&self,
|
||||
workspace_id: &str,
|
||||
@@ -4057,52 +3948,6 @@ fn read_worker_control_grant_by_operation(
|
||||
.map_err(Error::from)
|
||||
}
|
||||
|
||||
fn read_worker_control_delegation_operation_record(
|
||||
row: &rusqlite::Row<'_>,
|
||||
) -> rusqlite::Result<WorkerControlDelegationOperationRecord> {
|
||||
Ok(WorkerControlDelegationOperationRecord {
|
||||
workspace_id: row.get(0)?,
|
||||
source_controller: RuntimeWorkerRef::new(
|
||||
row.get::<_, String>(1)?,
|
||||
row.get::<_, u64>(2)?.to_string(),
|
||||
),
|
||||
source_grant_id: row.get(3)?,
|
||||
operation_id: row.get(4)?,
|
||||
input_fingerprint: row.get(5)?,
|
||||
delegated_grant_id: row.get(6)?,
|
||||
created_at: row.get(7)?,
|
||||
completed_at: row.get(8)?,
|
||||
})
|
||||
}
|
||||
|
||||
fn read_worker_control_delegation_operation_by_key(
|
||||
conn: &Connection,
|
||||
workspace_id: &str,
|
||||
source_controller: &RuntimeWorkerRef,
|
||||
operation_id: &str,
|
||||
) -> Result<Option<WorkerControlDelegationOperationRecord>> {
|
||||
conn.query_row(
|
||||
r#"SELECT workspace_id,
|
||||
source_controller_runtime_id, source_controller_worker_id,
|
||||
source_grant_id, operation_id, input_fingerprint,
|
||||
delegated_grant_id, created_at, completed_at
|
||||
FROM worker_control_delegation_operations
|
||||
WHERE workspace_id = ?1
|
||||
AND source_controller_runtime_id = ?2
|
||||
AND source_controller_worker_id = ?3
|
||||
AND operation_id = ?4"#,
|
||||
params![
|
||||
workspace_id,
|
||||
source_controller.runtime_id,
|
||||
source_controller.worker_id,
|
||||
operation_id,
|
||||
],
|
||||
read_worker_control_delegation_operation_record,
|
||||
)
|
||||
.optional()
|
||||
.map_err(Error::from)
|
||||
}
|
||||
|
||||
fn current_ticket_worker_assignment_select_sql() -> String {
|
||||
"SELECT a.workspace_id, a.ticket_id, a.assignment_id, a.runtime_id, a.worker_id, \
|
||||
a.assigned_by, a.assigned_at \
|
||||
@@ -5201,6 +5046,58 @@ fn create_worker_control_delegation_operation_authority(conn: &Connection) -> Re
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn remove_worker_control_delegation_authority(conn: &Connection) -> Result<()> {
|
||||
let mut statement =
|
||||
conn.prepare("SELECT workspace_id, grant_id, permissions_json FROM worker_control_grants")?;
|
||||
let rows = statement.query_map([], |row| {
|
||||
Ok((
|
||||
row.get::<_, String>(0)?,
|
||||
row.get::<_, String>(1)?,
|
||||
row.get::<_, String>(2)?,
|
||||
))
|
||||
})?;
|
||||
let mut permission_updates = Vec::new();
|
||||
for row in rows {
|
||||
let (workspace_id, grant_id, permissions_json) = row?;
|
||||
let mut permissions: Vec<String> =
|
||||
serde_json::from_str(&permissions_json).map_err(|error| {
|
||||
Error::Store(format!(
|
||||
"failed to decode Worker control grant `{grant_id}` permissions during delegation removal: {error}"
|
||||
))
|
||||
})?;
|
||||
let previous_len = permissions.len();
|
||||
permissions
|
||||
.retain(|permission| !matches!(permission.as_str(), "share" | "transfer" | "revoke"));
|
||||
if permissions.len() != previous_len {
|
||||
permission_updates.push((
|
||||
workspace_id,
|
||||
grant_id,
|
||||
serde_json::to_string(&permissions).map_err(|error| {
|
||||
Error::Store(format!(
|
||||
"failed to encode Worker control grant permissions during delegation removal: {error}"
|
||||
))
|
||||
})?,
|
||||
));
|
||||
}
|
||||
}
|
||||
drop(statement);
|
||||
|
||||
for (workspace_id, grant_id, permissions_json) in permission_updates {
|
||||
conn.execute(
|
||||
"UPDATE worker_control_grants SET permissions_json = ?3 WHERE workspace_id = ?1 AND grant_id = ?2",
|
||||
params![workspace_id, grant_id, permissions_json],
|
||||
)?;
|
||||
}
|
||||
conn.execute(
|
||||
r#"UPDATE worker_control_grants
|
||||
SET revoked_at = COALESCE(revoked_at, strftime('%Y-%m-%dT%H:%M:%SZ', 'now'))
|
||||
WHERE relation IN ('shared', 'transferred')"#,
|
||||
[],
|
||||
)?;
|
||||
conn.execute_batch("DROP TABLE IF EXISTS worker_control_delegation_operations;")?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn create_worker_mutation_source_proof_replay_guard(conn: &Connection) -> Result<()> {
|
||||
conn.execute_batch(
|
||||
r#"
|
||||
@@ -5852,6 +5749,82 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn schema_v35_removes_worker_control_delegation_authority() {
|
||||
let conn = Connection::open_in_memory().unwrap();
|
||||
configure_sqlite(&conn).unwrap();
|
||||
for migration in MIGRATIONS
|
||||
.iter()
|
||||
.filter(|migration| migration.version <= 34)
|
||||
{
|
||||
let tx = conn.unchecked_transaction().unwrap();
|
||||
(migration.apply)(&tx).unwrap();
|
||||
tx.execute(
|
||||
"INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)",
|
||||
params![migration.version, migration.name],
|
||||
)
|
||||
.unwrap();
|
||||
tx.commit().unwrap();
|
||||
}
|
||||
conn.execute_batch(
|
||||
r#"
|
||||
INSERT INTO workspaces (
|
||||
workspace_id, display_name, state, created_at, updated_at
|
||||
) VALUES ('workspace-a', 'Workspace A', 'active', '1', '1');
|
||||
INSERT INTO worker_registry (
|
||||
workspace_id, runtime_id, runtime_worker_id, display_name,
|
||||
retention_state, created_at, updated_at
|
||||
) VALUES
|
||||
('workspace-a', 'runtime-a', 1, 'Controller', 'normal', '1', '1'),
|
||||
('workspace-a', 'runtime-a', 2, 'Spawned Worker', 'normal', '1', '1'),
|
||||
('workspace-a', 'runtime-a', 3, 'Shared Worker', 'normal', '1', '1'),
|
||||
('workspace-a', 'runtime-a', 4, 'Transferred Worker', 'normal', '1', '1');
|
||||
INSERT INTO worker_control_grants (
|
||||
workspace_id, grant_id,
|
||||
controller_runtime_id, controller_worker_id,
|
||||
subject_runtime_id, subject_worker_id,
|
||||
relation, origin, permissions_json, operation_id, created_at, revoked_at
|
||||
) VALUES
|
||||
('workspace-a', 'spawned', 'runtime-a', 1, 'runtime-a', 2,
|
||||
'spawned', 'spawn', '["observe","share","transfer","revoke","stop"]', 'spawn-op', '1', NULL),
|
||||
('workspace-a', 'shared', 'runtime-a', 1, 'runtime-a', 3,
|
||||
'shared', 'share', '["observe"]', 'share-op', '1', NULL),
|
||||
('workspace-a', 'transferred', 'runtime-a', 1, 'runtime-a', 4,
|
||||
'transferred', 'transfer', '["observe"]', 'transfer-op', '1', NULL);
|
||||
"#,
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(current_schema_version(&conn).unwrap(), 34);
|
||||
assert!(table_exists(&conn, "worker_control_delegation_operations").unwrap());
|
||||
|
||||
apply_migrations(&conn).unwrap();
|
||||
|
||||
assert_eq!(current_schema_version(&conn).unwrap(), 35);
|
||||
assert!(!table_exists(&conn, "worker_control_delegation_operations").unwrap());
|
||||
let (permissions_json, revoked_at): (String, Option<String>) = conn
|
||||
.query_row(
|
||||
"SELECT permissions_json, revoked_at FROM worker_control_grants WHERE grant_id = 'spawned'",
|
||||
[],
|
||||
|row| Ok((row.get(0)?, row.get(1)?)),
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
serde_json::from_str::<Vec<String>>(&permissions_json).unwrap(),
|
||||
vec!["observe", "stop"]
|
||||
);
|
||||
assert!(revoked_at.is_none());
|
||||
for grant_id in ["shared", "transferred"] {
|
||||
let revoked_at: Option<String> = conn
|
||||
.query_row(
|
||||
"SELECT revoked_at FROM worker_control_grants WHERE grant_id = ?1",
|
||||
[grant_id],
|
||||
|row| row.get(0),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(revoked_at.is_some(), "{grant_id} grant remained active");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn schema_v24_adds_attachment_reservations_to_already_applied_v23() {
|
||||
let conn = Connection::open_in_memory().unwrap();
|
||||
@@ -5874,7 +5847,7 @@ CREATE TABLE ticket_worker_links (ticket_id TEXT, worker_ref_key TEXT);
|
||||
|
||||
apply_migrations(&conn).unwrap();
|
||||
|
||||
assert_eq!(current_schema_version(&conn).unwrap(), 34);
|
||||
assert_eq!(current_schema_version(&conn).unwrap(), 35);
|
||||
assert!(table_exists(&conn, "worker_workdir_attachment_reservations").unwrap());
|
||||
}
|
||||
|
||||
@@ -5907,7 +5880,7 @@ CREATE TABLE flow_events (event_id TEXT PRIMARY KEY);
|
||||
|
||||
apply_migrations(&conn).unwrap();
|
||||
|
||||
assert_eq!(current_schema_version(&conn).unwrap(), 34);
|
||||
assert_eq!(current_schema_version(&conn).unwrap(), 35);
|
||||
assert!(table_exists(&conn, "flow_sources").unwrap());
|
||||
assert!(table_exists(&conn, "flow_source_revisions").unwrap());
|
||||
assert!(!table_exists(&conn, "flow_instances").unwrap());
|
||||
@@ -5974,7 +5947,7 @@ INSERT INTO worker_workdir_attachment_reservations (
|
||||
|
||||
apply_migrations(&conn).unwrap();
|
||||
|
||||
assert_eq!(current_schema_version(&conn).unwrap(), 34);
|
||||
assert_eq!(current_schema_version(&conn).unwrap(), 35);
|
||||
let repositories_sql: String = conn
|
||||
.query_row(
|
||||
"SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'repositories'",
|
||||
@@ -6154,7 +6127,7 @@ INSERT INTO workdir_registry (
|
||||
let db = dir.path().join("control-plane.sqlite");
|
||||
let store = SqliteWorkspaceStore::open(&db).unwrap();
|
||||
|
||||
assert_eq!(store.schema_version().await.unwrap(), 34);
|
||||
assert_eq!(store.schema_version().await.unwrap(), 35);
|
||||
assert!(
|
||||
!store
|
||||
.with_conn(|conn| table_exists(conn, "worker_workspace_credentials"))
|
||||
@@ -6171,7 +6144,7 @@ INSERT INTO workdir_registry (
|
||||
store.upsert_workspace(&record).await.unwrap();
|
||||
|
||||
let reopened = SqliteWorkspaceStore::open(&db).unwrap();
|
||||
assert_eq!(reopened.schema_version().await.unwrap(), 34);
|
||||
assert_eq!(reopened.schema_version().await.unwrap(), 35);
|
||||
assert_eq!(
|
||||
reopened.get_workspace("local-dev").await.unwrap(),
|
||||
Some(record)
|
||||
@@ -6718,7 +6691,7 @@ INSERT INTO workdir_registry (
|
||||
.unwrap();
|
||||
|
||||
let store = SqliteWorkspaceStore::from_connection(conn).unwrap();
|
||||
assert_eq!(store.schema_version().await.unwrap(), 34);
|
||||
assert_eq!(store.schema_version().await.unwrap(), 35);
|
||||
|
||||
store
|
||||
.with_conn(|conn| {
|
||||
@@ -6907,7 +6880,7 @@ CREATE TABLE ticket_assignment_operations (
|
||||
#[tokio::test]
|
||||
async fn repository_records_round_trip() {
|
||||
let store = SqliteWorkspaceStore::in_memory().unwrap();
|
||||
assert_eq!(store.schema_version().await.unwrap(), 34);
|
||||
assert_eq!(store.schema_version().await.unwrap(), 35);
|
||||
let workspace = WorkspaceRecord {
|
||||
workspace_id: "local-dev".to_string(),
|
||||
owner_account_id: None,
|
||||
@@ -6973,7 +6946,7 @@ CREATE TABLE ticket_assignment_operations (
|
||||
#[tokio::test]
|
||||
async fn memory_authority_records_round_trip_and_close_staging() {
|
||||
let store = SqliteWorkspaceStore::in_memory().unwrap();
|
||||
assert_eq!(store.schema_version().await.unwrap(), 34);
|
||||
assert_eq!(store.schema_version().await.unwrap(), 35);
|
||||
let workspace = WorkspaceRecord {
|
||||
workspace_id: "local-dev".to_string(),
|
||||
owner_account_id: None,
|
||||
@@ -7234,7 +7207,7 @@ CREATE TABLE ticket_assignment_operations (
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn worker_control_grants_are_idempotent_scoped_and_revocable() {
|
||||
async fn worker_control_grants_are_idempotent_scoped_and_support_internal_invalidation() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let database = dir.path().join("control-grants.db");
|
||||
let store = SqliteWorkspaceStore::open(&database).unwrap();
|
||||
@@ -7364,7 +7337,7 @@ CREATE TABLE ticket_assignment_operations (
|
||||
#[tokio::test]
|
||||
async fn account_and_login_records_round_trip() {
|
||||
let store = SqliteWorkspaceStore::in_memory().unwrap();
|
||||
assert_eq!(store.schema_version().await.unwrap(), 34);
|
||||
assert_eq!(store.schema_version().await.unwrap(), 35);
|
||||
let now = "2026-07-22T00:00:00Z".to_string();
|
||||
let account = AccountRecord {
|
||||
account_id: "acct-user-alice".to_string(),
|
||||
|
||||
Reference in New Issue
Block a user