Merge commit '89856eb7c3dd6b659ff952d7e3be4a8c0885bf98' into work/T-584-agen-typed-interceptor
This commit is contained in:
+11
-239
@@ -9,7 +9,6 @@ use thiserror::Error;
|
||||
use uuid::Uuid;
|
||||
|
||||
const SCHEMA_VERSION: i64 = 12;
|
||||
const PREVIOUS_SCHEMA_VERSION: i64 = 11;
|
||||
const MAX_BODY_BYTES: usize = 16 * 1024;
|
||||
const DOMAIN_TABLES: [&str; 5] = [
|
||||
"merge_requests",
|
||||
@@ -37,7 +36,7 @@ impl MergeRequestState {
|
||||
|
||||
fn parse(v: &str) -> Result<Self, MergeRequestError> {
|
||||
match v {
|
||||
"draft" | "open" => Ok(Self::Open),
|
||||
"open" => Ok(Self::Open),
|
||||
"merged" => Ok(Self::Merged),
|
||||
"closed" => Ok(Self::Closed),
|
||||
_ => Err(MergeRequestError::Corrupt(format!("unknown state `{v}`"))),
|
||||
@@ -1355,14 +1354,9 @@ pub fn migrate(c: &Connection) -> Result<(), MergeRequestError> {
|
||||
match schema_state(c)? {
|
||||
SchemaState::Fresh => fresh(c),
|
||||
SchemaState::Current(SCHEMA_VERSION) => verify(c),
|
||||
SchemaState::Current(PREVIOUS_SCHEMA_VERSION) => from_v11(c, PreviousSchemaMarker::Current),
|
||||
SchemaState::Legacy(PREVIOUS_SCHEMA_VERSION) => from_v11(c, PreviousSchemaMarker::Legacy),
|
||||
SchemaState::Current(v) => Err(MergeRequestError::Operation(format!(
|
||||
"unsupported schema {v}"
|
||||
))),
|
||||
SchemaState::Legacy(v) => Err(MergeRequestError::Operation(format!(
|
||||
"unsupported legacy schema {v}"
|
||||
))),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1370,26 +1364,14 @@ pub fn migrate(c: &Connection) -> Result<(), MergeRequestError> {
|
||||
enum SchemaState {
|
||||
Fresh,
|
||||
Current(i64),
|
||||
Legacy(i64),
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
enum PreviousSchemaMarker {
|
||||
Current,
|
||||
Legacy,
|
||||
}
|
||||
|
||||
fn schema_state(c: &Connection) -> Result<SchemaState, MergeRequestError> {
|
||||
let (current, legacy): (bool, bool) = c.query_row(
|
||||
"SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name='merge_request_schema'),EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name='merge_request_schema_migrations')",
|
||||
let current: bool = c.query_row(
|
||||
"SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name='merge_request_schema')",
|
||||
[],
|
||||
|r| Ok((r.get(0)?, r.get(1)?)),
|
||||
|r| r.get(0),
|
||||
)?;
|
||||
if current && legacy {
|
||||
return Err(MergeRequestError::Corrupt(
|
||||
"both current and legacy schema markers exist".into(),
|
||||
));
|
||||
}
|
||||
if current {
|
||||
let (count, singleton, version): (i64, Option<i64>, Option<i64>) = c.query_row(
|
||||
"SELECT COUNT(*),MIN(singleton),MAX(version) FROM merge_request_schema",
|
||||
@@ -1406,22 +1388,6 @@ fn schema_state(c: &Connection) -> Result<SchemaState, MergeRequestError> {
|
||||
})?;
|
||||
return Ok(SchemaState::Current(version));
|
||||
}
|
||||
if legacy {
|
||||
let (count, version): (i64, Option<i64>) = c.query_row(
|
||||
"SELECT COUNT(*),MAX(version) FROM merge_request_schema_migrations",
|
||||
[],
|
||||
|r| Ok((r.get(0)?, r.get(1)?)),
|
||||
)?;
|
||||
if count != 1 {
|
||||
return Err(MergeRequestError::Corrupt(
|
||||
"legacy schema marker must contain exactly one version".into(),
|
||||
));
|
||||
}
|
||||
let version = version.ok_or_else(|| {
|
||||
MergeRequestError::Corrupt("legacy schema marker version is null".into())
|
||||
})?;
|
||||
return Ok(SchemaState::Legacy(version));
|
||||
}
|
||||
let domain_tables: bool = c.query_row(
|
||||
"SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name GLOB 'merge_request*')",
|
||||
[],
|
||||
@@ -1436,214 +1402,20 @@ fn schema_state(c: &Connection) -> Result<SchemaState, MergeRequestError> {
|
||||
}
|
||||
fn fresh(c: &Connection) -> Result<(), MergeRequestError> {
|
||||
let t = c.unchecked_transaction()?;
|
||||
tables(&t, true)?;
|
||||
t.execute("INSERT INTO merge_request_schema VALUES(1,12)", [])?;
|
||||
tables(&t)?;
|
||||
t.execute(
|
||||
"INSERT INTO merge_request_schema VALUES(1,?1)",
|
||||
params![SCHEMA_VERSION],
|
||||
)?;
|
||||
fk(&t)?;
|
||||
t.commit()?;
|
||||
Ok(())
|
||||
}
|
||||
fn tables(t: &Transaction<'_>, marker: bool) -> Result<(), MergeRequestError> {
|
||||
if marker {
|
||||
t.execute_batch("CREATE TABLE merge_request_schema(singleton INTEGER PRIMARY KEY CHECK(singleton=1),version INTEGER NOT NULL);")?
|
||||
}
|
||||
fn tables(t: &Transaction<'_>) -> Result<(), MergeRequestError> {
|
||||
t.execute_batch("CREATE TABLE merge_request_schema(singleton INTEGER PRIMARY KEY CHECK(singleton=1),version INTEGER NOT NULL);")?;
|
||||
t.execute_batch("CREATE TABLE merge_requests(workspace_id TEXT NOT NULL,merge_request_id TEXT NOT NULL,repository_id TEXT NOT NULL,state TEXT NOT NULL CHECK(state IN('open','merged','closed')),selector_from TEXT,selector_to TEXT NOT NULL,created_at TEXT NOT NULL,updated_at TEXT NOT NULL,PRIMARY KEY(workspace_id,merge_request_id),FOREIGN KEY(workspace_id,repository_id)REFERENCES repositories(workspace_id,repository_id));CREATE TABLE merge_request_ticket_relations(workspace_id TEXT NOT NULL,merge_request_id TEXT NOT NULL,ticket_id TEXT NOT NULL,relation_kind TEXT NOT NULL CHECK(relation_kind='implements'),created_at TEXT NOT NULL,PRIMARY KEY(workspace_id,merge_request_id,ticket_id),FOREIGN KEY(workspace_id,merge_request_id)REFERENCES merge_requests(workspace_id,merge_request_id)ON DELETE CASCADE,FOREIGN KEY(workspace_id,ticket_id)REFERENCES typed_tickets(workspace_id,ticket_id)ON DELETE CASCADE);CREATE TABLE merge_request_thread_events(workspace_id TEXT NOT NULL,merge_request_id TEXT NOT NULL,event_id TEXT NOT NULL,sequence INTEGER NOT NULL,kind TEXT NOT NULL CHECK(kind IN('review_requested','review','review_revoked','review_cancelled','comment','merge')),payload_json TEXT NOT NULL,operation_id TEXT,created_at TEXT NOT NULL,PRIMARY KEY(workspace_id,merge_request_id,event_id),UNIQUE(workspace_id,merge_request_id,sequence),FOREIGN KEY(workspace_id,merge_request_id)REFERENCES merge_requests(workspace_id,merge_request_id)ON DELETE CASCADE);CREATE UNIQUE INDEX merge_request_merge_operations ON merge_request_thread_events(workspace_id,operation_id)WHERE operation_id IS NOT NULL;CREATE TABLE merge_request_review_grants(workspace_id TEXT NOT NULL,merge_request_id TEXT NOT NULL,request_event_id TEXT NOT NULL,subject_ref TEXT NOT NULL,reviewer_runtime_id TEXT NOT NULL,reviewer_worker_id TEXT NOT NULL,capability_token TEXT PRIMARY KEY,issued_at TEXT NOT NULL,consumed_at TEXT,revoked_at TEXT,status TEXT NOT NULL CHECK(status IN('issued','consumed','revoked')),FOREIGN KEY(workspace_id,merge_request_id,request_event_id)REFERENCES merge_request_thread_events(workspace_id,merge_request_id,event_id)ON DELETE CASCADE);CREATE TABLE merge_request_reviewer_child_sessions(workspace_id TEXT NOT NULL,child_session_id TEXT NOT NULL,parent_runtime_id TEXT NOT NULL,parent_worker_id TEXT NOT NULL,reviewer_profile TEXT NOT NULL,registered_at TEXT NOT NULL,status TEXT NOT NULL CHECK(status IN('active','consumed')),PRIMARY KEY(workspace_id,child_session_id));")?;
|
||||
Ok(())
|
||||
}
|
||||
fn from_v11(
|
||||
c: &Connection,
|
||||
previous_marker: PreviousSchemaMarker,
|
||||
) -> Result<(), MergeRequestError> {
|
||||
let t = c.unchecked_transaction()?;
|
||||
if previous_marker == PreviousSchemaMarker::Legacy {
|
||||
t.execute_batch("CREATE TABLE merge_request_schema(singleton INTEGER PRIMARY KEY CHECK(singleton=1),version INTEGER NOT NULL);")?;
|
||||
t.execute(
|
||||
"INSERT INTO merge_request_schema VALUES(1,?1)",
|
||||
params![PREVIOUS_SCHEMA_VERSION],
|
||||
)?;
|
||||
}
|
||||
t.execute_batch("ALTER TABLE merge_requests RENAME TO merge_requests_v11;ALTER TABLE merge_request_ticket_relations RENAME TO merge_request_ticket_relations_v11;ALTER TABLE merge_request_revisions RENAME TO merge_request_revisions_v11;ALTER TABLE merge_request_revision_paths RENAME TO merge_request_revision_paths_v11;ALTER TABLE merge_request_reviewer_child_sessions RENAME TO merge_request_reviewer_child_sessions_v11;ALTER TABLE merge_request_review_attempts RENAME TO merge_request_review_attempts_v11;ALTER TABLE merge_request_reviews RENAME TO merge_request_reviews_v11;ALTER TABLE merge_request_review_findings RENAME TO merge_request_review_findings_v11;ALTER TABLE merge_request_completion_operations RENAME TO merge_request_completion_operations_v11;")?;
|
||||
tables(&t, false)?;
|
||||
t.execute("INSERT INTO merge_requests SELECT workspace_id,merge_request_id,repository_id,CASE state WHEN 'draft'THEN'open'ELSE state END,NULL,target_ref_selector,created_at,updated_at FROM merge_requests_v11",[])?;
|
||||
t.execute("INSERT INTO merge_request_ticket_relations SELECT * FROM merge_request_ticket_relations_v11",[])?;
|
||||
migrate_events(&t)?;
|
||||
if previous_marker == PreviousSchemaMarker::Legacy {
|
||||
t.execute("DROP TABLE merge_request_schema_migrations", [])?;
|
||||
}
|
||||
t.execute_batch("DROP TABLE merge_request_review_findings_v11;DROP TABLE merge_request_reviews_v11;DROP TABLE merge_request_review_attempts_v11;DROP TABLE merge_request_reviewer_child_sessions_v11;DROP TABLE merge_request_revision_paths_v11;DROP TABLE merge_request_revisions_v11;DROP TABLE merge_request_completion_operations_v11;DROP TABLE merge_request_ticket_relations_v11;DROP TABLE merge_requests_v11;UPDATE merge_request_schema SET version=12 WHERE singleton=1;")?;
|
||||
fk(&t)?;
|
||||
t.commit()?;
|
||||
Ok(())
|
||||
}
|
||||
fn migrate_events(t: &Transaction<'_>) -> Result<(), MergeRequestError> {
|
||||
let attempts = {
|
||||
let mut s=t.prepare("SELECT a.workspace_id,a.attempt_id,a.merge_request_id,a.parent_runtime_id,a.parent_worker_id,a.child_session_id,a.status,a.created_at,a.consumed_at,r.head_commit FROM merge_request_review_attempts_v11 a JOIN merge_request_revisions_v11 r ON r.workspace_id=a.workspace_id AND r.merge_request_id=a.merge_request_id AND r.revision_id=a.revision_id ORDER BY a.created_at")?;
|
||||
s.query_map([], |r| {
|
||||
Ok((
|
||||
r.get::<_, String>(0)?,
|
||||
r.get::<_, String>(1)?,
|
||||
r.get::<_, String>(2)?,
|
||||
r.get::<_, String>(3)?,
|
||||
r.get::<_, String>(4)?,
|
||||
r.get::<_, String>(5)?,
|
||||
r.get::<_, String>(6)?,
|
||||
r.get::<_, String>(7)?,
|
||||
r.get::<_, Option<String>>(8)?,
|
||||
r.get::<_, String>(9)?,
|
||||
))
|
||||
})?
|
||||
.collect::<Result<Vec<_>, _>>()?
|
||||
};
|
||||
for (ws, a, mr, pr, pw, child, status, created, consumed, subject) in attempts {
|
||||
let req = ReviewRequestedEvent {
|
||||
event_id: format!("migrated-request-{a}"),
|
||||
sequence: next_seq(t, &ws, &mr)?,
|
||||
subject_ref: subject.clone(),
|
||||
requested_by: WorkerIdentity {
|
||||
runtime_id: pr.clone(),
|
||||
worker_id: pw,
|
||||
},
|
||||
reviewer: WorkerIdentity {
|
||||
runtime_id: pr,
|
||||
worker_id: child,
|
||||
},
|
||||
created_at: time(&created)?,
|
||||
};
|
||||
insert_event(t, &ws, &mr, "review_requested", &req, req.created_at, None)?;
|
||||
if status == "submitted" {
|
||||
let(row_dec,row_body,row_at):(String,String,String)=t.query_row("SELECT decision,body,submitted_at FROM merge_request_reviews_v11 WHERE workspace_id=?1 AND attempt_id=?2",params![ws,a],|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?)))?;
|
||||
let findings = {
|
||||
let mut s=t.prepare("SELECT severity,code,path,line,body FROM merge_request_review_findings_v11 WHERE workspace_id=?1 AND attempt_id=?2 ORDER BY ordinal")?;
|
||||
s.query_map(params![ws, a], |r| {
|
||||
Ok(ReviewFinding {
|
||||
severity: match r.get::<_, String>(0)?.as_str() {
|
||||
"blocker" => FindingSeverity::Blocker,
|
||||
"major" => FindingSeverity::Major,
|
||||
"minor" => FindingSeverity::Minor,
|
||||
_ => FindingSeverity::Note,
|
||||
},
|
||||
code: r.get(1)?,
|
||||
path: r.get(2)?,
|
||||
line: r.get(3)?,
|
||||
body: r.get(4)?,
|
||||
})
|
||||
})?
|
||||
.collect::<Result<Vec<_>, _>>()?
|
||||
};
|
||||
let rev = ReviewEvent {
|
||||
event_id: format!("migrated-review-{a}"),
|
||||
sequence: next_seq(t, &ws, &mr)?,
|
||||
request_event_id: req.event_id,
|
||||
subject_ref: subject,
|
||||
decision: if row_dec == "approve" {
|
||||
ReviewDecision::Approve
|
||||
} else {
|
||||
ReviewDecision::RequestChanges
|
||||
},
|
||||
body: row_body,
|
||||
findings,
|
||||
reviewer: req.reviewer,
|
||||
created_at: time(&row_at)?,
|
||||
};
|
||||
insert_event(t, &ws, &mr, "review", &rev, rev.created_at, None)?
|
||||
} else {
|
||||
let at = consumed.as_deref().unwrap_or(&created);
|
||||
let e = ReviewCancelledEvent {
|
||||
event_id: format!("migrated-cancel-{a}"),
|
||||
sequence: next_seq(t, &ws, &mr)?,
|
||||
request_event_id: req.event_id,
|
||||
subject_ref: subject,
|
||||
reason: format!(
|
||||
"legacy `{status}` review request cancelled because its capability cannot be migrated"
|
||||
),
|
||||
created_at: time(at)?,
|
||||
};
|
||||
insert_event(t, &ws, &mr, "review_cancelled", &e, e.created_at, None)?
|
||||
}
|
||||
}
|
||||
let completed = {
|
||||
let mut q=t.prepare("SELECT c.workspace_id,c.operation_id,c.ticket_id,c.target_commit,c.source_commit,c.result_commit,c.strategy,c.resolution,c.completion_actor_runtime_id,c.completion_actor_worker_id,c.updated_at,rel.merge_request_id FROM merge_request_completion_operations_v11 c JOIN merge_request_ticket_relations_v11 rel ON rel.workspace_id=c.workspace_id AND rel.ticket_id=c.ticket_id WHERE c.status='completed' ORDER BY c.updated_at")?;
|
||||
q.query_map([], |r| {
|
||||
Ok((
|
||||
r.get::<_, String>(0)?,
|
||||
r.get::<_, String>(1)?,
|
||||
r.get::<_, String>(2)?,
|
||||
r.get::<_, Option<String>>(3)?,
|
||||
r.get::<_, Option<String>>(4)?,
|
||||
r.get::<_, Option<String>>(5)?,
|
||||
r.get::<_, Option<String>>(6)?,
|
||||
r.get::<_, Option<String>>(7)?,
|
||||
r.get::<_, Option<String>>(8)?,
|
||||
r.get::<_, Option<String>>(9)?,
|
||||
r.get::<_, String>(10)?,
|
||||
r.get::<_, String>(11)?,
|
||||
))
|
||||
})?
|
||||
.collect::<Result<Vec<_>, _>>()?
|
||||
};
|
||||
for (
|
||||
ws,
|
||||
op,
|
||||
_ticket,
|
||||
target,
|
||||
source,
|
||||
result,
|
||||
strategy,
|
||||
resolution,
|
||||
runtime,
|
||||
worker,
|
||||
updated,
|
||||
mr,
|
||||
) in completed
|
||||
{
|
||||
let subject = source.ok_or_else(|| {
|
||||
MergeRequestError::Operation(format!("completed operation {op} lacks source evidence"))
|
||||
})?;
|
||||
let approval:Option<String>=t.query_row("SELECT event_id FROM merge_request_thread_events WHERE workspace_id=?1 AND merge_request_id=?2 AND kind='review' AND json_extract(payload_json,'$.subject_ref')=?3 AND json_extract(payload_json,'$.decision')='approve' ORDER BY sequence DESC LIMIT 1",params![ws,mr,subject],|r|r.get(0)).optional()?;
|
||||
let approval = approval.ok_or_else(|| {
|
||||
MergeRequestError::Operation(format!(
|
||||
"completed operation {op} lacks approval evidence"
|
||||
))
|
||||
})?;
|
||||
let e = MergeEvent {
|
||||
event_id: format!("migrated-merge-{op}"),
|
||||
sequence: next_seq(t, &ws, &mr)?,
|
||||
operation_id: op,
|
||||
approval_event_id: approval,
|
||||
approved_source_ref: subject,
|
||||
target_ref_before: target.ok_or_else(|| {
|
||||
MergeRequestError::Operation("completed operation lacks target evidence".into())
|
||||
})?,
|
||||
target_ref_after: result.ok_or_else(|| {
|
||||
MergeRequestError::Operation("completed operation lacks result evidence".into())
|
||||
})?,
|
||||
strategy: if strategy.as_deref() == Some("merge") {
|
||||
MergeStrategy::Merge
|
||||
} else {
|
||||
MergeStrategy::FastForward
|
||||
},
|
||||
resolution: match resolution.as_deref() {
|
||||
Some("clean") => ConflictResolution::Clean,
|
||||
Some("conflicts_resolved") => ConflictResolution::ConflictsResolved,
|
||||
_ => ConflictResolution::None,
|
||||
},
|
||||
merged_by: WorkerIdentity {
|
||||
runtime_id: runtime.unwrap_or_else(|| "legacy".into()),
|
||||
worker_id: worker.unwrap_or_else(|| "legacy".into()),
|
||||
},
|
||||
created_at: time(&updated)?,
|
||||
};
|
||||
insert_event(
|
||||
t,
|
||||
&ws,
|
||||
&mr,
|
||||
"merge",
|
||||
&e,
|
||||
e.created_at,
|
||||
Some(&e.operation_id),
|
||||
)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
fn verify(c: &Connection) -> Result<(), MergeRequestError> {
|
||||
for n in DOMAIN_TABLES {
|
||||
let e: bool = c.query_row(
|
||||
|
||||
@@ -301,21 +301,13 @@ fn review_revocation_invalidates_readiness() {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn legacy_v11_migration_preserves_review_events_and_replaces_marker() {
|
||||
fn fresh_schema_uses_version_12_and_reopens_as_current() {
|
||||
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_migrations(version INTEGER PRIMARY KEY,applied_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP);INSERT INTO merge_request_schema_migrations(version) VALUES(11);CREATE TABLE merge_requests(workspace_id TEXT,merge_request_id TEXT,repository_id TEXT,state TEXT,target_ref_selector TEXT,current_revision_ordinal INTEGER,current_revision_id TEXT,created_at TEXT,updated_at TEXT,merged_revision_id TEXT,merged_at TEXT);CREATE TABLE merge_request_ticket_relations(workspace_id TEXT,merge_request_id TEXT,ticket_id TEXT,relation_kind TEXT,created_at TEXT);CREATE TABLE merge_request_revisions(workspace_id TEXT,merge_request_id TEXT,revision_id TEXT,ordinal INTEGER,base_commit TEXT,head_commit TEXT,diff_digest TEXT,summary TEXT,assignment_id TEXT,created_at TEXT);CREATE TABLE merge_request_revision_paths(workspace_id TEXT,merge_request_id TEXT,revision_id TEXT,ordinal INTEGER,path TEXT);CREATE TABLE merge_request_reviewer_child_sessions(workspace_id TEXT,child_session_id TEXT,parent_runtime_id TEXT,parent_worker_id TEXT,reviewer_profile TEXT,registered_at TEXT);CREATE TABLE merge_request_review_attempts(workspace_id TEXT,attempt_id TEXT,merge_request_id TEXT,ticket_id TEXT,revision_id TEXT,revision_ordinal INTEGER,parent_assignment_id TEXT,parent_runtime_id TEXT,parent_worker_id TEXT,child_session_id TEXT,reviewer_effective_profile TEXT,capability_token TEXT,status TEXT,created_at TEXT,consumed_at TEXT);CREATE TABLE merge_request_reviews(workspace_id TEXT,attempt_id TEXT,merge_request_id TEXT,revision_id TEXT,decision TEXT,body TEXT,submitted_at TEXT);CREATE TABLE merge_request_review_findings(workspace_id TEXT,attempt_id TEXT,ordinal INTEGER,severity TEXT,code TEXT,path TEXT,line INTEGER,body TEXT);CREATE TABLE merge_request_completion_operations(workspace_id TEXT,operation_id TEXT,ticket_id TEXT,revision_id TEXT,authority_kind TEXT,implementation_assignment_id TEXT,completion_actor_runtime_id TEXT,completion_actor_worker_id TEXT,target_commit TEXT,source_commit TEXT,result_commit TEXT,strategy TEXT,resolution TEXT,fingerprint TEXT,status TEXT,result_ticket_state TEXT,created_at TEXT,updated_at TEXT);INSERT INTO merge_requests VALUES('W','MR','R','open','develop',1,'V','2026-07-26T12:00:00Z','2026-07-26T12:00:00Z',NULL,NULL);INSERT INTO merge_request_ticket_relations VALUES('W','MR','T','implements','2026-07-26T12:00:00Z');INSERT INTO merge_request_revisions VALUES('W','MR','V',1,'base','subject','digest','summary','A','2026-07-26T12:00:00Z');INSERT INTO merge_request_review_attempts VALUES('W','AT','MR','T','V',1,'A','runtime','coder','child','builtin:reviewer','token','submitted','2026-07-26T12:00:00Z','2026-07-26T12:00:01Z');INSERT INTO merge_request_reviews VALUES('W','AT','MR','V','approve','approved','2026-07-26T12:00:01Z');INSERT INTO merge_request_review_attempts VALUES('W','PENDING','MR','T','V',1,'A','runtime','coder','pending-child','builtin:reviewer','pending-token','registered','2026-07-26T12:00:02Z',NULL);").unwrap();
|
||||
c.execute_batch(
|
||||
"CREATE TABLE unrelated_parent(left_id TEXT,right_id TEXT,PRIMARY KEY(left_id,right_id));CREATE TABLE unrelated_child(left_id TEXT REFERENCES unrelated_parent(left_id));",
|
||||
"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));",
|
||||
)
|
||||
.unwrap();
|
||||
let unrelated_mismatch = c
|
||||
.query_row("PRAGMA foreign_key_check", [], |_| Ok(()))
|
||||
.unwrap_err();
|
||||
assert!(
|
||||
unrelated_mismatch
|
||||
.to_string()
|
||||
.contains("foreign key mismatch")
|
||||
);
|
||||
|
||||
merge_request::migrate(&c).unwrap();
|
||||
assert_eq!(
|
||||
c.query_row("SELECT version FROM merge_request_schema", [], |r| {
|
||||
@@ -324,66 +316,26 @@ fn legacy_v11_migration_preserves_review_events_and_replaces_marker() {
|
||||
.unwrap(),
|
||||
12
|
||||
);
|
||||
let legacy_marker: bool = c
|
||||
.query_row(
|
||||
"SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name='merge_request_schema_migrations')",
|
||||
[],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(!legacy_marker);
|
||||
let selector: Option<String> = c
|
||||
.query_row("SELECT selector_from FROM merge_requests", [], |r| r.get(0))
|
||||
.unwrap();
|
||||
assert!(selector.is_none());
|
||||
let kinds: String = c
|
||||
.query_row(
|
||||
"SELECT group_concat(kind,',') FROM merge_request_thread_events ORDER BY sequence",
|
||||
[],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
kinds,
|
||||
"review_requested,review,review_requested,review_cancelled"
|
||||
);
|
||||
let old: bool = c
|
||||
.query_row(
|
||||
"SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE name='merge_request_revisions')",
|
||||
[],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(!old);
|
||||
merge_request::migrate(&c).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn failed_legacy_v11_migration_rolls_back_marker_bridge() {
|
||||
fn current_schema_validation_rejects_missing_tables() {
|
||||
let c = Connection::open_in_memory().unwrap();
|
||||
c.execute_batch(
|
||||
"CREATE TABLE merge_request_schema_migrations(version INTEGER PRIMARY KEY,applied_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP);INSERT INTO merge_request_schema_migrations(version) VALUES(11);CREATE TABLE merge_requests(merge_request_id TEXT);",
|
||||
"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));",
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
assert!(merge_request::migrate(&c).is_err());
|
||||
for table in ["merge_request_schema_migrations", "merge_requests"] {
|
||||
let exists: bool = c
|
||||
.query_row(
|
||||
"SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name=?1)",
|
||||
[table],
|
||||
|r| r.get(0),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(exists, "{table} was not rolled back");
|
||||
}
|
||||
let current_marker: bool = c
|
||||
.query_row(
|
||||
"SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name='merge_request_schema')",
|
||||
[],
|
||||
|r| r.get(0),
|
||||
)
|
||||
merge_request::migrate(&c).unwrap();
|
||||
c.execute_batch("DROP TABLE merge_request_review_grants;")
|
||||
.unwrap();
|
||||
assert!(!current_marker);
|
||||
|
||||
let error = merge_request::migrate(&c).unwrap_err();
|
||||
assert!(matches!(
|
||||
error,
|
||||
MergeRequestError::Corrupt(message)
|
||||
if message == "missing `merge_request_review_grants`"
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -0,0 +1,122 @@
|
||||
-- Canonical standalone Ticket schema. Workspace Server composes stricter cross-domain authority.
|
||||
CREATE TABLE typed_ticket_artifacts (
|
||||
workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL, relative_path TEXT NOT NULL, content BLOB NOT NULL,
|
||||
PRIMARY KEY (workspace_id, ticket_id, relative_path),
|
||||
FOREIGN KEY (workspace_id, ticket_id) REFERENCES typed_tickets(workspace_id, ticket_id) ON DELETE CASCADE
|
||||
);
|
||||
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),
|
||||
FOREIGN KEY (workspace_id, ticket_id, event_index) REFERENCES typed_ticket_events(workspace_id, ticket_id, event_index) ON DELETE CASCADE
|
||||
);
|
||||
CREATE TABLE typed_ticket_event_references (
|
||||
workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL, event_index INTEGER NOT NULL, ordinal INTEGER NOT NULL, kind TEXT NOT NULL, target TEXT NOT NULL,
|
||||
PRIMARY KEY (workspace_id, ticket_id, event_index, ordinal),
|
||||
FOREIGN KEY (workspace_id, ticket_id, event_index) REFERENCES typed_ticket_events(workspace_id, ticket_id, event_index) ON DELETE CASCADE
|
||||
);
|
||||
CREATE TABLE typed_ticket_events (
|
||||
workspace_id TEXT NOT NULL,
|
||||
ticket_id TEXT NOT NULL,
|
||||
event_index INTEGER NOT NULL,
|
||||
kind TEXT NOT NULL,
|
||||
author TEXT,
|
||||
at TEXT,
|
||||
status TEXT,
|
||||
from_state TEXT,
|
||||
to_state TEXT,
|
||||
reason TEXT,
|
||||
state_field TEXT,
|
||||
heading TEXT,
|
||||
body TEXT NOT NULL,
|
||||
PRIMARY KEY (workspace_id, ticket_id, event_index),
|
||||
FOREIGN KEY (workspace_id, ticket_id) REFERENCES typed_tickets(workspace_id, ticket_id) ON DELETE CASCADE
|
||||
);
|
||||
CREATE TABLE typed_ticket_labels (
|
||||
workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL, ordinal INTEGER NOT NULL, label TEXT NOT NULL,
|
||||
PRIMARY KEY (workspace_id, ticket_id, ordinal),
|
||||
FOREIGN KEY (workspace_id, ticket_id) REFERENCES typed_tickets(workspace_id, ticket_id) ON DELETE CASCADE
|
||||
);
|
||||
CREATE TABLE typed_ticket_orchestration_plans (
|
||||
workspace_id TEXT NOT NULL,
|
||||
ticket_id TEXT NOT NULL,
|
||||
record_id TEXT NOT NULL,
|
||||
kind TEXT NOT NULL,
|
||||
related_ticket TEXT,
|
||||
note TEXT,
|
||||
accepted_summary TEXT,
|
||||
accepted_branch TEXT,
|
||||
accepted_worktree TEXT,
|
||||
accepted_role_plan TEXT,
|
||||
author TEXT NOT NULL,
|
||||
at TEXT NOT NULL,
|
||||
PRIMARY KEY (workspace_id, ticket_id, record_id),
|
||||
FOREIGN KEY (workspace_id, ticket_id) REFERENCES typed_tickets(workspace_id, ticket_id) ON DELETE CASCADE
|
||||
);
|
||||
CREATE TABLE typed_ticket_raw_frontmatter (
|
||||
workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL, key TEXT NOT NULL, value TEXT NOT NULL,
|
||||
PRIMARY KEY (workspace_id, ticket_id, key),
|
||||
FOREIGN KEY (workspace_id, ticket_id) REFERENCES typed_tickets(workspace_id, ticket_id) ON DELETE CASCADE
|
||||
);
|
||||
CREATE TABLE typed_ticket_relations (
|
||||
workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL, kind TEXT NOT NULL, target TEXT NOT NULL, note TEXT, author TEXT NOT NULL, at TEXT NOT NULL,
|
||||
PRIMARY KEY (workspace_id, ticket_id, kind, target),
|
||||
FOREIGN KEY (workspace_id, ticket_id) REFERENCES typed_tickets(workspace_id, ticket_id) ON DELETE CASCADE
|
||||
);
|
||||
CREATE TABLE typed_ticket_risk_flags (
|
||||
workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL, ordinal INTEGER NOT NULL, risk_flag TEXT NOT NULL,
|
||||
PRIMARY KEY (workspace_id, ticket_id, ordinal),
|
||||
FOREIGN KEY (workspace_id, ticket_id) REFERENCES typed_tickets(workspace_id, ticket_id) ON DELETE CASCADE
|
||||
);
|
||||
CREATE TABLE typed_tickets (
|
||||
workspace_id TEXT NOT NULL,
|
||||
ticket_id TEXT NOT NULL,
|
||||
slug TEXT NOT NULL,
|
||||
title TEXT NOT NULL,
|
||||
status TEXT NOT NULL,
|
||||
kind TEXT NOT NULL,
|
||||
priority TEXT NOT NULL,
|
||||
body TEXT NOT NULL,
|
||||
created_at TEXT,
|
||||
updated_at TEXT,
|
||||
assignee TEXT,
|
||||
readiness TEXT,
|
||||
workflow_state TEXT NOT NULL,
|
||||
workflow_state_explicit INTEGER NOT NULL,
|
||||
queued_by TEXT,
|
||||
queued_at TEXT,
|
||||
resolution TEXT, repository_id TEXT, ref_selector TEXT,
|
||||
PRIMARY KEY (workspace_id, ticket_id)
|
||||
);
|
||||
CREATE TABLE "workspace_resource_key_counters" (
|
||||
workspace_id TEXT NOT NULL,
|
||||
resource_kind TEXT NOT NULL CHECK (resource_kind IN ('ticket', 'objective', 'worker')),
|
||||
next_sequence INTEGER NOT NULL CHECK (next_sequence > 0),
|
||||
PRIMARY KEY (workspace_id, resource_kind)
|
||||
);
|
||||
CREATE TABLE "workspace_resource_keys" (
|
||||
workspace_id TEXT NOT NULL,
|
||||
resource_kind TEXT NOT NULL CHECK (resource_kind IN ('ticket', 'objective', 'worker')),
|
||||
resource_id TEXT NOT NULL,
|
||||
sequence INTEGER NOT NULL CHECK (sequence > 0),
|
||||
resource_key TEXT NOT NULL,
|
||||
allocated_at TEXT NOT NULL,
|
||||
PRIMARY KEY (workspace_id, resource_kind, resource_id),
|
||||
UNIQUE (workspace_id, resource_kind, sequence),
|
||||
UNIQUE (workspace_id, resource_key)
|
||||
);
|
||||
CREATE INDEX idx_workspace_resource_keys_reverse
|
||||
ON workspace_resource_keys(workspace_id, resource_kind, resource_key);
|
||||
CREATE INDEX typed_ticket_events_workspace_kind_ticket
|
||||
ON typed_ticket_events(workspace_id, kind, ticket_id, event_index);
|
||||
CREATE INDEX typed_ticket_relations_workspace_source_kind
|
||||
ON typed_ticket_relations(workspace_id, ticket_id, kind, target);
|
||||
CREATE INDEX typed_ticket_relations_workspace_target_kind
|
||||
ON typed_ticket_relations(workspace_id, target, kind, ticket_id);
|
||||
CREATE INDEX typed_tickets_workspace_created
|
||||
ON typed_tickets(workspace_id, created_at DESC, ticket_id);
|
||||
CREATE INDEX typed_tickets_workspace_state_updated
|
||||
ON typed_tickets(workspace_id, workflow_state, updated_at DESC, ticket_id);
|
||||
CREATE INDEX typed_tickets_workspace_title
|
||||
ON typed_tickets(workspace_id, title COLLATE NOCASE, ticket_id);
|
||||
CREATE INDEX typed_tickets_workspace_updated
|
||||
ON typed_tickets(workspace_id, updated_at DESC, ticket_id);
|
||||
@@ -26,11 +26,7 @@ pub mod config;
|
||||
mod sqlite_schema;
|
||||
pub mod tool;
|
||||
|
||||
pub use sqlite_schema::{
|
||||
LATEST_SQLITE_TICKET_SCHEMA_VERSION, migrate_sqlite_ticket_resource_key_schema_in_transaction,
|
||||
migrate_sqlite_ticket_schema, migrate_sqlite_ticket_schema_through,
|
||||
verify_sqlite_ticket_schema,
|
||||
};
|
||||
pub use sqlite_schema::{migrate_sqlite_ticket_schema, verify_sqlite_ticket_schema};
|
||||
|
||||
const REQUIRED_FIELDS: [&str; 4] = ["title", "state", "created_at", "updated_at"];
|
||||
const MAX_STATE_CHANGE_REASON_BYTES: usize = 1024;
|
||||
@@ -2576,7 +2572,7 @@ impl SqliteTicketBackend {
|
||||
}
|
||||
}
|
||||
|
||||
/// Opens a standalone Ticket backend, applying all Ticket-owned migrations once.
|
||||
/// Opens a standalone Ticket backend at the current canonical schema baseline.
|
||||
pub fn open(db_path: impl Into<PathBuf>, workspace_id: impl Into<String>) -> Result<Self> {
|
||||
let backend = Self::configured(db_path, workspace_id);
|
||||
let connection = backend.connect()?;
|
||||
|
||||
@@ -7,7 +7,7 @@ use crate::{Result, TicketError, sqlite_err};
|
||||
|
||||
const MIGRATION_TABLE: &str = "ticket_schema_migrations";
|
||||
const MAX_SCHEMA_DIAGNOSTICS: usize = 32;
|
||||
pub const LATEST_SQLITE_TICKET_SCHEMA_VERSION: i64 = 6;
|
||||
const LATEST_SQLITE_TICKET_SCHEMA_VERSION: i64 = 6;
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
struct Migration {
|
||||
@@ -16,38 +16,11 @@ struct Migration {
|
||||
apply: fn(&Connection) -> Result<()>,
|
||||
}
|
||||
|
||||
const MIGRATIONS: &[Migration] = &[
|
||||
Migration {
|
||||
version: 1,
|
||||
name: "create_typed_ticket_tables",
|
||||
apply: create_typed_ticket_tables,
|
||||
},
|
||||
Migration {
|
||||
version: 2,
|
||||
name: "add_ticket_repository_target",
|
||||
apply: add_ticket_repository_target,
|
||||
},
|
||||
Migration {
|
||||
version: 3,
|
||||
name: "convert_legacy_reviews_to_comments",
|
||||
apply: retire_legacy_ticket_review_events,
|
||||
},
|
||||
Migration {
|
||||
version: 4,
|
||||
name: "add_ticket_query_indexes",
|
||||
apply: add_ticket_query_indexes,
|
||||
},
|
||||
Migration {
|
||||
version: 5,
|
||||
name: "add_workspace_human_keys",
|
||||
apply: add_workspace_human_keys,
|
||||
},
|
||||
Migration {
|
||||
version: 6,
|
||||
name: "rename_workspace_resource_keys",
|
||||
apply: rename_workspace_resource_keys,
|
||||
},
|
||||
];
|
||||
const MIGRATIONS: &[Migration] = &[Migration {
|
||||
version: LATEST_SQLITE_TICKET_SCHEMA_VERSION,
|
||||
name: "ticket schema baseline",
|
||||
apply: create_latest_ticket_schema,
|
||||
}];
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
struct ExpectedColumn {
|
||||
@@ -258,30 +231,12 @@ const fn column(
|
||||
}
|
||||
}
|
||||
|
||||
/// Applies the Ticket crate's SQLite migrations and verifies the resulting schema.
|
||||
/// Creates and verifies the Ticket crate's latest SQLite schema.
|
||||
///
|
||||
/// This is a startup/standalone-open operation. Normal Ticket request handling must
|
||||
/// use [`verify_sqlite_ticket_schema`] instead, so request paths never acquire DDL
|
||||
/// authority.
|
||||
pub fn migrate_sqlite_ticket_schema(connection: &Connection) -> Result<()> {
|
||||
migrate_sqlite_ticket_schema_through(connection, LATEST_SQLITE_TICKET_SCHEMA_VERSION)
|
||||
}
|
||||
|
||||
/// Applies Ticket migrations only through `target_version`.
|
||||
///
|
||||
/// This exists for the Workspace Server's ordered migration bridge: older Server
|
||||
/// migrations must materialize the Ticket schema shape they were written against
|
||||
/// before the current Ticket migration is applied at the matching Server version.
|
||||
#[doc(hidden)]
|
||||
pub fn migrate_sqlite_ticket_schema_through(
|
||||
connection: &Connection,
|
||||
target_version: i64,
|
||||
) -> Result<()> {
|
||||
if !(1..=LATEST_SQLITE_TICKET_SCHEMA_VERSION).contains(&target_version) {
|
||||
return Err(TicketError::Sqlite(format!(
|
||||
"unsupported Ticket schema migration target {target_version}"
|
||||
)));
|
||||
}
|
||||
connection
|
||||
.busy_timeout(Duration::from_secs(5))
|
||||
.map_err(sqlite_err)?;
|
||||
@@ -302,25 +257,10 @@ pub fn migrate_sqlite_ticket_schema_through(
|
||||
verify_table(connection, MIGRATION_TABLE, MIGRATION_COLUMNS, &[], false)?;
|
||||
|
||||
let applied = load_applied_migrations(connection)?;
|
||||
validate_applied_migrations(&applied)?;
|
||||
|
||||
if let Some(version) = applied
|
||||
.keys()
|
||||
.copied()
|
||||
.find(|version| *version > target_version)
|
||||
{
|
||||
return Err(TicketError::Sqlite(format!(
|
||||
"Ticket schema version {version} is newer than requested migration target {target_version}"
|
||||
)));
|
||||
}
|
||||
|
||||
for migration in MIGRATIONS
|
||||
.iter()
|
||||
.filter(|migration| migration.version <= target_version)
|
||||
{
|
||||
if applied.contains_key(&migration.version) {
|
||||
continue;
|
||||
}
|
||||
if applied.is_empty() {
|
||||
let migration = MIGRATIONS
|
||||
.first()
|
||||
.ok_or_else(|| TicketError::Sqlite("Ticket migration catalog is empty".into()))?;
|
||||
(migration.apply)(connection)?;
|
||||
connection
|
||||
.execute(
|
||||
@@ -333,24 +273,11 @@ pub fn migrate_sqlite_ticket_schema_through(
|
||||
],
|
||||
)
|
||||
.map_err(sqlite_err)?;
|
||||
} else {
|
||||
validate_applied_migrations(&applied)?;
|
||||
}
|
||||
|
||||
if target_version == LATEST_SQLITE_TICKET_SCHEMA_VERSION {
|
||||
verify_sqlite_ticket_schema(connection)
|
||||
} else {
|
||||
let applied = load_applied_migrations(connection)?;
|
||||
let expected = MIGRATIONS
|
||||
.iter()
|
||||
.filter(|migration| migration.version <= target_version)
|
||||
.map(|migration| (migration.version, migration.name.to_string()))
|
||||
.collect::<BTreeMap<_, _>>();
|
||||
if applied != expected {
|
||||
return Err(TicketError::Sqlite(format!(
|
||||
"Ticket schema migration history does not match target version {target_version}"
|
||||
)));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
verify_sqlite_ticket_schema(connection)
|
||||
})();
|
||||
|
||||
match result {
|
||||
@@ -362,47 +289,6 @@ pub fn migrate_sqlite_ticket_schema_through(
|
||||
}
|
||||
}
|
||||
|
||||
/// Applies the resource-key Ticket migration inside a transaction owned by the
|
||||
/// Workspace Server. The caller must provide an active transaction; this function
|
||||
/// deliberately does not begin or commit one so the Ticket and Server migration
|
||||
/// markers can be persisted atomically.
|
||||
#[doc(hidden)]
|
||||
pub fn migrate_sqlite_ticket_resource_key_schema_in_transaction(
|
||||
connection: &Connection,
|
||||
) -> Result<()> {
|
||||
connection
|
||||
.execute_batch(
|
||||
"CREATE TABLE IF NOT EXISTS ticket_schema_migrations (
|
||||
version INTEGER PRIMARY KEY,
|
||||
name TEXT NOT NULL,
|
||||
applied_at TEXT NOT NULL
|
||||
);",
|
||||
)
|
||||
.map_err(sqlite_err)?;
|
||||
let applied = load_applied_migrations(connection)?;
|
||||
validate_applied_migrations(&applied)?;
|
||||
if applied.contains_key(&LATEST_SQLITE_TICKET_SCHEMA_VERSION) {
|
||||
return verify_sqlite_ticket_schema(connection);
|
||||
}
|
||||
let expected_previous = LATEST_SQLITE_TICKET_SCHEMA_VERSION - 1;
|
||||
if applied.len() != expected_previous as usize || !applied.contains_key(&expected_previous) {
|
||||
return Err(TicketError::Sqlite(format!(
|
||||
"Ticket schema must be at version {expected_previous} before the resource-key migration"
|
||||
)));
|
||||
}
|
||||
let migration = MIGRATIONS
|
||||
.last()
|
||||
.ok_or_else(|| TicketError::Sqlite("Ticket migration catalog is empty".to_string()))?;
|
||||
(migration.apply)(connection)?;
|
||||
connection
|
||||
.execute(
|
||||
"INSERT INTO ticket_schema_migrations (version, name, applied_at) VALUES (?1, ?2, datetime('now'))",
|
||||
params![migration.version, migration.name],
|
||||
)
|
||||
.map_err(sqlite_err)?;
|
||||
verify_sqlite_ticket_schema(connection)
|
||||
}
|
||||
|
||||
/// Verifies the current Ticket-owned SQLite schema without executing DDL.
|
||||
pub fn verify_sqlite_ticket_schema(connection: &Connection) -> Result<()> {
|
||||
let mut diagnostics = Vec::new();
|
||||
@@ -539,238 +425,9 @@ pub fn verify_sqlite_ticket_schema(connection: &Connection) -> Result<()> {
|
||||
}
|
||||
}
|
||||
|
||||
fn create_typed_ticket_tables(connection: &Connection) -> Result<()> {
|
||||
fn create_latest_ticket_schema(connection: &Connection) -> Result<()> {
|
||||
connection
|
||||
.execute_batch(
|
||||
r#"
|
||||
CREATE TABLE IF NOT EXISTS typed_tickets (
|
||||
workspace_id TEXT NOT NULL,
|
||||
ticket_id TEXT NOT NULL,
|
||||
slug TEXT NOT NULL,
|
||||
title TEXT NOT NULL,
|
||||
status TEXT NOT NULL,
|
||||
kind TEXT NOT NULL,
|
||||
priority TEXT NOT NULL,
|
||||
body TEXT NOT NULL,
|
||||
created_at TEXT,
|
||||
updated_at TEXT,
|
||||
assignee TEXT,
|
||||
readiness TEXT,
|
||||
workflow_state TEXT NOT NULL,
|
||||
workflow_state_explicit INTEGER NOT NULL,
|
||||
queued_by TEXT,
|
||||
queued_at TEXT,
|
||||
resolution TEXT,
|
||||
PRIMARY KEY (workspace_id, ticket_id)
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS typed_ticket_labels (
|
||||
workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL, ordinal INTEGER NOT NULL, label TEXT NOT NULL,
|
||||
PRIMARY KEY (workspace_id, ticket_id, ordinal),
|
||||
FOREIGN KEY (workspace_id, ticket_id) REFERENCES typed_tickets(workspace_id, ticket_id) ON DELETE CASCADE
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS typed_ticket_risk_flags (
|
||||
workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL, ordinal INTEGER NOT NULL, risk_flag TEXT NOT NULL,
|
||||
PRIMARY KEY (workspace_id, ticket_id, ordinal),
|
||||
FOREIGN KEY (workspace_id, ticket_id) REFERENCES typed_tickets(workspace_id, ticket_id) ON DELETE CASCADE
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS typed_ticket_raw_frontmatter (
|
||||
workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL, key TEXT NOT NULL, value TEXT NOT NULL,
|
||||
PRIMARY KEY (workspace_id, ticket_id, key),
|
||||
FOREIGN KEY (workspace_id, ticket_id) REFERENCES typed_tickets(workspace_id, ticket_id) ON DELETE CASCADE
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS typed_ticket_events (
|
||||
workspace_id TEXT NOT NULL,
|
||||
ticket_id TEXT NOT NULL,
|
||||
event_index INTEGER NOT NULL,
|
||||
kind TEXT NOT NULL,
|
||||
author TEXT,
|
||||
at TEXT,
|
||||
status TEXT,
|
||||
from_state TEXT,
|
||||
to_state TEXT,
|
||||
reason TEXT,
|
||||
state_field TEXT,
|
||||
heading TEXT,
|
||||
body TEXT NOT NULL,
|
||||
PRIMARY KEY (workspace_id, ticket_id, event_index),
|
||||
FOREIGN KEY (workspace_id, ticket_id) REFERENCES typed_tickets(workspace_id, ticket_id) ON DELETE CASCADE
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS typed_ticket_event_references (
|
||||
workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL, event_index INTEGER NOT NULL, ordinal INTEGER NOT NULL, kind TEXT NOT NULL, target TEXT NOT NULL,
|
||||
PRIMARY KEY (workspace_id, ticket_id, event_index, ordinal),
|
||||
FOREIGN KEY (workspace_id, ticket_id, event_index) REFERENCES typed_ticket_events(workspace_id, ticket_id, event_index) ON DELETE CASCADE
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS 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),
|
||||
FOREIGN KEY (workspace_id, ticket_id, event_index) REFERENCES typed_ticket_events(workspace_id, ticket_id, event_index) ON DELETE CASCADE
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS typed_ticket_relations (
|
||||
workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL, kind TEXT NOT NULL, target TEXT NOT NULL, note TEXT, author TEXT NOT NULL, at TEXT NOT NULL,
|
||||
PRIMARY KEY (workspace_id, ticket_id, kind, target),
|
||||
FOREIGN KEY (workspace_id, ticket_id) REFERENCES typed_tickets(workspace_id, ticket_id) ON DELETE CASCADE
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS typed_ticket_orchestration_plans (
|
||||
workspace_id TEXT NOT NULL,
|
||||
ticket_id TEXT NOT NULL,
|
||||
record_id TEXT NOT NULL,
|
||||
kind TEXT NOT NULL,
|
||||
related_ticket TEXT,
|
||||
note TEXT,
|
||||
accepted_summary TEXT,
|
||||
accepted_branch TEXT,
|
||||
accepted_worktree TEXT,
|
||||
accepted_role_plan TEXT,
|
||||
author TEXT NOT NULL,
|
||||
at TEXT NOT NULL,
|
||||
PRIMARY KEY (workspace_id, ticket_id, record_id),
|
||||
FOREIGN KEY (workspace_id, ticket_id) REFERENCES typed_tickets(workspace_id, ticket_id) ON DELETE CASCADE
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS typed_ticket_artifacts (
|
||||
workspace_id TEXT NOT NULL, ticket_id TEXT NOT NULL, relative_path TEXT NOT NULL, content BLOB NOT NULL,
|
||||
PRIMARY KEY (workspace_id, ticket_id, relative_path),
|
||||
FOREIGN KEY (workspace_id, ticket_id) REFERENCES typed_tickets(workspace_id, ticket_id) ON DELETE CASCADE
|
||||
);
|
||||
"#,
|
||||
)
|
||||
.map_err(sqlite_err)
|
||||
}
|
||||
|
||||
fn add_ticket_repository_target(connection: &Connection) -> Result<()> {
|
||||
add_column_if_missing(connection, "typed_tickets", "repository_id", "TEXT")?;
|
||||
add_column_if_missing(connection, "typed_tickets", "ref_selector", "TEXT")
|
||||
}
|
||||
|
||||
fn retire_legacy_ticket_review_events(connection: &Connection) -> Result<()> {
|
||||
// Historical prose remains visible for audit, but it is explicitly converted to a
|
||||
// non-authoritative comment. Approval authority now lives only in Merge Requests.
|
||||
connection
|
||||
.execute_batch(
|
||||
r#"
|
||||
INSERT OR REPLACE INTO typed_ticket_event_attributes
|
||||
(workspace_id, ticket_id, event_index, key, value)
|
||||
SELECT workspace_id, ticket_id, event_index, 'legacy_event_kind', 'review'
|
||||
FROM typed_ticket_events WHERE kind = 'review';
|
||||
UPDATE typed_ticket_events
|
||||
SET kind = 'comment', status = NULL, heading = 'Legacy review (non-authoritative)'
|
||||
WHERE kind = 'review';
|
||||
DELETE FROM typed_ticket_event_attributes
|
||||
WHERE key IN ('result', 'review_result', 'status')
|
||||
AND EXISTS (
|
||||
SELECT 1 FROM typed_ticket_events event
|
||||
WHERE event.workspace_id = typed_ticket_event_attributes.workspace_id
|
||||
AND event.ticket_id = typed_ticket_event_attributes.ticket_id
|
||||
AND event.event_index = typed_ticket_event_attributes.event_index
|
||||
AND event.heading = 'Legacy review (non-authoritative)'
|
||||
);
|
||||
"#,
|
||||
)
|
||||
.map_err(sqlite_err)
|
||||
}
|
||||
|
||||
fn add_ticket_query_indexes(connection: &Connection) -> Result<()> {
|
||||
connection
|
||||
.execute_batch(
|
||||
r#"
|
||||
CREATE INDEX IF NOT EXISTS typed_tickets_workspace_state_updated
|
||||
ON typed_tickets(workspace_id, workflow_state, updated_at DESC, ticket_id);
|
||||
CREATE INDEX IF NOT EXISTS typed_tickets_workspace_updated
|
||||
ON typed_tickets(workspace_id, updated_at DESC, ticket_id);
|
||||
CREATE INDEX IF NOT EXISTS typed_tickets_workspace_created
|
||||
ON typed_tickets(workspace_id, created_at DESC, ticket_id);
|
||||
CREATE INDEX IF NOT EXISTS typed_tickets_workspace_title
|
||||
ON typed_tickets(workspace_id, title COLLATE NOCASE, ticket_id);
|
||||
CREATE INDEX IF NOT EXISTS typed_ticket_events_workspace_kind_ticket
|
||||
ON typed_ticket_events(workspace_id, kind, ticket_id, event_index);
|
||||
CREATE INDEX IF NOT EXISTS typed_ticket_relations_workspace_source_kind
|
||||
ON typed_ticket_relations(workspace_id, ticket_id, kind, target);
|
||||
CREATE INDEX IF NOT EXISTS typed_ticket_relations_workspace_target_kind
|
||||
ON typed_ticket_relations(workspace_id, target, kind, ticket_id);
|
||||
"#,
|
||||
)
|
||||
.map_err(sqlite_err)
|
||||
}
|
||||
|
||||
fn add_workspace_human_keys(connection: &Connection) -> Result<()> {
|
||||
connection
|
||||
.execute_batch(
|
||||
r#"
|
||||
CREATE TABLE IF NOT EXISTS workspace_resource_human_keys (
|
||||
workspace_id TEXT NOT NULL,
|
||||
resource_kind TEXT NOT NULL CHECK (resource_kind IN ('ticket', 'objective', 'worker')),
|
||||
resource_id TEXT NOT NULL,
|
||||
sequence INTEGER NOT NULL CHECK (sequence > 0),
|
||||
human_key TEXT NOT NULL,
|
||||
allocated_at TEXT NOT NULL,
|
||||
PRIMARY KEY (workspace_id, resource_kind, resource_id),
|
||||
UNIQUE (workspace_id, resource_kind, sequence),
|
||||
UNIQUE (workspace_id, human_key)
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS workspace_resource_human_key_counters (
|
||||
workspace_id TEXT NOT NULL,
|
||||
resource_kind TEXT NOT NULL CHECK (resource_kind IN ('ticket', 'objective', 'worker')),
|
||||
next_sequence INTEGER NOT NULL CHECK (next_sequence > 0),
|
||||
PRIMARY KEY (workspace_id, resource_kind)
|
||||
);
|
||||
|
||||
INSERT OR IGNORE INTO workspace_resource_human_keys (
|
||||
workspace_id, resource_kind, resource_id, sequence, human_key, allocated_at
|
||||
)
|
||||
SELECT workspace_id,
|
||||
'ticket',
|
||||
ticket_id,
|
||||
ROW_NUMBER() OVER (
|
||||
PARTITION BY workspace_id ORDER BY created_at ASC, ticket_id ASC
|
||||
),
|
||||
'T-' || ROW_NUMBER() OVER (
|
||||
PARTITION BY workspace_id ORDER BY created_at ASC, ticket_id ASC
|
||||
),
|
||||
COALESCE(created_at, updated_at)
|
||||
FROM typed_tickets;
|
||||
|
||||
INSERT INTO workspace_resource_human_key_counters (
|
||||
workspace_id, resource_kind, next_sequence
|
||||
)
|
||||
SELECT workspace_id, 'ticket', MAX(sequence) + 1
|
||||
FROM workspace_resource_human_keys
|
||||
WHERE resource_kind = 'ticket'
|
||||
GROUP BY workspace_id
|
||||
ON CONFLICT(workspace_id, resource_kind) DO UPDATE SET
|
||||
next_sequence = MAX(next_sequence, excluded.next_sequence);
|
||||
"#,
|
||||
)
|
||||
.map_err(sqlite_err)
|
||||
}
|
||||
|
||||
fn rename_workspace_resource_keys(connection: &Connection) -> Result<()> {
|
||||
connection
|
||||
.execute_batch(
|
||||
r#"
|
||||
ALTER TABLE workspace_resource_human_keys RENAME TO workspace_resource_keys;
|
||||
ALTER TABLE workspace_resource_keys RENAME COLUMN human_key TO resource_key;
|
||||
ALTER TABLE workspace_resource_human_key_counters RENAME TO workspace_resource_key_counters;
|
||||
DROP INDEX IF EXISTS idx_workspace_resource_human_keys_reverse;
|
||||
CREATE INDEX idx_workspace_resource_keys_reverse
|
||||
ON workspace_resource_keys(workspace_id, resource_kind, resource_key);
|
||||
"#,
|
||||
)
|
||||
.map_err(sqlite_err)
|
||||
}
|
||||
|
||||
fn add_column_if_missing(
|
||||
connection: &Connection,
|
||||
table: &str,
|
||||
column: &str,
|
||||
declaration: &str,
|
||||
) -> Result<()> {
|
||||
let columns = load_columns(connection, table)?;
|
||||
if columns.iter().any(|found| found.name == column) {
|
||||
return Ok(());
|
||||
}
|
||||
connection
|
||||
.execute_batch(&format!(
|
||||
"ALTER TABLE {table} ADD COLUMN {column} {declaration}"
|
||||
))
|
||||
.execute_batch(include_str!("latest_schema.sql"))
|
||||
.map_err(sqlite_err)
|
||||
}
|
||||
|
||||
@@ -796,33 +453,17 @@ fn load_applied_migrations(connection: &Connection) -> Result<BTreeMap<i64, Stri
|
||||
}
|
||||
|
||||
fn validate_applied_migrations(applied: &BTreeMap<i64, String>) -> Result<()> {
|
||||
for (&version, name) in applied {
|
||||
let Some(expected) = MIGRATIONS
|
||||
.iter()
|
||||
.find(|migration| migration.version == version)
|
||||
else {
|
||||
return Err(TicketError::Sqlite(format!(
|
||||
"unsupported Ticket schema migration version {version}; latest supported version is {LATEST_SQLITE_TICKET_SCHEMA_VERSION}"
|
||||
)));
|
||||
};
|
||||
if name != expected.name {
|
||||
return Err(TicketError::Sqlite(format!(
|
||||
"Ticket schema migration {version} is named {name:?}, expected {:?}",
|
||||
expected.name
|
||||
)));
|
||||
}
|
||||
let expected = BTreeMap::from([(
|
||||
LATEST_SQLITE_TICKET_SCHEMA_VERSION,
|
||||
MIGRATIONS[0].name.to_string(),
|
||||
)]);
|
||||
if applied == &expected {
|
||||
Ok(())
|
||||
} else {
|
||||
Err(TicketError::Sqlite(format!(
|
||||
"Ticket schema migration history must contain only the canonical version {LATEST_SQLITE_TICKET_SCHEMA_VERSION} baseline marker"
|
||||
)))
|
||||
}
|
||||
for migration in MIGRATIONS {
|
||||
if applied.keys().any(|version| *version > migration.version)
|
||||
&& !applied.contains_key(&migration.version)
|
||||
{
|
||||
return Err(TicketError::Sqlite(format!(
|
||||
"Ticket schema migration history has a gap at version {}",
|
||||
migration.version
|
||||
)));
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
@@ -1189,223 +830,16 @@ mod tests {
|
||||
verify_sqlite_ticket_schema(&connection).unwrap();
|
||||
|
||||
let versions = load_applied_migrations(&connection).unwrap();
|
||||
assert_eq!(versions.len(), 6);
|
||||
assert_eq!(
|
||||
versions.get(&LATEST_SQLITE_TICKET_SCHEMA_VERSION),
|
||||
Some(&"rename_workspace_resource_keys".to_string())
|
||||
versions,
|
||||
BTreeMap::from([(
|
||||
LATEST_SQLITE_TICKET_SCHEMA_VERSION,
|
||||
"ticket schema baseline".to_string(),
|
||||
)])
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn adopts_existing_current_schema_without_losing_data() {
|
||||
let connection = Connection::open_in_memory().unwrap();
|
||||
create_typed_ticket_tables(&connection).unwrap();
|
||||
add_ticket_repository_target(&connection).unwrap();
|
||||
connection
|
||||
.execute(
|
||||
"INSERT INTO typed_tickets (
|
||||
workspace_id, ticket_id, slug, title, status, kind, priority, body,
|
||||
workflow_state, workflow_state_explicit, repository_id, ref_selector
|
||||
) VALUES ('workspace-1', 'ticket-1', 'ticket-1', 'kept', 'open',
|
||||
'task', 'medium', 'body', 'ready', 1, 'main', 'develop')",
|
||||
[],
|
||||
)
|
||||
.unwrap();
|
||||
connection
|
||||
.execute_batch(
|
||||
"INSERT INTO typed_ticket_events (
|
||||
workspace_id, ticket_id, event_index, kind, author, at, heading, body
|
||||
) VALUES (
|
||||
'workspace-1', 'ticket-1', 0, 'comment', 'hare',
|
||||
'2026-08-10T00:00:00Z', 'Evidence', 'event kept'
|
||||
);
|
||||
INSERT INTO typed_ticket_event_references (
|
||||
workspace_id, ticket_id, event_index, ordinal, kind, target
|
||||
) VALUES ('workspace-1', 'ticket-1', 0, 0, 'commit', 'abc123');
|
||||
INSERT INTO typed_ticket_relations (
|
||||
workspace_id, ticket_id, kind, target, note, author, at
|
||||
) VALUES (
|
||||
'workspace-1', 'ticket-1', 'related', 'ticket-2', 'relation kept',
|
||||
'hare', '2026-08-10T00:00:00Z'
|
||||
);
|
||||
INSERT INTO typed_ticket_orchestration_plans (
|
||||
workspace_id, ticket_id, record_id, kind, note, author, at
|
||||
) VALUES (
|
||||
'workspace-1', 'ticket-1', 'plan-1', 'waiting_capacity_note',
|
||||
'plan kept', 'hare', '2026-08-10T00:00:00Z'
|
||||
);
|
||||
INSERT INTO typed_ticket_artifacts (
|
||||
workspace_id, ticket_id, relative_path, content
|
||||
) VALUES ('workspace-1', 'ticket-1', 'evidence.txt', X'6b657074');",
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
migrate_sqlite_ticket_schema(&connection).unwrap();
|
||||
|
||||
let row = connection
|
||||
.query_row(
|
||||
"SELECT title, repository_id, ref_selector FROM typed_tickets",
|
||||
[],
|
||||
|row| {
|
||||
Ok((
|
||||
row.get::<_, String>(0)?,
|
||||
row.get::<_, String>(1)?,
|
||||
row.get::<_, String>(2)?,
|
||||
))
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(row, ("kept".into(), "main".into(), "develop".into()));
|
||||
let preserved = connection
|
||||
.query_row(
|
||||
"SELECT
|
||||
(SELECT COUNT(*) FROM typed_ticket_events),
|
||||
(SELECT COUNT(*) FROM typed_ticket_event_references),
|
||||
(SELECT COUNT(*) FROM typed_ticket_relations),
|
||||
(SELECT COUNT(*) FROM typed_ticket_orchestration_plans),
|
||||
(SELECT COUNT(*) FROM typed_ticket_artifacts)",
|
||||
[],
|
||||
|row| {
|
||||
Ok((
|
||||
row.get::<_, i64>(0)?,
|
||||
row.get::<_, i64>(1)?,
|
||||
row.get::<_, i64>(2)?,
|
||||
row.get::<_, i64>(3)?,
|
||||
row.get::<_, i64>(4)?,
|
||||
))
|
||||
},
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(preserved, (1, 1, 1, 1, 1));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn v5_backfills_ticket_keys_and_v6_preserves_them_under_resource_key_schema() {
|
||||
let connection = Connection::open_in_memory().unwrap();
|
||||
migrate_sqlite_ticket_schema_through(&connection, 4).unwrap();
|
||||
connection.execute_batch(
|
||||
"INSERT INTO typed_tickets (
|
||||
workspace_id, ticket_id, slug, title, status, kind, priority, body,
|
||||
workflow_state, workflow_state_explicit, created_at, updated_at
|
||||
) VALUES
|
||||
('workspace-1', 'later', 'later', 'Later', 'open', 'task', 'medium', '', 'ready', 1, '2026-01-02T00:00:00Z', '2026-01-02T00:00:00Z'),
|
||||
('workspace-1', 'earlier', 'earlier', 'Earlier', 'open', 'task', 'medium', '', 'ready', 1, '2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z');"
|
||||
).unwrap();
|
||||
|
||||
migrate_sqlite_ticket_schema_through(&connection, 5).unwrap();
|
||||
let legacy_keys = connection
|
||||
.prepare(
|
||||
"SELECT resource_id, human_key FROM workspace_resource_human_keys
|
||||
WHERE workspace_id = 'workspace-1' AND resource_kind = 'ticket'
|
||||
ORDER BY sequence",
|
||||
)
|
||||
.unwrap()
|
||||
.query_map([], |row| {
|
||||
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
|
||||
})
|
||||
.unwrap()
|
||||
.collect::<std::result::Result<Vec<_>, _>>()
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
legacy_keys,
|
||||
vec![
|
||||
("earlier".into(), "T-1".into()),
|
||||
("later".into(), "T-2".into())
|
||||
]
|
||||
);
|
||||
let next: i64 = connection
|
||||
.query_row(
|
||||
"SELECT next_sequence FROM workspace_resource_human_key_counters
|
||||
WHERE workspace_id = 'workspace-1' AND resource_kind = 'ticket'",
|
||||
[],
|
||||
|row| row.get(0),
|
||||
)
|
||||
.unwrap();
|
||||
assert_eq!(next, 3);
|
||||
|
||||
migrate_sqlite_ticket_schema(&connection).unwrap();
|
||||
let resource_keys = connection
|
||||
.prepare(
|
||||
"SELECT resource_id, resource_key FROM workspace_resource_keys
|
||||
WHERE workspace_id = 'workspace-1' AND resource_kind = 'ticket'
|
||||
ORDER BY sequence",
|
||||
)
|
||||
.unwrap()
|
||||
.query_map([], |row| {
|
||||
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
|
||||
})
|
||||
.unwrap()
|
||||
.collect::<std::result::Result<Vec<_>, _>>()
|
||||
.unwrap();
|
||||
assert_eq!(resource_keys, legacy_keys);
|
||||
assert_eq!(
|
||||
connection
|
||||
.query_row(
|
||||
"SELECT next_sequence FROM workspace_resource_key_counters
|
||||
WHERE workspace_id = 'workspace-1' AND resource_kind = 'ticket'",
|
||||
[],
|
||||
|row| row.get::<_, i64>(0),
|
||||
)
|
||||
.unwrap(),
|
||||
3
|
||||
);
|
||||
for legacy_table in [
|
||||
"workspace_resource_human_keys",
|
||||
"workspace_resource_human_key_counters",
|
||||
] {
|
||||
assert!(
|
||||
connection
|
||||
.query_row(
|
||||
"SELECT 1 FROM sqlite_schema WHERE type = 'table' AND name = ?1",
|
||||
[legacy_table],
|
||||
|_| Ok(()),
|
||||
)
|
||||
.optional()
|
||||
.unwrap()
|
||||
.is_none(),
|
||||
"{legacy_table} still exists"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn upgrades_legacy_schema_without_repository_target_columns() {
|
||||
let connection = Connection::open_in_memory().unwrap();
|
||||
create_typed_ticket_tables(&connection).unwrap();
|
||||
connection
|
||||
.execute(
|
||||
"INSERT INTO typed_tickets (
|
||||
workspace_id, ticket_id, slug, title, status, kind, priority, body,
|
||||
workflow_state, workflow_state_explicit
|
||||
) VALUES ('workspace-1', 'ticket-1', 'ticket-1', 'legacy', 'open',
|
||||
'task', 'medium', 'body', 'ready', 1)",
|
||||
[],
|
||||
)
|
||||
.unwrap();
|
||||
connection
|
||||
.execute_batch(
|
||||
"CREATE TABLE ticket_schema_migrations (
|
||||
version INTEGER PRIMARY KEY,
|
||||
name TEXT NOT NULL,
|
||||
applied_at TEXT NOT NULL
|
||||
);
|
||||
INSERT INTO ticket_schema_migrations (version, name, applied_at)
|
||||
VALUES (1, 'create_typed_ticket_tables', '2026-08-10T00:00:00Z');",
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
migrate_sqlite_ticket_schema(&connection).unwrap();
|
||||
verify_sqlite_ticket_schema(&connection).unwrap();
|
||||
|
||||
let columns = load_columns(&connection, "typed_tickets").unwrap();
|
||||
assert!(columns.iter().any(|column| column.name == "repository_id"));
|
||||
assert!(columns.iter().any(|column| column.name == "ref_selector"));
|
||||
let title = connection
|
||||
.query_row("SELECT title FROM typed_tickets", [], |row| {
|
||||
row.get::<_, String>(0)
|
||||
})
|
||||
.unwrap();
|
||||
assert_eq!(title, "legacy");
|
||||
assert_eq!(load_applied_migrations(&connection).unwrap(), versions);
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -1421,12 +855,32 @@ mod tests {
|
||||
.unwrap();
|
||||
|
||||
let error = migrate_sqlite_ticket_schema(&connection).unwrap_err();
|
||||
assert!(
|
||||
error
|
||||
.to_string()
|
||||
.contains("unsupported Ticket schema migration version 99")
|
||||
);
|
||||
assert_eq!(load_applied_migrations(&connection).unwrap().len(), 7);
|
||||
assert!(error.to_string().contains(
|
||||
"migration history must contain only the canonical version 6 baseline marker"
|
||||
));
|
||||
assert_eq!(load_applied_migrations(&connection).unwrap().len(), 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_legacy_migration_marker() {
|
||||
let connection = Connection::open_in_memory().unwrap();
|
||||
connection
|
||||
.execute_batch(
|
||||
"CREATE TABLE ticket_schema_migrations (
|
||||
version INTEGER PRIMARY KEY,
|
||||
name TEXT NOT NULL,
|
||||
applied_at TEXT NOT NULL
|
||||
);
|
||||
INSERT INTO ticket_schema_migrations (version, name, applied_at)
|
||||
VALUES (6, 'rename_workspace_resource_keys', '2026-08-10T00:00:00Z');",
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let error = migrate_sqlite_ticket_schema(&connection).unwrap_err();
|
||||
assert!(error.to_string().contains(
|
||||
"migration history must contain only the canonical version 6 baseline marker"
|
||||
));
|
||||
assert!(!table_exists(&connection, "typed_tickets").unwrap());
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -1509,77 +963,6 @@ mod tests {
|
||||
verify_sqlite_ticket_schema(&connection).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn migration_rejects_constraint_drift_and_rolls_back_version_adoption() {
|
||||
let connection = Connection::open_in_memory().unwrap();
|
||||
connection
|
||||
.execute_batch(
|
||||
"CREATE TABLE typed_tickets (
|
||||
workspace_id TEXT NOT NULL,
|
||||
ticket_id TEXT NOT NULL,
|
||||
slug TEXT NOT NULL,
|
||||
title TEXT NOT NULL,
|
||||
status TEXT NOT NULL,
|
||||
kind TEXT NOT NULL,
|
||||
priority TEXT NOT NULL,
|
||||
body TEXT NOT NULL,
|
||||
created_at TEXT,
|
||||
updated_at TEXT,
|
||||
assignee TEXT,
|
||||
readiness TEXT,
|
||||
workflow_state TEXT NOT NULL,
|
||||
workflow_state_explicit INTEGER NOT NULL,
|
||||
queued_by TEXT,
|
||||
queued_at TEXT,
|
||||
resolution TEXT,
|
||||
PRIMARY KEY (ticket_id, workspace_id)
|
||||
);",
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let error = migrate_sqlite_ticket_schema(&connection).unwrap_err();
|
||||
assert!(error.to_string().contains("primary-key position"));
|
||||
let migration_table_exists = connection
|
||||
.query_row(
|
||||
"SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'ticket_schema_migrations'",
|
||||
[],
|
||||
|_| Ok(()),
|
||||
)
|
||||
.optional()
|
||||
.unwrap()
|
||||
.is_some();
|
||||
assert!(!migration_table_exists);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn legacy_review_upgrade_preserves_prose_as_non_authoritative_comment() {
|
||||
let connection = Connection::open_in_memory().unwrap();
|
||||
migrate_sqlite_ticket_schema(&connection).unwrap();
|
||||
connection.execute("INSERT INTO typed_tickets (workspace_id,ticket_id,slug,title,status,kind,priority,body,workflow_state,workflow_state_explicit) VALUES ('workspace-1','ticket-1','ticket-1','title','open','task','medium','body','inprogress',1)",[]).unwrap();
|
||||
connection.execute("INSERT INTO typed_ticket_events (workspace_id,ticket_id,event_index,kind,author,at,status,heading,body) VALUES ('workspace-1','ticket-1',0,'review','reviewer','2026-08-11T00:00:00Z','approve','Review','legacy evidence')",[]).unwrap();
|
||||
connection.execute("INSERT INTO typed_ticket_event_attributes (workspace_id,ticket_id,event_index,key,value) VALUES ('workspace-1','ticket-1',0,'result','approve')",[]).unwrap();
|
||||
connection
|
||||
.execute_batch(
|
||||
"DROP TABLE workspace_resource_key_counters;
|
||||
DROP TABLE workspace_resource_keys;
|
||||
DELETE FROM ticket_schema_migrations WHERE version >= 3;",
|
||||
)
|
||||
.unwrap();
|
||||
migrate_sqlite_ticket_schema(&connection).unwrap();
|
||||
let (kind,status,heading,body):(String,Option<String>,Option<String>,Option<String>)=connection.query_row("SELECT kind,status,heading,body FROM typed_ticket_events WHERE workspace_id='workspace-1' AND ticket_id='ticket-1' AND event_index=0",[],|row|Ok((row.get(0)?,row.get(1)?,row.get(2)?,row.get(3)?))).unwrap();
|
||||
assert_eq!(kind, "comment");
|
||||
assert_eq!(status, None);
|
||||
assert_eq!(
|
||||
heading.as_deref(),
|
||||
Some("Legacy review (non-authoritative)")
|
||||
);
|
||||
assert_eq!(body.as_deref(), Some("legacy evidence"));
|
||||
let attributes:i64=connection.query_row("SELECT COUNT(*) FROM typed_ticket_event_attributes WHERE workspace_id='workspace-1' AND ticket_id='ticket-1'",[],|row|row.get(0)).unwrap();
|
||||
assert_eq!(attributes, 1);
|
||||
let legacy:String=connection.query_row("SELECT value FROM typed_ticket_event_attributes WHERE workspace_id='workspace-1' AND ticket_id='ticket-1' AND key='legacy_event_kind'",[],|row|row.get(0)).unwrap();
|
||||
assert_eq!(legacy, "review");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn concurrent_migrators_converge_on_one_version_history() {
|
||||
let directory = tempdir().unwrap();
|
||||
@@ -1602,6 +985,6 @@ mod tests {
|
||||
|
||||
let connection = Connection::open(database).unwrap();
|
||||
verify_sqlite_ticket_schema(&connection).unwrap();
|
||||
assert_eq!(load_applied_migrations(&connection).unwrap().len(), 6);
|
||||
assert_eq!(load_applied_migrations(&connection).unwrap().len(), 1);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2733,6 +2733,18 @@ impl RuntimeState {
|
||||
.unwrap_or_else(|| "worker execution restore failed".to_string());
|
||||
let diagnostic_id = self.next_diagnostic_id;
|
||||
self.next_diagnostic_id += 1;
|
||||
let workspace_id = self
|
||||
.workers
|
||||
.get(&worker_ref.worker_id)
|
||||
.and_then(|worker| worker.workspace_id.as_deref())
|
||||
.unwrap_or("<unscoped>");
|
||||
eprintln!(
|
||||
"yoi-runtime: Worker execution restore failed: diagnostic_id={diagnostic_id} runtime_id={} workspace_id={workspace_id} worker_id={} operation={:?} outcome={:?}: {message}",
|
||||
self.runtime_identity.as_deref().unwrap_or("<unbound>"),
|
||||
worker_ref.worker_id,
|
||||
result.operation,
|
||||
result.outcome,
|
||||
);
|
||||
self.diagnostics.push(RuntimeDiagnostic {
|
||||
id: diagnostic_id,
|
||||
severity: DiagnosticSeverity::Warning,
|
||||
|
||||
@@ -1207,11 +1207,17 @@ impl RepositorySshAgent {
|
||||
working_directory_id: &str,
|
||||
access: &RepositorySshMaterializationAccess,
|
||||
) -> Result<Self, WorkingDirectoryDiagnostic> {
|
||||
let root = runtime_root.join(".repository-agents").join(format!(
|
||||
"{}-{}",
|
||||
sanitize_path_component(working_directory_id),
|
||||
next_working_directory_id("agent")
|
||||
));
|
||||
let mut digest = Sha256::new();
|
||||
digest.update(working_directory_id.as_bytes());
|
||||
digest.update([0]);
|
||||
digest.update(next_working_directory_id("agent").as_bytes());
|
||||
let agent_id = digest
|
||||
.finalize()
|
||||
.iter()
|
||||
.take(8)
|
||||
.map(|byte| format!("{byte:02x}"))
|
||||
.collect::<String>();
|
||||
let root = runtime_root.join(".repository-agents").join(agent_id);
|
||||
fs::create_dir_all(&root).map_err(|_| {
|
||||
WorkingDirectoryDiagnostic::new(
|
||||
"working_directory_repository_agent_failed",
|
||||
@@ -1862,11 +1868,20 @@ impl RepositoryCommandAccess {
|
||||
repository_id: &str,
|
||||
ssh: &RepositorySshMaterializationAccess,
|
||||
) -> Result<Self, WorkingDirectoryDiagnostic> {
|
||||
let root = runtime_root.join(REPOSITORY_ACCESS_DIR).join(format!(
|
||||
"{}-{}",
|
||||
sanitize_path_component(operation_id),
|
||||
next_working_directory_id(repository_id)
|
||||
));
|
||||
let mut digest = Sha256::new();
|
||||
digest.update(operation_id.as_bytes());
|
||||
digest.update([0]);
|
||||
digest.update(repository_id.as_bytes());
|
||||
digest.update([0]);
|
||||
digest.update(next_working_directory_id("access").as_bytes());
|
||||
let access_id = digest
|
||||
.finalize()
|
||||
.iter()
|
||||
.take(8)
|
||||
.map(|byte| format!("{byte:02x}"))
|
||||
.collect::<String>();
|
||||
// Keep operation-scoped Unix socket paths below sockaddr_un::sun_path on deep runtime roots.
|
||||
let root = runtime_root.join(REPOSITORY_ACCESS_DIR).join(access_id);
|
||||
fs::create_dir_all(&root).map_err(|_| {
|
||||
WorkingDirectoryDiagnostic::new(
|
||||
"working_directory_repository_access_setup_failed",
|
||||
|
||||
@@ -1513,33 +1513,6 @@ mod tests {
|
||||
assert_eq!(revision, first.snapshot);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn migration_materializes_main_for_existing_workspace_without_config() {
|
||||
let conn = rusqlite::Connection::open_in_memory().unwrap();
|
||||
crate::store::configure_sqlite(&conn).unwrap();
|
||||
crate::store::apply_migrations_through(&conn, 30).unwrap();
|
||||
conn.execute(
|
||||
"INSERT INTO workspaces (
|
||||
workspace_id, display_name, state, created_at, updated_at
|
||||
) VALUES ('legacy', 'Legacy', 'active', '2026-08-06T00:00:00Z', '2026-08-06T00:00:00Z')",
|
||||
[],
|
||||
)
|
||||
.unwrap();
|
||||
crate::store::persist_workspace_config_schema_bundles(&conn).unwrap();
|
||||
crate::store::materialize_main_config_entrypoint(&conn).unwrap();
|
||||
let state = load_state(&conn, "legacy").unwrap().unwrap();
|
||||
assert!(
|
||||
state
|
||||
.snapshot
|
||||
.entries
|
||||
.contains_key(&path(MAIN_CONFIG_ENTRYPOINT))
|
||||
);
|
||||
assert_eq!(
|
||||
state.contract.entrypoints,
|
||||
vec![path(MAIN_CONFIG_ENTRYPOINT)]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn exports_typescript_transport_contract() {
|
||||
use ts_rs::TS;
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -20,7 +20,6 @@ enum Command {
|
||||
Serve(ServeOptions),
|
||||
Identity(Vec<String>),
|
||||
TrustRuntime(Vec<String>),
|
||||
MigrateDryRun { database: Option<PathBuf> },
|
||||
Skills(SkillsCommand),
|
||||
Help,
|
||||
}
|
||||
@@ -71,17 +70,6 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
|
||||
Command::Serve(options) => run_serve(options).await,
|
||||
Command::Identity(args) => run_identity_command(args),
|
||||
Command::TrustRuntime(args) => run_trust_runtime_command(args),
|
||||
Command::MigrateDryRun { database } => {
|
||||
let database = database.unwrap_or_else(ServerConfig::default_server_database_path);
|
||||
let plan = SqliteWorkspaceStore::migration_plan(&database).map_err(|error| {
|
||||
CliError(format!(
|
||||
"migration dry-run failed for {}: {error}",
|
||||
database.display()
|
||||
))
|
||||
})?;
|
||||
println!("{}", serde_json::to_string_pretty(&plan)?);
|
||||
Ok(())
|
||||
}
|
||||
Command::Skills(command) => run_skills(command),
|
||||
Command::Help => Ok(()),
|
||||
}
|
||||
@@ -96,7 +84,6 @@ fn parse_command(args: &[String]) -> Result<Command, CliError> {
|
||||
match command.as_str() {
|
||||
"identity" => Ok(Command::Identity(rest.to_vec())),
|
||||
"trust-runtime" => Ok(Command::TrustRuntime(rest.to_vec())),
|
||||
"migrate" => parse_migrate_command(rest),
|
||||
"skills" => parse_skills_command(rest),
|
||||
"serve" => {
|
||||
if rest.iter().any(|arg| arg == "--help" || arg == "-h") {
|
||||
@@ -625,32 +612,6 @@ fn workspace_root_from_server_data(workspace: &WorkspaceRecord) -> Result<PathBu
|
||||
))
|
||||
}
|
||||
|
||||
fn parse_migrate_command(args: &[String]) -> Result<Command, CliError> {
|
||||
let mut dry_run = false;
|
||||
let mut database = None;
|
||||
let mut index = 0;
|
||||
while index < args.len() {
|
||||
match args[index].as_str() {
|
||||
"--dry-run" => dry_run = true,
|
||||
"--database" => {
|
||||
index += 1;
|
||||
database =
|
||||
Some(PathBuf::from(args.get(index).ok_or_else(|| {
|
||||
CliError("--database requires a path".to_string())
|
||||
})?));
|
||||
}
|
||||
value => {
|
||||
return Err(CliError(format!("unknown migrate option: {value}")));
|
||||
}
|
||||
}
|
||||
index += 1;
|
||||
}
|
||||
if !dry_run {
|
||||
return Err(CliError("migrate currently requires --dry-run".to_string()));
|
||||
}
|
||||
Ok(Command::MigrateDryRun { database })
|
||||
}
|
||||
|
||||
fn parse_skills_command(args: &[String]) -> Result<Command, CliError> {
|
||||
let Some((subcommand, rest)) = args.split_first() else {
|
||||
print_skills_help();
|
||||
@@ -770,8 +731,7 @@ fn parse_listen(value: &str) -> Result<SocketAddr, CliError> {
|
||||
|
||||
fn print_help() {
|
||||
println!(
|
||||
"yoi-server\n\nUsage:\n yoi-server identity init --server-id <SERVER_ID> [--replace]\n yoi-server identity show [--json]\n yoi-server trust-runtime add --runtime-id <RUNTIME_ID> --workspace-id <WORKSPACE_ID> --base-url <URL> --public-key <KEY> [--display-name <NAME>] [--replace]\n yoi-server trust-runtime list [--json] [--include-revoked]\n yoi-server trust-runtime revoke --runtime-id <RUNTIME_ID>\n yoi-server skills <COMMAND> [OPTIONS]\n yoi-server migrate --dry-run [--database <PATH>]
|
||||
yoi-server serve [OPTIONS]\n\nOptions:\n -h, --help Print help"
|
||||
"yoi-server\n\nUsage:\n yoi-server identity init --server-id <SERVER_ID> [--replace]\n yoi-server identity show [--json]\n yoi-server trust-runtime add --runtime-id <RUNTIME_ID> --workspace-id <WORKSPACE_ID> --base-url <URL> --public-key <KEY> [--display-name <NAME>] [--replace]\n yoi-server trust-runtime list [--json] [--include-revoked]\n yoi-server trust-runtime revoke --runtime-id <RUNTIME_ID>\n yoi-server skills <COMMAND> [OPTIONS]\n yoi-server serve [OPTIONS]\n\nOptions:\n -h, --help Print help"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -783,8 +743,7 @@ fn print_skills_help() {
|
||||
|
||||
fn print_serve_help() {
|
||||
println!(
|
||||
"yoi-server serve\n\nUsage:\n yoi-server migrate --dry-run [--database <PATH>]
|
||||
yoi-server serve [OPTIONS]\n\nDescription:\n Serves Workspaces recorded in the Yoi server DB. Host-level deployment settings are loaded from the explicit --config path or the canonical XDG yoi/server.toml path, and runtime sources are loaded from XDG runtimes.toml.\n\nOptions:\n --listen <ADDR> Listen address (default 127.0.0.1:8787)\n --config <PATH> Host-level Server config path\n -h, --help Print help"
|
||||
"yoi-server serve\n\nUsage:\n yoi-server serve [OPTIONS]\n\nDescription:\n Serves Workspaces recorded in the Yoi server DB. Host-level deployment settings are loaded from the explicit --config path or the canonical XDG yoi/server.toml path, and runtime sources are loaded from XDG runtimes.toml.\n\nOptions:\n --listen <ADDR> Listen address (default 127.0.0.1:8787)\n --config <PATH> Host-level Server config path\n -h, --help Print help"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -823,22 +782,6 @@ mod tests {
|
||||
assert_eq!(name, "debug-rust");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_migrate_requires_dry_run_and_accepts_database_path() {
|
||||
let error = parse_migrate_command(&[]).unwrap_err();
|
||||
assert_eq!(error.to_string(), "migrate currently requires --dry-run");
|
||||
let command = parse_migrate_command(&[
|
||||
"--dry-run".to_string(),
|
||||
"--database".to_string(),
|
||||
"/tmp/server.db".to_string(),
|
||||
])
|
||||
.unwrap();
|
||||
let Command::MigrateDryRun { database } = command else {
|
||||
panic!("expected migration dry-run command");
|
||||
};
|
||||
assert_eq!(database, Some(PathBuf::from("/tmp/server.db")));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_serve_accepts_listen_and_host_config() {
|
||||
let args = vec![
|
||||
|
||||
@@ -163,101 +163,6 @@ pub enum WorkerRetentionError {
|
||||
Invalid(String),
|
||||
}
|
||||
|
||||
pub(crate) fn repair_worker_diagnostics_archive_table(conn: &Connection) -> crate::Result<bool> {
|
||||
let existed: bool = conn.query_row(
|
||||
"SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name='worker_diagnostics_archives')",
|
||||
[],
|
||||
|row| row.get(0),
|
||||
)?;
|
||||
if !existed {
|
||||
conn.execute_batch(
|
||||
"CREATE TABLE worker_diagnostics_archives (
|
||||
operation_id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, runtime_id TEXT NOT NULL,
|
||||
worker_id TEXT NOT NULL, policy_id TEXT NOT NULL, policy_revision INTEGER NOT NULL,
|
||||
committed_at TEXT NOT NULL, expires_at TEXT NOT NULL,
|
||||
FOREIGN KEY(operation_id) REFERENCES worker_removal_operations(operation_id),
|
||||
FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE);",
|
||||
)?;
|
||||
}
|
||||
Ok(!existed)
|
||||
}
|
||||
|
||||
pub(crate) fn create_worker_retention_tables(conn: &Connection) -> crate::Result<()> {
|
||||
conn.execute_batch(r#"
|
||||
CREATE TABLE workspace_worker_retention_policy_revisions (
|
||||
workspace_id TEXT NOT NULL, policy_id TEXT NOT NULL, revision INTEGER NOT NULL CHECK(revision>0),
|
||||
session_disposition TEXT NOT NULL CHECK(session_disposition IN ('archive','purge')),
|
||||
metadata_disposition TEXT NOT NULL CHECK(metadata_disposition IN ('tombstone','purge')),
|
||||
archive_retention_kind TEXT NOT NULL CHECK(archive_retention_kind IN ('forever','for_seconds')),
|
||||
archive_retention_seconds INTEGER,
|
||||
diagnostics_disposition TEXT NOT NULL CHECK(diagnostics_disposition IN ('purge','retain')),
|
||||
diagnostics_retention_seconds INTEGER, created_at TEXT NOT NULL,
|
||||
PRIMARY KEY(workspace_id,policy_id,revision),
|
||||
FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE);
|
||||
CREATE TABLE workspace_worker_retention_policies (
|
||||
workspace_id TEXT PRIMARY KEY, policy_id TEXT NOT NULL, revision INTEGER NOT NULL, updated_at TEXT NOT NULL,
|
||||
FOREIGN KEY(workspace_id,policy_id,revision) REFERENCES workspace_worker_retention_policy_revisions(workspace_id,policy_id,revision),
|
||||
FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE);
|
||||
CREATE TABLE worker_removal_operations (
|
||||
operation_id TEXT PRIMARY KEY, plan_id TEXT NOT NULL UNIQUE, input_fingerprint TEXT NOT NULL,
|
||||
workspace_id TEXT NOT NULL, runtime_id TEXT NOT NULL, worker_id TEXT NOT NULL,
|
||||
worker_revision TEXT NOT NULL, run_generation INTEGER NOT NULL CHECK(run_generation>=0),
|
||||
policy_id TEXT NOT NULL, policy_revision INTEGER NOT NULL,
|
||||
session_disposition TEXT NOT NULL, metadata_disposition TEXT NOT NULL,
|
||||
archive_retention_kind TEXT NOT NULL, archive_retention_seconds INTEGER,
|
||||
diagnostics_disposition TEXT NOT NULL,
|
||||
diagnostics_retention_seconds INTEGER, archive_id TEXT UNIQUE, blockers_json TEXT NOT NULL,
|
||||
state TEXT NOT NULL CHECK(state IN ('planned','blocked','executing','failed','stale','succeeded')),
|
||||
reason TEXT NOT NULL, failure_category TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL,
|
||||
FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE);
|
||||
CREATE INDEX worker_removal_operations_worker_idx ON worker_removal_operations(workspace_id,runtime_id,worker_id,created_at);
|
||||
CREATE TABLE worker_session_archives (
|
||||
archive_id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, runtime_id TEXT NOT NULL, worker_id TEXT NOT NULL,
|
||||
session_id TEXT NOT NULL, checksum_sha256 TEXT NOT NULL, content_bytes INTEGER NOT NULL,
|
||||
policy_id TEXT NOT NULL, policy_revision INTEGER NOT NULL, operation_id TEXT NOT NULL UNIQUE,
|
||||
committed_at TEXT NOT NULL, expires_at TEXT,
|
||||
FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
|
||||
FOREIGN KEY(operation_id) REFERENCES worker_removal_operations(operation_id));
|
||||
CREATE TABLE worker_diagnostics_archives (
|
||||
operation_id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, runtime_id TEXT NOT NULL,
|
||||
worker_id TEXT NOT NULL, policy_id TEXT NOT NULL, policy_revision INTEGER NOT NULL,
|
||||
committed_at TEXT NOT NULL, expires_at TEXT NOT NULL,
|
||||
FOREIGN KEY(operation_id) REFERENCES worker_removal_operations(operation_id),
|
||||
FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE);
|
||||
CREATE TABLE worker_tombstones (
|
||||
workspace_id TEXT NOT NULL, runtime_id TEXT NOT NULL, worker_id TEXT NOT NULL,
|
||||
display_name TEXT NOT NULL, profile TEXT, worker_created_at TEXT NOT NULL, removed_at TEXT NOT NULL,
|
||||
archive_id TEXT, policy_id TEXT NOT NULL, policy_revision INTEGER NOT NULL, operation_id TEXT NOT NULL UNIQUE,
|
||||
PRIMARY KEY(workspace_id,runtime_id,worker_id),
|
||||
FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
|
||||
FOREIGN KEY(archive_id) REFERENCES worker_session_archives(archive_id),
|
||||
FOREIGN KEY(operation_id) REFERENCES worker_removal_operations(operation_id));
|
||||
CREATE TABLE worker_orphan_diagnostics (
|
||||
diagnostic_id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, runtime_id TEXT NOT NULL, worker_id TEXT NOT NULL,
|
||||
category TEXT NOT NULL, detail TEXT NOT NULL, observed_at TEXT NOT NULL,
|
||||
FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE);
|
||||
CREATE TABLE worker_retention_audit_events (
|
||||
event_id TEXT PRIMARY KEY, operation_id TEXT NOT NULL, workspace_id TEXT NOT NULL,
|
||||
event_kind TEXT NOT NULL, detail TEXT NOT NULL, created_at TEXT NOT NULL,
|
||||
FOREIGN KEY(operation_id) REFERENCES worker_removal_operations(operation_id),
|
||||
FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE);
|
||||
CREATE TRIGGER seed_worker_retention_policy_after_workspace_insert AFTER INSERT ON workspaces BEGIN
|
||||
INSERT INTO workspace_worker_retention_policy_revisions
|
||||
(workspace_id,policy_id,revision,session_disposition,metadata_disposition,archive_retention_kind,archive_retention_seconds,diagnostics_disposition,diagnostics_retention_seconds,created_at)
|
||||
VALUES(NEW.workspace_id,'workspace-default-conservative',1,'archive','tombstone','forever',NULL,'purge',NULL,NEW.created_at);
|
||||
INSERT INTO workspace_worker_retention_policies(workspace_id,policy_id,revision,updated_at)
|
||||
VALUES(NEW.workspace_id,'workspace-default-conservative',1,NEW.created_at);
|
||||
END;
|
||||
"#)?;
|
||||
let now = Utc::now().to_rfc3339();
|
||||
conn.execute("INSERT OR IGNORE INTO workspace_worker_retention_policy_revisions
|
||||
(workspace_id,policy_id,revision,session_disposition,metadata_disposition,archive_retention_kind,archive_retention_seconds,diagnostics_disposition,diagnostics_retention_seconds,created_at)
|
||||
SELECT workspace_id,?1,1,'archive','tombstone','forever',NULL,'purge',NULL,?2 FROM workspaces", params![CONSERVATIVE_POLICY_ID,now])?;
|
||||
conn.execute("INSERT OR IGNORE INTO workspace_worker_retention_policies(workspace_id,policy_id,revision,updated_at)
|
||||
SELECT workspace_id,?1,1,?2 FROM workspaces", params![CONSERVATIVE_POLICY_ID,now])?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
impl SqliteWorkspaceStore {
|
||||
pub fn worker_retention_policy(
|
||||
&self,
|
||||
@@ -1724,36 +1629,4 @@ mod tests {
|
||||
);
|
||||
assert_eq!(recovered.plan.state, WorkerRemovalPlanState::Failed);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn old_schema_upgrade_seeds_existing_workspace() {
|
||||
let temp = tempfile::tempdir().unwrap();
|
||||
let path = temp.path().join("server.db");
|
||||
{
|
||||
let connection = rusqlite::Connection::open(&path).unwrap();
|
||||
crate::store::configure_sqlite(&connection).unwrap();
|
||||
crate::store::apply_migrations_through(&connection, 27).unwrap();
|
||||
connection
|
||||
.execute(
|
||||
"INSERT INTO accounts(
|
||||
account_id, kind, handle, display_name, created_at, updated_at
|
||||
) VALUES ('owner-account', 'user', 'owner-account', 'Owner Account', 'old', 'old')",
|
||||
[],
|
||||
)
|
||||
.unwrap();
|
||||
connection
|
||||
.execute(
|
||||
"INSERT INTO workspaces(
|
||||
workspace_id, display_name, state, created_at, updated_at, owner_account_id
|
||||
) VALUES ('legacy', 'Legacy', 'active', 'old', 'old', 'owner-account')",
|
||||
[],
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
let reopened = SqliteWorkspaceStore::open(&path).unwrap();
|
||||
let p = reopened.worker_retention_policy("legacy").unwrap().unwrap();
|
||||
assert_eq!(p.policy_id, CONSERVATIVE_POLICY_ID);
|
||||
assert_eq!(p.session_disposition, SessionDisposition::Archive);
|
||||
assert_eq!(p.metadata_disposition, MetadataDisposition::Tombstone);
|
||||
}
|
||||
}
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -91,44 +91,6 @@ pub struct WorkdirRemovalGuard {
|
||||
pub detail: &'static str,
|
||||
}
|
||||
|
||||
pub(crate) fn create_workdir_removal_operations(conn: &Connection) -> Result<()> {
|
||||
conn.execute_batch(
|
||||
r#"
|
||||
CREATE TABLE workdir_removal_operations (
|
||||
workspace_id TEXT NOT NULL,
|
||||
operation_id TEXT NOT NULL,
|
||||
request_fingerprint TEXT NOT NULL,
|
||||
workdir_id TEXT NOT NULL,
|
||||
runtime_id TEXT NOT NULL,
|
||||
repository_id TEXT NOT NULL,
|
||||
materialization_fingerprint TEXT NOT NULL,
|
||||
source_actor TEXT NOT NULL,
|
||||
reason TEXT NOT NULL,
|
||||
state TEXT NOT NULL CHECK (state IN ('pending', 'failed', 'completed')),
|
||||
attempt_count INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0),
|
||||
retryable INTEGER NOT NULL CHECK (retryable IN (0, 1)),
|
||||
disposition TEXT CHECK (disposition IN ('removed', 'retained', 'attention_required')),
|
||||
failure_category TEXT,
|
||||
attempt_owner_pid INTEGER CHECK (attempt_owner_pid > 0),
|
||||
attempt_owner_start_marker INTEGER CHECK (attempt_owner_start_marker >= 0),
|
||||
created_at TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL,
|
||||
completed_at TEXT,
|
||||
PRIMARY KEY (workspace_id, operation_id),
|
||||
FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE
|
||||
);
|
||||
CREATE UNIQUE INDEX idx_workdir_removal_operations_one_pending
|
||||
ON workdir_removal_operations(workspace_id, workdir_id)
|
||||
WHERE state = 'pending';
|
||||
CREATE INDEX idx_workdir_removal_operations_recovery
|
||||
ON workdir_removal_operations(workspace_id, state, retryable, updated_at);
|
||||
CREATE INDEX idx_workdir_removal_operations_workdir
|
||||
ON workdir_removal_operations(workspace_id, workdir_id, created_at DESC);
|
||||
"#,
|
||||
)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn workdir_materialization_fingerprint(record: &WorkdirRegistryRecord) -> String {
|
||||
let bytes = serde_json::to_vec(&serde_json::json!([
|
||||
record.workspace_id,
|
||||
|
||||
Reference in New Issue
Block a user