Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

fix: align merge requests with selector thread contract

Browse Source
This commit is contained in:
Hare
2026-08-17 07:41:03 +09:00
parent 9e48cae759
commit 1fe1b7463f
13 changed files with 1354 additions and 2658 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/merge-request/Cargo.toml
+1
View File
@@ -10,6 +10,7 @@ rusqlite.workspace = true
serde = { workspace = true, features = ["derive"] }
serde_json.workspace = true
thiserror.workspace = true
uuid = { workspace = true, features = ["v7"] }
[dev-dependencies]
tempfile.workspace = true
crates/merge-request/src/lib.rs
+760 -1478
View File
File diff suppressed because it is too large Load Diff
crates/merge-request/tests/store.rs
+186 -378
View File
@@ -1,407 +1,215 @@
use std::sync::{Arc, Mutex};
use chrono::{TimeZone, Utc};
use merge_request::{
AssignmentSource, CompleteMergeRequest, ConflictResolution, CurrentAssignment, FindingSeverity,
MergeRequestAuth, MergeRequestState, MergeRequestStore, MergeRequestThreadEvent, MergeStrategy,
OpenMergeRequest, ReadinessCheck, RegisterReviewCapability, RegisterReviewerChildSession,
RepositorySource, RequestForReview, RequestMergeRequestReview, ReviewDecision, ReviewFinding,
SubmitMergeRequestReview,
};
use rusqlite::{Connection, params};
use merge_request::*;
use rusqlite::Connection;
use std::sync::{Arc, Mutex};
#[derive(Clone)]
struct Assignments {
current: Arc<Mutex<CurrentAssignment>>,
}
struct Assignments(Arc<Mutex<CurrentAssignment>>);
impl AssignmentSource for Assignments {
fn current_assignment(
&self,
_workspace_id: &str,
_ticket_id: &str,
) -> Result<Option<CurrentAssignment>, String> {
Ok(Some(self.current.lock().unwrap().clone()))
fn current_assignment(&self, _: &str, _: &str) -> Result<Option<CurrentAssignment>, String> {
Ok(Some(self.0.lock().unwrap().clone()))
}
}
struct Repositories;
impl RepositorySource for Repositories {
fn repository_belongs_to_workspace(
&self,
workspace_id: &str,
repository_id: &str,
) -> Result<bool, String> {
Ok(workspace_id == "W" && repository_id == "R")
}
fn is_ancestor(
&self,
_workspace_id: &str,
_repository_id: &str,
ancestor: &str,
descendant: &str,
) -> Result<bool, String> {
Ok(matches!(
(ancestor, descendant),
("base", "head-1") | ("base", "head-2")
))
fn repository_belongs_to_workspace(&self, w: &str, r: &str) -> Result<bool, String> {
Ok(w == "W" && r == "R")
}
}
fn now(second: u32) -> chrono::DateTime<Utc> {
Utc.with_ymd_and_hms(2026, 7, 26, 12, 0, second)
fn at(s: u32) -> chrono::DateTime<Utc> {
Utc.with_ymd_and_hms(2026, 7, 26, 12, 0, s)
.single()
.unwrap()
}
fn fixture() -> (tempfile::TempDir, MergeRequestStore, Assignments) {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("server.db");
let conn = Connection::open(&path).unwrap();
conn.execute_batch(
"PRAGMA foreign_keys = ON;
CREATE TABLE workspaces (workspace_id TEXT PRIMARY KEY);
CREATE TABLE repositories (
workspace_id TEXT NOT NULL,
repository_id TEXT NOT NULL,
PRIMARY KEY (workspace_id, repository_id),
FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_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,
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 NOT NULL, at TEXT NOT NULL,
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)
);
INSERT INTO workspaces VALUES ('W');
INSERT INTO repositories VALUES ('W', 'R');
INSERT INTO typed_tickets VALUES ('W', 'T', 'inprogress', 1, '2026-07-26T12:00:00Z');",
)
.unwrap();
drop(conn);
let assignments = Assignments {
current: Arc::new(Mutex::new(CurrentAssignment {
assignment_id: "A1".into(),
ticket_id: "T".into(),
runtime_id: "runtime".into(),
worker_id: "coder".into(),
})),
};
let store =
MergeRequestStore::open(&path, Arc::new(assignments.clone()), Arc::new(Repositories))
.unwrap();
(dir, store, assignments)
}
fn auth(assignment_id: &str) -> MergeRequestAuth {
fn auth() -> MergeRequestAuth {
MergeRequestAuth {
workspace_id: "W".into(),
repository_id: "R".into(),
runtime_id: "runtime".into(),
worker_id: "coder".into(),
assignment_id: assignment_id.into(),
assignment_id: "A".into(),
}
}
fn open(store: &MergeRequestStore) {
store
.open_merge_request(OpenMergeRequest {
merge_request_id: "MR".into(),
ticket_id: "T".into(),
repository_id: "R".into(),
selector_from: "work/t-feature".into(),
selector_to: "develop".into(),
request: RequestForReview {
base_commit: "base".into(),
head_commit: "head-1".into(),
changed_paths: vec!["src/lib.rs".into()],
summary: "first candidate".into(),
},
auth: auth("A1"),
now: now(1),
})
.unwrap();
}
fn approve(store: &MergeRequestStore, expected_head_commit: &str, token: &str) {
store
.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: now(2),
})
.unwrap();
store
.register_review_capability(RegisterReviewCapability {
ticket_id: "T".into(),
expected_head_commit: expected_head_commit.into(),
child_session_id: format!("child-{token}"),
capability_token: token.into(),
auth: auth("A1"),
now: now(3),
})
.unwrap();
store
.submit_review(SubmitMergeRequestReview {
ticket_id: "T".into(),
expected_head_commit: expected_head_commit.into(),
capability_token: token.into(),
decision: ReviewDecision::Approve,
body: "approved independently".into(),
findings: vec![ReviewFinding {
severity: FindingSeverity::Note,
path: None,
line: None,
message: "looks good".into(),
}],
now: now(4),
})
.unwrap();
}
#[test]
fn thread_drives_review_readiness_and_completion_without_public_revision_identity() {
let (_dir, store, _assignments) = fixture();
open(&store);
approve(&store, "head-1", "token-1");
let readiness = store
.readiness(ReadinessCheck {
ticket_id: "T".into(),
expected_head_commit: Some("head-1".into()),
auth: auth("A1"),
})
.unwrap();
assert!(readiness.ready, "{:?}", readiness.blockers);
let merged = store
.complete(CompleteMergeRequest {
ticket_id: "T".into(),
expected_head_commit: "head-1".into(),
operation_id: "op-1".into(),
target_commit: "base".into(),
source_commit: "head-1".into(),
result_commit: "head-1".into(),
strategy: MergeStrategy::FastForward,
resolution: ConflictResolution::None,
auth: auth("A1"),
now: now(5),
})
.unwrap();
assert_eq!(merged.result_commit, "head-1");
let mr = store.get("W", "T").unwrap();
assert_eq!(mr.state, MergeRequestState::Merged);
assert!(matches!(
mr.thread.last(),
Some(MergeRequestThreadEvent::Merge(_))
));
assert_eq!(mr.selector_from, "work/t-feature");
assert_eq!(mr.selector_to, "develop");
let json = serde_json::to_string(&mr).unwrap();
for forbidden in [
"revision_id",
"current_revision",
"attempt_id",
"review_attempt",
"head_tree",
"diff_digest",
"merged_revision_id",
] {
assert!(
!json.contains(forbidden),
"unexpected `{forbidden}` in {json}"
);
}
}
#[test]
fn new_review_request_invalidates_prior_approval_and_fences_stale_capability() {
let (_dir, store, _assignments) = fixture();
open(&store);
approve(&store, "head-1", "token-1");
store
.request_review(RequestMergeRequestReview {
ticket_id: "T".into(),
expected_head_commit: "head-1".into(),
request: RequestForReview {
base_commit: "base".into(),
head_commit: "head-2".into(),
changed_paths: vec!["src/lib.rs".into(), "tests/store.rs".into()],
summary: "address review".into(),
},
auth: auth("A1"),
now: now(6),
})
.unwrap();
let readiness = store
.readiness(ReadinessCheck {
ticket_id: "T".into(),
expected_head_commit: Some("head-2".into()),
auth: auth("A1"),
})
.unwrap();
assert!(!readiness.ready);
assert!(
readiness
.blockers
.iter()
.any(|value| value.contains("no review result"))
);
assert!(
store
.submit_review(SubmitMergeRequestReview {
ticket_id: "T".into(),
expected_head_commit: "head-1".into(),
capability_token: "token-1".into(),
decision: ReviewDecision::Approve,
body: "stale".into(),
findings: vec![],
now: now(7),
})
.is_err()
);
}
#[test]
fn assignment_change_rejects_candidate_mutation() {
let (_dir, store, assignments) = fixture();
open(&store);
*assignments.current.lock().unwrap() = CurrentAssignment {
assignment_id: "A2".into(),
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 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 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: "other".into(),
};
let error = store
.request_review(RequestMergeRequestReview {
worker_id: "coder".into(),
})));
let s = MergeRequestStore::open(&p, Arc::new(a), Arc::new(Repositories)).unwrap();
(d, s)
}
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();
}
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
}
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: 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(),
expected_head_commit: "head-1".into(),
request: RequestForReview {
base_commit: "base".into(),
head_commit: "head-2".into(),
changed_paths: vec![],
summary: String::new(),
},
auth: auth("A1"),
now: now(6),
current_subject_ref: Some("opaque-source-ref".into()),
auth: auth(),
})
.unwrap_err();
assert!(error.to_string().contains("current assigned worker"));
.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_builds_thread_events_and_removes_revision_tables() {
let conn = Connection::open_in_memory().unwrap();
conn.execute_batch(
"PRAGMA foreign_keys = OFF;
CREATE TABLE workspaces (workspace_id TEXT PRIMARY KEY);
CREATE TABLE repositories (
workspace_id TEXT NOT NULL, repository_id TEXT NOT NULL,
PRIMARY KEY (workspace_id, repository_id)
);
INSERT INTO workspaces VALUES ('W');
INSERT INTO repositories VALUES ('W', 'R');
CREATE TABLE typed_tickets (
workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL,
PRIMARY KEY (workspace_id, ticket_id)
);
INSERT INTO typed_tickets VALUES ('W', 'T');
CREATE TABLE merge_request_schema (singleton INTEGER PRIMARY KEY, version INTEGER NOT NULL);
INSERT INTO merge_request_schema VALUES (1, 11);
CREATE TABLE merge_requests (
workspace_id TEXT, merge_request_id TEXT, ticket_id TEXT, repository_id TEXT,
state TEXT, target_ref_selector TEXT, current_revision_id TEXT,
opened_by_worker_runtime_id TEXT, opened_by_worker_id TEXT, created_at TEXT, updated_at TEXT
);
CREATE TABLE merge_request_revisions (
workspace_id TEXT, revision_id TEXT, merge_request_id TEXT, base_commit TEXT,
head_commit TEXT, changed_paths_json TEXT, summary TEXT, assignment_id TEXT,
coder_worker_runtime_id TEXT, coder_worker_id TEXT, created_at TEXT
);
CREATE TABLE merge_request_review_attempts (attempt_id TEXT);
CREATE TABLE merge_request_reviews (
workspace_id TEXT, review_id TEXT, revision_id TEXT,
reviewer_worker_runtime_id TEXT, reviewer_worker_id TEXT, reviewer_profile TEXT,
decision TEXT, body TEXT, findings_json TEXT, created_at TEXT
);
CREATE TABLE merge_request_completion_operations (
workspace_id TEXT, merge_request_id TEXT, operation_id TEXT, target_commit TEXT,
source_commit TEXT, result_commit TEXT, strategy TEXT, resolution TEXT,
requested_by_runtime_id TEXT, requested_by_worker_id TEXT, completed_at TEXT, status TEXT
);
CREATE TABLE merge_request_reviewer_child_sessions (child_session_id TEXT);
INSERT INTO merge_requests VALUES (
'W', 'MR', 'T', 'R', 'open', 'develop', 'REV',
'runtime', 'coder', '2026-07-26T12:00:00Z', '2026-07-26T12:00:00Z'
);
INSERT INTO merge_request_revisions VALUES (
'W', 'REV', 'MR', 'base', 'head-1', '[\"src/lib.rs\"]', 'legacy', 'A1',
'runtime', 'coder', '2026-07-26T12:00:00Z'
);
INSERT INTO merge_request_reviews VALUES (
'W', 'REVIEW', 'REV', 'runtime', 'child', 'builtin:reviewer',
'approve', 'approved', '[]', '2026-07-26T12:00:01Z'
);",
)
.unwrap();
merge_request::migrate(&conn).unwrap();
assert_eq!(
conn.query_row("SELECT version FROM merge_request_schema", [], |row| row
.get::<_, i64>(0))
.unwrap(),
12
);
assert_eq!(
conn.query_row("SELECT selector_from FROM merge_requests", [], |row| row
.get::<_, String>(
0
))
.unwrap(),
"head-1"
);
assert_eq!(
conn.query_row(
"SELECT COUNT(*) FROM merge_request_thread_events",
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');").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",
[],
|row| row.get::<_, i64>(0)
|r| r.get(0),
)
.unwrap(),
2
);
for removed in [
"merge_request_revisions",
"merge_request_review_attempts",
"merge_request_reviews",
"merge_request_completion_operations",
] {
let exists: bool = conn
.query_row(
"SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name=?1)",
params![removed],
|row| row.get(0),
)
.unwrap();
assert!(!exists, "legacy table `{removed}` still exists");
}
.unwrap();
assert_eq!(kinds, "review_requested,review");
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);
}
crates/worker/src/feature/builtin/merge_request.rs
+98 -210
View File
@@ -11,77 +11,51 @@ pub const MERGE_REQUEST_COMMON_TOOL_NAMES: &[&str] = &[
"MergeRequestShow",
"MergeRequestReadinessCheck",
"MergeRequestOpen",
"MergeRequestRequestReview",
"MergeRequestComplete",
];
pub const MERGE_REQUEST_REVIEW_TOOL_NAME: &str = "MergeRequestReviewSubmit";
#[derive(Clone, Copy)]
enum Kind {
Show,
Readiness,
Open,
RequestReview,
Complete,
Review,
}
#[derive(Clone)]
struct MergeRequestTool {
client: Arc<dyn WorkspaceClient>,
kind: Kind,
}
#[derive(Debug, Deserialize, JsonSchema)]
struct ShowInput {
ticket: String,
}
#[derive(Debug, Deserialize, JsonSchema)]
struct OpenInput {
ticket: String,
repository_id: String,
selector_from: String,
selector_to: String,
base_commit: String,
head_commit: String,
#[serde(default)]
changed_paths: Vec<String>,
#[serde(default)]
summary: String,
}
#[derive(Debug, Deserialize, JsonSchema)]
struct RequestReviewInput {
ticket: String,
expected_head_commit: String,
base_commit: String,
head_commit: String,
#[serde(default)]
changed_paths: Vec<String>,
#[serde(default)]
summary: String,
}
#[derive(Debug, Deserialize, JsonSchema)]
struct CompleteInput {
ticket: String,
operation_id: String,
expected_head_commit: 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,
}
#[derive(Debug, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
enum MergeStrategyInput {
FastForward,
Merge,
}
#[derive(Debug, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
enum MergeResolutionInput {
@@ -89,7 +63,6 @@ enum MergeResolutionInput {
Clean,
ConflictsResolved,
}
#[derive(Debug, Deserialize, JsonSchema)]
struct ReviewInput {
decision: ReviewDecisionInput,
@@ -98,211 +71,133 @@ struct ReviewInput {
#[serde(default)]
findings: Vec<ReviewFindingInput>,
}
#[derive(Debug, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
enum ReviewDecisionInput {
Approve,
RequestChanges,
}
#[derive(Debug, Deserialize, JsonSchema)]
struct ReviewFindingInput {
severity: String,
#[serde(default)]
code: Option<String>,
#[serde(default)]
path: Option<String>,
#[serde(default)]
line: Option<u32>,
message: String,
body: String,
}
impl Kind {
fn name(self) -> &'static str {
match self {
Self::Show => "MergeRequestShow",
Self::Readiness => "MergeRequestReadinessCheck",
Self::Open => "MergeRequestOpen",
Self::RequestReview => "MergeRequestRequestReview",
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::RequestReview => json!(schemars::schema_for!(RequestReviewInput)),
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 value: ShowInput = parse(input)?;
nonempty(&value.ticket)?;
Kind::Show | Kind::Readiness => {
let v: ShowInput = parse(input)?;
nonempty(&v.ticket)?;
(
WorkspaceRequestMethod::Get,
format!(
"/api/w/{workspace_id}/tickets/{}/merge-request",
value.ticket
),
None,
)
}
Kind::Readiness => {
let value: ShowInput = parse(input)?;
nonempty(&value.ticket)?;
(
WorkspaceRequestMethod::Get,
format!(
"/api/w/{workspace_id}/tickets/{}/merge-request/readiness",
value.ticket
"/api/w/{ws}/tickets/{}/merge-request{}",
v.ticket,
if matches!(self.kind, Kind::Readiness) {
"/readiness"
} else {
""
}
),
None,
)
}
Kind::Open => {
let value: OpenInput = parse(input)?;
nonempty(&value.ticket)?;
let v: OpenInput = parse(input)?;
nonempty(&v.ticket)?;
(
WorkspaceRequestMethod::Post,
format!(
"/api/w/{workspace_id}/tickets/{}/merge-request",
value.ticket
format!("/api/w/{ws}/tickets/{}/merge-request", v.ticket),
Some(
json!({"repository_id":v.repository_id,"selector_from":v.selector_from,"selector_to":v.selector_to,"summary":v.summary}),
),
Some(json!({
"repository_id": value.repository_id,
"selector_from": value.selector_from,
"selector_to": value.selector_to,
"base_commit": value.base_commit,
"head_commit": value.head_commit,
"changed_paths": value.changed_paths,
"summary": value.summary,
})),
)
}
Kind::RequestReview => {
let value: RequestReviewInput = parse(input)?;
nonempty(&value.ticket)?;
(
WorkspaceRequestMethod::Post,
format!(
"/api/w/{workspace_id}/tickets/{}/merge-request/review-requests",
value.ticket
),
Some(json!({
"expected_head_commit": value.expected_head_commit,
"base_commit": value.base_commit,
"head_commit": value.head_commit,
"changed_paths": value.changed_paths,
"summary": value.summary,
})),
)
}
Kind::Complete => {
let value: CompleteInput = parse(input)?;
nonempty(&value.ticket)?;
let strategy = match value.strategy {
MergeStrategyInput::FastForward => "fast_forward",
MergeStrategyInput::Merge => "merge",
};
let resolution = match value.resolution {
MergeResolutionInput::None => "none",
MergeResolutionInput::Clean => "clean",
MergeResolutionInput::ConflictsResolved => "conflicts_resolved",
};
let v: CompleteInput = parse(input)?;
nonempty(&v.ticket)?;
(
WorkspaceRequestMethod::Post,
format!(
"/api/w/{workspace_id}/tickets/{}/merge-request/complete",
value.ticket
format!("/api/w/{ws}/tickets/{}/merge-request/complete", v.ticket),
Some(
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"}}),
),
Some(json!({
"operation_id": value.operation_id,
"expected_head_commit": value.expected_head_commit,
"target_commit": value.target_commit,
"source_commit": value.source_commit,
"result_commit": value.result_commit,
"strategy": strategy,
"resolution": resolution,
})),
)
}
Kind::Review => {
let value: ReviewInput = parse(input)?;
let context = self.client.reviewer_context().ok_or_else(|| {
let v: ReviewInput = parse(input)?;
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<_>>() }),
),
Some(json!({
"decision": match value.decision {
ReviewDecisionInput::Approve => "approve",
ReviewDecisionInput::RequestChanges => "request_changes",
},
"body": value.body,
"findings": value.findings.into_iter().map(|finding| json!({
"severity": finding.severity,
"path": finding.path,
"line": finding.line,
"message": finding.message,
})).collect::<Vec<_>>(),
})),
)
}
};
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)
.map_err(|error| ToolError::ExecutionFailed(error.to_string()))?;
if !response.is_success() {
.execute(req)
.map_err(|e| ToolError::ExecutionFailed(e.to_string()))?;
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(|error| ToolError::InvalidArgument(error.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(),
))
@@ -310,84 +205,77 @@ fn nonempty(value: &str) -> Result<(), ToolError> {
Ok(())
}
}
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::RequestReview),
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_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, selector pair, append-only thread, and current 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 review request.")
Some("Resolve current provider refs and derive readiness from valid review events.")
}
"MergeRequestOpen" => {
Some("Open a Merge Request with immutable source and target selectors.")
}
"MergeRequestComplete" => {
Some("Complete using an approved review event and final target-ref evidence.")
}
"MergeRequestOpen" => Some(
"Open a Merge Request with immutable source/target selectors and its first review request.",
),
"MergeRequestRequestReview" => Some(
"Append a RequestForReview event for new candidate evidence; prior approval cannot carry forward.",
),
"MergeRequestComplete" => Some(
"CAS-complete the approved current candidate with operation-id replay and crash fencing.",
),
"MergeRequestReviewSubmit" => {
Some("Submit the attested direct-child Reviewer result for the current candidate.")
Some("Submit the injected Reviewer capability result for its captured subject ref.")
}
_ => None,
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn merge_request_tool_contract_uses_selectors_and_commit_fences_without_revision_ids() {
let open = serde_json::to_string(&schemars::schema_for!(OpenInput)).unwrap();
let request = serde_json::to_string(&schemars::schema_for!(RequestReviewInput)).unwrap();
let complete = serde_json::to_string(&schemars::schema_for!(CompleteInput)).unwrap();
assert!(open.contains("selector_from"));
assert!(open.contains("selector_to"));
assert!(request.contains("expected_head_commit"));
assert!(complete.contains("expected_head_commit"));
for schema in [&open, &request, &complete] {
assert!(!schema.contains("revision_id"));
assert!(!schema.contains("attempt_id"));
assert!(!schema.contains("head_tree"));
assert!(!schema.contains("diff_digest"));
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(&"MergeRequestAddRevision"));
assert!(!MERGE_REQUEST_COMMON_TOOL_NAMES.contains(&"MergeRequestRequestReview"));
}
}
crates/worker/src/spawn/tool.rs
+7 -12
View File
@@ -67,7 +67,6 @@ struct SubWorkerSpawnInput {
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct ReviewerHandoffInput {
ticket_id: String,
expected_head_commit: 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.expected_head_commit.trim().is_empty() {
if review.ticket_id.trim().is_empty() {
return Err(ToolError::InvalidArgument(
"reviewer handoff requires non-empty ticket_id and expected_head_commit".to_string(),
"reviewer handoff requires non-empty ticket_id".to_string(),
));
}
if input.profile.as_deref() != Some("builtin:reviewer") {
@@ -421,7 +420,6 @@ impl Tool for SubWorkerSpawnTool {
let reviewer_capability = input.review.as_ref().map(|review| {
(
review.ticket_id.clone(),
review.expected_head_commit.clone(),
format!(
"{}{}",
uuid::Uuid::now_v7().simple(),
@@ -430,8 +428,7 @@ impl Tool for SubWorkerSpawnTool {
)
});
let child_workspace_context =
if let Some((ticket_id, expected_head_commit, capability_token)) = &reviewer_capability
{
if let Some((ticket_id, capability_token)) = &reviewer_capability {
let workspace_id =
self.workspace_context
.workspace_id()
@@ -452,7 +449,6 @@ impl Tool for SubWorkerSpawnTool {
parent_client.clone(),
ReviewerContext {
ticket_id: ticket_id.clone(),
expected_head_commit: expected_head_commit.clone(),
},
capability_token.clone(),
));
@@ -547,7 +543,7 @@ impl Tool for SubWorkerSpawnTool {
}
};
if let Some((ticket_id, expected_head_commit, capability_token)) = &reviewer_capability {
if let Some((ticket_id, capability_token)) = &reviewer_capability {
let workspace_id = self.workspace_context.workspace_id().ok_or_else(|| {
ToolError::ExecutionFailed("review capability lost Workspace identity".to_string())
})?;
@@ -577,7 +573,6 @@ impl Tool for SubWorkerSpawnTool {
)));
}
let body = serde_json::json!({
"expected_head_commit": expected_head_commit,
"child_session_id": child_session_id,
"capability_token": capability_token,
});
@@ -1043,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","expected_head_commit":"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","expected_head_commit":"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","expected_head_commit":"V1"}
"review":{"ticket_id":"T1"}
}))
.unwrap();
assert!(validate_reviewer_handoff(&writable).is_err());
crates/worker/src/worker.rs
-6
View File
@@ -248,7 +248,6 @@ pub trait WorkspaceClient: std::fmt::Debug + Send + Sync {
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReviewerContext {
pub ticket_id: String,
pub expected_head_commit: String,
}
#[derive(Debug)]
@@ -306,10 +305,6 @@ impl WorkspaceClient for ReviewerChildWorkspaceClient {
"review submission body must be an object".to_string(),
)
})?;
object.insert(
"expected_head_commit".to_string(),
serde_json::Value::String(self.context.expected_head_commit.clone()),
);
object.insert(
"capability_token".to_string(),
serde_json::Value::String(self.capability_token.clone()),
@@ -452,7 +447,6 @@ mod reviewer_client_tests {
inner,
ReviewerContext {
ticket_id: "T1".into(),
expected_head_commit: "head".into(),
},
"secret".into(),
);
crates/workspace-server/src/server.rs
+240 -542
View File
@@ -94,8 +94,8 @@ use crate::observation::{
use crate::profile_settings::UpdateWorkspaceMetadataRequest;
use crate::records::{ObjectiveDetail, ProjectRecordList, TicketDetail};
use crate::repositories::{
ConfiguredRepository, MergeTargetObservation, RepositoryListProjection, RepositoryLogRead,
RepositoryLookupError, RepositoryRegistryReader, RepositorySummary,
ConfiguredRepository, RepositoryListProjection, RepositoryLogRead, RepositoryLookupError,
RepositoryRegistryReader, RepositorySummary,
};
use crate::resource_broker::BackendResourceBroker;
use crate::runtime_subscription::RuntimeSubscriptionBroker;
@@ -1324,8 +1324,12 @@ pub fn build_router(api: WorkspaceApi) -> Router {
get(scoped_merge_request_readiness),
)
.route(
"/api/w/{workspace_id}/tickets/{id}/merge-request/review-requests",
post(scoped_request_merge_request_review),
"/api/w/{workspace_id}/tickets/{id}/merge-request/thread",
get(scoped_merge_request_thread),
)
.route(
"/api/w/{workspace_id}/tickets/{id}/merge-request/repair-source",
post(scoped_repair_merge_request_selector),
)
.route(
"/api/w/{workspace_id}/internal/reviewer-child-sessions",
@@ -1339,18 +1343,15 @@ pub fn build_router(api: WorkspaceApi) -> Router {
"/api/w/{workspace_id}/tickets/{id}/merge-request/reviews",
post(scoped_submit_merge_request_review),
)
.route(
"/api/w/{workspace_id}/tickets/{id}/merge-request/reviews/revoke",
post(scoped_revoke_merge_request_review),
)
.route(
"/api/w/{workspace_id}/tickets/{id}/merge-request/complete",
post(scoped_complete_merge_request),
)
.route(
"/api/w/{workspace_id}/tickets/{id}/merge-request/close",
post(scoped_close_merge_request),
)
.route(
"/api/w/{workspace_id}/tickets/{id}/merge-request/reopen",
post(scoped_reopen_merge_request),
)
.route(
"/api/w/{workspace_id}/tickets/{id}/workflow/close",
post(scoped_close_ticket_record),
@@ -3583,23 +3584,28 @@ struct OpenMergeRequestRequest {
repository_id: String,
selector_from: String,
selector_to: String,
base_commit: String,
head_commit: String,
#[serde(default)]
changed_paths: Vec<String>,
#[serde(default)]
summary: String,
}
#[derive(Debug, serde::Deserialize)]
struct RequestMergeRequestReviewRequest {
expected_head_commit: String,
base_commit: String,
head_commit: String,
#[serde(default)]
changed_paths: Vec<String>,
#[serde(default)]
summary: String,
struct RepairMergeRequestSelectorRequest {
selector_from: String,
reason: String,
explicit_confirmation: bool,
}
#[derive(Debug, serde::Deserialize)]
struct RevokeMergeRequestReviewRequest {
review_event_id: String,
reason: String,
explicit_confirmation: bool,
}
#[derive(Debug, serde::Deserialize)]
struct MergeRequestThreadQuery {
after: Option<u64>,
limit: Option<usize>,
}
#[derive(Debug, serde::Deserialize)]
@@ -3609,14 +3615,12 @@ struct RegisterReviewerChildSessionRequest {
#[derive(Debug, serde::Deserialize)]
struct RegisterMergeRequestReviewCapabilityRequest {
expected_head_commit: String,
child_session_id: String,
capability_token: String,
}
#[derive(Debug, serde::Deserialize)]
struct SubmitMergeRequestReviewRequest {
expected_head_commit: String,
capability_token: String,
decision: merge_request::ReviewDecision,
#[serde(default)]
@@ -3628,21 +3632,13 @@ struct SubmitMergeRequestReviewRequest {
#[derive(Debug, serde::Deserialize)]
struct CompleteMergeRequestRequest {
operation_id: String,
expected_head_commit: String,
target_commit: String,
source_commit: String,
result_commit: String,
approval_event_id: String,
target_ref_before: String,
target_ref_after: String,
strategy: merge_request::MergeStrategy,
resolution: merge_request::ConflictResolution,
}
#[derive(Debug, serde::Deserialize)]
struct MergeRequestStateRequest {
#[serde(default)]
body: String,
explicit_confirmation: bool,
}
fn parse_workspace_id(value: &str) -> ApiResult<String> {
if value.trim().is_empty() {
return Err(Error::InvalidInput("workspace_id must not be empty".to_string()).into());
@@ -3699,26 +3695,6 @@ impl merge_request::RepositorySource for MergeRequestRepositorySource {
}
Ok(self.reader.summary(repository_id).is_ok())
}
fn is_ancestor(
&self,
workspace_id: &str,
repository_id: &str,
ancestor: &str,
descendant: &str,
) -> std::result::Result<bool, String> {
if workspace_id != self.workspace_id {
return Ok(false);
}
match self
.reader
.ensure_ancestor(repository_id, ancestor, descendant)
{
Ok(()) => Ok(true),
Err(RepositoryLookupError::InvalidCommitRelation { .. }) => Ok(false),
Err(error) => Err(format!("{error:?}")),
}
}
}
fn merge_request_store(
@@ -3747,65 +3723,36 @@ fn repository_merge_evidence_error(error: RepositoryLookupError) -> ApiError {
.into()
}
fn validate_open_merge_request_evidence(
api: &WorkspaceApi,
ticket_id: &str,
repository_id: &str,
base_commit: &str,
head_commit: &str,
) -> ApiResult<(String, String, MergeTargetObservation)> {
let ticket = browser_ticket_backend(api)?
.show(TicketIdOrSlug::Id(ticket_id.into()))
.map_err(Error::from)?;
if ticket.meta.repository_id.as_deref() != Some(repository_id) {
return Err(Error::InvalidInput(
"Merge Request repository must match the authoritative Ticket target".into(),
)
.into());
}
let reader = api.repository_reader();
let target = reader
.observe_merge_target(repository_id, ticket.meta.ref_selector.as_deref())
.map_err(repository_merge_evidence_error)?;
let base = reader
.observe_commit(repository_id, base_commit)
.map_err(repository_merge_evidence_error)?;
let source = reader
.observe_commit(repository_id, head_commit)
.map_err(repository_merge_evidence_error)?;
reader
.ensure_ancestor(repository_id, &base.commit, &source.commit)
.map_err(repository_merge_evidence_error)?;
Ok((base.commit, source.commit, target))
}
fn validate_revision_evidence(
api: &WorkspaceApi,
repository_id: &str,
base_commit: &str,
head_commit: &str,
) -> ApiResult<(String, String)> {
let reader = api.repository_reader();
let base = reader
.observe_commit(repository_id, base_commit)
.map_err(repository_merge_evidence_error)?;
let source = reader
.observe_commit(repository_id, head_commit)
.map_err(repository_merge_evidence_error)?;
reader
.ensure_ancestor(repository_id, &base.commit, &source.commit)
.map_err(repository_merge_evidence_error)?;
Ok((base.commit, source.commit))
}
async fn scoped_show_merge_request(
State(api): State<WorkspaceApi>,
AxumPath((workspace_id, ticket_id)): AxumPath<(String, String)>,
) -> ApiResult<Json<merge_request::MergeRequest>> {
) -> ApiResult<Json<serde_json::Value>> {
let workspace_id = parse_workspace_id(&workspace_id)?;
Ok(Json(
merge_request_store(&api, &workspace_id)?.get(&workspace_id, &ticket_id)?,
))
let mr = merge_request_store(&api, &workspace_id)?.get(&workspace_id, &ticket_id)?;
let reader = api.repository_reader();
let observed_at = Utc::now().to_rfc3339();
let source = match mr.selector_from.as_deref() {
Some(selector) => match reader.observe_merge_target(&mr.repository_id, Some(selector)) {
Ok(value) => {
serde_json::json!({"status":"known","ref":value.commit,"observed_at":observed_at})
}
Err(_) => serde_json::json!({"status":"unknown","observed_at":observed_at}),
},
None => serde_json::json!({"status":"requires_repair","observed_at":observed_at}),
};
let target = match reader.observe_merge_target(&mr.repository_id, Some(&mr.selector_to)) {
Ok(value) => {
serde_json::json!({"status":"known","ref":value.commit,"observed_at":observed_at})
}
Err(_) => serde_json::json!({"status":"unknown","observed_at":observed_at}),
};
let mut response =
serde_json::to_value(mr).map_err(|error| Error::InvalidInput(error.to_string()))?;
if let Some(object) = response.as_object_mut() {
object.insert("source".into(), source);
object.insert("target".into(), target);
}
Ok(Json(response))
}
async fn scoped_merge_request_readiness(
@@ -3815,9 +3762,15 @@ async fn scoped_merge_request_readiness(
let workspace_id = parse_workspace_id(&workspace_id)?;
let store = merge_request_store(&api, &workspace_id)?;
let mr = store.get(&workspace_id, &ticket_id)?;
let current_subject_ref = mr.selector_from.as_deref().and_then(|selector| {
api.repository_reader()
.observe_merge_target(&mr.repository_id, Some(selector))
.ok()
.map(|v| v.commit)
});
Ok(Json(store.readiness(merge_request::ReadinessCheck {
ticket_id,
expected_head_commit: None,
current_subject_ref,
auth: merge_request::MergeRequestAuth {
workspace_id,
repository_id: mr.repository_id,
@@ -3847,117 +3800,94 @@ async fn scoped_open_merge_request(
|| assignment.worker.worker_id != source.worker_id
{
return Err(Error::TicketAssignmentConflict(
"authenticated Worker is not the current Ticket assignee".into(),
"authenticated Worker is not current assignee".into(),
)
.into());
}
let (base_commit, source_commit, target) = validate_open_merge_request_evidence(
&api,
&ticket_id,
&input.repository_id,
&input.base_commit,
&input.head_commit,
)?;
if input.selector_to != target.selector {
let ticket = browser_ticket_backend(&api)?
.show(TicketIdOrSlug::Id(ticket_id.clone().into()))
.map_err(Error::from)?;
if ticket.meta.repository_id.as_deref() != Some(input.repository_id.as_str())
|| ticket.meta.ref_selector.as_deref() != Some(input.selector_to.as_str())
{
return Err(Error::InvalidInput(
"selector_to must match the authoritative Ticket target selector".into(),
"selectors must match the authoritative Ticket repository target".into(),
)
.into());
}
let observed_source = api
.repository_reader()
let reader = api.repository_reader();
reader
.observe_merge_target(&input.repository_id, Some(&input.selector_from))
.map_err(repository_merge_evidence_error)?;
if observed_source.commit != source_commit {
return Err(Error::InvalidInput(
"selector_from does not resolve to the nominated head commit".into(),
)
.into());
}
let mr = merge_request_store(&api, &workspace_id)?.open_merge_request(
merge_request::OpenMergeRequest {
merge_request_id: Uuid::now_v7().to_string(),
ticket_id,
repository_id: input.repository_id.clone(),
selector_from: input.selector_from,
selector_to: input.selector_to,
request: merge_request::RequestForReview {
base_commit,
head_commit: source_commit,
changed_paths: input.changed_paths,
reader
.observe_merge_target(&input.repository_id, Some(&input.selector_to))
.map_err(repository_merge_evidence_error)?;
Ok(Json(
merge_request_store(&api, &workspace_id)?.open_merge_request(
merge_request::OpenMergeRequest {
merge_request_id: Uuid::now_v7().to_string(),
ticket_id,
repository_id: input.repository_id.clone(),
selector_from: input.selector_from,
selector_to: input.selector_to,
summary: input.summary,
auth: merge_request::MergeRequestAuth {
workspace_id,
repository_id: input.repository_id,
runtime_id: source.runtime_id,
worker_id: source.worker_id,
assignment_id: assignment.assignment_id,
},
now: Utc::now(),
},
auth: merge_request::MergeRequestAuth {
workspace_id,
repository_id: input.repository_id,
runtime_id: source.runtime_id,
worker_id: source.worker_id,
assignment_id: assignment.assignment_id,
},
now: Utc::now(),
},
)?;
Ok(Json(mr))
)?,
))
}
async fn scoped_request_merge_request_review(
async fn scoped_merge_request_thread(
State(api): State<WorkspaceApi>,
AxumPath((workspace_id, ticket_id)): AxumPath<(String, String)>,
Query(query): Query<MergeRequestThreadQuery>,
) -> ApiResult<Json<Vec<merge_request::MergeRequestThreadEvent>>> {
let workspace_id = parse_workspace_id(&workspace_id)?;
Ok(Json(
merge_request_store(&api, &workspace_id)?.thread_page(
&workspace_id,
&ticket_id,
query.after,
query.limit.unwrap_or(100),
)?,
))
}
async fn scoped_repair_merge_request_selector(
State(api): State<WorkspaceApi>,
headers: HeaderMap,
AxumPath((workspace_id, ticket_id)): AxumPath<(String, String)>,
Json(input): Json<RequestMergeRequestReviewRequest>,
) -> ApiResult<Json<merge_request::RequestForReviewEvent>> {
Json(input): Json<RepairMergeRequestSelectorRequest>,
) -> ApiResult<Json<merge_request::MergeRequest>> {
let workspace_id = parse_workspace_id(&workspace_id)?;
require_workspace_access(&workspace_id, &api)?;
let source = authenticate_worker_mutation_source(&api, &workspace_id, &headers)?;
let assignment = api
.store
.get_current_ticket_worker_assignment(&workspace_id, &ticket_id)?
.ok_or_else(|| {
Error::TicketAssignmentConflict("Ticket has no current assigned Coder".into())
})?;
if assignment.worker.runtime_id != source.runtime_id
|| assignment.worker.worker_id != source.worker_id
{
return Err(Error::TicketAssignmentConflict(
"authenticated Worker is not the current Ticket assignee".into(),
)
.into());
reject_non_browser_reopen_auth(&headers)?;
let _actor = require_actor(&api, &headers).await?;
if !input.explicit_confirmation {
return Err(Error::BrowserReopenConfirmationRequired.into());
}
let store = merge_request_store(&api, &workspace_id)?;
let current = store.get(&workspace_id, &ticket_id)?;
let (base_commit, source_commit) = validate_revision_evidence(
&api,
&current.repository_id,
&input.base_commit,
&input.head_commit,
)?;
let observed_source = api
.repository_reader()
.observe_merge_target(&current.repository_id, Some(&current.selector_from))
let mr = store.get(&workspace_id, &ticket_id)?;
api.repository_reader()
.observe_merge_target(&mr.repository_id, Some(&input.selector_from))
.map_err(repository_merge_evidence_error)?;
if observed_source.commit != source_commit {
return Err(Error::InvalidInput(
"selector_from does not resolve to the nominated head commit".into(),
)
.into());
}
Ok(Json(store.request_review(
merge_request::RequestMergeRequestReview {
Ok(Json(store.repair_selector_from(
merge_request::RepairSelectorFrom {
workspace_id,
ticket_id,
expected_head_commit: input.expected_head_commit,
request: merge_request::RequestForReview {
base_commit,
head_commit: source_commit,
changed_paths: input.changed_paths,
summary: input.summary,
},
auth: merge_request::MergeRequestAuth {
workspace_id,
repository_id: current.repository_id,
runtime_id: source.runtime_id,
worker_id: source.worker_id,
assignment_id: assignment.assignment_id,
selector_from: input.selector_from,
repaired_by: merge_request::WorkerIdentity {
runtime_id: "browser".into(),
worker_id: "authenticated-user".into(),
},
reason: input.reason,
now: Utc::now(),
},
)?))
@@ -4004,20 +3934,29 @@ async fn scoped_register_merge_request_review_capability(
|| assignment.worker.worker_id != source.worker_id
{
return Err(Error::TicketAssignmentConflict(
"authenticated Worker is not the current Ticket assignee".into(),
"authenticated Worker is not current assignee".into(),
)
.into());
}
let store = merge_request_store(&api, &workspace_id)?;
let current = store.get(&workspace_id, &ticket_id)?;
store.register_review_capability(merge_request::RegisterReviewCapability {
let mr = store.get(&workspace_id, &ticket_id)?;
let selector = mr
.selector_from
.as_deref()
.ok_or_else(|| Error::InvalidInput("selector_from requires repair".into()))?;
let subject_ref = api
.repository_reader()
.observe_merge_target(&mr.repository_id, Some(selector))
.map_err(repository_merge_evidence_error)?
.commit;
store.request_review(merge_request::RequestMergeRequestReview {
ticket_id,
expected_head_commit: input.expected_head_commit,
subject_ref,
child_session_id: input.child_session_id,
capability_token: input.capability_token,
auth: merge_request::MergeRequestAuth {
workspace_id,
repository_id: current.repository_id,
repository_id: mr.repository_id,
runtime_id: source.runtime_id,
worker_id: source.worker_id,
assignment_id: assignment.assignment_id,
@@ -4033,19 +3972,73 @@ async fn scoped_submit_merge_request_review(
Json(input): Json<SubmitMergeRequestReviewRequest>,
) -> ApiResult<Json<merge_request::ReviewEvent>> {
let workspace_id = parse_workspace_id(&workspace_id)?;
Ok(Json(
merge_request_store(&api, &workspace_id)?.submit_review(
merge_request::SubmitMergeRequestReview {
ticket_id,
expected_head_commit: input.expected_head_commit,
capability_token: input.capability_token,
decision: input.decision,
body: input.body,
findings: input.findings,
now: Utc::now(),
let store = merge_request_store(&api, &workspace_id)?;
let mr = store.get(&workspace_id, &ticket_id)?;
let selector = mr
.selector_from
.as_deref()
.ok_or_else(|| Error::InvalidInput("selector_from requires repair".into()))?;
let current_subject_ref = api
.repository_reader()
.observe_merge_target(&mr.repository_id, Some(selector))
.map_err(repository_merge_evidence_error)?
.commit;
Ok(Json(store.submit_review(
merge_request::SubmitMergeRequestReview {
ticket_id,
current_subject_ref,
capability_token: input.capability_token,
decision: input.decision,
body: input.body,
findings: input.findings,
now: Utc::now(),
},
)?))
}
async fn scoped_revoke_merge_request_review(
State(api): State<WorkspaceApi>,
headers: HeaderMap,
AxumPath((workspace_id, ticket_id)): AxumPath<(String, String)>,
Json(input): Json<RevokeMergeRequestReviewRequest>,
) -> ApiResult<Json<merge_request::ReviewRevokedEvent>> {
let workspace_id = parse_workspace_id(&workspace_id)?;
require_workspace_access(&workspace_id, &api)?;
if !input.explicit_confirmation {
return Err(Error::BrowserReopenConfirmationRequired.into());
}
let source = authenticate_worker_mutation_source(&api, &workspace_id, &headers)?;
let assignment = api
.store
.get_current_ticket_worker_assignment(&workspace_id, &ticket_id)?
.ok_or_else(|| {
Error::TicketAssignmentConflict("Ticket has no current assigned Coder".into())
})?;
if assignment.worker.runtime_id != source.runtime_id
|| assignment.worker.worker_id != source.worker_id
{
return Err(Error::TicketAssignmentConflict(
"authenticated Worker is not current assignee".into(),
)
.into());
}
let store = merge_request_store(&api, &workspace_id)?;
let mr = store.get(&workspace_id, &ticket_id)?;
Ok(Json(store.revoke_review(
merge_request::RevokeMergeRequestReview {
ticket_id,
review_event_id: input.review_event_id,
reason: input.reason,
auth: merge_request::MergeRequestAuth {
workspace_id,
repository_id: mr.repository_id,
runtime_id: source.runtime_id,
worker_id: source.worker_id,
assignment_id: assignment.assignment_id,
},
)?,
))
now: Utc::now(),
},
)?))
}
async fn scoped_complete_merge_request(
@@ -4066,48 +4059,42 @@ async fn scoped_complete_merge_request(
})?;
let store = merge_request_store(&api, &workspace_id)?;
let mr = store.get(&workspace_id, &ticket_id)?;
let current_request = mr.current_request().ok_or_else(|| {
Error::InvalidInput("Merge Request has no current RequestForReview event".into())
})?;
if current_request.head_commit != input.expected_head_commit
|| input.source_commit != current_request.head_commit
{
let selector = mr
.selector_from
.as_deref()
.ok_or_else(|| Error::InvalidInput("selector_from requires repair".into()))?;
let repositories = api.repository_reader();
let current_source_ref = repositories
.observe_merge_target(&mr.repository_id, Some(selector))
.map_err(repository_merge_evidence_error)?
.commit;
let observed = repositories
.observe_merge_target(&mr.repository_id, Some(&mr.selector_to))
.map_err(repository_merge_evidence_error)?;
if observed.commit != input.target_ref_before && observed.commit != input.target_ref_after {
return Err(Error::InvalidInput(
"source commit does not match the current review request".into(),
"target selector moved outside completion evidence".into(),
)
.into());
}
let repositories = api.repository_reader();
let observed_target = repositories
.observe_merge_target(&mr.repository_id, Some(&mr.selector_to))
.map_err(repository_merge_evidence_error)?;
if observed_target.commit != input.target_commit
&& observed_target.commit != input.result_commit
{
return Err(Error::InvalidInput(format!(
"Merge Request target moved: expected {}, observed {}",
input.target_commit, observed_target.commit
))
.into());
}
let target_was_already_updated = observed_target.commit == input.result_commit;
if !target_was_already_updated {
let already = observed.commit == input.target_ref_after;
if !already {
repositories
.update_merge_target(
&mr.repository_id,
&mr.selector_to,
&input.target_commit,
&input.result_commit,
&input.target_ref_before,
&input.target_ref_after,
)
.map_err(repository_merge_evidence_error)?;
.map_err(repository_merge_evidence_error)?
}
let completion = merge_request::CompleteMergeRequest {
ticket_id,
expected_head_commit: input.expected_head_commit,
operation_id: input.operation_id,
target_commit: input.target_commit.clone(),
source_commit: input.source_commit,
result_commit: input.result_commit.clone(),
approval_event_id: input.approval_event_id,
current_subject_ref: current_source_ref,
target_ref_before: input.target_ref_before.clone(),
target_ref_after: input.target_ref_after.clone(),
strategy: input.strategy,
resolution: input.resolution,
auth: merge_request::MergeRequestAuth {
@@ -4120,75 +4107,21 @@ async fn scoped_complete_merge_request(
now: Utc::now(),
};
match store.complete(completion) {
Ok(event) => Ok(Json(event)),
Err(error) => {
if !target_was_already_updated {
Ok(v) => Ok(Json(v)),
Err(e) => {
if !already {
let _ = repositories.update_merge_target(
&mr.repository_id,
&mr.selector_to,
&input.result_commit,
&input.target_commit,
&input.target_ref_after,
&input.target_ref_before,
);
}
Err(error.into())
Err(e.into())
}
}
}
async fn scoped_close_merge_request(
State(api): State<WorkspaceApi>,
headers: HeaderMap,
AxumPath((workspace_id, ticket_id)): AxumPath<(String, String)>,
Json(input): Json<MergeRequestStateRequest>,
) -> ApiResult<Json<merge_request::MergeRequest>> {
scoped_change_merge_request_state(api, headers, workspace_id, ticket_id, input, false).await
}
async fn scoped_reopen_merge_request(
State(api): State<WorkspaceApi>,
headers: HeaderMap,
AxumPath((workspace_id, ticket_id)): AxumPath<(String, String)>,
Json(input): Json<MergeRequestStateRequest>,
) -> ApiResult<Json<merge_request::MergeRequest>> {
scoped_change_merge_request_state(api, headers, workspace_id, ticket_id, input, true).await
}
async fn scoped_change_merge_request_state(
api: WorkspaceApi,
headers: HeaderMap,
workspace_id: String,
ticket_id: String,
input: MergeRequestStateRequest,
reopen: bool,
) -> ApiResult<Json<merge_request::MergeRequest>> {
let workspace_id = parse_workspace_id(&workspace_id)?;
require_workspace_access(&workspace_id, &api)?;
reject_non_browser_reopen_auth(&headers)?;
let _actor = require_actor(&api, &headers).await?;
if !input.explicit_confirmation {
return Err(Error::BrowserReopenConfirmationRequired.into());
}
let store = merge_request_store(&api, &workspace_id)?;
let mr = store.get(&workspace_id, &ticket_id)?;
let operation = merge_request::ChangeMergeRequestState {
ticket_id,
body: input.body,
auth: merge_request::MergeRequestAuth {
workspace_id,
repository_id: mr.repository_id,
runtime_id: "browser".into(),
worker_id: "authenticated-user".into(),
assignment_id: String::new(),
},
now: Utc::now(),
};
Ok(Json(if reopen {
store.reopen(operation)?
} else {
store.close(operation)?
}))
}
fn reject_non_browser_reopen_auth(headers: &HeaderMap) -> Result<()> {
if headers.contains_key("authorization") {
return Err(Error::BrowserReopenConfirmationRequired);
@@ -12738,241 +12671,6 @@ mod tests {
));
}
#[tokio::test]
async fn merge_request_completion_endpoint_rejects_coder_and_accepts_orchestrator() {
let workspace = tempfile::tempdir().unwrap();
init_clean_git_workspace(workspace.path());
let git_value = |args: &[&str]| {
let output = std::process::Command::new("git")
.arg("-C")
.arg(workspace.path())
.args(args)
.output()
.unwrap();
assert!(output.status.success());
String::from_utf8(output.stdout).unwrap().trim().to_string()
};
let target_commit = git_value(&["rev-parse", "HEAD"]);
let target_ref = git_value(&["symbolic-ref", "HEAD"]);
std::fs::write(workspace.path().join("README.md"), "merge source\n").unwrap();
for args in [&["add", "README.md"][..], &["commit", "-m", "source"][..]] {
assert!(
std::process::Command::new("git")
.arg("-C")
.arg(workspace.path())
.args(args)
.status()
.unwrap()
.success()
);
}
let source_commit = git_value(&["rev-parse", "HEAD"]);
assert!(
std::process::Command::new("git")
.arg("-C")
.arg(workspace.path())
.args(["reset", "--hard", &target_commit])
.status()
.unwrap()
.success()
);
let api = test_api(workspace.path()).await;
let workspace_id = api.config.workspace_id.clone();
let backend = browser_ticket_backend(&api).unwrap();
let mut input = ticket::NewTicket::new("Orchestrator completion authority");
input.workflow_state = Some(TicketWorkflowState::InProgress);
let ticket = backend.create(input).unwrap();
let Json(coder) = create_workspace_worker(
State(api.clone()),
HeaderMap::new(),
Json(CreateWorkspaceWorkerRequest {
runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(),
display_name: "Assigned Coder".to_string(),
profile: Some("builtin:coder".to_string()),
ticket_assignment: Some(CreateWorkspaceWorkerTicketAssignmentRequest {
ticket_id: ticket.id.clone(),
operation_id: "completion-coder-assignment".to_string(),
}),
initial_submit: vec![Segment::Flow {
selector: "builtin:coder-review".to_string(),
}],
working_directory: None,
control_operation_id: None,
resolved_control_operation: None,
}),
)
.await
.unwrap();
let assignment = api
.store
.get_current_ticket_worker_assignment(&workspace_id, &ticket.id)
.unwrap()
.unwrap();
let mr_store = merge_request_store(&api, &workspace_id).unwrap();
mr_store
.open_merge_request(merge_request::OpenMergeRequest {
merge_request_id: "MR-server-completion".into(),
ticket_id: ticket.id.clone(),
repository_id: TEST_REPOSITORY_ID.into(),
selector_from: source_commit.clone(),
selector_to: target_ref.clone(),
request: merge_request::RequestForReview {
base_commit: target_commit.clone(),
head_commit: source_commit.clone(),
changed_paths: vec!["src/lib.rs".into()],
summary: "approved candidate".into(),
},
auth: merge_request::MergeRequestAuth {
workspace_id: workspace_id.clone(),
repository_id: TEST_REPOSITORY_ID.into(),
runtime_id: coder.worker_ref.runtime_id.clone(),
worker_id: coder.worker_ref.worker_id.clone(),
assignment_id: assignment.assignment_id.clone(),
},
now: Utc::now(),
})
.unwrap();
mr_store
.register_reviewer_child_session(merge_request::RegisterReviewerChildSession {
workspace_id: workspace_id.clone(),
parent_runtime_id: coder.worker_ref.runtime_id.clone(),
parent_worker_id: coder.worker_ref.worker_id.clone(),
child_session_id: "reviewer-child".into(),
reviewer_profile: "builtin:reviewer".into(),
now: Utc::now(),
})
.unwrap();
mr_store
.register_review_capability(merge_request::RegisterReviewCapability {
ticket_id: ticket.id.clone(),
expected_head_commit: source_commit.clone(),
child_session_id: "reviewer-child".into(),
capability_token: "review-token".into(),
auth: merge_request::MergeRequestAuth {
workspace_id: workspace_id.clone(),
repository_id: TEST_REPOSITORY_ID.into(),
runtime_id: coder.worker_ref.runtime_id.clone(),
worker_id: coder.worker_ref.worker_id.clone(),
assignment_id: assignment.assignment_id.clone(),
},
now: Utc::now(),
})
.unwrap();
mr_store
.submit_review(merge_request::SubmitMergeRequestReview {
ticket_id: ticket.id.clone(),
expected_head_commit: source_commit.clone(),
capability_token: "review-token".into(),
decision: merge_request::ReviewDecision::Approve,
body: "approved".into(),
findings: Vec::new(),
now: Utc::now(),
})
.unwrap();
let worker_headers = |worker: &RuntimeWorkerRef| {
let mut headers = HeaderMap::new();
headers.insert(
"x-yoi-runtime-id",
axum::http::HeaderValue::from_str(&worker.runtime_id).unwrap(),
);
headers.insert(
"x-yoi-worker-id",
axum::http::HeaderValue::from_str(&worker.worker_id).unwrap(),
);
headers
};
let request = || CompleteMergeRequestRequest {
operation_id: "complete-operation".into(),
expected_head_commit: source_commit.clone(),
target_commit: target_commit.clone(),
source_commit: source_commit.clone(),
result_commit: source_commit.clone(),
strategy: merge_request::MergeStrategy::FastForward,
resolution: merge_request::ConflictResolution::None,
};
let coder_error = scoped_complete_merge_request(
State(api.clone()),
worker_headers(&coder.worker_ref),
AxumPath((workspace_id.clone(), ticket.id.clone())),
Json(request()),
)
.await
.unwrap_err();
assert!(matches!(
coder_error.error,
Error::TicketAssignmentConflict(_)
));
let Json(started) = scoped_start_workspace_orchestrator(
State(api.clone()),
AxumPath(ScopedWorkspacePath {
workspace_id: workspace_id.clone(),
}),
)
.await
.unwrap();
let orchestrator = started.worker.unwrap().worker;
let Json(completed) = scoped_complete_merge_request(
State(api.clone()),
worker_headers(&orchestrator),
AxumPath((workspace_id, ticket.id.clone())),
Json(request()),
)
.await
.unwrap();
assert_eq!(completed.result_commit, source_commit);
assert_eq!(
backend
.show(ticket.id.clone().into())
.unwrap()
.meta
.workflow_state,
TicketWorkflowState::Done
);
assert_eq!(
api.repository_reader()
.observe_merge_target(TEST_REPOSITORY_ID, Some(&target_ref))
.unwrap()
.commit,
source_commit
);
let mr = mr_store.get(&api.config.workspace_id, &ticket.id).unwrap();
assert_eq!(mr.state, merge_request::MergeRequestState::Merged);
let merge = mr.thread.iter().find_map(|event| match event {
merge_request::MergeRequestThreadEvent::Merge(value) => Some(value),
_ => None,
});
assert_eq!(
merge.map(|value| value.result_commit.as_str()),
Some(source_commit.as_str())
);
let Json(replayed) = scoped_complete_merge_request(
State(api.clone()),
worker_headers(&orchestrator),
AxumPath((api.config.workspace_id.clone(), ticket.id.clone())),
Json(request()),
)
.await
.unwrap();
assert_eq!(replayed.result_commit, source_commit);
let conn = rusqlite::Connection::open(&api.config.database_path).unwrap();
let actor: String = conn
.query_row(
"SELECT author FROM typed_ticket_events WHERE workspace_id=?1 AND ticket_id=?2 AND kind='state_changed'",
rusqlite::params![api.config.workspace_id, ticket.id],
|row| row.get(0),
)
.unwrap();
assert_eq!(
actor,
format!(
"worker:{}:{}",
orchestrator.runtime_id, orchestrator.worker_id
)
);
}
#[tokio::test]
async fn production_profile_backend_launches_and_restores_workspace_orchestrator() {
let workspace = tempfile::tempdir().unwrap();