v3.30.4: preserve sync data on invalid edits and read failures

Cargo.toml

@@ -11,7 +11,7 @@ members = [
 resolver = "2"
 
 [workspace.package]
-version = "3.30.3"
+version = "3.30.4"
 edition = "2024"
 license-file = "LICENSE"
 authors = ["PureTensor, Inc. <[email protected]>"]
@@ -52,12 +52,12 @@ rmcp = { version = "3.3.0", features = ["server", "macros", "schemars", "transpo
 futures-util = { version = "0.3", default-features = false }
 
 # Internal crates
-ptask-core    = { path = "crates/ptask-core",    version = "3.30.3" }
-ptask-tui     = { path = "crates/ptask-tui",     version = "3.30.3" }
-ptask-server  = { path = "crates/ptask-server",  version = "3.30.3" }
-ptask-bot     = { path = "crates/ptask-bot",     version = "3.30.3" }
-ptask-distill = { path = "crates/ptask-distill", version = "3.30.3" }
-ptask-notify  = { path = "crates/ptask-notify",  version = "3.30.3" }
+ptask-core    = { path = "crates/ptask-core",    version = "3.30.4" }
+ptask-tui     = { path = "crates/ptask-tui",     version = "3.30.4" }
+ptask-server  = { path = "crates/ptask-server",  version = "3.30.4" }
+ptask-bot     = { path = "crates/ptask-bot",     version = "3.30.4" }
+ptask-distill = { path = "crates/ptask-distill", version = "3.30.4" }
+ptask-notify  = { path = "crates/ptask-notify",  version = "3.30.4" }
 
 # HMAC for outbound webhook signing (v0.3.5).
 sha2 = "0.10"

crates/ptask-server/src/routes/sync.rs

@@ -174,10 +174,12 @@ fn read_delta(state: &AppState, full_sync: bool, since: i64) -> DeltaRead {
         event_log::deleted_task_uuids_since(&state.db, since).map_err(|e| ("tombstone read", e))?;
     let mut rows = Vec::new();
     for u in &delta_uuids {
-        // Deleted tasks legitimately fail the row fetch — they're
-        // reported through the tombstone list instead.
-        if let Ok(t) = task_by_uuid(&state.db, u) {
-            rows.push(t);
+        // Missing rows are delivered as tombstones. Any other read failure
+        // must abort the response so its cursor cannot hide an unread update.
+        match task_by_uuid(&state.db, u) {
+            Ok(t) => rows.push(t),
+            Err(ptask_core::Error::Sqlite(rusqlite::Error::QueryReturnedNoRows)) => {}
+            Err(e) => return Err(("delta task read", e)),
         }
     }
     Ok((new_cursor, rows, deleted))
@@ -437,12 +439,15 @@ fn apply_command(
             let task = resolve_task(state, &cmd.args)?;
             // `deadline` present as a string sets it; present as null clears it;
             // absent is an error (this command edits the deadline).
-            if cmd.args.get("deadline").is_none() {
-                return Err(anyhow::anyhow!(
-                    "task_edit: args.deadline required (ISO string to set, null to clear)"
-                ));
-            }
-            let new_deadline = cmd.args.get("deadline").and_then(Value::as_str);
+            let new_deadline = match cmd.args.get("deadline") {
+                Some(Value::String(deadline)) => Some(deadline.as_str()),
+                Some(Value::Null) => None,
+                _ => {
+                    return Err(anyhow::anyhow!(
+                        "task_edit: args.deadline required (ISO string to set, null to clear)"
+                    ));
+                }
+            };
             tasks::update_deadline(
                 &state.db,
                 &task.id,
@@ -607,7 +612,7 @@ fn resolve_query(state: &AppState, query: &str) -> Result<tasks::Task, anyhow::E
 
 fn resolve_task(state: &AppState, args: &Value) -> Result<tasks::Task, anyhow::Error> {
     if let Some(s) = args.get("task_uuid").and_then(Value::as_str) {
-        return task_by_uuid(&state.db, s);
+        return Ok(task_by_uuid(&state.db, s)?);
     }
     if let Some(s) = args.get("pt_id").and_then(Value::as_str) {
         let t = tasks::resolve(&state.db, s)?;
@@ -617,7 +622,7 @@ fn resolve_task(state: &AppState, args: &Value) -> Result<tasks::Task, anyhow::E
 }
 
 /// Direct fetch by UUID (no PT-N indirection).
-fn task_by_uuid(db: &ptask_core::Db, uuid: &str) -> Result<tasks::Task, anyhow::Error> {
+fn task_by_uuid(db: &ptask_core::Db, uuid: &str) -> ptask_core::Result<tasks::Task> {
     let conn = db.get()?;
     let row = conn.query_row(
         "SELECT t.id, t.pt_id, t.title, t.description, t.priority, t.status_v2 AS status,
@@ -653,6 +658,143 @@ mod tests {
     use axum::extract::State;
     use std::time::{Duration, Instant};
 
+    fn test_state() -> (tempfile::TempDir, AppState) {
+        let dir = tempfile::tempdir().unwrap();
+        let db = ptask_core::Db::open(dir.path().join("sync.db")).unwrap();
+        (
+            dir,
+            AppState::new(db, Default::default(), Default::default()),
+        )
+    }
+
+    #[tokio::test]
+    async fn sync_rejects_non_string_deadlines_without_mutation() {
+        let (_dir, state) = test_state();
+        let mut new = tasks::NewTask::minimal("retain my deadline");
+        new.deadline = Some("2099-01-01".into());
+        let task = tasks::create(&state.db, new, &EventCtx::test()).unwrap();
+        let cursor = event_log::current_cursor(&state.db).unwrap();
+        for value in [
+            serde_json::json!(false),
+            serde_json::json!(7),
+            serde_json::json!(1.5),
+            serde_json::json!([]),
+            serde_json::json!({}),
+        ] {
+            let response = sync(
+                State(state.clone()),
+                HeaderMap::new(),
+                Json(SyncReq {
+                    sync_token: Some(cursor.to_string()),
+                    resource_types: vec![],
+                    commands: vec![Command {
+                        kind: "task_edit".into(),
+                        uuid: "invalid-deadline".into(),
+                        temp_id: None,
+                        args: serde_json::json!({"task_uuid": task.id, "deadline": value}),
+                    }],
+                }),
+            )
+            .await
+            .into_response();
+            assert_eq!(response.status(), StatusCode::OK);
+            let bytes = axum::body::to_bytes(response.into_body(), usize::MAX)
+                .await
+                .unwrap();
+            let body: Value = serde_json::from_slice(&bytes).unwrap();
+            assert!(
+                body["sync_status"]["invalid-deadline"]["error"].is_string(),
+                "{body}"
+            );
+            assert_eq!(
+                task_by_uuid(&state.db, &task.id).unwrap().deadline,
+                task.deadline
+            );
+            assert_eq!(event_log::current_cursor(&state.db).unwrap(), cursor);
+        }
+        // Rejected command IDs remain retryable with a valid value.
+        for deadline in [serde_json::json!("2099-02-01"), Value::Null] {
+            let cmd = Command {
+                kind: "task_edit".into(),
+                uuid: if deadline.is_null() {
+                    "clear-deadline"
+                } else {
+                    "invalid-deadline"
+                }
+                .into(),
+                temp_id: None,
+                args: serde_json::json!({"task_uuid": task.id, "deadline": deadline}),
+            };
+            apply_command(&state, &cmd, "test").unwrap();
+            assert_eq!(
+                task_by_uuid(&state.db, &task.id)
+                    .unwrap()
+                    .deadline
+                    .as_deref(),
+                deadline.as_str()
+            );
+        }
+        let cmd = Command {
+            kind: "task_edit".into(),
+            uuid: "missing-deadline".into(),
+            temp_id: None,
+            args: serde_json::json!({"task_uuid": task.id}),
+        };
+        assert!(apply_command(&state, &cmd, "test").is_err());
+    }
+
+    #[tokio::test]
+    async fn sync_delta_read_failure_does_not_advance_client_cursor() {
+        let (_dir, state) = test_state();
+        tasks::create(
+            &state.db,
+            tasks::NewTask::minimal("first"),
+            &EventCtx::test(),
+        )
+        .unwrap();
+        let cursor = event_log::current_cursor(&state.db).unwrap();
+        let second = tasks::create(
+            &state.db,
+            tasks::NewTask::minimal("second"),
+            &EventCtx::test(),
+        )
+        .unwrap();
+        state
+            .db
+            .with_conn(|c| {
+                c.execute_batch("ALTER TABLE tasks RENAME TO tasks_unavailable")?;
+                Ok(())
+            })
+            .unwrap();
+        let response = sync(
+            State(state.clone()),
+            HeaderMap::new(),
+            Json(SyncReq {
+                sync_token: Some(cursor.to_string()),
+                resource_types: vec![],
+                commands: vec![],
+            }),
+        )
+        .await
+        .into_response();
+        assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
+        state
+            .db
+            .with_conn(|c| {
+                c.execute_batch("ALTER TABLE tasks_unavailable RENAME TO tasks")?;
+                Ok(())
+            })
+            .unwrap();
+        let (_, rows, deleted) = read_delta(&state, false, cursor).unwrap();
+        assert_eq!(rows.len(), 1);
+        assert_eq!(rows[0].id, second.id);
+        assert!(deleted.is_empty());
+        tasks::delete_task(&state.db, &second.id, &EventCtx::test()).unwrap();
+        let (_, rows, deleted) = read_delta(&state, false, cursor).unwrap();
+        assert!(rows.is_empty());
+        assert_eq!(deleted, vec![second.id]);
+    }
+
     /// Regression (#39.2): /sync is the heaviest writer on the fleet and ran
     /// every SQLite leg — auth lookup, per-command mutation + rescore, delta
     /// read — inline on an async worker. A caller that had to wait for a

docs/gen/pt.1

@@ -1,6 +1,6 @@
 .ie \n(.g .ds Aq \(aq
 .el .ds Aq '
-.TH pt 1  "pt 3.30.3"
+.TH pt 1  "pt 3.30.4"
 .SH NAME
 pt \- Sovereign task manager for PureTensor
 .SH SYNOPSIS
@@ -166,4 +166,4 @@ Generate shell completions (bash/zsh/fish) to stdout
 pt\-help(1)
 Print this message or the help of the given subcommand(s)
 .SH VERSION
-v3.30.3
+v3.30.4

docs/sync-api.md

@@ -101,7 +101,7 @@ the environment variable is set on the client node.
 | `task_create` | `{ text, source_type? }` | runs quick-add parser, inserts to `tasks` + `pt_extensions`, optional `pt_recurrence`. |
 | `task_done` | `{ task_uuid }` or `{ pt_id }` | flips status to `done` or advances recurrence in-place, logs an `interaction` row. |
 | `task_priority` (v1.8.0) | `{ task_uuid \| pt_id, priority }` | sets priority (1..=5), logs a `priority_change` interaction, rescores. |
-| `task_edit` (v1.8.0) | `{ task_uuid \| pt_id, deadline }` | sets the deadline (ISO string) or clears it (JSON `null`); rescores. |
+| `task_edit` (v1.8.0) | `{ task_uuid \| pt_id, deadline }` | sets the deadline (ISO string) or clears it (JSON `null`); other JSON types or an omitted deadline are rejected without mutation; rescores. |
 | `task_reopen` (v1.8.0) | `{ task_uuid \| pt_id }` | flips a done/dismissed task back to `pending` (logs the neglect-score reopen signal). |
 | `task_retext` (v1.9.0) | `{ task_uuid \| pt_id, title?, description? }` | replaces the title and/or description (at least one required). |
 | `task_dismiss`, `task_start`, `task_snooze` (args.until ISO), `task_depend` (args.on query, args.clear bool), `task_delete` (v1.10.0) | `{ task_uuid \| pt_id }` | soft-closes a task (`status → dismissed`); reversible via `task_reopen`. |