fix(tasks): a create that names a status is stamped like an update to it (#3683)

build_note (create_note and create_records) set the status and nothing it
implies, so create_task(status="in_progress") wrote a started task with no
started_at, and done/cancelled had no completed_at or next recurrence.
The transition is now one function, apply_status_transition, called by
both paths; tests pin the invariant for every status.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/scribe/services/notes.py

@@ -284,7 +284,7 @@ def build_note(
         except ValueError:
             raise ValueError(f"Invalid priority: {priority!r}. Must be one of: {[p.value for p in TaskPriority]}")
 
-    return Note(
+    note = Note(
         user_id=user_id,
         title=title,
         body=body,
@@ -304,6 +304,39 @@ def build_note(
         verify_with=verify_with,
         expires_when=expires_when,
     )
+    # A create that names a status is the same transition an update to it is
+    # (#3683) — otherwise `create_task(status="in_progress")` writes a row no
+    # update could produce: started, with no `started_at`.
+    if status is not None:
+        apply_status_transition(note)
+    return note
+
+
+def apply_status_transition(note: Note) -> None:
+    """Stamp what reaching `note.status` implies — the ONE statement of it.
+
+    Called by the update path whenever `status` is written and by `build_note`
+    whenever a create names one, so a task created at a status is
+    indistinguishable from one that reached it by update. Two copies of "what
+    a status implies" are where the two drift, and #3683 was that drift.
+    """
+    _now = datetime.now(timezone.utc)
+    if note.status == TaskStatus.in_progress.value:
+        if note.started_at is None:
+            note.started_at = _now
+    elif note.status in (TaskStatus.done.value, TaskStatus.cancelled.value):
+        note.completed_at = _now
+        if note.recurrence_rule:
+            from scribe.services.recurrence import calculate_next_due
+            base = note.due_date or _now.date()
+            next_due = calculate_next_due(note.recurrence_rule, base)
+            note.recurrence_next_spawn_at = datetime(
+                next_due.year, next_due.month, next_due.day, tzinfo=timezone.utc
+            )
+    elif note.status == TaskStatus.todo.value:
+        note.started_at = None
+        note.completed_at = None
+        note.recurrence_next_spawn_at = None
 
 
 async def get_note(user_id: int, note_id: int) -> Note | None:
@@ -653,25 +686,8 @@ async def update_note(
             recompose = _mirror_recomposers().get(note.note_type or "")
             if recompose is not None:
                 note.data = recompose(note)
-        # Auto-set lifecycle timestamps on status transitions
         if "status" in fields:
-            _now = datetime.now(timezone.utc)
-            if note.status == TaskStatus.in_progress.value:
-                if note.started_at is None:
-                    note.started_at = _now
-            elif note.status in (TaskStatus.done.value, TaskStatus.cancelled.value):
-                note.completed_at = _now
-                if note.recurrence_rule:
-                    from scribe.services.recurrence import calculate_next_due
-                    base = note.due_date or _now.date()
-                    next_due = calculate_next_due(note.recurrence_rule, base)
-                    note.recurrence_next_spawn_at = datetime(
-                        next_due.year, next_due.month, next_due.day, tzinfo=timezone.utc
-                    )
-            elif note.status == TaskStatus.todo.value:
-                note.started_at = None
-                note.completed_at = None
-                note.recurrence_next_spawn_at = None
+            apply_status_transition(note)
         note.updated_at = datetime.now(timezone.utc)
         await session.commit()
         await session.refresh(note)

tests/test_services_notes.py

@@ -8,6 +8,8 @@ MCP, recurrence, snippets — gets it by construction rather than by remembering
 These test the helper directly. The point of the change is that there is now ONE
 place to test.
 """
+import pytest
+
 from scribe.services import notes as notes_svc
 
 # --- inline embedding (#2056) -----------------------------------------------
@@ -69,3 +71,36 @@ def test_embed_note_swallows_an_indexing_failure():
     note = MagicMock(id=5, user_id=42, title="T", body="B")
     with patch("asyncio.create_task", side_effect=ValueError("model gone")):
         notes_svc.embed_note(note)          # must not raise
+
+
+# --- #3683: a create that names a status is the transition an update is ------
+
+_LIFECYCLE = ("started_at", "completed_at", "recurrence_next_spawn_at")
+
+
+@pytest.mark.parametrize("status", ["todo", "in_progress", "done", "cancelled"])
+def test_a_task_created_at_a_status_matches_one_updated_to_it(status):
+    """The invariant, not the instance that surfaced it: which lifecycle
+    stamps a row carries must not depend on whether it was created at a status
+    or reached it by update. Recurrence is included because done/cancelled
+    schedule the next spawn."""
+    rule = {"type": "interval", "unit": "week", "every": 1}
+    created = notes_svc.build_note(1, title="t", status=status, recurrence_rule=rule)
+
+    updated = notes_svc.build_note(1, title="t", recurrence_rule=rule)
+    updated.status = status
+    notes_svc.apply_status_transition(updated)
+
+    for field in _LIFECYCLE:
+        assert (getattr(created, field) is None) == (getattr(updated, field) is None), (status, field)
+
+
+def test_a_task_created_in_progress_knows_when_it_started():
+    note = notes_svc.build_note(1, title="t", status="in_progress")
+    assert note.started_at is not None
+    assert note.completed_at is None
+
+
+def test_a_note_with_no_status_gets_no_lifecycle_stamps():
+    note = notes_svc.build_note(1, title="t")
+    assert all(getattr(note, f) is None for f in _LIFECYCLE)