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

fix: integrate develop and preserve invocation notification recovery

Browse Source
This commit is contained in:
Hare
2026-10-05 22:25:32 +09:00
parent c847c2ffad 3deef49897
commit 5670c20a6a
92 changed files with 14528 additions and 1005 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/server.rs
+2449 -152
View File
File diff suppressed because it is too large Load Diff
crates/workspace-server/src/store.rs
+658 -30
View File
@@ -28,7 +28,7 @@ use crate::workspace_deletion::WorkspaceDeletionStore;
use crate::{Error, Result};
const OLDEST_SCHEMA_VERSION: i64 = 50;
const LATEST_SCHEMA_VERSION: i64 = 77;
const LATEST_SCHEMA_VERSION: i64 = 78;
const SCHEMA_BASELINE_NAME: &str = "workspace schema baseline";
const WORKSPACE_RUNTIME_BINDINGS_MIGRATION_NAME: &str = "workspace runtime bindings";
const RUNTIME_BINDING_AUDIT_MIGRATION_NAME: &str = "workspace Runtime binding revision and audit";
@@ -77,6 +77,7 @@ const ARCHIVE_OBSERVE_GRANTS_MIGRATION_NAME: &str =
"preserve explicit observe grants for committed Worker Session archives";
const BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME: &str =
"durable terminal Backend Job Worker cleanup";
const WORKDIR_CONNECTION_ID_MIGRATION_NAME: &str = "durable Workdir attachment connection identity";
const TICKET_SCHEMA_VERSION_WITH_TARGETS: i64 = 7;
const TICKET_SCHEMA_BASELINE_NAME: &str = "ticket schema baseline";
const WORKER_REGISTRY_PROJECTION_SCHEMA: &str = r#"
@@ -259,6 +260,11 @@ const MIGRATIONS: &[Migration] = &[
name: BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME,
apply: migrate_backend_job_worker_cleanup_v76_to_v77,
},
Migration {
version: 78,
name: WORKDIR_CONNECTION_ID_MIGRATION_NAME,
apply: migrate_workdir_connection_id_v77_to_v78,
},
];
#[derive(Clone, Copy)]
@@ -945,6 +951,23 @@ pub struct WorkdirRegistryRecord {
pub updated_at: String,
}
/// Projection inputs captured together with the catalog revision in one SQLite snapshot.
#[derive(Debug, Clone)]
pub struct WorkdirCatalogEntry {
pub record: WorkdirRegistryRecord,
pub repository_key: Option<String>,
/// None for Repository records and unsupported legacy External grants.
pub external_grant_permissions: Option<workdir::workspace::WorkdirPermissionSummary>,
pub occupied_by: Option<workdir::workspace::WorkingDirectoryOccupancy>,
}
#[derive(Debug, Clone)]
pub struct WorkdirCatalogPage {
pub entries: Vec<WorkdirCatalogEntry>,
pub next_cursor: Option<String>,
pub revision: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct ExternalWorkdirGrantRecord {
pub grant_id: String,
@@ -963,6 +986,8 @@ pub struct ExternalWorkdirGrantRecord {
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct WorkerWorkdirLinkRecord {
/// Assigned by the ledger on attach; preserved only by explicit compensation.
pub connection_id: String,
pub workspace_id: String,
pub worker: RuntimeWorkerRef,
pub workdir_id: String,
@@ -1888,6 +1913,12 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore {
workspace_id: &str,
limit: usize,
) -> Result<Vec<WorkdirRegistryRecord>>;
fn workdir_catalog_page(
&self,
workspace_id: &str,
limit: u32,
cursor: Option<&str>,
) -> Result<WorkdirCatalogPage>;
fn delete_workdir_registry(&self, workspace_id: &str, workdir_id: &str) -> Result<bool>;
fn create_external_workdir_grant(
@@ -1983,6 +2014,31 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore {
expected_workdir_id: Option<&str>,
unlinked_at: &str,
) -> Result<Option<WorkerWorkdirLinkRecord>>;
fn detach_worker_workdir_connection(
&self,
workspace_id: &str,
worker: &RuntimeWorkerRef,
alias: &str,
expected_connection_id: &str,
unlinked_at: &str,
) -> Result<WorkerWorkdirLinkRecord>;
fn restore_worker_workdir_connection(&self, record: &WorkerWorkdirLinkRecord) -> Result<()>;
fn list_worker_workdir_links_page(
&self,
workspace_id: &str,
worker: &RuntimeWorkerRef,
limit: u32,
offset: u32,
connection_id: Option<&str>,
) -> Result<Vec<WorkerWorkdirLinkRecord>>;
fn list_worker_workdir_links_page_with_revision(
&self,
workspace_id: &str,
worker: &RuntimeWorkerRef,
limit: u32,
offset: u32,
connection_id: Option<&str>,
) -> Result<(Vec<WorkerWorkdirLinkRecord>, String)>;
fn worker_workdir_link_history_exists(
&self,
workspace_id: &str,
@@ -8876,6 +8932,72 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
})
}
fn workdir_catalog_page(
&self,
workspace_id: &str,
limit: u32,
cursor: Option<&str>,
) -> Result<WorkdirCatalogPage> {
if limit == 0 || limit > 100 {
return Err(Error::InvalidInput(
"catalog page limit must be 1..=100".to_string(),
));
}
if cursor.is_some_and(|id| {
id.trim().is_empty() || id.len() > 128 || id.chars().any(char::is_control)
}) {
return Err(Error::InvalidInput(
"catalog cursor must contain 1..=128 bytes without control characters".to_string(),
));
}
self.with_conn(|conn| {
// Also fence writers using other SQLite connections, not only this mutex.
let tx = conn.unchecked_transaction()?;
let sql = workdir_catalog_select_sql();
let mut hasher = Sha256::new();
update_public_collection_digest(&mut hasher, &("workdir-catalog:v1", workspace_id))?;
{
let mut stmt = tx.prepare(&format!("{sql} WHERE wr.workspace_id = ?1 ORDER BY wr.workdir_id"))?;
let mut rows = stmt.query(params![workspace_id])?;
while let Some(row) = rows.next()? {
let (entry, connection_id) = read_workdir_catalog_entry(row)?;
let record = &entry.record;
// Safe projection fields and timestamps only: no paths, provider identity,
// credentials, Repository UUIDs, or raw internal records.
update_public_collection_digest(&mut hasher, &(
&record.workdir_id, &record.display_name, &entry.repository_key,
record.source.external_grant_id(), &entry.external_grant_permissions,
(&record.creation_selector, &record.creation_ref, &record.creation_tree),
(&record.current_selector, &record.current_ref, &record.current_tree),
record.observed_at_epoch_seconds,
(&record.materialization_status, &record.cleanliness),
(&record.created_at, &record.updated_at),
&entry.occupied_by, connection_id,
))?;
}
}
let mut stmt = tx.prepare(&format!(
"{sql} WHERE wr.workspace_id = ?1 AND (?2 IS NULL OR wr.workdir_id > ?2) ORDER BY wr.workdir_id ASC LIMIT ?3"
))?;
let mut rows = stmt.query(params![workspace_id, cursor, limit + 1])?;
let mut entries = Vec::new();
while let Some(row) = rows.next()? {
entries.push(read_workdir_catalog_entry(row)?.0);
}
let next_cursor = if entries.len() > limit as usize {
entries.truncate(limit as usize);
entries.last().map(|entry| entry.record.workdir_id.clone())
} else {
None
};
Ok(WorkdirCatalogPage {
entries,
next_cursor,
revision: hasher.finalize().iter().map(|byte| format!("{byte:02x}")).collect::<String>(),
})
})
}
fn delete_workdir_registry(&self, workspace_id: &str, workdir_id: &str) -> Result<bool> {
self.with_conn(|conn| {
let tx = rusqlite::Transaction::new_unchecked(
@@ -9349,6 +9471,8 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
"reserved attachment finalization requires an active attachment".to_string(),
));
}
let mut record = record.clone();
record.connection_id = uuid::Uuid::now_v7().to_string();
self.with_conn(|conn| {
let tx = rusqlite::Transaction::new_unchecked(
conn,
@@ -9370,11 +9494,12 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
}
tx.execute(
r#"INSERT INTO worker_workdir_links (
workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL)
workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL, ?8)
ON CONFLICT(workspace_id, runtime_id, worker_id, workdir_id, alias) DO UPDATE SET
capabilities = excluded.capabilities,
linked_at = excluded.linked_at,
connection_id = excluded.connection_id,
unlinked_at = NULL"#,
params![
record.workspace_id,
@@ -9384,6 +9509,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
record.alias,
encode_workdir_link_capabilities(record.capabilities)?,
record.linked_at,
record.connection_id,
],
)
.map_err(|error| {
@@ -9421,6 +9547,8 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
"a new attachment cannot already be unlinked".to_string(),
));
}
let mut record = record.clone();
record.connection_id = uuid::Uuid::now_v7().to_string();
self.with_conn(|conn| {
let tx = rusqlite::Transaction::new_unchecked(
conn,
@@ -9456,7 +9584,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
}
let active_for_alias = tx
.query_row(
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
FROM worker_workdir_links
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3
AND alias = ?4 AND unlinked_at IS NULL"#,
@@ -9499,7 +9627,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
}
let active_for_workdir = tx
.query_row(
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
FROM worker_workdir_links
WHERE workspace_id = ?1 AND workdir_id = ?2 AND unlinked_at IS NULL"#,
params![record.workspace_id, record.workdir_id],
@@ -9514,11 +9642,12 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
}
let write = tx.execute(
r#"INSERT INTO worker_workdir_links (
workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL)
workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL, ?8)
ON CONFLICT(workspace_id, runtime_id, worker_id, workdir_id, alias) DO UPDATE SET
capabilities = excluded.capabilities,
linked_at = excluded.linked_at,
connection_id = excluded.connection_id,
unlinked_at = NULL"#,
params![
record.workspace_id,
@@ -9528,6 +9657,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
record.alias,
encode_workdir_link_capabilities(record.capabilities)?,
record.linked_at,
record.connection_id,
],
);
if let Err(error) = write {
@@ -9556,7 +9686,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
rusqlite::TransactionBehavior::Immediate,
)?;
let mut stmt = tx.prepare(
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
FROM worker_workdir_links
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3
AND unlinked_at IS NULL"#,
@@ -9571,7 +9701,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
let mut persisted_identity = persisted
.iter()
.map(|link| (link.alias.as_str(), link.workdir_id.as_str()))
.map(|link| (link.alias.as_str(), link.workdir_id.as_str(), link.connection_id.as_str()))
.collect::<Vec<_>>();
persisted_identity.sort_unstable();
let mut requested_identity = active_links
@@ -9592,7 +9722,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
link.alias
))
})?;
Ok((link.alias.as_str(), link.workdir_id.as_str()))
Ok((link.alias.as_str(), link.workdir_id.as_str(), link.connection_id.as_str()))
})
.collect::<Result<Vec<_>>>()?;
requested_identity.sort_unstable();
@@ -9644,7 +9774,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
)?;
let active = tx
.query_row(
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
FROM worker_workdir_links
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3
AND (?4 IS NULL OR workdir_id = ?4) AND unlinked_at IS NULL"#,
@@ -9683,6 +9813,202 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
})
}
fn detach_worker_workdir_connection(
&self,
workspace_id: &str,
worker: &RuntimeWorkerRef,
alias: &str,
expected_connection_id: &str,
unlinked_at: &str,
) -> Result<WorkerWorkdirLinkRecord> {
self.with_conn(|conn| {
let tx = rusqlite::Transaction::new_unchecked(
conn,
rusqlite::TransactionBehavior::Immediate,
)?;
let active = tx
.query_row(
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities,
linked_at, unlinked_at, connection_id
FROM worker_workdir_links
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3
AND alias = ?4 AND connection_id = ?5 AND unlinked_at IS NULL"#,
params![
workspace_id,
worker.runtime_id,
worker.worker_id,
alias,
expected_connection_id
],
read_worker_workdir_link_record,
)
.optional()?
.ok_or_else(|| {
Error::WorkdirAttachmentConflict(format!(
"Workdir attachment `{alias}` connection changed during detach"
))
})?;
let changed = tx.execute(
r#"UPDATE worker_workdir_links SET unlinked_at = ?6
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3
AND alias = ?4 AND connection_id = ?5 AND unlinked_at IS NULL"#,
params![
workspace_id,
worker.runtime_id,
worker.worker_id,
alias,
expected_connection_id,
unlinked_at
],
)?;
if changed != 1 {
return Err(Error::WorkdirAttachmentConflict(format!(
"Workdir attachment `{alias}` connection changed during detach"
)));
}
tx.commit()?;
Ok(WorkerWorkdirLinkRecord {
unlinked_at: Some(unlinked_at.to_string()),
..active
})
})
}
fn restore_worker_workdir_connection(&self, record: &WorkerWorkdirLinkRecord) -> Result<()> {
self.with_conn(|conn| {
// Compensation reactivates only the same lifetime, never an intervening attach.
let changed = conn.execute(
r#"UPDATE worker_workdir_links SET unlinked_at = NULL
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3
AND workdir_id = ?4 AND alias = ?5 AND connection_id = ?6
AND unlinked_at IS NOT NULL
AND NOT EXISTS (
SELECT 1 FROM worker_workdir_attachment_reservations
WHERE workspace_id = ?1 AND workdir_id = ?4
)"#,
params![record.workspace_id, record.worker.runtime_id, record.worker.worker_id,
record.workdir_id, record.alias, record.connection_id],
).map_err(|error| {
if matches!(error, rusqlite::Error::SqliteFailure(ref code, _) if code.extended_code == rusqlite::ffi::SQLITE_CONSTRAINT_UNIQUE) {
Error::WorkdirAttachmentConflict("Workdir acquired another active attachment during compensation".to_string())
} else {
error.into()
}
})?;
if changed != 1 {
return Err(Error::WorkdirAttachmentConflict(
"Workdir attachment connection changed during compensation".to_string(),
));
}
Ok(())
})
}
fn list_worker_workdir_links_page(
&self,
workspace_id: &str,
worker: &RuntimeWorkerRef,
limit: u32,
offset: u32,
connection_id: Option<&str>,
) -> Result<Vec<WorkerWorkdirLinkRecord>> {
self.list_worker_workdir_links_page_with_revision(
workspace_id,
worker,
limit,
offset,
connection_id,
)
.map(|(links, _)| links)
}
fn list_worker_workdir_links_page_with_revision(
&self,
workspace_id: &str,
worker: &RuntimeWorkerRef,
limit: u32,
offset: u32,
connection_id: Option<&str>,
) -> Result<(Vec<WorkerWorkdirLinkRecord>, String)> {
if limit == 0 || limit > 101 {
return Err(Error::InvalidInput(
"attachment page limit must be 1..=101".to_string(),
));
}
if connection_id.is_some_and(|id| {
id.trim().is_empty() || id.len() > 128 || id.chars().any(char::is_control)
}) {
return Err(Error::InvalidInput(
"attachment connection_id must contain 1..=128 bytes without control characters"
.to_string(),
));
}
self.with_conn(|conn| {
let tx = conn.unchecked_transaction()?;
let mut hasher = Sha256::new();
update_public_collection_digest(
&mut hasher,
&(
"workdir-connections:v1",
workspace_id,
&worker.runtime_id,
&worker.worker_id,
),
)?;
{
let mut stmt = tx.prepare(
"SELECT alias, connection_id, workdir_id, capabilities FROM worker_workdir_links
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3
AND unlinked_at IS NULL ORDER BY alias"
)?;
let mut rows =
stmt.query(params![workspace_id, worker.runtime_id, worker.worker_id])?;
while let Some(row) = rows.next()? {
let alias: String = row.get(0)?;
let id: String = row.get(1)?;
let workdir: String = row.get(2)?;
let capabilities =
decode_workdir_link_capabilities(&row.get::<_, String>(3)?, 3).map_err(
|_| Error::Store("invalid public connection capabilities".to_string()),
)?;
update_public_collection_digest(
&mut hasher,
&(alias, id, workdir, capabilities),
)?;
}
}
let mut stmt = tx.prepare(
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities,
linked_at, unlinked_at, connection_id
FROM worker_workdir_links
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3
AND unlinked_at IS NULL
AND (?6 IS NULL OR connection_id = ?6)
ORDER BY alias LIMIT ?4 OFFSET ?5"#,
)?;
let rows = stmt.query_map(
params![
workspace_id,
worker.runtime_id,
worker.worker_id,
limit,
offset,
connection_id
],
read_worker_workdir_link_record,
)?;
let links = rows.collect::<std::result::Result<Vec<_>, _>>()?;
Ok((
links,
hasher
.finalize()
.iter()
.map(|byte| format!("{byte:02x}"))
.collect::<String>(),
))
})
}
fn worker_workdir_link_history_exists(
&self,
workspace_id: &str,
@@ -9708,7 +10034,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
) -> Result<Vec<WorkerWorkdirLinkRecord>> {
self.with_conn(|conn| {
let mut stmt = conn.prepare(
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
FROM worker_workdir_links
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 AND unlinked_at IS NULL
ORDER BY linked_at DESC"#,
@@ -9729,7 +10055,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
) -> Result<Vec<WorkerWorkdirLinkRecord>> {
self.with_conn(|conn| {
let mut stmt = conn.prepare(
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
FROM worker_workdir_links
WHERE workspace_id = ?1 AND workdir_id = ?2 AND unlinked_at IS NULL
ORDER BY linked_at DESC"#,
@@ -9750,7 +10076,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
) -> Result<Option<WorkerWorkdirLinkRecord>> {
self.with_conn(|conn| {
conn.query_row(
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
FROM worker_workdir_links
WHERE workspace_id = ?1 AND workdir_id = ?2
ORDER BY linked_at DESC, rowid DESC
@@ -10497,6 +10823,7 @@ fn read_worker_workdir_link_record(
capabilities,
linked_at: row.get(6)?,
unlinked_at: row.get(7)?,
connection_id: row.get(8)?,
})
}
@@ -11442,6 +11769,96 @@ fn require_expected_ticket_assignment(
)))
}
// Length-framed canonical JSON prevents ambiguous concatenations and never hashes raw ledgers.
fn update_public_collection_digest(hasher: &mut Sha256, value: &impl Serialize) -> Result<()> {
let bytes = serde_json::to_vec(value)
.map_err(|_| Error::Store("unable to encode public collection metadata".to_string()))?;
hasher.update((bytes.len() as u64).to_be_bytes());
hasher.update(bytes);
Ok(())
}
fn workdir_catalog_select_sql() -> String {
format!(
"SELECT wr.*, r.repository_key, g.permissions,
l.runtime_id, l.worker_id, w.display_name, l.linked_at, l.connection_id
FROM ({}) wr
LEFT JOIN repositories r ON r.workspace_id = wr.workspace_id AND r.repository_id = wr.repository_id
LEFT JOIN external_workdir_grants g ON g.workspace_id = wr.workspace_id AND g.grant_id = wr.external_grant_id
LEFT JOIN worker_workdir_links l ON l.rowid = (
SELECT rowid FROM worker_workdir_links
WHERE workspace_id = wr.workspace_id AND workdir_id = wr.workdir_id AND unlinked_at IS NULL
ORDER BY linked_at DESC, rowid DESC LIMIT 1
)
LEFT JOIN worker_registry w ON w.workspace_id = l.workspace_id AND w.runtime_id = l.runtime_id AND w.worker_id = l.worker_id",
workdir_registry_select_sql("")
)
}
fn read_workdir_catalog_entry(
row: &rusqlite::Row<'_>,
) -> Result<(WorkdirCatalogEntry, Option<String>)> {
let record = read_workdir_registry_record(row)?;
let repository_key: Option<String> = row.get(18)?;
let external_grant_permissions = match &record.source {
WorkdirRegistrySource::Repository { .. } => {
if repository_key.is_none() {
return Err(Error::Store(
"Workdir catalog Repository is unavailable".to_string(),
));
}
None
}
WorkdirRegistrySource::ExternalGrant { .. } => {
let permissions: Option<String> = row.get(19)?;
match permissions.as_deref() {
Some("read_only") => Some(workdir::workspace::WorkdirPermissionSummary {
read: true,
write: false,
command: false,
}),
Some("read_write") => Some(workdir::workspace::WorkdirPermissionSummary {
read: true,
write: true,
command: false,
}),
Some("read_write_command") => Some(workdir::workspace::WorkdirPermissionSummary {
read: true,
write: true,
command: true,
}),
Some("command_only" | "read_command") => None,
_ => {
return Err(Error::Store(
"Workdir catalog External grant permissions are unavailable".to_string(),
));
}
}
}
};
let runtime_id: Option<String> = row.get(20)?;
let occupied_by = match runtime_id {
Some(runtime_id) => Some(workdir::workspace::WorkingDirectoryOccupancy {
runtime_id,
worker_id: row.get(21)?,
display_name: row.get::<_, Option<String>>(22)?.ok_or_else(|| {
Error::Store("Workdir catalog occupant is unavailable".to_string())
})?,
linked_at: row.get(23)?,
}),
None => None,
};
Ok((
WorkdirCatalogEntry {
record,
repository_key,
external_grant_permissions,
occupied_by,
},
row.get(24)?,
))
}
fn workdir_registry_select_sql(where_clause: &str) -> String {
format!(
"SELECT workspace_id, workdir_id, display_name, source_kind, runtime_id, repository_id, external_grant_id, \
@@ -14486,6 +14903,27 @@ fn migrate_backend_job_worker_cleanup_v76_to_v77(conn: &Connection) -> Result<()
Ok(())
}
fn migrate_workdir_connection_id_v77_to_v78(conn: &Connection) -> Result<()> {
use rusqlite::TransactionBehavior;
let current = current_schema_version(conn)?;
if current != 77 {
return Err(Error::Store(format!(
"expected schema version 77 before {WORKDIR_CONNECTION_ID_MIGRATION_NAME} migration, found {current}"
)));
}
let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Exclusive)?;
tx.execute_batch(
"ALTER TABLE worker_workdir_links ADD COLUMN connection_id TEXT NOT NULL DEFAULT '';
UPDATE worker_workdir_links SET connection_id = lower(hex(randomblob(16)));",
)?;
tx.execute(
"INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)",
params![78_i64, WORKDIR_CONNECTION_ID_MIGRATION_NAME],
)?;
tx.commit()?;
Ok(())
}
fn create_latest_workspace_schema(conn: &Connection) -> Result<()> {
conn.execute_batch("DROP TABLE IF EXISTS typed_ticket_targets;")?;
conn.execute_batch(include_str!("latest_schema.sql"))?;
@@ -14546,6 +14984,7 @@ fn create_latest_workspace_schema(conn: &Connection) -> Result<()> {
);
CREATE INDEX typed_ticket_targets_workspace_repository
ON typed_ticket_targets(workspace_id, repository_key, ticket_id);
ALTER TABLE worker_workdir_links ADD COLUMN connection_id TEXT NOT NULL DEFAULT '';
ALTER TABLE worker_workdir_links
ADD COLUMN capabilities TEXT NOT NULL
CHECK (capabilities IN (
@@ -15610,6 +16049,46 @@ mod tests {
Ok(())
}
#[test]
fn workdir_connection_migration_backfills_distinct_durable_lifetimes() {
let conn = Connection::open_in_memory().unwrap();
conn.execute_batch(
"CREATE TABLE __yoi_schema_migrations (version INTEGER PRIMARY KEY, name TEXT NOT NULL);
INSERT INTO __yoi_schema_migrations VALUES (77, 'workspace schema baseline');
CREATE TABLE worker_workdir_links (alias TEXT, linked_at TEXT, unlinked_at TEXT);
INSERT INTO worker_workdir_links VALUES ('checkout', 'same-time', NULL), ('docs', 'same-time', 'removed');",
).unwrap();
migrate_workdir_connection_id_v77_to_v78(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 78);
let rows = || {
let mut stmt = conn.prepare("SELECT alias, linked_at, unlinked_at, connection_id FROM worker_workdir_links ORDER BY alias").unwrap();
stmt.query_map([], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, Option<String>>(2)?,
row.get::<_, String>(3)?,
))
})
.unwrap()
.collect::<std::result::Result<Vec<_>, _>>()
.unwrap()
};
let migrated = rows();
assert_eq!(migrated[0].0, "checkout");
assert_eq!(migrated[0].1, "same-time");
assert_eq!(migrated[0].2, None);
assert_eq!(migrated[1].2.as_deref(), Some("removed"));
assert!(!migrated[0].3.is_empty());
assert_ne!(migrated[0].3, migrated[1].3);
assert!(migrate_workdir_connection_id_v77_to_v78(&conn).is_err());
assert_eq!(
rows(),
migrated,
"a retry cannot replace an already durable connection ID"
);
}
#[test]
fn worker_workdir_link_capability_codec_round_trips_only_canonical_sets() {
for (capabilities, encoded) in WORKDIR_LINK_CAPABILITY_ENCODINGS {
@@ -17052,6 +17531,10 @@ mod tests {
version: 77,
name: BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME.to_string(),
},
WorkspaceSchemaMigrationStep {
version: 78,
name: WORKDIR_CONNECTION_ID_MIGRATION_NAME.to_string(),
},
]
);
@@ -17139,6 +17622,7 @@ mod tests {
77,
BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME.to_string(),
),
(78, WORKDIR_CONNECTION_ID_MIGRATION_NAME.to_string()),
]
);
assert!(!table_exists(conn, "trusted_runtime_records")?);
@@ -17490,7 +17974,7 @@ mod tests {
.collect::<Vec<_>>(),
vec![
52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72,
73, 74, 75, 76, 77
73, 74, 75, 76, 77, 78
]
);
SqliteWorkspaceStore::migrate_database(&path).unwrap();
@@ -17499,7 +17983,7 @@ mod tests {
current_schema_version(&conn).unwrap(),
LATEST_SCHEMA_VERSION
);
assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 28);
assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 29);
assert!(column_exists(&conn, "worker_workdir_links", "alias").unwrap());
assert!(column_exists(&conn, "worker_workdir_links", "capabilities").unwrap());
assert!(!column_exists(&conn, "worker_workdir_links", "role").unwrap());
@@ -20855,6 +21339,7 @@ INSERT INTO worker_registry (
store.upsert_workdir_registry(&unmanaged_workdir).unwrap();
let link = WorkerWorkdirLinkRecord {
connection_id: String::new(),
workspace_id: "local-dev".to_string(),
worker: worker.worker.clone(),
workdir_id: workdir.workdir_id.clone(),
@@ -20883,12 +21368,10 @@ INSERT INTO worker_registry (
store.attach_worker_workdir(&link),
Err(Error::WorkdirAttachmentConflict(_))
));
assert_eq!(
store
.finalize_reserved_worker_workdir_attachment(&link, "spawn-1")
.unwrap(),
link
);
let link = store
.finalize_reserved_worker_workdir_attachment(&link, "spawn-1")
.unwrap();
assert!(!link.connection_id.is_empty());
assert_eq!(store.attach_worker_workdir(&link).unwrap(), link);
let downgraded_link = WorkerWorkdirLinkRecord {
capabilities: workdir::WorkdirSessionCapabilities::READ_ONLY,
@@ -20958,10 +21441,7 @@ INSERT INTO worker_registry (
store.attach_worker_workdir(&invalid_capabilities),
Err(Error::InvalidInput(_))
));
assert_eq!(
store.attach_worker_workdir(&second_attachment).unwrap(),
second_attachment
);
let second_attachment = store.attach_worker_workdir(&second_attachment).unwrap();
assert_eq!(
store
.list_worker_workdir_links("local-dev", &worker.worker)
@@ -20991,6 +21471,157 @@ INSERT INTO worker_registry (
store.detach_worker_workdir("local-dev", &worker.worker, Some("wrong-workdir"), "6",),
Err(Error::WorkdirAttachmentConflict(_))
));
let old_connection = downgraded_link.connection_id.clone();
let removed = store
.detach_worker_workdir_connection(
"local-dev",
&worker.worker,
"checkout",
&old_connection,
"5",
)
.unwrap();
store.restore_worker_workdir_connection(&removed).unwrap();
assert_eq!(
store
.list_worker_workdir_links_page("local-dev", &worker.worker, 1, 0, None)
.unwrap()[0],
downgraded_link
);
store
.detach_worker_workdir_connection(
"local-dev",
&worker.worker,
"checkout",
&old_connection,
"5",
)
.unwrap();
// Reusing the exact record, including alias, Workdir ID and timestamp, starts a fresh lifetime.
let replacement = store.attach_worker_workdir(&downgraded_link).unwrap();
assert_ne!(replacement.connection_id, old_connection);
assert!(matches!(
store.detach_worker_workdir_connection(
"local-dev",
&worker.worker,
"checkout",
&old_connection,
"6",
),
Err(Error::WorkdirAttachmentConflict(_))
));
assert!(matches!(
store.restore_worker_workdir_connection(&removed),
Err(Error::WorkdirAttachmentConflict(_))
));
assert!(matches!(
store.replace_worker_workdir_link_capabilities(
"local-dev",
&worker.worker,
&[downgraded_link.clone(), second_attachment.clone()],
),
Err(Error::WorkdirAttachmentConflict(_))
));
assert_eq!(store.attach_worker_workdir(&link).unwrap(), replacement);
assert_eq!(
store
.list_worker_workdir_links_page("local-dev", &worker.worker, 1, 0, None)
.unwrap(),
vec![replacement.clone()]
);
assert_eq!(
store
.list_worker_workdir_links_page("local-dev", &worker.worker, 1, 1, None)
.unwrap(),
vec![second_attachment.clone()]
);
assert!(
store
.list_worker_workdir_links_page("local-dev", &worker.worker, 1, 2, None)
.unwrap()
.is_empty()
);
assert!(
store
.list_worker_workdir_links_page("foreign", &worker.worker, 1, 0, None)
.unwrap()
.is_empty()
);
assert!(
store
.list_worker_workdir_links_page(
"local-dev",
&RuntimeWorkerRef::new("foreign", worker.worker.worker_id.clone()),
1,
0,
None
)
.unwrap()
.is_empty()
);
assert_eq!(
store
.list_worker_workdir_links_page(
"local-dev",
&worker.worker,
1,
0,
Some(&second_attachment.connection_id)
)
.unwrap(),
vec![second_attachment.clone()],
"connection filtering must precede paging"
);
for (workspace_id, identity, connection_id) in [
(
"foreign",
worker.worker.clone(),
second_attachment.connection_id.as_str(),
),
(
"local-dev",
second_worker.worker.clone(),
second_attachment.connection_id.as_str(),
),
(
"local-dev",
RuntimeWorkerRef::new("foreign", worker.worker.worker_id.clone()),
second_attachment.connection_id.as_str(),
),
("local-dev", worker.worker.clone(), old_connection.as_str()),
("local-dev", worker.worker.clone(), "nonexistent"),
] {
assert!(
store
.list_worker_workdir_links_page(
workspace_id,
&identity,
1,
0,
Some(connection_id)
)
.unwrap()
.is_empty()
);
}
drop(store);
let store = SqliteWorkspaceStore::open(&db).unwrap();
assert_eq!(
store
.list_worker_workdir_links_page("local-dev", &worker.worker, 1, 0, None)
.unwrap(),
vec![replacement]
);
assert!(matches!(
store.detach_worker_workdir_connection(
"local-dev",
&worker.worker,
"checkout",
&old_connection,
"6",
),
Err(Error::WorkdirAttachmentConflict(_))
));
let detached = store
.detach_worker_workdir("local-dev", &worker.worker, Some(&workdir.workdir_id), "6")
.unwrap()
@@ -21001,10 +21632,7 @@ INSERT INTO worker_registry (
.worker_workdir_link_history_exists("local-dev", &worker.worker)
.unwrap()
);
assert_eq!(
store.attach_worker_workdir(&workdir_conflict).unwrap(),
workdir_conflict
);
let workdir_conflict = store.attach_worker_workdir(&workdir_conflict).unwrap();
drop(store);
let reopened = SqliteWorkspaceStore::open(&db).unwrap();
crates/workspace-server/src/subjektiv.rs
+310 -64
View File
@@ -11,7 +11,7 @@ use std::collections::HashSet;
use chrono::{SecondsFormat, Utc};
use memory::extract::{CandidateKind, STAGING_SCHEMA_VERSION, StagingEvidence, StagingRecord};
use memory::schema::{SourceEvidenceRef, SourceRef};
use rusqlite::{OptionalExtension, Transaction, params};
use rusqlite::{Connection, OptionalExtension, Transaction, params};
use serde::{Deserialize, Deserializer, Serialize};
use uuid::Uuid;
@@ -24,6 +24,8 @@ pub const SUBJEKTIV_SCHEMA_VERSION: u32 = 1;
pub const SUBJEKTIV_FEATURE_ID: &str = "subjektiv";
pub const SUBJECT_WORKER_SINGLETON_PREFIX: &str = "subjektiv:";
pub const MAX_STAGING_ANCHORS: usize = 10;
/// Maximum UTF-8 size of the user-managed Subject behavior document.
pub const MAX_SUBJECT_BEHAVIOR_BYTES: usize = server_api::SUBJEKTIV_MAX_BEHAVIOR_BYTES;
/// Deterministic surface-generation policy. Input estimates use the repository's
/// provider-independent UTF-8 byte estimate (`ceil(bytes / 4)`). The material
/// payload is capped below the whole-input budget to reserve room for the fixed
@@ -74,6 +76,14 @@ pub enum SubjektivError {
expected: u64,
actual: u64,
},
#[error(
"subject `{subject_id}` behavior revision conflict: expected {expected}, current {actual}"
)]
SubjectBehaviorConflict {
subject_id: String,
expected: u64,
actual: u64,
},
#[error("surface generation conflict: {0}")]
SurfaceGenerationConflict(String),
#[error("memory `{memory_id}` cannot transition from {from:?} to {to:?}")]
@@ -136,6 +146,14 @@ pub struct SubjectRecord {
pub schema_version: u32,
pub id: String,
pub role: SubjectRole,
/// User-managed, host-injected behavior document. This is not Memory and
/// Memory lifecycle operations must preserve it byte-for-byte.
#[serde(default)]
pub behavior_md: String,
/// Monotonic CAS generation for behavior updates, independent of Memory's
/// `store_revision`.
#[serde(default)]
pub behavior_revision: u64,
pub state: SubjectState,
/// Monotonic generation incremented exactly once for every committed Memory
/// creation or revision for this subject.
@@ -500,6 +518,12 @@ pub struct ResidentSurface {
pub snapshot: Option<SurfaceSnapshot>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SubjectResidentContext {
pub subject: SubjectRecord,
pub surface: ResidentSurface,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum MemoryRevisionTarget {
@@ -986,6 +1010,17 @@ END;
Ok(())
}
fn add_subject_behavior(transaction: &Transaction<'_>) -> crate::feature_storage::Result<()> {
transaction.execute_batch(
r#"
ALTER TABLE subjects ADD COLUMN behavior_md TEXT NOT NULL DEFAULT '';
ALTER TABLE subjects ADD COLUMN behavior_revision INTEGER NOT NULL DEFAULT 0
CHECK (behavior_revision >= 0);
"#,
)?;
Ok(())
}
static MIGRATIONS: &[FeatureMigration] = &[
FeatureMigration::new(1, "create subjektiv subject memory store", create_schema),
FeatureMigration::new(
@@ -1003,6 +1038,7 @@ static MIGRATIONS: &[FeatureMigration] = &[
"add bounded surface generation state and runs",
add_surface_generation_state,
),
FeatureMigration::new(5, "add user-managed subject behavior", add_subject_behavior),
];
pub const REGISTRATION: FeatureRegistration =
@@ -1070,11 +1106,23 @@ impl SubjektivStore {
/// Issues a random host-owned subject id. It is never derived from Worker,
/// Profile, Session, or Memory content.
pub fn create_subject(&self, role: SubjectRole) -> Result<SubjectRecord> {
self.create_subject_with_behavior(role, String::new())
}
/// Creates a Subject with its optional user-managed behavior document.
pub fn create_subject_with_behavior(
&self,
role: SubjectRole,
behavior_md: String,
) -> Result<SubjectRecord> {
validate_subject_behavior(&behavior_md)?;
let timestamp = now();
let record = SubjectRecord {
schema_version: SUBJEKTIV_SCHEMA_VERSION,
id: issued_id("subject"),
role,
behavior_md,
behavior_revision: 0,
state: SubjectState::Active,
store_revision: 0,
created_at: timestamp.clone(),
@@ -1084,11 +1132,13 @@ impl SubjektivStore {
self.database.try_transaction(|transaction| {
transaction.execute(
"INSERT INTO subjects (
subject_id, role, state, store_revision, record_json, created_at, updated_at
) VALUES (?1, ?2, 'active', 0, ?3, ?4, ?5)",
subject_id, role, behavior_md, behavior_revision, state,
store_revision, record_json, created_at, updated_at
) VALUES (?1, ?2, ?3, 0, 'active', 0, ?4, ?5, ?6)",
params![
record.id,
record.role.as_str(),
record.behavior_md,
raw,
record.created_at,
record.updated_at
@@ -1111,6 +1161,51 @@ impl SubjektivStore {
})
}
/// Replaces or clears the user-managed behavior document using its own CAS
/// generation. Memory revisions do not participate in this generation.
pub fn update_subject_behavior(
&self,
subject_id: &str,
expected_behavior_revision: u64,
behavior_md: String,
) -> Result<SubjectRecord> {
validate_subject_behavior(&behavior_md)?;
self.database.try_transaction(|transaction| {
let mut subject = require_subject(transaction, subject_id)?;
if subject.behavior_revision != expected_behavior_revision {
return Err(SubjektivError::SubjectBehaviorConflict {
subject_id: subject_id.to_string(),
expected: expected_behavior_revision,
actual: subject.behavior_revision,
});
}
if subject.behavior_md == behavior_md {
return Ok(subject);
}
subject.behavior_md = behavior_md;
subject.behavior_revision =
subject.behavior_revision.checked_add(1).ok_or_else(|| {
SubjektivError::InvalidRecord("subject behavior revision overflow".to_string())
})?;
subject.updated_at = now();
let raw = serde_json::to_string(&subject)?;
transaction.execute(
"UPDATE subjects
SET behavior_md = ?2, behavior_revision = ?3,
record_json = ?4, updated_at = ?5
WHERE subject_id = ?1",
params![
subject.id,
subject.behavior_md,
to_i64(subject.behavior_revision)?,
raw,
subject.updated_at
],
)?;
Ok(subject)
})
}
/// Validates an explicit product launch connection against this Workspace's
/// subject store and the already-resolved Profile policy. Subject knowledge
/// alone is insufficient: only an active local record produces host launch material.
@@ -2021,60 +2116,17 @@ impl SubjektivStore {
pub fn resident_surface(&self, subject_id: &str) -> Result<ResidentSurface> {
self.database.try_with_connection(|connection| {
let subject = require_subject_in_connection(connection, subject_id)?;
let state = connection
.query_row(
"SELECT store_revision, status, snapshot_id
FROM surface_generation_state WHERE subject_id = ?1",
[subject_id],
|row| {
Ok((
row.get::<_, i64>(0)?,
row.get::<_, String>(1)?,
row.get::<_, Option<String>>(2)?,
))
},
)
.optional()?;
let Some((revision, status, snapshot_id)) = state else {
return Ok(ResidentSurface {
availability: SurfaceAvailability::Ungenerated,
snapshot: None,
});
};
let revision = u64::try_from(revision).map_err(|_| {
SubjektivError::InvalidRecord("surface state revision is negative".into())
})?;
if revision != subject.store_revision || status == "dirty" {
return Ok(ResidentSurface {
availability: SurfaceAvailability::Stale,
snapshot: None,
});
}
if status == "failed" {
return Ok(ResidentSurface {
availability: SurfaceAvailability::Failed,
snapshot: None,
});
}
let snapshot_id = snapshot_id.ok_or_else(|| {
SubjektivError::InvalidRecord("ready surface state has no snapshot id".into())
})?;
let raw = connection.query_row(
"SELECT snapshot_json FROM surface_snapshots
WHERE subject_id = ?1 AND snapshot_id = ?2",
params![subject_id, snapshot_id],
|row| row.get::<_, String>(0),
)?;
let snapshot = parse_surface_snapshot(&raw)?;
if snapshot.built_from_store_revision != subject.store_revision {
return Err(SubjektivError::InvalidRecord(
"ready surface snapshot generation does not match subject".into(),
));
}
Ok(ResidentSurface {
availability: SurfaceAvailability::Ready,
snapshot: Some(snapshot),
})
resident_surface_in_connection(connection, &subject)
})
}
/// Atomically reads user-managed behavior and the independently-fresh
/// generated Memory surface from one SQLite connection snapshot.
pub fn resident_context(&self, subject_id: &str) -> Result<SubjectResidentContext> {
self.database.try_transaction(|transaction| {
let subject = require_subject(transaction, subject_id)?;
let surface = resident_surface_in_connection(transaction, &subject)?;
Ok(SubjectResidentContext { subject, surface })
})
}
@@ -3414,6 +3466,87 @@ fn validate_range(kind: &str, range: [u64; 2]) -> Result<()> {
}
}
fn resident_surface_in_connection(
connection: &Connection,
subject: &SubjectRecord,
) -> Result<ResidentSurface> {
let state = connection
.query_row(
"SELECT store_revision, status, snapshot_id
FROM surface_generation_state WHERE subject_id = ?1",
[&subject.id],
|row| {
Ok((
row.get::<_, i64>(0)?,
row.get::<_, String>(1)?,
row.get::<_, Option<String>>(2)?,
))
},
)
.optional()?;
let Some((revision, status, snapshot_id)) = state else {
return Ok(ResidentSurface {
availability: SurfaceAvailability::Ungenerated,
snapshot: None,
});
};
let revision = u64::try_from(revision)
.map_err(|_| SubjektivError::InvalidRecord("surface state revision is negative".into()))?;
if revision != subject.store_revision || status == "dirty" {
return Ok(ResidentSurface {
availability: SurfaceAvailability::Stale,
snapshot: None,
});
}
if status == "failed" {
return Ok(ResidentSurface {
availability: SurfaceAvailability::Failed,
snapshot: None,
});
}
let snapshot_id = snapshot_id.ok_or_else(|| {
SubjektivError::InvalidRecord("ready surface state has no snapshot id".into())
})?;
let raw = connection.query_row(
"SELECT snapshot_json FROM surface_snapshots
WHERE subject_id = ?1 AND snapshot_id = ?2",
params![subject.id, snapshot_id],
|row| row.get::<_, String>(0),
)?;
let snapshot = parse_surface_snapshot(&raw)?;
if snapshot.built_from_store_revision != subject.store_revision {
return Err(SubjektivError::InvalidRecord(
"ready surface snapshot generation does not match subject".into(),
));
}
Ok(ResidentSurface {
availability: SurfaceAvailability::Ready,
snapshot: Some(snapshot),
})
}
pub fn validate_subject_behavior(value: &str) -> Result<()> {
if value.len() > MAX_SUBJECT_BEHAVIOR_BYTES {
return Err(SubjektivError::InvalidRecord(format!(
"subject behavior must be at most {MAX_SUBJECT_BEHAVIOR_BYTES} bytes"
)));
}
if !value.is_empty() && value.trim().is_empty() {
return Err(SubjektivError::InvalidRecord(
"subject behavior must be empty or contain non-whitespace text".to_string(),
));
}
if value
.chars()
.any(|character| character.is_control() && !matches!(character, '\n' | '\r' | '\t'))
{
return Err(SubjektivError::InvalidRecord(
"subject behavior contains an unsupported control character".to_string(),
));
}
Ok(())
}
fn validate_label(kind: &str, value: &str) -> Result<()> {
validate_nonempty(kind, value)?;
if value.len() > 256 || value.chars().any(char::is_control) {
@@ -3668,6 +3801,67 @@ mod tests {
assert!(!complete.has_more);
}
#[test]
fn subject_behavior_is_bounded_revisioned_and_independent_from_memory_revision() {
let temp = tempfile::tempdir().unwrap();
let (_manager, _workspace, store) = open_store(temp.path(), "workspace-a");
let subject = store
.create_subject_with_behavior(role(), "Prefer explicit evidence.".to_string())
.unwrap();
assert_eq!(subject.behavior_revision, 0);
assert_eq!(subject.store_revision, 0);
let updated = store
.update_subject_behavior(
&subject.id,
0,
"Prefer explicit evidence.\nAsk when uncertain.".to_string(),
)
.unwrap();
assert_eq!(updated.behavior_revision, 1);
assert_eq!(updated.store_revision, 0);
store
.create_memory(
&subject.id,
draft(
"A confirmed observation",
"exercise Memory lifecycle isolation",
),
)
.unwrap();
let after_memory = store.subject(&subject.id).unwrap().unwrap();
assert_eq!(after_memory.behavior_md, updated.behavior_md);
assert_eq!(after_memory.behavior_revision, 1);
assert_eq!(after_memory.store_revision, 1);
let (_reopened_manager, _reopened_workspace, reopened) =
open_store(temp.path(), "workspace-a");
let persisted = reopened.subject(&subject.id).unwrap().unwrap();
assert_eq!(persisted.behavior_md, updated.behavior_md);
assert_eq!(persisted.behavior_revision, 1);
assert!(matches!(
store.update_subject_behavior(&subject.id, 0, String::new()),
Err(SubjektivError::SubjectBehaviorConflict {
expected: 0,
actual: 1,
..
})
));
let unchanged = store
.update_subject_behavior(&subject.id, 1, updated.behavior_md.clone())
.unwrap();
assert_eq!(unchanged.behavior_revision, 1);
let cleared = store
.update_subject_behavior(&subject.id, 1, String::new())
.unwrap();
assert_eq!(cleared.behavior_revision, 2);
assert!(cleared.behavior_md.is_empty());
assert!(validate_subject_behavior(&"x".repeat(MAX_SUBJECT_BEHAVIOR_BYTES + 1)).is_err());
assert!(validate_subject_behavior(" ").is_err());
}
#[test]
fn attributed_candidates_preserve_revision_and_surface_reference_continuity() {
let temp = tempfile::tempdir().unwrap();
@@ -3892,10 +4086,38 @@ mod tests {
))
.unwrap();
let store = SubjektivStore::open(&workspace, &registration).unwrap();
store.create_subject(role()).unwrap().id
let timestamp = now();
let subject_id = issued_id("subject");
let mut raw = serde_json::json!({
"schema_version": SUBJEKTIV_SCHEMA_VERSION,
"id": subject_id,
"role": role(),
"state": SubjectState::Active,
"store_revision": 0,
"created_at": timestamp.clone(),
"updated_at": timestamp.clone(),
});
let raw = serde_json::to_string(&raw.take()).unwrap();
store
.database
.try_transaction(|transaction| -> Result<()> {
transaction.execute(
"INSERT INTO subjects (
subject_id, role, state, store_revision,
record_json, created_at, updated_at
) VALUES (?1, ?2, 'active', 0, ?3, ?4, ?5)",
params![subject_id, role().as_str(), raw, timestamp, timestamp],
)?;
Ok(())
})
.unwrap();
subject_id
};
let (_manager, _workspace, store) = open_store(&root, "workspace-a");
let migrated_subject = store.subject(&subject_id).unwrap().unwrap();
assert!(migrated_subject.behavior_md.is_empty());
assert_eq!(migrated_subject.behavior_revision, 0);
let lifecycle_attribution = attribution(
&subject_id,
"runtime-1",
@@ -3935,26 +4157,50 @@ mod tests {
))
.unwrap();
let store = SubjektivStore::open(&workspace, &registration).unwrap();
let subject = store.create_subject(role()).unwrap();
let subject_id = issued_id("subject");
let timestamp = now();
let subject_json = serde_json::to_string(&serde_json::json!({
"schema_version": SUBJEKTIV_SCHEMA_VERSION,
"id": subject_id,
"role": role(),
"state": SubjectState::Active,
"store_revision": 0,
"created_at": timestamp,
"updated_at": timestamp,
}))
.unwrap();
let snapshot = SurfaceSnapshot {
schema_version: SUBJEKTIV_SCHEMA_VERSION,
id: "legacy-surface".into(),
subject_id: subject.id.clone(),
subject_id: subject_id.clone(),
body_md: "Legacy summary without generation-policy evidence.".into(),
memory_refs: Vec::new(),
built_from_store_revision: subject.store_revision,
built_from_store_revision: 0,
created_at: now(),
};
store
.database
.try_transaction(|transaction| -> Result<()> {
transaction.execute(
"INSERT INTO subjects (
subject_id, role, state, store_revision,
record_json, created_at, updated_at
) VALUES (?1, ?2, 'active', 0, ?3, ?4, ?5)",
params![
subject_id,
role().as_str(),
subject_json,
timestamp,
timestamp
],
)?;
transaction.execute(
"INSERT INTO surface_snapshots (
subject_id, snapshot_id, built_from_store_revision,
snapshot_json, created_at
) VALUES (?1, ?2, ?3, ?4, ?5)",
params![
subject.id,
subject_id,
snapshot.id,
to_i64(snapshot.built_from_store_revision)?,
serde_json::to_string(&snapshot)?,
@@ -3964,12 +4210,12 @@ mod tests {
transaction.execute(
"INSERT INTO surface_snapshot_seals (subject_id, snapshot_id)
VALUES (?1, ?2)",
params![subject.id, snapshot.id],
params![subject_id, snapshot.id],
)?;
Ok(())
})
.unwrap();
(subject.id, snapshot.id)
(subject_id, snapshot.id)
};
let (_manager, _workspace, store) = open_store(&root, "workspace-a");