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

feat: adopt Workspace resource keys

Browse Source
This commit is contained in:
Hare
2026-08-21 10:20:32 +09:00
parent 9989aed916
commit 85d1815dcf
33 changed files with 678 additions and 294 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/client/src/backend_runtime.rs
+2 -2
View File
@@ -137,8 +137,7 @@ pub struct BackendWorkerCapabilitySummary {
pub struct BackendWorkerSummary {
pub runtime_id: String,
pub worker_id: String,
#[serde(default)]
pub human_key: Option<String>,
pub resource_key: String,
pub host_id: String,
#[serde(default)]
pub display_name: String,
@@ -651,6 +650,7 @@ mod tests {
let payload = serde_json::json!({
"runtime_id": "arcadia",
"worker_id": "worker-opaque-64",
"resource_key": "W-64",
"host_id": "host",
"display_name": "Coder",
"label": "Coder",
crates/protocol/src/subscription.rs
+8
View File
@@ -551,6 +551,10 @@ pub struct SubscriptionWorker {
/// Runtime producers leave this unset because the connection identifies the Runtime.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub runtime_id: Option<String>,
/// Workspace-scoped canonical resource key. Runtime producers leave this unset;
/// Workspace-facing projections must populate it before publishing the Worker.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub resource_key: Option<String>,
/// Producer-owned monotonic revision for this Worker subject.
pub subject_revision: u64,
pub state: SubscriptionWorkerState,
@@ -574,6 +578,9 @@ impl SubscriptionWorker {
if let Some(runtime_id) = &self.runtime_id {
validate_identifier("runtime_id", runtime_id, MAX_RESOURCE_ID_BYTES)?;
}
if let Some(resource_key) = &self.resource_key {
validate_identifier("resource_key", resource_key, MAX_RESOURCE_ID_BYTES)?;
}
if let Some(repository_id) = &self.repository_id {
validate_identifier("repository_id", repository_id, MAX_RESOURCE_ID_BYTES)?;
}
@@ -796,6 +803,7 @@ mod tests {
SubscriptionWorker {
worker_id: worker_id(value),
runtime_id: None,
resource_key: None,
subject_revision: 0,
state: SubscriptionWorkerState::Idle,
has_running_internal_workers: false,
crates/ticket/src/lib.rs
+49 -38
View File
@@ -27,7 +27,9 @@ mod sqlite_schema;
pub mod tool;
pub use sqlite_schema::{
LATEST_SQLITE_TICKET_SCHEMA_VERSION, migrate_sqlite_ticket_schema, verify_sqlite_ticket_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,
};
const REQUIRED_FIELDS: [&str; 4] = ["title", "state", "created_at", "updated_at"];
@@ -124,7 +126,7 @@ fn read_ticket_summary_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<TicketSu
let workflow_state = row.get::<_, String>(7)?;
Ok(TicketSummary {
id: row.get(0)?,
human_key: None,
resource_key: None,
slug: row.get(1)?,
title: row.get(2)?,
status: ExtensibleTicketStatus::from(row.get::<_, String>(3)?.as_str()),
@@ -928,7 +930,7 @@ impl TicketListQuery {
pub struct TicketRef {
pub id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub human_key: Option<String>,
pub resource_key: Option<String>,
pub slug: String,
pub status: TicketStatus,
}
@@ -1544,7 +1546,7 @@ pub struct OrchestrationPlanRecord {
pub struct TicketMeta {
pub id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub human_key: Option<String>,
pub resource_key: Option<String>,
pub slug: String,
pub title: String,
pub status: ExtensibleTicketStatus,
@@ -1569,7 +1571,7 @@ pub struct TicketMeta {
pub struct TicketSummary {
pub id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub human_key: Option<String>,
pub resource_key: Option<String>,
pub slug: String,
pub title: String,
pub status: ExtensibleTicketStatus,
@@ -2677,7 +2679,7 @@ impl SqliteTicketBackend {
.collect::<std::result::Result<Vec<_>, _>>()
.map_err(sqlite_err)?;
for summary in &mut summaries {
summary.human_key = Self::human_key_for(conn, &self.workspace_id, &summary.id)?;
summary.resource_key = Self::resource_key_for(conn, &self.workspace_id, &summary.id)?;
}
let has_more = summaries.len() > query.limit;
summaries.truncate(query.limit);
@@ -2762,7 +2764,7 @@ impl SqliteTicketBackend {
updated_at,
) = row.map_err(sqlite_err)?;
summaries.push(TicketSummary {
human_key: Self::human_key_for(conn, &self.workspace_id, &id)?,
resource_key: Self::resource_key_for(conn, &self.workspace_id, &id)?,
id,
slug,
title,
@@ -2944,8 +2946,8 @@ impl SqliteTicketBackend {
"SELECT ticket_id FROM typed_tickets
WHERE workspace_id = ?1 AND (ticket_id = ?2 OR slug = ?2)
UNION
SELECT resource_id FROM workspace_resource_human_keys
WHERE workspace_id = ?1 AND resource_kind = 'ticket' AND human_key = ?2
SELECT resource_id FROM workspace_resource_keys
WHERE workspace_id = ?1 AND resource_kind = 'ticket' AND resource_key = ?2
ORDER BY 1",
)
.map_err(sqlite_err)?;
@@ -2967,13 +2969,13 @@ impl SqliteTicketBackend {
}
}
fn human_key_for(
fn resource_key_for(
conn: &Connection,
workspace_id: &str,
ticket_id: &str,
) -> Result<Option<String>> {
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 = 'ticket' AND resource_id = ?2",
params![workspace_id, ticket_id],
|row| row.get(0),
@@ -2982,44 +2984,50 @@ impl SqliteTicketBackend {
.map_err(sqlite_err)
}
fn allocate_human_key(
fn allocate_resource_key(
conn: &Connection,
workspace_id: &str,
ticket_id: &str,
allocated_at: &str,
) -> Result<String> {
if let Some(existing) = Self::human_key_for(conn, workspace_id, ticket_id)? {
if let Some(existing) = Self::resource_key_for(conn, workspace_id, ticket_id)? {
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, 'ticket', 1)",
params![workspace_id],
)
.map_err(sqlite_err)?;
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 = 'ticket'",
params![workspace_id],
|row| row.get(0),
)
.map_err(sqlite_err)?;
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 = 'ticket' AND next_sequence = ?2",
params![workspace_id, sequence, sequence + 1],
)
.map_err(sqlite_err)?;
let human_key = format!("T-{sequence}");
let resource_key = format!("T-{sequence}");
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, 'ticket', ?2, ?3, ?4, ?5)",
params![workspace_id, ticket_id, sequence, human_key, allocated_at],
params![
workspace_id,
ticket_id,
sequence,
resource_key,
allocated_at
],
)
.map_err(sqlite_err)?;
Ok(human_key)
Ok(resource_key)
}
fn ticket_exists(&self, conn: &Connection, id: &str) -> Result<bool> {
@@ -3165,7 +3173,7 @@ impl SqliteTicketBackend {
let state_raw: String = row.get(12)?;
Ok(TicketMeta {
id: row.get(0)?,
human_key: None,
resource_key: None,
slug: row.get(1)?,
title: row.get(2)?,
status: ExtensibleTicketStatus::from(row.get::<_, String>(3)?.as_str()),
@@ -3193,7 +3201,7 @@ impl SqliteTicketBackend {
self.full_ticket_load_count.fetch_add(1, Ordering::SeqCst);
let (mut meta, body, resolution): (TicketMeta, String, Option<String>) = conn.query_row(r#"SELECT ticket_id, slug, title, status, kind, priority, created_at, updated_at, assignee, readiness, body, resolution, workflow_state, workflow_state_explicit, queued_by, queued_at, repository_id, ref_selector FROM typed_tickets WHERE workspace_id = ?1 AND ticket_id = ?2"#,
params![self.workspace_id, ticket_id], |row| Ok((Self::ticket_meta_from_row(row)?, row.get(10)?, row.get(11)?))).optional().map_err(sqlite_err)?.ok_or_else(|| TicketError::NotFound(ticket_id.to_string()))?;
meta.human_key = Self::human_key_for(conn, &self.workspace_id, ticket_id)?;
meta.resource_key = Self::resource_key_for(conn, &self.workspace_id, ticket_id)?;
meta.labels = self.load_ordered_values(conn, "typed_ticket_labels", "label", ticket_id)?;
meta.risk_flags =
self.load_ordered_values(conn, "typed_ticket_risk_flags", "risk_flag", ticket_id)?;
@@ -3365,7 +3373,7 @@ impl SqliteTicketBackend {
let mut summaries = Vec::new();
for row in rows {
let mut meta = row.map_err(sqlite_err)?;
meta.human_key = Self::human_key_for(conn, &self.workspace_id, &meta.id)?;
meta.resource_key = Self::resource_key_for(conn, &self.workspace_id, &meta.id)?;
if !filter.matches_state(meta.workflow_state) {
continue;
}
@@ -3515,7 +3523,7 @@ impl TicketBackend for SqliteTicketBackend {
};
let meta = TicketMeta {
id: id.clone(),
human_key: None,
resource_key: None,
slug: input.slug.clone().unwrap_or_else(|| id.clone()),
title: input.title,
status,
@@ -3567,11 +3575,11 @@ impl TicketBackend for SqliteTicketBackend {
relations: TicketRelationView::default(),
resolution: None,
};
let human_key = Self::allocate_human_key(conn, &self.workspace_id, &id, &now)?;
let resource_key = Self::allocate_resource_key(conn, &self.workspace_id, &id, &now)?;
self.insert_ticket(conn, &ticket)?;
Ok(TicketRef {
id: id.clone(),
human_key: Some(human_key),
resource_key: Some(resource_key),
slug: id,
status: TicketStatus::Open,
})
@@ -4218,7 +4226,7 @@ impl TicketBackend for LocalTicketBackend {
atomic_write(&dir.join("thread.md"), thread.as_bytes())?;
Ok(TicketRef {
id: id.clone(),
human_key: None,
resource_key: None,
slug: id,
status: TicketStatus::Open,
})
@@ -5256,7 +5264,7 @@ fn ticket_meta(frontmatter: TicketItemFrontmatter, id: String) -> TicketMeta {
};
TicketMeta {
id: id.clone(),
human_key: None,
resource_key: None,
slug: id,
title: frontmatter.title.unwrap_or_default(),
status,
@@ -5281,7 +5289,7 @@ fn ticket_meta(frontmatter: TicketItemFrontmatter, id: String) -> TicketMeta {
fn ticket_summary_from_meta(meta: TicketMeta) -> TicketSummary {
TicketSummary {
id: meta.id,
human_key: meta.human_key,
resource_key: meta.resource_key,
slug: meta.slug,
title: meta.title,
status: meta.status,
@@ -6890,7 +6898,7 @@ mod tests {
fn summary_with_state(state: TicketWorkflowState) -> TicketSummary {
TicketSummary {
id: "000TEST".to_string(),
human_key: Some("T-1".to_string()),
resource_key: Some("T-1".to_string()),
slug: "000TEST".to_string(),
title: "Test Ticket".to_string(),
status: ExtensibleTicketStatus::Open,
@@ -7296,14 +7304,14 @@ state: planning
}
#[test]
fn sqlite_human_keys_are_workspace_scoped_monotonic_and_resolvable() {
fn sqlite_resource_keys_are_workspace_scoped_monotonic_and_resolvable() {
let tmp = TempDir::new().unwrap();
let db_path = tmp.path().join("workspace.db");
let backend = SqliteTicketBackend::open(&db_path, "workspace-a").unwrap();
let first = backend.create(NewTicket::new("First")).unwrap();
let second = backend.create(NewTicket::new("Second")).unwrap();
assert_eq!(first.human_key.as_deref(), Some("T-1"));
assert_eq!(second.human_key.as_deref(), Some("T-2"));
assert_eq!(first.resource_key.as_deref(), Some("T-1"));
assert_eq!(second.resource_key.as_deref(), Some("T-2"));
assert_eq!(backend.show("T-1".into()).unwrap().meta.id, first.id);
let projection = backend.list_workspace_projection(100).unwrap();
let projected_second = projection
@@ -7311,16 +7319,19 @@ state: planning
.iter()
.find(|item| item.summary.id == second.id)
.unwrap();
assert_eq!(projected_second.summary.human_key.as_deref(), Some("T-2"));
assert_eq!(
projected_second.summary.resource_key.as_deref(),
Some("T-2")
);
let other = SqliteTicketBackend::open(&db_path, "workspace-b").unwrap();
let other_first = other.create(NewTicket::new("Other")).unwrap();
assert_eq!(other_first.human_key.as_deref(), Some("T-1"));
assert_eq!(other_first.resource_key.as_deref(), Some("T-1"));
assert_eq!(other.show("T-1".into()).unwrap().meta.id, other_first.id);
}
#[test]
fn sqlite_human_key_allocation_is_concurrency_safe() {
fn sqlite_resource_key_allocation_is_concurrency_safe() {
let tmp = TempDir::new().unwrap();
let db_path = tmp.path().join("workspace.db");
SqliteTicketBackend::open(&db_path, "workspace-a").unwrap();
@@ -7335,7 +7346,7 @@ state: planning
backend
.create(NewTicket::new(format!("Ticket {index}")))
.unwrap()
.human_key
.resource_key
.unwrap()
})
})
crates/ticket/src/sqlite_schema.rs
+172 -20
View File
@@ -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 = 5;
pub const LATEST_SQLITE_TICKET_SCHEMA_VERSION: i64 = 6;
#[derive(Clone, Copy)]
struct Migration {
@@ -42,6 +42,11 @@ const MIGRATIONS: &[Migration] = &[
name: "add_workspace_human_keys",
apply: add_workspace_human_keys,
},
Migration {
version: 6,
name: "rename_workspace_resource_keys",
apply: rename_workspace_resource_keys,
},
];
#[derive(Clone, Copy)]
@@ -243,6 +248,24 @@ const fn column(
/// 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)?;
@@ -265,7 +288,20 @@ pub fn migrate_sqlite_ticket_schema(connection: &Connection) -> Result<()> {
let applied = load_applied_migrations(connection)?;
validate_applied_migrations(&applied)?;
for migration in MIGRATIONS {
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;
}
@@ -283,7 +319,22 @@ pub fn migrate_sqlite_ticket_schema(connection: &Connection) -> Result<()> {
.map_err(sqlite_err)?;
}
verify_sqlite_ticket_schema(connection)
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(())
}
})();
match result {
@@ -295,6 +346,47 @@ pub fn migrate_sqlite_ticket_schema(connection: &Connection) -> Result<()> {
}
}
/// 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();
@@ -591,6 +683,21 @@ fn add_workspace_human_keys(connection: &Connection) -> Result<()> {
.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,
@@ -923,10 +1030,10 @@ mod tests {
verify_sqlite_ticket_schema(&connection).unwrap();
let versions = load_applied_migrations(&connection).unwrap();
assert_eq!(versions.len(), 5);
assert_eq!(versions.len(), 6);
assert_eq!(
versions.get(&LATEST_SQLITE_TICKET_SCHEMA_VERSION),
Some(&"add_workspace_human_keys".to_string())
Some(&"rename_workspace_resource_keys".to_string())
);
}
@@ -1014,14 +1121,11 @@ mod tests {
}
#[test]
fn v5_backfills_ticket_keys_by_creation_order_and_advances_counter() {
fn v5_backfills_ticket_keys_and_v6_preserves_them_under_resource_key_schema() {
let connection = Connection::open_in_memory().unwrap();
migrate_sqlite_ticket_schema(&connection).unwrap();
migrate_sqlite_ticket_schema_through(&connection, 4).unwrap();
connection.execute_batch(
"DROP TABLE workspace_resource_human_key_counters;
DROP TABLE workspace_resource_human_keys;
DELETE FROM ticket_schema_migrations WHERE version = 5;
INSERT INTO typed_tickets (
"INSERT INTO typed_tickets (
workspace_id, ticket_id, slug, title, status, kind, priority, body,
workflow_state, workflow_state_explicit, created_at, updated_at
) VALUES
@@ -1029,12 +1133,12 @@ mod tests {
('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(&connection).unwrap();
let keys = connection
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",
WHERE workspace_id = 'workspace-1' AND resource_kind = 'ticket'
ORDER BY sequence",
)
.unwrap()
.query_map([], |row| {
@@ -1044,7 +1148,7 @@ mod tests {
.collect::<std::result::Result<Vec<_>, _>>()
.unwrap();
assert_eq!(
keys,
legacy_keys,
vec![
("earlier".into(), "T-1".into()),
("later".into(), "T-2".into())
@@ -1053,12 +1157,56 @@ mod tests {
let next: i64 = connection
.query_row(
"SELECT next_sequence FROM workspace_resource_human_key_counters
WHERE workspace_id = 'workspace-1' AND resource_kind = 'ticket'",
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]
@@ -1119,7 +1267,7 @@ mod tests {
.to_string()
.contains("unsupported Ticket schema migration version 99")
);
assert_eq!(load_applied_migrations(&connection).unwrap().len(), 6);
assert_eq!(load_applied_migrations(&connection).unwrap().len(), 7);
}
#[test]
@@ -1220,7 +1368,11 @@ mod tests {
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("DELETE FROM ticket_schema_migrations WHERE version>=3", [])
.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();
@@ -1259,6 +1411,6 @@ mod tests {
let connection = Connection::open(database).unwrap();
verify_sqlite_ticket_schema(&connection).unwrap();
assert_eq!(load_applied_migrations(&connection).unwrap().len(), 5);
assert_eq!(load_applied_migrations(&connection).unwrap().len(), 6);
}
}
crates/tui/src/backend_worker_picker.rs
+6 -13
View File
@@ -348,17 +348,6 @@ fn state_style(state: &str) -> Style {
}
}
fn short_worker_id(worker: &BackendWorkerSummary) -> String {
if let Some(human_key) = worker.human_key.as_ref() {
return human_key.clone();
}
format!(
"{}:{}",
short_text(&worker.runtime_id),
short_text(&worker.worker_id)
)
}
fn short_text(text: &str) -> String {
const MAX: usize = 24;
let mut chars = text.chars();
@@ -370,6 +359,10 @@ fn short_text(text: &str) -> String {
}
}
fn short_worker_id(worker: &BackendWorkerSummary) -> String {
worker.resource_key.clone()
}
fn working_directory_text(worker: &BackendWorkerSummary) -> String {
let Some(wd) = worker.working_directory.as_ref() else {
return "wd:—".to_string();
@@ -393,7 +386,7 @@ mod tests {
BackendWorkerSummary {
runtime_id: runtime_id.to_string(),
worker_id: worker_id.to_string(),
human_key: None,
resource_key: "W-1".to_string(),
host_id: "host".to_string(),
label: "label".to_string(),
display_name: "label".to_string(),
@@ -429,7 +422,7 @@ mod tests {
.into_iter()
.map(|span| span.content)
.collect::<String>();
assert!(text.starts_with("▶ runtime-a:worker-b"));
assert!(text.starts_with("▶ W-1"));
assert!(text.contains("[running]"));
assert!(text.contains("profile:default"));
assert!(text.contains("wd:—"));
crates/tui/src/dashboard/render.rs
+2 -2
View File
@@ -537,9 +537,9 @@ pub(super) fn panel_ticket_reference(row: &PanelRow) -> String {
.as_ref()
.map(|ticket| {
ticket
.human_key
.resource_key
.clone()
.unwrap_or_else(|| ticket.id.clone())
.unwrap_or_else(|| "resource key unavailable".to_string())
})
.unwrap_or_else(|| match &row.key {
PanelRowKey::Ticket(id) | PanelRowKey::InvalidTicket(id) => id.clone(),
crates/tui/src/dashboard/tests.rs
+9 -3
View File
@@ -1737,7 +1737,13 @@ fn panel_ticket_rows_render_state_title_then_detail_line() {
let state_start = 2;
let title_start = state_start + TICKET_STATE_COLUMN_WIDTH + 1;
let row_id = row.ticket.as_ref().unwrap().id.as_str();
let human_key = row.ticket.as_ref().unwrap().human_key.as_deref().unwrap();
let resource_key = row
.ticket
.as_ref()
.unwrap()
.resource_key
.as_deref()
.unwrap();
assert!(title_line.starts_with("▶ "));
assert!(detail_line.starts_with("│ meta "));
@@ -1747,7 +1753,7 @@ fn panel_ticket_rows_render_state_title_then_detail_line() {
display_column(&title_line, "Workspace Dashboard composer targets"),
title_start
);
assert!(detail_line.contains(human_key));
assert!(detail_line.contains(resource_key));
assert!(detail_line.contains("Gate: clear"));
assert!(detail_line.contains("Action: Wait"));
}
@@ -3266,7 +3272,7 @@ fn panel_test_ticket_row(
) -> PanelRow {
let ticket = crate::workspace_panel::TicketPanelEntry {
id: id.to_string(),
human_key: Some("T-1".to_string()),
resource_key: Some("T-1".to_string()),
title: title.to_string(),
priority: "P2".to_string(),
workflow_state: TicketWorkflowState::parse(state).unwrap_or(TicketWorkflowState::Planning),
crates/tui/src/workspace_panel.rs
+3 -3
View File
@@ -253,7 +253,7 @@ impl NextUserAction {
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct TicketPanelEntry {
pub(crate) id: String,
pub(crate) human_key: Option<String>,
pub(crate) resource_key: Option<String>,
pub(crate) title: String,
pub(crate) priority: String,
pub(crate) workflow_state: TicketWorkflowState,
@@ -1064,7 +1064,7 @@ pub(crate) fn build_current_ticket_row(
fn ticket_summary_from_meta(meta: &TicketMeta) -> TicketSummary {
TicketSummary {
id: meta.id.clone(),
human_key: meta.human_key.clone(),
resource_key: meta.resource_key.clone(),
slug: meta.slug.clone(),
title: meta.title.clone(),
status: meta.status.clone(),
@@ -1240,7 +1240,7 @@ fn ticket_row(
let next_action = projection.next_action.map(next_user_action_from_workspace);
let entry = TicketPanelEntry {
id: summary.id.clone(),
human_key: summary.human_key.clone(),
resource_key: summary.resource_key.clone(),
title: summary.title.clone(),
priority: summary.priority.clone(),
workflow_state: summary.workflow_state,
crates/worker-runtime/src/runtime.rs
+1
View File
@@ -2262,6 +2262,7 @@ impl RuntimeState {
Ok(SubscriptionWorker {
worker_id,
runtime_id: None,
resource_key: None,
subject_revision: self
.worker_subject_revisions
.get(&worker.worker_id)
crates/worker/src/feature/builtin/ticket.rs
+1 -1
View File
@@ -1781,7 +1781,7 @@ provider = "github"
assert!(request.contains("\"title\":\"HTTP ticket\""));
let response_body = serde_json::to_string(&TicketRef {
id: "01TEST".to_string(),
human_key: None,
resource_key: None,
slug: "http-ticket".to_string(),
status: ticket::TicketStatus::Open,
})
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)]
@@ -327,7 +327,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>,
@@ -379,6 +379,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>,
@@ -414,7 +415,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,
}
@@ -422,7 +423,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>,
@@ -435,7 +436,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
+82 -34
View File
@@ -9930,11 +9930,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 }))
}
@@ -9952,12 +9961,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(),
@@ -11041,11 +11060,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 {
@@ -11914,7 +11942,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(),
@@ -16065,10 +16093,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 {
@@ -18114,7 +18143,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",
@@ -18150,7 +18191,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(),
}],
@@ -18347,7 +18388,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,
@@ -18363,7 +18404,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 =
@@ -19256,16 +19297,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(
@@ -19602,12 +19650,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);
"#,
)
@@ -19824,13 +19872,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
+172 -53
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,
@@ -1179,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,
@@ -1312,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,
@@ -1320,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),
@@ -1339,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),
)
@@ -1544,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],
@@ -1987,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,
@@ -2765,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
@@ -2782,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,
@@ -2824,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(())
})
}
@@ -6230,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,
@@ -6239,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),
@@ -6249,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<()> {
@@ -6359,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")?;
@@ -6943,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(),
@@ -6969,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(),
@@ -7574,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);
@@ -7588,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();
@@ -7631,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();
@@ -7667,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()
@@ -7796,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(
@@ -7914,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());
}
@@ -7947,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());
@@ -8014,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'",
@@ -8192,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"))
@@ -8209,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)
@@ -8217,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
@@ -8244,7 +8352,7 @@ INSERT INTO workdir_registry (
.unwrap();
assert_eq!(
store
.resource_human_key(
.resource_key(
"workspace-a",
WorkspaceResourceKind::Objective,
"objective-internal"
@@ -8298,7 +8406,7 @@ INSERT INTO workdir_registry (
);
assert_eq!(
store
.resource_human_key(
.resource_key(
"workspace-a",
WorkspaceResourceKind::Worker,
&reserved.to_string()
@@ -8318,7 +8426,7 @@ INSERT INTO workdir_registry (
.unwrap();
assert_eq!(
store
.resource_human_key(
.resource_key(
"workspace-a",
WorkspaceResourceKind::Worker,
&second.to_string()
@@ -8708,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}");
}
@@ -8826,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#"
@@ -8872,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) \
@@ -8935,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'",
@@ -8952,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(
@@ -9236,7 +9344,7 @@ INSERT INTO ticket_worker_assignment_events (
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(
@@ -9552,7 +9660,7 @@ WHERE workspace_id = 'workspace-a'
.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| {
@@ -9741,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,
@@ -9807,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,
@@ -9927,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();
@@ -10198,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);