Fix/result types atomic add

.gitignore

@@ -0,0 +1,34 @@
+# Generated by Cargo
+# will have compiled files and executables
+debug
+target
+
+# These are backup files generated by rustfmt
+**/*.rs.bk
+
+# MSVC Windows builds of rustc generate these, which store debugging information
+*.pdb
+
+# Generated by cargo mutants
+# Contains mutation testing data
+**/mutants.out*/
+
+# rustc will dump stack traces when hitting an internal compiler error to PWD
+rustc-ice-*.txt
+
+# RustRover
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+#.idea/
+
+# taskgraph
+*.db
+*.db-shm
+*.db-wal
+*.db-journal
+.taskgraph/
+.env
+.env.*
+!.env.example

src/cli/task.rs

@@ -4,13 +4,13 @@ use crate::cli::{
 };
 use crate::db::{
     add_dependency, add_note, add_task_files, amend_task_description, approve_task,
-    batch_create_tasks, cancel_task, check_file_conflicts, claim_next_task, claim_task,
-    complete_task, compute_effects, create_task, fail_task, fuzzy_find_task, get_handoff_context,
-    get_lookahead, get_task, insert_task_between, list_dependencies, list_due_wakes, list_notes,
-    list_task_files, list_tasks, parse_sleep_duration_ms, pause_task, pivot_subtree, project_state,
-    promote_ready_tasks, remove_dependency, resume_task, sleep_task, snapshot_task_statuses,
-    split_task, start_task, update_heartbeat, update_progress, update_task, Database, NewSubtask,
-    SplitPart, TaskListFilters, TaskgraphError,
+    batch_create_tasks_with_edges, cancel_task, check_file_conflicts, claim_next_task, claim_task,
+    complete_task, compute_effects, create_task, create_task_with_deps, fail_task, fuzzy_find_task,
+    get_handoff_context, get_lookahead, get_task, insert_task_between, list_dependencies,
+    list_due_wakes, list_notes, list_task_files, list_tasks, parse_sleep_duration_ms, pause_task,
+    pivot_subtree, project_state, promote_ready_tasks, remove_dependency, resume_task, sleep_task,
+    snapshot_task_statuses, split_task, start_task, update_heartbeat, update_progress, update_task,
+    Database, NewSubtask, SplitPart, TaskListFilters, TaskgraphError,
 };
 use crate::models::{
     generate_id, DependencyCondition, DependencyKind, RetryBackoff, Task, TaskKind, TaskStatus,
@@ -1056,18 +1056,12 @@ pub fn create_task_cmd(
         updated_at: now,
     };
 
-    let created = create_task(db, &task, &args.tags)?;
-    for dep in &args.deps {
-        let (from_task, kind) = parse_dep_arg(dep)?;
-        add_dependency(
-            db,
-            &from_task,
-            &created.id,
-            kind,
-            DependencyCondition::All,
-            None,
-        )?;
-    }
+    let deps = args
+        .deps
+        .iter()
+        .map(|dep| parse_dep_arg(dep))
+        .collect::<Result<Vec<_>>>()?;
+    let created = create_task_with_deps(db, &task, &args.tags, &deps)?;
     let _ = promote_ready_tasks(db)?;
 
     if json {
@@ -1145,35 +1139,25 @@ fn create_batch_cmd(db: &Database, args: CreateBatchArgs, json: bool) -> Result<
         });
     }
 
-    let inserted = batch_create_tasks(db, &tasks)?;
-
+    let mut edges = Vec::new();
+    let mut tags = Vec::new();
     for (task_id, spec) in &task_specs {
-        if let Some(deps) = &spec.deps {
-            for dep in deps {
-                let from_task = id_aliases
-                    .get(&dep.from)
-                    .cloned()
-                    .unwrap_or_else(|| dep.from.clone());
-                add_dependency(
-                    db,
-                    &from_task,
-                    task_id,
-                    dep.kind.clone().unwrap_or(DependencyKind::FeedsInto),
-                    DependencyCondition::All,
-                    None,
-                )?;
-            }
+        for dep in spec.deps.iter().flatten() {
+            let from_task = id_aliases
+                .get(&dep.from)
+                .cloned()
+                .unwrap_or_else(|| dep.from.clone());
+            edges.push((
+                from_task,
+                task_id.clone(),
+                dep.kind.clone().unwrap_or(DependencyKind::FeedsInto),
+            ));
         }
-        if let Some(tags) = &spec.tags {
-            let conn = db.lock()?;
-            for tag in tags {
-                conn.execute(
-                    "INSERT OR IGNORE INTO task_tags(task_id, tag) VALUES (?1, ?2)",
-                    rusqlite::params![task_id, tag],
-                )?;
-            }
+        for tag in spec.tags.iter().flatten() {
+            tags.push((task_id.clone(), tag.clone()));
         }
     }
+    let inserted = batch_create_tasks_with_edges(db, &tasks, &tags, &edges)?;
 
     let _ = promote_ready_tasks(db)?;
 
@@ -1302,18 +1286,17 @@ pub fn go_cmd(db: &Database, args: &GoArgs, json: bool) -> Result<()> {
                 eprintln!("upstream:");
                 for h in handoff {
                     let from_id = h["from_task"].as_str().unwrap_or("");
-                    let result = h["result"]
-                        .as_str()
-                        .or_else(|| {
-                            if h["result"].is_null() {
-                                None
-                            } else {
-                                Some("(structured)")
-                            }
-                        })
-                        .unwrap_or("(no result)");
-                    let result_display = if result.len() > 120 {
-                        format!("{}…", &result[..117])
+                    let rendered = if h["result"].is_null() {
+                        "(no result)".to_string()
+                    } else {
+                        h["result"]
+                            .as_str()
+                            .map(str::to_string)
+                            .unwrap_or_else(|| h["result"].to_string())
+                    };
+                    let result = rendered.as_str();
+                    let result_display = if result.chars().count() > 120 {
+                        format!("{}…", result.chars().take(117).collect::<String>())
                     } else {
                         result.to_string()
                     };
@@ -1344,9 +1327,9 @@ pub fn go_cmd(db: &Database, args: &GoArgs, json: bool) -> Result<()> {
 pub fn done_cmd(db: &Database, args: DoneArgs, json: bool, compact: bool) -> Result<()> {
     let result_provided = args.result.is_some();
     let result = match args.result {
-        Some(text) => match serde_json::from_str(&text) {
-            Ok(v) => Some(v),
-            Err(_) => Some(serde_json::Value::String(text)),
+        Some(text) => match serde_json::from_str::<serde_json::Value>(&text) {
+            Ok(v @ (serde_json::Value::Object(_) | serde_json::Value::Array(_))) => Some(v),
+            _ => Some(serde_json::Value::String(text)),
         },
         None => None,
     };
@@ -1409,9 +1392,17 @@ pub fn done_cmd(db: &Database, args: DoneArgs, json: bool, compact: bool) -> Res
             if let Some(handoff) = n["handoff"].as_array() {
                 for h in handoff {
                     let from_id = h["from_task"].as_str().unwrap_or("");
-                    let result_str = h["result"].as_str().unwrap_or("(no result)");
-                    let display = if result_str.len() > 120 {
-                        format!("{}…", &result_str[..117])
+                    let rendered = if h["result"].is_null() {
+                        "(no result)".to_string()
+                    } else {
+                        h["result"]
+                            .as_str()
+                            .map(str::to_string)
+                            .unwrap_or_else(|| h["result"].to_string())
+                    };
+                    let result_str = rendered.as_str();
+                    let display = if result_str.chars().count() > 120 {
+                        format!("{}…", result_str.chars().take(117).collect::<String>())
                     } else {
                         result_str.to_string()
                     };

src/db/artifacts.rs

@@ -1,4 +1,4 @@
-use crate::db::{dt_to_sql, json_to_sql, parse_dt, parse_json, Database};
+use crate::db::{dt_to_sql, json_text, json_to_sql, parse_dt, parse_json, Database};
 use crate::models::Artifact;
 use anyhow::Result;
 use rusqlite::params;
@@ -40,7 +40,7 @@ fn row_to_artifact(row: &rusqlite::Row<'_>) -> rusqlite::Result<Artifact> {
             )),
         )
     };
-    let metadata: Option<String> = row.get(8)?;
+    let metadata = json_text(row, 8)?;
     Ok(Artifact {
         id: row.get(0)?,
         task_id: row.get(1)?,

src/db/dependencies.rs

@@ -1,9 +1,9 @@
-use crate::db::{json_to_sql, parse_json, Database};
+use crate::db::{json_text, json_to_sql, parse_json, Database};
 use crate::models::{Dependency, DependencyCondition, DependencyKind};
 use anyhow::Result;
 use rusqlite::params;
 
-const INSERT_DEPENDENCY: &str = r#"
+pub(crate) const INSERT_DEPENDENCY: &str = r#"
 INSERT INTO dependencies(from_task, to_task, kind, condition, metadata)
 VALUES (?1, ?2, ?3, ?4, ?5);
 "#;
@@ -42,7 +42,7 @@ fn row_to_dependency(row: &rusqlite::Row<'_>) -> rusqlite::Result<Dependency> {
             )),
         )
     };
-    let metadata: Option<String> = row.get(5)?;
+    let metadata = json_text(row, 5)?;
     Ok(Dependency {
         id: row.get(0)?,
         from_task: row.get(1)?,

src/db/events.rs

@@ -1,4 +1,4 @@
-use crate::db::{dt_to_sql, json_to_sql, parse_dt, parse_json, Database};
+use crate::db::{dt_to_sql, json_text, json_to_sql, parse_dt, parse_json, Database};
 use crate::models::{Event, EventType};
 use anyhow::Result;
 use chrono::NaiveDateTime;
@@ -44,7 +44,7 @@ fn row_to_event(row: &rusqlite::Row<'_>) -> rusqlite::Result<Event> {
             )),
         )
     };
-    let payload: Option<String> = row.get(5)?;
+    let payload = json_text(row, 5)?;
     Ok(Event {
         id: row.get(0)?,
         task_id: row.get(1)?,

src/db/mod.rs

@@ -43,13 +43,14 @@ pub use projects::{
 pub use schema::init_db;
 pub use sweeper::{run_sweep, SweepResult};
 pub use tasks::{
-    amend_task_description, approve_task, batch_create_tasks, cancel_task, claim_next_task,
-    claim_task, complete_task, create_task, emit_due_wakes, fail_task, fuzzy_find_task,
-    get_handoff_context, get_lookahead, get_task, insert_task_between, list_due_wakes, list_tasks,
-    next_sleep_due_at, parse_sleep_duration_ms, pause_task, pivot_subtree, project_state,
-    promote_ready_tasks, resume_task, sleep_task, split_task, start_task, update_heartbeat,
-    update_progress, update_task, DueWake, HandoffEntry, LookaheadResult, NewSubtask, PivotResult,
-    ProjectState, ResumeResult, SleepResult, SplitPart, SplitResult, TaskListFilters,
+    amend_task_description, approve_task, batch_create_tasks, batch_create_tasks_with_edges,
+    cancel_task, claim_next_task, claim_task, complete_task, create_task, create_task_with_deps,
+    emit_due_wakes, fail_task, fuzzy_find_task, get_handoff_context, get_lookahead, get_task,
+    insert_task_between, list_due_wakes, list_tasks, next_sleep_due_at, parse_sleep_duration_ms,
+    pause_task, pivot_subtree, project_state, promote_ready_tasks, resume_task, sleep_task,
+    split_task, start_task, update_heartbeat, update_progress, update_task, DueWake, HandoffEntry,
+    LookaheadResult, NewSubtask, PivotResult, ProjectState, ResumeResult, SleepResult, SplitPart,
+    SplitResult, TaskListFilters,
 };
 
 #[derive(Debug, Error)]
@@ -129,6 +130,18 @@ pub(crate) fn json_to_sql(value: &Option<Value>) -> Result<Option<String>> {
     }
 }
 
+/// Reads a `JSON` column as text. The type name gives the column NUMERIC affinity, so SQLite stores a numeric
+/// JSON value as INTEGER or REAL, and reading it as `String` fails.
+pub(crate) fn json_text(row: &rusqlite::Row<'_>, idx: usize) -> rusqlite::Result<Option<String>> {
+    use rusqlite::types::ValueRef;
+    Ok(match row.get_ref(idx)? {
+        ValueRef::Null => None,
+        ValueRef::Integer(i) => Some(i.to_string()),
+        ValueRef::Real(f) => Some(f.to_string()),
+        ValueRef::Text(t) | ValueRef::Blob(t) => Some(String::from_utf8_lossy(t).into_owned()),
+    })
+}
+
 pub(crate) fn parse_json(value: Option<String>) -> Result<Option<Value>> {
     match value {
         Some(v) => Ok(Some(serde_json::from_str(&v)?)),

src/db/processes.rs

@@ -1,6 +1,6 @@
 use crate::db::{
-    get_task, json_to_sql, now_utc_naive, parse_dt, parse_json, sleep_dt_to_sql, Database,
-    TaskgraphError,
+    get_task, json_text, json_to_sql, now_utc_naive, parse_dt, parse_json, sleep_dt_to_sql,
+    Database, TaskgraphError,
 };
 use crate::models::{generate_id, Event, EventType, TaskStatus};
 use anyhow::{anyhow, Result};
@@ -264,7 +264,7 @@ fn row_to_process_run(row: &rusqlite::Row<'_>) -> rusqlite::Result<ProcessRun> {
     let ended_at: Option<String> = row.get(11)?;
     let last_heartbeat_at: Option<String> = row.get(12)?;
     let last_output_at: Option<String> = row.get(13)?;
-    let state_ref: Option<String> = row.get(22)?;
+    let state_ref = json_text(row, 22)?;
     Ok(ProcessRun {
         id: row.get(0)?,
         task_id: row.get(1)?,
@@ -340,7 +340,7 @@ fn row_to_task_wait(row: &rusqlite::Row<'_>) -> rusqlite::Result<TaskWait> {
             )),
         )
     };
-    let state_ref: Option<String> = row.get(7)?;
+    let state_ref = json_text(row, 7)?;
     let due_at: Option<String> = row.get(9)?;
     let resumed_at: Option<String> = row.get(10)?;
     Ok(TaskWait {

src/db/projects.rs

@@ -1,5 +1,6 @@
 use crate::db::{
-    dt_to_sql, json_to_sql, now_utc_naive, parse_dt, parse_json, Database, TaskgraphError,
+    dt_to_sql, json_text, json_to_sql, now_utc_naive, parse_dt, parse_json, Database,
+    TaskgraphError,
 };
 use crate::models::{generate_id, Project, ProjectStatus};
 use anyhow::Result;
@@ -69,7 +70,7 @@ WHERE id = ?1;
 "#;
 
 fn row_to_project(row: &rusqlite::Row<'_>) -> Result<Project> {
-    let metadata: Option<String> = row.get(5)?;
+    let metadata = json_text(row, 5)?;
     Ok(Project {
         id: row.get(0)?,
         user_id: row.get(1)?,