Skip to content

Define the task types once and route the typed task update through the shared pipeline (closes #3441) #3470

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
malibio merged 6 commits into from
Oct 1, 2026
Merged

Define the task types once and route the typed task update through the shared pipeline (closes #3441) #3470

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
8c6f916
Define task types once in nodespace-types; route the typed task updat…
malibio Oct 1, 2026
13edc76
Typed task update carries schema fields only across proto, daemon, Ta…
malibio Oct 1, 2026
55f578e
Refresh comments that named the removed task write path
malibio Oct 1, 2026
eee4e29
Address review: typed updates refuse unknown keys on every route
malibio Oct 1, 2026
cc197bc
Address review: match own keys only in the typed update field check
malibio Oct 1, 2026
0256824
Address re-review: hold the typed update field lists to the keys the …
malibio Oct 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions packages/agent/src/local_agent/tools.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3760,8 +3760,8 @@ impl GraphToolExecutor {
// (`core_schemas.rs`), and ADR-076's `add_field_values` lets a
// methodology bundle append real values to it at install time
// (`backlog`, `in_review`, ...). A literal match here would reject
// exactly those values while the service layer
// (`node_service::schema::validate_task_status`) accepts them —
// exactly those values while the service layer (the update
// pipeline's enum check) accepts them —
// leaving the agent unable to write a status the schema declares
// valid.
//
Expand Down
2 changes: 1 addition & 1 deletion packages/cli/src/commands/node.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -319,7 +319,7 @@ async fn set_status(client: &mut NodeClient, args: SetStatusArgs, json: bool) ->
// not a list baked into this binary — is the only source of truth for
// which values are valid. An invalid value comes back from the RPC
// below as an InvalidArgument status naming the value and the current
// list (`NodeService::validate_task_status`).
// list (the update pipeline's enum check).
let properties = json!({ "status": args.status }).to_string();

let response = client
Expand Down
2 changes: 1 addition & 1 deletion packages/cli/tests/it/cli_integration.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2147,7 +2147,7 @@ async fn node_set_status_rejects_invalid_status() {
.node_id;

// No CLI-side vocabulary check remains, so this exercises the daemon's
// live-schema validation (`NodeService::validate_task_status`) via the
// live-schema validation (the update pipeline's enum check) via the
// RPC — same error-mapping path as `schema_create_rejects_malformed_params`.
let err = commands::node::run(
&mut client,
Expand Down
107 changes: 8 additions & 99 deletions packages/core/src/behaviors/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -704,9 +704,8 @@ impl NodeBehavior for ProjectNodeBehavior {

// Typed fields live under `properties.project.*` — `project` is a
// schema-typed node type, so NodeService hoists its schema-defined
// fields there on write, the same as `task` under `properties.task.*`
// (see `TaskNode::from_node`, which uses the same nested-first/
// flat-fallback lookup for the same reason). Falling back to the flat
// fields there on write, the same as `task` under `properties.task.*`.
// Falling back to the flat
// top level keeps this correct for a node that was constructed
// directly. TYPE checks only — the schema system validates enum
// membership and allowed values.
Expand Down Expand Up @@ -2539,7 +2538,6 @@ impl Default for NodeBehaviorRegistry {
#[cfg(test)]
mod tests {
use super::*;
use crate::models::TaskNode;
use serde_json::json;

#[test]
Expand Down Expand Up @@ -3015,8 +3013,8 @@ mod tests {
assert!(behavior.validate(&complete_node).is_ok());

// Priority is a string enum. A non-string priority is not a supported
// format: `from_node` reads it with `as_str()`, so it is ignored and the
// task falls back to having no priority rather than failing validation.
// format: the typed conversion reads it as a string, so it is ignored and
// the task falls back to having no priority rather than failing validation.
let integer_priority_node = Node::new(
"task".to_string(),
"Task with non-string priority".to_string(),
Expand All @@ -3026,9 +3024,9 @@ mod tests {
// any task-typed node validates. The assertion below is the load-bearing
// one: the value is dropped, not interpreted.
assert!(behavior.validate(&integer_priority_node).is_ok());
let parsed = TaskNode::from_node(integer_priority_node).unwrap();
let typed = crate::models::node_to_typed_value(integer_priority_node).unwrap();
assert!(
parsed.priority.is_none(),
typed.get("priority").is_none(),
"non-string priority should be ignored, not interpreted as a legacy format"
);

Expand All @@ -3055,8 +3053,8 @@ mod tests {
"Task".to_string(),
json!({"task": {"status": 123}}), // number instead of string
);
// Note: TaskNode::from_node() ignores invalid status and uses default (Open)
// This is graceful degradation - it passes validation
// The typed conversion ignores an invalid status and uses the default
// (open). This is graceful degradation - it passes validation
assert!(behavior.validate(&bad_status_type).is_ok());

// Priority is now a string enum (highest, high, medium, low, lowest) with user-extensibility
Expand Down Expand Up @@ -4514,95 +4512,6 @@ mod tests {
));
}

// =========================================================================
// TaskNode from_node Format Tests
// =========================================================================

#[test]
fn test_task_node_from_node_nested_format() {
use crate::models::{Priority, TaskNode};

// Test that TaskNode::from_node handles nested format
// Priority now uses string enum format
let nested_node = Node::new(
"task".to_string(),
"Nested format task".to_string(),
json!({
"task": {
"status": "in_progress",
"priority": "medium"
}
}),
);

let task = TaskNode::from_node(nested_node).unwrap();
assert_eq!(task.content, "Nested format task");
assert_eq!(task.status.as_str(), "in_progress");
assert_eq!(task.priority, Some(Priority::Medium));
}

#[test]
fn test_task_node_from_node_flat_format() {
use crate::models::{Priority, TaskNode};

// Test flat format (old property structure, but string priority)
let flat_node = Node::new(
"task".to_string(),
"Flat format task".to_string(),
json!({
"status": "done",
"priority": "low"
}),
);

let task = TaskNode::from_node(flat_node).unwrap();
assert_eq!(task.content, "Flat format task");
assert_eq!(task.status.as_str(), "done");
assert_eq!(task.priority, Some(Priority::Low));
}

#[test]
fn test_task_node_from_node_string_priority() {
use crate::models::{Priority, TaskNode};

// Test string priority - now the canonical format
let string_priority_node = Node::new(
"task".to_string(),
"High priority task".to_string(),
json!({
"task": {
"status": "open",
"priority": "high"
}
}),
);

let task = TaskNode::from_node(string_priority_node).unwrap();
assert_eq!(task.priority, Some(Priority::High));

// Test core string priorities
let low_node = Node::new(
"task".to_string(),
"Low".to_string(),
json!({"priority": "low"}),
);
let low_task = TaskNode::from_node(low_node).unwrap();
assert_eq!(low_task.priority, Some(Priority::Low));

// Test user-defined priority (schema extensibility)
let urgent_node = Node::new(
"task".to_string(),
"Urgent".to_string(),
json!({"priority": "urgent"}),
);
let urgent_task = TaskNode::from_node(urgent_node).unwrap();
assert_eq!(
urgent_task.priority,
Some(Priority::User("urgent".to_string())),
"Unknown priorities become User-defined"
);
}

#[test]
fn test_query_node_behavior_validation() {
let behavior = QueryNodeBehavior;
Expand Down
18 changes: 1 addition & 17 deletions packages/core/src/db/sqlite_store/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
use crate::db::fractional_ordering::FractionalOrderCalculator;
use crate::models::{DeleteResult, Node, NodeQuery, NodeUpdate, OrderBy};
use anyhow::{Context, Result};
use chrono::{DateTime, NaiveDate, Utc};
use chrono::{DateTime, Utc};
use connections::{Connections, ReadConn, WriteGuard};
use serde_json::Value;
use std::collections::HashMap;
Expand All @@ -11,22 +11,6 @@ use std::sync::Arc;
use tokio::sync::broadcast;
use tx::Tx;

/// Normalise a stored date string to YYYY-MM-DD on read.
/// Accepts YYYY-MM-DD (pass-through) or RFC 3339 (extract date portion).
/// Returns the original string for unrecognised values so callers can surface them.
fn normalize_date_field(s: &str) -> String {
if NaiveDate::parse_from_str(s, "%Y-%m-%d").is_ok() {
return s.to_string();
}
if let Ok(dt) = DateTime::parse_from_rfc3339(s) {
return dt.format("%Y-%m-%d").to_string();
}
if let Ok(dt) = s.parse::<DateTime<Utc>>() {
return dt.format("%Y-%m-%d").to_string();
}
s.to_string()
}

const DOMAIN_EVENT_CHANNEL_CAPACITY: usize = 128;
/// KNN over-fetch factor: vec0 returns top-k *chunks*, but search results are grouped
/// by node and many chunks map to one node, so we fetch `limit * this` to cover enough
Expand Down
161 changes: 0 additions & 161 deletions packages/core/src/db/sqlite_store/nodes.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3779,167 +3779,6 @@ impl SqliteStore {
Ok(id)
}

/// Convert a raw [`Node`] into a [`crate::models::TaskNode`], reading its
/// typed fields back out of the namespaced `properties.task.*` storage
/// shape (the inverse of [`crate::models::TaskNodeUpdate::apply_to_properties`]).
/// Returns `None` for a non-task node. Shared by [`Self::get_task_node`]
/// and [`Self::update_task_node_with_version_check_in_tx`] so the two
/// read paths can never parse the stored shape differently.
pub(crate) fn node_to_task_node(n: Node) -> Option<crate::models::TaskNode> {
// The typed task shape is `task`'s own: a subtype is never converted
// to its base's struct (ADR-086 §5).
if !crate::models::CoreNodeType::Task.is_exactly(&n.node_type) {
return None;
}
let props = &n.properties;
let task_props = props.get("task").cloned().unwrap_or(serde_json::json!({}));
Some(crate::models::TaskNode {
id: n.id,
node_type: n.node_type,
content: n.content,
version: n.version,
created_at: n.created_at,
modified_at: n.modified_at,
properties: n.properties,
status: task_props
.get("status")
.and_then(|v| v.as_str())
.and_then(|s| s.parse().ok())
.unwrap_or_default(),
priority: task_props
.get("priority")
.and_then(|v| v.as_str())
.and_then(|s| s.parse().ok()),
due_date: task_props
.get("due_date")
.and_then(|v| v.as_str())
.map(normalize_date_field),
started_at: task_props
.get("started_at")
.and_then(|v| v.as_str())
.map(normalize_date_field),
completed_at: task_props
.get("completed_at")
.and_then(|v| v.as_str())
.map(normalize_date_field),
})
}

pub async fn get_task_node(&self, id: &str) -> Result<Option<crate::models::TaskNode>> {
let node = self.get_node(id).await?;
Ok(node.and_then(Self::node_to_task_node))
}

/// `_in_tx` counterpart of the removed non-tx `update_task_node` (ADR-060
/// §2), same read-check-write shape as
/// [`Self::update_node_with_version_check_in_tx`]: the pre-read and the
/// version-gated `UPDATE` both run against `tx`, so the OCC check is
/// sound against this transaction's own view — no TOCTOU window between
/// reading `current` and writing. The `WHERE ... AND version = ?` guard
/// on the `UPDATE` is new relative to the removed method (which checked
/// the version only via a separate, non-atomic pre-read): a strict
/// correctness improvement that comes for free from running inside a
/// real transaction, matching the generic node-update path's own guard.
///
/// Returns `Ok(Err(actual_version))` on a version mismatch rather than
/// `Err` — an expected, common outcome the caller maps to
/// `NodeServiceError::VersionConflict` itself, exactly like
/// `update_node_with_version_check_in_tx`'s own convention (ADR-069
/// §2a: a version mismatch is not a transaction failure). `actual_version`
/// is the real persisted version observed by this call, not a placeholder
/// — an improvement over the removed method, which could only ever report
/// `0` for a task-node conflict.
///
/// Returns the raw [`Node`], not [`crate::models::TaskNode`] — the
/// caller (`NodeService::update_task_node_in_tx`) needs the namespaced
/// `properties.task.*` shape for its own changed-properties diff and
/// invariant-rule dispatch, not `TaskNode::into_node`'s flattened wire
/// shape (that shape exists for the RPC boundary only, never storage —
/// see that method's own doc).
pub(crate) async fn update_task_node_with_version_check_in_tx(
tx: &Tx<'_>,
id: &str,
expected_version: i64,
update: crate::models::TaskNodeUpdate,
title: Option<String>,
) -> Result<std::result::Result<Node, i64>> {
let current = match Self::get_node_in_tx(tx, id).await? {
Some(n) => n,
None => return Err(anyhow::anyhow!("Task node not found: {}", id)),
};

if current.version != expected_version {
return Ok(Err(current.version));
}

// Merge the update's task-property fields into the node's properties via the
// shared helper (also used by the service layer to compute a title_template
// title against the post-merge node — keeping the two in lockstep). The
// helper is a best-effort no-op on malformed shapes; the persist path keeps
// the stricter contract and rejects them loudly, as before this refactor.
let mut props = current.properties.clone();
let root = props
.as_object()
.ok_or_else(|| anyhow::anyhow!("Invalid properties"))?;
if root.get("task").is_some_and(|task| !task.is_object()) {
return Err(anyhow::anyhow!("Invalid task properties"));
}
update.apply_to_properties(&mut props);

let props_json = serde_json::to_string(&props).context("Failed to serialize properties")?;
let now = Utc::now().to_rfc3339();
let new_version = expected_version + 1;

let mut sql = "UPDATE node SET properties = ?1, version = ?2, modified_at = ?3".to_string();
let mut sql_params: Vec<libsql::Value> = vec![
libsql::Value::Text(props_json),
libsql::Value::Integer(new_version),
libsql::Value::Text(now),
];

if let Some(ref content) = update.content {
sql.push_str(&format!(", content = ?{}", sql_params.len() + 1));
sql_params.push(libsql::Value::Text(content.clone()));
}

// Persist the recomputed title (when the caller determined content changed)
// in the same UPDATE as content, so a task's indexed `title` column never
// goes stale relative to its content.
if let Some(ref new_title) = title {
sql.push_str(&format!(", title = ?{}", sql_params.len() + 1));
sql_params.push(libsql::Value::Text(new_title.clone()));
}

sql.push_str(&format!(
" WHERE id = ?{} AND version = ?{}",
sql_params.len() + 1,
sql_params.len() + 2
));
sql_params.push(libsql::Value::Text(id.to_string()));
sql_params.push(libsql::Value::Integer(expected_version));

let rows_affected = tx
.conn()
.execute(&sql, sql_params)
.await
.context("Failed to update task node")?;

if rows_affected == 0 {
// Lost a race between the read above and this write despite
// running inside the same transaction — vanishingly unlikely
// (nothing else can observe or mutate this row mid-transaction),
// same defensive posture as `update_node_with_version_check_in_tx`'s
// own WHERE clause.
return Ok(Err(current.version));
}

let node = Self::get_node_in_tx(tx, id)
.await?
.ok_or_else(|| anyhow::anyhow!("Task node '{}' not found after update", id))?;

Ok(Ok(node))
}

/// Fetch a schema with `relationships` hydrated from its declaration edges
/// in the `relationship` table. One caller-facing fetch: fields come from
/// the node row, relationships from `get_schema_declarations` — no consumer
Expand Down
2 changes: 1 addition & 1 deletion packages/core/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ pub use db::{
};
pub use models::{
FilterOperator, Node, NodeFilter, NodeQuery, NodeUpdate, OrderBy, PropertyFilter, SchemaNode,
TaskNode, TaskNodeUpdate, TaskStatus, ValidationError,
TaskNodeUpdate, TaskStatus, ValidationError,
};
pub use playbook::PlaybookEngine;
pub use services::{
Expand Down
Loading