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

merge-request: finalize one guarded merge outcome

Browse Source
This commit is contained in:
Hare
2026-08-16 21:03:30 +09:00
parent 1479148f84
commit 5ea2792df7
9 changed files with 846 additions and 1519 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/merge-request/src/lib.rs
+364 -581
View File
File diff suppressed because it is too large Load Diff
crates/merge-request/tests/store.rs
+160 -642
View File
@@ -1,5 +1,7 @@
use merge_request::*;
use rusqlite::{Connection, params};
use std::sync::{Arc, Barrier};
use std::thread;
use tempfile::TempDir;
fn setup() -> (TempDir, SqliteMergeRequestStore) {
@@ -38,6 +40,7 @@ fn setup() -> (TempDir, SqliteMergeRequestStore) {
let store = SqliteMergeRequestStore::open(&path, "ws-a").unwrap();
(dir, store)
}
fn revision(id: &str, ordinal: u64, head: &str) -> MergeRequestRevision {
MergeRequestRevision {
revision_id: id.into(),
@@ -51,653 +54,199 @@ fn revision(id: &str, ordinal: u64, head: &str) -> MergeRequestRevision {
created_at: format!("t{ordinal}"),
}
}
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: "develop".into(),
revision: revision("V1", 1, "h1"),
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, id: &str, revision: &str, token: &str, child: &str) {
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.into(),
now: "t".into(),
})
.unwrap();
store
.register_review_attempt(RegisterReviewAttempt {
attempt_id: id.into(),
ticket_id: "T1".into(),
revision_id: revision.into(),
merge_result_id: None,
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: "t".into(),
})
.unwrap();
}
fn review(
store: &SqliteMergeRequestStore,
revision: &str,
token: &str,
decision: ReviewDecision,
) -> Result<MergeRequestReview> {
store.submit_review(SubmitReview {
ticket_id: "T1".into(),
revision_id: revision.into(),
merge_result_id: None,
capability_token: token.into(),
decision,
body: "evidence".into(),
findings: vec![],
now: "tr".into(),
})
}
fn record_result(
store: &SqliteMergeRequestStore,
revision: &str,
target: &str,
source: &str,
result: &str,
strategy: MergeStrategy,
resolution: MergeResolution,
operation: &str,
) -> RecordMergeResultOutcome {
store
.record_merge_result(RecordMergeResult {
merge_result_id: format!("result-{operation}"),
ticket_id: "T1".into(),
expected_revision_id: revision.into(),
target_commit: target.into(),
source_commit: source.into(),
result_commit: result.into(),
strategy,
resolution,
operation_id: operation.into(),
actor_runtime_id: "runtime-orchestrator".into(),
actor_worker_id: "workspace-orchestrator".into(),
created_at: "2026-07-26T00:00:02Z".into(),
})
.unwrap()
}
#[test]
fn storage_allows_multiple_merge_requests_for_one_ticket() {
let (_dir, store) = setup();
open(&store);
store
.open_merge_request(OpenMergeRequest {
merge_request_id: "MR2".into(),
ticket_id: "T1".into(),
repository_id: "repo".into(),
target_ref_selector: "develop".into(),
revision: revision("V2", 1, "h2"),
authenticated_runtime_id: "R1".into(),
authenticated_worker_id: "W1".into(),
child_session_id: child.clone(),
now: "t2".into(),
})
.unwrap();
let conn = Connection::open(store.db_path()).unwrap();
let count:i64=conn.query_row("SELECT COUNT(*) FROM merge_request_ticket_relations WHERE workspace_id='ws-a' AND ticket_id='T1'",[],|row|row.get(0)).unwrap();
assert_eq!(count, 2);
assert_eq!(
store
.show_for_ticket("T1")
.unwrap()
.unwrap()
.merge_request_id,
"MR2"
);
}
#[test]
fn merge_result_is_idempotent_target_fenced_and_non_ff_reviewed_independently() {
let (_tmp, store) = setup();
open(&store);
attempt(
&store,
"attempt-source",
"V1",
"token-source",
"child-source",
);
review(&store, "V1", "token-source", ReviewDecision::Approve).unwrap();
let first = record_result(
&store,
"V1",
"T0",
"h1",
"h1",
MergeStrategy::FastForward,
MergeResolution::None,
"op-ff",
);
assert!(!first.replayed);
let replay = record_result(
&store,
"V1",
"T0",
"h1",
"h1",
MergeStrategy::FastForward,
MergeResolution::None,
"op-ff",
);
assert!(replay.replayed);
assert_eq!(
first.merge_result.merge_result_id,
replay.merge_result.merge_result_id
);
let ready = store
.readiness_for_ticket_with_target("T1", Some("T0"))
.unwrap();
assert!(ready.ready, "{:?}", ready.blockers);
assert_eq!(ready.merge_result_id.as_deref(), Some("result-op-ff"));
let stale = store
.readiness_for_ticket_with_target("T1", Some("T1"))
.unwrap();
assert!(!stale.ready);
assert!(
stale
.blockers
.iter()
.any(|blocker| blocker.contains("target moved"))
);
let changed = store.record_merge_result(RecordMergeResult {
merge_result_id: "result-conflict".into(),
ticket_id: "T1".into(),
expected_revision_id: "V1".into(),
target_commit: "different".into(),
source_commit: "h1".into(),
result_commit: "h1".into(),
strategy: MergeStrategy::FastForward,
resolution: MergeResolution::None,
operation_id: "op-ff".into(),
actor_runtime_id: "runtime-orchestrator".into(),
actor_worker_id: "workspace-orchestrator".into(),
created_at: "2026-07-26T00:00:03Z".into(),
});
assert!(matches!(
changed,
Err(MergeRequestError::MergeResultOperationConflict)
));
// Multiple valid candidates for the same target are retained as history. The
// most recently recorded candidate is the one explicit final result.
record_result(
&store,
"V1",
"T1",
"h1",
"h1",
MergeStrategy::FastForward,
MergeResolution::None,
"op-same-target-old",
);
record_result(
&store,
"V1",
"T1",
"h1",
"M1",
MergeStrategy::Merge,
MergeResolution::ConflictsResolved,
"op-merge",
);
let pending = store
.readiness_for_ticket_with_target("T1", Some("T1"))
.unwrap();
assert!(!pending.ready);
assert_eq!(
pending.merge_result_review_status,
Some(ReviewStatus::Pending)
);
let candidates = store
.show_for_ticket_with_target("T1", Some("T1"))
.unwrap()
.unwrap();
assert_eq!(candidates.merge_results.len(), 3);
assert_eq!(
candidates
.final_merge_result
.as_ref()
.map(|result| result.merge_result_id.as_str()),
Some("result-op-merge")
);
store
.register_reviewer_child_session(RegisterReviewerChildSession {
parent_runtime_id: "runtime-orchestrator".into(),
parent_worker_id: "workspace-orchestrator".into(),
child_session_id: "child-old-candidate".into(),
now: "2026-07-26T00:00:04Z".into(),
})
.unwrap();
let old_candidate_review = store.register_review_attempt(RegisterReviewAttempt {
attempt_id: "attempt-old-candidate".into(),
ticket_id: "T1".into(),
revision_id: "V1".into(),
merge_result_id: Some("result-op-same-target-old".into()),
parent_assignment_id: "A1".into(),
parent_runtime_id: "runtime-orchestrator".into(),
parent_worker_id: "workspace-orchestrator".into(),
child_session_id: "child-old-candidate".into(),
capability_token: "token-old-candidate".into(),
now: "2026-07-26T00:00:04Z".into(),
});
assert!(matches!(
old_candidate_review,
Err(MergeRequestError::MergeResultNotFinal)
));
store
.register_reviewer_child_session(RegisterReviewerChildSession {
parent_runtime_id: "runtime-orchestrator".into(),
parent_worker_id: "workspace-orchestrator".into(),
child_session_id: "child-merge".into(),
now: "2026-07-26T00:00:04Z".into(),
})
.unwrap();
store
.register_review_attempt(RegisterReviewAttempt {
attempt_id: "attempt-merge".into(),
attempt_id: format!("attempt-{revision}"),
ticket_id: "T1".into(),
revision_id: "V1".into(),
merge_result_id: Some("result-op-merge".into()),
revision_id: revision.into(),
parent_assignment_id: "A1".into(),
parent_runtime_id: "runtime-orchestrator".into(),
parent_worker_id: "workspace-orchestrator".into(),
child_session_id: "child-merge".into(),
capability_token: "token-merge".into(),
now: "2026-07-26T00:00:04Z".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: "V1".into(),
merge_result_id: Some("result-op-merge".into()),
capability_token: "token-merge".into(),
revision_id: revision.into(),
capability_token: token.into(),
decision: ReviewDecision::Approve,
body: "merge evidence is valid".into(),
body: "approved".into(),
findings: vec![],
now: "2026-07-26T00:00:05Z".into(),
now: "t3".into(),
})
.unwrap();
let approved = store
.readiness_for_ticket_with_target("T1", Some("T1"))
.unwrap();
assert!(approved.ready, "{:?}", approved.blockers);
assert_eq!(
approved.merge_result_review_status,
Some(ReviewStatus::Approved)
);
let applied = store
.show_for_ticket_with_target("T1", Some("M1"))
.unwrap()
.unwrap();
assert_eq!(
applied
.final_merge_result
.as_ref()
.map(|result| result.target_status),
Some(MergeResultTargetStatus::Applied)
);
assert!(
store
.readiness_for_ticket_with_target("T1", Some("M1"))
.unwrap()
.ready
);
}
#[test]
fn bounded_context_rejects_oversized_revision_evidence() {
let (_dir, store) = setup();
let mut oversized = revision("V1", 1, "h1");
oversized.changed_paths = (0..=1_000).map(|i| format!("src/{i}.rs")).collect();
let result = store.open_merge_request(OpenMergeRequest {
merge_request_id: "MR1".into(),
fn completion(operation_id: &str) -> CompleteMergeRequest {
CompleteMergeRequest {
operation_id: operation_id.into(),
ticket_id: "T1".into(),
repository_id: "repo".into(),
target_ref_selector: "develop".into(),
revision: oversized,
authenticated_runtime_id: "R1".into(),
authenticated_worker_id: "W1".into(),
now: "t".into(),
});
assert!(matches!(
result,
Err(MergeRequestError::TooLarge {
field: "revision.changed_paths",
..
})
));
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(),
}
}
#[test]
fn v6_legacy_schema_fails_closed_without_archiving() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("legacy.db");
let conn = Connection::open(&path).unwrap();
conn.execute_batch(
"CREATE TABLE merge_request_schema_migrations(version INTEGER PRIMARY KEY,name TEXT NOT NULL,applied_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP);\
INSERT INTO merge_request_schema_migrations(version,name) VALUES(6,'rejected_merge_request_v6');\
CREATE TABLE merge_requests(workspace_id TEXT NOT NULL,merge_request_id TEXT NOT NULL,repository_id TEXT NOT NULL,state TEXT NOT NULL,lifecycle_generation INTEGER NOT NULL,current_revision_id TEXT NOT NULL,created_at TEXT NOT NULL,updated_at TEXT NOT NULL,PRIMARY KEY(workspace_id,merge_request_id));",
).unwrap();
drop(conn);
let error = SqliteMergeRequestStore::open(&path, "ws-a").unwrap_err();
assert!(
error
.to_string()
.contains("unsupported legacy merge request schema version 6")
);
let conn = Connection::open(&path).unwrap();
let version: i64 = conn
.query_row(
"SELECT MAX(version) FROM merge_request_schema_migrations",
[],
|row| row.get(0),
)
.unwrap();
assert_eq!(version, 6);
let original: i64 = conn
.query_row(
"SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='merge_requests'",
[],
|row| row.get(0),
)
.unwrap();
let archived: i64 = conn
.query_row(
"SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name LIKE 'legacy_v6_%'",
[],
|row| row.get(0),
)
.unwrap();
assert_eq!(original, 1);
assert_eq!(archived, 0);
}
#[test]
fn v7_schema_is_rejected_without_mutating_completion_evidence() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("v7.db");
let conn = Connection::open(&path).unwrap();
conn.execute_batch(
"CREATE TABLE merge_request_schema_migrations(version INTEGER PRIMARY KEY,name TEXT NOT NULL,applied_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP);\
INSERT INTO merge_request_schema_migrations(version,name) VALUES(7,'fresh_bounded_context_authority');\
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));\
INSERT INTO typed_tickets VALUES('ws-a','T1','done',1,'t');\
CREATE TABLE merge_request_completion_operations(workspace_id TEXT NOT NULL,operation_id TEXT NOT NULL,ticket_id TEXT NOT NULL,revision_id TEXT NOT NULL,assignment_id TEXT NOT NULL,fingerprint TEXT NOT NULL,status TEXT NOT NULL CHECK(status IN ('pending','completed')),result_ticket_state TEXT,created_at TEXT NOT NULL,updated_at TEXT NOT NULL,PRIMARY KEY(workspace_id,operation_id),FOREIGN KEY(workspace_id,ticket_id) REFERENCES typed_tickets(workspace_id,ticket_id));\
INSERT INTO merge_request_completion_operations VALUES('ws-a','legacy-op','T1','V1','A1','legacy-fingerprint','completed','done','t','t');",
).unwrap();
drop(conn);
let error = SqliteMergeRequestStore::open(&path, "ws-a").unwrap_err();
assert!(
error
.to_string()
.contains("unsupported legacy merge request schema version 7")
);
let conn = Connection::open(&path).unwrap();
let row: (String, String) = conn.query_row(
"SELECT assignment_id,fingerprint FROM merge_request_completion_operations WHERE operation_id='legacy-op'",
[],
|row| Ok((row.get(0)?, row.get(1)?)),
).unwrap();
assert_eq!(row, ("A1".into(), "legacy-fingerprint".into()));
let version: i64 = conn
.query_row(
"SELECT MAX(version) FROM merge_request_schema_migrations",
[],
|row| row.get(0),
)
.unwrap();
assert_eq!(version, 7);
}
#[test]
fn request_changes_new_revision_resets_and_exact_completion_replay_converges() {
fn target_movement_does_not_invalidate_source_revision_approval() {
let (_dir, store) = setup();
open(&store);
attempt(&store, "AT1", "V1", "tok1", "child1");
review(&store, "V1", "tok1", ReviewDecision::RequestChanges).unwrap();
assert_eq!(
store.show_for_ticket("T1").unwrap().unwrap().review_status,
ReviewStatus::ChangesRequested
);
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));
}
store
.add_revision(AddRevision {
ticket_id: "T1".into(),
expected_current_revision_id: "V1".into(),
revision: revision("V2", 2, "h2"),
revision: revision("V2", 2, "head2"),
authenticated_runtime_id: "R1".into(),
authenticated_worker_id: "W1".into(),
now: "t2".into(),
now: "t5".into(),
})
.unwrap();
assert_eq!(
store.show_for_ticket("T1").unwrap().unwrap().review_status,
store.readiness_for_ticket("T1").unwrap().review_status,
ReviewStatus::Pending
);
assert!(review(&store, "V1", "tok1", ReviewDecision::Approve).is_err());
attempt(&store, "AT2", "V2", "tok2", "child2");
review(&store, "V2", "tok2", ReviewDecision::Approve).unwrap();
let missing_result = store.complete(CompleteMergeRequest {
operation_id: "OP-missing-result".into(),
ticket_id: "T1".into(),
expected_revision_id: "V2".into(),
expected_merge_result_id: "missing".into(),
observed_target_commit: "h2".into(),
implementation_assignment_id: "A1".into(),
completion_actor_runtime_id: "OR".into(),
completion_actor_worker_id: "OW".into(),
now: "tc".into(),
});
assert!(matches!(
missing_result,
Err(MergeRequestError::FinalMergeResultMissing)
));
record_result(
&store,
"V2",
"base",
"h2",
"h2",
MergeStrategy::FastForward,
MergeResolution::None,
"final-v2",
);
let not_applied = store.complete(CompleteMergeRequest {
operation_id: "OP-not-applied".into(),
ticket_id: "T1".into(),
expected_revision_id: "V2".into(),
expected_merge_result_id: "result-final-v2".into(),
observed_target_commit: "base".into(),
implementation_assignment_id: "A1".into(),
completion_actor_runtime_id: "OR".into(),
completion_actor_worker_id: "OW".into(),
now: "tc".into(),
});
assert!(matches!(
not_applied,
Err(MergeRequestError::FinalMergeResultNotApplied)
));
let input = CompleteMergeRequest {
operation_id: "OP1".into(),
ticket_id: "T1".into(),
expected_revision_id: "V2".into(),
expected_merge_result_id: "result-final-v2".into(),
observed_target_commit: "h2".into(),
implementation_assignment_id: "A1".into(),
completion_actor_runtime_id: "OR".into(),
completion_actor_worker_id: "OW".into(),
now: "tc".into(),
};
let first = store.complete(input.clone()).unwrap();
}
#[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)
));
}
#[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::Merged
MergeRequestState::Open
);
assert_eq!(
store
.show_for_ticket("T1")
.unwrap()
.unwrap()
.merged_at
.as_deref(),
Some("tc")
);
let replay = store.complete(input).unwrap();
assert!(replay.replayed);
assert_eq!(replay.ticket_state, "done");
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'",
[],
|r| r.get::<_, String>(0)
|row| row.get::<_, String>(0)
)
.unwrap(),
"done"
"inprogress"
);
assert_eq!(
conn.query_row(
"SELECT COUNT(*) FROM typed_ticket_events WHERE workspace_id='ws-a' AND ticket_id='T1'",
[],
|r| r.get::<_, i64>(0)
)
.unwrap(),
1
);
assert_eq!(
conn.query_row(
"SELECT authority_kind || ':' || implementation_assignment_id || ':' || completion_actor_runtime_id || ':' || completion_actor_worker_id FROM merge_request_completion_operations WHERE workspace_id='ws-a' AND operation_id='OP1'",
[],
|r| r.get::<_, String>(0)
)
.unwrap(),
"workspace_orchestrator:A1:OR:OW"
);
assert_eq!(
conn.query_row(
"SELECT author FROM typed_ticket_events WHERE workspace_id='ws-a' AND ticket_id='T1' AND kind='state_changed'",
[],
|r| r.get::<_, String>(0)
)
.unwrap(),
"worker:OR:OW"
);
let authority: String = conn
.query_row(
"SELECT value FROM typed_ticket_event_attributes WHERE workspace_id='ws-a' AND ticket_id='T1' AND key='completion_authority'",
[],
|r| r.get(0),
)
.unwrap();
assert_eq!(authority, "workspace_orchestrator");
}
#[test]
fn spoof_self_approval_replay_and_cross_workspace_are_rejected() {
fn concurrent_completion_converges_on_one_operation() {
let (_dir, store) = setup();
open(&store);
let mut bad = RegisterReviewAttempt {
attempt_id: "bad".into(),
ticket_id: "T1".into(),
revision_id: "V1".into(),
merge_result_id: None,
parent_assignment_id: "A1".into(),
parent_runtime_id: "R1".into(),
parent_worker_id: "W1".into(),
child_session_id: "W1".into(),
capability_token: "bad".into(),
now: "t".into(),
};
assert!(matches!(
store.register_review_attempt(bad.clone()),
Err(MergeRequestError::SelfApproval)
));
bad.child_session_id = "child".into();
assert!(matches!(
store.register_review_attempt(bad),
Err(MergeRequestError::InvalidReviewer)
));
attempt(&store, "AT", "V1", "secret", "child");
assert!(review(&store, "V1", "spoof", ReviewDecision::Approve).is_err());
review(&store, "V1", "secret", ReviewDecision::Approve).unwrap();
assert!(review(&store, "V1", "secret", ReviewDecision::Approve).is_err());
let other = SqliteMergeRequestStore::open_verified(store.db_path(), "ws-b").unwrap();
assert!(other.show_for_ticket("T1").unwrap().is_none());
}
#[test]
fn reopen_resets_approval() {
let (_dir, store) = setup();
open(&store);
attempt(&store, "AT", "V1", "token", "child");
review(&store, "V1", "token", ReviewDecision::Approve).unwrap();
store.close("T1", "V1", "tc").unwrap();
let reopened = store.reopen("T1", "V1", "tr").unwrap();
assert_eq!(reopened.review_status, ReviewStatus::Pending);
}
#[test]
fn concurrent_exact_completion_replays_commit_one_ticket_side_effect() {
let (_dir, store) = setup();
open(&store);
attempt(&store, "AT", "V1", "token", "child");
review(&store, "V1", "token", ReviewDecision::Approve).unwrap();
record_result(
&store,
"V1",
"base",
"h1",
"h1",
MergeStrategy::FastForward,
MergeResolution::None,
"final-concurrent",
);
let input = CompleteMergeRequest {
operation_id: "OP-concurrent".into(),
ticket_id: "T1".into(),
expected_revision_id: "V1".into(),
expected_merge_result_id: "result-final-concurrent".into(),
observed_target_commit: "h1".into(),
implementation_assignment_id: "A1".into(),
completion_actor_runtime_id: "OR".into(),
completion_actor_worker_id: "OW".into(),
now: "t".into(),
};
let left_store = store.clone();
let left_input = input.clone();
let left = std::thread::spawn(move || left_store.complete(left_input));
let right_store = store.clone();
let right = std::thread::spawn(move || right_store.complete(input));
let outcomes = [
left.join().unwrap().unwrap(),
right.join().unwrap().unwrap(),
];
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
@@ -706,75 +255,44 @@ fn concurrent_exact_completion_replays_commit_one_ticket_side_effect() {
outcomes.iter().filter(|outcome| outcome.replayed).count(),
1
);
let conn = Connection::open(store.db_path()).unwrap();
let events: i64 = conn
.query_row(
"SELECT COUNT(*) FROM typed_ticket_events WHERE workspace_id='ws-a' AND ticket_id='T1'",
[],
|row| row.get(0),
)
.unwrap();
assert_eq!(events, 1);
}
#[test]
fn operation_key_mismatch_and_actor_or_assignment_change_are_fenced() {
fn reviewer_attempt_is_bound_to_direct_child_and_current_assignment() {
let (_dir, store) = setup();
open(&store);
attempt(&store, "AT", "V1", "token", "child");
review(&store, "V1", "token", ReviewDecision::Approve).unwrap();
record_result(
&store,
"V1",
"base",
"h1",
"h1",
MergeStrategy::FastForward,
MergeResolution::None,
"final-operation",
);
let mut input = CompleteMergeRequest {
operation_id: "OP".into(),
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(),
expected_revision_id: "V1".into(),
expected_merge_result_id: "result-final-operation".into(),
observed_target_commit: "h1".into(),
implementation_assignment_id: "A1".into(),
completion_actor_runtime_id: "OR".into(),
completion_actor_worker_id: "OW".into(),
now: "t".into(),
};
let conn = Connection::open(store.db_path()).unwrap();
conn.execute(
"UPDATE ticket_current_worker_assignments SET assignment_id='A2',runtime_id='R2',worker_id='W2' WHERE workspace_id='ws-a' AND ticket_id='T1'",
[],
)
.unwrap();
revision_id: "V1".into(),
capability_token: "wrong".into(),
decision: ReviewDecision::Approve,
body: "approved".into(),
findings: vec![],
now: "t3".into(),
});
assert!(matches!(
store.complete(input.clone()),
Err(MergeRequestError::AssignmentMismatch)
));
conn.execute(
"UPDATE ticket_current_worker_assignments SET assignment_id='A1',runtime_id='R1',worker_id='W1' WHERE workspace_id='ws-a' AND ticket_id='T1'",
[],
)
.unwrap();
store.complete(input.clone()).unwrap();
input.completion_actor_worker_id = "other".into();
assert!(matches!(
store.complete(input.clone()),
Err(MergeRequestError::OperationConflict)
));
input.completion_actor_worker_id = "OW".into();
input.implementation_assignment_id = "A2".into();
assert!(matches!(
store.complete(input.clone()),
Err(MergeRequestError::OperationConflict)
));
input.implementation_assignment_id = "A1".into();
input.expected_revision_id = "other".into();
assert!(matches!(
store.complete(input),
Err(MergeRequestError::OperationConflict)
wrong_token,
Err(MergeRequestError::InvalidReviewAttempt)
));
}
crates/worker/src/feature/builtin/merge_request.rs
+10 -38
View File
@@ -12,7 +12,6 @@ pub const MERGE_REQUEST_COMMON_TOOL_NAMES: &[&str] = &[
"MergeRequestReadinessCheck",
"MergeRequestOpen",
"MergeRequestAddRevision",
"MergeRequestRecordMergeResult",
"MergeRequestComplete",
];
pub const MERGE_REQUEST_REVIEW_TOOL_NAME: &str = "MergeRequestReviewSubmit";
@@ -22,7 +21,6 @@ enum Kind {
Readiness,
Open,
AddRevision,
RecordMergeResult,
Complete,
Review,
}
@@ -63,10 +61,10 @@ struct AddRevisionInput {
summary: String,
}
#[derive(Debug, Deserialize, JsonSchema)]
struct RecordMergeResultInput {
struct CompleteInput {
ticket: String,
expected_current_revision_id: String,
operation_id: String,
expected_revision_id: String,
target_commit: String,
source_commit: String,
result_commit: String,
@@ -87,12 +85,6 @@ enum MergeResolutionInput {
ConflictsResolved,
}
#[derive(Debug, Deserialize, JsonSchema)]
struct CompleteInput {
ticket: String,
operation_id: String,
expected_revision_id: String,
}
#[derive(Debug, Deserialize, JsonSchema)]
struct ReviewInput {
decision: ReviewDecisionInput,
#[serde(default)]
@@ -125,7 +117,6 @@ impl Kind {
Self::Readiness => "MergeRequestReadinessCheck",
Self::Open => "MergeRequestOpen",
Self::AddRevision => "MergeRequestAddRevision",
Self::RecordMergeResult => "MergeRequestRecordMergeResult",
Self::Complete => "MergeRequestComplete",
Self::Review => "MergeRequestReviewSubmit",
}
@@ -138,7 +129,6 @@ impl Kind {
Self::Show | Self::Readiness => json!(schemars::schema_for!(ShowInput)),
Self::Open => json!(schemars::schema_for!(OpenInput)),
Self::AddRevision => json!(schemars::schema_for!(AddRevisionInput)),
Self::RecordMergeResult => json!(schemars::schema_for!(RecordMergeResultInput)),
Self::Complete => json!(schemars::schema_for!(CompleteInput)),
Self::Review => json!(schemars::schema_for!(ReviewInput)),
}
@@ -202,8 +192,8 @@ impl Tool for MergeRequestTool {
),
)
}
Kind::RecordMergeResult => {
let v: RecordMergeResultInput = parse(input)?;
Kind::Complete => {
let v: CompleteInput = parse(input)?;
nonempty(&v.ticket)?;
let strategy = match v.strategy {
MergeStrategyInput::FastForward => "fast_forward",
@@ -214,20 +204,6 @@ impl Tool for MergeRequestTool {
MergeResolutionInput::Clean => "clean",
MergeResolutionInput::ConflictsResolved => "conflicts_resolved",
};
(
WorkspaceRequestMethod::Post,
format!(
"/api/w/{workspace_id}/tickets/{}/merge-request/merge-results",
v.ticket
),
Some(
json!({"expected_current_revision_id":v.expected_current_revision_id,"operation_id":v.operation_id,"target_commit":v.target_commit,"source_commit":v.source_commit,"result_commit":v.result_commit,"strategy":strategy,"resolution":resolution}),
),
)
}
Kind::Complete => {
let v: CompleteInput = parse(input)?;
nonempty(&v.ticket)?;
(
WorkspaceRequestMethod::Post,
format!(
@@ -235,7 +211,7 @@ impl Tool for MergeRequestTool {
v.ticket
),
Some(
json!({"operation_id":v.operation_id,"expected_revision_id":v.expected_revision_id}),
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}),
),
)
}
@@ -310,7 +286,6 @@ pub fn common_tools(client: Arc<dyn WorkspaceClient>) -> Vec<ToolDefinition> {
definition(client.clone(), Kind::Readiness),
definition(client.clone(), Kind::Open),
definition(client.clone(), Kind::AddRevision),
definition(client.clone(), Kind::RecordMergeResult),
definition(client, Kind::Complete),
]
}
@@ -338,9 +313,6 @@ pub fn description(name: &str) -> Option<&'static str> {
"MergeRequestAddRevision" => {
Some("Append an immutable revision; prior approval cannot carry to the new revision.")
}
"MergeRequestRecordMergeResult" => Some(
"Record validated immutable integration evidence for the current source revision and target tip.",
),
"MergeRequestComplete" => {
Some("CAS-complete an approved revision with operation-id replay and crash fencing.")
}
@@ -356,14 +328,14 @@ mod tests {
use super::*;
#[test]
fn merge_request_tool_contract_omits_tree_hashes_and_exposes_merge_result() {
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 result = serde_json::to_string(&schemars::schema_for!(RecordMergeResultInput)).unwrap();
let complete = serde_json::to_string(&schemars::schema_for!(CompleteInput)).unwrap();
assert!(!open.contains("head_tree"));
assert!(!add.contains("head_tree"));
assert!(result.contains("fast_forward"));
assert!(result.contains("conflicts_resolved"));
assert!(MERGE_REQUEST_COMMON_TOOL_NAMES.contains(&"MergeRequestRecordMergeResult"));
assert!(complete.contains("result_commit"));
assert!(complete.contains("conflicts_resolved"));
assert!(!MERGE_REQUEST_COMMON_TOOL_NAMES.contains(&"MergeRequestRecordMergeResult"));
}
}
crates/worker/src/spawn/tool.rs
+2 -12
View File
@@ -68,9 +68,6 @@ struct SubWorkerSpawnInput {
struct ReviewerHandoffInput {
ticket_id: String,
revision_id: String,
/// Optional immutable MergeResult subject. Omit for source-revision review.
#[serde(default)]
merge_result_id: Option<String>,
}
#[derive(Debug, Deserialize, schemars::JsonSchema)]
@@ -425,7 +422,6 @@ impl Tool for SubWorkerSpawnTool {
(
review.ticket_id.clone(),
review.revision_id.clone(),
review.merge_result_id.clone(),
uuid::Uuid::now_v7().to_string(),
format!(
"{}{}",
@@ -435,9 +431,7 @@ impl Tool for SubWorkerSpawnTool {
)
});
let child_workspace_context =
if let Some((ticket_id, revision_id, merge_result_id, _, capability_token)) =
&reviewer_attempt
{
if let Some((ticket_id, revision_id, _, capability_token)) = &reviewer_attempt {
let workspace_id =
self.workspace_context
.workspace_id()
@@ -459,7 +453,6 @@ impl Tool for SubWorkerSpawnTool {
ReviewerAttemptContext {
ticket_id: ticket_id.clone(),
revision_id: revision_id.clone(),
merge_result_id: merge_result_id.clone(),
},
capability_token.clone(),
));
@@ -554,9 +547,7 @@ impl Tool for SubWorkerSpawnTool {
}
};
if let Some((ticket_id, revision_id, merge_result_id, attempt_id, capability_token)) =
&reviewer_attempt
{
if let Some((ticket_id, revision_id, attempt_id, capability_token)) = &reviewer_attempt {
let workspace_id = self.workspace_context.workspace_id().ok_or_else(|| {
ToolError::ExecutionFailed("reviewer attempt lost Workspace identity".to_string())
})?;
@@ -588,7 +579,6 @@ impl Tool for SubWorkerSpawnTool {
let body = serde_json::json!({
"attempt_id": attempt_id,
"revision_id": revision_id,
"merge_result_id": merge_result_id,
"child_session_id": child_session_id,
"capability_token": capability_token,
});
crates/worker/src/worker.rs
-10
View File
@@ -249,7 +249,6 @@ pub trait WorkspaceClient: std::fmt::Debug + Send + Sync {
pub struct ReviewerAttemptContext {
pub ticket_id: String,
pub revision_id: String,
pub merge_result_id: Option<String>,
}
#[derive(Debug)]
@@ -311,14 +310,6 @@ impl WorkspaceClient for ReviewerChildWorkspaceClient {
"revision_id".to_string(),
serde_json::Value::String(self.context.revision_id.clone()),
);
object.insert(
"merge_result_id".to_string(),
self.context
.merge_result_id
.clone()
.map(serde_json::Value::String)
.unwrap_or(serde_json::Value::Null),
);
object.insert(
"capability_token".to_string(),
serde_json::Value::String(self.capability_token.clone()),
@@ -462,7 +453,6 @@ mod reviewer_client_tests {
ReviewerAttemptContext {
ticket_id: "T1".into(),
revision_id: "V1".into(),
merge_result_id: None,
},
"secret".into(),
);
crates/workspace-server/src/repositories.rs
+88 -7
View File
@@ -97,13 +97,38 @@ pub struct CommitObservation {
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RepositoryLookupError {
UnknownRepository { id: RepositoryId },
UnsupportedProvider { id: RepositoryId, provider: String },
MissingDefaultSelector { id: RepositoryId },
InvalidSelector { id: RepositoryId, selector: String },
CommitNotFound { id: RepositoryId, commit: String },
InvalidCommitRelation { id: RepositoryId, detail: String },
ProviderFailure { id: RepositoryId, operation: String },
UnknownRepository {
id: RepositoryId,
},
UnsupportedProvider {
id: RepositoryId,
provider: String,
},
MissingDefaultSelector {
id: RepositoryId,
},
InvalidSelector {
id: RepositoryId,
selector: String,
},
CommitNotFound {
id: RepositoryId,
commit: String,
},
InvalidCommitRelation {
id: RepositoryId,
detail: String,
},
TargetMoved {
id: RepositoryId,
selector: String,
expected: String,
observed: Option<String>,
},
ProviderFailure {
id: RepositoryId,
operation: String,
},
}
#[derive(Debug, Clone)]
@@ -286,6 +311,48 @@ impl RepositoryRegistryReader {
}
}
pub fn update_merge_target(
&self,
id: &str,
selector: &str,
expected_target: &str,
result_commit: &str,
) -> Result<(), RepositoryLookupError> {
let repository = self.merge_repository(id)?;
if !selector.starts_with("refs/heads/")
|| selector.starts_with('-')
|| selector.as_bytes().contains(&0)
{
return Err(RepositoryLookupError::InvalidSelector {
id: id.into(),
selector: selector.into(),
});
}
self.observe_commit(id, result_commit)?;
let status = Command::new("git")
.arg("-C")
.arg(&repository.path)
.args(["update-ref", selector, result_commit, expected_target])
.status()
.map_err(|_| RepositoryLookupError::ProviderFailure {
id: id.into(),
operation: "guarded target update".into(),
})?;
if status.success() {
return Ok(());
}
let observed = self
.observe_merge_target(id, Some(selector))
.ok()
.map(|target| target.commit);
Err(RepositoryLookupError::TargetMoved {
id: id.into(),
selector: selector.into(),
expected: expected_target.into(),
observed,
})
}
fn merge_repository(&self, id: &str) -> Result<&ConfiguredRepository, RepositoryLookupError> {
let repository = self
.find(id)
@@ -656,6 +723,20 @@ mod tests {
vec![base.clone()]
);
reader.ensure_ancestor("main", &base, &source).unwrap();
reader
.update_merge_target("main", "refs/heads/main", &base, &source)
.unwrap();
assert_eq!(
reader
.observe_merge_target("main", Some("refs/heads/main"))
.unwrap()
.commit,
source
);
assert!(matches!(
reader.update_merge_target("main", "refs/heads/main", &base, &base),
Err(RepositoryLookupError::TargetMoved { .. })
));
assert!(matches!(
reader.ensure_ancestor("main", &source, &base),
Err(RepositoryLookupError::InvalidCommitRelation { .. })
crates/workspace-server/src/server.rs
+204 -211
View File
@@ -1290,10 +1290,6 @@ pub fn build_router(api: WorkspaceApi) -> Router {
"/api/w/{workspace_id}/tickets/{id}/merge-request/revisions",
post(scoped_add_merge_request_revision),
)
.route(
"/api/w/{workspace_id}/tickets/{id}/merge-request/merge-results",
post(scoped_record_merge_request_result),
)
.route(
"/api/w/{workspace_id}/internal/reviewer-child-sessions",
post(scoped_register_reviewer_child_session),
@@ -3541,17 +3537,6 @@ struct AddMergeRequestRevisionRequest {
summary: String,
}
#[derive(Debug, serde::Deserialize)]
struct RecordMergeResultRequest {
expected_current_revision_id: String,
operation_id: String,
target_commit: String,
source_commit: String,
result_commit: String,
strategy: merge_request::MergeStrategy,
resolution: merge_request::MergeResolution,
}
#[derive(Debug, serde::Deserialize)]
struct RegisterReviewerChildSessionRequest {
child_session_id: String,
@@ -3561,8 +3546,6 @@ struct RegisterReviewerChildSessionRequest {
struct RegisterMergeRequestReviewAttemptRequest {
attempt_id: String,
revision_id: String,
#[serde(default)]
merge_result_id: Option<String>,
child_session_id: String,
capability_token: String,
}
@@ -3570,8 +3553,6 @@ struct RegisterMergeRequestReviewAttemptRequest {
#[derive(Debug, serde::Deserialize)]
struct SubmitMergeRequestReviewRequest {
revision_id: String,
#[serde(default)]
merge_result_id: Option<String>,
capability_token: String,
decision: merge_request::ReviewDecision,
#[serde(default)]
@@ -3584,6 +3565,11 @@ struct SubmitMergeRequestReviewRequest {
struct CompleteMergeRequestRequest {
operation_id: String,
expected_revision_id: String,
target_commit: String,
source_commit: String,
result_commit: String,
strategy: merge_request::MergeStrategy,
resolution: merge_request::MergeResolution,
}
#[derive(Debug, serde::Deserialize)]
@@ -3840,105 +3826,6 @@ async fn scoped_add_merge_request_revision(
Ok(Json(mr))
}
async fn scoped_record_merge_request_result(
State(api): State<WorkspaceApi>,
AxumPath((workspace_id, ticket_id)): AxumPath<(String, String)>,
headers: HeaderMap,
Json(input): Json<RecordMergeResultRequest>,
) -> ApiResult<Json<merge_request::RecordMergeResultOutcome>> {
require_workspace_access(&workspace_id, &api)?;
let source = authenticate_worker_mutation_source(&api, &workspace_id, &headers)?;
require_online_workspace_orchestrator_source(&api, &source)?;
let store = merge_request_store(&api, &workspace_id)?;
let mr = store.show_for_ticket(&ticket_id)?.ok_or_else(|| {
Error::from(merge_request::MergeRequestError::NotFound(
ticket_id.clone(),
))
})?;
if mr.current_revision.revision_id != input.expected_current_revision_id {
return Err(
Error::from(merge_request::MergeRequestError::StaleRevision {
expected: input.expected_current_revision_id,
current: mr.current_revision.revision_id,
})
.into(),
);
}
let target = api
.repository_reader()
.observe_merge_target(&mr.repository_id, mr.target_ref_selector.as_deref())
.map_err(repository_merge_evidence_error)?;
let supplied_target = api
.repository_reader()
.observe_commit(&mr.repository_id, &input.target_commit)
.map_err(repository_merge_evidence_error)?;
let supplied_source = api
.repository_reader()
.observe_commit(&mr.repository_id, &input.source_commit)
.map_err(repository_merge_evidence_error)?;
let supplied_result = api
.repository_reader()
.observe_commit(&mr.repository_id, &input.result_commit)
.map_err(repository_merge_evidence_error)?;
if target.commit != supplied_target.commit {
return Err(Error::InvalidInput(
"MergeResult target_commit is not the current target tip".into(),
)
.into());
}
if mr.current_revision.head_commit != supplied_source.commit {
return Err(Error::InvalidInput(
"MergeResult source_commit is not the current source revision".into(),
)
.into());
}
match input.strategy {
merge_request::MergeStrategy::FastForward => {
if input.resolution != merge_request::MergeResolution::None
|| supplied_result.commit != supplied_source.commit
{
return Err(Error::InvalidInput(
"fast-forward MergeResult must use the source commit and resolution=none"
.into(),
)
.into());
}
api.repository_reader()
.ensure_ancestor(&mr.repository_id, &target.commit, &supplied_source.commit)
.map_err(repository_merge_evidence_error)?;
}
merge_request::MergeStrategy::Merge => {
if input.resolution == merge_request::MergeResolution::None {
return Err(Error::InvalidInput(
"merge MergeResult requires clean or conflicts_resolved resolution".into(),
)
.into());
}
if supplied_result.parents.len() != 2
|| !supplied_result.parents.contains(&target.commit)
|| !supplied_result.parents.contains(&supplied_source.commit)
{
return Err(Error::InvalidInput("merge result commit must have exactly the target and source commits as parents".into()).into());
}
}
}
let outcome = store.record_merge_result(merge_request::RecordMergeResult {
merge_result_id: format!("MRG-{}", Uuid::new_v4()),
ticket_id,
expected_revision_id: mr.current_revision.revision_id,
target_commit: target.commit,
source_commit: supplied_source.commit,
result_commit: supplied_result.commit,
strategy: input.strategy,
resolution: input.resolution,
operation_id: input.operation_id,
actor_runtime_id: source.runtime_id,
actor_worker_id: source.worker_id,
created_at: Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true),
})?;
Ok(Json(outcome))
}
async fn scoped_register_reviewer_child_session(
State(api): State<WorkspaceApi>,
headers: HeaderMap,
@@ -3974,9 +3861,7 @@ async fn scoped_register_merge_request_review_attempt(
.ok_or_else(|| {
Error::TicketAssignmentConflict("Ticket has no current assigned Coder".into())
})?;
if input.merge_result_id.is_some() {
require_online_workspace_orchestrator_source(&api, &source)?;
} else if assignment.worker.runtime_id != source.runtime_id
if assignment.worker.runtime_id != source.runtime_id
|| assignment.worker.worker_id != source.worker_id
{
return Err(Error::TicketAssignmentConflict(
@@ -3989,7 +3874,6 @@ async fn scoped_register_merge_request_review_attempt(
attempt_id: input.attempt_id,
ticket_id,
revision_id: input.revision_id,
merge_result_id: input.merge_result_id,
parent_assignment_id: assignment.assignment_id,
parent_runtime_id: source.runtime_id,
parent_worker_id: source.worker_id,
@@ -4011,7 +3895,6 @@ async fn scoped_submit_merge_request_review(
merge_request_store(&api, &workspace_id)?.submit_review(merge_request::SubmitReview {
ticket_id,
revision_id: input.revision_id,
merge_result_id: input.merge_result_id,
capability_token: input.capability_token,
decision: input.decision,
body: input.body,
@@ -4038,47 +3921,141 @@ async fn scoped_complete_merge_request(
Error::TicketAssignmentConflict("Ticket has no current assigned Coder".into())
})?;
let store = merge_request_store(&api, &workspace_id)?;
let current = store.show_for_ticket(&ticket_id)?.ok_or_else(|| {
Error::from(merge_request::MergeRequestError::NotFound(
ticket_id.clone(),
))
})?;
let target_commit = observe_merge_request_target(&api, &current);
let observed = store
.show_for_ticket_with_target(&ticket_id, target_commit.as_deref())?
.ok_or_else(|| {
Error::from(merge_request::MergeRequestError::NotFound(
ticket_id.clone(),
))
})?;
let final_result = observed
.final_merge_result
.as_ref()
.ok_or(merge_request::MergeRequestError::FinalMergeResultMissing)?;
if final_result.target_status != merge_request::MergeResultTargetStatus::Applied {
return Err(merge_request::MergeRequestError::FinalMergeResultNotApplied.into());
let mr = store
.show_for_ticket(&ticket_id)?
.ok_or_else(|| merge_request::MergeRequestError::NotFound(ticket_id.clone()))?;
if mr.current_revision.revision_id != input.expected_revision_id {
return Err(merge_request::MergeRequestError::StaleRevision {
expected: input.expected_revision_id,
current: mr.current_revision.revision_id,
}
.into());
}
let readiness = store.readiness_for_ticket_with_target(&ticket_id, target_commit.as_deref())?;
if !readiness.ready {
if mr.review_status != merge_request::ReviewStatus::Approved {
return Err(merge_request::MergeRequestError::NotApproved.into());
}
if input.source_commit != mr.current_revision.head_commit {
return Err(merge_request::MergeRequestError::InvalidMergeOutcome(
"source commit does not match the current approved revision".into(),
)
.into());
}
if mr.state == merge_request::MergeRequestState::Merged {
return Ok(Json(store.complete(
merge_request::CompleteMergeRequest {
operation_id: input.operation_id,
ticket_id,
expected_revision_id: mr.current_revision.revision_id,
target_commit: input.target_commit,
source_commit: input.source_commit,
result_commit: input.result_commit,
strategy: input.strategy,
resolution: input.resolution,
implementation_assignment_id: assignment.assignment_id,
completion_actor_runtime_id: source.runtime_id,
completion_actor_worker_id: source.worker_id,
now: Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true),
},
)?));
}
let selector = mr
.target_ref_selector
.as_deref()
.ok_or(merge_request::MergeRequestError::UnknownTarget)?;
let repositories = api.repository_reader();
let observed_target = repositories
.observe_merge_target(&mr.repository_id, Some(selector))
.map_err(repository_merge_evidence_error)?;
let source_commit = repositories
.observe_commit(&mr.repository_id, &input.source_commit)
.map_err(repository_merge_evidence_error)?;
if source_commit.commit != input.source_commit {
return Err(Error::InvalidInput("source commit must be canonical".into()).into());
}
let result_commit = repositories
.observe_commit(&mr.repository_id, &input.result_commit)
.map_err(repository_merge_evidence_error)?;
if result_commit.commit != input.result_commit {
return Err(Error::InvalidInput("result commit must be canonical".into()).into());
}
match input.strategy {
merge_request::MergeStrategy::FastForward => {
if input.resolution != merge_request::MergeResolution::None
|| input.result_commit != input.source_commit
{
return Err(merge_request::MergeRequestError::InvalidMergeOutcome(
"fast-forward result must equal the approved source and use resolution=none"
.into(),
)
.into());
}
repositories
.ensure_ancestor(
&mr.repository_id,
&input.target_commit,
&input.source_commit,
)
.map_err(repository_merge_evidence_error)?;
}
merge_request::MergeStrategy::Merge => {
if input.resolution == merge_request::MergeResolution::None
|| result_commit.parents
!= vec![input.target_commit.clone(), input.source_commit.clone()]
{
return Err(merge_request::MergeRequestError::InvalidMergeOutcome(
"merge result must have the expected target and approved source as its two ordered parents"
.into(),
)
.into());
}
}
}
let target_was_already_updated = observed_target.commit == input.result_commit;
if observed_target.commit != input.target_commit && !target_was_already_updated {
return Err(Error::InvalidInput(format!(
"Merge Request is not completion-ready: {}",
readiness.blockers.join("; ")
"Merge Request target moved: expected {}, observed {}",
input.target_commit, observed_target.commit
))
.into());
}
let outcome = store.complete(merge_request::CompleteMergeRequest {
if !target_was_already_updated {
repositories
.update_merge_target(
&mr.repository_id,
selector,
&input.target_commit,
&input.result_commit,
)
.map_err(repository_merge_evidence_error)?;
}
let completion = merge_request::CompleteMergeRequest {
operation_id: input.operation_id,
ticket_id,
expected_revision_id: input.expected_revision_id,
expected_merge_result_id: final_result.merge_result_id.clone(),
observed_target_commit: target_commit
.ok_or(merge_request::MergeRequestError::FinalMergeResultNotApplied)?,
expected_revision_id: mr.current_revision.revision_id,
target_commit: input.target_commit.clone(),
source_commit: input.source_commit,
result_commit: input.result_commit.clone(),
strategy: input.strategy,
resolution: input.resolution,
implementation_assignment_id: assignment.assignment_id,
completion_actor_runtime_id: source.runtime_id,
completion_actor_worker_id: source.worker_id,
now: Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true),
})?;
Ok(Json(outcome))
};
match store.complete(completion) {
Ok(outcome) => Ok(Json(outcome)),
Err(error) => {
if !target_was_already_updated {
let _ = repositories.update_merge_target(
&mr.repository_id,
selector,
&input.result_commit,
&input.target_commit,
);
}
Err(error.into())
}
}
}
async fn scoped_reopen_merge_request(
@@ -4089,9 +4066,7 @@ async fn scoped_reopen_merge_request(
) -> ApiResult<Json<merge_request::MergeRequest>> {
let workspace_id = parse_workspace_id(&workspace_id)?;
require_workspace_access(&workspace_id, &api)?;
if headers.contains_key("authorization") {
return Err(Error::BrowserReopenConfirmationRequired.into());
}
reject_non_browser_reopen_auth(&headers)?;
let _actor = require_actor(&api, &headers).await?;
if !input.explicit_confirmation {
return Err(Error::BrowserReopenConfirmationRequired.into());
@@ -4103,6 +4078,13 @@ async fn scoped_reopen_merge_request(
)?))
}
fn reject_non_browser_reopen_auth(headers: &HeaderMap) -> Result<()> {
if headers.contains_key("authorization") {
return Err(Error::BrowserReopenConfirmationRequired);
}
Ok(())
}
async fn scoped_close_ticket_record(
State(api): State<WorkspaceApi>,
AxumPath((workspace_id, id)): AxumPath<(String, String)>,
@@ -11719,6 +11701,17 @@ mod tests {
ObjectiveTicketLinkRecord, SqliteWorkspaceStore, WorkspaceRecord,
};
#[test]
fn reopen_confirmation_rejects_api_token_actor_before_session_resolution() {
let mut headers = HeaderMap::new();
headers.insert("authorization", "Bearer api-token".parse().unwrap());
assert!(matches!(
reject_non_browser_reopen_auth(&headers),
Err(Error::BrowserReopenConfirmationRequired)
));
assert!(reject_non_browser_reopen_auth(&HeaderMap::new()).is_ok());
}
#[test]
fn flow_or_generic_worker_state_change_is_not_ticket_completion_authority() {
let operation = TicketBackendOperation::SetWorkflowState {
@@ -12365,42 +12358,36 @@ mod tests {
async fn merge_request_completion_endpoint_rejects_coder_and_accepts_orchestrator() {
let workspace = tempfile::tempdir().unwrap();
init_clean_git_workspace(workspace.path());
let git_output = |args: &[&str]| {
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(), "git {:?} failed", args);
assert!(output.status.success());
String::from_utf8(output.stdout).unwrap().trim().to_string()
};
let base_commit = git_output(&["rev-parse", "HEAD"]);
std::fs::write(workspace.path().join("README.md"), "completion candidate\n").unwrap();
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(["add", "README.md"])
.status()
.unwrap()
.success()
);
assert!(
std::process::Command::new("git")
.arg("-C")
.arg(workspace.path())
.args(["commit", "-m", "candidate"])
.status()
.unwrap()
.success()
);
let head_commit = git_output(&["rev-parse", "HEAD"]);
assert!(
std::process::Command::new("git")
.arg("-C")
.arg(workspace.path())
.args(["reset", "--hard", &base_commit])
.args(["reset", "--hard", &target_commit])
.status()
.unwrap()
.success()
@@ -12441,12 +12428,12 @@ mod tests {
merge_request_id: "MR-server-completion".into(),
ticket_id: ticket.id.clone(),
repository_id: TEST_REPOSITORY_ID.into(),
target_ref_selector: "develop".into(),
target_ref_selector: target_ref.clone(),
revision: merge_request::MergeRequestRevision {
revision_id: "V1".into(),
ordinal: 1,
base_commit: base_commit.clone(),
head_commit: head_commit.clone(),
base_commit: target_commit.clone(),
head_commit: source_commit.clone(),
diff_digest: "sha256:diff".into(),
changed_paths: vec!["src/lib.rs".into()],
@@ -12472,7 +12459,6 @@ mod tests {
attempt_id: "attempt".into(),
ticket_id: ticket.id.clone(),
revision_id: "V1".into(),
merge_result_id: None,
parent_assignment_id: assignment.assignment_id.clone(),
parent_runtime_id: coder.worker_ref.runtime_id.clone(),
parent_worker_id: coder.worker_ref.worker_id.clone(),
@@ -12485,7 +12471,6 @@ mod tests {
.submit_review(merge_request::SubmitReview {
ticket_id: ticket.id.clone(),
revision_id: "V1".into(),
merge_result_id: None,
capability_token: "review-token".into(),
decision: merge_request::ReviewDecision::Approve,
body: "approved".into(),
@@ -12493,31 +12478,6 @@ mod tests {
now: "t3".into(),
})
.unwrap();
mr_store
.record_merge_result(merge_request::RecordMergeResult {
operation_id: "record-complete-result".into(),
merge_result_id: "result-complete".into(),
ticket_id: ticket.id.clone(),
expected_revision_id: "V1".into(),
target_commit: base_commit.clone(),
source_commit: head_commit.clone(),
result_commit: head_commit.clone(),
strategy: merge_request::MergeStrategy::FastForward,
resolution: merge_request::MergeResolution::None,
actor_runtime_id: coder.worker_ref.runtime_id.clone(),
actor_worker_id: coder.worker_ref.worker_id.clone(),
created_at: "t4".into(),
})
.unwrap();
assert!(
std::process::Command::new("git")
.arg("-C")
.arg(workspace.path())
.args(["reset", "--hard", &head_commit])
.status()
.unwrap()
.success()
);
let worker_headers = |worker: &RuntimeWorkerRef| {
let mut headers = HeaderMap::new();
@@ -12534,6 +12494,11 @@ mod tests {
let request = || CompleteMergeRequestRequest {
operation_id: "complete-operation".into(),
expected_revision_id: "V1".into(),
target_commit: target_commit.clone(),
source_commit: source_commit.clone(),
result_commit: source_commit.clone(),
strategy: merge_request::MergeStrategy::FastForward,
resolution: merge_request::MergeResolution::None,
};
let coder_error = scoped_complete_merge_request(
State(api.clone()),
@@ -12574,6 +12539,35 @@ mod tests {
.workflow_state,
TicketWorkflowState::Done
);
assert_eq!(
api.repository_reader()
.observe_merge_target(TEST_REPOSITORY_ID, Some(&target_ref))
.unwrap()
.commit,
source_commit
);
let merged = mr_store.show_for_ticket(&ticket.id).unwrap().unwrap();
assert_eq!(
merged.merged_target_commit.as_deref(),
Some(target_commit.as_str())
);
assert_eq!(
merged.merged_result_commit.as_deref(),
Some(source_commit.as_str())
);
assert_eq!(
merged.merge_strategy,
Some(merge_request::MergeStrategy::FastForward)
);
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!(replayed.replayed);
let conn = rusqlite::Connection::open(&api.config.database_path).unwrap();
let actor: String = conn
.query_row(
@@ -13418,7 +13412,6 @@ mod tests {
for args in [
vec!["add", "README.md", ".gitignore"],
vec!["commit", "-m", "init"],
vec!["branch", "-M", "develop"],
] {
let status = std::process::Command::new("git")
.arg("-C")