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 diff --git a/src/cli/task.rs b/src/cli/task.rs index b9019b6..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)?; @@ -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::()) } 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::(&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::()) } else { result_str.to_string() }; 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..fa9f1e2 100644 --- a/src/db/dependencies.rs +++ b/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 { )), ) }; - 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..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)] @@ -129,6 +130,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..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_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}; +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}; @@ -383,12 +386,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)?, @@ -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)?; @@ -1036,7 +1092,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 +1143,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 +1321,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)?, @@ -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 4353a50..0a62d86 100644 --- a/src/db/tests.rs +++ b/src/db/tests.rs @@ -1109,3 +1109,112 @@ 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)) + ); +} + +#[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 a2139cc..0170ad2 100755 --- a/tests/functional_test.sh +++ b/tests/functional_test.sh @@ -762,6 +762,52 @@ 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" +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") +$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 "" + +# ───────────────────────────────────────────── +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 "=========================================="