diff --git a/crates/workspace-server/src/authority.rs b/crates/workspace-server/src/authority.rs index 6483cdb7..5b5dba0e 100644 --- a/crates/workspace-server/src/authority.rs +++ b/crates/workspace-server/src/authority.rs @@ -1,5 +1,7 @@ use std::path::PathBuf; +use chrono::Utc; + use ticket::{ SqliteTicketBackend, TicketBackend, TicketIdOrSlug, TicketListQuery, TicketWorkspaceActionPriority, project_ticket_workspace_item, @@ -9,22 +11,24 @@ use crate::records::{ ObjectiveDetail, ObjectiveResourceSummary, ObjectiveSummary, ProjectRecordList, TicketDetail, TicketSummary, summarize_body, truncate_body, validate_project_id, }; -use crate::store::{ControlPlaneStore, SqliteWorkspaceStore}; +use crate::store::{ + ControlPlaneStore, MemoryDocumentRecord, MemoryStagingRecord, MemoryStagingResolutionRecord, + SqliteWorkspaceStore, +}; use crate::{Error, Result}; const DETAIL_BODY_LIMIT: usize = 64 * 1024; +const DEFAULT_MEMORY_DOCUMENT_BODY: &str = "# Memory\n\n"; +const RECORD_SOURCE_WORKSPACE_SQLITE: &str = "workspace-sqlite"; /// Workspace-scoped runtime authority for project resources. /// /// Normal Backend API handlers should depend on this authority abstraction, not /// on legacy filesystem layouts. Filesystem readers belong in explicitly /// temporary migration/repair tools and tests, not normal runtime authority paths. -pub trait WorkspaceAuthority: ObjectiveAuthority + TicketAuthority + MemoryAuthorityMarker {} +pub trait WorkspaceAuthority: ObjectiveAuthority + TicketAuthority + MemoryAuthority {} -impl WorkspaceAuthority for T where - T: ObjectiveAuthority + TicketAuthority + MemoryAuthorityMarker -{ -} +impl WorkspaceAuthority for T where T: ObjectiveAuthority + TicketAuthority + MemoryAuthority {} pub trait TicketAuthority { fn list_tickets(&self, limit: usize) -> Result>; @@ -36,19 +40,57 @@ pub trait ObjectiveAuthority { fn objective(&self, id: &str) -> Result; } -/// Marker for the pending Memory authority split. -/// -/// Memory still routes through the legacy `memory` crate backend in this codebase. -/// This marker makes the missing SQL-backed Memory authority explicit so callers -/// do not mistake the current Backend API boundary for a completed storage -/// authority migration. -pub trait MemoryAuthorityMarker { - fn memory_authority_status(&self) -> MemoryAuthorityStatus; +pub trait MemoryAuthority { + fn ensure_memory_document(&self) -> Result; + fn memory_document(&self) -> Result; + fn update_memory_document(&self, body_md: &str) -> Result; + fn list_memory_staging_records(&self, limit: usize) -> Result>; + fn memory_staging_record(&self, candidate_id: &str) -> Result; + fn upsert_memory_staging_record( + &self, + candidate_id: &str, + raw_json: &str, + source_path: Option<&str>, + ) -> Result; + fn close_memory_staging_record( + &self, + candidate_id: &str, + action: &str, + reason: &str, + affected_refs_json: &str, + ) -> Result; + fn list_memory_staging_resolutions(&self, limit: usize) + -> Result>; } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum MemoryAuthorityStatus { - LegacyFilesystemBackendPendingSqliteAuthority, +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct MemoryDocument { + pub body_md: String, + pub created_at: String, + pub updated_at: String, + pub record_source: String, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct MemoryStagingEntry { + pub candidate_id: String, + pub raw_json: String, + pub source_path: Option, + pub imported_at: String, + pub record_source: String, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct MemoryStagingResolution { + pub candidate_id: String, + pub action: String, + pub reason: String, + pub affected_refs_json: String, + pub staging_raw_json: String, + pub source_path: Option, + pub imported_at: String, + pub resolved_at: String, + pub record_source: String, } #[derive(Clone)] @@ -196,9 +238,192 @@ impl ObjectiveAuthority for SqliteWorkspaceAuthority { } } -impl MemoryAuthorityMarker for SqliteWorkspaceAuthority { - fn memory_authority_status(&self) -> MemoryAuthorityStatus { - MemoryAuthorityStatus::LegacyFilesystemBackendPendingSqliteAuthority +impl MemoryAuthority for SqliteWorkspaceAuthority { + fn ensure_memory_document(&self) -> Result { + let now = now_rfc3339(); + self.store + .ensure_memory_document(&self.workspace_id, DEFAULT_MEMORY_DOCUMENT_BODY, &now) + .map(memory_document_from_record) + } + + fn memory_document(&self) -> Result { + self.store + .get_memory_document(&self.workspace_id)? + .map(memory_document_from_record) + .ok_or_else(|| Error::Store("workspace memory document is not initialized".to_string())) + } + + fn update_memory_document(&self, body_md: &str) -> Result { + let existing = self.ensure_memory_document()?; + let updated_at = now_rfc3339(); + let record = MemoryDocumentRecord { + workspace_id: self.workspace_id.clone(), + body_md: body_md.to_string(), + created_at: existing.created_at, + updated_at, + }; + self.store.upsert_memory_document(&record)?; + Ok(memory_document_from_record(record)) + } + + fn list_memory_staging_records(&self, limit: usize) -> Result> { + self.store + .list_memory_staging_records(&self.workspace_id, limit) + .map(|records| { + records + .into_iter() + .map(memory_staging_from_record) + .collect() + }) + } + + fn memory_staging_record(&self, candidate_id: &str) -> Result { + validate_memory_candidate_id(candidate_id)?; + self.store + .get_memory_staging_record(&self.workspace_id, candidate_id)? + .map(memory_staging_from_record) + .ok_or_else(|| { + Error::Store(format!("unknown memory staging candidate '{candidate_id}'")) + }) + } + + fn upsert_memory_staging_record( + &self, + candidate_id: &str, + raw_json: &str, + source_path: Option<&str>, + ) -> Result { + validate_memory_candidate_id(candidate_id)?; + validate_json_object(raw_json, "raw_json")?; + let imported_at = now_rfc3339(); + let record = MemoryStagingRecord { + workspace_id: self.workspace_id.clone(), + candidate_id: candidate_id.to_string(), + raw_json: raw_json.to_string(), + source_path: source_path.map(str::to_string), + imported_at, + }; + self.store.upsert_memory_staging_record(&record)?; + Ok(memory_staging_from_record(record)) + } + + fn close_memory_staging_record( + &self, + candidate_id: &str, + action: &str, + reason: &str, + affected_refs_json: &str, + ) -> Result { + validate_memory_candidate_id(candidate_id)?; + validate_non_empty(action, "action")?; + validate_non_empty(reason, "reason")?; + validate_json_array(affected_refs_json, "affected_refs_json")?; + let staging = self + .store + .get_memory_staging_record(&self.workspace_id, candidate_id)? + .ok_or_else(|| { + Error::Store(format!("unknown memory staging candidate '{candidate_id}'")) + })?; + let record = MemoryStagingResolutionRecord { + workspace_id: self.workspace_id.clone(), + candidate_id: staging.candidate_id, + action: action.trim().to_string(), + reason: reason.trim().to_string(), + affected_refs_json: affected_refs_json.to_string(), + staging_raw_json: staging.raw_json, + source_path: staging.source_path, + imported_at: staging.imported_at, + resolved_at: now_rfc3339(), + }; + self.store.insert_memory_staging_resolution(&record)?; + self.store + .delete_memory_staging_record(&self.workspace_id, candidate_id)?; + Ok(memory_resolution_from_record(record)) + } + + fn list_memory_staging_resolutions( + &self, + limit: usize, + ) -> Result> { + self.store + .list_memory_staging_resolutions(&self.workspace_id, limit) + .map(|records| { + records + .into_iter() + .map(memory_resolution_from_record) + .collect() + }) + } +} + +fn now_rfc3339() -> String { + Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true) +} + +fn validate_memory_candidate_id(candidate_id: &str) -> Result<()> { + validate_non_empty(candidate_id, "candidate_id") +} + +fn validate_non_empty(value: &str, field: &str) -> Result<()> { + if value.trim().is_empty() { + Err(Error::Store(format!("memory {field} must not be empty"))) + } else { + Ok(()) + } +} + +fn validate_json_object(raw_json: &str, field: &str) -> Result<()> { + let value: serde_json::Value = serde_json::from_str(raw_json) + .map_err(|err| Error::Store(format!("memory {field} must be valid JSON: {err}")))?; + if value.is_object() { + Ok(()) + } else { + Err(Error::Store(format!( + "memory {field} must be a JSON object" + ))) + } +} + +fn validate_json_array(raw_json: &str, field: &str) -> Result<()> { + let value: serde_json::Value = serde_json::from_str(raw_json) + .map_err(|err| Error::Store(format!("memory {field} must be valid JSON: {err}")))?; + if value.is_array() { + Ok(()) + } else { + Err(Error::Store(format!("memory {field} must be a JSON array"))) + } +} + +fn memory_document_from_record(record: MemoryDocumentRecord) -> MemoryDocument { + MemoryDocument { + body_md: record.body_md, + created_at: record.created_at, + updated_at: record.updated_at, + record_source: RECORD_SOURCE_WORKSPACE_SQLITE.to_string(), + } +} + +fn memory_staging_from_record(record: MemoryStagingRecord) -> MemoryStagingEntry { + MemoryStagingEntry { + candidate_id: record.candidate_id, + raw_json: record.raw_json, + source_path: record.source_path, + imported_at: record.imported_at, + record_source: RECORD_SOURCE_WORKSPACE_SQLITE.to_string(), + } +} + +fn memory_resolution_from_record(record: MemoryStagingResolutionRecord) -> MemoryStagingResolution { + MemoryStagingResolution { + candidate_id: record.candidate_id, + action: record.action, + reason: record.reason, + affected_refs_json: record.affected_refs_json, + staging_raw_json: record.staging_raw_json, + source_path: record.source_path, + imported_at: record.imported_at, + resolved_at: record.resolved_at, + record_source: RECORD_SOURCE_WORKSPACE_SQLITE.to_string(), } } @@ -293,9 +518,36 @@ mod tests { let objective = authority.objective("00000000001J3").unwrap(); assert!(objective.body.contains("Objective body")); + let memory = authority.ensure_memory_document().unwrap(); + assert_eq!(memory.body_md, DEFAULT_MEMORY_DOCUMENT_BODY); + let updated = authority + .update_memory_document("# Memory\n\n- Durable fact.\n") + .unwrap(); + assert!(updated.body_md.contains("Durable fact")); + + let staging = authority + .upsert_memory_staging_record( + "candidate-1", + r#"{"kind":"summary","body_md":"Candidate"}"#, + Some("memory/_staging/candidate-1.json"), + ) + .unwrap(); + assert_eq!(staging.record_source, "workspace-sqlite"); + assert_eq!(authority.list_memory_staging_records(20).unwrap().len(), 1); + let resolution = authority + .close_memory_staging_record("candidate-1", "apply", "accepted", r#"["memory"]"#) + .unwrap(); + assert_eq!(resolution.action, "apply"); + assert_eq!(resolution.reason, "accepted"); + assert!( + authority + .list_memory_staging_records(20) + .unwrap() + .is_empty() + ); assert_eq!( - authority.memory_authority_status(), - MemoryAuthorityStatus::LegacyFilesystemBackendPendingSqliteAuthority + authority.list_memory_staging_resolutions(20).unwrap().len(), + 1 ); } diff --git a/crates/workspace-server/src/lib.rs b/crates/workspace-server/src/lib.rs index 95985766..2bdaeded 100644 --- a/crates/workspace-server/src/lib.rs +++ b/crates/workspace-server/src/lib.rs @@ -21,8 +21,8 @@ pub mod skills; pub mod store; pub use authority::{ - MemoryAuthorityMarker, MemoryAuthorityStatus, ObjectiveAuthority, SqliteWorkspaceAuthority, - TicketAuthority, WorkspaceAuthority, + MemoryAuthority, MemoryDocument, MemoryStagingEntry, MemoryStagingResolution, + ObjectiveAuthority, SqliteWorkspaceAuthority, TicketAuthority, WorkspaceAuthority, }; pub use config::{ BackendRuntimesConfigFile, ConfigDiff, ResolvedWorkspaceBackendConfig, diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index 2a464eb6..a62ac803 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -67,6 +67,11 @@ const MIGRATIONS: &[Migration] = &[ name: "sqlite objective authority and memory staging import", apply: create_objective_sqlite_authority_tables, }, + Migration { + version: 11, + name: "sqlite memory authority documents and staging resolutions", + apply: create_memory_authority_tables, + }, ]; struct Migration { @@ -257,6 +262,14 @@ pub struct ObjectiveResourceRecord { pub updated_at: String, } +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct MemoryDocumentRecord { + pub workspace_id: String, + pub body_md: String, + pub created_at: String, + pub updated_at: String, +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct MemoryStagingRecord { pub workspace_id: String, @@ -266,6 +279,19 @@ pub struct MemoryStagingRecord { pub imported_at: String, } +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct MemoryStagingResolutionRecord { + pub workspace_id: String, + pub candidate_id: String, + pub action: String, + pub reason: String, + pub affected_refs_json: String, + pub staging_raw_json: String, + pub source_path: Option, + pub imported_at: String, + pub resolved_at: String, +} + #[async_trait] pub trait ControlPlaneStore: Send + Sync { async fn schema_version(&self) -> Result; @@ -299,8 +325,36 @@ pub trait ControlPlaneStore: Send + Sync { workspace_id: &str, objective_id: &str, ) -> Result>; + fn ensure_memory_document( + &self, + workspace_id: &str, + default_body_md: &str, + now: &str, + ) -> Result; + fn get_memory_document(&self, workspace_id: &str) -> Result>; + fn upsert_memory_document(&self, record: &MemoryDocumentRecord) -> Result<()>; fn upsert_memory_staging_record(&self, record: &MemoryStagingRecord) -> Result<()>; + fn list_memory_staging_records( + &self, + workspace_id: &str, + limit: usize, + ) -> Result>; + fn get_memory_staging_record( + &self, + workspace_id: &str, + candidate_id: &str, + ) -> Result>; + fn delete_memory_staging_record(&self, workspace_id: &str, candidate_id: &str) -> Result; fn count_memory_staging_records(&self, workspace_id: &str) -> Result; + fn insert_memory_staging_resolution( + &self, + record: &MemoryStagingResolutionRecord, + ) -> Result<()>; + fn list_memory_staging_resolutions( + &self, + workspace_id: &str, + limit: usize, + ) -> Result>; fn upsert_account(&self, record: &AccountRecord) -> Result<()>; fn get_account(&self, account_id: &str) -> Result>; @@ -712,6 +766,64 @@ impl ControlPlaneStore for SqliteWorkspaceStore { }) } + fn ensure_memory_document( + &self, + workspace_id: &str, + default_body_md: &str, + now: &str, + ) -> Result { + self.with_conn(|conn| { + conn.execute( + r#"INSERT OR IGNORE INTO workspace_memory_documents ( + workspace_id, body_md, created_at, updated_at + ) VALUES (?1, ?2, ?3, ?3)"#, + params![workspace_id, default_body_md, now], + )?; + conn.query_row( + r#"SELECT workspace_id, body_md, created_at, updated_at + FROM workspace_memory_documents + WHERE workspace_id = ?1"#, + params![workspace_id], + read_memory_document_record, + ) + .map_err(Error::from) + }) + } + + fn get_memory_document(&self, workspace_id: &str) -> Result> { + self.with_conn(|conn| { + conn.query_row( + r#"SELECT workspace_id, body_md, created_at, updated_at + FROM workspace_memory_documents + WHERE workspace_id = ?1"#, + params![workspace_id], + read_memory_document_record, + ) + .optional() + .map_err(Error::from) + }) + } + + fn upsert_memory_document(&self, record: &MemoryDocumentRecord) -> Result<()> { + self.with_conn(|conn| { + conn.execute( + r#"INSERT INTO workspace_memory_documents ( + workspace_id, body_md, created_at, updated_at + ) VALUES (?1, ?2, ?3, ?4) + ON CONFLICT(workspace_id) DO UPDATE SET + body_md = excluded.body_md, + updated_at = excluded.updated_at"#, + params![ + record.workspace_id, + record.body_md, + record.created_at, + record.updated_at, + ], + )?; + Ok(()) + }) + } + fn upsert_memory_staging_record(&self, record: &MemoryStagingRecord) -> Result<()> { self.with_conn(|conn| { conn.execute( @@ -734,6 +846,56 @@ impl ControlPlaneStore for SqliteWorkspaceStore { }) } + fn list_memory_staging_records( + &self, + workspace_id: &str, + limit: usize, + ) -> Result> { + self.with_conn(|conn| { + let mut stmt = conn.prepare( + r#"SELECT workspace_id, candidate_id, raw_json, source_path, imported_at + FROM memory_staging_records + WHERE workspace_id = ?1 + ORDER BY imported_at DESC, candidate_id ASC + LIMIT ?2"#, + )?; + let rows = stmt.query_map( + params![workspace_id, limit as i64], + read_memory_staging_record, + )?; + rows.collect::, _>>() + .map_err(Error::from) + }) + } + + fn get_memory_staging_record( + &self, + workspace_id: &str, + candidate_id: &str, + ) -> Result> { + self.with_conn(|conn| { + conn.query_row( + r#"SELECT workspace_id, candidate_id, raw_json, source_path, imported_at + FROM memory_staging_records + WHERE workspace_id = ?1 AND candidate_id = ?2"#, + params![workspace_id, candidate_id], + read_memory_staging_record, + ) + .optional() + .map_err(Error::from) + }) + } + + fn delete_memory_staging_record(&self, workspace_id: &str, candidate_id: &str) -> Result { + self.with_conn(|conn| { + let changed = conn.execute( + "DELETE FROM memory_staging_records WHERE workspace_id = ?1 AND candidate_id = ?2", + params![workspace_id, candidate_id], + )?; + Ok(changed > 0) + }) + } + fn count_memory_staging_records(&self, workspace_id: &str) -> Result { self.with_conn(|conn| { conn.query_row( @@ -746,6 +908,55 @@ impl ControlPlaneStore for SqliteWorkspaceStore { }) } + fn insert_memory_staging_resolution( + &self, + record: &MemoryStagingResolutionRecord, + ) -> Result<()> { + self.with_conn(|conn| { + conn.execute( + r#"INSERT INTO memory_staging_resolutions ( + workspace_id, candidate_id, action, reason, affected_refs_json, + staging_raw_json, source_path, imported_at, resolved_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)"#, + params![ + record.workspace_id, + record.candidate_id, + record.action, + record.reason, + record.affected_refs_json, + record.staging_raw_json, + record.source_path, + record.imported_at, + record.resolved_at, + ], + )?; + Ok(()) + }) + } + + fn list_memory_staging_resolutions( + &self, + workspace_id: &str, + limit: usize, + ) -> Result> { + self.with_conn(|conn| { + let mut stmt = conn.prepare( + r#"SELECT workspace_id, candidate_id, action, reason, affected_refs_json, + staging_raw_json, source_path, imported_at, resolved_at + FROM memory_staging_resolutions + WHERE workspace_id = ?1 + ORDER BY resolved_at DESC, candidate_id ASC + LIMIT ?2"#, + )?; + let rows = stmt.query_map( + params![workspace_id, limit as i64], + read_memory_staging_resolution_record, + )?; + rows.collect::, _>>() + .map_err(Error::from) + }) + } + fn upsert_account(&self, record: &AccountRecord) -> Result<()> { self.with_conn(|conn| { conn.execute( @@ -1569,6 +1780,41 @@ fn read_objective_resource_record( }) } +fn read_memory_document_record(row: &rusqlite::Row<'_>) -> rusqlite::Result { + Ok(MemoryDocumentRecord { + workspace_id: row.get(0)?, + body_md: row.get(1)?, + created_at: row.get(2)?, + updated_at: row.get(3)?, + }) +} + +fn read_memory_staging_record(row: &rusqlite::Row<'_>) -> rusqlite::Result { + Ok(MemoryStagingRecord { + workspace_id: row.get(0)?, + candidate_id: row.get(1)?, + raw_json: row.get(2)?, + source_path: row.get(3)?, + imported_at: row.get(4)?, + }) +} + +fn read_memory_staging_resolution_record( + row: &rusqlite::Row<'_>, +) -> rusqlite::Result { + Ok(MemoryStagingResolutionRecord { + workspace_id: row.get(0)?, + candidate_id: row.get(1)?, + action: row.get(2)?, + reason: row.get(3)?, + affected_refs_json: row.get(4)?, + staging_raw_json: row.get(5)?, + source_path: row.get(6)?, + imported_at: row.get(7)?, + resolved_at: row.get(8)?, + }) +} + fn worker_registry_select_sql(where_clause: &str) -> String { format!( "SELECT workspace_id, runtime_id, runtime_worker_id, display_name, profile, \ @@ -1732,6 +1978,33 @@ CREATE TABLE IF NOT EXISTS memory_staging_records ( Ok(()) } +fn create_memory_authority_tables(conn: &Connection) -> Result<()> { + conn.execute_batch( + r#" +CREATE TABLE IF NOT EXISTS workspace_memory_documents ( + workspace_id TEXT PRIMARY KEY REFERENCES workspaces(workspace_id) ON DELETE CASCADE, + body_md TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS memory_staging_resolutions ( + workspace_id TEXT NOT NULL REFERENCES workspaces(workspace_id) ON DELETE CASCADE, + candidate_id TEXT NOT NULL, + action TEXT NOT NULL, + reason TEXT NOT NULL, + affected_refs_json TEXT NOT NULL, + staging_raw_json TEXT NOT NULL, + source_path TEXT, + imported_at TEXT NOT NULL, + resolved_at TEXT NOT NULL, + PRIMARY KEY (workspace_id, candidate_id, resolved_at) +); +"#, + )?; + Ok(()) +} + fn create_account_identity_tables(conn: &Connection) -> Result<()> { conn.execute_batch( r#" @@ -2293,6 +2566,26 @@ CREATE TABLE IF NOT EXISTS memory_staging_records ( PRIMARY KEY (workspace_id, candidate_id) ); +CREATE TABLE IF NOT EXISTS workspace_memory_documents ( + workspace_id TEXT PRIMARY KEY REFERENCES workspaces(workspace_id) ON DELETE CASCADE, + body_md TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS memory_staging_resolutions ( + workspace_id TEXT NOT NULL REFERENCES workspaces(workspace_id) ON DELETE CASCADE, + candidate_id TEXT NOT NULL, + action TEXT NOT NULL, + reason TEXT NOT NULL, + affected_refs_json TEXT NOT NULL, + staging_raw_json TEXT NOT NULL, + source_path TEXT, + imported_at TEXT NOT NULL, + resolved_at TEXT NOT NULL, + PRIMARY KEY (workspace_id, candidate_id, resolved_at) +); + CREATE TABLE IF NOT EXISTS repositories ( workspace_id TEXT NOT NULL REFERENCES workspaces(workspace_id) ON DELETE CASCADE, repository_id TEXT PRIMARY KEY, @@ -2424,7 +2717,7 @@ mod tests { let db = dir.path().join("control-plane.sqlite"); let store = SqliteWorkspaceStore::open(&db).unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 10); + assert_eq!(store.schema_version().await.unwrap(), 11); let record = WorkspaceRecord { workspace_id: "local-dev".to_string(), @@ -2437,7 +2730,7 @@ mod tests { store.upsert_workspace(&record).await.unwrap(); let reopened = SqliteWorkspaceStore::open(&db).unwrap(); - assert_eq!(reopened.schema_version().await.unwrap(), 10); + assert_eq!(reopened.schema_version().await.unwrap(), 11); assert_eq!( reopened.get_workspace("local-dev").await.unwrap(), Some(record) @@ -2458,6 +2751,10 @@ mod tests { "ticket_relations", "objectives", "objective_ticket_links", + "objective_resources", + "memory_staging_records", + "workspace_memory_documents", + "memory_staging_resolutions", "repositories", "ticket_targets", "ticket_target_paths", @@ -2654,7 +2951,7 @@ mod tests { .unwrap(); let store = SqliteWorkspaceStore::from_connection(conn).unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 10); + assert_eq!(store.schema_version().await.unwrap(), 11); store .with_conn(|conn| { @@ -2667,6 +2964,9 @@ mod tests { "ticket_worker_links", "artifacts", "audit_events", + "workspace_memory_documents", + "memory_staging_records", + "memory_staging_resolutions", "legacy_workspaces", "legacy_repositories", "legacy_runs", @@ -2745,7 +3045,7 @@ mod tests { #[tokio::test] async fn repository_records_round_trip() { let store = SqliteWorkspaceStore::in_memory().unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 10); + assert_eq!(store.schema_version().await.unwrap(), 11); let workspace = WorkspaceRecord { workspace_id: "local-dev".to_string(), owner_account_id: None, @@ -2780,6 +3080,81 @@ mod tests { ); } + #[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(), 11); + let workspace = WorkspaceRecord { + workspace_id: "local-dev".to_string(), + owner_account_id: None, + display_name: "Local Dev".to_string(), + state: "active".to_string(), + created_at: "1".to_string(), + updated_at: "1".to_string(), + }; + store.upsert_workspace(&workspace).await.unwrap(); + + let document = store + .ensure_memory_document("local-dev", "# Memory\n", "2") + .unwrap(); + assert_eq!(document.body_md, "# Memory\n"); + let updated = MemoryDocumentRecord { + workspace_id: "local-dev".to_string(), + body_md: "# Memory\n\n- fact\n".to_string(), + created_at: document.created_at.clone(), + updated_at: "3".to_string(), + }; + store.upsert_memory_document(&updated).unwrap(); + assert_eq!( + store.get_memory_document("local-dev").unwrap(), + Some(updated) + ); + + let staging = MemoryStagingRecord { + workspace_id: "local-dev".to_string(), + candidate_id: "candidate-a".to_string(), + raw_json: r#"{"kind":"summary","body":"candidate"}"#.to_string(), + source_path: Some("memory/_staging/candidate-a.json".to_string()), + imported_at: "4".to_string(), + }; + store.upsert_memory_staging_record(&staging).unwrap(); + assert_eq!( + store + .get_memory_staging_record("local-dev", "candidate-a") + .unwrap(), + Some(staging.clone()) + ); + assert_eq!( + store.list_memory_staging_records("local-dev", 10).unwrap(), + vec![staging.clone()] + ); + + let resolution = MemoryStagingResolutionRecord { + workspace_id: "local-dev".to_string(), + candidate_id: staging.candidate_id.clone(), + action: "apply".to_string(), + reason: "accepted".to_string(), + affected_refs_json: r#"["summary"]"#.to_string(), + staging_raw_json: staging.raw_json.clone(), + source_path: staging.source_path.clone(), + imported_at: staging.imported_at.clone(), + resolved_at: "5".to_string(), + }; + store.insert_memory_staging_resolution(&resolution).unwrap(); + assert!( + store + .delete_memory_staging_record("local-dev", "candidate-a") + .unwrap() + ); + assert_eq!(store.count_memory_staging_records("local-dev").unwrap(), 0); + assert_eq!( + store + .list_memory_staging_resolutions("local-dev", 10) + .unwrap(), + vec![resolution] + ); + } + #[tokio::test] async fn worker_workdir_registry_round_trips_and_preserves_pinned_retention() { let temp = tempfile::tempdir().unwrap(); @@ -2882,7 +3257,7 @@ mod tests { #[tokio::test] async fn account_and_login_records_round_trip() { let store = SqliteWorkspaceStore::in_memory().unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 10); + assert_eq!(store.schema_version().await.unwrap(), 11); let now = "2026-07-22T00:00:00Z".to_string(); let account = AccountRecord { account_id: "acct-user-alice".to_string(),