From c3e020c963c596f1dad55df31928daa53d4d7cc1 Mon Sep 17 00:00:00 2001 From: Varun S R <97205393+vsraccubits@users.noreply.github.com> Date: Mon, 19 Oct 2026 12:15:04 +0530 Subject: [PATCH 1/5] Read numeric JSON columns as text Columns declared as JSON get NUMERIC affinity in SQLite, so scalar results and metadata such as 500500 or 6.92 are stored as INTEGER or REAL. Reading them back as String failed, breaking get_task, list_tasks and handoff context for tasks with numeric results. Add a json_text helper that reads INTEGER, REAL, TEXT and BLOB values as text, and use it for every JSON column read (task result, metadata, sleep state refs, process/wait state refs, artifact, dependency, event and project JSON fields). Add tests covering scalar and object results through completion and handoff, and numeric task metadata. Co-Authored-By: Claude Opus 5.5 --- src/db/artifacts.rs | 4 +-- src/db/dependencies.rs | 4 +-- src/db/events.rs | 4 +-- src/db/mod.rs | 12 ++++++++ src/db/processes.rs | 8 +++--- src/db/projects.rs | 5 ++-- src/db/tasks.rs | 14 ++++----- src/db/tests.rs | 65 ++++++++++++++++++++++++++++++++++++++++++ 8 files changed, 97 insertions(+), 19 deletions(-) diff --git a/src/db/artifacts.rs b/src/db/artifacts.rs index e1d5d53..71e6f06 100644 --- a/src/db/artifacts.rs +++ b/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 { )), ) }; - let metadata: Option = row.get(8)?; + let metadata = json_text(row, 8)?; Ok(Artifact { id: row.get(0)?, task_id: row.get(1)?, diff --git a/src/db/dependencies.rs b/src/db/dependencies.rs index 6eb0b13..630c564 100644 --- a/src/db/dependencies.rs +++ b/src/db/dependencies.rs @@ -1,4 +1,4 @@ -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; @@ -42,7 +42,7 @@ fn row_to_dependency(row: &rusqlite::Row<'_>) -> rusqlite::Result { )), ) }; - let metadata: Option = row.get(5)?; + let metadata = json_text(row, 5)?; Ok(Dependency { id: row.get(0)?, from_task: row.get(1)?, diff --git a/src/db/events.rs b/src/db/events.rs index e71ce99..1627f3a 100644 --- a/src/db/events.rs +++ b/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 { )), ) }; - let payload: Option = row.get(5)?; + let payload = json_text(row, 5)?; Ok(Event { id: row.get(0)?, task_id: row.get(1)?, diff --git a/src/db/mod.rs b/src/db/mod.rs index 1e25200..c9dc259 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -129,6 +129,18 @@ pub(crate) fn json_to_sql(value: &Option) -> Result> { } } +/// 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> { + 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) -> Result> { match value { Some(v) => Ok(Some(serde_json::from_str(&v)?)), diff --git a/src/db/processes.rs b/src/db/processes.rs index 028dffe..1737e34 100644 --- a/src/db/processes.rs +++ b/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 { let ended_at: Option = row.get(11)?; let last_heartbeat_at: Option = row.get(12)?; let last_output_at: Option = row.get(13)?; - let state_ref: Option = 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 { )), ) }; - let state_ref: Option = row.get(7)?; + let state_ref = json_text(row, 7)?; let due_at: Option = row.get(9)?; let resumed_at: Option = row.get(10)?; Ok(TaskWait { diff --git a/src/db/projects.rs b/src/db/projects.rs index 804bcc5..4eb42d0 100644 --- a/src/db/projects.rs +++ b/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 { - let metadata: Option = row.get(5)?; + let metadata = json_text(row, 5)?; Ok(Project { id: row.get(0)?, user_id: row.get(1)?, diff --git a/src/db/tasks.rs b/src/db/tasks.rs index 14f2b23..8be7c96 100644 --- a/src/db/tasks.rs +++ b/src/db/tasks.rs @@ -1,6 +1,6 @@ use crate::db::dependencies::{add_dependency, remove_dependency}; use crate::db::{ - dt_to_sql, json_to_sql, now_utc_naive, parse_dt, parse_json, sleep_dt_to_sql, + dt_to_sql, json_text, json_to_sql, now_utc_naive, parse_dt, parse_json, sleep_dt_to_sql, truncate_to_millis, Database, TaskgraphError, }; use crate::models::{generate_id, EventType, RetryBackoff, Task, TaskKind, TaskStatus}; @@ -383,12 +383,12 @@ pub(crate) fn row_to_task(row: &rusqlite::Row<'_>) -> rusqlite::Result { let claimed_at: Option = row.get(10)?; let started_at: Option = row.get(11)?; let completed_at: Option = row.get(12)?; - let result: Option = row.get(13)?; + let result = json_text(row, 13)?; let last_heartbeat: Option = row.get(23)?; let sleep_until: Option = row.get(25)?; - let sleep_state_ref: Option = row.get(26)?; + let sleep_state_ref = json_text(row, 26)?; let wake_emitted_at: Option = row.get(28)?; - let metadata: Option = row.get(33)?; + let metadata = json_text(row, 33)?; Ok(Task { id: row.get(0)?, project_id: row.get(1)?, @@ -1036,7 +1036,7 @@ pub fn list_due_wakes(db: &Database, project_id: Option<&str>) -> Result = row.get(3)?; let sleep_until: String = row.get(4)?; - let state_ref: Option = row.get(5)?; + let state_ref = json_text(row, 5)?; let wake_emitted_at: Option = row.get(7)?; wakes.push(DueWake { task_id: row.get(0)?, @@ -1087,7 +1087,7 @@ pub fn emit_due_wakes(db: &Database) -> Result> { let mut rows = stmt.query(params![now_s.clone()])?; let mut wakes = Vec::new(); while let Some(row) = rows.next()? { - let state_ref_raw: Option = row.get(5)?; + let state_ref_raw = json_text(row, 5)?; let sleep_until: String = row.get(4)?; let state_ref = parse_json(state_ref_raw)?; wakes.push(DueWake { @@ -1265,7 +1265,7 @@ pub fn get_handoff_context(db: &Database, task_id: &str) -> Result = row.get(2)?; + let result_raw = json_text(row, 2)?; entries.push(HandoffEntry { from_task_id: row.get(0)?, from_title: row.get(1)?, diff --git a/src/db/tests.rs b/src/db/tests.rs index 4353a50..c520bb7 100644 --- a/src/db/tests.rs +++ b/src/db/tests.rs @@ -1109,3 +1109,68 @@ fn done_rejects_pending() { assert!(msg.contains("is 'pending'")); assert!(msg.contains("ready/claimed/running")); } + +fn project_tasks(db: &Database, project_id: &str) -> Vec { + let filters = TaskListFilters { + project_id: Some(project_id.to_owned()), + ..Default::default() + }; + list_tasks(db, filters).unwrap() +} + +#[test] +fn scalar_and_object_results_read_back_and_hand_off() { + let db_path = test_db_path(); + let db = init_db(&db_path).unwrap(); + let project = create_project(&db, "ResultTypes", None, None, None).unwrap(); + + for result in [json!(500500), json!(6.92), json!(true), json!({"sum": 5})] { + let upstream = create_task( + &db, + &make_task(&project.id, "upstream", TaskStatus::Ready), + &[], + ) + .unwrap(); + let downstream = create_task( + &db, + &make_task(&project.id, "downstream", TaskStatus::Pending), + &[], + ) + .unwrap(); + add_dependency( + &db, + &upstream.id, + &downstream.id, + DependencyKind::FeedsInto, + DependencyCondition::All, + None, + ) + .unwrap(); + complete_task(&db, &upstream.id, Some(result.clone())).unwrap(); + + assert_eq!( + get_task(&db, &upstream.id).unwrap().result, + Some(result.clone()) + ); + let handoff = get_handoff_context(&db, &downstream.id).unwrap(); + assert_eq!(handoff.len(), 1); + assert_eq!(handoff[0].result, Some(result)); + } + assert_eq!(project_tasks(&db, &project.id).len(), 8); +} + +#[test] +fn numeric_metadata_reads_back() { + let db_path = test_db_path(); + let db = init_db(&db_path).unwrap(); + let project = create_project(&db, "NumericMetadata", None, None, None).unwrap(); + let mut task = make_task(&project.id, "with metadata", TaskStatus::Ready); + task.metadata = Some(json!(42)); + + let created = create_task(&db, &task, &[]).unwrap(); + + assert_eq!( + get_task(&db, &created.id).unwrap().metadata, + Some(json!(42)) + ); +} From 0ba5523f51a584bae68719c0d36fea42981927ab Mon Sep 17 00:00:00 2001 From: Varun S R <97205393+vsraccubits@users.noreply.github.com> Date: Mon, 19 Oct 2026 12:29:51 +0530 Subject: [PATCH 2/5] Show non-string handoff results in go and done output The upstream handoff display only rendered string results: done printed "(no result)" for numeric, boolean or object results, and go printed "(structured)". Both truncated long results by byte index, which panics when the cut lands inside a multi-byte UTF-8 character. Render non-string results as compact JSON, keep "(no result)" for null, and truncate by characters instead of bytes. Add functional tests for numeric, object and long non-ASCII results handed off through done --next. Co-Authored-By: Claude Opus 5.5 --- src/cli/task.rs | 37 ++++++++++++++++++++++--------------- tests/functional_test.sh | 31 +++++++++++++++++++++++++++++++ 2 files changed, 53 insertions(+), 15 deletions(-) diff --git a/src/cli/task.rs b/src/cli/task.rs index b9019b6..4b2a5da 100644 --- a/src/cli/task.rs +++ b/src/cli/task.rs @@ -1302,18 +1302,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::()) } else { result.to_string() }; @@ -1409,9 +1408,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::()) } else { result_str.to_string() }; diff --git a/tests/functional_test.sh b/tests/functional_test.sh index a2139cc..03b095b 100755 --- a/tests/functional_test.sh +++ b/tests/functional_test.sh @@ -762,6 +762,37 @@ assert_contains "task split response has effect" '"effect"' "$SPLIT_OUT" echo "" +# ───────────────────────────────────────────── +echo "RESULT TYPES" + +RT_PROJECT=$($TASKGRAPH --db "$DB" --json project create "result-types") +RT_PROJECT_ID=$(jq_field "$RT_PROJECT" "id") +RT_A=$(jq_field "$($TASKGRAPH --db "$DB" --json task create --project "$RT_PROJECT_ID" --title "RT A")" "id") +RT_B=$(jq_field "$($TASKGRAPH --db "$DB" --json task create --project "$RT_PROJECT_ID" --title "RT B" --dep "$RT_A")" "id") +RT_C=$(jq_field "$($TASKGRAPH --db "$DB" --json task create --project "$RT_PROJECT_ID" --title "RT C" --dep "$RT_B")" "id") +$TASKGRAPH --db "$DB" go --project "$RT_PROJECT_ID" --agent rt >/dev/null +NUM_OUT=$($TASKGRAPH --db "$DB" done "$RT_A" --result 500500 --next --agent rt 2>&1) +assert_contains "numeric result is handed off" "500500" "$NUM_OUT" +RT_LIST=$($TASKGRAPH --db "$DB" --json list --project "$RT_PROJECT_ID") +assert_eq "list reads a project holding numeric results" "3" "$(jq_len "$RT_LIST")" +LONG_RESULT="Größenänderung über alle Prüfungen hinweg — äöüß äöüß äöüß äöüß äöüß äöüß äöüß äöüß äöüß äöüß äöüß äöüß" +NA_PROJECT_ID=$(jq_field "$($TASKGRAPH --db "$DB" --json project create "non-ascii-results")" "id") +NA_A=$(jq_field "$($TASKGRAPH --db "$DB" --json task create --project "$NA_PROJECT_ID" --title "NA A")" "id") +$TASKGRAPH --db "$DB" --json task create --project "$NA_PROJECT_ID" --title "NA B" --dep "$NA_A" >/dev/null +$TASKGRAPH --db "$DB" go --project "$NA_PROJECT_ID" --agent rt >/dev/null +NA_EXIT=0 +$TASKGRAPH --db "$DB" done "$NA_A" --result "$LONG_RESULT" --next --agent rt >/dev/null 2>&1 || NA_EXIT=$? +assert_eq "long non-ASCII result is handed off" "0" "$NA_EXIT" + +OBJ_PROJECT_ID=$(jq_field "$($TASKGRAPH --db "$DB" --json project create "object-results")" "id") +OBJ_A=$(jq_field "$($TASKGRAPH --db "$DB" --json task create --project "$OBJ_PROJECT_ID" --title "OBJ A")" "id") +$TASKGRAPH --db "$DB" --json task create --project "$OBJ_PROJECT_ID" --title "OBJ B" --dep "$OBJ_A" >/dev/null +$TASKGRAPH --db "$DB" go --project "$OBJ_PROJECT_ID" --agent rt >/dev/null +OBJ_OUT=$($TASKGRAPH --db "$DB" done "$OBJ_A" --result '{"sum": 5}' --next --agent rt 2>&1) +assert_contains "object result is shown downstream" '{"sum":5}' "$OBJ_OUT" + +echo "" + # ───────────────────────────────────────────── # SUMMARY echo "==========================================" From 6c35c9271419aefc8f8bb9732f23fe1e6874d85f Mon Sep 17 00:00:00 2001 From: Varun S R <97205393+vsraccubits@users.noreply.github.com> Date: Mon, 19 Oct 2026 12:38:23 +0530 Subject: [PATCH 3/5] Keep scalar done results as written done --result parsed any text that was valid JSON, so scalar results lost their original form: 6.90 was stored as 6.9, and "null" cleared the result entirely. Only parse objects and arrays as JSON; store everything else as the string the caller passed. Add functional tests that a float result keeps its text in done output and in show. Co-Authored-By: Claude Opus 5.5 --- src/cli/task.rs | 6 +++--- tests/functional_test.sh | 3 +++ 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/src/cli/task.rs b/src/cli/task.rs index 4b2a5da..b4c297c 100644 --- a/src/cli/task.rs +++ b/src/cli/task.rs @@ -1343,9 +1343,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::(&text) { + Ok(v @ (serde_json::Value::Object(_) | serde_json::Value::Array(_))) => Some(v), + _ => Some(serde_json::Value::String(text)), }, None => None, }; diff --git a/tests/functional_test.sh b/tests/functional_test.sh index 03b095b..9d8033a 100755 --- a/tests/functional_test.sh +++ b/tests/functional_test.sh @@ -773,8 +773,11 @@ RT_C=$(jq_field "$($TASKGRAPH --db "$DB" --json task create --project "$RT_PROJE $TASKGRAPH --db "$DB" go --project "$RT_PROJECT_ID" --agent rt >/dev/null NUM_OUT=$($TASKGRAPH --db "$DB" done "$RT_A" --result 500500 --next --agent rt 2>&1) assert_contains "numeric result is handed off" "500500" "$NUM_OUT" +FLOAT_OUT=$($TASKGRAPH --db "$DB" done "$RT_B" --result 6.90 --next --agent rt 2>&1) +assert_contains "float result keeps its text" "6.90" "$FLOAT_OUT" RT_LIST=$($TASKGRAPH --db "$DB" --json list --project "$RT_PROJECT_ID") assert_eq "list reads a project holding numeric results" "3" "$(jq_len "$RT_LIST")" +assert_eq "scalar result is stored as written" "6.90" "$(jq_field "$($TASKGRAPH --db "$DB" --json show "$RT_B")" "result")" LONG_RESULT="Größenänderung über alle Prüfungen hinweg — äöüß äöüß äöüß äöüß äöüß äöüß äöüß äöüß äöüß äöüß äöüß äöüß" NA_PROJECT_ID=$(jq_field "$($TASKGRAPH --db "$DB" --json project create "non-ascii-results")" "id") NA_A=$(jq_field "$($TASKGRAPH --db "$DB" --json task create --project "$NA_PROJECT_ID" --title "NA A")" "id") From 5b399fc98f8ede0fbb8ed321c0a17d866d120128 Mon Sep 17 00:00:00 2001 From: Varun S R <97205393+vsraccubits@users.noreply.github.com> Date: Mon, 19 Oct 2026 12:47:57 +0530 Subject: [PATCH 4/5] Create tasks and their dependencies atomically task add and batch creation inserted the task rows first and added dependencies afterwards in separate statements. An unknown dependency id failed the command after the task was already written, leaving an orphan task with partial edges and tags behind. Add create_task_with_deps and batch_create_tasks_with_edges, which insert tasks, tags and dependency edges in one transaction. Each edge checks that its upstream task exists and fails with a NotFound error naming the id, which rolls back the whole call. create_task and batch_create_tasks delegate to the new functions, and the shared INSERT_TASK binding moves into insert_task_row. Add unit tests that single and batch creation with an unknown dependency create nothing, and a functional test for task add. Co-Authored-By: Claude Opus 5.5 --- src/cli/task.rs | 72 +++++-------- src/db/dependencies.rs | 2 +- src/db/mod.rs | 15 +-- src/db/tasks.rs | 219 ++++++++++++++++++++++----------------- src/db/tests.rs | 44 ++++++++ tests/functional_test.sh | 12 +++ 6 files changed, 216 insertions(+), 148 deletions(-) diff --git a/src/cli/task.rs b/src/cli/task.rs index b4c297c..204e1a3 100644 --- a/src/cli/task.rs +++ b/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::>>()?; + 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)?; diff --git a/src/db/dependencies.rs b/src/db/dependencies.rs index 630c564..fa9f1e2 100644 --- a/src/db/dependencies.rs +++ b/src/db/dependencies.rs @@ -3,7 +3,7 @@ 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); "#; diff --git a/src/db/mod.rs b/src/db/mod.rs index c9dc259..779b2c3 100644 --- a/src/db/mod.rs +++ b/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)] diff --git a/src/db/tasks.rs b/src/db/tasks.rs index 8be7c96..5c4496d 100644 --- a/src/db/tasks.rs +++ b/src/db/tasks.rs @@ -1,12 +1,15 @@ -use crate::db::dependencies::{add_dependency, remove_dependency}; +use crate::db::dependencies::{add_dependency, remove_dependency, INSERT_DEPENDENCY}; use crate::db::{ dt_to_sql, json_text, json_to_sql, now_utc_naive, parse_dt, parse_json, sleep_dt_to_sql, truncate_to_millis, Database, TaskgraphError, }; -use crate::models::{generate_id, EventType, RetryBackoff, Task, TaskKind, TaskStatus}; +use crate::models::{ + generate_id, DependencyCondition, DependencyKind, EventType, RetryBackoff, Task, TaskKind, + TaskStatus, +}; use anyhow::{anyhow, Result}; use chrono::Duration; -use rusqlite::{params, OptionalExtension}; +use rusqlite::{params, Connection, OptionalExtension}; use serde_json::{Map, Value}; use std::collections::{HashMap, HashSet, VecDeque}; @@ -448,7 +451,17 @@ pub(crate) fn row_to_task(row: &rusqlite::Row<'_>) -> rusqlite::Result { } pub fn create_task(db: &Database, task: &Task, tags: &[String]) -> Result { - let conn = db.lock()?; + create_task_with_deps(db, task, tags, &[]) +} + +/// Creates a task with its tags and incoming dependencies in one transaction: a missing dependency task fails the +/// whole call and leaves nothing behind. +pub fn create_task_with_deps( + db: &Database, + task: &Task, + tags: &[String], + deps: &[(String, DependencyKind)], +) -> Result { let mut task_with_defaults = task.clone(); if task_with_defaults.max_retries < 0 { task_with_defaults.max_retries = 0; @@ -463,53 +476,16 @@ pub fn create_task(db: &Database, task: &Task, tags: &[String]) -> Result task_with_defaults.heartbeat_interval = 30; } let task_id = task_with_defaults.id.clone(); - let result = json_to_sql(&task_with_defaults.result)?; - let sleep_state_ref = json_to_sql(&task_with_defaults.sleep_state_ref)?; - let metadata = json_to_sql(&task_with_defaults.metadata)?; - conn.execute( - INSERT_TASK, - params![ - &task_with_defaults.id, - &task_with_defaults.project_id, - &task_with_defaults.parent_task_id, - task_with_defaults.is_composite, - &task_with_defaults.title, - &task_with_defaults.description, - &task_with_defaults.status, - &task_with_defaults.kind, - task_with_defaults.priority, - &task_with_defaults.agent_id, - task_with_defaults.claimed_at.map(dt_to_sql), - task_with_defaults.started_at.map(dt_to_sql), - task_with_defaults.completed_at.map(dt_to_sql), - &result, - &task_with_defaults.error, - task_with_defaults.progress, - &task_with_defaults.progress_note, - task_with_defaults.max_retries, - task_with_defaults.retry_count, - &task_with_defaults.retry_backoff, - task_with_defaults.retry_delay_ms, - task_with_defaults.timeout_seconds, - task_with_defaults.heartbeat_interval, - task_with_defaults.last_heartbeat.map(dt_to_sql), - &task_with_defaults.sleep_id, - task_with_defaults.sleep_until.map(dt_to_sql), - &sleep_state_ref, - &task_with_defaults.sleep_reason, - task_with_defaults.wake_emitted_at.map(dt_to_sql), - task_with_defaults.requires_approval, - &task_with_defaults.approval_status, - &task_with_defaults.approved_by, - &task_with_defaults.approval_comment, - &metadata, - dt_to_sql(task_with_defaults.created_at), - dt_to_sql(task_with_defaults.updated_at) - ], - )?; + let mut conn = db.lock()?; + let tx = conn.transaction()?; + insert_task_row(&tx, &task_with_defaults)?; for tag in tags { - conn.execute(INSERT_TASK_TAG, params![&task_id, tag])?; + tx.execute(INSERT_TASK_TAG, params![&task_id, tag])?; + } + for (from_task, kind) in deps { + insert_edge(&tx, from_task, &task_id, kind)?; } + tx.commit()?; drop(conn); let task = get_task(db, &task_id)?; let _ = crate::db::insert_event( @@ -524,6 +500,86 @@ pub fn create_task(db: &Database, task: &Task, tags: &[String]) -> Result Ok(task) } +fn insert_task_row(conn: &Connection, task: &Task) -> Result<()> { + let result = json_to_sql(&task.result)?; + let sleep_state_ref = json_to_sql(&task.sleep_state_ref)?; + let metadata = json_to_sql(&task.metadata)?; + conn.execute( + INSERT_TASK, + params![ + &task.id, + &task.project_id, + &task.parent_task_id, + task.is_composite, + &task.title, + &task.description, + &task.status, + &task.kind, + task.priority, + &task.agent_id, + task.claimed_at.map(dt_to_sql), + task.started_at.map(dt_to_sql), + task.completed_at.map(dt_to_sql), + &result, + &task.error, + task.progress, + &task.progress_note, + task.max_retries, + task.retry_count, + &task.retry_backoff, + task.retry_delay_ms, + task.timeout_seconds, + task.heartbeat_interval, + task.last_heartbeat.map(dt_to_sql), + &task.sleep_id, + task.sleep_until.map(dt_to_sql), + &sleep_state_ref, + &task.sleep_reason, + task.wake_emitted_at.map(dt_to_sql), + task.requires_approval, + &task.approval_status, + &task.approved_by, + &task.approval_comment, + &metadata, + dt_to_sql(task.created_at), + dt_to_sql(task.updated_at) + ], + )?; + Ok(()) +} + +fn insert_edge( + conn: &Connection, + from_task: &str, + to_task: &str, + kind: &DependencyKind, +) -> Result<()> { + let exists: Option = conn + .query_row( + "SELECT 1 FROM tasks WHERE id = ?1", + params![from_task], + |row| row.get(0), + ) + .optional()?; + if exists.is_none() { + return Err(TaskgraphError::NotFound(format!( + "dependency task {from_task}; nothing was created. Check the id with `taskgraph list`" + )) + .into()); + } + conn.execute( + INSERT_DEPENDENCY, + params![ + from_task, + to_task, + kind, + DependencyCondition::All, + None:: + ], + )?; + Ok(()) +} + pub fn get_task(db: &Database, task_id: &str) -> Result { let conn = db.lock()?; let mut stmt = conn.prepare(SELECT_TASK_BY_ID)?; @@ -1277,58 +1333,29 @@ pub fn get_handoff_context(db: &Database, task_id: &str) -> Result Result { + batch_create_tasks_with_edges(db, tasks, &[], &[]) +} + +/// Creates tasks, their tags and their dependencies in one transaction. +pub fn batch_create_tasks_with_edges( + db: &Database, + tasks: &[Task], + tags: &[(String, String)], + edges: &[(String, String, DependencyKind)], +) -> Result { let mut conn = db.lock()?; let tx = conn.transaction()?; - let mut inserted = 0usize; for task in tasks { - let result = json_to_sql(&task.result)?; - let sleep_state_ref = json_to_sql(&task.sleep_state_ref)?; - let metadata = json_to_sql(&task.metadata)?; - tx.execute( - INSERT_TASK, - params![ - &task.id, - &task.project_id, - &task.parent_task_id, - task.is_composite, - &task.title, - &task.description, - &task.status, - &task.kind, - task.priority, - &task.agent_id, - task.claimed_at.map(dt_to_sql), - task.started_at.map(dt_to_sql), - task.completed_at.map(dt_to_sql), - &result, - &task.error, - task.progress, - &task.progress_note, - task.max_retries, - task.retry_count, - &task.retry_backoff, - task.retry_delay_ms, - task.timeout_seconds, - task.heartbeat_interval, - task.last_heartbeat.map(dt_to_sql), - &task.sleep_id, - task.sleep_until.map(dt_to_sql), - &sleep_state_ref, - &task.sleep_reason, - task.wake_emitted_at.map(dt_to_sql), - task.requires_approval, - &task.approval_status, - &task.approved_by, - &task.approval_comment, - &metadata, - dt_to_sql(task.created_at), - dt_to_sql(task.updated_at) - ], - )?; - inserted += 1; + insert_task_row(&tx, task)?; + } + for (task_id, tag) in tags { + tx.execute(INSERT_TASK_TAG, params![task_id, tag])?; + } + for (from_task, to_task, kind) in edges { + insert_edge(&tx, from_task, to_task, kind)?; } tx.commit()?; - Ok(inserted) + Ok(tasks.len()) } /// Returns (task_id, title) pairs for tasks promoted from pending → ready. diff --git a/src/db/tests.rs b/src/db/tests.rs index c520bb7..0a62d86 100644 --- a/src/db/tests.rs +++ b/src/db/tests.rs @@ -1174,3 +1174,47 @@ fn numeric_metadata_reads_back() { Some(json!(42)) ); } + +#[test] +fn add_with_unknown_dependency_creates_nothing() { + let db_path = test_db_path(); + let db = init_db(&db_path).unwrap(); + let project = create_project(&db, "AtomicAdd", None, None, None).unwrap(); + + let err = create_task_with_deps( + &db, + &make_task(&project.id, "orphan", TaskStatus::Pending), + &["tag-a".to_owned()], + &[("t-bogus1".to_owned(), DependencyKind::FeedsInto)], + ) + .unwrap_err(); + + assert!(err.to_string().contains("t-bogus1"), "{err}"); + assert!(project_tasks(&db, &project.id).is_empty()); +} + +#[test] +fn batch_with_unknown_dependency_creates_nothing() { + let db_path = test_db_path(); + let db = init_db(&db_path).unwrap(); + let project = create_project(&db, "AtomicBatch", None, None, None).unwrap(); + let first = make_task(&project.id, "first", TaskStatus::Pending); + let second = make_task(&project.id, "second", TaskStatus::Pending); + let edges = vec![ + ( + first.id.clone(), + second.id.clone(), + DependencyKind::FeedsInto, + ), + ( + "t-bogus2".to_owned(), + second.id.clone(), + DependencyKind::FeedsInto, + ), + ]; + + let err = batch_create_tasks_with_edges(&db, &[first, second], &[], &edges).unwrap_err(); + + assert!(err.to_string().contains("t-bogus2"), "{err}"); + assert!(project_tasks(&db, &project.id).is_empty()); +} diff --git a/tests/functional_test.sh b/tests/functional_test.sh index 9d8033a..0170ad2 100755 --- a/tests/functional_test.sh +++ b/tests/functional_test.sh @@ -796,6 +796,18 @@ assert_contains "object result is shown downstream" '{"sum":5}' "$OBJ_OUT" echo "" +# ───────────────────────────────────────────── +echo "ATOMIC ADD" + +ADD_BEFORE=$(jq_len "$($TASKGRAPH --db "$DB" --json list --project "$RT_PROJECT_ID")") +ADD_EXIT=0 +ADD_OUT=$($TASKGRAPH --db "$DB" add --project "$RT_PROJECT_ID" --title "Orphan" --dep t-bogus1 2>&1) || ADD_EXIT=$? +assert_regex "add with an unknown dependency fails" "^[1-9]" "$ADD_EXIT" +assert_contains "add names the unknown dependency" "t-bogus1" "$ADD_OUT" +assert_eq "add with an unknown dependency creates nothing" "$ADD_BEFORE" "$(jq_len "$($TASKGRAPH --db "$DB" --json list --project "$RT_PROJECT_ID")")" + +echo "" + # ───────────────────────────────────────────── # SUMMARY echo "==========================================" From 238e87a98021c5c33c420ff0bfcbb020f95a60d1 Mon Sep 17 00:00:00 2001 From: Varun S R <97205393+vsraccubits@users.noreply.github.com> Date: Mon, 19 Oct 2026 12:49:24 +0530 Subject: [PATCH 5/5] Add .gitignore Ignore Cargo build output and Rust tooling artifacts (rustfmt backups, MSVC debug info, cargo-mutants output, rustc ICE dumps), plus local taskgraph state: SQLite databases with their WAL, SHM and journal files, the .taskgraph/ directory, and .env files other than .env.example. Co-Authored-By: Claude Opus 5.5 --- .gitignore | 34 ++++++++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) create mode 100644 .gitignore diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..f276b5a --- /dev/null +++ b/.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