| 1 | //! Compatibility receipts on the existing StateStore connection. These records |
| 2 | //! describe committed canonical work; they never execute or own a conversation. |
| 3 | use super::*; |
| 4 | use codewhale_protocol::{ |
| 5 | CanonicalThreadReceipt, LegacyThreadHistory, MAX_CANONICAL_HISTORY_BYTES, |
| 6 | MAX_CANONICAL_HISTORY_ENTRIES, RuntimeOwnerReceipt, |
| 7 | }; |
| 8 | |
| 9 | impl StateStore { |
| 10 | /// Restore an immutable historical archive into an absent SQLite thread. |
| 11 | /// Validation and every row write share one transaction. Existing targets, |
| 12 | /// canonical aliases and partial/cyclic/oversized archives are refused. |
| 13 | pub fn restore_legacy_thread_archive(&self, archive: &LegacyThreadArchive) -> Result<()> { |
| 14 | validate_legacy_archive(archive)?; |
| 15 | let mut conn = self.conn()?; |
| 16 | let tx = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; |
| 17 | let occupied: bool = tx.query_row( |
| 18 | "SELECT EXISTS(SELECT 1 FROM threads WHERE id = ?1) OR EXISTS(SELECT 1 FROM thread_runtime_links WHERE thread_id = ?1) OR EXISTS(SELECT 1 FROM thread_runtime_receipts WHERE thread_id = ?1)", |
| 19 | params![archive.thread.id], |row| row.get(0), |
| 20 | )?; |
| 21 | anyhow::ensure!( |
| 22 | !occupied, |
| 23 | "legacy archive target is already present; no replacement or append is permitted" |
| 24 | ); |
| 25 | write_thread_metadata_on(&tx, &archive.thread)?; |
| 26 | for message in &archive.messages { |
| 27 | tx.execute( |
| 28 | "INSERT INTO messages(id, thread_id, role, content, item_json, created_at, parent_entry_id) VALUES(?1,?2,?3,?4,?5,?6,?7)", |
| 29 | params![message.id, message.thread_id, message.role, message.content, |
| 30 | message.item.as_ref().map(serde_json::to_string).transpose()?, message.created_at, message.parent_entry_id], |
| 31 | )?; |
| 32 | } |
| 33 | tx.execute( |
| 34 | "UPDATE threads SET current_leaf_id = ?1 WHERE id = ?2", |
| 35 | params![archive.thread.current_leaf_id, archive.thread.id], |
| 36 | )?; |
| 37 | if let Some(goal) = &archive.goal { |
| 38 | write_thread_goal_on(&tx, goal)?; |
| 39 | } |
| 40 | for checkpoint in &archive.checkpoints { |
| 41 | tx.execute("INSERT INTO checkpoints(thread_id, checkpoint_id, state_json, created_at) VALUES(?1,?2,?3,?4)", params![checkpoint.thread_id, checkpoint.checkpoint_id, serde_json::to_string(&checkpoint.state)?, checkpoint.created_at])?; |
| 42 | } |
| 43 | tx.commit()?; |
| 44 | Ok(()) |
| 45 | } |
| 46 | |
| 47 | /// Read all branches and the active leaf in one SQLite read transaction. |
| 48 | /// Bounds are checked against SQLite lengths before retaining payloads. |
| 49 | pub fn snapshot_legacy_thread_history(&self, thread_id: &str) -> Result<LegacyThreadHistory> { |
| 50 | let mut conn = self.conn()?; |
| 51 | let tx = conn.transaction()?; |
| 52 | let snapshot = snapshot_history_on(&tx, thread_id)?; |
| 53 | tx.commit()?; |
| 54 | Ok(snapshot) |
| 55 | } |
| 56 | |
| 57 | pub fn get_canonical_runtime_link( |
| 58 | &self, |
| 59 | thread_id: &str, |
| 60 | owner: &RuntimeOwnerReceipt, |
| 61 | ) -> Result<Option<CanonicalThreadReceipt>> { |
| 62 | let conn = self.conn()?; |
| 63 | let record: Option<(String, Option<String>)> = conn.query_row( |
| 64 | "SELECT r.receipt_json, l.runtime_thread_id FROM thread_runtime_receipts r LEFT JOIN thread_runtime_links l ON l.thread_id = r.thread_id WHERE r.thread_id = ?1", |
| 65 | params![thread_id], |row| Ok((row.get(0)?, row.get(1)?)), |
| 66 | ).optional()?; |
| 67 | let Some((encoded, linked)) = record else { |
| 68 | return Ok(None); |
| 69 | }; |
| 70 | let receipt: CanonicalThreadReceipt = serde_json::from_str(&encoded)?; |
| 71 | validate_canonical_receipt(&receipt)?; |
| 72 | anyhow::ensure!( |
| 73 | Some(receipt.runtime_thread_id.as_str()) == linked.as_deref(), |
| 74 | "legacy link changed behind its bound canonical receipt; recovery required" |
| 75 | ); |
| 76 | anyhow::ensure!( |
| 77 | receipt.data_dir == owner.data_dir && receipt.execution_scope == owner.execution_scope, |
| 78 | "thread belongs to a different canonical Runtime store; owner attachment required" |
| 79 | ); |
| 80 | Ok(Some(receipt)) |
| 81 | } |
| 82 | |
| 83 | /// Publish only an already committed owner receipt, comparing the previous |
| 84 | /// link under the same write transaction. Cancellation of the outer waiter |
| 85 | /// cannot roll back a committed canonical operation into a remint. |
| 86 | pub fn publish_canonical_runtime_link( |
| 87 | &self, |
| 88 | thread_id: &str, |
| 89 | expected_previous: Option<&str>, |
| 90 | expected_history: &LegacyThreadHistory, |
| 91 | owner: &RuntimeOwnerReceipt, |
| 92 | receipt: &CanonicalThreadReceipt, |
| 93 | ) -> Result<()> { |
| 94 | validate_canonical_receipt(receipt)?; |
| 95 | anyhow::ensure!( |
| 96 | receipt.data_dir == owner.data_dir && receipt.execution_scope == owner.execution_scope, |
| 97 | "canonical receipt does not match the authenticated owner" |
| 98 | ); |
| 99 | let encoded = serde_json::to_string(receipt)?; |
| 100 | let mut conn = self.conn()?; |
| 101 | let tx = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; |
| 102 | anyhow::ensure!( |
| 103 | expected_history.thread_id == thread_id |
| 104 | && serde_json::to_value(snapshot_history_on(&tx, thread_id)?)? |
| 105 | == serde_json::to_value(expected_history)?, |
| 106 | "legacy history changed before canonical publication; source and canonical result retained for recovery" |
| 107 | ); |
| 108 | let exists: bool = tx.query_row( |
| 109 | "SELECT EXISTS(SELECT 1 FROM threads WHERE id = ?1)", |
| 110 | params![thread_id], |
| 111 | |row| row.get(0), |
| 112 | )?; |
| 113 | anyhow::ensure!( |
| 114 | exists, |
| 115 | "cannot publish canonical link for an absent compatibility thread" |
| 116 | ); |
| 117 | let previous: Option<String> = tx |
| 118 | .query_row( |
| 119 | "SELECT runtime_thread_id FROM thread_runtime_links WHERE thread_id = ?1", |
| 120 | params![thread_id], |
| 121 | |row| row.get(0), |
| 122 | ) |
| 123 | .optional()?; |
| 124 | let saved: Option<String> = tx |
| 125 | .query_row( |
| 126 | "SELECT receipt_json FROM thread_runtime_receipts WHERE thread_id = ?1", |
| 127 | params![thread_id], |
| 128 | |row| row.get(0), |
| 129 | ) |
| 130 | .optional()?; |
| 131 | if saved.as_deref() == Some(&encoded) |
| 132 | && previous.as_deref() == Some(&receipt.runtime_thread_id) |
| 133 | { |
| 134 | tx.commit()?; |
| 135 | return Ok(()); |
| 136 | } |
| 137 | anyhow::ensure!( |
| 138 | saved.is_none(), |
| 139 | "canonical alias is already committed to another operation; recovery required" |
| 140 | ); |
| 141 | anyhow::ensure!( |
| 142 | previous.as_deref() == expected_previous, |
| 143 | "compatibility link changed before canonical publication" |
| 144 | ); |
| 145 | let operation: Option<(String, String)> = tx.query_row("SELECT request_digest, receipt_json FROM thread_runtime_operations WHERE operation_key = ?1", params![receipt.operation_key], |row| Ok((row.get(0)?, row.get(1)?))).optional()?; |
| 146 | if let Some((digest, saved)) = operation { |
| 147 | anyhow::ensure!( |
| 148 | digest == receipt.request_digest && saved == encoded, |
| 149 | "canonical operation key was used for different data" |
| 150 | ); |
| 151 | } |
| 152 | tx.execute("INSERT OR IGNORE INTO thread_runtime_operations(operation_key, request_digest, receipt_json) VALUES(?1, ?2, ?3)", params![receipt.operation_key, receipt.request_digest, encoded])?; |
| 153 | tx.execute("INSERT INTO thread_runtime_links(thread_id, runtime_thread_id, created_at) VALUES(?1, ?2, ?3) ON CONFLICT(thread_id) DO UPDATE SET runtime_thread_id = excluded.runtime_thread_id, created_at = excluded.created_at", params![thread_id, receipt.runtime_thread_id, Utc::now().timestamp()])?; |
| 154 | tx.execute( |
| 155 | "INSERT INTO thread_runtime_receipts(thread_id, receipt_json) VALUES(?1, ?2)", |
| 156 | params![thread_id, encoded], |
| 157 | )?; |
| 158 | tx.commit()?; |
| 159 | Ok(()) |
| 160 | } |
| 161 | } |
| 162 | |
| 163 | fn validate_canonical_receipt(receipt: &CanonicalThreadReceipt) -> Result<()> { |
| 164 | anyhow::ensure!( |
| 165 | receipt.version == 1 |
| 166 | && receipt.data_dir.is_absolute() |
| 167 | && !receipt.execution_scope.is_empty() |
| 168 | && receipt.execution_scope.len() <= 128, |
| 169 | "invalid canonical store receipt" |
| 170 | ); |
| 171 | anyhow::ensure!( |
| 172 | !receipt.operation_key.is_empty() |
| 173 | && receipt.operation_key.len() <= 128 |
| 174 | && !receipt.operation_key.chars().any(char::is_control), |
| 175 | "invalid canonical operation key" |
| 176 | ); |
| 177 | for digest in [&receipt.request_digest, &receipt.history_digest] { |
| 178 | anyhow::ensure!( |
| 179 | digest.len() == 64 |
| 180 | && digest |
| 181 | .bytes() |
| 182 | .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()), |
| 183 | "invalid canonical operation digest" |
| 184 | ); |
| 185 | } |
| 186 | for id in [&receipt.runtime_thread_id, &receipt.session_id] { |
| 187 | anyhow::ensure!( |
| 188 | !id.is_empty() |
| 189 | && id.len() <= 128 |
| 190 | && id |
| 191 | .bytes() |
| 192 | .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-')), |
| 193 | "invalid canonical record identity" |
| 194 | ); |
| 195 | } |
| 196 | Ok(()) |
| 197 | } |
| 198 | |
| 199 | fn snapshot_history_on(conn: &Connection, thread_id: &str) -> Result<LegacyThreadHistory> { |
| 200 | let leaf = conn.query_row( |
| 201 | "SELECT current_leaf_id FROM threads WHERE id = ?1", |
| 202 | params![thread_id], |
| 203 | |row| row.get::<_, Option<i64>>(0), |
| 204 | )?; |
| 205 | let identity: String = conn.query_row( |
| 206 | "SELECT identity FROM state_store_identity WHERE singleton = 1", |
| 207 | [], |
| 208 | |row| row.get(0), |
| 209 | )?; |
| 210 | let (count, bytes): (i64, i64) = conn.query_row( |
| 211 | "SELECT count(*), coalesce(sum(length(cast(role AS BLOB)) + length(cast(content AS BLOB)) + coalesce(length(cast(item_json AS BLOB)), 0) + 128), 0) FROM messages WHERE thread_id = ?1", |
| 212 | params![thread_id], |row| Ok((row.get(0)?, row.get(1)?)), |
| 213 | )?; |
| 214 | anyhow::ensure!( |
| 215 | count >= 0 |
| 216 | && count as u64 <= MAX_CANONICAL_HISTORY_ENTRIES as u64 |
| 217 | && bytes >= 0 |
| 218 | && bytes as u64 <= MAX_CANONICAL_HISTORY_BYTES as u64, |
| 219 | "legacy history exceeds canonical import bounds; source retained for recovery" |
| 220 | ); |
| 221 | let mut messages = Vec::with_capacity(count as usize); |
| 222 | { |
| 223 | let mut statement = conn.prepare("SELECT id, role, content, item_json, created_at, parent_entry_id FROM messages WHERE thread_id = ?1 ORDER BY id")?; |
| 224 | let mut rows = statement.query(params![thread_id])?; |
| 225 | while let Some(row) = rows.next()? { |
| 226 | let raw: Option<String> = row.get(3)?; |
| 227 | messages.push(MessageRecord { |
| 228 | id: row.get(0)?, |
| 229 | thread_id: thread_id.into(), |
| 230 | role: row.get(1)?, |
| 231 | content: row.get(2)?, |
| 232 | item: raw |
| 233 | .as_deref() |
| 234 | .map(serde_json::from_str) |
| 235 | .transpose() |
| 236 | .context("unsupported legacy JSON; source retained for recovery")?, |
| 237 | created_at: row.get(4)?, |
| 238 | parent_entry_id: row.get(5)?, |
| 239 | }); |
| 240 | } |
| 241 | } |
| 242 | let snapshot = LegacyThreadHistory { |
| 243 | version: 1, |
| 244 | state_store_id: identity, |
| 245 | thread_id: thread_id.into(), |
| 246 | current_leaf_id: leaf, |
| 247 | messages, |
| 248 | goal: StateStore::read_thread_goal_snapshot(conn, thread_id)? |
| 249 | .map(|goal| serde_json::from_value(serde_json::to_value(goal)?)) |
| 250 | .transpose()?, |
| 251 | }; |
| 252 | anyhow::ensure!( |
| 253 | serde_json::to_vec(&snapshot)?.len() <= MAX_CANONICAL_HISTORY_BYTES, |
| 254 | "encoded legacy history exceeds canonical import bounds; source retained for recovery" |
| 255 | ); |
| 256 | Ok(snapshot) |
| 257 | } |
| 258 | |
| 259 | fn validate_legacy_archive(archive: &LegacyThreadArchive) -> Result<()> { |
| 260 | anyhow::ensure!( |
| 261 | !archive.thread.id.is_empty() |
| 262 | && archive.thread.id.len() <= 128 |
| 263 | && !archive.thread.id.chars().any(char::is_control), |
| 264 | "invalid legacy archive thread identity" |
| 265 | ); |
| 266 | anyhow::ensure!( |
| 267 | archive |
| 268 | .messages |
| 269 | .len() |
| 270 | .saturating_add(archive.checkpoints.len()) |
| 271 | <= MAX_CANONICAL_HISTORY_ENTRIES, |
| 272 | "legacy archive exceeds entry bounds; source retained" |
| 273 | ); |
| 274 | struct Measure(usize); |
| 275 | impl std::io::Write for Measure { |
| 276 | fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> { |
| 277 | if bytes.len() > self.0 { |
| 278 | return Err(std::io::Error::other("legacy archive exceeds byte bounds")); |
| 279 | } |
| 280 | self.0 -= bytes.len(); |
| 281 | Ok(bytes.len()) |
| 282 | } |
| 283 | fn flush(&mut self) -> std::io::Result<()> { |
| 284 | Ok(()) |
| 285 | } |
| 286 | } |
| 287 | serde_json::to_writer(&mut Measure(MAX_CANONICAL_HISTORY_BYTES), archive)?; |
| 288 | let mut parents = HashMap::new(); |
| 289 | for message in &archive.messages { |
| 290 | anyhow::ensure!( |
| 291 | message.thread_id == archive.thread.id && message.id > 0 && !message.role.is_empty(), |
| 292 | "invalid or foreign legacy archive entry" |
| 293 | ); |
| 294 | anyhow::ensure!( |
| 295 | parents |
| 296 | .insert(message.id, message.parent_entry_id) |
| 297 | .is_none(), |
| 298 | "duplicate legacy archive entry" |
| 299 | ); |
| 300 | } |
| 301 | anyhow::ensure!( |
| 302 | match archive.thread.current_leaf_id { |
| 303 | Some(leaf) => parents.contains_key(&leaf), |
| 304 | None => parents.is_empty(), |
| 305 | }, |
| 306 | "legacy archive has no valid selected leaf" |
| 307 | ); |
| 308 | let mut completed = std::collections::HashSet::new(); |
| 309 | for &id in parents.keys() { |
| 310 | let mut path = std::collections::HashSet::new(); |
| 311 | let mut cursor = Some(id); |
| 312 | while let Some(id) = cursor { |
| 313 | if completed.contains(&id) { |
| 314 | break; |
| 315 | } |
| 316 | anyhow::ensure!(path.insert(id), "cyclic legacy archive graph"); |
| 317 | cursor = *parents |
| 318 | .get(&id) |
| 319 | .ok_or_else(|| anyhow::anyhow!("missing legacy archive parent"))?; |
| 320 | } |
| 321 | completed.extend(path); |
| 322 | } |
| 323 | if let Some(goal) = &archive.goal { |
| 324 | anyhow::ensure!( |
| 325 | goal.thread_id == archive.thread.id |
| 326 | && !goal.goal_id.is_empty() |
| 327 | && !goal.objective.trim().is_empty() |
| 328 | && goal.tokens_used >= 0 |
| 329 | && goal.time_used_seconds >= 0 |
| 330 | && goal.continuation_count >= 0 |
| 331 | && goal.token_budget.is_none_or(|budget| budget >= 0), |
| 332 | "invalid or foreign legacy archive goal" |
| 333 | ); |
| 334 | codewhale_protocol::validate_goal_stall_state( |
| 335 | goal.last_gap_fingerprint.as_deref(), |
| 336 | goal.repeated_gap_count, |
| 337 | goal.last_gap_pass, |
| 338 | u32::try_from(goal.continuation_count).map_err(anyhow::Error::from)?, |
| 339 | ) |
| 340 | .map_err(anyhow::Error::msg)?; |
| 341 | } |
| 342 | let mut checkpoints = std::collections::HashSet::new(); |
| 343 | for checkpoint in &archive.checkpoints { |
| 344 | anyhow::ensure!( |
| 345 | checkpoint.thread_id == archive.thread.id |
| 346 | && !checkpoint.checkpoint_id.is_empty() |
| 347 | && checkpoints.insert(&checkpoint.checkpoint_id), |
| 348 | "invalid, foreign or duplicate legacy archive checkpoint" |
| 349 | ); |
| 350 | } |
| 351 | Ok(()) |
| 352 | } |
| 353 |