use std::ffi::OsStr;
use std::io::Write;
use std::path::{Path, PathBuf};
use std::process::{Command, Output, Stdio};
use serde_json::Value;
pub const BRANCH: &str = "main";
pub const REMOTE: &str = "origin";
pub const REMOTE_BRANCH_NAME: &str = "remotes/origin/main";
pub const NOT_INSTALLED: &str = "error: dolt is not installed.\n \
The SessionStart hook (ops/local/claude-session-start.sh) installs it; \
otherwise download a release from https://github.com/dolthub/dolt/releases";
pub const SCHEMA: &[&str] = &[
"create table manifest (k varchar(64) primary key, v text)",
"create table threads (thread_id varchar(64) primary key, briefing_path varchar(1024) null, mission_id varchar(64) null)",
"create table continuation_entries (thread_id varchar(64), seq int, entry longtext not null, primary key (thread_id, seq))",
"create table bindings (actor varchar(255) primary key, kind enum('cloud','worktree') not null, thread_id varchar(64) not null, registered_at varchar(32) not null)",
"create table external_events (position bigint primary key, envelope longtext not null)",
"create table watermark (id int primary key, processed_through bigint not null, updated_at varchar(32) not null)",
];
#[allow(dead_code)]
pub const LEGACY_DOCUMENTS_TABLE: &str =
"create table documents (path varchar(1024) primary key, body longtext not null)";
pub const MISSION_SCHEMA: &[&str] = &[
"create table missions (\
id varchar(64) not null primary key, \
name varchar(255) not null, \
category enum('operational','administrative') not null, \
status enum('todo','in-progress','done') not null default 'todo', \
parent varchar(64) null, \
purpose longtext null, \
decision_authority longtext null, \
next_objective int not null default 1, \
next_task int not null default 1, \
next_question int not null default 1, \
next_constraint int not null default 1, \
next_out_of_scope int not null default 1, \
constraint missions_parent_fk foreign key (parent) references missions (id))",
"create table objective_outcomes (\
mission_id varchar(64) not null, \
seq int not null, \
outcome longtext not null, \
met boolean not null default false, \
primary key (mission_id, seq), \
constraint objective_outcomes_mission_fk foreign key (mission_id) references missions (id))",
"create table mission_blocks (\
mission_id varchar(64) not null, \
blocked_by_id varchar(64) not null, \
primary key (mission_id, blocked_by_id), \
constraint mission_blocks_mission_fk foreign key (mission_id) references missions (id), \
constraint mission_blocks_by_fk foreign key (blocked_by_id) references missions (id))",
"create table mission_documents (\
mission_id varchar(64) not null, \
path varchar(1024) not null, \
role enum('intelligence','report','design') not null, \
primary key (mission_id, path), \
constraint mission_documents_mission_fk foreign key (mission_id) references missions (id))",
];
pub const LIST_SCHEMA: &[&str] = &[
"create table mission_constraints (\
mission_id varchar(64) not null, \
seq int not null, \
text longtext not null, \
primary key (mission_id, seq), \
constraint mission_constraints_mission_fk foreign key (mission_id) references missions (id))",
"create table mission_out_of_scope (\
mission_id varchar(64) not null, \
seq int not null, \
text longtext not null, \
primary key (mission_id, seq), \
constraint mission_out_of_scope_mission_fk foreign key (mission_id) references missions (id))",
];
pub const TASK_SCHEMA: &[&str] = &[
"create table tasks (\
mission_id varchar(64) not null, \
seq int not null, \
description longtext not null, \
status enum('todo','in-progress','done') not null default 'todo', \
primary key (mission_id, seq), \
constraint tasks_mission_fk foreign key (mission_id) references missions (id))",
"create table task_blocks (\
mission_id varchar(64) not null, \
task_seq int not null, \
blocked_by_seq int not null, \
primary key (mission_id, task_seq, blocked_by_seq), \
constraint task_blocks_task_fk foreign key (mission_id, task_seq) references tasks (mission_id, seq), \
constraint task_blocks_by_fk foreign key (mission_id, blocked_by_seq) references tasks (mission_id, seq))",
"create table decisions (\
mission_id varchar(64) not null, \
seq int not null, \
text longtext not null, \
decided_by varchar(255) not null, \
decided_on date not null, \
task_seq int null, \
adr_path varchar(1024) null, \
primary key (mission_id, seq), \
constraint decisions_mission_fk foreign key (mission_id) references missions (id), \
constraint decisions_task_fk foreign key (mission_id, task_seq) references tasks (mission_id, seq))",
"create table questions (\
mission_id varchar(64) not null, \
seq int not null, \
text longtext not null, \
status enum('open','answered') not null default 'open', \
answer longtext null, \
answered_by varchar(255) null, \
answered_on date null, \
asked_on date not null, \
primary key (mission_id, seq), \
constraint questions_mission_fk foreign key (mission_id) references missions (id))",
];
pub const LAYOUT_VERSION: i64 = 8;
pub const BINDINGS_LAYOUT_VERSION: i64 = 8;
pub const THREAD_MISSION_LAYOUT_VERSION: i64 = 5;
pub const ADR_LAYOUT_VERSION: i64 = 6;
pub const LIST_LAYOUT_VERSION: i64 = 7;
pub const TASKS_LAYOUT_VERSION: i64 = 4;
pub const MISSIONS_LAYOUT_VERSION: i64 = 3;
pub const DOCS_LAYOUT_VERSION: i64 = 2;
pub const LEGACY_LAYOUT_VERSION: i64 = 1;
pub fn upgrade_sql_moving(from: i64, moved: &str) -> String {
let mut sql = String::new();
if from <= DOCS_LAYOUT_VERSION {
for statement in MISSION_SCHEMA.iter().chain(LIST_SCHEMA).chain(TASK_SCHEMA) {
sql.push_str(statement);
sql.push_str(";\n");
}
} else if from == MISSIONS_LAYOUT_VERSION {
sql.push_str(
"alter table missions add column next_task int not null default 1;\n\
alter table missions add column next_question int not null default 1;\n",
);
for statement in TASK_SCHEMA {
sql.push_str(statement);
sql.push_str(";\n");
}
}
if from < THREAD_MISSION_LAYOUT_VERSION {
sql.push_str(
"alter table threads modify column briefing_path varchar(1024) null;\n\
alter table threads add column mission_id varchar(64) null;\n",
);
}
if from == TASKS_LAYOUT_VERSION || from == THREAD_MISSION_LAYOUT_VERSION {
sql.push_str("alter table decisions add column adr_path varchar(1024) null;\n");
}
if (MISSIONS_LAYOUT_VERSION..LIST_LAYOUT_VERSION).contains(&from) {
sql.push_str(
"alter table missions add column next_constraint int not null default 1;\n\
alter table missions add column next_out_of_scope int not null default 1;\n",
);
for statement in LIST_SCHEMA {
sql.push_str(statement);
sql.push_str(";\n");
}
sql.push_str(moved);
sql.push_str(
"alter table missions drop column constraints_text;\n\
alter table missions drop column out_of_scope;\n",
);
}
if from < BINDINGS_LAYOUT_VERSION {
sql.push_str(BINDINGS_UPGRADE_SQL);
}
sql.push_str(&version_sql());
sql
}
const BINDINGS_UPGRADE_SQL: &str = "create table bindings (actor varchar(255) primary key, kind enum('cloud','worktree') not null, thread_id varchar(64) not null, registered_at varchar(32) not null);\n\
insert into bindings (actor, kind, thread_id, registered_at) select session_id, 'cloud', thread_id, registered_at from cloud_session_bindings;\n\
drop table cloud_session_bindings;\n";
pub fn upgrade_sql(from: i64) -> String {
upgrade_sql_moving(from, "")
}
pub fn move_lists_sql(rows: &[(String, String, String)]) -> String {
let mut sql = String::new();
for (id, constraints, out_of_scope) in rows {
for (table, column, text) in [
("mission_constraints", "next_constraint", constraints),
("mission_out_of_scope", "next_out_of_scope", out_of_scope),
] {
let items = crate::mission::model::lines_to_items(text);
let Some(last) = items.last() else { continue };
let values: Vec<String> = items
.iter()
.map(|i| format!("({}, {}, {})", lit(id), i.seq, lit(&i.text)))
.collect();
sql.push_str(&format!(
"insert into {} (mission_id, seq, text) values {};\n\
update missions set {} = {} where id = {};\n",
table,
values.join(", "),
column,
last.seq + 1,
lit(id)
));
}
}
sql
}
fn version_sql() -> String {
format!(
"replace into manifest (k, v) values ('version', '{}');\n",
LAYOUT_VERSION
)
}
pub fn init_sql(repo_id: Option<&str>) -> String {
let mut sql = String::new();
for statement in SCHEMA
.iter()
.chain(MISSION_SCHEMA)
.chain(LIST_SCHEMA)
.chain(TASK_SCHEMA)
{
sql.push_str(statement);
sql.push_str(";\n");
}
sql.push_str(&format!(
"insert into manifest (k, v) values ('version', '{}');\n",
LAYOUT_VERSION
));
if let Some(id) = repo_id {
sql.push_str(&format!(
"insert into manifest (k, v) values ('repo_id', {});\n",
lit(id)
));
}
sql
}
pub fn lit(text: &str) -> String {
if text.is_empty() {
return "''".to_string();
}
let mut hex = String::with_capacity(text.len() * 2);
for byte in text.as_bytes() {
hex.push_str(&format!("{:02x}", byte));
}
format!("CONVERT(X'{}' USING utf8mb4)", hex)
}
#[derive(Clone)]
pub struct Dolt {
dir: PathBuf,
}
impl Dolt {
pub fn new(dir: impl Into<PathBuf>) -> Self {
Dolt { dir: dir.into() }
}
pub fn dir(&self) -> &Path {
&self.dir
}
fn command(&self) -> Command {
let mut command = Command::new("dolt");
command
.current_dir(&self.dir)
.env("DOLT_DISABLE_EVENT_FLUSH", "1")
.env("DOLT_REMOTE_INFO_BRANCH", "");
crate::gitenv::apply(&mut command);
command
}
fn spawn_error(e: std::io::Error) -> String {
if e.kind() == std::io::ErrorKind::NotFound {
NOT_INSTALLED.to_string()
} else {
format!("error: could not run dolt: {}", e)
}
}
pub fn output<I, S>(&self, args: I) -> Result<(Vec<String>, Output), String>
where
I: IntoIterator<Item = S>,
S: AsRef<OsStr>,
{
let args: Vec<String> = args
.into_iter()
.map(|a| a.as_ref().to_string_lossy().into_owned())
.collect();
let output = self
.command()
.args(&args)
.output()
.map_err(Self::spawn_error)?;
Ok((args, output))
}
pub fn run<I, S>(&self, args: I) -> Result<String, String>
where
I: IntoIterator<Item = S>,
S: AsRef<OsStr>,
{
let (args, output) = self.output(args)?;
if !output.status.success() {
return Err(failure_message(&args, &output));
}
Ok(String::from_utf8_lossy(&output.stdout).into_owned())
}
pub fn rows(&self, sql: &str) -> Result<Vec<Value>, String> {
let args = vec!["sql".to_string(), "-r".to_string(), "json".to_string()];
let stdout = self.pipe(&args, sql)?;
parse_rows(&stdout)
}
pub fn rows_each(&self, queries: &[String]) -> Result<Vec<Vec<Value>>, String> {
if queries.is_empty() {
return Ok(Vec::new());
}
let script: String = queries.iter().map(|q| format!("{};\n", q)).collect();
let args = vec!["sql".to_string(), "-r".to_string(), "json".to_string()];
let stdout = self.pipe(&args, &script)?;
let results = parse_result_sets(&stdout)?;
if results.len() != queries.len() {
return Err(format!(
"error: dolt printed {} result sets for {} queries",
results.len(),
queries.len()
));
}
Ok(results)
}
pub fn exec(&self, sql: &str) -> Result<(), String> {
let script = format!("start transaction;\n{}\ncommit;\n", sql);
let args = vec!["sql".to_string()];
self.pipe(&args, &script).map(|_| ())
}
fn pipe(&self, args: &[String], input: &str) -> Result<String, String> {
let mut child = self
.command()
.args(args)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.map_err(Self::spawn_error)?;
if let Some(mut stdin) = child.stdin.take() {
stdin
.write_all(input.as_bytes())
.map_err(|e| format!("error: could not write to dolt {}: {}", args.join(" "), e))?;
}
let output = child
.wait_with_output()
.map_err(|e| format!("error: could not run dolt: {}", e))?;
if !output.status.success() {
return Err(failure_message(args, &output));
}
Ok(String::from_utf8_lossy(&output.stdout).into_owned())
}
pub fn count(&self, sql: &str) -> Result<i64, String> {
Ok(first_count(&self.rows(sql)?))
}
pub fn head_hash(&self) -> Result<String, String> {
self.hash_of("HEAD")
}
pub fn hash_of(&self, revision: &str) -> Result<String, String> {
let rows = self.rows(&format!("select hashof({}) as h", lit(revision)))?;
rows.first()
.and_then(|row| text(row, "h"))
.ok_or_else(|| format!("error: dolt could not resolve {}", revision))
}
}
pub fn first_count(rows: &[Value]) -> i64 {
rows.first()
.and_then(|row| row.as_object())
.and_then(|map| map.values().next())
.and_then(Value::as_i64)
.unwrap_or(0)
}
pub fn ensure_identity() -> Result<(), String> {
static ENSURED: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
if ENSURED.load(std::sync::atomic::Ordering::Relaxed) {
return Ok(());
}
ensure_identity_once()?;
ENSURED.store(true, std::sync::atomic::Ordering::Relaxed);
Ok(())
}
fn ensure_identity_once() -> Result<(), String> {
let here = Dolt::new(".");
let (_, output) = here.output(["config", "--global", "--get", "user.name"])?;
let name = String::from_utf8_lossy(&output.stdout);
if output.status.success() && !name.trim().is_empty() {
return Ok(());
}
here.run(["config", "--global", "--add", "user.name", "tsk"])?;
here.run(["config", "--global", "--add", "user.email", "tsk@localhost"])?;
Ok(())
}
pub fn text(row: &Value, column: &str) -> Option<String> {
row.get(column).and_then(Value::as_str).map(str::to_string)
}
pub fn integer(row: &Value, column: &str) -> Option<i64> {
row.get(column).and_then(Value::as_i64)
}
pub fn flag(row: &Value, column: &str) -> bool {
match row.get(column) {
Some(Value::Bool(b)) => *b,
Some(Value::Number(n)) => n.as_i64() == Some(1),
_ => false,
}
}
fn parse_result_sets(stdout: &str) -> Result<Vec<Vec<Value>>, String> {
let mut sets = Vec::new();
for result in serde_json::Deserializer::from_str(stdout).into_iter::<Value>() {
let result =
result.map_err(|e| format!("error: dolt printed output that is not JSON: {}", e))?;
sets.push(match result.get("rows") {
Some(Value::Array(found)) => found.clone(),
_ => Vec::new(),
});
}
Ok(sets)
}
fn parse_rows(stdout: &str) -> Result<Vec<Value>, String> {
let mut rows = Vec::new();
for result in serde_json::Deserializer::from_str(stdout).into_iter::<Value>() {
let result =
result.map_err(|e| format!("error: dolt printed output that is not JSON: {}", e))?;
if let Some(Value::Array(found)) = result.get("rows") {
rows.extend(found.iter().cloned());
}
}
Ok(rows)
}
pub fn failure_message(args: &[String], output: &Output) -> String {
let stderr = String::from_utf8_lossy(&output.stderr);
let stdout = String::from_utf8_lossy(&output.stdout);
let detail = if stderr.trim().is_empty() {
stdout.trim().to_string()
} else {
stderr.trim().to_string()
};
let status = match output.status.code() {
Some(code) => format!("exit status {}", code),
None => "terminated by a signal".to_string(),
};
let shown = args.join(" ");
let shown: String = shown.chars().take(200).collect();
if detail.is_empty() {
format!("error: dolt {} failed ({})", shown, status)
} else {
format!("error: dolt {} failed ({}):\n{}", shown, status, detail)
}
}
#[allow(dead_code)]
pub fn available() -> bool {
Command::new("dolt")
.arg("version")
.stdout(Stdio::null())
.stderr(Stdio::null())
.status()
.map(|s| s.success())
.unwrap_or(false)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn lit_hex_encodes_so_quotes_and_semicolons_stay_data() {
assert_eq!(lit(""), "''");
assert_eq!(lit("a"), "CONVERT(X'61' USING utf8mb4)");
let hostile = lit("x'); drop table threads; --");
assert!(!hostile.contains("drop"));
assert!(hostile.starts_with("CONVERT(X'"));
}
#[test]
fn rows_are_read_from_every_result_set() {
let rows =
parse_rows("{\"rows\": [{\"a\":1}]}\n{}\n{\"rows\": [{\"b\":\"x\"}]}\n").unwrap();
assert_eq!(rows.len(), 2);
assert_eq!(integer(&rows[0], "a"), Some(1));
assert_eq!(text(&rows[1], "b").as_deref(), Some("x"));
assert!(parse_rows("{}\n").unwrap().is_empty());
assert!(parse_rows("").unwrap().is_empty());
}
#[test]
fn init_sql_creates_every_table_and_the_manifest_rows() {
let sql = init_sql(Some("work-api"));
for table in [
"manifest",
"threads",
"continuation_entries",
"bindings",
"external_events",
"watermark",
"missions",
"objective_outcomes",
"mission_blocks",
"mission_documents",
"tasks",
"task_blocks",
"decisions",
"questions",
] {
assert!(
sql.contains(&format!("create table {} ", table)),
"{}",
table
);
}
assert!(!sql.contains("create table documents"));
assert!(!sql.contains("cloud_session_bindings"));
assert!(sql.contains("kind enum('cloud','worktree') not null"));
assert!(sql.contains("('version', '8')"));
assert!(sql.contains("mission_id varchar(64) null"));
assert!(sql.contains("briefing_path varchar(1024) null"));
assert!(sql.contains("'repo_id'"));
assert!(!init_sql(None).contains("repo_id"));
}
#[test]
fn the_upgrade_from_each_older_version_ends_at_the_current_one() {
let from_four = upgrade_sql(4);
assert!(from_four.contains("alter table threads add column mission_id"));
assert!(from_four.contains("modify column briefing_path varchar(1024) null"));
assert!(!from_four.contains("create table missions "));
assert!(!from_four.contains("create table tasks "));
assert!(from_four.contains("alter table decisions add column adr_path varchar(1024) null"));
assert!(from_four.ends_with("('version', '8');\n"));
let from_five = upgrade_sql(5);
assert!(!from_five.contains("alter table threads"));
assert!(!from_five.contains("create table tasks "));
assert!(from_five.contains("alter table decisions add column adr_path"));
assert!(from_five.ends_with("('version', '8');\n"));
let from_six = upgrade_sql(6);
assert!(!from_six.contains("alter table decisions"));
assert!(!from_six.contains("alter table threads"));
assert!(from_six.contains("create table mission_constraints "));
assert!(from_six.contains("create table mission_out_of_scope "));
assert!(from_six.contains("add column next_constraint"));
assert!(from_six.contains("drop column constraints_text"));
assert!(from_six.contains("drop column out_of_scope"));
assert!(!from_six.contains("create table missions "));
assert!(from_six.ends_with("('version', '8');\n"));
let from_seven = upgrade_sql(7);
assert!(from_seven.contains("create table bindings "));
assert!(from_seven.contains(
"select session_id, 'cloud', thread_id, registered_at from cloud_session_bindings"
));
assert!(from_seven.contains("drop table cloud_session_bindings"));
assert!(!from_seven.contains("mission_constraints"));
assert!(!from_seven.contains("alter table"));
assert!(from_seven.ends_with("('version', '8');\n"));
assert!(from_six.contains("drop table cloud_session_bindings"));
let from_three = upgrade_sql(3);
assert!(from_three.contains("alter table missions add column next_task"));
assert!(from_three.contains("create table tasks "));
assert!(!from_three.contains("create table missions "));
assert!(from_three.contains("alter table threads add column mission_id"));
assert!(from_three.contains("adr_path varchar(1024) null"));
assert!(!from_three.contains("alter table decisions"));
assert!(from_three.contains("drop column constraints_text"));
let from_two = upgrade_sql(2);
assert!(from_two.contains("create table missions "));
assert!(from_two.contains("create table questions "));
assert!(from_two.contains("alter table threads add column mission_id"));
assert!(!from_two.contains("alter table decisions"));
assert!(from_two.contains("create table mission_constraints "));
assert!(!from_two.contains("constraints_text"));
assert!(!from_two.contains("drop column"));
assert!(from_two.contains("drop table cloud_session_bindings"));
assert_eq!(upgrade_sql(1), from_two);
}
#[test]
fn the_old_text_moves_into_the_list_tables_one_item_per_non_empty_line() {
let rows = vec![
(
"M-A".to_string(),
" first\n\n second \n\t\nthird\r\n".to_string(),
String::new(),
),
("M-B".to_string(), String::new(), "only".to_string()),
("M-C".to_string(), " \n ".to_string(), String::new()),
];
let sql = move_lists_sql(&rows);
assert_eq!(sql.matches("insert into mission_constraints").count(), 1);
assert_eq!(sql.matches("insert into mission_out_of_scope").count(), 1);
assert!(sql.contains("update missions set next_constraint = 4 where id = "));
assert!(sql.contains("update missions set next_out_of_scope = 2 where id = "));
assert_eq!(sql.matches(", 3, ").count(), 1);
assert!(!sql.contains(&lit("M-C")));
let upgrade = upgrade_sql_moving(6, &sql);
let at = |needle: &str| upgrade.find(needle).unwrap();
assert!(at("create table mission_constraints") < at("insert into mission_constraints"));
assert!(at("insert into mission_constraints") < at("drop column constraints_text"));
}
#[test]
fn flag_reads_booleans_and_ones() {
let row: Value = serde_json::from_str("{\"a\":true,\"b\":1,\"c\":0}").unwrap();
assert!(flag(&row, "a"));
assert!(flag(&row, "b"));
assert!(!flag(&row, "c"));
assert!(!flag(&row, "missing"));
}
}