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

chore: merge develop into work/companion

Browse Source
This commit is contained in:
Hare
2026-08-21 11:38:47 +09:00
parent 0cd5ca11cc 41cd2d044a
commit 72cae33ea6
34 changed files with 1124 additions and 319 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/authority.rs
+43 -39
View File
@@ -228,10 +228,10 @@ impl SqliteWorkspaceAuthority {
self
}
fn human_key(&self, kind: WorkspaceResourceKind, resource_id: &str) -> Result<String> {
fn resource_key(&self, kind: WorkspaceResourceKind, resource_id: &str) -> Result<String> {
self.store
.resource_human_key(&self.workspace_id, kind, resource_id)?
.ok_or_else(|| Error::Store(format!("missing human key for {resource_id}")))
.resource_key(&self.workspace_id, kind, resource_id)?
.ok_or_else(|| Error::Store(format!("missing resource key for {resource_id}")))
}
fn objective_record(&self, reference: &str) -> Result<ObjectiveRecord> {
@@ -262,7 +262,7 @@ impl SqliteWorkspaceAuthority {
.filter(|ticket| linked_tickets.iter().any(|id| id == &ticket.id))
.map(|ticket| ObjectiveLinkedTicketSummary {
id: ticket.id,
human_key: ticket.human_key,
resource_key: ticket.resource_key,
title: ticket.title,
state: ticket.state,
})
@@ -304,7 +304,8 @@ impl SqliteWorkspaceAuthority {
.unwrap_or("none")
);
Ok(ObjectiveDetail {
human_key: self.human_key(WorkspaceResourceKind::Objective, &record.objective_id)?,
resource_key: self
.resource_key(WorkspaceResourceKind::Objective, &record.objective_id)?,
id: record.objective_id,
title: record.title,
state: record.state,
@@ -746,8 +747,8 @@ impl SqliteWorkspaceAuthority {
.into_iter()
.map(|objective| {
Ok::<_, Error>(ObjectiveLinkSummary {
human_key: self
.human_key(WorkspaceResourceKind::Objective, &objective.objective_id)?,
resource_key: self
.resource_key(WorkspaceResourceKind::Objective, &objective.objective_id)?,
id: objective.objective_id,
title: objective.title,
state: objective.state,
@@ -765,7 +766,7 @@ impl SqliteWorkspaceAuthority {
.store
.get_current_ticket_worker_assignment(&self.workspace_id, id)?
.map(|assignment| {
let worker_human_key = self.store.resource_human_key(
let worker_resource_key = self.store.resource_key(
&self.workspace_id,
WorkspaceResourceKind::Worker,
&assignment.worker.worker_id,
@@ -774,7 +775,7 @@ impl SqliteWorkspaceAuthority {
assignment_id: assignment.assignment_id,
runtime_id: assignment.worker.runtime_id,
worker_id: assignment.worker.worker_id,
worker_human_key,
worker_resource_key,
})
})
.transpose()?;
@@ -802,33 +803,33 @@ impl SqliteWorkspaceAuthority {
.and_then(|event| event.attributes.get("event_id").cloned())
.or_else(|| ticket.meta.updated_at.clone())
.unwrap_or_else(|| format!("{}:0", ticket.meta.id));
let human_key = ticket
let resource_key = ticket
.meta
.human_key
.resource_key
.clone()
.or(self.store.resource_human_key(
.or(self.store.resource_key(
&self.workspace_id,
WorkspaceResourceKind::Ticket,
&ticket.meta.id,
)?)
.ok_or_else(|| Error::Store(format!("missing human key for {}", ticket.meta.id)))?;
.ok_or_else(|| Error::Store(format!("missing resource key for {}", ticket.meta.id)))?;
let mut relations: TicketRelationView = ticket.relations.into();
for relation in &mut relations.outgoing {
relation.target_human_key = self.store.resource_human_key(
relation.target_resource_key = self.store.resource_key(
&self.workspace_id,
WorkspaceResourceKind::Ticket,
&relation.target,
)?;
}
for relation in &mut relations.incoming {
relation.source_human_key = self.store.resource_human_key(
relation.source_resource_key = self.store.resource_key(
&self.workspace_id,
WorkspaceResourceKind::Ticket,
&relation.source_ticket,
)?;
}
for blocker in &mut relations.blockers {
blocker.blocking_human_key = self.store.resource_human_key(
blocker.blocking_resource_key = self.store.resource_key(
&self.workspace_id,
WorkspaceResourceKind::Ticket,
&blocker.blocking_ticket,
@@ -836,7 +837,7 @@ impl SqliteWorkspaceAuthority {
}
Ok(TicketDetail {
id: ticket.meta.id,
human_key,
resource_key,
title: ticket.meta.title,
state: ticket.meta.workflow_state.as_str().to_string(),
readiness: ticket.meta.readiness,
@@ -892,11 +893,11 @@ impl TicketAuthority for SqliteWorkspaceAuthority {
.map(|item| {
let projection =
project_ticket_workspace_item(&item.summary, &item.relation_blockers, None);
let human_key = item.summary.human_key.clone().ok_or_else(|| {
Error::Store(format!("missing human key for {}", item.summary.id))
let resource_key = item.summary.resource_key.clone().ok_or_else(|| {
Error::Store(format!("missing resource key for {}", item.summary.id))
})?;
Ok::<_, Error>(TicketSummary {
human_key,
resource_key,
id: item.summary.id,
title: item.summary.title,
state: item.summary.workflow_state.as_str().to_string(),
@@ -1063,8 +1064,8 @@ impl ObjectiveAuthority for SqliteWorkspaceAuthority {
.map(|link| link.ticket_id)
.collect::<Vec<_>>();
items.push(ObjectiveSummary {
human_key: self
.human_key(WorkspaceResourceKind::Objective, &record.objective_id)?,
resource_key: self
.resource_key(WorkspaceResourceKind::Objective, &record.objective_id)?,
id: record.objective_id,
title: record.title,
state: record.state,
@@ -1110,8 +1111,8 @@ impl ObjectiveAuthority for SqliteWorkspaceAuthority {
.collect::<Vec<_>>();
let body_md = record.body_md.clone();
let objective = ObjectiveSummary {
human_key: self
.human_key(WorkspaceResourceKind::Objective, &record.objective_id)?,
resource_key: self
.resource_key(WorkspaceResourceKind::Objective, &record.objective_id)?,
id: record.objective_id,
title: record.title,
state: record.state,
@@ -2137,7 +2138,7 @@ fn ticket_query_item(
}
TicketQueryItem {
id: summary.id,
human_key: summary.human_key,
resource_key: summary.resource_key,
title: summary.title,
state: summary.state,
readiness: detail.readiness.clone(),
@@ -2288,6 +2289,7 @@ fn objective_query_item(
}
ObjectiveQueryItem {
id: objective.id,
resource_key: objective.resource_key,
title: objective.title,
state: objective.state,
created_at: objective.created_at,
@@ -2480,7 +2482,7 @@ fn memory_resolution_from_record(record: MemoryStagingResolutionRecord) -> Memor
fn ticket_summary_from_ticket(ticket: &ticket::Ticket) -> Result<TicketSummary> {
let summary = ticket::TicketSummary {
id: ticket.meta.id.clone(),
human_key: ticket.meta.human_key.clone(),
resource_key: ticket.meta.resource_key.clone(),
slug: ticket.meta.slug.clone(),
title: ticket.meta.title.clone(),
status: ticket.meta.status.clone(),
@@ -2502,13 +2504,13 @@ fn ticket_summary_from_ticket(ticket: &ticket::Ticket) -> Result<TicketSummary>
fn ticket_summary_from_sqlite_item(item: SqliteTicketListItem) -> Result<TicketSummary> {
let projection = project_ticket_workspace_item(&item.summary, &item.relation_blockers, None);
let human_key = item
let resource_key = item
.summary
.human_key
.resource_key
.clone()
.ok_or_else(|| Error::Store(format!("missing human key for {}", item.summary.id)))?;
.ok_or_else(|| Error::Store(format!("missing resource key for {}", item.summary.id)))?;
Ok(TicketSummary {
human_key,
resource_key,
id: item.summary.id,
title: item.summary.title,
state: item.summary.workflow_state.as_str().to_string(),
@@ -2865,13 +2867,13 @@ mod tests {
.unwrap()
.execute_batch(
r#"
INSERT INTO workspace_resource_human_keys (
workspace_id, resource_kind, resource_id, sequence, human_key, allocated_at
INSERT INTO workspace_resource_keys (
workspace_id, resource_kind, resource_id, sequence, resource_key, allocated_at
) VALUES
('workspace-test', 'ticket', '00000000001J2', 1, 'T-1', '2026-01-01T00:00:00Z'),
('workspace-test', 'ticket', '00000000001J5', 2, 'T-2', '2026-01-01T00:00:00Z'),
('workspace-test', 'ticket', '00000000001J6', 3, 'T-3', '2026-01-01T00:00:00Z');
INSERT INTO workspace_resource_human_key_counters (workspace_id, resource_kind, next_sequence)
INSERT INTO workspace_resource_key_counters (workspace_id, resource_kind, next_sequence)
VALUES ('workspace-test', 'ticket', 4);
"#,
)
@@ -2965,7 +2967,7 @@ VALUES ('workspace-test', 'ticket', 4);
assert_eq!(tickets.items[0].id, "00000000001J2");
assert_eq!(tickets.items[0].state, "ready");
assert_eq!(tickets.items[0].workspace_action_priority, "background");
let ticket_by_key = authority.ticket(&tickets.items[0].human_key).unwrap();
let ticket_by_key = authority.ticket(&tickets.items[0].resource_key).unwrap();
assert_eq!(ticket_by_key.id, tickets.items[0].id);
let ticket = authority.ticket("00000000001J2").unwrap();
@@ -3138,12 +3140,14 @@ VALUES ('workspace-test', 'ticket', 4);
assert_eq!(objectives.items.len(), 1);
assert_eq!(objectives.items[0].id, "00000000001J3");
assert_eq!(objectives.items[0].linked_tickets, vec!["00000000001J2"]);
let objective_by_key = authority.objective(&objectives.items[0].human_key).unwrap();
let objective_by_key = authority
.objective(&objectives.items[0].resource_key)
.unwrap();
assert_eq!(objective_by_key.id, objectives.items[0].id);
assert_eq!(
authority
.show_objective(
&objectives.items[0].human_key,
&objectives.items[0].resource_key,
ObjectiveShowRequest::default(),
)
.unwrap()
@@ -3241,12 +3245,12 @@ INSERT INTO typed_tickets (
) VALUES
('workspace-test', '00000000001J2', 'ticket-j2', 'Ticket J2', 'open', 'task', 'normal', '', 'planning', 1),
('workspace-test', '00000000001J3', 'ticket-j3', 'Ticket J3', 'open', 'task', 'normal', '', 'planning', 1);
INSERT INTO workspace_resource_human_keys (
workspace_id, resource_kind, resource_id, sequence, human_key, allocated_at
INSERT INTO workspace_resource_keys (
workspace_id, resource_kind, resource_id, sequence, resource_key, allocated_at
) VALUES
('workspace-test', 'ticket', '00000000001J2', 1, 'T-1', '2026-01-01T00:00:00Z'),
('workspace-test', 'ticket', '00000000001J3', 2, 'T-2', '2026-01-01T00:00:00Z');
INSERT INTO workspace_resource_human_key_counters (workspace_id, resource_kind, next_sequence)
INSERT INTO workspace_resource_key_counters (workspace_id, resource_kind, next_sequence)
VALUES ('workspace-test', 'ticket', 3);
"#,
)
crates/workspace-server/src/hosts.rs
+7 -7
View File
@@ -247,7 +247,7 @@ pub struct WorkerSummary {
#[serde(flatten)]
pub worker: RuntimeWorkerRef,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub human_key: Option<String>,
pub resource_key: Option<String>,
pub host_id: String,
/// Human-readable display name. This is not identity and may be duplicated.
pub display_name: String,
@@ -1680,7 +1680,7 @@ impl EmbeddedWorkerRuntime {
);
WorkerSummary {
worker: RuntimeWorkerRef::new(&self.runtime_id, worker_id.clone()),
human_key: None,
resource_key: None,
host_id: self.host_id.clone(),
display_name: display.display_name.clone(),
label: display.display_name,
@@ -1720,7 +1720,7 @@ impl EmbeddedWorkerRuntime {
);
WorkerSummary {
worker: RuntimeWorkerRef::new(&self.runtime_id, worker_id.clone()),
human_key: None,
resource_key: None,
host_id: self.host_id.clone(),
display_name: display.display_name.clone(),
label: display.display_name,
@@ -2806,7 +2806,7 @@ impl RemoteWorkerRuntime {
);
WorkerSummary {
worker: RuntimeWorkerRef::new(&self.runtime_id, worker_id.clone()),
human_key: None,
resource_key: None,
host_id: self.host_id.clone(),
display_name: display.display_name.clone(),
label: display.display_name,
@@ -2850,7 +2850,7 @@ impl RemoteWorkerRuntime {
);
WorkerSummary {
worker: RuntimeWorkerRef::new(&self.runtime_id, worker_id.clone()),
human_key: None,
resource_key: None,
host_id: self.host_id.clone(),
display_name: display.display_name.clone(),
label: display.display_name,
@@ -4222,7 +4222,7 @@ pub fn placeholder_worker(host_id: impl Into<String>) -> WorkerSummary {
let host_id = host_id.into();
WorkerSummary {
worker: RuntimeWorkerRef::new("placeholder", "worker-placeholder"),
human_key: None,
resource_key: None,
host_id,
display_name: "Worker runtime actions are not implemented".to_string(),
label: "Worker runtime actions are not implemented".to_string(),
@@ -4616,7 +4616,7 @@ mod tests {
host_id: host_id.to_string(),
workers: vec![WorkerSummary {
worker: RuntimeWorkerRef::new(runtime_id, worker_id),
human_key: None,
resource_key: None,
host_id: host_id.to_string(),
display_name: label.to_string(),
label: label.to_string(),
crates/workspace-server/src/records.rs
+15 -14
View File
@@ -31,7 +31,7 @@ pub struct InvalidProjectRecord {
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
pub struct TicketSummary {
pub id: String,
pub human_key: String,
pub resource_key: String,
pub title: String,
pub state: String,
pub priority: String,
@@ -67,7 +67,7 @@ pub struct TicketListResponse {
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
pub struct TicketDetail {
pub id: String,
pub human_key: String,
pub resource_key: String,
pub title: String,
pub state: String,
pub readiness: Option<String>,
@@ -124,7 +124,7 @@ pub struct TicketRelation {
pub kind: String,
pub target: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub target_human_key: Option<String>,
pub target_resource_key: Option<String>,
pub note: Option<String>,
pub author: String,
pub at: String,
@@ -135,7 +135,7 @@ pub struct TicketRelation {
pub struct DerivedTicketRelation {
pub source_ticket: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub source_human_key: Option<String>,
pub source_resource_key: Option<String>,
pub inverse_kind: String,
pub forward_kind: String,
pub note: Option<String>,
@@ -148,7 +148,7 @@ pub struct DerivedTicketRelation {
pub struct TicketRelationBlocker {
pub blocking_ticket: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub blocking_human_key: Option<String>,
pub blocking_resource_key: Option<String>,
pub reason_kind: String,
pub relation_kind: String,
pub note: Option<String>,
@@ -182,7 +182,7 @@ impl From<ticket::TicketRelationView> for TicketRelationView {
ticket_id: relation.ticket_id,
kind: relation.kind.as_str().to_string(),
target: relation.target,
target_human_key: None,
target_resource_key: None,
note: relation.note,
author: relation.author,
at: relation.at,
@@ -193,7 +193,7 @@ impl From<ticket::TicketRelationView> for TicketRelationView {
.into_iter()
.map(|relation| DerivedTicketRelation {
source_ticket: relation.source_ticket,
source_human_key: None,
source_resource_key: None,
inverse_kind: relation.inverse_kind,
forward_kind: relation.forward_kind.as_str().to_string(),
note: relation.note,
@@ -206,7 +206,7 @@ impl From<ticket::TicketRelationView> for TicketRelationView {
.into_iter()
.map(|blocker| TicketRelationBlocker {
blocking_ticket: blocker.blocking_ticket,
blocking_human_key: None,
blocking_resource_key: None,
reason_kind: blocker.reason_kind,
relation_kind: blocker.relation_kind.as_str().to_string(),
note: blocker.note,
@@ -242,7 +242,7 @@ pub struct QueryPage {
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
pub struct ObjectiveLinkSummary {
pub id: String,
pub human_key: String,
pub resource_key: String,
pub title: String,
pub state: String,
}
@@ -265,7 +265,7 @@ pub struct TicketAssignmentSummary {
pub runtime_id: String,
pub worker_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub worker_human_key: Option<String>,
pub worker_resource_key: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
@@ -342,7 +342,7 @@ pub struct TicketQueryRequest {
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
pub struct TicketQueryItem {
pub id: String,
pub human_key: String,
pub resource_key: String,
pub title: String,
pub state: String,
pub readiness: Option<String>,
@@ -394,6 +394,7 @@ pub struct ObjectiveQueryRequest {
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct ObjectiveQueryItem {
pub id: String,
pub resource_key: String,
pub title: String,
pub state: String,
pub created_at: Option<String>,
@@ -429,7 +430,7 @@ pub struct ObjectiveEventDetail {
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct ObjectiveLinkedTicketSummary {
pub id: String,
pub human_key: String,
pub resource_key: String,
pub title: String,
pub state: String,
}
@@ -437,7 +438,7 @@ pub struct ObjectiveLinkedTicketSummary {
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct ObjectiveSummary {
pub id: String,
pub human_key: String,
pub resource_key: String,
pub title: String,
pub state: String,
pub created_at: Option<String>,
@@ -450,7 +451,7 @@ pub struct ObjectiveSummary {
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct ObjectiveDetail {
pub id: String,
pub human_key: String,
pub resource_key: String,
pub title: String,
pub state: String,
pub revision: String,
crates/workspace-server/src/server.rs
+83 -35
View File
@@ -4562,7 +4562,7 @@ async fn scoped_show_merge_request(
.map(|ticket_id| {
Ok(MergeRequestLinkedTicketResponse {
ticket_id: ticket_id.clone(),
key: api.store.resource_human_key(
key: api.store.resource_key(
&workspace_id,
WorkspaceResourceKind::Ticket,
ticket_id,
@@ -10072,11 +10072,20 @@ async fn get_runtime_worker(
.list_workdir_registry(&api.config.workspace_id, 500)?;
let updated_at = record.updated_at.clone();
let mut worker = merge_worker_registry_projection(Some(&worker), &record, links, &workdirs);
worker.human_key = api.store.resource_human_key(
&api.config.workspace_id,
WorkspaceResourceKind::Worker,
&worker_ref.worker_id,
)?;
worker.resource_key = Some(
api.store
.resource_key(
&api.config.workspace_id,
WorkspaceResourceKind::Worker,
&worker_ref.worker_id,
)?
.ok_or_else(|| {
Error::Store(format!(
"Workspace Worker `{}` has no resource key",
worker_ref.worker_id
))
})?,
);
Ok(Json(WorkerShowProjection { worker, updated_at }))
}
@@ -10094,12 +10103,22 @@ async fn restore_runtime_worker(
let workdirs = api
.store
.list_workdir_registry(&api.config.workspace_id, 500)?;
result.worker = Some(merge_worker_registry_projection(
Some(worker),
&record,
links,
&workdirs,
));
let mut summary = merge_worker_registry_projection(Some(worker), &record, links, &workdirs);
summary.resource_key = Some(
api.store
.resource_key(
&api.config.workspace_id,
WorkspaceResourceKind::Worker,
&record.worker.worker_id,
)?
.ok_or_else(|| {
Error::Store(format!(
"Workspace Worker `{}` has no resource key",
record.worker.worker_id
))
})?,
);
result.worker = Some(summary);
}
Ok(Json(WorkerRestoreResponse {
workspace_id: api.workspace_id().to_string(),
@@ -11183,11 +11202,20 @@ fn workers_response(api: WorkspaceApi) -> ApiResult<RuntimeListResponse<WorkerSu
links,
&workdir_records,
);
summary.human_key = api.store.resource_human_key(
&api.config.workspace_id,
WorkspaceResourceKind::Worker,
&record.worker.worker_id,
)?;
summary.resource_key = Some(
api.store
.resource_key(
&api.config.workspace_id,
WorkspaceResourceKind::Worker,
&record.worker.worker_id,
)?
.ok_or_else(|| {
Error::Store(format!(
"Workspace Worker `{}` has no resource key",
record.worker.worker_id
))
})?,
);
items.push(summary);
}
Ok(RuntimeListResponse {
@@ -12056,7 +12084,7 @@ fn record_worker_summary(
fn worker_summary_from_registry(record: &WorkerRegistryRecord) -> WorkerSummary {
WorkerSummary {
worker: record.worker.clone(),
human_key: None,
resource_key: None,
host_id: "backend-registry".to_string(),
display_name: record.display_name.clone(),
label: record.display_name.clone(),
@@ -16207,10 +16235,11 @@ mod tests {
.unwrap()
.create(ticket::NewTicket::new("Browser Ticket API"))
.unwrap();
let ticket_human_key = ticket_ref.human_key.clone().unwrap();
let ticket_resource_key = ticket_ref.resource_key.clone().unwrap();
let ticket_id = ticket_ref.id;
assert_eq!(
resolve_workspace_ticket_reference(&api, TEST_WORKSPACE_ID, &ticket_human_key).unwrap(),
resolve_workspace_ticket_reference(&api, TEST_WORKSPACE_ID, &ticket_resource_key)
.unwrap(),
ticket_id
);
let path = || ScopedRecordPath {
@@ -18317,7 +18346,19 @@ mod tests {
let store = SqliteWorkspaceStore::in_memory().unwrap();
let mut config = test_server_config(dir.path());
write_ticket(
let sqlite_store = SqliteWorkspaceStore::open(&config.database_path).unwrap();
sqlite_store
.upsert_workspace(&WorkspaceRecord {
workspace_id: TEST_WORKSPACE_ID.to_string(),
owner_account_id: None,
display_name: "Test Workspace".to_string(),
state: "active".to_string(),
created_at: "2026-01-01T00:00:00Z".to_string(),
updated_at: "2026-01-01T00:00:00Z".to_string(),
})
.await
.unwrap();
let ticket_id = write_ticket(
&config.database_path,
TEST_WORKSPACE_ID,
"API Ticket",
@@ -18353,7 +18394,7 @@ mod tests {
&[ObjectiveTicketLinkRecord {
workspace_id: TEST_WORKSPACE_ID.to_string(),
objective_id: "00000000001J3".to_string(),
ticket_id: "00000000001J2".to_string(),
ticket_id: ticket_id.clone(),
kind: "linked".to_string(),
created_at: "2026-01-01T00:00:00Z".to_string(),
}],
@@ -18550,7 +18591,7 @@ mod tests {
&format!("/api/w/{TEST_WORKSPACE_ID}/objectives/query"),
Some(json!({
"query": "Objective body",
"linked_ticket_id": "00000000001J2",
"linked_ticket_id": ticket_id,
"limit": 1
})),
StatusCode::OK,
@@ -18566,7 +18607,7 @@ mod tests {
StatusCode::OK,
)
.await;
assert_eq!(shown_objective["linked_tickets"][0], "00000000001J2");
assert_eq!(shown_objective["linked_tickets"][0], ticket_id);
assert!(shown_objective["event_page"]["returned"].is_number());
let memory_document =
@@ -19459,16 +19500,23 @@ mod tests {
};
let response: protocol::subscription::SubscriptionFrame =
serde_json::from_str(text.as_str()).unwrap();
assert!(matches!(
response.payload,
let workers = match response.payload {
protocol::subscription::SubscriptionFramePayload::Response(
protocol::subscription::SubscriptionResponse::Subscribed {
selector: protocol::subscription::EventSubscriptionSelector::WorkspaceWorkers,
snapshot: protocol::subscription::SubscriptionSnapshot::Workers { .. },
snapshot: protocol::subscription::SubscriptionSnapshot::Workers { workers },
..
}
)
));
},
) => workers,
other => panic!("expected Workspace Worker snapshot, got {other:?}"),
};
assert_eq!(
workers
.iter()
.find(|worker| worker.worker_id.as_str() == worker_id)
.and_then(|worker| worker.resource_key.as_deref()),
Some("W-1")
);
let subscribe_protocol = protocol::subscription::SubscriptionFrame::new(
protocol::subscription::SubscriptionFramePayload::Request(
@@ -19805,12 +19853,12 @@ INSERT INTO typed_tickets (
) VALUES
('0192f0e8-4d84-7d6e-a000-000000000001', '00000000001J2', 'ticket-j2', 'Ticket J2', 'open', 'task', 'normal', '', 'planning', 1),
('0192f0e8-4d84-7d6e-a000-000000000001', '00000000001J3', 'ticket-j3', 'Ticket J3', 'open', 'task', 'normal', '', 'planning', 1);
INSERT INTO workspace_resource_human_keys (
workspace_id, resource_kind, resource_id, sequence, human_key, allocated_at
INSERT INTO workspace_resource_keys (
workspace_id, resource_kind, resource_id, sequence, resource_key, allocated_at
) VALUES
('0192f0e8-4d84-7d6e-a000-000000000001', 'ticket', '00000000001J2', 1, 'T-1', '2026-01-01T00:00:00Z'),
('0192f0e8-4d84-7d6e-a000-000000000001', 'ticket', '00000000001J3', 2, 'T-2', '2026-01-01T00:00:00Z');
INSERT INTO workspace_resource_human_key_counters (workspace_id, resource_kind, next_sequence)
INSERT INTO workspace_resource_key_counters (workspace_id, resource_kind, next_sequence)
VALUES ('0192f0e8-4d84-7d6e-a000-000000000001', 'ticket', 3);
"#,
)
@@ -20027,13 +20075,13 @@ VALUES ('0192f0e8-4d84-7d6e-a000-000000000001', 'ticket', 3);
workspace_id: &str,
title: &str,
state: ticket::TicketWorkflowState,
) {
) -> String {
use ticket::TicketBackend as _;
let backend = ticket::SqliteTicketBackend::open(database_path, workspace_id).unwrap();
let mut input = ticket::NewTicket::new(title);
input.workflow_state = Some(state);
backend.create(input).unwrap();
backend.create(input).unwrap().id
}
fn write_objective(root: &Path, id: &str, title: &str, state: &str) {
crates/workspace-server/src/store.rs
+425 -77
View File
@@ -225,6 +225,11 @@ const MIGRATIONS: &[Migration] = &[
name: "create atomic Workspace catalog operations",
apply: create_workspace_catalog_operations,
},
Migration {
version: 41,
name: "rename Workspace resource keys",
apply: verify_workspace_resource_key_schema,
},
];
struct Migration {
@@ -595,7 +600,7 @@ impl WorkspaceResourceKind {
#[async_trait]
pub trait ControlPlaneStore: Send + Sync {
async fn schema_version(&self) -> Result<i64>;
fn resource_human_key(
fn resource_key(
&self,
workspace_id: &str,
kind: WorkspaceResourceKind,
@@ -1059,6 +1064,14 @@ impl SqliteWorkspaceStore {
} else {
Vec::new()
};
apply_migrations_through(&candidate, 38)?;
let assignment_worker_tombstone_repairs =
legacy_assignment_worker_tombstone_repairs(&candidate)?.len();
if assignment_worker_tombstone_repairs > 0 {
repairs.push(format!(
"materialize {assignment_worker_tombstone_repairs} legacy Ticket assignment Worker tombstone(s)"
));
}
apply_migrations_through(&candidate, i64::MAX)?;
ticket::migrate_sqlite_ticket_schema(&candidate)?;
merge_request::migrate(&candidate).map_err(|error| Error::Store(error.to_string()))?;
@@ -1171,7 +1184,7 @@ impl SqliteWorkspaceStore {
let worker_id = WorkerId::now_v7();
let now = chrono::Utc::now().to_rfc3339();
allocate_resource_human_key(
allocate_resource_key(
&tx,
workspace_id,
WorkspaceResourceKind::Worker,
@@ -1304,7 +1317,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
self.with_conn(current_schema_version)
}
fn resource_human_key(
fn resource_key(
&self,
workspace_id: &str,
kind: WorkspaceResourceKind,
@@ -1312,7 +1325,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
) -> Result<Option<String>> {
self.with_conn(|conn| {
conn.query_row(
"SELECT human_key FROM workspace_resource_human_keys
"SELECT resource_key FROM workspace_resource_keys
WHERE workspace_id = ?1 AND resource_kind = ?2 AND resource_id = ?3",
params![workspace_id, kind.as_str(), resource_id],
|row| row.get(0),
@@ -1331,8 +1344,8 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
self.with_conn(|conn| {
if let Some(resource_id) = conn
.query_row(
"SELECT resource_id FROM workspace_resource_human_keys
WHERE workspace_id = ?1 AND resource_kind = ?2 AND human_key = ?3",
"SELECT resource_id FROM workspace_resource_keys
WHERE workspace_id = ?1 AND resource_kind = ?2 AND resource_key = ?3",
params![workspace_id, kind.as_str(), reference],
|row| row.get(0),
)
@@ -1536,7 +1549,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
}
for resource_kind in ["ticket", "objective", "worker"] {
tx.execute(
r#"INSERT OR IGNORE INTO workspace_resource_human_key_counters (
r#"INSERT OR IGNORE INTO workspace_resource_key_counters (
workspace_id, resource_kind, next_sequence
) VALUES (?1, ?2, 1)"#,
params![record.workspace.workspace_id, resource_kind],
@@ -1979,7 +1992,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
fn upsert_objective(&self, record: &ObjectiveRecord) -> Result<()> {
self.with_conn(|conn| {
let tx = conn.unchecked_transaction()?;
allocate_resource_human_key(
allocate_resource_key(
&tx,
&record.workspace_id,
WorkspaceResourceKind::Objective,
@@ -2757,7 +2770,8 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
fn upsert_worker_registry(&self, record: &WorkerRegistryRecord) -> Result<()> {
self.with_conn(|conn| {
let removal_blocks_upsert: bool = conn.query_row(
let tx = conn.unchecked_transaction()?;
let removal_blocks_upsert: bool = tx.query_row(
"SELECT EXISTS(
SELECT 1 FROM worker_removal_operations
WHERE workspace_id = ?1 AND runtime_id = ?2
@@ -2774,7 +2788,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
if removal_blocks_upsert {
return Ok(());
}
conn.execute(
tx.execute(
r#"INSERT INTO worker_registry (
workspace_id, runtime_id, worker_id, display_name, profile,
retention_state, transcript_ref, session_ref, summary_ref,
@@ -2816,6 +2830,14 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
record.updated_at,
],
)?;
allocate_resource_key(
&tx,
&record.workspace_id,
WorkspaceResourceKind::Worker,
&record.worker.worker_id,
&record.created_at,
)?;
tx.commit()?;
Ok(())
})
}
@@ -5527,8 +5549,11 @@ fn workspace_resource_reference_diagnostics(conn: &Connection) -> Result<Vec<Str
}
// Assignment and operation rows are historical soft references. Schema v39 records an
// explicit tombstone before a live Ticket or Worker parent is deleted/moved; older schemas
// have no tombstone authority, so every missing live parent remains migration-blocking drift.
// explicit tombstone before a live Ticket or Worker parent is deleted/moved. A pre-v39
// assignment with a valid Worker UUID and no contradictory Worker authority in another
// Workspace is repairable legacy evidence; the migration materializes its tombstone.
// Missing Ticket parents remain migration-blocking because no equivalent legacy repair is
// currently defined.
if table_exists(conn, "ticket_worker_assignments")? && table_exists(conn, "typed_tickets")? {
let tombstone_filter = if table_exists(conn, "ticket_assignment_ticket_tombstones")? {
"AND NOT EXISTS (SELECT 1 FROM ticket_assignment_ticket_tombstones AS tombstone \
@@ -5556,28 +5581,7 @@ fn workspace_resource_reference_diagnostics(conn: &Connection) -> Result<Vec<Str
&& column_exists(conn, "ticket_worker_assignments", "worker_id")?
&& column_exists(conn, "worker_registry", "worker_id")?
{
let tombstone_filter = if table_exists(conn, "ticket_assignment_worker_tombstones")? {
"AND NOT EXISTS (SELECT 1 FROM ticket_assignment_worker_tombstones AS tombstone \
WHERE tombstone.workspace_id = assignment.workspace_id \
AND tombstone.runtime_id = assignment.runtime_id \
AND tombstone.worker_id = assignment.worker_id)"
} else {
""
};
collect_reference_diagnostics(
conn,
&format!(
"SELECT assignment.workspace_id || '/' || assignment.assignment_id || ' -> ' || assignment.runtime_id || '/' || assignment.worker_id \
FROM ticket_worker_assignments AS assignment \
WHERE NOT EXISTS (SELECT 1 FROM worker_registry AS worker \
WHERE worker.workspace_id = assignment.workspace_id \
AND worker.runtime_id = assignment.runtime_id \
AND worker.worker_id = assignment.worker_id) \
{tombstone_filter} LIMIT 100"
),
"ticket_worker_assignments.worker_id",
&mut diagnostics,
)?;
collect_assignment_worker_reference_diagnostics(conn, &mut diagnostics)?;
}
if table_exists(conn, "ticket_assignment_operations")? && table_exists(conn, "typed_tickets")? {
let tombstone_filter = if table_exists(conn, "ticket_assignment_ticket_tombstones")? {
@@ -5604,6 +5608,114 @@ fn workspace_resource_reference_diagnostics(conn: &Connection) -> Result<Vec<Str
Ok(diagnostics)
}
fn collect_assignment_worker_reference_diagnostics(
conn: &Connection,
diagnostics: &mut Vec<String>,
) -> Result<()> {
let has_assignment_tombstones = table_exists(conn, "ticket_assignment_worker_tombstones")?;
let legacy_tombstone_repairs = legacy_assignment_worker_tombstone_repairs(conn)?;
let tombstone_filter = if has_assignment_tombstones {
"AND NOT EXISTS (SELECT 1 FROM ticket_assignment_worker_tombstones AS tombstone \
WHERE tombstone.workspace_id = assignment.workspace_id \
AND tombstone.runtime_id = assignment.runtime_id \
AND tombstone.worker_id = assignment.worker_id)"
} else {
""
};
let sql = format!(
"SELECT assignment.workspace_id, assignment.assignment_id, \
assignment.runtime_id, assignment.worker_id \
FROM ticket_worker_assignments AS assignment \
WHERE NOT EXISTS (SELECT 1 FROM worker_registry AS worker \
WHERE worker.workspace_id = assignment.workspace_id \
AND worker.runtime_id = assignment.runtime_id \
AND worker.worker_id = assignment.worker_id) \
{tombstone_filter}"
);
let mut statement = conn.prepare(&sql)?;
let rows = statement.query_map([], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
))
})?;
let mut worker_diagnostic_count = 0;
for row in rows {
let (workspace_id, assignment_id, runtime_id, worker_id) = row?;
if legacy_tombstone_repairs.contains(&(
workspace_id.clone(),
runtime_id.clone(),
worker_id.clone(),
)) {
continue;
}
diagnostics.push(format!(
"ticket_worker_assignments.worker_id: \
{workspace_id}/{assignment_id} -> {runtime_id}/{worker_id}"
));
worker_diagnostic_count += 1;
if worker_diagnostic_count == 100 {
break;
}
}
Ok(())
}
fn legacy_assignment_worker_tombstone_repairs(
conn: &Connection,
) -> Result<std::collections::BTreeSet<(String, String, String)>> {
if current_schema_version(conn)? >= 39
|| table_exists(conn, "ticket_assignment_worker_tombstones")?
|| !table_exists(conn, "ticket_worker_assignments")?
|| !table_exists(conn, "worker_registry")?
|| !column_exists(conn, "ticket_worker_assignments", "worker_id")?
|| !column_exists(conn, "worker_registry", "worker_id")?
{
return Ok(std::collections::BTreeSet::new());
}
let mut repairs = std::collections::BTreeSet::new();
let mut statement = conn.prepare(
"SELECT DISTINCT assignment.workspace_id, assignment.runtime_id, assignment.worker_id \
FROM ticket_worker_assignments AS assignment \
WHERE NOT EXISTS (SELECT 1 FROM worker_registry AS worker \
WHERE worker.workspace_id = assignment.workspace_id \
AND worker.runtime_id = assignment.runtime_id \
AND worker.worker_id = assignment.worker_id)",
)?;
let rows = statement.query_map([], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
))
})?;
for row in rows {
let (workspace_id, runtime_id, worker_id) = row?;
if WorkerId::parse(&worker_id).is_none() {
continue;
}
let exists_only_outside_workspace: bool = conn.query_row(
"SELECT EXISTS(SELECT 1 FROM worker_registry \
WHERE worker_id = ?1 AND workspace_id != ?2) \
AND NOT EXISTS(SELECT 1 FROM worker_registry \
WHERE worker_id = ?1 AND workspace_id = ?2)",
params![worker_id, workspace_id],
|row| row.get(0),
)?;
if !exists_only_outside_workspace {
// Before v39, supported cleanup and Runtime-placement changes could remove or move a
// Worker without recording an assignment-specific tombstone. A valid,
// non-cross-Workspace Worker identity is sufficient legacy evidence; v39
// materializes the missing tombstone in the migration transaction.
repairs.insert((workspace_id, runtime_id, worker_id));
}
}
Ok(repairs)
}
fn collect_reference_diagnostics(
conn: &Connection,
sql: &str,
@@ -6132,7 +6244,7 @@ fn promote_workspace_worker_uuid_identity(
Ok(mappings)
}
fn allocate_resource_human_key(
fn allocate_resource_key(
conn: &Connection,
workspace_id: &str,
kind: WorkspaceResourceKind,
@@ -6141,7 +6253,7 @@ fn allocate_resource_human_key(
) -> Result<String> {
if let Some(existing) = conn
.query_row(
"SELECT human_key FROM workspace_resource_human_keys
"SELECT resource_key FROM workspace_resource_keys
WHERE workspace_id = ?1 AND resource_kind = ?2 AND resource_id = ?3",
params![workspace_id, kind.as_str(), resource_id],
|row| row.get(0),
@@ -6151,36 +6263,36 @@ fn allocate_resource_human_key(
return Ok(existing);
}
conn.execute(
"INSERT OR IGNORE INTO workspace_resource_human_key_counters
"INSERT OR IGNORE INTO workspace_resource_key_counters
(workspace_id, resource_kind, next_sequence) VALUES (?1, ?2, 1)",
params![workspace_id, kind.as_str()],
)?;
let sequence: i64 = conn.query_row(
"SELECT next_sequence FROM workspace_resource_human_key_counters
"SELECT next_sequence FROM workspace_resource_key_counters
WHERE workspace_id = ?1 AND resource_kind = ?2",
params![workspace_id, kind.as_str()],
|row| row.get(0),
)?;
conn.execute(
"UPDATE workspace_resource_human_key_counters SET next_sequence = ?3
"UPDATE workspace_resource_key_counters SET next_sequence = ?3
WHERE workspace_id = ?1 AND resource_kind = ?2",
params![workspace_id, kind.as_str(), sequence + 1],
)?;
let human_key = format!("{}-{sequence}", kind.prefix());
let resource_key = format!("{}-{sequence}", kind.prefix());
conn.execute(
"INSERT INTO workspace_resource_human_keys
(workspace_id, resource_kind, resource_id, sequence, human_key, allocated_at)
"INSERT INTO workspace_resource_keys
(workspace_id, resource_kind, resource_id, sequence, resource_key, allocated_at)
VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
params![
workspace_id,
kind.as_str(),
resource_id,
sequence,
human_key,
resource_key,
allocated_at
],
)?;
Ok(human_key)
Ok(resource_key)
}
fn add_workspace_resource_human_keys(conn: &Connection) -> Result<()> {
@@ -6261,6 +6373,47 @@ fn add_workspace_resource_human_keys(conn: &Connection) -> Result<()> {
Ok(())
}
fn verify_workspace_resource_key_schema(conn: &Connection) -> Result<()> {
ticket::migrate_sqlite_ticket_resource_key_schema_in_transaction(conn).map_err(|error| {
Error::Store(format!(
"migration 41 Ticket resource-key schema failed: {error}"
))
})?;
for legacy_table in [
"workspace_resource_human_keys",
"workspace_resource_human_key_counters",
] {
if table_exists(conn, legacy_table)? {
return Err(Error::Store(format!(
"migration 41 left legacy table `{legacy_table}`"
)));
}
}
if !table_exists(conn, "workspace_resource_keys")?
|| !column_exists(conn, "workspace_resource_keys", "resource_key")?
|| column_exists(conn, "workspace_resource_keys", "human_key")?
|| !table_exists(conn, "workspace_resource_key_counters")?
{
return Err(Error::Store(
"migration 41 did not materialize the Workspace resource key schema".to_string(),
));
}
let index_exists = conn
.query_row(
"SELECT 1 FROM sqlite_schema WHERE type = 'index' AND name = 'idx_workspace_resource_keys_reverse'",
[],
|_| Ok(()),
)
.optional()?
.is_some();
if !index_exists {
return Err(Error::Store(
"migration 41 did not create the Workspace resource key reverse index".to_string(),
));
}
Ok(())
}
fn remove_worker_control_delegation_authority(conn: &Connection) -> Result<()> {
let mut statement =
conn.prepare("SELECT workspace_id, grant_id, permissions_json FROM worker_control_grants")?;
@@ -6530,6 +6683,21 @@ CREATE TABLE ticket_worker_assignments_v39 (
FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE
);
INSERT INTO ticket_worker_assignments_v39 SELECT * FROM ticket_worker_assignments;
INSERT OR IGNORE INTO ticket_assignment_worker_tombstones (
workspace_id, runtime_id, worker_id, deleted_at
)
SELECT DISTINCT
assignment.workspace_id,
assignment.runtime_id,
assignment.worker_id,
CURRENT_TIMESTAMP
FROM ticket_worker_assignments_v39 AS assignment
WHERE NOT EXISTS (
SELECT 1 FROM worker_registry AS worker
WHERE worker.workspace_id = assignment.workspace_id
AND worker.runtime_id = assignment.runtime_id
AND worker.worker_id = assignment.worker_id
);
CREATE TABLE ticket_worker_assignment_events_v39 (
workspace_id TEXT NOT NULL,
@@ -6830,17 +6998,66 @@ END;
Ok(())
}
fn rebuild_workspace_scoped_references_from_resource_keys(conn: &Connection) -> Result<()> {
if table_exists(conn, "workspace_resource_human_keys")? {
conn.execute_batch(
r#"
INSERT OR IGNORE INTO workspace_resource_human_keys (
workspace_id, resource_kind, resource_id, sequence, human_key, allocated_at
)
SELECT workspace_id, resource_kind, resource_id, sequence, resource_key, allocated_at
FROM workspace_resource_keys;
INSERT INTO workspace_resource_human_key_counters (
workspace_id, resource_kind, next_sequence
)
SELECT workspace_id, resource_kind, next_sequence
FROM workspace_resource_key_counters
WHERE true
ON CONFLICT(workspace_id, resource_kind) DO UPDATE SET
next_sequence = max(next_sequence, excluded.next_sequence);
DROP INDEX IF EXISTS idx_workspace_resource_keys_reverse;
DROP TABLE workspace_resource_keys;
DROP TABLE workspace_resource_key_counters;
"#,
)?;
} else {
conn.execute_batch(
r#"
DROP INDEX IF EXISTS idx_workspace_resource_keys_reverse;
ALTER TABLE workspace_resource_keys RENAME COLUMN resource_key TO human_key;
ALTER TABLE workspace_resource_keys RENAME TO workspace_resource_human_keys;
ALTER TABLE workspace_resource_key_counters RENAME TO workspace_resource_human_key_counters;
"#,
)?;
}
enforce_workspace_resource_foreign_keys(conn)?;
conn.execute_batch(
r#"
DROP INDEX IF EXISTS idx_workspace_resource_human_keys_reverse;
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;
CREATE INDEX idx_workspace_resource_keys_reverse
ON workspace_resource_keys(workspace_id, resource_kind, resource_key);
"#,
)?;
Ok(())
}
pub(crate) fn apply_migrations_through(conn: &Connection, through_version: i64) -> Result<()> {
let current = current_schema_version(conn)?;
for migration in MIGRATIONS.iter().filter(|migration| {
i64::from(migration.version) > current && i64::from(migration.version) <= through_version
}) {
if migration.version == 39 {
ticket::migrate_sqlite_ticket_schema(conn).map_err(|error| {
Error::Store(format!(
"migration 39 Ticket schema preparation failed: {error}"
))
})?;
let resource_key_schema_current = table_exists(conn, "workspace_resource_keys")?;
if !resource_key_schema_current {
ticket::migrate_sqlite_ticket_schema_through(conn, 5).map_err(|error| {
Error::Store(format!(
"migration 39 Ticket schema preparation failed: {error}"
))
})?;
}
if !table_exists(conn, "typed_tickets")? {
return Err(Error::Store(
"migration 39 Ticket schema preparation created no typed_tickets".to_string(),
@@ -6856,7 +7073,11 @@ pub(crate) fn apply_migrations_through(conn: &Connection, through_version: i64)
conn.execute_batch("PRAGMA foreign_keys = OFF; PRAGMA legacy_alter_table = ON;")?;
let result = (|| -> Result<()> {
let tx = conn.unchecked_transaction()?;
(migration.apply)(&tx)?;
if resource_key_schema_current {
rebuild_workspace_scoped_references_from_resource_keys(&tx)?;
} else {
(migration.apply)(&tx)?;
}
if !table_exists(&tx, "typed_tickets")? {
return Err(Error::Store(
"migration 39 did not materialize `typed_tickets`".to_string(),
@@ -7461,7 +7682,7 @@ mod tests {
let before = std::fs::read(&path).unwrap();
let plan = SqliteWorkspaceStore::migration_plan(&path).unwrap();
assert_eq!(plan.current_schema_version, 36);
assert_eq!(plan.target_schema_version, 40);
assert_eq!(plan.target_schema_version, 41);
assert!(plan.migration_required);
assert_eq!(plan.worker_count, 1);
assert_eq!(plan.mappings[0].legacy_worker_id, 7);
@@ -7475,14 +7696,14 @@ mod tests {
store
.with_conn(|conn| {
assert!(table_exists(conn, "worker_diagnostics_archives")?);
assert_eq!(current_schema_version(conn)?, 40);
assert_eq!(current_schema_version(conn)?, 41);
Ok(())
})
.unwrap();
}
#[test]
fn v38_backfills_workspace_scoped_objective_and_worker_human_keys() {
fn v38_backfills_workspace_scoped_objective_and_worker_resource_keys() {
let conn = Connection::open_in_memory().unwrap();
configure_sqlite(&conn).unwrap();
apply_migrations_through(&conn, 37).unwrap();
@@ -7518,11 +7739,11 @@ mod tests {
).unwrap();
}
ticket::migrate_sqlite_ticket_schema(&conn).unwrap();
ticket::migrate_sqlite_ticket_schema_through(&conn, 5).unwrap();
apply_migrations(&conn).unwrap();
let mut statement = conn
.prepare(
"SELECT resource_kind, resource_id, human_key FROM workspace_resource_human_keys
"SELECT resource_kind, resource_id, resource_key FROM workspace_resource_keys
ORDER BY resource_kind, sequence",
)
.unwrap();
@@ -7554,7 +7775,7 @@ mod tests {
),
]
);
assert_eq!(current_schema_version(&conn).unwrap(), 40);
assert_eq!(current_schema_version(&conn).unwrap(), 41);
let foreign_key_error: Option<String> = conn
.query_row("PRAGMA foreign_key_check", [], |row| row.get(0))
.optional()
@@ -7683,7 +7904,7 @@ INSERT INTO worker_orphan_diagnostics (
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 40);
assert_eq!(current_schema_version(&conn).unwrap(), 41);
assert!(!table_exists(&conn, "worker_control_delegation_operations").unwrap());
let controller_worker_id: String = conn
.query_row(
@@ -7801,7 +8022,7 @@ INSERT INTO worker_orphan_diagnostics (
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 40);
assert_eq!(current_schema_version(&conn).unwrap(), 41);
assert!(table_exists(&conn, "worker_workdir_attachment_reservations").unwrap());
}
@@ -7834,7 +8055,7 @@ CREATE TABLE flow_events (event_id TEXT PRIMARY KEY);
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 40);
assert_eq!(current_schema_version(&conn).unwrap(), 41);
assert!(table_exists(&conn, "flow_sources").unwrap());
assert!(table_exists(&conn, "flow_source_revisions").unwrap());
assert!(!table_exists(&conn, "flow_instances").unwrap());
@@ -7901,7 +8122,7 @@ INSERT INTO worker_workdir_attachment_reservations (
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 40);
assert_eq!(current_schema_version(&conn).unwrap(), 41);
let repositories_sql: String = conn
.query_row(
"SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'repositories'",
@@ -8079,7 +8300,7 @@ INSERT INTO workdir_registry (
let db = dir.path().join("control-plane.sqlite");
let store = SqliteWorkspaceStore::open(&db).unwrap();
assert_eq!(store.schema_version().await.unwrap(), 40);
assert_eq!(store.schema_version().await.unwrap(), 41);
assert!(
!store
.with_conn(|conn| table_exists(conn, "worker_workspace_credentials"))
@@ -8096,7 +8317,7 @@ INSERT INTO workdir_registry (
store.upsert_workspace(&record).await.unwrap();
let reopened = SqliteWorkspaceStore::open(&db).unwrap();
assert_eq!(reopened.schema_version().await.unwrap(), 40);
assert_eq!(reopened.schema_version().await.unwrap(), 41);
assert_eq!(
reopened.get_workspace("local-dev").await.unwrap(),
Some(record)
@@ -8104,7 +8325,7 @@ INSERT INTO workdir_registry (
}
#[tokio::test]
async fn objective_creation_allocates_and_resolves_workspace_human_key() {
async fn objective_creation_allocates_and_resolves_workspace_resource_key() {
let dir = tempfile::tempdir().unwrap();
let store = SqliteWorkspaceStore::open(dir.path().join("server.db")).unwrap();
store
@@ -8131,7 +8352,7 @@ INSERT INTO workdir_registry (
.unwrap();
assert_eq!(
store
.resource_human_key(
.resource_key(
"workspace-a",
WorkspaceResourceKind::Objective,
"objective-internal"
@@ -8185,7 +8406,7 @@ INSERT INTO workdir_registry (
);
assert_eq!(
store
.resource_human_key(
.resource_key(
"workspace-a",
WorkspaceResourceKind::Worker,
&reserved.to_string()
@@ -8205,7 +8426,7 @@ INSERT INTO workdir_registry (
.unwrap();
assert_eq!(
store
.resource_human_key(
.resource_key(
"workspace-a",
WorkspaceResourceKind::Worker,
&second.to_string()
@@ -8595,13 +8816,13 @@ INSERT INTO worker_registry (
configure_sqlite(&conn).unwrap();
apply_migrations(&conn).unwrap();
conn.execute(
"INSERT INTO __yoi_schema_migrations (version, name) VALUES (41, 'future')",
"INSERT INTO __yoi_schema_migrations (version, name) VALUES (42, 'future')",
[],
)
.unwrap();
let error = apply_migrations(&conn).unwrap_err().to_string();
assert!(error.contains("schema version 41 is newer"), "{error}");
assert!(error.contains("schema version 42 is newer"), "{error}");
assert!(error.contains("refusing to serve"), "{error}");
}
@@ -8713,7 +8934,7 @@ INSERT INTO worker_registry (
let conn = Connection::open(&path).unwrap();
configure_sqlite(&conn).unwrap();
apply_migrations_through(&conn, 38).unwrap();
ticket::migrate_sqlite_ticket_schema(&conn).unwrap();
ticket::migrate_sqlite_ticket_schema_through(&conn, 5).unwrap();
merge_request::migrate(&conn).unwrap();
conn.execute_batch(
r#"
@@ -8759,7 +8980,7 @@ VALUES ('workspace-b', 'ticket-b', 'related', 'ticket-a', NULL, 'tester', '2026-
let conn = Connection::open_in_memory().unwrap();
configure_sqlite(&conn).unwrap();
apply_migrations_through(&conn, 38).unwrap();
ticket::migrate_sqlite_ticket_schema(&conn).unwrap();
ticket::migrate_sqlite_ticket_schema_through(&conn, 5).unwrap();
merge_request::migrate(&conn).unwrap();
conn.execute(
"INSERT INTO workspaces (workspace_id, display_name, state, created_at, updated_at) \
@@ -8822,7 +9043,7 @@ VALUES ('workspace-b', 'ticket-b', 'related', 'ticket-a', NULL, 'tester', '2026-
apply_migrations(&mut conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 40);
assert_eq!(current_schema_version(&conn).unwrap(), 41);
let workspace_id: Option<String> = conn
.query_row(
"SELECT workspace_id FROM trusted_runtime_records WHERE runtime_id = 'runtime-a'",
@@ -8839,7 +9060,7 @@ VALUES ('workspace-b', 'ticket-b', 'related', 'ticket-a', NULL, 'tester', '2026-
let conn = Connection::open_in_memory().unwrap();
configure_sqlite(&conn).unwrap();
apply_migrations_through(&conn, 38).unwrap();
ticket::migrate_sqlite_ticket_schema(&conn).unwrap();
ticket::migrate_sqlite_ticket_schema_through(&conn, 5).unwrap();
merge_request::migrate(&conn).unwrap();
conn.execute_batch(
@@ -8937,7 +9158,6 @@ INSERT INTO ticket_worker_assignment_events (
"{error}"
);
assert!(error.contains("assignment-cross-worker"), "{error}");
assert!(error.contains("assignment-runtime-mismatch"), "{error}");
assert!(error.contains("assignment-missing-parents"), "{error}");
assert!(
error.contains("ticket_worker_assignment_events.assignment_id"),
@@ -9117,6 +9337,123 @@ INSERT INTO ticket_worker_assignment_events (
assert_eq!(integrity, "ok");
}
#[test]
fn workspace_resource_fk_migration_preserves_assignments_for_legacy_absent_workers() {
let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("server.db");
let conn = Connection::open(&path).unwrap();
configure_sqlite(&conn).unwrap();
apply_migrations_through(&conn, 38).unwrap();
ticket::migrate_sqlite_ticket_schema_through(&conn, 5).unwrap();
merge_request::migrate(&conn).unwrap();
conn.execute_batch(
r#"
INSERT INTO workspaces (
workspace_id, display_name, state, created_at, updated_at
) VALUES ('workspace-a', 'A', 'active', '2026-01-01', '2026-01-01');
INSERT INTO typed_tickets (
workspace_id, ticket_id, slug, title, status, kind, priority, body,
workflow_state, workflow_state_explicit
) VALUES (
'workspace-a', 'ticket-a', 'ticket-a', 'A', 'open', 'task', 'normal', '',
'planning', 1
);
INSERT INTO worker_registry (
workspace_id, runtime_id, worker_id, display_name, retention_state, created_at, updated_at
) VALUES
(
'workspace-a', 'runtime-a', '00000000-0000-7000-8000-000000000001',
'Worker A', 'normal', '2026-01-01', '2026-01-01'
),
(
'workspace-a', 'runtime-old', '00000000-0000-7000-8000-000000000002',
'Worker B', 'normal', '2026-01-01', '2026-01-01'
);
INSERT INTO ticket_worker_assignments (
workspace_id, ticket_id, assignment_id, runtime_id, worker_id, assigned_by, assigned_at
) VALUES
(
'workspace-a', 'ticket-a', 'assignment-a', 'runtime-a',
'00000000-0000-7000-8000-000000000001', 'tester', '2026-01-01'
),
(
'workspace-a', 'ticket-a', 'assignment-b', 'runtime-old',
'00000000-0000-7000-8000-000000000002', 'tester', '2026-01-01'
);
DELETE FROM worker_registry
WHERE workspace_id = 'workspace-a'
AND runtime_id = 'runtime-a'
AND worker_id = '00000000-0000-7000-8000-000000000001';
UPDATE worker_registry
SET runtime_id = 'runtime-new'
WHERE workspace_id = 'workspace-a'
AND runtime_id = 'runtime-old'
AND worker_id = '00000000-0000-7000-8000-000000000002';
"#,
)
.unwrap();
assert_eq!(
legacy_assignment_worker_tombstone_repairs(&conn)
.unwrap()
.len(),
2
);
drop(conn);
let plan = SqliteWorkspaceStore::migration_plan(&path).unwrap();
assert!(
plan.repairs.iter().any(
|repair| repair == "materialize 2 legacy Ticket assignment Worker tombstone(s)"
),
"{:?}",
plan.repairs
);
let conn = Connection::open(&path).unwrap();
configure_sqlite(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 38);
assert!(!table_exists(&conn, "ticket_assignment_worker_tombstones").unwrap());
apply_migrations_through(&conn, 39).unwrap();
assert_eq!(
conn.query_row(
"SELECT COUNT(*) FROM ticket_worker_assignments \
WHERE workspace_id = 'workspace-a' AND assignment_id = 'assignment-a'",
[],
|row| row.get::<_, i64>(0),
)
.unwrap(),
1
);
assert_eq!(
conn.query_row(
"SELECT COUNT(*) FROM ticket_assignment_worker_tombstones \
WHERE workspace_id = 'workspace-a' \
AND runtime_id = 'runtime-a' \
AND worker_id = '00000000-0000-7000-8000-000000000001'",
[],
|row| row.get::<_, i64>(0),
)
.unwrap(),
1
);
assert_eq!(
conn.query_row(
"SELECT COUNT(*) FROM ticket_assignment_worker_tombstones \
WHERE workspace_id = 'workspace-a' \
AND runtime_id = 'runtime-old' \
AND worker_id = '00000000-0000-7000-8000-000000000002'",
[],
|row| row.get::<_, i64>(0),
)
.unwrap(),
1
);
validate_workspace_resource_references(&conn).unwrap();
}
#[test]
fn fresh_schema_matches_workspace_db_v0_boundaries() {
let conn = Connection::open_in_memory().unwrap();
@@ -9323,7 +9660,7 @@ INSERT INTO ticket_worker_assignment_events (
.unwrap();
let store = SqliteWorkspaceStore::from_connection(conn).unwrap();
assert_eq!(store.schema_version().await.unwrap(), 40);
assert_eq!(store.schema_version().await.unwrap(), 41);
store
.with_conn(|conn| {
@@ -9512,7 +9849,7 @@ CREATE TABLE ticket_assignment_operations (
#[tokio::test]
async fn repository_records_round_trip() {
let store = SqliteWorkspaceStore::in_memory().unwrap();
assert_eq!(store.schema_version().await.unwrap(), 40);
assert_eq!(store.schema_version().await.unwrap(), 41);
let workspace = WorkspaceRecord {
workspace_id: "local-dev".to_string(),
owner_account_id: None,
@@ -9578,7 +9915,7 @@ CREATE TABLE ticket_assignment_operations (
#[tokio::test]
async fn memory_authority_records_round_trip_and_close_staging() {
let store = SqliteWorkspaceStore::in_memory().unwrap();
assert_eq!(store.schema_version().await.unwrap(), 40);
assert_eq!(store.schema_version().await.unwrap(), 41);
let workspace = WorkspaceRecord {
workspace_id: "local-dev".to_string(),
owner_account_id: None,
@@ -9698,6 +10035,17 @@ CREATE TABLE ticket_assignment_operations (
runtime_sync_worker.retention_state = "normal".to_string();
runtime_sync_worker.updated_at = "5".to_string();
store.upsert_worker_registry(&runtime_sync_worker).unwrap();
assert_eq!(
store
.resource_key(
"local-dev",
WorkspaceResourceKind::Worker,
&worker.worker.worker_id,
)
.unwrap()
.as_deref(),
Some("W-1")
);
let mut expected_worker = worker.clone();
expected_worker.updated_at = "5".to_string();
@@ -9969,7 +10317,7 @@ CREATE TABLE ticket_assignment_operations (
#[tokio::test]
async fn account_and_login_records_round_trip() {
let store = SqliteWorkspaceStore::in_memory().unwrap();
assert_eq!(store.schema_version().await.unwrap(), 40);
assert_eq!(store.schema_version().await.unwrap(), 41);
let now = "2026-07-22T00:00:00Z".to_string();
let account = AccountRecord {
account_id: "acct-user-alice".to_string(),
crates/workspace-server/src/workspace_subscription.rs
+22 -2
View File
@@ -12,6 +12,7 @@ use worker_runtime::identity::RuntimeWorkerRef;
use crate::runtime_subscription::{BrokerSubscriptionEvent, RuntimeSubscriptionBroker};
use crate::server::{WorkspaceApi, connect_workspace_worker_protocol};
use crate::store::WorkspaceResourceKind;
const OUTBOUND_CAPACITY: usize = 256;
@@ -65,6 +66,7 @@ pub(crate) async fn serve_workspace_subscription(api: WorkspaceApi, socket: WebS
match selector {
EventSubscriptionSelector::WorkspaceWorkers => {
let task = tokio::spawn(run_workspace_workers(
api.clone(),
broker.clone(),
request_id,
subscription_id.clone(),
@@ -273,6 +275,7 @@ async fn run_worker_protocol(
}
async fn run_workspace_workers(
api: WorkspaceApi,
broker: RuntimeSubscriptionBroker,
request_id: protocol::subscription::SubscriptionRequestId,
subscription_id: SubscriptionId,
@@ -307,7 +310,7 @@ async fn run_workspace_workers(
};
match event {
BrokerSubscriptionEvent::Snapshot { snapshot, .. } => {
install_snapshot(&mut workers, &runtime_id, snapshot);
install_snapshot(&api, &mut workers, &runtime_id, snapshot);
pending.remove(&runtime_id);
}
BrokerSubscriptionEvent::Disconnected { .. }
@@ -371,7 +374,7 @@ async fn run_workspace_workers(
return;
}
}
install_snapshot(&mut workers, &runtime_id, snapshot);
install_snapshot(&api, &mut workers, &runtime_id, snapshot);
if let Some(current) = workers.get_mut(&runtime_id) {
for worker in current.values_mut() {
let worker_ref =
@@ -397,6 +400,14 @@ async fn run_workspace_workers(
BrokerSubscriptionEvent::Event { payload, .. } => match payload {
SubscriptionEventPayload::WorkerUpserted { mut worker } => {
worker.runtime_id = Some(runtime_id.clone());
let Ok(Some(resource_key)) = api.store.resource_key(
&api.config.workspace_id,
WorkspaceResourceKind::Worker,
worker.worker_id.as_str(),
) else {
continue;
};
worker.resource_key = Some(resource_key);
let worker_ref = RuntimeWorkerRef::new(&runtime_id, worker.worker_id.as_str());
let revision = next_revision(&mut revisions, &worker_ref);
worker.subject_revision = revision;
@@ -474,6 +485,7 @@ async fn run_workspace_workers(
}
fn install_snapshot(
api: &WorkspaceApi,
workers: &mut HashMap<String, BTreeMap<String, SubscriptionWorker>>,
runtime_id: &str,
snapshot: SubscriptionSnapshot,
@@ -487,6 +499,14 @@ fn install_snapshot(
let mut projected = BTreeMap::new();
for mut worker in snapshot_workers {
worker.runtime_id = Some(runtime_id.to_string());
let Ok(Some(resource_key)) = api.store.resource_key(
&api.config.workspace_id,
WorkspaceResourceKind::Worker,
worker.worker_id.as_str(),
) else {
continue;
};
worker.resource_key = Some(resource_key);
projected.insert(worker.worker_id.to_string(), worker);
}
workers.insert(runtime_id.to_string(), projected);