mirror of
https://github.com/farion1231/cc-switch.git
synced 2026-10-25 17:22:27 +08:00
fix(hermes): read live WSL databases inside their distribution (#8254)
Windows SQLite cannot reliably read a live Hermes WAL database through a WSL UNC path. For wsl.localhost / wsl$ Hermes paths only, read the database inside the owning distribution via wsl.exe and Python SQLite (mode=ro, query_only, one short read transaction), streaming the result back in bounded JSON frames into the existing session projection. Non-WSL paths and macOS/Linux are unchanged. Requires python3 (3.7+) and timeout inside the distribution. Fixes #8205.
This commit is contained in:
1 parent
d18564481d
commit
d23cee4c11
6 files changed
+655
-1
No files matched your search
@@ -208,6 +208,7 @@ jobs:
|
||||
with:
|
||||
distribution: Ubuntu-24.04
|
||||
wsl-version: 2
|
||||
additional-packages: python3
|
||||
|
||||
# 只恢复不保存:这里只编 lib 测试,若由它先存,同 key 的 Windows 缓存就只剩一小半。
|
||||
# 完整缓存由 backend 矩阵的 Windows job 保存(同一个 shared-key)。
|
||||
@@ -285,3 +286,15 @@ jobs:
|
||||
if ($LASTEXITCODE -ne 0) {
|
||||
exit $LASTEXITCODE
|
||||
}
|
||||
|
||||
- name: Run Hermes live WAL reads inside WSL2
|
||||
env:
|
||||
CC_SWITCH_WSL_DISTRO: Ubuntu-24.04
|
||||
shell: pwsh
|
||||
run: |
|
||||
$testName = "session_manager::providers::hermes::tests::wsl2_live_wal_readonly_reference"
|
||||
$testList = & $env:CC_SWITCH_TEST_EXE --list --ignored
|
||||
if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE }
|
||||
if ($testList -notcontains "${testName}: test") { throw "Hermes WSL test not discovered" }
|
||||
& $env:CC_SWITCH_TEST_EXE $testName --ignored --exact --nocapture
|
||||
if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE }
|
||||
@@ -454,6 +454,20 @@ fn read_sqlite_cell(
|
||||
if !sqlite_allowed(provider_id, table, column) {
|
||||
return Err("不支持的内容引用".to_string());
|
||||
}
|
||||
#[cfg(windows)]
|
||||
if provider_id == "hermes" && crate::session_manager::hermes_wsl::is_wsl(db) {
|
||||
let value = crate::session_manager::hermes_wsl::read(
|
||||
db,
|
||||
serde_json::json!({
|
||||
"operation":"cell", "session_id":session_id, "id":id,
|
||||
"column":column, "max_bytes":MAX_TEXT_BYTES,
|
||||
}),
|
||||
)?;
|
||||
return value
|
||||
.as_str()
|
||||
.map(str::to_string)
|
||||
.ok_or_else(|| NOT_TEXT.to_string());
|
||||
}
|
||||
let conn = Connection::open_with_flags(
|
||||
db,
|
||||
OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
|
||||
|
||||
@@ -0,0 +1,199 @@
|
||||
use std::io::{self, BufRead, BufReader, Read};
|
||||
use std::path::Path;
|
||||
use std::process::{Command, Stdio};
|
||||
use std::thread;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use serde_json::Value;
|
||||
|
||||
const READER: &str = include_str!("hermes_wsl_reader.py");
|
||||
const MAX_FRAME: u64 = 1024 * 1024;
|
||||
|
||||
/// Decode bounded JSON-string frames as a byte stream. Deserializing directly
|
||||
/// avoids retaining another complete encoded copy of a large raw transcript.
|
||||
/// Projection and duplicate selection remain shared with the native provider.
|
||||
struct FramedReader<R> {
|
||||
source: BufReader<R>,
|
||||
frame: io::Cursor<Vec<u8>>,
|
||||
}
|
||||
|
||||
impl<R: Read> FramedReader<R> {
|
||||
fn new(source: R) -> Self {
|
||||
Self {
|
||||
source: BufReader::new(source),
|
||||
frame: io::Cursor::new(Vec::new()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<R: Read> Read for FramedReader<R> {
|
||||
fn read(&mut self, output: &mut [u8]) -> io::Result<usize> {
|
||||
if output.is_empty() {
|
||||
return Ok(0);
|
||||
}
|
||||
loop {
|
||||
let count = self.frame.read(output)?;
|
||||
if count != 0 {
|
||||
return Ok(count);
|
||||
}
|
||||
let mut line = Vec::new();
|
||||
let count = self
|
||||
.source
|
||||
.by_ref()
|
||||
.take(MAX_FRAME + 1)
|
||||
.read_until(b'\n', &mut line)?;
|
||||
if count == 0 {
|
||||
return Ok(0);
|
||||
}
|
||||
if count as u64 > MAX_FRAME || line.last() != Some(&b'\n') {
|
||||
return Err(io::Error::new(
|
||||
io::ErrorKind::InvalidData,
|
||||
"Hermes WSL frame exceeds its limit or is incomplete",
|
||||
));
|
||||
}
|
||||
let decoded: String = serde_json::from_slice(&line)
|
||||
.map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
|
||||
self.frame = io::Cursor::new(decoded.into_bytes());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Only the two WSL UNC hosts are routed to a Linux reader. Keep the public
|
||||
/// source reference as UNC so normal provider/path validation still applies.
|
||||
fn target(path: &Path) -> Option<(String, String)> {
|
||||
let text = path.to_str()?;
|
||||
let rest = text
|
||||
.strip_prefix(r"\\?\UNC\")
|
||||
.or_else(|| text.strip_prefix(r"\\"))?;
|
||||
let mut parts = rest.split('\\');
|
||||
let host = parts.next()?;
|
||||
if !host.eq_ignore_ascii_case("wsl.localhost") && !host.eq_ignore_ascii_case("wsl$") {
|
||||
return None;
|
||||
}
|
||||
let distro = parts.next()?;
|
||||
if distro.is_empty() || distro.starts_with('-') {
|
||||
return None;
|
||||
}
|
||||
let rest: Vec<_> = parts.collect();
|
||||
if rest.is_empty() || rest.iter().any(|p| p.is_empty() || *p == "." || *p == "..") {
|
||||
return None;
|
||||
}
|
||||
Some((distro.to_string(), format!("/{}", rest.join("/"))))
|
||||
}
|
||||
|
||||
pub(super) fn is_wsl(path: &Path) -> bool {
|
||||
target(path).is_some()
|
||||
}
|
||||
|
||||
pub(super) fn read(path: &Path, request: Value) -> Result<Value, String> {
|
||||
let (distro, native) = target(path).ok_or("Invalid Hermes WSL database path")?;
|
||||
let mut command = Command::new("wsl.exe");
|
||||
command.args([
|
||||
"--distribution",
|
||||
&distro,
|
||||
"--exec",
|
||||
"timeout",
|
||||
"8s",
|
||||
"python3",
|
||||
"-I",
|
||||
"-c",
|
||||
READER,
|
||||
&native,
|
||||
&request.to_string(),
|
||||
]);
|
||||
// Do not show a console window on each session refresh.
|
||||
use std::os::windows::process::CommandExt;
|
||||
command.creation_flags(0x08000000);
|
||||
let mut child = command
|
||||
.stdin(Stdio::null())
|
||||
.stdout(Stdio::piped())
|
||||
.stderr(Stdio::piped())
|
||||
.spawn()
|
||||
.map_err(|e| format!("Failed to start Hermes WSL reader: {e}"))?;
|
||||
let stdout = child.stdout.take().unwrap();
|
||||
let stderr = child.stderr.take().unwrap();
|
||||
let out = thread::spawn(move || {
|
||||
serde_json::from_reader::<_, Value>(BufReader::new(FramedReader::new(stdout)))
|
||||
});
|
||||
let err = thread::spawn(move || {
|
||||
let mut bytes = Vec::new();
|
||||
stderr.take(4096).read_to_end(&mut bytes).map(|_| bytes)
|
||||
});
|
||||
let deadline = Instant::now() + Duration::from_secs(10);
|
||||
let status = loop {
|
||||
match child.try_wait() {
|
||||
Ok(Some(status)) => break Ok(status),
|
||||
Ok(None) if Instant::now() < deadline => thread::sleep(Duration::from_millis(20)),
|
||||
result => {
|
||||
let _ = child.kill();
|
||||
let _ = child.wait();
|
||||
break Err(match result {
|
||||
Err(e) => format!("Failed to wait for Hermes WSL reader: {e}"),
|
||||
_ => "Hermes WSL reader timed out".to_string(),
|
||||
});
|
||||
}
|
||||
}
|
||||
};
|
||||
let stdout = out
|
||||
.join()
|
||||
.map_err(|_| "Hermes WSL reader thread failed")?
|
||||
.map_err(|e| format!("Invalid Hermes WSL response: {e}"));
|
||||
let stderr = err
|
||||
.join()
|
||||
.map_err(|_| "Hermes WSL reader thread failed")?
|
||||
.map_err(|e| e.to_string())?;
|
||||
if !status?.success() {
|
||||
return Err(format!(
|
||||
"Hermes WSL read failed (python3 is required): {}",
|
||||
String::from_utf8_lossy(&stderr).trim()
|
||||
));
|
||||
}
|
||||
stdout
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn framed_json_preserves_escape_boundaries_and_rejects_invalid_frames() {
|
||||
let frames = ["[\"", "\\", "n中文\"]"];
|
||||
let stream = frames
|
||||
.iter()
|
||||
.map(|s| format!("{}\n", serde_json::to_string(s).unwrap()))
|
||||
.collect::<String>();
|
||||
let value: Value = serde_json::from_reader(FramedReader::new(stream.as_bytes())).unwrap();
|
||||
assert_eq!(value, serde_json::json!(["\n中文"]));
|
||||
for invalid in ["\"unfinished\"", "123\n"] {
|
||||
assert!(
|
||||
serde_json::from_reader::<_, Value>(FramedReader::new(invalid.as_bytes())).is_err()
|
||||
);
|
||||
}
|
||||
let oversized = format!("\"{}\"\n", "x".repeat(MAX_FRAME as usize));
|
||||
assert!(
|
||||
serde_json::from_reader::<_, Value>(FramedReader::new(oversized.as_bytes())).is_err()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn routes_only_wsl_hosts_and_preserves_linux_path() {
|
||||
for path in [
|
||||
r"\\wsl.localhost\Ubuntu-22.04\home\user\.hermes\state.db",
|
||||
r"\\wsl$\Ubuntu-22.04\home\user\.hermes\state.db",
|
||||
r"\\?\UNC\wsl.localhost\Ubuntu-22.04\home\user\.hermes\state.db",
|
||||
] {
|
||||
assert_eq!(
|
||||
target(Path::new(path)),
|
||||
Some(("Ubuntu-22.04".into(), "/home/user/.hermes/state.db".into()))
|
||||
);
|
||||
}
|
||||
for path in [
|
||||
r"C:\Users\user\.hermes\state.db",
|
||||
r"\\server\share\state.db",
|
||||
r"\\wsl$\-option\state.db",
|
||||
r"\\wsl$\Ubuntu\..\state.db",
|
||||
] {
|
||||
assert!(target(Path::new(path)).is_none());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,88 @@
|
||||
"""Fixed read-only Hermes operations; invoked through WSL, never a shell."""
|
||||
import json
|
||||
import pathlib
|
||||
import sqlite3
|
||||
import sys
|
||||
|
||||
sys.stdout.reconfigure(encoding="utf-8")
|
||||
|
||||
FRAME_CHARS = 128 * 1024
|
||||
request = json.loads(sys.argv[2])
|
||||
connection = sqlite3.connect(
|
||||
pathlib.Path(sys.argv[1]).as_uri() + "?mode=ro", uri=True, timeout=2.0,
|
||||
isolation_level=None,
|
||||
)
|
||||
connection.row_factory = sqlite3.Row
|
||||
connection.execute("PRAGMA query_only=ON")
|
||||
connection.execute("BEGIN")
|
||||
|
||||
|
||||
def rows(sql, parameters=()):
|
||||
result = []
|
||||
for row in connection.execute(sql, parameters):
|
||||
item = {key: (None if isinstance(row[key], bytes) else row[key]) for key in row.keys()}
|
||||
result.append(item)
|
||||
return result
|
||||
|
||||
|
||||
def visible():
|
||||
columns = {row["name"] for row in connection.execute("PRAGMA table_info(messages)")}
|
||||
if "active" not in columns:
|
||||
return ""
|
||||
return " AND (active=1 OR compacted=1)" if "compacted" in columns else " AND active=1"
|
||||
|
||||
|
||||
operation = request["operation"]
|
||||
if operation == "scan":
|
||||
sessions = rows("SELECT * FROM sessions ORDER BY rowid DESC LIMIT 500")
|
||||
edges = {}
|
||||
for name, pick, condition in (
|
||||
("first", "MIN", "role='user'"),
|
||||
("last", "MAX", "role IN ('user','assistant') AND content IS NOT NULL AND content!=''"),
|
||||
):
|
||||
edges[name] = rows(
|
||||
"SELECT m.session_id,m.content FROM messages m JOIN "
|
||||
"(SELECT session_id," + pick + "(id) edge_id FROM messages WHERE "
|
||||
+ condition + visible()
|
||||
+ " AND session_id IN (SELECT id FROM sessions ORDER BY rowid DESC LIMIT 500) "
|
||||
"GROUP BY session_id) f ON m.id=f.edge_id"
|
||||
) if connection.execute("PRAGMA table_info(messages)").fetchone() else []
|
||||
response = {"sessions": sessions, **edges}
|
||||
elif operation == "messages":
|
||||
response = rows(
|
||||
"SELECT * FROM messages WHERE session_id=?" + visible() + " ORDER BY id ASC",
|
||||
(request["session_id"],),
|
||||
)
|
||||
elif operation == "cell":
|
||||
column = request["column"]
|
||||
if column not in ("content", "reasoning", "tool_calls"):
|
||||
raise ValueError("Unsupported Hermes content column")
|
||||
# Fetch the size before the value, preserving the caller's existing limit.
|
||||
parameters = (request["id"], request["session_id"])
|
||||
size = connection.execute(
|
||||
"SELECT length(CAST(" + column + " AS BLOB)) FROM messages WHERE id=? AND session_id=?",
|
||||
parameters,
|
||||
).fetchone()
|
||||
if size is None:
|
||||
raise ValueError("Hermes content is no longer available")
|
||||
if (size[0] or 0) > request["max_bytes"]:
|
||||
raise ValueError("Hermes content is too large")
|
||||
value = connection.execute(
|
||||
"SELECT " + column + " FROM messages WHERE id=? AND session_id=?", parameters,
|
||||
).fetchone()[0]
|
||||
if isinstance(value, bytes):
|
||||
value = value.decode("utf-8", errors="replace")
|
||||
if not isinstance(value, str):
|
||||
raise ValueError("Hermes content is not text")
|
||||
response = value
|
||||
else:
|
||||
raise ValueError("Unsupported Hermes read operation")
|
||||
connection.rollback()
|
||||
connection.close()
|
||||
# Frame the JSON stream, rather than rejecting a whole scan/session before
|
||||
# Rust can apply its display projection. Each frame is bounded even when JSON
|
||||
# escaping expands a cell; cell expansion retains its separate max_bytes guard.
|
||||
# All queries above still use the same fresh, read-only SQLite transaction.
|
||||
for part in json.JSONEncoder(ensure_ascii=False).iterencode(response):
|
||||
for start in range(0, len(part), FRAME_CHARS):
|
||||
print(json.dumps(part[start:start + FRAME_CHARS], ensure_ascii=False))
|
||||
@@ -1,5 +1,7 @@
|
||||
pub mod cache;
|
||||
pub mod content;
|
||||
#[cfg(windows)]
|
||||
pub(crate) mod hermes_wsl;
|
||||
pub mod model;
|
||||
pub mod providers;
|
||||
pub mod terminal;
|
||||
|
||||
@@ -67,6 +67,13 @@ fn scan_sessions_sqlite() -> Vec<SessionMeta> {
|
||||
}
|
||||
|
||||
fn scan_sessions_sqlite_at(db_path: &Path) -> Vec<SessionMeta> {
|
||||
#[cfg(windows)]
|
||||
if crate::session_manager::hermes_wsl::is_wsl(db_path) {
|
||||
return scan_sessions_wsl(db_path).unwrap_or_else(|error| {
|
||||
log::warn!("Failed to scan Hermes WSL database: {error}");
|
||||
Vec::new()
|
||||
});
|
||||
}
|
||||
if !db_path.exists() {
|
||||
return Vec::new();
|
||||
}
|
||||
@@ -130,6 +137,49 @@ fn scan_sessions_sqlite_at(db_path: &Path) -> Vec<SessionMeta> {
|
||||
sessions
|
||||
}
|
||||
|
||||
#[cfg(windows)]
|
||||
fn scan_sessions_wsl(db_path: &Path) -> Result<Vec<SessionMeta>, String> {
|
||||
let data =
|
||||
crate::session_manager::hermes_wsl::read(db_path, serde_json::json!({"operation":"scan"}))?;
|
||||
let edges = |name: &str| -> HashMap<String, String> {
|
||||
data[name]
|
||||
.as_array()
|
||||
.into_iter()
|
||||
.flatten()
|
||||
.filter_map(|row| {
|
||||
let text = decode_content(row["content"].as_str()?);
|
||||
(!text.trim().is_empty()).then(|| {
|
||||
(
|
||||
row["session_id"].as_str().unwrap_or_default().to_string(),
|
||||
text,
|
||||
)
|
||||
})
|
||||
})
|
||||
.collect()
|
||||
};
|
||||
let first = edges("first");
|
||||
let last = edges("last");
|
||||
let source = format!("sqlite:{}", db_path.display());
|
||||
Ok(data["sessions"]
|
||||
.as_array()
|
||||
.ok_or("Invalid Hermes session response")?
|
||||
.iter()
|
||||
.filter_map(|row| {
|
||||
let mut meta = sqlite_row_to_session_meta(row, &source)?;
|
||||
if meta.title.is_none() {
|
||||
meta.title = first
|
||||
.get(&meta.session_id)
|
||||
.map(|text| truncate_summary(text, TITLE_MAX_CHARS));
|
||||
}
|
||||
meta.summary = last
|
||||
.get(&meta.session_id)
|
||||
.or_else(|| first.get(&meta.session_id))
|
||||
.map(|text| truncate_summary(text, 160));
|
||||
Some(meta)
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
/// First displayable user message of each listed session, decoded to text.
|
||||
/// Hermes auto-titles most sessions, so this is the title only when
|
||||
/// `sessions.title` is empty (and the summary fallback when a session has no
|
||||
@@ -290,6 +340,34 @@ pub fn load_messages_sqlite(source: &str) -> Result<Vec<SessionMessage>, String>
|
||||
let (db_path, session_id) = parse_sqlite_source(source)
|
||||
.ok_or_else(|| format!("Invalid SQLite source reference: {source}"))?;
|
||||
|
||||
#[cfg(windows)]
|
||||
if crate::session_manager::hermes_wsl::is_wsl(&db_path) {
|
||||
let data = crate::session_manager::hermes_wsl::read(
|
||||
&db_path,
|
||||
serde_json::json!({"operation":"messages","session_id":session_id}),
|
||||
)?;
|
||||
let rows: Vec<MessageRow> = data
|
||||
.as_array()
|
||||
.ok_or("Invalid Hermes message response")?
|
||||
.iter()
|
||||
.filter_map(|row| {
|
||||
let text = |name: &str| row[name].as_str().map(str::to_string);
|
||||
Some(MessageRow {
|
||||
id: row["id"].as_i64()?,
|
||||
role: row["role"].as_str()?.to_string(),
|
||||
content: text("content"),
|
||||
timestamp: row["timestamp"].as_f64(),
|
||||
tool_call_id: text("tool_call_id"),
|
||||
tool_calls: text("tool_calls"),
|
||||
tool_name: text("tool_name"),
|
||||
active: row["active"].as_i64().unwrap_or(1),
|
||||
display_metadata: text("display_metadata"),
|
||||
reasoning: text("reasoning"),
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
return messages_from_rows(rows);
|
||||
}
|
||||
let conn = Connection::open_with_flags(
|
||||
&db_path,
|
||||
rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
|
||||
@@ -411,11 +489,17 @@ fn load_messages_from_conn(
|
||||
})
|
||||
.map_err(|e| format!("Failed to query messages: {e}"))?;
|
||||
|
||||
messages_from_rows(rows.flatten())
|
||||
}
|
||||
|
||||
fn messages_from_rows(
|
||||
rows: impl IntoIterator<Item = MessageRow>,
|
||||
) -> Result<Vec<SessionMessage>, String> {
|
||||
// Collapse duplicates the way Hermes' display projection does: keep the
|
||||
// first position, show the most live copy.
|
||||
let mut order: Vec<MessageRow> = Vec::new();
|
||||
let mut index: HashMap<DedupeKey, usize> = HashMap::new();
|
||||
for row in rows.flatten() {
|
||||
for row in rows {
|
||||
if row.is_model_only() {
|
||||
continue;
|
||||
}
|
||||
@@ -958,6 +1042,260 @@ mod tests {
|
||||
.collect()
|
||||
}
|
||||
|
||||
// Native-filesystem reference for #8205; this does not exercise Windows
|
||||
// access to WSL's UNC share or its locking/shared-memory implementation.
|
||||
#[test]
|
||||
fn readonly_reader_sees_committed_wal_without_immutable() {
|
||||
if crate::config::sqlite_unsupported_in_temp_dir() {
|
||||
return;
|
||||
}
|
||||
let dir = tempdir().expect("tempdir");
|
||||
let (path, writer) = hermes_db(dir.path());
|
||||
writer
|
||||
.execute_batch("PRAGMA journal_mode=WAL; PRAGMA wal_autocheckpoint=0;")
|
||||
.expect("enable WAL");
|
||||
writer
|
||||
.execute_batch("PRAGMA wal_checkpoint(TRUNCATE);")
|
||||
.expect("checkpoint schema");
|
||||
insert_message(&writer, "s1", "user", "committed in WAL", 1.0);
|
||||
|
||||
let reader = Connection::open_with_flags(
|
||||
&path,
|
||||
rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
|
||||
)
|
||||
.expect("open readonly reader");
|
||||
assert_eq!(contents(&reader, "s1"), vec!["committed in WAL"]);
|
||||
assert!(reader.execute("DELETE FROM messages", []).is_err());
|
||||
|
||||
writer.execute_batch("BEGIN IMMEDIATE;").unwrap();
|
||||
insert_message(&writer, "s1", "assistant", "not yet committed", 2.0);
|
||||
assert_eq!(contents(&reader, "s1"), vec!["committed in WAL"]);
|
||||
writer.execute_batch("COMMIT;").unwrap();
|
||||
assert_eq!(
|
||||
contents(&reader, "s1"),
|
||||
vec!["committed in WAL", "not yet committed"]
|
||||
);
|
||||
|
||||
// immutable disables change detection and assumes the database is
|
||||
// stable. It must not be a fallback for a live Hermes database.
|
||||
let immutable = Connection::open_with_flags(
|
||||
format!("file:{}?mode=ro&immutable=1", path.display()),
|
||||
rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_URI,
|
||||
)
|
||||
.expect("open immutable reader");
|
||||
assert!(contents(&immutable, "s1").is_empty());
|
||||
assert_eq!(
|
||||
load_messages_sqlite(&format!("sqlite:{}#s1", path.display()))
|
||||
.unwrap()
|
||||
.len(),
|
||||
2
|
||||
);
|
||||
}
|
||||
|
||||
// Diagnostic reference only: requires a real Windows host and WSL2 distro.
|
||||
#[cfg(windows)]
|
||||
#[test]
|
||||
#[ignore]
|
||||
fn wsl2_live_wal_readonly_reference() {
|
||||
use std::io::{BufRead, BufReader};
|
||||
use std::process::{Command, Stdio};
|
||||
|
||||
let distro = std::env::var("CC_SWITCH_WSL_DISTRO").expect("WSL distro");
|
||||
let native_path = format!("/tmp/ccswitch8205-{}.db", std::process::id());
|
||||
let mut writer = Command::new("wsl.exe")
|
||||
.args(["--distribution", &distro, "--exec", "python3", "-u", "-c"])
|
||||
.arg(
|
||||
r#"import sqlite3,sys,os
|
||||
p=sys.argv[1]
|
||||
c=sqlite3.connect(p,isolation_level=None)
|
||||
c.execute('CREATE TABLE sessions(id TEXT PRIMARY KEY,cwd TEXT,title TEXT,started_at REAL,last_activity_at REAL)')
|
||||
c.execute("INSERT INTO sessions VALUES('s1','/tmp',NULL,1,1)")
|
||||
c.execute('CREATE TABLE messages(id INTEGER PRIMARY KEY,session_id TEXT,role TEXT,content TEXT,timestamp REAL)')
|
||||
c.execute('PRAGMA journal_mode=WAL')
|
||||
c.execute('PRAGMA wal_autocheckpoint=0')
|
||||
c.execute('PRAGMA wal_checkpoint(TRUNCATE)')
|
||||
c.execute("INSERT INTO messages VALUES(1,'s1','user','first',1)")
|
||||
print('ready',flush=True)
|
||||
for action in sys.stdin:
|
||||
action=action.strip()
|
||||
if action=='begin':
|
||||
c.execute('BEGIN IMMEDIATE')
|
||||
c.execute("INSERT INTO messages VALUES(2,'s1','assistant','second',2)")
|
||||
elif action=='commit': c.execute('COMMIT')
|
||||
elif action=='large':
|
||||
import json
|
||||
for i in range(9):
|
||||
sid='image'+str(i)
|
||||
c.execute("INSERT INTO sessions VALUES(?, '/tmp', NULL, 1, 1)",(sid,))
|
||||
content='\0json:'+json.dumps([{'type':'image_url','image_url':{'url':'data:image/png;base64,'+'A'*(2*1024*1024)}},{'type':'text','text':'visible text'}])
|
||||
c.execute("INSERT INTO messages(session_id,role,content,timestamp) VALUES(?,'user',?,1)",(sid,content))
|
||||
c.execute("INSERT INTO sessions VALUES('tools','/tmp',NULL,1,1)")
|
||||
for i in range(40):
|
||||
# Distinct outputs preserve the native duplicate-selection semantics.
|
||||
c.execute("INSERT INTO messages(session_id,role,content,timestamp) VALUES('tools','tool',?,?)",(str(i)+'\n'+('x'*1024*1024),i+1))
|
||||
elif action=='exit': break
|
||||
print(action,flush=True)
|
||||
c.close()
|
||||
for suffix in ('','-wal','-shm'):
|
||||
try: os.unlink(p+suffix)
|
||||
except FileNotFoundError: pass
|
||||
"#,
|
||||
)
|
||||
.arg(&native_path)
|
||||
.stdin(Stdio::piped())
|
||||
.stdout(Stdio::piped())
|
||||
.stderr(Stdio::inherit())
|
||||
.spawn()
|
||||
.expect("WSL Python writer");
|
||||
let mut output = BufReader::new(writer.stdout.take().unwrap());
|
||||
let mut line = String::new();
|
||||
output.read_line(&mut line).unwrap();
|
||||
assert_eq!(line.trim(), "ready");
|
||||
|
||||
let read_native = || {
|
||||
let result = Command::new("wsl.exe")
|
||||
.args(["--distribution", &distro, "--exec", "python3", "-c"])
|
||||
.arg(
|
||||
r#"import sqlite3,sys,json,pathlib
|
||||
c=sqlite3.connect(pathlib.Path(sys.argv[1]).as_uri()+'?mode=ro',uri=True,isolation_level=None)
|
||||
rows=c.execute('SELECT content FROM messages ORDER BY id').fetchall()
|
||||
try: c.execute('DELETE FROM messages'); raise AssertionError('write succeeded')
|
||||
except sqlite3.OperationalError as e: assert 'readonly' in str(e),str(e)
|
||||
print(json.dumps([row[0] for row in rows]))
|
||||
"#,
|
||||
)
|
||||
.arg(&native_path)
|
||||
.output()
|
||||
.expect("WSL native readonly query");
|
||||
assert!(result.status.success(), "{:?}", result);
|
||||
serde_json::from_slice::<Vec<String>>(&result.stdout).expect("reader JSON")
|
||||
};
|
||||
assert_eq!(read_native(), vec!["first"]);
|
||||
|
||||
let unc = PathBuf::from(format!(
|
||||
r"\\wsl.localhost\{}{}",
|
||||
distro,
|
||||
native_path.replace('/', "\\")
|
||||
));
|
||||
let direct = Connection::open_with_flags(
|
||||
&unc,
|
||||
rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
|
||||
);
|
||||
match direct {
|
||||
Ok(conn) => println!(
|
||||
"Windows rusqlite {} UNC query: {:?}",
|
||||
rusqlite::version(),
|
||||
conn.query_row("SELECT COUNT(*) FROM messages", [], |row| row
|
||||
.get::<_, i64>(0))
|
||||
),
|
||||
Err(error) => println!(
|
||||
"Windows rusqlite {} UNC open: {error:?}",
|
||||
rusqlite::version()
|
||||
),
|
||||
}
|
||||
let source = format!("sqlite:{}#s1", unc.display());
|
||||
let verify_production = |expected: &[&str]| {
|
||||
let messages = load_messages_sqlite(&source).expect("production WSL message route");
|
||||
assert_eq!(
|
||||
messages
|
||||
.iter()
|
||||
.map(|m| m.content.as_str())
|
||||
.collect::<Vec<_>>(),
|
||||
expected
|
||||
);
|
||||
let sessions = scan_sessions_sqlite_at(&unc);
|
||||
assert_eq!(sessions.len(), 1);
|
||||
assert_eq!(sessions[0].title.as_deref(), Some("first"));
|
||||
assert_eq!(sessions[0].summary.as_deref(), expected.last().copied());
|
||||
// Content validation canonicalizes paths to the extended UNC form.
|
||||
let canonical = unc.canonicalize().expect("canonical WSL UNC path");
|
||||
let value = crate::session_manager::hermes_wsl::read(&canonical, serde_json::json!({
|
||||
"operation":"cell", "session_id":"s1", "id":"1", "column":"content", "max_bytes":1024,
|
||||
})).expect("production canonical content route");
|
||||
assert_eq!(value, "first");
|
||||
let validated = crate::session_manager::content::ValidatedSource {
|
||||
provider_id: "hermes".to_string(),
|
||||
raw: source.clone(),
|
||||
location: crate::session_manager::content::SourceLocation::Sqlite {
|
||||
db: canonical.clone(),
|
||||
session_id: "s1".to_string(),
|
||||
},
|
||||
};
|
||||
let content_ref = ContentRef::Sqlite {
|
||||
table: "messages".to_string(),
|
||||
id: "1".to_string(),
|
||||
column: "content".to_string(),
|
||||
pointer: String::new(),
|
||||
};
|
||||
assert_eq!(
|
||||
crate::session_manager::content::resolve_content_ref(&validated, &content_ref)
|
||||
.unwrap(),
|
||||
"first"
|
||||
);
|
||||
let forbidden = ContentRef::Sqlite {
|
||||
table: "sqlite_master".to_string(),
|
||||
id: "1".to_string(),
|
||||
column: "sql".to_string(),
|
||||
pointer: String::new(),
|
||||
};
|
||||
assert!(
|
||||
crate::session_manager::content::resolve_content_ref(&validated, &forbidden)
|
||||
.is_err()
|
||||
);
|
||||
assert!(crate::session_manager::hermes_wsl::read(&canonical, serde_json::json!({
|
||||
"operation":"cell", "session_id":"wrong-session", "id":"1", "column":"content", "max_bytes":1024,
|
||||
})).is_err());
|
||||
assert!(crate::session_manager::hermes_wsl::read(&canonical, serde_json::json!({
|
||||
"operation":"cell", "session_id":"s1", "id":"1", "column":"content", "max_bytes":1,
|
||||
})).is_err());
|
||||
};
|
||||
verify_production(&["first"]);
|
||||
for (action, expected) in [
|
||||
("begin", vec!["first"]),
|
||||
("commit", vec!["first", "second"]),
|
||||
] {
|
||||
writeln!(writer.stdin.as_mut().unwrap(), "{action}").unwrap();
|
||||
line.clear();
|
||||
output.read_line(&mut line).unwrap();
|
||||
assert_eq!(line.trim(), action);
|
||||
assert_eq!(read_native(), expected);
|
||||
verify_production(&expected);
|
||||
}
|
||||
// Review regression: small display projections must not disappear when
|
||||
// the raw image edges or aggregate tool transcript exceed 32 MiB.
|
||||
writeln!(writer.stdin.as_mut().unwrap(), "large").unwrap();
|
||||
line.clear();
|
||||
output.read_line(&mut line).unwrap();
|
||||
assert_eq!(line.trim(), "large");
|
||||
let sessions = scan_sessions_sqlite_at(&unc);
|
||||
for i in 0..9 {
|
||||
let session = sessions
|
||||
.iter()
|
||||
.find(|s| s.session_id == format!("image{i}"))
|
||||
.unwrap();
|
||||
assert_eq!(session.title.as_deref(), Some("visible text"));
|
||||
assert_eq!(session.summary.as_deref(), Some("visible text"));
|
||||
}
|
||||
let tools = load_messages_sqlite(&format!("sqlite:{}#tools", unc.display())).unwrap();
|
||||
assert_eq!(tools.len(), 40);
|
||||
assert!(tools.iter().all(|m| m.content.chars().count() < 1300));
|
||||
let tool_ref = messages_cell(12, "content", String::new());
|
||||
let validated = crate::session_manager::content::ValidatedSource {
|
||||
provider_id: "hermes".into(),
|
||||
raw: format!("sqlite:{}#tools", unc.display()),
|
||||
location: crate::session_manager::content::SourceLocation::Sqlite {
|
||||
db: unc.canonicalize().unwrap(),
|
||||
session_id: "tools".into(),
|
||||
},
|
||||
};
|
||||
let expanded =
|
||||
crate::session_manager::content::resolve_content_ref(&validated, &tool_ref).unwrap();
|
||||
assert!(expanded.len() > 1024 * 1024);
|
||||
assert!(expanded.starts_with("0\n"));
|
||||
writeln!(writer.stdin.as_mut().unwrap(), "exit").unwrap();
|
||||
assert!(writer.wait().unwrap().success());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn load_messages_sqlite_reads_timestamp_in_insertion_order() {
|
||||
if crate::config::sqlite_unsupported_in_temp_dir() {
|
||||
|
||||
Reference in new issue
Block a user