[6/7] Record task status transitions

docs/api.md

@@ -177,7 +177,8 @@ than the rest since it grows without bound.
 ```json
 Response: {
   "statuses": [{ "name": "pending", "label": "Pending", "color": "#...", ... }],
-  "lanes": [{ "status": "pending", "total": 12, "tasks": [ ... ] }]
+  "lanes": [{ "status": "pending", "total": 12, "tasks": [ ... ] }],
+  "status_since": { "2026-03-01-fix-bug": "2026-03-02T10:00:00Z" }
 }
 ```
 
@@ -223,6 +224,16 @@ Request:  { "deadline": "" }
 Response: { "task": { ... }, "task_id": "2026-03-01-fix-bug", "updated": true }
 ```
 
+#### `GET /api/tasks/{id}/events`
+A task's status history, oldest first.
+
+```json
+Response: { "events": [
+  { "id": 1, "task_id": "...", "from_status": null, "to_status": "pending", "actor": "system", "created_at": "..." },
+  { "id": 2, "task_id": "...", "from_status": "pending", "to_status": "in_progress", "actor": "impl-abc123", "created_at": "..." }
+] }
+```
+
 #### `POST /api/tasks/{id}/move`
 Place a task in a lane — the board's drag-and-drop write. Send *intent*
 ("between these two cards"), not a computed rank: the server resolves the

docs/tasks.md

@@ -114,6 +114,34 @@ Two rules keep ranks meaningful:
   lane, so a status change places the card at the top of its destination
   instead of carrying over a number it was never ordered against.
 
+### Status History
+
+Every status transition is appended to `task_events` (`task_id`,
+`from_status`, `to_status`, `actor`, `created_at`), added in migration v044
+and seeded with one origin row per existing task. `from_status` is NULL on
+the row recording a task's creation, so an aging calculation can tell
+"created here" from "moved here".
+
+Recording happens inside the same transaction as the status write, from all
+three paths that can change one: `update_task_status`, `move_task`, and the
+full-row `upsert_task` (which is how `task_update` flips status when it also
+writes a note). A no-op transition records nothing — that single rule is
+what keeps `reindex()` from doubling the table on every run, since it
+rewrites every row with the status it already had. The one case where
+reindex *does* change a status, resetting an orphaned row to match its
+directory, is a real correction and gets a row.
+
+`actor` is the session id for agent-driven changes, `web` for the HTTP API,
+`backfill` on the origin rows v044 seeded for tasks that predate the table,
+and `system` otherwise. `backfill` is load-bearing rather than cosmetic: a
+real creation also records a NULL `from_status`, so the actor is the only
+thing distinguishing a synthesized origin from one that actually happened.
+
+This powers the aging indicator on board cards and the timeline in the task
+detail view. Tasks whose last transition predates v044 report no entry time
+rather than a guessed one, so the UI stays silent instead of showing an age
+that isn't true.
+
 ### Live Updates
 
 Every task mutation broadcasts a `task_updated` WebSocket event on the

nerve/agent/tools/handlers/tasks.py

@@ -256,6 +256,7 @@ async def task_create_handler(ctx: ToolContext, args: dict) -> ToolResult:
             deadline=deadline or None,
             tags=tags_to_string(tags),
             content=content,
+            actor=ctx.session_id,
         )
 
     _tasks_read.add(task_id)
@@ -427,14 +428,15 @@ async def task_update_handler(ctx: ToolContext, args: dict) -> ToolResult:
                     title=final_title,
                     status=final_status,
                     content=content,
+                    actor=ctx.session_id,
                     **edits,
                 )
                 await _emit_task_event(ctx, task_id, "updated")
                 return ToolResult.text(f"Task {task_id} updated.")
 
         # Fall back to metadata updates when no task file was changed.
         if status:
-            await ctx.db.update_task_status(task_id, status)
+            await ctx.db.update_task_status(task_id, status, actor=ctx.session_id)
         if raw_tags is not None:
             await ctx.db.update_task_tags(task_id, new_tags_str)
         await _emit_task_event(ctx, task_id, "updated")
@@ -540,7 +542,7 @@ async def task_done_handler(ctx: ToolContext, args: dict) -> ToolResult:
         if ctx.workspace:
             ensure_path_not_tracked_config(ctx.workspace / task["file_path"], "move")
 
-        await ctx.db.update_task_status(task_id, "done")
+        await ctx.db.update_task_status(task_id, "done", actor=ctx.session_id)
 
         # Mark any implementing plans for this task as done
         implementing_plans = await ctx.db.get_plans_for_task(task_id)
@@ -635,7 +637,7 @@ async def task_reopen_handler(ctx: ToolContext, args: dict) -> ToolResult:
                 is_error=True,
             )
 
-    await ctx.db.update_task_status(task_id, status)
+    await ctx.db.update_task_status(task_id, status, actor=ctx.session_id)
 
     if src is not None:
         # Re-checked because the guard above is not atomic with the move.

nerve/db/migrations/v044_task_events.py

@@ -0,0 +1,72 @@
+"""V44: status-transition history for tasks.
+
+``tasks`` stores only the current status, so every transition overwrote
+the last one and nothing recorded that it happened. The markdown file's
+``## Updates`` section was the closest thing to a history, and it only
+gains a line when a note is passed or on done/reopen — a plain status
+flip left no trace at all, at day granularity and with no actor.
+
+That was tolerable while status changes were rare and deliberate. The
+task board makes them a drag, so they're frequent, and it raises the
+questions this table answers: how long has this been in progress, which
+cards are aging, how often does work bounce back out of review.
+
+Rows are append-only. ``from_status`` is NULL for the row recording a
+task's creation, so a task's first event is its birth rather than an
+implicit gap before the first transition.
+
+Timestamps are TEXT written Python-side as UTC ISO strings, matching every
+other table here. They are not fixed width — ``isoformat()`` drops the
+fractional part on an exact second — but lexicographic order is still
+chronological, because the only characters that can follow the seconds
+field are ``+`` (0x2B) and ``.`` (0x2E), and both sort below every digit.
+"""
+
+from __future__ import annotations
+
+import logging
+
+import aiosqlite
+
+logger = logging.getLogger(__name__)
+
+SQL = """
+CREATE TABLE IF NOT EXISTS task_events (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    task_id TEXT NOT NULL,
+    from_status TEXT,
+    to_status TEXT NOT NULL,
+    actor TEXT NOT NULL DEFAULT 'system',
+    created_at TEXT NOT NULL
+);
+
+-- The only read pattern: one task's history, oldest first.
+CREATE INDEX IF NOT EXISTS idx_task_events_task
+    ON task_events(task_id, created_at);
+
+-- Aging queries sweep by time across all tasks.
+CREATE INDEX IF NOT EXISTS idx_task_events_created
+    ON task_events(created_at);
+"""
+
+
+async def up(db: aiosqlite.Connection) -> None:
+    await db.executescript(SQL)
+
+    # Seed one event per existing task so history has a floor rather than
+    # starting mid-air. created_at (not updated_at) is the honest stamp:
+    # we know when the task appeared, not when it reached its current
+    # status. The `backfill` actor is what marks these as synthesized — not
+    # the NULL from_status, which a genuine creation records too.
+    await db.execute(
+        """
+        INSERT INTO task_events (task_id, from_status, to_status, actor, created_at)
+        SELECT id, NULL, status, 'backfill', COALESCE(created_at, updated_at)
+        FROM tasks
+        WHERE COALESCE(created_at, updated_at) IS NOT NULL
+        """
+    )
+
+    async with db.execute("SELECT COUNT(*) FROM task_events") as cursor:
+        seeded = (await cursor.fetchone())[0]
+    logger.info("v044: task_events created, seeded %d origin row(s)", seeded)

nerve/db/tasks.py

@@ -60,6 +60,7 @@ async def upsert_task(
         tags: str | _Keep = KEEP,
         content: str = "",
         position: float | None = None,
+        actor: str = "system",
     ) -> None:
         """Insert or update a task row and its FTS entry.
 
@@ -84,14 +85,18 @@ async def upsert_task(
         """
         now = datetime.now(timezone.utc).isoformat()
         async with self._atomic():
-            # One read resolves every preserve-on-omit column. The statement
-            # below stays a plain full replace, and the FTS text sees the
-            # tags that actually land in the row.
+            # One read serves every column this method resolves before the
+            # write. The statement below stays a plain full replace, and the
+            # FTS text sees the tags that actually land in the row.
             stored = await self._stored_task_columns(task_id)
             source = _resolve_kept(source, stored, "source", None)
             source_url = _resolve_kept(source_url, stored, "source_url", None)
             deadline = _resolve_kept(deadline, stored, "deadline", None)
             tags = _resolve_kept(tags, stored, "tags", "")
+            # A full-row upsert is one of the paths that can change status
+            # (task_update with a note, the PATCH route saving content), so
+            # it has to record transitions like the dedicated methods do.
+            previous_status = stored["status"] if stored else None
             if position is None:
                 position = await self._resolve_upsert_position(stored, status)
             await self.db.execute(
@@ -109,6 +114,8 @@ async def upsert_task(
             # The FTS5 tokenizer already splits the hyphenated slug into words
             # ("2026-03-10-distribution" → 2026, 03, 10, distribution), so slug
             # search works without rewriting the key — and the key stays joinable.
+            await self._record_status_event(task_id, previous_status, status, actor)
+
             fts_content = f"{content} {tags.replace(',', ' ')}" if tags else content
             await self.db.execute("DELETE FROM tasks_fts WHERE task_id = ?", (task_id,))
             await self.db.execute(
@@ -121,6 +128,79 @@ async def get_task(self, task_id: str) -> dict | None:
             row = await cursor.fetchone()
             return dict(row) if row else None
 
+    # ── Status history (task_events) ─────────────────────────────────────
+
+    async def _record_status_event(
+        self,
+        task_id: str,
+        from_status: str | None,
+        to_status: str,
+        actor: str,
+    ) -> None:
+        """Append a transition to ``task_events``, if it is one.
+
+        Called from inside ``_atomic()`` by every path that can change a
+        task's status, so the event and the status commit together — there
+        is no window in which a task has moved but its history hasn't.
+
+        A no-op transition writes nothing, and that one rule is what keeps
+        ``TaskManager.reindex()`` from flooding the table: reindex rewrites
+        every row with the status it already had, so from == to and nothing
+        is recorded. The single case where reindex *does* change a status —
+        resetting an orphaned row to match its directory — is a real
+        correction and worth a row.
+        """
+        if from_status == to_status:
+            return
+        await self.db.execute(
+            "INSERT INTO task_events (task_id, from_status, to_status, actor, created_at) "
+            "VALUES (?, ?, ?, ?, ?)",
+            (
+                task_id, from_status, to_status, actor or "system",
+                datetime.now(timezone.utc).isoformat(),
+            ),
+        )
+
+    async def list_task_events(self, task_id: str, limit: int = 200) -> list[dict]:
+        """A task's transitions, oldest first — the order a timeline reads."""
+        async with self.db.execute(
+            "SELECT * FROM task_events WHERE task_id = ? "
+            "ORDER BY created_at ASC, id ASC LIMIT ?",
+            (task_id, limit),
+        ) as cursor:
+            return [dict(row) async for row in cursor]
+
+    async def get_status_entry_times(self, task_ids: list[str]) -> dict[str, str]:
+        """When each task last *entered* its current status.
+
+        The basis for card aging. Takes the newest event per task and keeps
+        it only if it still matches the live status: a row whose status was
+        last changed by a path predating this table, or by a direct DB edit,
+        has no truthful entry time. Omitting it leaves the indicator silent,
+        which beats inventing an age that reads as fact.
+        """
+        if not task_ids:
+            return {}
+        placeholders = ",".join("?" * len(task_ids))
+        async with self.db.execute(
+            f"SELECT task_id, to_status, MAX(created_at) FROM task_events "
+            f"WHERE task_id IN ({placeholders}) GROUP BY task_id",
+            tuple(task_ids),
+        ) as cursor:
+            newest = {row[0]: (row[1], row[2]) async for row in cursor}
+
+        async with self.db.execute(
+            f"SELECT id, status FROM tasks WHERE id IN ({placeholders})",
+            tuple(task_ids),
+        ) as cursor:
+            live = {row[0]: row[1] async for row in cursor}
+
+        return {
+            task_id: entered_at
+            for task_id, (to_status, entered_at) in newest.items()
+            if live.get(task_id) == to_status
+        }
+
     # ── Board ordering (tasks.position) ──────────────────────────────────
     #
     # Lanes sort by ``position ASC``, so a *lower* rank is *higher* in the
@@ -239,6 +319,7 @@ async def move_task(
         status: str | None = None,
         before_id: str | None = None,
         after_id: str | None = None,
+        actor: str = "system",
     ) -> dict | None:
         """Place a task at an explicit slot in a lane; return the new row.
 
@@ -276,6 +357,10 @@ async def move_task(
                 "UPDATE tasks SET status = ?, position = ?, updated_at = ? WHERE id = ?",
                 (lane, position, now, task_id),
             )
+            # A pure reorder leaves status alone, and _record_status_event
+            # drops the no-op — so dragging within a lane doesn't pollute
+            # the history with rows that say nothing changed.
+            await self._record_status_event(task_id, row[0], lane, actor)
             async with self.db.execute(
                 "SELECT * FROM tasks WHERE id = ?", (task_id,),
             ) as cursor:
@@ -368,7 +453,9 @@ async def count_tasks(
             row = await cursor.fetchone()
             return row[0] if row else 0
 
-    async def update_task_status(self, task_id: str, status: str) -> None:
+    async def update_task_status(
+        self, task_id: str, status: str, actor: str = "system",
+    ) -> None:
         """Move a task to another status, re-ranking it into the new lane.
 
         A rank only means anything relative to its own lane, so carrying
@@ -396,6 +483,7 @@ async def update_task_status(self, task_id: str, status: str) -> None:
                 "UPDATE tasks SET status = ?, position = ?, updated_at = ? WHERE id = ?",
                 (status, position, now, task_id),
             )
+            await self._record_status_event(task_id, row[0], status, actor)
 
     async def update_task_tags(self, task_id: str, tags: str) -> None:
         now = datetime.now(timezone.utc).isoformat()

nerve/gateway/routes/tasks.py

@@ -186,7 +186,14 @@ async def task_board(
             )
         lanes.append({"status": name, "total": total, "tasks": tasks})
 
-    return {"statuses": statuses, "lanes": lanes}
+    # When each visible card last entered its current status — one query
+    # for the whole board rather than one per card. Tasks whose last
+    # transition predates task_events are simply absent, so the UI shows
+    # no age rather than a wrong one.
+    visible = [t["id"] for lane in lanes for t in lane["tasks"]]
+    status_since = await deps.db.get_status_entry_times(visible)
+
+    return {"statuses": statuses, "lanes": lanes, "status_since": status_since}
 
 
 @router.get("/api/tasks/tags")
@@ -266,14 +273,18 @@ async def move_task(
     if target != task["status"]:
         result = await get_tool_registry().invoke(
             "task_update",
-            build_route_tool_context(),
+            # "web" rather than the default "system" sentinel: this string
+            # lands in task_events.actor, and the point of that column is
+            # telling a person dragging a card from the agent moving it.
+            build_route_tool_context("web"),
             {"task_id": task_id, "status": target},
         )
         if result.is_error:
             raise HTTPException(status_code=409, detail=result.text_content)
 
     moved = await deps.db.move_task(
         task_id, status=target, before_id=req.before_id, after_id=req.after_id,
+        actor="web",
     )
     if not moved:
         raise HTTPException(status_code=404, detail="Task not found")
@@ -282,6 +293,16 @@ async def move_task(
     return {"task": moved}
 
 
+@router.get("/api/tasks/{task_id}/events")
+async def list_task_events(
+    task_id: str, limit: int = 200, user: dict = Depends(require_auth),
+):
+    """A task's status history, oldest first."""
+    deps = get_deps()
+    limit = max(1, min(limit, 500))
+    return {"events": await deps.db.list_task_events(task_id, limit=limit)}
+
+
 @router.get("/api/tasks/{task_id}")
 async def get_task(task_id: str, user: dict = Depends(require_auth)):
     deps = get_deps()
@@ -365,7 +386,7 @@ async def update_task(task_id: str, req: TaskUpdateRequest, user: dict = Depends
 
     if len(payload) > 1:
         result = await get_tool_registry().invoke(
-            "task_update", build_route_tool_context(), payload,
+            "task_update", build_route_tool_context("web"), payload,
         )
         # Previously swallowed: an unknown status returned {"updated": true}
         # while nothing had changed. Surface the handler's own message.