use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use async_stream::stream;
use chrono::{DateTime, Utc};
use rusqlite::{Connection, OptionalExtension};
use serde_json::{Value, json};
use tokio::sync::mpsc;
use crate::{
sessions::IngestEvent,
wire::{FileData, Message, Part, PartKind, Provenance, ProviderOptions, Session},
};
use super::{
Adapter, AdapterError, AdapterFactory, AdapterYield, AdapterYieldStream, DiscoverFuture, Env,
RestoreFidelity, RestoredFile, SkipOracle, SkipReason, by_timestamp_then_id, compact_json,
config_path,
extract::{Extracted, extract_str, json_or_string},
jsonl::{RECORD_CAP, parse_bounded},
part_id, part_ordinal, raw_record, source_options,
sqlite::{self, CHANNEL_CAP, emit},
validate_path_id,
};
const NAME: &str = "opencode";
pub struct OpencodeFactory;
impl AdapterFactory for OpencodeFactory {
fn name(&self) -> &'static str {
NAME
}
fn open(&self, config: Value) -> Result<Box<dyn Adapter>, AdapterError> {
Ok(Box::new(OpencodeAdapter::new(config_path(NAME, config)?)))
}
fn probe_default(&self, env: &Env) -> Option<Value> {
let data_dir = env.home.join(".local").join("share").join("opencode");
if !data_dir.exists() {
return None;
}
let has_db = db_paths(&data_dir).is_ok_and(|dbs| !dbs.is_empty());
let has_tree = data_dir.join("storage").is_dir();
(has_db || has_tree).then(|| json!({ "path": data_dir }))
}
fn serialize(
&self,
session: &crate::sessions::SessionWithMessages,
fidelity: RestoreFidelity,
) -> Result<Vec<RestoredFile>, AdapterError> {
match fidelity {
RestoreFidelity::Native => serialize_native(session),
RestoreFidelity::Foreign => serialize_foreign(session),
}
}
}
#[derive(Debug, Clone)]
pub struct OpencodeAdapter {
root: PathBuf,
}
impl OpencodeAdapter {
pub fn new(root: impl Into<PathBuf>) -> Self {
let root = root.into();
let root = if root.file_name().and_then(|name| name.to_str()) == Some("storage") {
root.parent()
.map_or_else(|| root.clone(), Path::to_path_buf)
} else {
root
};
Self { root }
}
}
fn tree_base(root: &Path) -> PathBuf {
let nested = root.join("storage");
if nested.join("session").is_dir() {
nested
} else {
root.to_path_buf()
}
}
impl Adapter for OpencodeAdapter {
fn discover(&self) -> DiscoverFuture<'_> {
let root = self.root.clone();
Box::pin(async move {
tokio::task::spawn_blocking(move || {
Ok(enumerate_and_peek(&root, false).entries.len())
})
.await
.map_err(join_error)?
})
}
fn plan<'a>(&'a self, oracle: &'a dyn SkipOracle) -> super::PlanFuture<'a> {
let root = self.root.clone();
Box::pin(async move {
let peek = !oracle.is_empty();
let peeked = tokio::task::spawn_blocking(move || enumerate_and_peek(&root, peek))
.await
.map_err(join_error)?;
if !peek {
return Ok(Some(super::SyncPlan::all_pending(peeked.entries.len())));
}
Ok(Some(super::SyncPlan::from_heads(
oracle,
peeked.entries.iter().map(|entry| {
let watermark = match entry.source_ts {
Some(ts) => super::SourceWatermark::At(ts),
None => super::SourceWatermark::Opaque,
};
(Some(entry.source.session_id()), watermark)
}),
)))
})
}
fn events_with<'a>(&'a self, oracle: &'a dyn SkipOracle) -> AdapterYieldStream<'a> {
let adapter = self.clone();
Box::pin(stream! {
let root = adapter.root.clone();
let peek = !oracle.is_empty();
let peeked = tokio::task::spawn_blocking(move || enumerate_and_peek(&root, peek)).await;
let Peeked { tree_base, entries, duplicates, errors } = match peeked {
Ok(peeked) => peeked,
Err(join) => { yield Err(join_error(join)); return; }
};
for error in errors {
yield Err(error);
}
if duplicates > 0 {
yield Ok(AdapterYield::SkippedBatch {
reason: SkipReason::Superseded,
count: duplicates,
});
}
let mut survivors = Vec::with_capacity(entries.len());
for entry in entries {
if crate::adapter::is_session_fresh(oracle, entry.source.session_id(), entry.source_ts) {
yield Ok(AdapterYield::Skipped {
session_id: Some(entry.source.session_id().to_owned()),
project: None,
reason: SkipReason::Fresh,
});
continue;
}
survivors.push(entry.source);
}
let (tx, mut rx) = mpsc::channel(CHANNEL_CAP);
let handle =
tokio::task::spawn_blocking(move || read_survivors(&tree_base, survivors, &tx));
while let Some(item) = rx.recv().await {
yield item;
}
if let Err(join) = handle.await {
yield Err(join_error(join));
}
})
}
}
pub(crate) struct Attribution {
pub source_agent_root: String,
pub project: Extracted<String>,
pub extra_options: (String, Value),
}
pub(crate) fn composed_events_with<'a>(
root: PathBuf,
oracle: &'a dyn SkipOracle,
attribution: Attribution,
) -> AdapterYieldStream<'a> {
use tokio_stream::StreamExt;
Box::pin(stream! {
let adapter = OpencodeAdapter::new(root);
let mut inner = adapter.events_with(oracle);
while let Some(item) = inner.next().await {
yield item.map(|yielded| reattribute(yielded, &attribution));
}
})
}
fn reattribute(yielded: AdapterYield, attribution: &Attribution) -> AdapterYield {
let AdapterYield::Event(IngestEvent::Session(mut session)) = yielded else {
return yielded;
};
session.source_agent = if session.source_agent == NAME {
attribution.source_agent_root.clone()
} else {
format!("{}/subagent", attribution.source_agent_root)
};
session.project = attribution.project.clone();
let (key, value) = &attribution.extra_options;
session.options.insert(key.clone(), value.clone());
AdapterYield::Event(IngestEvent::Session(session))
}
struct HeadEntry {
source: SessionSource,
source_ts: Option<i64>,
}
enum SessionSource {
Db(DbLight),
Tree(SessionFile),
}
impl SessionSource {
fn session_id(&self) -> &str {
match self {
SessionSource::Db(light) => &light.session_id,
SessionSource::Tree(file) => &file.session_id,
}
}
}
struct DbLight {
db_path: PathBuf,
session_id: String,
store: DbStore,
}
enum DbStore {
V1,
V2 { head: &'static str },
Migrated,
}
struct Peeked {
tree_base: PathBuf,
entries: Vec<HeadEntry>,
duplicates: usize,
errors: Vec<AdapterError>,
}
fn enumerate_and_peek(root: &Path, peek: bool) -> Peeked {
let tree_base = tree_base(root);
let mut errors = Vec::new();
let mut conns: HashMap<PathBuf, Connection> = HashMap::new();
let mut db_entries: Vec<DbLight> = Vec::new();
let mut db_ids: HashSet<String> = HashSet::new();
let mut duplicates = 0usize;
let paths = match db_paths(root) {
Ok(paths) => paths,
Err(error) => {
tracing::warn!(path = %root.display(), %error, "opencode: listing DB files failed");
errors.push(error);
Vec::new()
}
};
for db_path in paths {
match db_session_ids(&mut conns, &db_path) {
Ok(ids) => {
for (id, store) in ids {
if !db_ids.insert(id.clone()) {
duplicates += 1;
continue;
}
db_entries.push(DbLight {
db_path: db_path.clone(),
session_id: id,
store,
});
}
}
Err(error) => {
tracing::warn!(
path = %db_path.display(),
%error,
"opencode: enumerating DB sessions failed"
);
errors.push(error);
}
}
}
let mut tree_entries: Vec<SessionFile> = Vec::new();
match collect_session_files(&tree_base) {
Ok(files) => {
for file in files {
if db_ids.contains(&file.session_id) {
duplicates += 1;
} else {
tree_entries.push(file);
}
}
}
Err(error) => {
tracing::warn!(
path = %tree_base.display(),
%error,
"opencode: listing tree sessions failed"
);
errors.push(error);
}
}
let mut entries = Vec::with_capacity(db_entries.len() + tree_entries.len());
for light in db_entries {
let source_ts = if peek {
match connection(&mut conns, &light.db_path) {
Ok(conn) => db_store_watermark(conn, &light).unwrap_or(None),
Err(_) => None,
}
} else {
None
};
entries.push(HeadEntry {
source: SessionSource::Db(light),
source_ts,
});
}
for mut file in tree_entries {
let source_ts = if peek {
match walk_session_subtree(&tree_base, &file.session_id) {
Ok(walk) => {
let ts = newest_message_ts(&walk);
file.cached_subtree = Some(walk);
ts
}
Err(_) => None,
}
} else {
None
};
entries.push(HeadEntry {
source: SessionSource::Tree(file),
source_ts,
});
}
Peeked {
tree_base,
entries,
duplicates,
errors,
}
}
fn read_survivors(
tree_base: &Path,
survivors: Vec<SessionSource>,
tx: &mpsc::Sender<Result<AdapterYield, AdapterError>>,
) {
let mut conns: HashMap<PathBuf, Connection> = HashMap::new();
for source in survivors {
let keep = match source {
SessionSource::Db(light) => read_db_survivor(&mut conns, light, tx),
SessionSource::Tree(file) => read_one_session(tree_base, file, tx),
};
if !keep {
return;
}
}
}
fn read_db_survivor(
conns: &mut HashMap<PathBuf, Connection>,
light: DbLight,
tx: &mpsc::Sender<Result<AdapterYield, AdapterError>>,
) -> bool {
if let Err(error) = validate_path_id(
NAME,
"session id",
&light.session_id,
record_location(&light.db_path, &light.session_id, &light.session_id),
) {
return tx.blocking_send(Err(error)).is_ok();
}
let conn = match connection(conns, &light.db_path) {
Ok(conn) => conn,
Err(error) => return tx.blocking_send(Err(error)).is_ok(),
};
let head = match light.store {
DbStore::V1 | DbStore::Migrated => {
fetch_db_session_head(conn, &light.db_path, &light.session_id)
}
DbStore::V2 { head } => {
fetch_v2_session_head(conn, &light.db_path, &light.session_id, head)
}
};
match head {
Ok(Some(head)) => {
let session_id = head.id.clone();
let anchor = head.created_at;
emit!(tx, Ok(AdapterYield::Event(IngestEvent::Session(head))));
match light.store {
DbStore::V1 => read_db_session(conn, &light.db_path, &session_id, anchor, tx),
DbStore::V2 { .. } => {
read_v2_messages(conn, &light.db_path, &session_id, anchor, None, tx)
}
DbStore::Migrated => {
if !read_db_session(conn, &light.db_path, &session_id, anchor, tx) {
return false;
}
let copies = match migration_copies(conn, &light.db_path, &session_id) {
Ok(copies) => copies,
Err(error) => return tx.blocking_send(Err(error)).is_ok(),
};
read_v2_messages(conn, &light.db_path, &session_id, anchor, Some(&copies), tx)
}
}
}
Ok(None) => {
let error = AdapterError::schema(
NAME,
record_location(&light.db_path, &light.session_id, &light.session_id),
"session row vanished between enumeration and read",
);
tx.blocking_send(Err(error)).is_ok()
}
Err(error) => tx.blocking_send(Err(error)).is_ok(),
}
}
fn join_error(join: tokio::task::JoinError) -> AdapterError {
sqlite::join_error(NAME, join)
}
struct SessionFile {
session_id: String,
path: PathBuf,
cached_subtree: Option<SubtreeWalk>,
}
struct SubtreeWalk {
message_files: Vec<PathBuf>,
part_files_by_message: Vec<Vec<PathBuf>>,
}
fn collect_session_files(root: &Path) -> Result<Vec<SessionFile>, AdapterError> {
let session_root = root.join("session");
let io = |path: &Path, source| AdapterError::io(NAME, path.display().to_string(), source);
let entries = match std::fs::read_dir(&session_root) {
Ok(entries) => entries,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
Err(error) => return Err(io(&session_root, error)),
};
let mut out = Vec::new();
for project in entries {
let project = project.map_err(|error| io(&session_root, error))?;
if !project
.file_type()
.map_err(|error| io(&project.path(), error))?
.is_dir()
{
continue;
}
let project_dir = project.path();
for session in std::fs::read_dir(&project_dir).map_err(|error| io(&project_dir, error))? {
let session = session.map_err(|error| io(&project_dir, error))?;
let path = session.path();
if path.extension().and_then(|ext| ext.to_str()) != Some("json") {
continue;
}
let Some(session_id) = path
.file_stem()
.and_then(|s| s.to_str())
.map(ToOwned::to_owned)
else {
continue;
};
validate_path_id(
NAME,
"session file name",
&session_id,
path.display().to_string(),
)?;
out.push(SessionFile {
session_id,
path,
cached_subtree: None,
});
}
}
out.sort_by(|a, b| a.path.cmp(&b.path));
Ok(out)
}
fn walk_session_subtree(tree_base: &Path, session_id: &str) -> Result<SubtreeWalk, AdapterError> {
let message_dir = tree_base.join("message").join(session_id);
let message_files = list_json_sorted(&message_dir)?;
let mut part_files_by_message = Vec::with_capacity(message_files.len());
for message_path in &message_files {
let Some(message_id) = message_path.file_stem().and_then(|stem| stem.to_str()) else {
part_files_by_message.push(Vec::new());
continue;
};
validate_path_id(
NAME,
"message file name",
message_id,
message_path.display().to_string(),
)?;
let part_dir = tree_base.join("part").join(message_id);
let parts = list_json_sorted(&part_dir)?;
part_files_by_message.push(parts);
}
Ok(SubtreeWalk {
message_files,
part_files_by_message,
})
}
fn newest_message_ts(walk: &SubtreeWalk) -> Option<i64> {
let message = read_json(walk.message_files.last()?).ok()?;
let parts = walk
.part_files_by_message
.last()
.into_iter()
.flatten()
.filter_map(|part_path| read_json(part_path).ok());
watermark_micros(&message, parts)
}
fn watermark_micros(message: &Value, parts: impl IntoIterator<Item = Value>) -> Option<i64> {
let mut newest = millis_at(message, &["time", "created"])?;
for part in parts {
if let Some(end) = millis_at(&part, &["state", "time", "end"]) {
newest = newest.max(end);
}
}
Some(newest.timestamp_micros())
}
fn read_one_session(
tree_base: &Path,
file: SessionFile,
tx: &mpsc::Sender<Result<AdapterYield, AdapterError>>,
) -> bool {
let session_value = match read_json(&file.path) {
Ok(value) => value,
Err(error) => {
emit!(tx, Err(error));
return true;
}
};
let session = match session_from_value(&session_value, &file.path) {
Ok(session) => session,
Err(error) => {
emit!(tx, Err(error));
return true;
}
};
let session_id = session.id.clone();
if let Err(error) = validate_path_id(
NAME,
"session id",
&session_id,
file.path.display().to_string(),
) {
emit!(tx, Err(error));
return true;
}
let session_created_at = session.created_at;
emit!(tx, Ok(AdapterYield::Event(IngestEvent::Session(session))));
let (message_files, mut part_files_by_message) = match file.cached_subtree {
Some(walk) => (walk.message_files, walk.part_files_by_message),
None => {
let message_dir = tree_base.join("message").join(&session_id);
let files = match list_json_sorted(&message_dir) {
Ok(files) => files,
Err(error) => {
emit!(tx, Err(error));
return true;
}
};
(files, Vec::new())
}
};
let use_cache = !part_files_by_message.is_empty();
for (index, message_path) in message_files.iter().enumerate() {
let message_value = match read_json(message_path) {
Ok(value) => value,
Err(error) => {
emit!(tx, Err(error));
continue;
}
};
let Some(message_id) = message_value.get("id").and_then(Value::as_str) else {
emit!(
tx,
Err(AdapterError::schema(
NAME,
message_path.display().to_string(),
"message file missing `id`",
))
);
continue;
};
if let Err(error) = validate_path_id(
NAME,
"message id",
message_id,
message_path.display().to_string(),
) {
emit!(tx, Err(error));
continue;
}
let part_files = if use_cache {
std::mem::take(&mut part_files_by_message[index])
} else {
let part_dir = tree_base.join("part").join(message_id);
match list_json_sorted(&part_dir) {
Ok(files) => files,
Err(error) => {
emit!(tx, Err(error));
continue;
}
}
};
let mut parts = Vec::with_capacity(part_files.len());
for part_path in part_files {
match read_json(&part_path) {
Ok(value) => parts.push(value),
Err(error) => emit!(tx, Err(error)),
}
}
match build_message_events(&session_id, &message_value, &parts, session_created_at) {
Ok(events) => {
for event in events {
emit!(tx, Ok(AdapterYield::Event(event)));
}
}
Err(error) => emit!(tx, Err(error)),
}
}
true
}
const SESSION_COLUMNS: &str = "id, project_id, workspace_id, parent_id, slug, directory, path, \
title, version, share_url, summary_additions, summary_deletions, summary_files, \
summary_diffs, metadata, cost, tokens_input, tokens_output, tokens_reasoning, \
tokens_cache_read, tokens_cache_write, revert, permission, agent, model, time_created, \
time_updated, time_compacting, time_archived";
fn list_files_sorted(
dir: &Path,
extension: &str,
name_filter: impl Fn(&str) -> bool,
) -> Result<Vec<PathBuf>, AdapterError> {
let entries = match std::fs::read_dir(dir) {
Ok(entries) => entries,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
Err(error) => return Err(AdapterError::io(NAME, dir.display().to_string(), error)),
};
let mut out = Vec::new();
for entry in entries {
let entry =
entry.map_err(|error| AdapterError::io(NAME, dir.display().to_string(), error))?;
let path = entry.path();
if path.extension().and_then(|ext| ext.to_str()) != Some(extension) {
continue;
}
if path
.file_name()
.and_then(|name| name.to_str())
.is_some_and(&name_filter)
{
out.push(path);
}
}
out.sort();
Ok(out)
}
fn db_paths(root: &Path) -> Result<Vec<PathBuf>, AdapterError> {
let mut paths = list_files_sorted(root, "db", |name| name.starts_with("opencode"))?;
paths.sort_by_key(|path| path.file_name() != Some(std::ffi::OsStr::new("opencode.db")));
Ok(paths)
}
#[cfg(test)]
fn open_db(path: &Path) -> Result<Connection, AdapterError> {
sqlite::open_db(NAME, path)
}
fn db_error(path: &Path, op: &str, error: &rusqlite::Error) -> AdapterError {
sqlite::db_error(NAME, path, op, error)
}
fn connection<'a>(
conns: &'a mut HashMap<PathBuf, Connection>,
path: &Path,
) -> Result<&'a Connection, AdapterError> {
sqlite::connection(NAME, conns, path)
}
fn db_session_ids(
conns: &mut HashMap<PathBuf, Connection>,
db_path: &Path,
) -> Result<Vec<(String, DbStore)>, AdapterError> {
let conn = connection(conns, db_path)?;
let v1 = table_ids(conn, db_path, "session")?;
let v2 = table_ids(conn, db_path, "session_v2")?;
let v2_set: HashSet<&str> = v2.iter().map(String::as_str).collect();
let has_v2_body = table_exists(conn, db_path, "session_message")?;
let prerelease = has_v2_body && !table_exists(conn, db_path, "message")?;
let with_v2_body: HashSet<String> = if has_v2_body {
conn.prepare_cached("SELECT DISTINCT session_id FROM session_message")
.and_then(|mut stmt| {
stmt.query_map([], |row| row.get::<_, String>(0))?
.collect::<rusqlite::Result<_>>()
})
.map_err(|error| db_error(db_path, "list v2 message sessions", &error))?
} else {
HashSet::new()
};
let v1_set: HashSet<&str> = v1.iter().map(String::as_str).collect();
let mut ids: Vec<(String, DbStore)> = v1
.iter()
.map(|id| {
let store = if v2_set.contains(id.as_str()) {
DbStore::Migrated
} else if prerelease || with_v2_body.contains(id) {
DbStore::V2 { head: "session" }
} else {
DbStore::V1
};
(id.clone(), store)
})
.chain(
v2.iter()
.filter(|id| !v1_set.contains(id.as_str()))
.map(|id| (id.clone(), DbStore::V2 { head: "session_v2" })),
)
.collect();
ids.sort_unstable_by(|a, b| a.0.cmp(&b.0));
Ok(ids)
}
fn table_exists(conn: &Connection, db_path: &Path, table: &str) -> Result<bool, AdapterError> {
sqlite::has_table(conn, table)
.map_err(|error| db_error(db_path, &format!("probe {table} table"), &error))
}
fn table_ids(conn: &Connection, db_path: &Path, table: &str) -> Result<Vec<String>, AdapterError> {
if !table_exists(conn, db_path, table)? {
return Ok(Vec::new());
}
let mut stmt = conn
.prepare_cached(&format!("SELECT id FROM {table}"))
.map_err(|error| db_error(db_path, &format!("prepare {table} ids"), &error))?;
let rows = stmt
.query_map([], |row| row.get::<_, String>(0))
.map_err(|error| db_error(db_path, &format!("query {table} ids"), &error))?;
rows.collect::<rusqlite::Result<Vec<_>>>()
.map_err(|error| db_error(db_path, &format!("read {table} id"), &error))
}
fn fetch_db_session_head(
conn: &Connection,
db_path: &Path,
session_id: &str,
) -> Result<Option<Session>, AdapterError> {
let mut stmt = conn
.prepare_cached(&format!(
"SELECT {SESSION_COLUMNS} FROM session WHERE id = ?1"
))
.map_err(|error| db_error(db_path, "prepare session by id", &error))?;
let info = stmt
.query_row([session_id], session_info_from_row)
.optional()
.map_err(|error| db_error(db_path, "query session by id", &error))?;
info.map(|info| db_session_head(db_path, &info)).transpose()
}
fn db_session_head(db_path: &Path, info: &Value) -> Result<Session, AdapterError> {
let mut session = session_from_value(info, db_path)?;
if info.get("parentID").and_then(Value::as_str).is_some() {
session.source_agent = match info.get("agent").and_then(Value::as_str) {
Some(agent) => format!("{NAME}/{agent}"),
None => format!("{NAME}/subagent"),
};
}
Ok(session)
}
fn session_info_from_row(row: &rusqlite::Row) -> rusqlite::Result<Value> {
let mut info = serde_json::Map::new();
info.insert("id".to_owned(), json!(row.get::<_, String>("id")?));
info.insert("slug".to_owned(), json!(row.get::<_, String>("slug")?));
info.insert(
"projectID".to_owned(),
json!(row.get::<_, String>("project_id")?),
);
if let Some(value) = row.get::<_, Option<String>>("workspace_id")? {
info.insert("workspaceID".to_owned(), json!(value));
}
info.insert(
"directory".to_owned(),
json!(row.get::<_, String>("directory")?),
);
if let Some(value) = row.get::<_, Option<String>>("path")? {
info.insert("path".to_owned(), json!(value));
}
if let Some(value) = row.get::<_, Option<String>>("parent_id")? {
info.insert("parentID".to_owned(), json!(value));
}
info.insert("title".to_owned(), json!(row.get::<_, String>("title")?));
if let Some(value) = row.get::<_, Option<String>>("agent")? {
info.insert("agent".to_owned(), json!(value));
}
if let Some(value) = row.get::<_, Option<String>>("model")? {
info.insert("model".to_owned(), json_or_string(&value));
}
info.insert(
"version".to_owned(),
json!(row.get::<_, String>("version")?),
);
let summary_additions = row.get::<_, Option<i64>>("summary_additions")?;
let summary_deletions = row.get::<_, Option<i64>>("summary_deletions")?;
let summary_files = row.get::<_, Option<i64>>("summary_files")?;
let summary_diffs = row.get::<_, Option<String>>("summary_diffs")?;
if summary_additions.is_some() || summary_deletions.is_some() || summary_files.is_some() {
let mut summary = serde_json::Map::new();
summary.insert(
"additions".to_owned(),
json!(summary_additions.unwrap_or(0)),
);
summary.insert(
"deletions".to_owned(),
json!(summary_deletions.unwrap_or(0)),
);
summary.insert("files".to_owned(), json!(summary_files.unwrap_or(0)));
if let Some(value) = &summary_diffs {
summary.insert("diffs".to_owned(), json_or_string(value));
}
info.insert("summary".to_owned(), Value::Object(summary));
}
info.insert("cost".to_owned(), json!(row.get::<_, f64>("cost")?));
info.insert(
"tokens".to_owned(),
json!({
"input": row.get::<_, i64>("tokens_input")?,
"output": row.get::<_, i64>("tokens_output")?,
"reasoning": row.get::<_, i64>("tokens_reasoning")?,
"cache": {
"read": row.get::<_, i64>("tokens_cache_read")?,
"write": row.get::<_, i64>("tokens_cache_write")?,
},
}),
);
if let Some(value) = row.get::<_, Option<String>>("share_url")? {
info.insert("share".to_owned(), json!({ "url": value }));
}
if let Some(value) = row.get::<_, Option<String>>("metadata")? {
info.insert("metadata".to_owned(), json_or_string(&value));
}
if let Some(value) = row.get::<_, Option<String>>("revert")? {
info.insert("revert".to_owned(), json_or_string(&value));
}
if let Some(value) = row.get::<_, Option<String>>("permission")? {
info.insert("permission".to_owned(), json_or_string(&value));
}
let mut time = serde_json::Map::new();
time.insert(
"created".to_owned(),
json!(row.get::<_, i64>("time_created")?),
);
time.insert(
"updated".to_owned(),
json!(row.get::<_, i64>("time_updated")?),
);
if let Some(value) = row.get::<_, Option<i64>>("time_compacting")? {
time.insert("compacting".to_owned(), json!(value));
}
if let Some(value) = row.get::<_, Option<i64>>("time_archived")? {
time.insert("archived".to_owned(), json!(value));
}
info.insert("time".to_owned(), Value::Object(time));
Ok(Value::Object(info))
}
fn db_session_watermark(
conn: &Connection,
db_path: &Path,
session_id: &str,
) -> Result<Option<i64>, AdapterError> {
let mut newest_stmt = conn
.prepare_cached("SELECT id FROM message WHERE session_id = ?1 ORDER BY id DESC LIMIT 1")
.map_err(|error| db_error(db_path, "prepare newest message id", &error))?;
let newest = newest_stmt
.query_row([session_id], |row| row.get::<_, String>(0))
.optional()
.map_err(|error| db_error(db_path, "query newest message id", &error))?;
let Some(message_id) = newest else {
return Ok(None);
};
let mut data_stmt = conn
.prepare_cached("SELECT data FROM message WHERE id = ?1")
.map_err(|error| db_error(db_path, "prepare newest message data", &error))?;
let data = data_stmt
.query_row([&message_id], |row| row.get::<_, String>(0))
.optional()
.map_err(|error| db_error(db_path, "query newest message data", &error))?;
let Some(data) = data else {
return Ok(None);
};
let Ok(message) = serde_json::from_str::<Value>(&data) else {
return Ok(None);
};
let mut stmt = conn
.prepare_cached("SELECT data FROM part WHERE message_id = ?1 ORDER BY id")
.map_err(|error| db_error(db_path, "prepare newest parts", &error))?;
let rows = stmt
.query_map([&message_id], |row| row.get::<_, String>(0))
.map_err(|error| db_error(db_path, "query newest parts", &error))?;
let datas = rows
.collect::<rusqlite::Result<Vec<_>>>()
.map_err(|error| db_error(db_path, "read newest part row", &error))?;
let parts = datas
.iter()
.filter_map(|data| serde_json::from_str::<Value>(data).ok());
Ok(watermark_micros(&message, parts))
}
struct RecordRow {
id: String,
data: String,
}
fn fetch_records(
conn: &Connection,
db_path: &Path,
sql: &str,
param: &str,
kind: &str,
) -> Result<Vec<RecordRow>, AdapterError> {
let mut stmt = conn
.prepare_cached(sql)
.map_err(|error| db_error(db_path, &format!("prepare {kind}"), &error))?;
let rows = stmt
.query_map([param], |row| {
Ok(RecordRow {
id: row.get(0)?,
data: row.get(1)?,
})
})
.map_err(|error| db_error(db_path, &format!("query {kind}"), &error))?;
rows.collect::<rusqlite::Result<Vec<_>>>()
.map_err(|error| db_error(db_path, &format!("read {kind} row"), &error))
}
fn read_db_session(
conn: &Connection,
db_path: &Path,
session_id: &str,
session_anchor: DateTime<Utc>,
tx: &mpsc::Sender<Result<AdapterYield, AdapterError>>,
) -> bool {
let mut messages = match fetch_records(
conn,
db_path,
"SELECT id, data FROM message WHERE session_id = ?1",
session_id,
"message",
) {
Ok(rows) => rows,
Err(error) => {
emit!(tx, Err(error));
return true;
}
};
messages.sort_unstable_by(|a, b| a.id.cmp(&b.id));
for message in messages {
let message_id = message.id;
let default_ts = session_anchor;
let message_value = match reconstruct_record(
&message.data,
[("id", json!(message_id)), ("sessionID", json!(session_id))],
db_path,
session_id,
&message_id,
) {
Ok(value) => value,
Err(error) => {
emit!(tx, Err(error));
continue;
}
};
let part_rows = match fetch_records(
conn,
db_path,
"SELECT id, data FROM part WHERE message_id = ?1 ORDER BY id",
&message_id,
"part",
) {
Ok(rows) => rows,
Err(error) => {
emit!(tx, Err(error));
continue;
}
};
let mut parts = Vec::with_capacity(part_rows.len());
for part in part_rows {
match reconstruct_record(
&part.data,
[
("id", json!(part.id)),
("sessionID", json!(session_id)),
("messageID", json!(message_id)),
],
db_path,
session_id,
&part.id,
) {
Ok(value) => parts.push(value),
Err(error) => emit!(tx, Err(error)),
}
}
match build_message_events(session_id, &message_value, &parts, default_ts) {
Ok(events) => {
for event in events {
emit!(tx, Ok(AdapterYield::Event(event)));
}
}
Err(error) => emit!(tx, Err(error)),
}
}
true
}
fn reconstruct_record(
data: &str,
inject: impl IntoIterator<Item = (&'static str, Value)>,
db_path: &Path,
session_id: &str,
record_id: &str,
) -> Result<Value, AdapterError> {
let mut value = parse_bounded(NAME, data.as_bytes(), || {
record_location(db_path, session_id, record_id)
})?;
if let Value::Object(map) = &mut value {
for (key, injected) in inject {
map.insert(key.to_owned(), injected);
}
}
Ok(value)
}
struct MigrationCopies {
ids: HashSet<String>,
through_seq: Option<i64>,
}
impl MigrationCopies {
fn contains(&self, id: &str, seq: i64) -> bool {
self.ids.contains(id) || self.through_seq.is_some_and(|through| seq <= through)
}
fn live_from(copies: Option<&Self>) -> i64 {
copies
.and_then(|copies| copies.through_seq)
.map_or(i64::MIN, |through| through.saturating_add(1))
}
}
fn migration_copies(
conn: &Connection,
db_path: &Path,
session_id: &str,
) -> Result<MigrationCopies, AdapterError> {
let read = |sql: &str, op: &str| -> Result<Vec<(String, Option<i64>)>, AdapterError> {
conn.prepare_cached(sql)
.and_then(|mut stmt| {
stmt.query_map([session_id], |row| Ok((row.get(0)?, row.get(1)?)))?
.collect::<rusqlite::Result<_>>()
})
.map_err(|error| db_error(db_path, op, &error))
};
let mut ids = HashSet::new();
for (id, _) in read(
"SELECT id, NULL FROM message WHERE session_id = ?1",
"read migrated message ids",
)? {
let split = synthetic_id(&id);
ids.insert(synthetic_id(&split));
ids.insert(split);
ids.insert(id);
}
let through_seq = read(
"SELECT id, seq FROM session_message WHERE session_id = ?1",
"read migrated v2 ids",
)?
.into_iter()
.filter(|(id, _)| ids.contains(id))
.filter_map(|(_, seq)| seq)
.max();
Ok(MigrationCopies { ids, through_seq })
}
fn synthetic_id(source: &str) -> String {
use sha2::{Digest, Sha256};
const ALPHABET: &[u8; 62] = b"0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz";
let mut value = Sha256::digest(format!("v1-synthetic:{source}")).to_vec();
let mut suffix = [0u8; 14];
for slot in suffix.iter_mut().rev() {
let mut remainder = 0u32;
for byte in &mut value {
let acc = (remainder << 8) | u32::from(*byte);
*byte = u8::try_from(acc / 62).unwrap_or(u8::MAX);
remainder = acc % 62;
}
*slot = ALPHABET[remainder as usize];
}
let prefix: String = source.chars().take(16).collect();
format!("{prefix}{}", String::from_utf8_lossy(&suffix))
}
fn is_settled(kind: &str, record: &Value) -> bool {
match kind {
"assistant" => record.pointer("/time/completed").is_some(),
"shell" | "compaction" => record.get("status").and_then(Value::as_str) != Some("running"),
_ => true,
}
}
fn db_store_watermark(conn: &Connection, light: &DbLight) -> Result<Option<i64>, AdapterError> {
let (db_path, session_id) = (light.db_path.as_path(), light.session_id.as_str());
match light.store {
DbStore::V1 => db_session_watermark(conn, db_path, session_id),
DbStore::V2 { .. } => Ok(match v2_session_watermark(conn, session_id, None)? {
V2Watermark::At(ts) => Some(ts),
V2Watermark::Empty | V2Watermark::Unknown => None,
}),
DbStore::Migrated => {
let copies = migration_copies(conn, db_path, session_id)?;
let v1 = db_session_watermark(conn, db_path, session_id)?;
Ok(
match v2_session_watermark(conn, session_id, Some(&copies))? {
V2Watermark::At(ts) => v1.max(Some(ts)),
V2Watermark::Empty => v1,
V2Watermark::Unknown => None,
},
)
}
}
}
enum V2Watermark {
Empty,
At(i64),
Unknown,
}
fn v2_session_watermark(
conn: &Connection,
session_id: &str,
copies: Option<&MigrationCopies>,
) -> Result<V2Watermark, AdapterError> {
let rows = (|| -> rusqlite::Result<(Vec<V2Row>, Vec<String>)> {
let open_turn = conn
.prepare_cached(
"SELECT MAX(seq) FROM session_message WHERE session_id = ?1 AND type = 'idle'",
)?
.query_row([session_id], |row| row.get::<_, Option<i64>>(0))?
.unwrap_or(i64::MIN)
.max(MigrationCopies::live_from(copies));
let turn = conn
.prepare_cached(
"SELECT id, type, seq, data FROM session_message \
WHERE session_id = ?1 AND seq >= ?2 ORDER BY seq",
)?
.query_map(rusqlite::params![session_id, open_turn], v2_row)?
.collect::<rusqlite::Result<_>>()?;
let shells = conn
.prepare_cached(
"SELECT data FROM session_message WHERE session_id = ?1 AND type = 'shell'",
)?
.query_map([session_id], |row| row.get(0))?
.collect::<rusqlite::Result<_>>()?;
Ok((turn, shells))
})();
let Ok((turn, shells)) = rows else {
return Ok(V2Watermark::Unknown);
};
let mut newest: Option<DateTime<Utc>> = None;
for data in &shells {
let Ok(record) = serde_json::from_str::<Value>(data) else {
return Ok(V2Watermark::Unknown);
};
if !is_settled("shell", &record) {
return Ok(V2Watermark::Unknown);
}
newest = newest.max(millis_at(&record, &["time", "completed"]));
}
for (id, kind, seq, data) in turn {
if copies.is_some_and(|copies| copies.contains(&id, seq)) {
continue;
}
let Ok(record) = serde_json::from_str::<Value>(&data) else {
return Ok(V2Watermark::Unknown);
};
let Some(created) = millis_at(&record, &["time", "created"]) else {
return Ok(V2Watermark::Unknown);
};
if matches!(kind.as_str(), "assistant" | "compaction") || !is_settled(&kind, &record) {
return Ok(V2Watermark::Unknown);
}
let emitted = v2_tool_items(&record)
.filter_map(tool_result_time)
.fold(created, DateTime::max);
newest = newest.max(Some(emitted));
}
Ok(newest.map_or(V2Watermark::Empty, |ts| {
V2Watermark::At(ts.timestamp_micros())
}))
}
const V2_JSON_COLUMNS: &[&str] = &[
"fork_boundary",
"summary_diffs",
"metadata",
"revert",
"permission",
"model",
];
fn fetch_v2_session_head(
conn: &Connection,
db_path: &Path,
session_id: &str,
table: &'static str,
) -> Result<Option<Session>, AdapterError> {
let mut stmt = conn
.prepare_cached(&format!("SELECT * FROM {table} WHERE id = ?1"))
.map_err(|error| db_error(db_path, "prepare v2 session by id", &error))?;
let row = stmt
.query_row([session_id], v2_row_value)
.optional()
.map_err(|error| db_error(db_path, "query v2 session by id", &error))?;
row.map(|row| {
let location = record_location(db_path, session_id, session_id);
let row = parse_bounded(NAME, compact_json(&row).as_bytes(), || location.clone())?;
session_from_v2_row(&row, &location, table)
})
.transpose()
}
fn v2_row_value(row: &rusqlite::Row) -> rusqlite::Result<Value> {
let mut map = serde_json::Map::new();
for (index, name) in row.as_ref().column_names().into_iter().enumerate() {
let value = match sqlite::value_json(row.get_ref(index)?) {
Value::Null => continue,
Value::String(text) if V2_JSON_COLUMNS.contains(&name) => json_or_string(&text),
value => value,
};
map.insert(name.to_owned(), value);
}
Ok(Value::Object(map))
}
fn set_column<'a>(row: &'a Value, name: &str) -> Option<&'a str> {
row.get(name)
.and_then(Value::as_str)
.filter(|value| !value.is_empty())
}
fn session_from_v2_row(
row: &Value,
location: &str,
table: &'static str,
) -> Result<Session, AdapterError> {
let schema =
|message: &str| AdapterError::schema(NAME, location.to_owned(), message.to_owned());
let id = row
.get("id")
.and_then(Value::as_str)
.ok_or_else(|| schema("v2 session row missing `id`"))?
.to_owned();
let created_at = millis_at(row, &["time_created"])
.ok_or_else(|| schema("v2 session row missing `time_created`"))?;
let project = extract_str(row, "directory")
.ok_or_else(|| schema("v2 session row missing `directory`"))?;
let parent = set_column(row, "parent_id");
let fork = set_column(row, "fork_session_id");
let parent_message_id = fork
.and(row.get("fork_boundary"))
.filter(|boundary| boundary.get("type").and_then(Value::as_str) == Some("through"))
.and_then(|boundary| boundary.get("messageID"))
.and_then(Value::as_str)
.map(ToOwned::to_owned);
let source_agent = match (parent, set_column(row, "agent")) {
(Some(_), Some(agent)) => format!("{NAME}/{agent}"),
(Some(_), None) => format!("{NAME}/subagent"),
(None, _) => NAME.to_owned(),
};
let mut options = opencode_raw(row);
mark_source(&mut options, "store", json!(table));
Ok(Session {
id,
parent_session_id: parent.or(fork).map(ToOwned::to_owned),
parent_message_id,
source_agent,
created_at,
project,
options,
})
}
fn mark_source(options: &mut ProviderOptions, key: &str, value: Value) {
if let Some(Value::Object(source)) = options.get_mut("source") {
source.insert(key.to_owned(), value);
}
}
type V2Row = (String, String, i64, String);
fn v2_row(row: &rusqlite::Row) -> rusqlite::Result<V2Row> {
Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?))
}
fn read_v2_messages(
conn: &Connection,
db_path: &Path,
session_id: &str,
session_anchor: DateTime<Utc>,
copies: Option<&MigrationCopies>,
tx: &mpsc::Sender<Result<AdapterYield, AdapterError>>,
) -> bool {
let rows = (|| -> rusqlite::Result<Vec<rusqlite::Result<V2Row>>> {
let mut stmt = conn.prepare_cached(
"SELECT id, type, seq, data FROM session_message \
WHERE session_id = ?1 AND seq >= ?2 ORDER BY seq",
)?;
let rows = stmt.query_map(
rusqlite::params![session_id, MigrationCopies::live_from(copies)],
v2_row,
)?;
Ok(rows.collect())
})();
let rows = match rows {
Ok(rows) => rows,
Err(error) => {
emit!(tx, Err(db_error(db_path, "read v2 messages", &error)));
return true;
}
};
let open_turn = rows
.iter()
.flatten()
.filter(|(_, kind, ..)| kind == "idle")
.map(|(_, _, seq, ..)| *seq)
.max()
.unwrap_or(i64::MIN);
for row in rows {
let (id, kind, seq, data) = match row {
Ok(row) => row,
Err(error) => {
emit!(tx, Err(db_error(db_path, "read v2 message row", &error)));
continue;
}
};
if copies.is_some_and(|copies| copies.contains(&id, seq)) {
continue;
}
let record = match reconstruct_record(
&data,
[("id", json!(id)), ("type", json!(kind))],
db_path,
session_id,
&id,
) {
Ok(record) => record,
Err(error) => {
emit!(tx, Err(error));
continue;
}
};
if !is_settled(&kind, &record) && (seq > open_turn || kind == "shell") {
continue;
}
for event in build_v2_message_events(session_id, &id, &record, seq, session_anchor) {
emit!(tx, Ok(AdapterYield::Event(event)));
}
}
true
}
fn v2_tool_items(record: &Value) -> impl Iterator<Item = &Value> {
record
.get("content")
.and_then(Value::as_array)
.into_iter()
.flatten()
.filter(|item| item.get("type").and_then(Value::as_str) == Some("tool"))
}
fn tool_result_time(item: &Value) -> Option<DateTime<Utc>> {
let status = item.pointer("/state/status").and_then(Value::as_str);
if !matches!(status, Some("completed" | "error")) {
return None;
}
millis_at(item, &["time", "completed"])
.or_else(|| millis_at(item, &["time", "ran"]))
.or_else(|| millis_at(item, &["time", "created"]))
}
fn build_v2_message_events(
session_id: &str,
message_id: &str,
record: &Value,
seq: i64,
session_anchor: DateTime<Utc>,
) -> Vec<IngestEvent> {
let timestamp = millis_at(record, &["time", "created"]).unwrap_or(session_anchor);
let mut options = opencode_raw(record);
mark_source(&mut options, "seq", json!(seq));
let user = |options| Message::User {
id: message_id.to_owned(),
session_id: session_id.to_owned(),
timestamp,
options,
};
let carrier = |options, content| Message::System {
id: message_id.to_owned(),
session_id: session_id.to_owned(),
timestamp,
content,
options,
};
let mut parts = V2Parts::new(session_id, message_id);
let message = match record.get("type").and_then(Value::as_str) {
Some("user") => {
for skill in record
.get("skills")
.and_then(Value::as_array)
.into_iter()
.flatten()
{
parts.text(Provenance::Injected, skill, "text");
}
if record
.get("text")
.and_then(Value::as_str)
.is_some_and(|text| !text.is_empty())
{
parts.text(Provenance::Conversational, record, "text");
}
for file in record
.get("files")
.and_then(Value::as_array)
.into_iter()
.flatten()
{
parts.file(file);
}
user(options)
}
Some("synthetic" | "skill") => {
parts.text(Provenance::Injected, record, "text");
user(options)
}
Some("shell") => {
parts.text(Provenance::Conversational, record, "command");
let mut events = vec![IngestEvent::Message(user(options))];
events.extend(parts.parts.into_iter().map(IngestEvent::Part));
if let Some(ended) = millis_at(record, &["time", "completed"]) {
events.push(IngestEvent::Message(Message::System {
id: format!("{message_id}/output"),
session_id: session_id.to_owned(),
timestamp: ended.max(timestamp),
content: record
.get("output")
.and_then(|output| extract_str(output, "output")),
options: synthetic_options(),
}));
}
return events;
}
Some("compaction") if record.get("summary").is_some() => {
parts.text(Provenance::Injected, record, "summary");
if record
.get("recent")
.and_then(Value::as_str)
.is_some_and(|recent| !recent.is_empty())
{
parts.text(Provenance::Injected, record, "recent");
}
user(options)
}
Some("system") => carrier(options, extract_str(record, "text")),
Some("assistant") => {
let mut results = Vec::new();
for item in record
.get("content")
.and_then(Value::as_array)
.into_iter()
.flatten()
{
match item.get("type").and_then(Value::as_str) {
Some("text") => parts.text(Provenance::Conversational, item, "text"),
Some("reasoning") => parts.push(
Provenance::Conversational,
PartKind::Reasoning {
text: extract_str(item, "text"),
},
),
Some("tool") => {
if let Some(result) = parts.tool(item, timestamp) {
results.push(result);
}
}
_ => parts.push(Provenance::Injected, PartKind::Text { text: None }),
}
}
let mut events = vec![IngestEvent::Message(Message::Assistant {
id: message_id.to_owned(),
session_id: session_id.to_owned(),
timestamp,
options,
})];
events.extend(parts.parts.into_iter().map(IngestEvent::Part));
for ToolSplit { message, result } in results {
events.push(IngestEvent::Message(message));
events.push(IngestEvent::Part(result));
}
return events;
}
_ => carrier(options, None),
};
let mut events = vec![IngestEvent::Message(message)];
events.extend(parts.parts.into_iter().map(IngestEvent::Part));
events
}
struct V2Parts<'a> {
session_id: &'a str,
message_id: &'a str,
parts: Vec<Part>,
}
impl<'a> V2Parts<'a> {
fn new(session_id: &'a str, message_id: &'a str) -> Self {
Self {
session_id,
message_id,
parts: Vec::new(),
}
}
fn next_id(&self) -> (String, i32) {
let ordinal = self.parts.len();
(part_id(self.message_id, ordinal), part_ordinal(ordinal))
}
fn push(&mut self, provenance: Provenance, kind: PartKind) {
let (id, ordinal) = self.next_id();
self.parts.push(Part {
session_id: self.session_id.to_owned(),
id,
message_id: self.message_id.to_owned(),
ordinal,
provenance,
options: ProviderOptions::new(),
kind,
});
}
fn text(&mut self, provenance: Provenance, source: &Value, key: &str) {
self.push(
provenance,
PartKind::Text {
text: extract_str(source, key),
},
);
}
fn file(&mut self, file: &Value) {
let data = match (
file.get("data").and_then(Value::as_str),
file.pointer("/source/uri").and_then(Value::as_str),
) {
(Some(data), _) if !data.is_empty() => FileData::String(data.to_owned()),
(_, Some(uri)) => FileData::Url(uri.to_owned()),
_ => FileData::String(compact_json(file)),
};
self.push(
Provenance::Conversational,
PartKind::File {
media_type: file
.get("mime")
.and_then(Value::as_str)
.map(ToOwned::to_owned),
file_name: file
.get("name")
.and_then(Value::as_str)
.map(ToOwned::to_owned),
data,
},
);
}
fn tool(&mut self, item: &Value, message_ts: DateTime<Utc>) -> Option<ToolSplit> {
let (id, _) = self.next_id();
let call_id = extract_str(item, "id");
let name = extract_str(item, "name");
let state = item.get("state");
let status = state
.and_then(|state| state.get("status"))
.and_then(Value::as_str);
self.push(
Provenance::Conversational,
PartKind::ToolCall {
call_id: call_id.clone(),
name: name.clone(),
params: state
.and_then(|state| state.get("input"))
.cloned()
.unwrap_or(Value::Null),
provider_executed: item.get("executed").and_then(Value::as_bool) == Some(true),
},
);
let result_ts = tool_result_time(item)?.max(message_ts);
let result = match status {
Some("error") => state.and_then(|state| state.get("error")),
_ => state.and_then(|state| state.get("content")),
};
let tool_message_id = format!("{id}/result");
Some(ToolSplit {
message: Message::Tool {
id: tool_message_id.clone(),
session_id: self.session_id.to_owned(),
timestamp: result_ts,
options: synthetic_options(),
},
result: Part {
session_id: self.session_id.to_owned(),
id: part_id(&tool_message_id, 0),
message_id: tool_message_id,
ordinal: 0,
provenance: Provenance::Injected,
options: synthetic_options(),
kind: PartKind::ToolResult {
call_id,
name,
is_failure: status == Some("error"),
result: result.cloned().unwrap_or(Value::Null),
},
},
})
}
}
fn record_location(db_path: &Path, session_id: &str, record_id: &str) -> String {
format!(
"{}::session={session_id}::record={record_id}",
db_path.display()
)
}
fn read_json(path: &Path) -> Result<Value, AdapterError> {
let io = |source| AdapterError::io(NAME, path.display().to_string(), source);
let len = std::fs::metadata(path).map_err(io)?.len();
if len > RECORD_CAP as u64 {
return Err(AdapterError::schema(
NAME,
path.display().to_string(),
format!("record data exceeds adapter record cap: {len} bytes > {RECORD_CAP}"),
));
}
let bytes = std::fs::read(path).map_err(io)?;
parse_bounded(NAME, &bytes, || path.display().to_string())
}
fn list_json_sorted(dir: &Path) -> Result<Vec<PathBuf>, AdapterError> {
list_files_sorted(dir, "json", |_| true)
}
fn session_from_value(value: &Value, path: &Path) -> Result<Session, AdapterError> {
let display = path.display().to_string();
let id = value
.get("id")
.and_then(Value::as_str)
.ok_or_else(|| AdapterError::schema(NAME, display.clone(), "session missing `id`"))?
.to_owned();
let created_at = millis_at(value, &["time", "created"]).ok_or_else(|| {
AdapterError::schema(NAME, display.clone(), "session missing `time.created`")
})?;
let project = extract_str(value, "directory")
.ok_or_else(|| AdapterError::schema(NAME, display, "session missing `directory`"))?;
let options = opencode_raw(value);
Ok(Session {
id,
parent_session_id: value
.get("parentID")
.and_then(Value::as_str)
.map(ToOwned::to_owned),
parent_message_id: None,
source_agent: NAME.to_owned(),
created_at,
project,
options,
})
}
fn build_message_events(
session_id: &str,
message_value: &Value,
part_values: &[Value],
default_timestamp: DateTime<Utc>,
) -> Result<Vec<IngestEvent>, AdapterError> {
let message_id = message_value
.get("id")
.and_then(Value::as_str)
.ok_or_else(|| AdapterError::schema(NAME, session_id.to_owned(), "message missing `id`"))?;
let role = message_value.get("role").and_then(Value::as_str);
let timestamp = millis_at(message_value, &["time", "created"]).unwrap_or(default_timestamp);
let options = opencode_raw(message_value);
let message = match role {
Some("user") => Message::User {
id: message_id.to_owned(),
session_id: session_id.to_owned(),
timestamp,
options,
},
Some("assistant") => Message::Assistant {
id: message_id.to_owned(),
session_id: session_id.to_owned(),
timestamp,
options,
},
_ => Message::System {
id: message_id.to_owned(),
session_id: session_id.to_owned(),
timestamp,
content: extract_str(message_value, "role"),
options,
},
};
let mut events = vec![IngestEvent::Message(message)];
let mut deferred = Vec::new();
for (ordinal, part_value) in part_values.iter().enumerate() {
let mapped = map_part(session_id, message_id, ordinal, part_value, timestamp)?;
events.push(IngestEvent::Part(mapped.part));
if let Some(split) = mapped.tool_split {
deferred.push(split);
}
}
for ToolSplit {
message: tool_message,
result,
} in deferred
{
events.push(IngestEvent::Message(tool_message));
events.push(IngestEvent::Part(result));
}
Ok(events)
}
struct MappedPart {
part: Part,
tool_split: Option<ToolSplit>,
}
struct ToolSplit {
message: Message,
result: Part,
}
fn map_part(
session_id: &str,
message_id: &str,
ordinal: usize,
value: &Value,
message_ts: DateTime<Utc>,
) -> Result<MappedPart, AdapterError> {
let kind = value.get("type").and_then(Value::as_str);
let id = value
.get("id")
.and_then(Value::as_str)
.ok_or_else(|| AdapterError::schema(NAME, message_id.to_owned(), "part missing `id`"))?
.to_owned();
if kind == Some("tool") {
return Ok(tool_part(
session_id, message_id, &id, ordinal, value, message_ts,
));
}
let (provenance, part_kind) = match kind {
Some("text") => (text_provenance(value), text_kind(value)),
Some("reasoning") => (Provenance::Conversational, reasoning_kind(value)),
Some("file") => (Provenance::Conversational, file_kind(value)),
_ => (Provenance::Injected, PartKind::Text { text: None }),
};
Ok(MappedPart {
part: Part {
session_id: session_id.to_owned(),
id,
message_id: message_id.to_owned(),
ordinal: part_ordinal(ordinal),
provenance,
options: opencode_raw(value),
kind: part_kind,
},
tool_split: None,
})
}
fn text_provenance(value: &Value) -> Provenance {
if value.get("synthetic").and_then(Value::as_bool) == Some(true) {
Provenance::Injected
} else {
Provenance::Conversational
}
}
fn text_kind(value: &Value) -> PartKind {
PartKind::Text {
text: extract_str(value, "text"),
}
}
fn reasoning_kind(value: &Value) -> PartKind {
PartKind::Reasoning {
text: extract_str(value, "text"),
}
}
fn file_kind(value: &Value) -> PartKind {
let media_type = value
.get("mime")
.and_then(Value::as_str)
.map(ToOwned::to_owned);
let file_name = value
.get("filename")
.and_then(Value::as_str)
.map(ToOwned::to_owned);
let data = match value.get("url").and_then(Value::as_str) {
Some(url) => FileData::Url(url.to_owned()),
None => FileData::String(compact_json(value)),
};
PartKind::File {
media_type,
file_name,
data,
}
}
fn tool_part(
session_id: &str,
message_id: &str,
id: &str,
ordinal: usize,
value: &Value,
message_ts: DateTime<Utc>,
) -> MappedPart {
let call_id = extract_str(value, "callID");
let name = extract_str(value, "tool");
let state = value.get("state");
let status = state.and_then(|s| s.get("status")).and_then(Value::as_str);
let result_ts = millis_at(value, &["state", "time", "end"]).unwrap_or(message_ts);
let mut owned_state = state.cloned().unwrap_or(Value::Null);
let (input, result) = match owned_state.as_object_mut() {
Some(map) => {
let input = map.remove("input").unwrap_or(Value::Null);
let result = map
.remove("output")
.or_else(|| map.remove("error"))
.unwrap_or_else(|| {
std::mem::take(&mut owned_state)
});
(input, result)
}
None => (Value::Null, Value::Null),
};
let tool_call = Part {
session_id: session_id.to_owned(),
id: id.to_owned(),
message_id: message_id.to_owned(),
ordinal: part_ordinal(ordinal),
provenance: Provenance::Conversational,
options: opencode_raw(value),
kind: PartKind::ToolCall {
call_id: call_id.clone(),
name: name.clone(),
params: input,
provider_executed: false,
},
};
let tool_message_id = format!("{id}/result");
let tool_message = Message::Tool {
id: tool_message_id.clone(),
session_id: session_id.to_owned(),
timestamp: result_ts,
options: synthetic_options(),
};
let result_part = Part {
session_id: session_id.to_owned(),
id: part_id(&tool_message_id, 0),
message_id: tool_message_id,
ordinal: 0,
provenance: Provenance::Injected,
options: synthetic_options(),
kind: PartKind::ToolResult {
call_id,
name,
is_failure: status == Some("error"),
result,
},
};
MappedPart {
part: tool_call,
tool_split: Some(ToolSplit {
message: tool_message,
result: result_part,
}),
}
}
#[inline]
fn opencode_raw(value: &Value) -> ProviderOptions {
source_options(NAME, value)
}
fn synthetic_options() -> ProviderOptions {
let mut options = ProviderOptions::new();
options.insert("opencode".to_owned(), json!({ "synthetic": true }));
options
}
fn millis_at(value: &Value, path: &[&str]) -> Option<DateTime<Utc>> {
let mut cursor = value;
for key in path {
cursor = cursor.get(key)?;
}
DateTime::from_timestamp_millis(cursor.as_i64()?)
}
fn is_synthetic(options: &ProviderOptions) -> bool {
options
.get("opencode")
.and_then(|o| o.get("synthetic"))
.and_then(Value::as_bool)
== Some(true)
}
fn serialize_native(
session: &crate::sessions::SessionWithMessages,
) -> Result<Vec<RestoredFile>, AdapterError> {
let Some(session_raw) = raw_record(&session.session.options) else {
return serialize_foreign(session);
};
if session_store(&session.session.options).is_some() {
return serialize_native_v2(session, &session_raw);
}
let mut ordered = session.messages.clone();
ordered.sort_by(by_timestamp_then_id);
let mut messages = Vec::with_capacity(ordered.len());
for message in &ordered {
if is_synthetic(message.message.options()) {
continue;
}
if message_seq(message.message.options()).is_some() {
return serialize_foreign(session);
}
let Some(info) = raw_record(message.message.options()) else {
return serialize_foreign(session);
};
let mut parts = Vec::with_capacity(message.parts.len());
for part in &message.parts {
if let Some(raw) = raw_record(&part.options) {
parts.push(raw);
}
}
messages.push(json!({ "info": info, "parts": parts }));
}
import_file(
&session.session.id,
session_raw,
messages,
RestoreFidelity::Native,
)
}
fn session_store(options: &ProviderOptions) -> Option<&str> {
options.get("source")?.get("store")?.as_str()
}
fn message_seq(options: &ProviderOptions) -> Option<i64> {
options.get("source")?.get("seq")?.as_i64()
}
fn serialize_native_v2(
session: &crate::sessions::SessionWithMessages,
row: &Value,
) -> Result<Vec<RestoredFile>, AdapterError> {
let mut records = Vec::with_capacity(session.messages.len());
for message in &session.messages {
if is_synthetic(message.message.options()) {
continue;
}
let (Some(raw), Some(seq)) = (
raw_record(message.message.options()),
message_seq(message.message.options()),
) else {
return serialize_foreign(session);
};
records.push((seq, raw));
}
records.sort_by_key(|(seq, _)| *seq);
let messages = records.into_iter().map(|(_, raw)| raw).collect();
import_file(
&session.session.id,
v2_session_info(row),
messages,
RestoreFidelity::Native,
)
}
fn v2_session_info(row: &Value) -> Value {
let column = |name: &str| row.get(name).filter(|value| !value.is_null());
let truthy = |name: &str| {
column(name).filter(|value| {
!matches!(value, Value::String(text) if text.is_empty()) && value.as_f64() != Some(0.0)
})
};
let mut info = serde_json::Map::new();
let mut set = |key: &str, value: Option<Value>| {
if let Some(value) = value {
info.insert(key.to_owned(), value);
}
};
set("id", column("id").cloned());
set("projectID", column("project_id").cloned());
set("title", column("title").cloned());
set("parentID", truthy("parent_id").cloned());
if let (Some(session_id), Some(boundary)) = (truthy("fork_session_id"), truthy("fork_boundary"))
{
set(
"fork",
Some(json!({ "sessionID": session_id, "boundary": boundary })),
);
}
set("agent", truthy("agent").cloned());
set(
"model",
truthy("model").map(|model| {
let mut model = model.clone();
if let Value::Object(map) = &mut model {
map.entry("variant").or_insert_with(|| json!("default"));
}
model
}),
);
set("cost", column("cost").cloned());
let token = |name: &str| column(name).cloned().unwrap_or(Value::Null);
set(
"tokens",
Some(json!({
"input": token("tokens_input"),
"output": token("tokens_output"),
"reasoning": token("tokens_reasoning"),
"cache": { "read": token("tokens_cache_read"), "write": token("tokens_cache_write") },
})),
);
let mut location = serde_json::Map::new();
if let Some(directory) = column("directory") {
location.insert("directory".to_owned(), directory.clone());
}
if let Some(workspace) = column("workspace_id") {
location.insert("workspaceID".to_owned(), workspace.clone());
}
set("location", Some(Value::Object(location)));
set("subpath", truthy("path").cloned());
set("metadata", column("metadata").cloned());
set("permissions", column("permission").cloned());
set("revert", truthy("revert").map(current_revert));
set("outcome", column("idle_outcome").cloned());
let mut time = serde_json::Map::new();
for (key, name) in [
("created", "time_created"),
("updated", "time_updated"),
("idle", "time_idle"),
("viewed", "time_viewed"),
("archived", "time_archived"),
] {
let value = if name == "time_archived" {
truthy(name)
} else {
column(name)
};
if let Some(value) = value {
time.insert(key.to_owned(), value.clone());
}
}
set("time", Some(Value::Object(time)));
Value::Object(info)
}
fn current_revert(revert: &Value) -> Value {
let files = revert.get("files").and_then(Value::as_array);
let legacy = revert.get("diff").is_some()
|| files.is_some_and(|files| files.iter().any(|file| file.get("path").is_some()));
if !legacy {
return revert.clone();
}
let mut current = serde_json::Map::new();
for key in ["messageID", "partID", "snapshot"] {
if let Some(value) = revert.get(key).filter(|value| !value.is_null()) {
current.insert(key.to_owned(), value.clone());
}
}
if let Some(files) = files {
let files = files
.iter()
.map(|file| {
json!({
"file": file.get("path"),
"status": file.get("status"),
"additions": file.get("additions"),
"deletions": file.get("deletions"),
"patch": file.get("patch"),
})
})
.collect();
current.insert("files".to_owned(), Value::Array(files));
}
Value::Object(current)
}
fn serialize_foreign(
session: &crate::sessions::SessionWithMessages,
) -> Result<Vec<RestoredFile>, AdapterError> {
let created = session.session.created_at.timestamp_millis();
let info = json!({
"id": session.session.id,
"projectID": encode_project(&session.session.project),
"directory": &*session.session.project,
"time": { "created": created, "updated": created },
});
let mut ordered = session.messages.clone();
ordered.sort_by(by_timestamp_then_id);
let mut messages = Vec::with_capacity(ordered.len());
for message in &ordered {
let role = match message.message {
Message::User { .. } => "user",
Message::Assistant { .. } => "assistant",
Message::Tool { .. } | Message::System { .. } => continue,
};
let created = message.message.timestamp().timestamp_millis();
let msg_info = json!({
"id": message.message.id(),
"sessionID": session.session.id,
"role": role,
"time": { "created": created },
});
let mut parts = Vec::with_capacity(message.parts.len());
for part in &message.parts {
if let Some(record) = foreign_part(&session.session.id, part) {
parts.push(record);
}
}
messages.push(json!({ "info": msg_info, "parts": parts }));
}
import_file(
&session.session.id,
info,
messages,
RestoreFidelity::Foreign,
)
}
fn import_file(
session_id: &str,
info: Value,
messages: Vec<Value>,
fidelity: RestoreFidelity,
) -> Result<Vec<RestoredFile>, AdapterError> {
let body = json!({ "info": info, "messages": messages });
Ok(vec![RestoredFile::new(
PathBuf::from(format!("{session_id}.json")),
encode(&body, session_id)?,
fidelity,
)])
}
fn foreign_part(session_id: &str, part: &Part) -> Option<Value> {
let mut record = match &part.kind {
PartKind::Text { text } => json!({
"type": "text",
"text": text.as_deref().map(|t| &**t),
"synthetic": part.provenance == Provenance::Injected,
}),
PartKind::Reasoning { text } => json!({
"type": "reasoning",
"text": text.as_deref().map(|t| &**t),
}),
PartKind::File {
media_type,
file_name,
data,
} => json!({
"type": "file",
"mime": media_type,
"filename": file_name,
"url": match data {
FileData::Url(url) => Some(url.clone()),
_ => None,
},
}),
PartKind::ToolCall {
call_id,
name,
params,
..
} => json!({
"type": "tool",
"callID": call_id.as_deref().map(|c| &**c),
"tool": name.as_deref().map(|n| &**n),
"state": { "status": "completed", "input": params },
}),
_ => return None,
};
if let Value::Object(map) = &mut record {
map.insert("id".to_owned(), json!(part.id));
map.insert("sessionID".to_owned(), json!(session_id));
map.insert("messageID".to_owned(), json!(part.message_id));
}
Some(record)
}
fn encode(value: &Value, location: &str) -> Result<Vec<u8>, AdapterError> {
serde_json::to_vec(value).map_err(|error| {
AdapterError::schema(
NAME,
location.to_owned(),
format!("json encode failed: {error}"),
)
})
}
fn encode_project(project: &str) -> String {
project
.chars()
.map(|c| if c.is_ascii_alphanumeric() { c } else { '-' })
.collect()
}
#[cfg(test)]
mod tests {
#![expect(
clippy::expect_used,
clippy::unwrap_used,
reason = "tests fail by panicking"
)]
use super::*;
use crate::{
adapter::{SyncPlan, extract::LEAF_CAP, test_support::MaxWatermarkOracle},
handlers::{SyncEvent, SyncStatus, ingest_adapter},
sessions::Store,
wire::PartKind,
};
use tempfile::TempDir;
fn fixtures() -> std::path::PathBuf {
crate::adapter::test_support::manifest_dir().join("tests/fixtures/adapter/opencode/storage")
}
fn data_dir() -> std::path::PathBuf {
crate::adapter::test_support::manifest_dir().join("tests/fixtures/adapter/opencode")
}
fn db_fixture() -> std::path::PathBuf {
crate::adapter::test_support::manifest_dir()
.join("tests/fixtures/adapter/opencode/opencode.db")
}
const FRESH_SESSION_ID: &str = "ses_6405e5a5cffeIG2QHRuTmm4mA7";
const FRESH_MESSAGE_ID: &str = "msg_zzzzfresh0001";
const FRESH_PART_ID: &str = "prt_zzzzfresh0001";
const DB_SESSION_COUNT: usize = 10;
const CHILD_SESSION_ID: &str = "ses_09fbc7fc1ffeUaBz77QiNdJbXa";
const CHILD_PARENT_ID: &str = "ses_09fbc87f2ffeyYCZq51oN2HfGa";
const DOCTORED_SESSION_ID: &str = "ses_09fbe676bffe9nYsSBi5xhBlaD";
const DOCTORED_MESSAGE_ID: &str = "msg_f6041991e001BOYvwdP1iI0H0A";
const DOCTORED_DATA_CREATED_MS: i64 = 1_784_026_339_614;
const DOCTORED_COLUMN_CREATED_MS: i64 = 1_794_394_339_614;
fn db_data_dir(temp: &std::path::Path) -> anyhow::Result<PathBuf> {
let dir = temp.join("data");
std::fs::create_dir_all(&dir)?;
std::fs::copy(db_fixture(), dir.join("opencode.db"))?;
Ok(dir)
}
struct FixedOracle {
session_id: &'static str,
watermark_micros: i64,
}
impl crate::adapter::SkipOracle for FixedOracle {
fn session_max_ts(&self, session_id: &str) -> Option<i64> {
(session_id == self.session_id).then_some(self.watermark_micros)
}
}
#[test]
fn probe_default_finds_opencode_data_dir() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let data_dir = temp.path().join(".local").join("share").join("opencode");
std::fs::create_dir_all(data_dir.join("storage"))?;
let env = Env::with_home(temp.path());
let probe = OpencodeFactory.probe_default(&env);
let got = probe
.as_ref()
.and_then(|value| value.get("path"))
.and_then(Value::as_str);
assert_eq!(got, data_dir.to_str(), "probe must return the data dir");
std::fs::remove_dir_all(data_dir.join("storage"))?;
assert!(
OpencodeFactory.probe_default(&env).is_none(),
"an empty data dir must not be offered as a source",
);
Ok(())
}
fn expected_envelope_from_tree(tree_base: &std::path::Path, session_id: &str) -> Value {
let read = |path: &std::path::Path| -> Value {
let raw: Value =
serde_json::from_slice(&std::fs::read(path).expect("read fixture file"))
.expect("fixture file is JSON");
crate::adapter::extract::extract_raw_record(&raw)
};
let mut session_json = None;
for project in std::fs::read_dir(tree_base.join("session")).expect("session dir") {
let candidate = project
.expect("session project entry")
.path()
.join(format!("{session_id}.json"));
if candidate.exists() {
session_json = Some(read(&candidate));
break;
}
}
let session_json = session_json.expect("session file in fixture tree");
let mut message_files =
list_json_sorted(&tree_base.join("message").join(session_id)).expect("message dir");
message_files.sort_by(|a, b| {
let (ma, mb) = (read(a), read(b));
ma["time"]["created"]
.as_i64()
.cmp(&mb["time"]["created"].as_i64())
.then_with(|| ma["id"].as_str().cmp(&mb["id"].as_str()))
});
let mut messages = Vec::new();
for message_path in &message_files {
let msg = read(message_path);
let message_id = msg["id"].as_str().expect("message id").to_owned();
let part_files =
list_json_sorted(&tree_base.join("part").join(&message_id)).expect("part dir");
let parts: Vec<Value> = part_files.iter().map(|p| read(p)).collect();
messages.push(json!({ "info": msg, "parts": parts }));
}
json!({ "info": session_json, "messages": messages })
}
fn serialize_native_body(session: &crate::sessions::SessionWithMessages) -> Value {
let files = serialize_native(session).expect("native serialize");
assert_eq!(files.len(), 1, "one file per session");
let file = &files[0];
assert_eq!(file.actual_fidelity, RestoreFidelity::Native);
assert_eq!(
file.relative_path,
PathBuf::from(format!("{}.json", session.session.id)),
"native emits <session_id>.json at the root",
);
let body: Value = serde_json::from_slice(&file.bytes).expect("emitted body is JSON");
assert!(body.get("info").is_some(), "envelope carries session info");
assert!(
body.get("messages").and_then(Value::as_array).is_some(),
"envelope carries a messages array",
);
body
}
#[tokio::test(flavor = "multi_thread")]
async fn native_restore_emits_import_shape_from_tree() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let source = temp.path().join("storage");
copy_dir(&fixtures(), &source)?;
let store = Store::open_local(temp.path().join("store")).await?;
let adapter = OpencodeAdapter::new(&source);
ingest_adapter(&store, &adapter, &crate::adapter::NoopOracle, |_| {}).await?;
let session_ids = store.session_ids().await?;
assert!(!session_ids.is_empty(), "tree fixture ingests sessions");
for session_id in session_ids {
let session = store
.get_session(&session_id)
.await?
.expect("session round-trips");
let body = serialize_native_body(&session);
assert_eq!(
body,
expected_envelope_from_tree(&source, &session_id),
"tree session {session_id} embeds its raw_records into the envelope",
);
let real_ids: std::collections::HashSet<&str> = session
.messages
.iter()
.filter(|m| !is_synthetic(m.message.options()))
.map(|m| m.message.id())
.collect();
for message in body["messages"].as_array().unwrap() {
let id = message["info"]["id"].as_str().expect("message info id");
assert!(
real_ids.contains(id),
"emitted message {id} must be a real canonical message",
);
}
}
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn native_restore_conformance_against_db_fixture() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let data = db_data_dir(temp.path())?;
let store = Store::open_local(temp.path().join("store")).await?;
ingest_adapter(
&store,
&OpencodeAdapter::new(&data),
&crate::adapter::NoopOracle,
|_| {},
)
.await?;
let db = db_fixture();
let conn = open_db(&db)?;
let expected_session: HashMap<String, Value> = conn
.prepare(&format!("SELECT {SESSION_COLUMNS} FROM session"))?
.query_map([], session_info_from_row)?
.map(|row| {
let info = row.expect("session row reconstructs");
let id = info["id"].as_str().expect("session id").to_owned();
(id, crate::adapter::extract::extract_raw_record(&info))
})
.collect();
let reconstruct_expected = |data: &str, inject: Vec<(&'static str, Value)>| -> Value {
reconstruct_record(data, inject, std::path::Path::new(""), "", "").unwrap()
};
let mut file_count = 0usize;
let mut saw_child = false;
let mut saw_parent = false;
for session_id in store.session_ids().await? {
let session = store
.get_session(&session_id)
.await?
.expect("session round-trips");
let body = serialize_native_body(&session);
file_count += 1;
if session_id == CHILD_SESSION_ID {
saw_child = true;
}
if session_id == CHILD_PARENT_ID {
saw_parent = true;
}
assert_eq!(
&body["info"],
expected_session
.get(&session_id)
.unwrap_or_else(|| panic!("session {session_id} in DB")),
"session info equals the row-reconstructed SessionInfo",
);
let mut expected_msg_ids: Vec<String> = Vec::new();
let mut stmt =
conn.prepare("SELECT id, data FROM message WHERE session_id = ?1 ORDER BY id")?;
let msg_rows: Vec<(String, String)> = stmt
.query_map([&session_id], |row| Ok((row.get(0)?, row.get(1)?)))?
.collect::<rusqlite::Result<_>>()?;
for (message_id, data) in &msg_rows {
expected_msg_ids.push(message_id.clone());
let expected_info = reconstruct_expected(
data,
vec![("id", json!(message_id)), ("sessionID", json!(session_id))],
);
let entry = body["messages"]
.as_array()
.unwrap()
.iter()
.find(|m| m["info"]["id"].as_str() == Some(message_id.as_str()))
.unwrap_or_else(|| panic!("message {message_id} emitted"));
assert_eq!(&entry["info"], &expected_info, "message info matches row");
let mut pstmt =
conn.prepare("SELECT id, data FROM part WHERE message_id = ?1 ORDER BY id")?;
let expected_parts: Vec<Value> = pstmt
.query_map([message_id], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
})?
.map(|r| {
let (part_id, data) = r.unwrap();
reconstruct_expected(
&data,
vec![
("id", json!(part_id)),
("sessionID", json!(session_id)),
("messageID", json!(message_id)),
],
)
})
.collect();
assert_eq!(
entry["parts"].as_array().unwrap(),
&expected_parts,
"parts for message {message_id} match the rows in id order",
);
}
let emitted_ids: Vec<&str> = body["messages"]
.as_array()
.unwrap()
.iter()
.map(|m| m["info"]["id"].as_str().unwrap())
.collect();
assert_eq!(
emitted_ids.len(),
expected_msg_ids.len(),
"no synthetic message carriers leak into the envelope",
);
}
assert_eq!(file_count, DB_SESSION_COUNT, "one file per DB session");
assert!(
saw_child,
"child session {CHILD_SESSION_ID} gets its own file"
);
assert!(
saw_parent,
"parent session {CHILD_PARENT_ID} gets its own file",
);
Ok(())
}
async fn plan_gate_agreement(
adapter: &OpencodeAdapter,
) -> anyhow::Result<(SyncPlan, SyncPlan)> {
use tokio_stream::StreamExt;
let first_sync = adapter
.plan(&crate::adapter::NoopOracle)
.await?
.expect("opencode supports plan");
assert_eq!(
first_sync.pending, first_sync.sessions,
"a first sync reads every session"
);
assert_eq!(first_sync.fresh, 0);
let max_plan = adapter
.plan(&MaxWatermarkOracle)
.await?
.expect("opencode supports plan");
assert_eq!(
max_plan.sessions, first_sync.sessions,
"session count is stable across oracles",
);
let mut gate_fresh = 0usize;
let mut stream = adapter.events_with(&MaxWatermarkOracle);
while let Some(item) = stream.next().await {
match item? {
AdapterYield::Skipped {
reason: SkipReason::Fresh,
..
} => gate_fresh += 1,
AdapterYield::SkippedBatch {
reason: SkipReason::Fresh,
count,
} => gate_fresh += count,
_ => {}
}
}
assert_eq!(gate_fresh, max_plan.fresh, "plan and gate must agree");
Ok((first_sync, max_plan))
}
#[tokio::test(flavor = "multi_thread")]
async fn plan_matches_the_events_gate() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let source = temp.path().join("storage");
copy_dir(&fixtures(), &source)?;
let empty_dir = source.join("session").join("proj-empty");
std::fs::create_dir_all(&empty_dir)?;
std::fs::write(
empty_dir.join("ses_zzzzemptysession00000000.json"),
"{\"id\":\"ses_zzzzemptysession00000000\",\"projectID\":\"proj-empty\",\
\"directory\":\"/tmp/pond-test\",\"time\":{\"created\":1759859990000}}",
)?;
let adapter = OpencodeAdapter::new(&source);
let (first_sync, max_plan) = plan_gate_agreement(&adapter).await?;
assert!(first_sync.sessions > 1);
assert_eq!(
max_plan.pending, 1,
"the message-less session must stay pending - its Session row only \
lands by reading it",
);
Ok(())
}
const FRESH_TURN_MICROS: i64 = 1_759_859_999_000 * 1_000;
#[tokio::test(flavor = "multi_thread")]
async fn freshness_re_reads_a_session_that_gained_a_newer_message() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let source = temp.path().join("storage");
copy_dir(&fixtures(), &source)?;
let store = Store::open_local(temp.path().join("store")).await?;
let adapter = OpencodeAdapter::new(&source);
ingest_adapter(&store, &adapter, &crate::adapter::NoopOracle, |_| {}).await?;
append_fresh_opencode_turn(&source)?;
let oracle = FixedOracle {
session_id: FRESH_SESSION_ID,
watermark_micros: FRESH_TURN_MICROS - 1,
};
ingest_adapter(&store, &adapter, &oracle, |_| {}).await?;
let session = store
.get_session(FRESH_SESSION_ID)
.await?
.expect("fixture session round-trips");
let fresh = session
.messages
.iter()
.find(|stored| stored.message.id() == FRESH_MESSAGE_ID)
.expect("message newer than the watermark must land");
assert!(
fresh.parts.iter().any(|part| matches!(
&part.kind,
PartKind::Text { text } if text.as_deref().map(|value| value.as_str()) == Some("fresh opencode text")
)),
"fresh message part must land with the re-read session",
);
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn freshness_skips_a_session_not_newer_than_the_watermark() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let source = temp.path().join("storage");
copy_dir(&fixtures(), &source)?;
let store = Store::open_local(temp.path().join("store")).await?;
let adapter = OpencodeAdapter::new(&source);
ingest_adapter(&store, &adapter, &crate::adapter::NoopOracle, |_| {}).await?;
append_fresh_opencode_turn(&source)?;
let oracle = FixedOracle {
session_id: FRESH_SESSION_ID,
watermark_micros: FRESH_TURN_MICROS,
};
let summary = ingest_adapter(&store, &adapter, &oracle, |_| {}).await?;
assert!(
summary.skipped_fresh >= 1,
"the unchanged-vs-watermark session must be skipped, got {summary:?}",
);
let session = store
.get_session(FRESH_SESSION_ID)
.await?
.expect("fixture session round-trips");
assert!(
!session
.messages
.iter()
.any(|stored| stored.message.id() == FRESH_MESSAGE_ID),
"a skipped session must not re-read the appended turn",
);
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn malformed_part_file_drops_only_that_part() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let source = temp.path().join("storage");
write_minimal_session(&source, "ses_badpart", "msg_badpart")?;
let part_dir = source.join("part").join("msg_badpart");
std::fs::write(part_dir.join("prt_000_bad.json"), b"{not json")?;
write_json_file(
&part_dir.join("prt_999_good.json"),
&json!({
"id": "prt_999_good",
"sessionID": "ses_badpart",
"messageID": "msg_badpart",
"type": "text",
"text": "valid sibling survives",
"synthetic": false,
}),
)?;
let store = Store::open_local(temp.path().join("store")).await?;
let summary = ingest_adapter(
&store,
&OpencodeAdapter::new(&source),
&crate::adapter::NoopOracle,
|_| {},
)
.await?;
assert_eq!(summary.dropped_events, 1);
let session = store
.get_session("ses_badpart")
.await?
.expect("session with one malformed part still lands");
let message = session
.messages
.iter()
.find(|stored| stored.message.id() == "msg_badpart")
.expect("message with valid sibling part still lands");
assert!(message.parts.iter().any(|part| {
matches!(
&part.kind,
PartKind::Text { text }
if text.as_deref().map(String::as_str) == Some("valid sibling survives")
)
}));
Ok(())
}
#[test]
fn missing_message_timestamp_uses_session_anchor() -> anyhow::Result<()> {
let session_anchor =
DateTime::parse_from_rfc3339("2026-05-05T12:13:14Z")?.with_timezone(&Utc);
let events = build_message_events(
"ses_anchor",
&json!({"id": "msg_no_time", "role": "user"}),
&[],
session_anchor,
)?;
let IngestEvent::Message(message) = &events[0] else {
panic!("first event is the message");
};
assert_eq!(message.timestamp(), session_anchor);
Ok(())
}
#[test]
fn source_part_without_id_is_schema_error() {
let session_anchor = DateTime::from_timestamp_millis(1_765_000_000_000).unwrap();
let error = build_message_events(
"ses_missing_part_id",
&json!({
"id": "msg_missing_part_id",
"role": "assistant",
"time": { "created": 1_765_000_000_000i64 },
}),
&[json!({"type": "text", "text": "cannot restore its filename"})],
session_anchor,
)
.expect_err("part ids are required for native filename replay");
assert!(error.to_string().contains("part missing `id`"));
}
#[test]
fn read_json_bounds_oversized_string_leaves() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let path = temp.path().join("oversized.json");
write_json_file(
&path,
&json!({
"id": "oversized",
"text": "x".repeat(LEAF_CAP + 100),
}),
)?;
let value = read_json(&path)?;
let text = value
.get("text")
.and_then(Value::as_str)
.expect("text leaf survives as a bounded marker");
assert!(text.len() <= LEAF_CAP);
assert!(text.ends_with(&format!("{} bytes>", LEAF_CAP + 100)));
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn opencode_adapter_ingests_fixture_corpus_into_canonical_shape() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let store = Store::open_local(temp.path()).await?;
let adapter = OpencodeAdapter::new(data_dir());
let summary = ingest_adapter(&store, &adapter, &crate::adapter::NoopOracle, |_| {}).await?;
assert!(summary.accepted() > 0, "ingest must accept rows");
assert_eq!(summary.dropped_events, 0, "no per-event drops expected");
assert_eq!(
summary.dropped_sessions, 0,
"no session-level rejections expected"
);
let (sessions, messages, parts) = store.row_counts().await?;
assert!(
sessions >= DB_SESSION_COUNT,
"the DB sessions plus the tree beside them all ingest",
);
assert!(messages > 0, "at least one opencode message");
assert!(parts > 0, "at least one opencode Part");
let mut saw_call = false;
let mut saw_result = false;
let mut saw_injected_text = false;
for session_id in store.session_ids().await? {
let session = store
.get_session(&session_id)
.await?
.expect("session round-trips");
assert!(
session.session.source_agent.starts_with(NAME),
"source_agent is `opencode` or `opencode/<agent>`, got {}",
session.session.source_agent,
);
assert!(
!(*session.session.project).is_empty(),
"spec.md#model-project-non-empty",
);
for stored in &session.messages {
for part in &stored.parts {
match &part.kind {
PartKind::ToolCall { .. } => saw_call = true,
PartKind::ToolResult { .. } => saw_result = true,
PartKind::Text { .. } if part.provenance == Provenance::Injected => {
saw_injected_text = true;
}
_ => {}
}
}
}
}
assert!(saw_call, "fused tool parts yield ToolCall on the assistant");
assert!(
saw_result,
"fused tool parts split off a ToolResult on a Tool message",
);
assert!(
saw_injected_text,
"spec.md#model-part-provenance: synthetic text parts are injected",
);
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn fused_tool_part_splits_into_call_and_result() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let store = Store::open_local(temp.path()).await?;
let adapter = OpencodeAdapter::new(data_dir());
ingest_adapter(&store, &adapter, &crate::adapter::NoopOracle, |_| {}).await?;
let mut call_ids = std::collections::HashSet::new();
let mut result_ids = std::collections::HashSet::new();
let mut saw_failure = false;
for session_id in store.session_ids().await? {
let session = store
.get_session(&session_id)
.await?
.expect("session round-trips");
for stored in &session.messages {
for part in &stored.parts {
match &part.kind {
PartKind::ToolCall { call_id, .. } => {
if let Some(id) = call_id.as_deref() {
call_ids.insert(id.clone());
}
}
PartKind::ToolResult {
call_id,
is_failure,
result,
..
} => {
assert!(
matches!(stored.message, Message::Tool { .. }),
"a ToolResult must live on a Tool-role message",
);
if *is_failure {
saw_failure = true;
assert_ne!(
result,
&Value::Null,
"failed tool results must carry the source error/output payload",
);
}
if let Some(id) = call_id.as_deref() {
result_ids.insert(id.clone());
}
}
_ => {}
}
}
}
}
assert!(!call_ids.is_empty(), "corpus has tool calls");
assert_eq!(
call_ids, result_ids,
"every tool call's id is matched by its split-off result",
);
assert!(
saw_failure,
"fixture has at least one failed opencode tool result"
);
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn db_fixture_ingests_expected_census() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let data = db_data_dir(temp.path())?;
let store = Store::open_local(temp.path().join("store")).await?;
let summary = ingest_adapter(
&store,
&OpencodeAdapter::new(&data),
&crate::adapter::NoopOracle,
|_| {},
)
.await?;
assert_eq!(summary.dropped_events, 0, "no per-event drops expected");
assert_eq!(summary.dropped_sessions, 0, "no session-level rejections");
let (sessions, messages, parts) = store.row_counts().await?;
assert_eq!(sessions, DB_SESSION_COUNT, "every DB session ingests");
assert!(
messages >= 27,
"27 message rows plus synthetic tool carriers"
);
assert!(parts >= 69, "69 part rows plus split-off tool results");
assert!(
store.get_session(CHILD_SESSION_ID).await?.is_some(),
"child session ingests",
);
assert!(
store.get_session(DOCTORED_SESSION_ID).await?.is_some(),
"migration-stamped session ingests",
);
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn child_session_labeled_as_subagent() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let data = db_data_dir(temp.path())?;
let store = Store::open_local(temp.path().join("store")).await?;
ingest_adapter(
&store,
&OpencodeAdapter::new(&data),
&crate::adapter::NoopOracle,
|_| {},
)
.await?;
let child = store
.get_session(CHILD_SESSION_ID)
.await?
.expect("child session ingests");
assert_eq!(
child.session.parent_session_id.as_deref(),
Some(CHILD_PARENT_ID),
"child carries its parent pointer",
);
assert_eq!(
child.session.source_agent, "opencode/general",
"spec.md: a subagent session is labeled opencode/<agent>",
);
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn doctored_row_uses_data_time_not_column() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let data = db_data_dir(temp.path())?;
let store = Store::open_local(temp.path().join("store")).await?;
ingest_adapter(
&store,
&OpencodeAdapter::new(&data),
&crate::adapter::NoopOracle,
|_| {},
)
.await?;
let session = store
.get_session(DOCTORED_SESSION_ID)
.await?
.expect("doctored session ingests");
let message = session
.messages
.iter()
.find(|stored| stored.message.id() == DOCTORED_MESSAGE_ID)
.expect("doctored message ingests");
assert_eq!(
message.message.timestamp().timestamp_millis(),
DOCTORED_DATA_CREATED_MS,
"canonical timestamp is data.time.created, not the migration column",
);
let conn = open_db(&db_fixture())?;
let watermark = db_session_watermark(&conn, &db_fixture(), DOCTORED_SESSION_ID)?
.expect("session has a message");
assert!(
watermark < DOCTORED_COLUMN_CREATED_MS * 1_000,
"watermark ({watermark} micros) must ignore the future time_created column",
);
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn injected_carrier_part_types_land() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let data = db_data_dir(temp.path())?;
{
let conn = Connection::open(data.join("opencode.db"))?;
conn.execute_batch("PRAGMA foreign_keys = OFF;")?;
insert_session(
&conn,
"ses_carrier00000000000000000",
"/tmp/carrier",
None,
None,
1_784_100_000_000,
)?;
insert_message(
&conn,
"msg_carrier00000000000000000",
"ses_carrier00000000000000000",
1_784_100_000_100,
&json!({ "role": "assistant", "time": { "created": 1_784_100_000_100i64 } }),
)?;
for (part_id, body) in [
(
"prt_carrier_a_subtask000000",
json!({ "type": "subtask", "prompt": "p", "description": "d", "agent": "general" }),
),
(
"prt_carrier_b_compaction00",
json!({ "type": "compaction", "auto": true }),
),
(
"prt_carrier_c_agent0000000",
json!({ "type": "agent", "name": "build" }),
),
(
"prt_carrier_d_snapshot0000",
json!({ "type": "snapshot", "snapshot": "abcdef" }),
),
] {
insert_part(
&conn,
part_id,
"msg_carrier00000000000000000",
"ses_carrier00000000000000000",
1_784_100_000_100,
&body,
)?;
}
}
let store = Store::open_local(temp.path().join("store")).await?;
ingest_adapter(
&store,
&OpencodeAdapter::new(&data),
&crate::adapter::NoopOracle,
|_| {},
)
.await?;
let session = store
.get_session("ses_carrier00000000000000000")
.await?
.expect("carrier session ingests");
let mut seen_types = std::collections::HashSet::new();
for stored in &session.messages {
for part in &stored.parts {
let Some(raw) = raw_record(&part.options) else {
continue;
};
if let Some(kind) = raw.get("type").and_then(Value::as_str) {
seen_types.insert(kind.to_owned());
assert_eq!(
part.provenance,
Provenance::Injected,
"carrier part {kind} is injected turn machinery",
);
}
}
}
for expected in ["subtask", "compaction", "agent", "snapshot"] {
assert!(
seen_types.contains(expected),
"carrier type {expected} must land with its raw_record, saw {seen_types:?}",
);
}
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn dual_source_dedup_prefers_db() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let data = db_data_dir(temp.path())?;
let overlap = "ses_09fb956e1ffeKLKCXceMNLRsS0";
let tree_only = "ses_treeonly0000000000000000";
write_tree_session(&data.join("storage"), overlap, "TREE COPY SHOULD LOSE")?;
write_tree_session(&data.join("storage"), tree_only, "tree-only survives")?;
let store = Store::open_local(temp.path().join("store")).await?;
let summary = ingest_adapter(
&store,
&OpencodeAdapter::new(&data),
&crate::adapter::NoopOracle,
|_| {},
)
.await?;
assert!(
summary.skipped_superseded >= 1,
"the deduped tree copy must be counted as superseded, got {summary:?}",
);
assert_eq!(
summary.skipped_empty, 0,
"the dedup must not fold into skipped_empty, got {summary:?}",
);
let overlapped = store
.get_session(overlap)
.await?
.expect("overlapping session ingests (DB copy)");
assert!(
!overlapped.messages.iter().any(|stored| stored.parts.iter().any(|part| matches!(
&part.kind,
PartKind::Text { text } if text.as_deref().map(String::as_str) == Some("TREE COPY SHOULD LOSE")
))),
"the DB copy wins; the tree body must not leak in",
);
assert!(
store.get_session(tree_only).await?.is_some(),
"the tree-only session still ingests",
);
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn hostile_db_session_id_is_rejected_at_ingest() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let data = db_data_dir(temp.path())?;
let conn = Connection::open(data.join("opencode.db"))?;
conn.pragma_update(None, "foreign_keys", "OFF")?;
conn.execute(
"UPDATE session SET id = '../evil' WHERE id = ?1",
[CHILD_SESSION_ID],
)?;
drop(conn);
let store = Store::open_local(temp.path().join("store")).await?;
let skip_reasons = std::sync::Mutex::new(Vec::new());
let summary = ingest_adapter(
&store,
&OpencodeAdapter::new(&data),
&crate::adapter::NoopOracle,
|event| {
if let SyncEvent::SessionDone(outcome) = &event
&& let SyncStatus::Skipped { reason } = &outcome.status
{
skip_reasons.lock().unwrap().push(reason.clone());
}
},
)
.await?;
let skip_reasons = skip_reasons.into_inner().unwrap();
assert!(
skip_reasons
.iter()
.any(|reason| reason.contains("traversal")),
"the hostile id must surface as a typed skip, got {skip_reasons:?}",
);
assert_eq!(summary.skipped_files, 1, "got {summary:?}");
assert!(store.get_session("../evil").await?.is_none());
let (sessions, ..) = store.row_counts().await?;
assert_eq!(
sessions,
DB_SESSION_COUNT - 1,
"every well-formed session still ingests",
);
Ok(())
}
#[test]
fn oversized_json_file_is_rejected_by_the_record_cap() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let path = temp.path().join("big.json");
std::fs::write(&path, vec![b' '; RECORD_CAP + 1])?;
let error = read_json(&path).unwrap_err();
assert!(
error.to_string().contains("record cap"),
"the pre-read size gate must reject the file, got: {error}",
);
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn config_path_ending_in_storage_reads_parent_db() -> anyhow::Result<()> {
let temp = TempDir::new()?;
std::fs::copy(db_fixture(), temp.path().join("opencode.db"))?;
let store = Store::open_local(temp.path().join("store")).await?;
let adapter = OpencodeAdapter::new(temp.path().join("storage"));
ingest_adapter(&store, &adapter, &crate::adapter::NoopOracle, |_| {}).await?;
let (sessions, ..) = store.row_counts().await?;
assert_eq!(
sessions, DB_SESSION_COUNT,
"the normalized parent's DB is read",
);
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn plan_matches_the_events_gate_db_source() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let data = db_data_dir(temp.path())?;
let adapter = OpencodeAdapter::new(&data);
let (first, max_plan) = plan_gate_agreement(&adapter).await?;
assert_eq!(first.sessions, DB_SESSION_COUNT);
assert_eq!(
max_plan.fresh, DB_SESSION_COUNT,
"every DB session is fresh under a max watermark",
);
Ok(())
}
const APPEND_SESSION_ID: &str = "ses_09fb956e1ffeKLKCXceMNLRsS0";
const APPEND_MESSAGE_ID: &str = "msg_zzzzfreshdb0000000000000";
const APPEND_CREATED_MS: i64 = 1_790_000_000_000;
const APPEND_MICROS: i64 = APPEND_CREATED_MS * 1_000;
fn append_db_turn(data: &std::path::Path) -> anyhow::Result<()> {
let conn = Connection::open(data.join("opencode.db"))?;
insert_message(
&conn,
APPEND_MESSAGE_ID,
APPEND_SESSION_ID,
APPEND_CREATED_MS,
&json!({ "role": "user", "time": { "created": APPEND_CREATED_MS } }),
)?;
insert_part(
&conn,
"prt_zzzzfreshdb0000000000000",
APPEND_MESSAGE_ID,
APPEND_SESSION_ID,
APPEND_CREATED_MS,
&json!({ "type": "text", "text": "fresh db text", "synthetic": false }),
)?;
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn freshness_db_re_reads_a_session_with_a_newer_message() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let data = db_data_dir(temp.path())?;
let store = Store::open_local(temp.path().join("store")).await?;
let adapter = OpencodeAdapter::new(&data);
ingest_adapter(&store, &adapter, &crate::adapter::NoopOracle, |_| {}).await?;
append_db_turn(&data)?;
let oracle = FixedOracle {
session_id: APPEND_SESSION_ID,
watermark_micros: APPEND_MICROS - 1,
};
ingest_adapter(&store, &adapter, &oracle, |_| {}).await?;
let session = store
.get_session(APPEND_SESSION_ID)
.await?
.expect("session round-trips");
assert!(
session
.messages
.iter()
.any(|stored| stored.message.id() == APPEND_MESSAGE_ID),
"a message newer than the watermark must land",
);
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn freshness_db_skips_a_session_not_newer_than_the_watermark() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let data = db_data_dir(temp.path())?;
let store = Store::open_local(temp.path().join("store")).await?;
let adapter = OpencodeAdapter::new(&data);
ingest_adapter(&store, &adapter, &crate::adapter::NoopOracle, |_| {}).await?;
append_db_turn(&data)?;
let oracle = FixedOracle {
session_id: APPEND_SESSION_ID,
watermark_micros: APPEND_MICROS,
};
let summary = ingest_adapter(&store, &adapter, &oracle, |_| {}).await?;
assert!(
summary.skipped_fresh >= 1,
"the unchanged-vs-watermark session must be skipped, got {summary:?}",
);
let session = store
.get_session(APPEND_SESSION_ID)
.await?
.expect("session round-trips");
assert!(
!session
.messages
.iter()
.any(|stored| stored.message.id() == APPEND_MESSAGE_ID),
"a skipped session must not re-read the appended turn",
);
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn malformed_db_data_drops_only_that_record() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let data = db_data_dir(temp.path())?;
{
let conn = Connection::open(data.join("opencode.db"))?;
conn.execute(
"UPDATE part SET data = '{not json' WHERE id = (SELECT id FROM part ORDER BY id LIMIT 1)",
[],
)?;
}
let store = Store::open_local(temp.path().join("store")).await?;
let summary = ingest_adapter(
&store,
&OpencodeAdapter::new(&data),
&crate::adapter::NoopOracle,
|_| {},
)
.await?;
assert_eq!(
summary.dropped_events, 1,
"only the malformed part is dropped, got {summary:?}",
);
let (sessions, ..) = store.row_counts().await?;
assert_eq!(sessions, DB_SESSION_COUNT, "all sessions still ingest");
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn legacy_tree_only_root_still_ingests() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let source = temp.path().join("storage");
copy_dir(&fixtures(), &source)?;
let store = Store::open_local(temp.path().join("store")).await?;
ingest_adapter(
&store,
&OpencodeAdapter::new(&source),
&crate::adapter::NoopOracle,
|_| {},
)
.await?;
let (sessions, ..) = store.row_counts().await?;
assert!(sessions > 0, "the bare legacy tree still ingests");
assert!(
store.get_session(FRESH_SESSION_ID).await?.is_some(),
"a known tree session ingests with no DB present",
);
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn foreign_serialization_emits_import_shape() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let origin_store = Store::open_local(temp.path().join("origin-store")).await?;
let origin = crate::adapter::PiCodingAgentAdapter::new(
crate::adapter::test_support::manifest_dir()
.join("tests/fixtures/adapter/pi-coding-agent/sessions"),
);
ingest_adapter(&origin_store, &origin, &crate::adapter::NoopOracle, |_| {}).await?;
let mut saw_tool = false;
for session_id in origin_store.session_ids().await? {
let session = origin_store
.get_session(&session_id)
.await?
.expect("fixture session is readable");
let mut files = OpencodeFactory.serialize(&session, RestoreFidelity::Foreign)?;
assert_eq!(files.len(), 1, "foreign emits one file per session");
let file = files.remove(0);
assert_eq!(file.actual_fidelity, RestoreFidelity::Foreign);
assert_eq!(
file.relative_path,
PathBuf::from(format!("{session_id}.json")),
"foreign emits <session_id>.json at the root",
);
let body: Value = serde_json::from_slice(&file.bytes)?;
crate::adapter::write_restored_files(
&temp.path().join("opencode-restore"),
vec![file],
)?;
assert_eq!(
body["info"]["id"].as_str(),
Some(session_id.as_str()),
"envelope info carries the session id",
);
assert!(
body["info"]["directory"].is_string(),
"foreign session info carries a directory",
);
let messages = body["messages"]
.as_array()
.expect("envelope carries a messages array");
for message in messages {
let role = message["info"]["role"].as_str().expect("message info role");
assert!(
role == "user" || role == "assistant",
"only user/assistant carriers survive foreign restore, got {role}",
);
assert_eq!(
message["info"]["sessionID"].as_str(),
Some(session_id.as_str()),
);
for part in message["parts"].as_array().expect("parts array") {
if part["type"].as_str() == Some("tool") {
saw_tool = true;
assert_eq!(
part["state"]["status"].as_str(),
Some("completed"),
"a re-fused tool part is completed",
);
assert!(
part["state"].get("input").is_some(),
"a re-fused tool part carries its input",
);
}
}
}
}
assert!(
saw_tool,
"the pi fixture's tool call re-fuses into a `tool` part",
);
Ok(())
}
#[test]
fn path_ids_reject_separators_and_traversal() {
let where_ = "session/project/session.json";
assert!(validate_path_id(NAME, "session id", "ses_safe", where_).is_ok());
assert!(validate_path_id(NAME, "session id", "../ses", where_).is_err());
assert!(validate_path_id(NAME, "session id", "/tmp/ses", where_).is_err());
assert!(validate_path_id(NAME, "message id", "msg/a", where_).is_err());
assert!(validate_path_id(NAME, "message id", "msg\\a", where_).is_err());
}
fn append_fresh_opencode_turn(root: &std::path::Path) -> anyhow::Result<()> {
let message_dir = root.join("message").join(FRESH_SESSION_ID);
let part_dir = root.join("part").join(FRESH_MESSAGE_ID);
std::fs::create_dir_all(&message_dir)?;
std::fs::create_dir_all(&part_dir)?;
std::fs::write(
message_dir.join(format!("{FRESH_MESSAGE_ID}.json")),
serde_json::to_vec(&json!({
"id": FRESH_MESSAGE_ID,
"sessionID": FRESH_SESSION_ID,
"role": "user",
"time": { "created": 1759859999000i64 }
}))?,
)?;
std::fs::write(
part_dir.join(format!("{FRESH_PART_ID}.json")),
serde_json::to_vec(&json!({
"id": FRESH_PART_ID,
"sessionID": FRESH_SESSION_ID,
"messageID": FRESH_MESSAGE_ID,
"type": "text",
"text": "fresh opencode text",
"synthetic": false
}))?,
)?;
Ok(())
}
fn write_minimal_session(
root: &std::path::Path,
session_id: &str,
message_id: &str,
) -> anyhow::Result<()> {
write_json_file(
&root
.join("session")
.join("project")
.join(format!("{session_id}.json")),
&json!({
"id": session_id,
"projectID": "project",
"directory": "/tmp/project",
"time": { "created": 1_765_000_000_000i64, "updated": 1_765_000_000_000i64 },
}),
)?;
write_json_file(
&root
.join("message")
.join(session_id)
.join(format!("{message_id}.json")),
&json!({
"id": message_id,
"sessionID": session_id,
"role": "assistant",
"time": { "created": 1_765_000_000_001i64 },
}),
)?;
std::fs::create_dir_all(root.join("part").join(message_id))?;
Ok(())
}
fn write_json_file(path: &std::path::Path, value: &Value) -> anyhow::Result<()> {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
std::fs::write(path, serde_json::to_vec(value)?)?;
Ok(())
}
fn copy_dir(from: &std::path::Path, to: &std::path::Path) -> anyhow::Result<()> {
std::fs::create_dir_all(to)?;
for entry in std::fs::read_dir(from)? {
let entry = entry?;
let source = entry.path();
let target = to.join(entry.file_name());
if entry.file_type()?.is_dir() {
copy_dir(&source, &target)?;
} else {
std::fs::copy(&source, &target)?;
}
}
Ok(())
}
fn insert_session(
conn: &Connection,
id: &str,
directory: &str,
parent: Option<&str>,
agent: Option<&str>,
created: i64,
) -> anyhow::Result<()> {
conn.execute(
"INSERT INTO session (id, project_id, parent_id, slug, directory, title, version, \
agent, time_created, time_updated) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?9)",
rusqlite::params![
id,
"proj-test",
parent,
"slug",
directory,
"title",
"1.0.0",
agent,
created
],
)?;
Ok(())
}
fn insert_message(
conn: &Connection,
id: &str,
session_id: &str,
created: i64,
data: &Value,
) -> anyhow::Result<()> {
conn.execute(
"INSERT INTO message (id, session_id, time_created, time_updated, data) \
VALUES (?1, ?2, ?3, ?3, ?4)",
rusqlite::params![id, session_id, created, serde_json::to_string(data)?],
)?;
Ok(())
}
fn insert_part(
conn: &Connection,
id: &str,
message_id: &str,
session_id: &str,
created: i64,
data: &Value,
) -> anyhow::Result<()> {
conn.execute(
"INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) \
VALUES (?1, ?2, ?3, ?4, ?4, ?5)",
rusqlite::params![
id,
message_id,
session_id,
created,
serde_json::to_string(data)?
],
)?;
Ok(())
}
fn write_tree_session(
tree_base: &std::path::Path,
session_id: &str,
marker: &str,
) -> anyhow::Result<()> {
let message_id = format!("msg_tree_{session_id}");
let part_id_value = format!("prt_tree_{session_id}");
write_json_file(
&tree_base
.join("session")
.join("proj-tree")
.join(format!("{session_id}.json")),
&json!({
"id": session_id,
"projectID": "proj-tree",
"directory": "/tmp/tree",
"time": { "created": 1_780_000_000_000i64 },
}),
)?;
write_json_file(
&tree_base
.join("message")
.join(session_id)
.join(format!("{message_id}.json")),
&json!({
"id": message_id,
"sessionID": session_id,
"role": "assistant",
"time": { "created": 1_780_000_000_001i64 },
}),
)?;
write_json_file(
&tree_base
.join("part")
.join(&message_id)
.join(format!("{part_id_value}.json")),
&json!({
"id": part_id_value,
"sessionID": session_id,
"messageID": message_id,
"type": "text",
"text": marker,
"synthetic": false,
}),
)?;
Ok(())
}
const V2_SESSION_COUNT: usize = 12;
const V2_MIGRATED_CONTINUED: &str = "ses_ece44e9c1ffeJWYCR23BcS31lX";
const V2_MIGRATED_SPLIT_COPY: &str = "msg_131bb1661001TtVuWkGzyg3n04";
const V2_POST_MIGRATION_USER: &str = "msg_131c0e706001jV7Smy0gYSlb2L";
const V2_MIGRATED_WITH_NOTICE: &str = "ses_ece45e791ffekNVGH2hsZMQvl5";
const V2_LEGACY_TOOL_NOTICE: &str = "msg_131ba23ec001TKOqvkoEn1JeVb";
const V2_MIGRATED_UNTOUCHED: &str = "ses_ece45fa45ffewtiD8DSPLj3P62";
const V2_TOOLS_SESSION: &str = "ses_ece44ba32ffevzFy2aZZD3h7TM";
const V2_CHILD: &str = "ses_ece3f27c0ffeXmTUqsOvLDGlbM";
const V2_CHILD_PARENT: &str = "ses_ece3f2e9affezCws7GfDhhKVSE";
const V2_FORK: &str = "ses_ece3c3812ffeMp6n6hGikNUmC3";
const V2_FORK_THROUGH: &str = "msg_131bb5340001PD3DzEht2R4zp9";
const V2_OPS_SESSION: &str = "ses_ece3c826dffeLR2B3sqhOHBVYG";
fn v2_db_fixture() -> PathBuf {
crate::adapter::test_support::manifest_dir()
.join("tests/fixtures/adapter/opencode/v2/opencode.db")
}
fn v2_data_dir(temp: &std::path::Path) -> anyhow::Result<PathBuf> {
let dir = temp.join("data");
std::fs::create_dir_all(&dir)?;
std::fs::copy(v2_db_fixture(), dir.join("opencode.db"))?;
Ok(dir)
}
async fn ingest_v2(
temp: &std::path::Path,
) -> anyhow::Result<(Store, PathBuf, crate::handlers::IngestSummary)> {
let data = v2_data_dir(temp)?;
let store = Store::open_local(temp.join("store")).await?;
let summary = ingest_adapter(
&store,
&OpencodeAdapter::new(&data),
&crate::adapter::NoopOracle,
|_| {},
)
.await?;
Ok((store, data, summary))
}
fn v2_type(message: &crate::sessions::MessageWithParts) -> Option<String> {
raw_record(message.message.options()).and_then(|raw| {
raw.get("type")
.and_then(Value::as_str)
.map(ToOwned::to_owned)
})
}
fn texts(message: &crate::sessions::MessageWithParts) -> Vec<(Provenance, String)> {
message
.parts
.iter()
.filter_map(|part| match &part.kind {
PartKind::Text { text } => Some((
part.provenance,
text.as_deref()
.map(String::as_str)
.unwrap_or_default()
.to_owned(),
)),
_ => None,
})
.collect()
}
#[tokio::test(flavor = "multi_thread")]
async fn v2_fixture_merges_both_table_generations() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let (store, _, summary) = ingest_v2(temp.path()).await?;
assert_eq!(summary.dropped_events, 0, "{summary:?}");
assert_eq!(summary.dropped_sessions, 0, "{summary:?}");
assert_eq!(store.session_ids().await?.len(), V2_SESSION_COUNT);
let continued = store
.get_session(V2_MIGRATED_CONTINUED)
.await?
.expect("migrated session ingests");
let ids: Vec<&str> = continued.messages.iter().map(|m| m.message.id()).collect();
assert!(
ids.contains(&V2_POST_MIGRATION_USER),
"post-migration turn lands"
);
assert!(
!ids.contains(&V2_MIGRATED_SPLIT_COPY),
"the migration's split-off copy of a V1 record is not re-ingested",
);
assert!(
continued
.messages
.iter()
.flat_map(|m| &m.parts)
.any(|part| part.id.starts_with("prt_")),
"migrated history keeps its V1 part rows",
);
assert!(
raw_record(&continued.session.options)
.is_some_and(|raw| raw.get("projectID").is_some()),
"a migrated session keeps its V1 head",
);
let noticed = store
.get_session(V2_MIGRATED_WITH_NOTICE)
.await?
.expect("migrated session ingests");
assert!(
noticed
.messages
.iter()
.all(|m| m.message.id() != V2_LEGACY_TOOL_NOTICE),
"the migration's appended legacy-tool notice is a copy, not activity",
);
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn v2_lineage_covers_spawned_children_and_forks() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let (store, _, _) = ingest_v2(temp.path()).await?;
let child = store.get_session(V2_CHILD).await?.expect("child ingests");
assert_eq!(
child.session.parent_session_id.as_deref(),
Some(V2_CHILD_PARENT)
);
assert_eq!(child.session.parent_message_id, None);
assert_eq!(child.session.source_agent, "opencode/general");
let fork = store.get_session(V2_FORK).await?.expect("fork ingests");
assert_eq!(
fork.session.parent_session_id.as_deref(),
Some(V2_TOOLS_SESSION)
);
assert_eq!(
fork.session.parent_message_id.as_deref(),
Some(V2_FORK_THROUGH),
"a `through` boundary is the fork's cut-point",
);
assert_eq!(
fork.session.source_agent, "opencode",
"a fork is not a subagent"
);
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn v2_message_types_land_in_their_model_slots() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let (store, _, _) = ingest_v2(temp.path()).await?;
let ops = store
.get_session(V2_OPS_SESSION)
.await?
.expect("ops session ingests");
let by_type = |kind: &str| {
ops.messages
.iter()
.filter(|m| v2_type(m).as_deref() == Some(kind))
.collect::<Vec<_>>()
};
let one = |kind: &str| {
let found = by_type(kind);
assert!(!found.is_empty(), "fixture has a {kind} row");
found[0]
};
for message in &ops.messages {
if !is_synthetic(message.message.options()) {
assert!(
message_seq(message.message.options()).is_some(),
"V2 message {} carries its seq",
message.message.id(),
);
}
}
let skill = one("skill");
assert_eq!(skill.message.role(), crate::wire::Role::User);
assert!(matches!(
texts(skill).as_slice(),
[(Provenance::Injected, _)]
));
let shell = one("shell");
assert_eq!(shell.message.role(), crate::wire::Role::User);
assert_eq!(
texts(shell),
vec![(Provenance::Conversational, "ls".to_owned())]
);
let output = ops
.messages
.iter()
.find(|m| m.message.id() == format!("{}/output", shell.message.id()))
.expect("a finished shell's output carrier");
assert!(matches!(
&output.message,
Message::System { content: Some(text), .. } if text.as_str() == "README.md\ndot.png\n"
));
assert!(is_synthetic(output.message.options()));
for synthetic in by_type("synthetic") {
assert_eq!(synthetic.message.role(), crate::wire::Role::User);
assert!(
texts(synthetic)
.iter()
.all(|(p, _)| *p == Provenance::Injected)
);
}
let compaction = one("compaction");
assert_eq!(compaction.message.role(), crate::wire::Role::User);
assert!(
texts(compaction)
.iter()
.all(|(p, _)| *p == Provenance::Injected)
);
for kind in ["idle", "agent-switched", "model-switched"] {
let marker = one(kind);
assert!(
matches!(&marker.message, Message::System { content: None, .. }),
"{kind} is an empty carrier",
);
assert!(marker.parts.is_empty());
}
let prompt = one("user");
assert!(matches!(
texts(prompt).as_slice(),
[(Provenance::Conversational, text)] if text.contains("Say hi")
));
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn v2_tool_items_split_into_call_and_result() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let (store, _, _) = ingest_v2(temp.path()).await?;
let session = store
.get_session(V2_TOOLS_SESSION)
.await?
.expect("tools session ingests");
let calls: Vec<String> = session
.messages
.iter()
.flat_map(|m| &m.parts)
.filter_map(|part| match &part.kind {
PartKind::ToolCall { call_id, .. } => call_id.as_deref().cloned(),
_ => None,
})
.collect();
assert_eq!(calls.len(), 2, "two read calls");
let results: Vec<(String, bool, Value)> = session
.messages
.iter()
.flat_map(|m| &m.parts)
.filter_map(|part| match &part.kind {
PartKind::ToolResult {
call_id,
is_failure,
result,
..
} => Some((call_id.as_deref().cloned()?, *is_failure, result.clone())),
_ => None,
})
.collect();
assert_eq!(results.len(), 2, "each settled call has a result");
for (call_id, _, _) in &results {
assert!(
calls.contains(call_id),
"result {call_id} correlates to a call"
);
}
let failed: Vec<_> = results.iter().filter(|(_, failure, _)| *failure).collect();
assert_eq!(failed.len(), 1);
assert_eq!(failed[0].2["type"], "tool.execution");
assert!(
results
.iter()
.any(|(_, failure, result)| !failure && result[0]["text"].is_string()),
"a completed call's result is its content",
);
Ok(())
}
#[test]
fn v2_unfinished_and_unknown_records_land_without_synthesis() {
let anchor = DateTime::from_timestamp_millis(1).expect("valid");
let assistant = json!({
"id": "msg_a", "type": "assistant", "time": { "created": 1000 },
"content": [
{ "type": "tool", "id": "call_s", "name": "bash",
"state": { "status": "streaming", "input": "{\"comm" }, "time": { "created": 1001 } },
{ "type": "tool", "id": "call_r", "name": "bash",
"state": { "status": "running", "input": { "command": "ls" }, "metadata": {} },
"time": { "created": 1002, "ran": 1003 } },
{ "type": "future-kind", "data": 1 },
],
});
let events = build_v2_message_events("ses_x", "msg_a", &assistant, 7, anchor);
let messages = events
.iter()
.filter(|event| matches!(event, IngestEvent::Message(_)))
.count();
assert_eq!(messages, 1, "no result carrier for an unfinished call");
let parts: Vec<&Part> = events
.iter()
.filter_map(|event| match event {
IngestEvent::Part(part) => Some(part),
_ => None,
})
.collect();
assert_eq!(parts.len(), 3);
assert!(matches!(&parts[0].kind, PartKind::ToolCall { params, .. } if params == "{\"comm"));
assert!(matches!(&parts[2].kind, PartKind::Text { text: None }));
let future = json!({ "id": "msg_f", "type": "teleported", "time": { "created": 5 } });
let events = build_v2_message_events("ses_x", "msg_f", &future, 8, anchor);
assert!(matches!(
events.as_slice(),
[IngestEvent::Message(Message::System { content: None, options, .. })]
if raw_record(options).is_some_and(|raw| raw["type"] == "teleported")
));
}
#[tokio::test(flavor = "multi_thread")]
async fn v2_native_restore_replays_the_transfer_shape() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let (store, data, _) = ingest_v2(temp.path()).await?;
let conn = open_db(&data.join("opencode.db"))?;
let v1_ids: HashSet<String> = conn
.prepare("SELECT id FROM session")?
.query_map([], |row| row.get(0))?
.collect::<rusqlite::Result<_>>()?;
let mut native_v2 = 0usize;
for session_id in store.session_ids().await? {
let session = store.get_session(&session_id).await?.expect("round-trips");
if v1_ids.contains(&session_id) {
continue;
}
let body = serialize_native_body(&session);
native_v2 += 1;
let expected: Vec<Value> = conn
.prepare(
"SELECT id, type, data FROM session_message WHERE session_id = ?1 ORDER BY seq",
)?
.query_map([&session_id], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
))
})?
.map(|row| {
let (id, kind, data) = row.expect("row");
let mut value: Value = serde_json::from_str(&data).expect("json");
value["id"] = json!(id);
value["type"] = json!(kind);
crate::adapter::extract::extract_raw_record(&value)
})
.collect();
assert_eq!(
body["messages"].as_array().expect("messages"),
&expected,
"{session_id}: messages replay the session_message records in seq order",
);
let row = conn.query_row(
"SELECT * FROM session_v2 WHERE id = ?1",
[&session_id],
v2_row_value,
)?;
assert_eq!(body["info"]["id"], json!(session_id));
assert_eq!(body["info"]["projectID"], row["project_id"]);
assert_eq!(body["info"]["location"]["directory"], row["directory"]);
assert_eq!(body["info"]["time"]["created"], row["time_created"]);
assert!(
body["info"]["model"]["variant"].is_string(),
"fromRow fills variant"
);
if session_id == V2_FORK {
assert_eq!(body["info"]["fork"]["sessionID"], json!(V2_TOOLS_SESSION));
assert_eq!(body["info"]["fork"]["boundary"]["type"], "through");
}
}
assert_eq!(native_v2, V2_SESSION_COUNT - v1_ids.len());
let continued = store
.get_session(V2_MIGRATED_CONTINUED)
.await?
.expect("round-trips");
let files = serialize_native(&continued)?;
assert_eq!(files[0].actual_fidelity, RestoreFidelity::Foreign);
let untouched = store
.get_session(V2_MIGRATED_UNTOUCHED)
.await?
.expect("round-trips");
let body = serialize_native_body(&untouched);
assert!(
body["messages"][0]["info"]["role"].is_string(),
"V1 envelope"
);
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn plan_matches_the_events_gate_v2_source() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let data = v2_data_dir(temp.path())?;
let (first, max_plan) = plan_gate_agreement(&OpencodeAdapter::new(&data)).await?;
assert_eq!(first.sessions, V2_SESSION_COUNT);
assert_eq!(max_plan.fresh, V2_SESSION_COUNT);
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn v2_freshness_tracks_post_migration_rows() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let (store, data, _) = ingest_v2(temp.path()).await?;
let db = data.join("opencode.db");
let light = |session_id: &str, store| DbLight {
db_path: db.clone(),
session_id: session_id.to_owned(),
store,
};
{
let conn = open_db(&db)?;
let notice = light(V2_MIGRATED_WITH_NOTICE, DbStore::Migrated);
assert_eq!(
db_store_watermark(&conn, ¬ice)?,
db_session_watermark(&conn, &db, V2_MIGRATED_WITH_NOTICE)?,
"a migration copy never raises a migrated session's watermark",
);
}
let appended_ms: i64 = 1_800_000_000_000;
{
let conn = Connection::open(&db)?;
conn.execute(
"INSERT INTO session_message (id, session_id, type, seq, time_created, time_updated, data)
VALUES ('msg_zzzzv2fresh00000000000000', ?1, 'user', 1000, ?2, ?2, ?3)",
rusqlite::params![
V2_MIGRATED_WITH_NOTICE,
appended_ms,
json!({ "time": { "created": appended_ms }, "text": "fresh v2 turn" }).to_string(),
],
)?;
let conn = open_db(&db)?;
assert_eq!(
db_store_watermark(&conn, &light(V2_MIGRATED_WITH_NOTICE, DbStore::Migrated))?,
Some(appended_ms * 1_000),
);
}
let before = store
.get_session(V2_MIGRATED_WITH_NOTICE)
.await?
.expect("ingested")
.messages
.len();
let oracle = FixedOracle {
session_id: V2_MIGRATED_WITH_NOTICE,
watermark_micros: appended_ms * 1_000 - 1,
};
ingest_adapter(&store, &OpencodeAdapter::new(&data), &oracle, |_| {}).await?;
let after = store
.get_session(V2_MIGRATED_WITH_NOTICE)
.await?
.expect("ingested");
assert_eq!(
after.messages.len(),
before + 1,
"only the new V2 row is added"
);
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn session_repeated_across_dbs_lands_once() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let data = v2_data_dir(temp.path())?;
std::fs::copy(v2_db_fixture(), data.join("opencode-next.db"))?;
let store = Store::open_local(temp.path().join("store")).await?;
let summary = ingest_adapter(
&store,
&OpencodeAdapter::new(&data),
&crate::adapter::NoopOracle,
|_| {},
)
.await?;
assert_eq!(store.session_ids().await?.len(), V2_SESSION_COUNT);
assert_eq!(summary.skipped_superseded, V2_SESSION_COUNT, "{summary:?}");
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn v2_only_db_without_v1_tables_ingests() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let data = v2_data_dir(temp.path())?;
Connection::open(data.join("opencode.db"))?.execute_batch(
"PRAGMA foreign_keys = OFF; DROP TABLE part; DROP TABLE message; DROP TABLE session;",
)?;
let store = Store::open_local(temp.path().join("store")).await?;
let summary = ingest_adapter(
&store,
&OpencodeAdapter::new(&data),
&crate::adapter::NoopOracle,
|_| {},
)
.await?;
assert_eq!(summary.dropped_events, 0, "{summary:?}");
assert_eq!(store.session_ids().await?.len(), V2_SESSION_COUNT);
let migrated = store
.get_session(V2_MIGRATED_CONTINUED)
.await?
.expect("ingests from V2 alone");
assert!(
migrated
.messages
.iter()
.any(|m| m.message.id() == V2_MIGRATED_SPLIT_COPY),
"with no V1 rows to prefer, the V2 copies are the only record",
);
Ok(())
}
#[test]
fn synthetic_id_matches_the_migration() {
assert_eq!(
synthetic_id("msg_131bb1661001jevwkFX4ZpwQDM"),
V2_MIGRATED_SPLIT_COPY,
"the user/synthetic split",
);
assert_eq!(
synthetic_id("msg_131ba23ec001D2OA1U5ijz8c0y"),
V2_LEGACY_TOOL_NOTICE,
"the legacy-tool notice after the last projected row",
);
}
fn insert_v2_row(
conn: &Connection,
id: &str,
session_id: &str,
kind: &str,
seq: i64,
created_ms: i64,
data: &Value,
) -> anyhow::Result<()> {
conn.execute(
"INSERT INTO session_message (id, session_id, type, seq, time_created, time_updated, data)
VALUES (?1, ?2, ?3, ?4, ?5, ?5, ?6)",
rusqlite::params![id, session_id, kind, seq, created_ms, data.to_string()],
)?;
Ok(())
}
fn message_ids(session: &crate::sessions::SessionWithMessages) -> Vec<&str> {
session.messages.iter().map(|m| m.message.id()).collect()
}
#[tokio::test(flavor = "multi_thread")]
async fn migration_copies_are_exact_and_survive_v1_deletes() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let data = v2_data_dir(temp.path())?;
let deleted_v1 = "msg_131ba198c001kCcV4OpDtOChtC";
let lookalike = "msg_131ba23ec001ZZZZZZZZZZZZZZ";
{
let conn = Connection::open(data.join("opencode.db"))?;
conn.execute_batch("PRAGMA foreign_keys = OFF;")?;
conn.execute("DELETE FROM part WHERE message_id = ?1", [deleted_v1])?;
conn.execute("DELETE FROM message WHERE id = ?1", [deleted_v1])?;
insert_v2_row(
&conn,
lookalike,
V2_MIGRATED_WITH_NOTICE,
"user",
500,
1_700_000_000_000,
&json!({ "time": { "created": 1_800_000_000_000i64 }, "text": "after the upgrade" }),
)?;
}
let store = Store::open_local(temp.path().join("store")).await?;
ingest_adapter(
&store,
&OpencodeAdapter::new(&data),
&crate::adapter::NoopOracle,
|_| {},
)
.await?;
let session = store
.get_session(V2_MIGRATED_WITH_NOTICE)
.await?
.expect("ingests");
let ids = message_ids(&session);
assert!(
!ids.contains(&deleted_v1),
"the deleted V1 message's copy stays skipped"
);
assert!(!ids.contains(&V2_LEGACY_TOOL_NOTICE));
assert!(
ids.contains(&lookalike),
"a prefix lookalike is real activity"
);
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn prerelease_v2_db_reads_heads_from_its_session_table() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let data = v2_data_dir(temp.path())?;
let next = data.join("opencode-next.db");
std::fs::copy(v2_db_fixture(), &next)?;
let unique = "ses_zzzznextonly000000000000";
let empty = "ses_zzzznextempty00000000000";
{
let conn = Connection::open(&next)?;
conn.execute_batch(
"PRAGMA foreign_keys = OFF;
DROP TABLE part; DROP TABLE message; DROP TABLE session;
ALTER TABLE session_v2 RENAME TO session;",
)?;
conn.execute(
"UPDATE session SET id = ?1 WHERE id = ?2",
[unique, V2_OPS_SESSION],
)?;
conn.execute(
"UPDATE session_message SET session_id = ?1 WHERE session_id = ?2",
[unique, V2_OPS_SESSION],
)?;
conn.execute(
"CREATE TEMP TABLE head AS SELECT * FROM session WHERE id = ?1",
[unique],
)?;
conn.execute("UPDATE head SET id = ?1", [empty])?;
conn.execute_batch("INSERT INTO session SELECT * FROM head;")?;
}
let store = Store::open_local(temp.path().join("store")).await?;
let summary = ingest_adapter(
&store,
&OpencodeAdapter::new(&data),
&crate::adapter::NoopOracle,
|_| {},
)
.await?;
assert_eq!(summary.dropped_events, 0, "{summary:?}");
assert_eq!(summary.unreadable_events, 0, "{summary:?}");
assert_eq!(store.session_ids().await?.len(), V2_SESSION_COUNT + 2);
assert_eq!(
summary.skipped_superseded,
V2_SESSION_COUNT - 1,
"{summary:?}"
);
let session = store
.get_session(unique)
.await?
.expect("unique session lands");
let ops = store.get_session(V2_OPS_SESSION).await?.expect("ops lands");
assert_eq!(
session.messages.len(),
ops.messages.len(),
"with its whole transcript"
);
assert_eq!(session_store(&session.session.options), Some("session"));
let empty = store
.get_session(empty)
.await?
.expect("empty session lands");
assert_eq!(session_store(&empty.session.options), Some("session"));
assert_eq!(
serialize_native(&session)?[0].actual_fidelity,
RestoreFidelity::Native
);
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn unsettled_rows_wait_for_the_turn_to_close() -> anyhow::Result<()> {
const SESSION: &str = "ses_ece40409affezvqxNOKCv7GG6X";
const ASSISTANT: &str = "msg_zzzzv2streaming0000000000";
let temp = TempDir::new()?;
let (store, data, _) = ingest_v2(temp.path()).await?;
let db = data.join("opencode.db");
let adapter = OpencodeAdapter::new(&data);
let started: i64 = 1_800_000_000_000;
let assistant = |text: &str, completed: Option<i64>| {
let mut time = json!({ "created": started + 1 });
if let Some(completed) = completed {
time["completed"] = json!(completed);
}
json!({
"time": time, "agent": "build",
"model": { "id": "m", "providerID": "p" },
"content": [{ "type": "text", "text": text }],
})
};
{
let conn = Connection::open(&db)?;
insert_v2_row(
&conn,
"msg_zzzzv2prompt0000000000000",
SESSION,
"user",
100,
started,
&json!({ "time": { "created": started }, "text": "next question" }),
)?;
insert_v2_row(
&conn,
ASSISTANT,
SESSION,
"assistant",
101,
started + 1,
&assistant("par", None),
)?;
}
let light = DbLight {
db_path: db.clone(),
session_id: SESSION.to_owned(),
store: DbStore::V2 { head: "session_v2" },
};
assert_eq!(
db_store_watermark(&open_db(&db)?, &light)?,
None,
"open turn: unknown"
);
let fresh_everywhere = FixedOracle {
session_id: SESSION,
watermark_micros: i64::MAX,
};
ingest_adapter(&store, &adapter, &fresh_everywhere, |_| {}).await?;
let mid = store.get_session(SESSION).await?.expect("ingested");
assert!(message_ids(&mid).contains(&"msg_zzzzv2prompt0000000000000"));
assert!(
!message_ids(&mid).contains(&ASSISTANT),
"the running row is deferred"
);
{
let conn = Connection::open(&db)?;
conn.execute(
"UPDATE session_message SET data = ?1 WHERE id = ?2",
rusqlite::params![
assistant("final answer", Some(started + 5)).to_string(),
ASSISTANT
],
)?;
insert_v2_row(
&conn,
"msg_zzzzv2idle000000000000000",
SESSION,
"idle",
102,
started + 6,
&json!({ "time": { "created": started + 6 }, "outcome": "succeeded" }),
)?;
}
let oracle = FixedOracle {
session_id: SESSION,
watermark_micros: started * 1_000,
};
ingest_adapter(&store, &adapter, &oracle, |_| {}).await?;
let done = store.get_session(SESSION).await?.expect("ingested");
let finished = done
.messages
.iter()
.find(|m| m.message.id() == ASSISTANT)
.expect("the settled row lands");
assert_eq!(
texts(finished),
vec![(Provenance::Conversational, "final answer".to_owned())],
);
Ok(())
}
#[test]
fn unreadable_v2_tail_makes_a_migrated_session_unknown() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let data = v2_data_dir(temp.path())?;
let db = data.join("opencode.db");
Connection::open(&db)?.execute(
"UPDATE session_message SET data = '{not json' WHERE id = ?1",
[V2_POST_MIGRATION_USER],
)?;
Connection::open(&db)?.execute(
"DELETE FROM session_message WHERE session_id = ?1 AND type = 'idle'",
[V2_MIGRATED_CONTINUED],
)?;
let light = DbLight {
db_path: db.clone(),
session_id: V2_MIGRATED_CONTINUED.to_owned(),
store: DbStore::Migrated,
};
assert_eq!(db_store_watermark(&open_db(&db)?, &light)?, None);
Ok(())
}
#[test]
fn v2_session_info_follows_from_row() {
let info = v2_session_info(&json!({
"id": "ses_a", "project_id": "p", "directory": "/w", "title": "",
"parent_id": "", "agent": "", "path": "", "time_archived": 0,
"time_created": 1, "time_updated": 2, "cost": 0.5,
"tokens_input": 1, "tokens_output": 2, "tokens_reasoning": 0,
"tokens_cache_read": 0, "tokens_cache_write": 0,
"revert": { "messageID": "msg_a", "diff": "d",
"files": [{ "path": "f", "status": "modified", "additions": 1, "deletions": 0, "patch": "p" }] },
}));
assert_eq!(info["title"], "", "title uses `??`, so an empty one stays");
for key in ["parentID", "agent", "subpath"] {
assert!(info.get(key).is_none(), "{key} is unset when empty");
}
assert!(info["time"].get("archived").is_none());
assert_eq!(
info["revert"],
json!({ "messageID": "msg_a", "files": [{ "file": "f", "status": "modified", "additions": 1, "deletions": 0, "patch": "p" }] }),
);
}
fn v2_light(db: &std::path::Path, session_id: &str) -> DbLight {
DbLight {
db_path: db.to_path_buf(),
session_id: session_id.to_owned(),
store: DbStore::V2 { head: "session_v2" },
}
}
#[test]
fn out_of_order_insert_keeps_the_open_turn_unknown() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let data = v2_data_dir(temp.path())?;
let db = data.join("opencode.db");
let conn = Connection::open(&db)?;
conn.execute(
"DELETE FROM session_message WHERE session_id = ?1 AND seq > 21",
[V2_OPS_SESSION],
)?;
let light = v2_light(&db, V2_OPS_SESSION);
assert_eq!(db_store_watermark(&conn, &light)?, None);
conn.execute(
"UPDATE session_message SET seq = 13 WHERE session_id = ?1 AND seq = 21",
[V2_OPS_SESSION],
)?;
assert_eq!(db_store_watermark(&conn, &light)?, None, "in order too");
insert_v2_row(
&conn,
"msg_zzzzv2idleaftermath000000",
V2_OPS_SESSION,
"idle",
22,
1_800_000_000_000,
&json!({ "time": { "created": 1_800_000_000_000i64 }, "outcome": "succeeded" }),
)?;
assert_eq!(
db_store_watermark(&conn, &light)?,
Some(1_800_000_000_000 * 1_000),
"a closed turn reads fresh again",
);
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn running_background_shell_waits_even_behind_idle() -> anyhow::Result<()> {
let temp = TempDir::new()?;
let data = v2_data_dir(temp.path())?;
let db = data.join("opencode.db");
let shell_id: String = Connection::open(&db)?.query_row(
"SELECT id FROM session_message WHERE session_id = ?1 AND type = 'shell'",
[V2_OPS_SESSION],
|row| row.get(0),
)?;
let running = json!({
"metadata": { "background": true }, "time": { "created": 1_791_836_259_452i64 },
"shellID": "sh_x", "command": "ls", "status": "running",
});
Connection::open(&db)?.execute(
"UPDATE session_message SET data = ?1 WHERE id = ?2",
rusqlite::params![running.to_string(), shell_id],
)?;
assert_eq!(
db_store_watermark(&open_db(&db)?, &v2_light(&db, V2_OPS_SESSION))?,
None
);
let store = Store::open_local(temp.path().join("store")).await?;
let adapter = OpencodeAdapter::new(&data);
ingest_adapter(&store, &adapter, &crate::adapter::NoopOracle, |_| {}).await?;
let mid = store.get_session(V2_OPS_SESSION).await?.expect("ingests");
assert!(
!message_ids(&mid).contains(&shell_id.as_str()),
"the running shell waits"
);
let mut ended = running;
ended["status"] = json!("exited");
ended["exit"] = json!(0);
ended["time"]["completed"] = json!(1_800_000_000_000i64);
ended["output"] = json!({ "output": "README.md\n" });
Connection::open(&db)?.execute(
"UPDATE session_message SET data = ?1 WHERE id = ?2",
rusqlite::params![ended.to_string(), shell_id],
)?;
let oracle = crate::sessions::RowmapOracle(store.rowmap_snapshot());
ingest_adapter(&store, &adapter, &oracle, |_| {}).await?;
let done = store.get_session(V2_OPS_SESSION).await?.expect("ingests");
assert!(
message_ids(&done).contains(&shell_id.as_str()),
"the ended shell lands"
);
let output = done
.messages
.iter()
.find(|m| m.message.id() == format!("{shell_id}/output"))
.expect("with its output carrier");
assert!(matches!(
&output.message,
Message::System { content: Some(text), .. } if text.as_str() == "README.md\n"
));
let summary = ingest_adapter(
&store,
&adapter,
&crate::sessions::RowmapOracle(store.rowmap_snapshot()),
|_| {},
)
.await?;
assert_eq!(
summary.inserted, 0,
"then the session reads fresh: {summary:?}"
);
Ok(())
}
}