fix(pi-task): serialize state updates and remind about open tasks

pi/extensions/pi-task/README.md

@@ -13,7 +13,8 @@ in progress at a time.
 
 State is stored in Pi tool-result details. It survives reloads and compaction,
 follows session branches correctly, and does not create project files. A new Pi
-session starts with an empty list.
+session starts with an empty list. Calls are serialized, and failed or cancelled
+calls do not replace the saved list. Invalid saved snapshots are ignored.
 
 While work is active, the widget uses themed symbols and colors to distinguish
 the current task (`●`) from pending tasks (`○`). The footer shows compact
@@ -21,4 +22,12 @@ overall progress. When every task is complete, Pi displays a success
 notification and removes both the widget and footer indicator to reclaim space.
 The completed list remains available in session history.
 
-Run `/tasks` to refresh the task UI manually.
+The agent must call `task` to update statuses; finishing code does not update
+items automatically. After compaction or two turns without a task call, Pi gives
+the agent a short, temporary reminder of open items. If a run ends with open
+items, Pi shows a warning. Neither reminder nor warning marks work complete.
+
+Run `/tasks` to refresh the task UI manually. Run the state regression tests
+with `node --experimental-strip-types --test pi/extensions/pi-task/state.test.ts`. Pi
+loads the deployed extension from Home Manager, so apply the configuration and
+reload Pi to use repository changes in normal sessions.

pi/extensions/pi-task/index.ts

@@ -6,17 +6,7 @@ import type {
 import { StringEnum } from "@earendil-works/pi-ai";
 import { Text } from "@earendil-works/pi-tui";
 import { Type } from "typebox";
-
+import { TaskStore, type Task, type TaskDetails } from "./state.js";
-type TaskStatus = "pending" | "in_progress" | "completed";
-
-interface Task {
-  text: string;
-  status: TaskStatus;
-}
-
-interface TaskDetails {
-  tasks: Task[];
-}
 
 const TaskParams = Type.Object({
   tasks: Type.Optional(
@@ -33,10 +23,6 @@ const TaskParams = Type.Object({
   ),
 });
 
-function cloneTasks(tasks: Task[]): Task[] {
-  return tasks.map((task) => ({ ...task }));
-}
-
 function formatTask(task: Task): string {
   const marker = {
     pending: "○",
@@ -59,11 +45,14 @@ function formatThemedTask(task: Task, theme: Theme): string {
 }
 
 export default function (pi: ExtensionAPI) {
-  let tasks: Task[] = [];
+  const store = new TaskStore();
+  let turnsSinceTask = 0;
+  let reminderDue = false;
 
   const refreshTaskUI = (ctx: ExtensionContext) => {
-    if (!ctx.hasUI) return;
+    if (!ctx.hasUI || ctx.mode !== "tui") return;
 
+    const tasks = store.read();
     const completed = tasks.filter((task) => task.status === "completed").length;
     const remaining = tasks.filter((task) => task.status !== "completed");
 
@@ -91,23 +80,45 @@ export default function (pi: ExtensionAPI) {
   };
 
   const reconstructState = (ctx: ExtensionContext) => {
-    tasks = [];
+    store.restore(ctx.sessionManager.getBranch());
-
+    turnsSinceTask = 0;
-    for (const entry of ctx.sessionManager.getBranch()) {
+    reminderDue = store.read().some((task) => task.status !== "completed");
-      if (entry.type !== "message") continue;
-      const message = entry.message;
-      if (message.role !== "toolResult" || message.toolName !== "task") continue;
-
-      const saved = message.details as TaskDetails | undefined;
-      if (saved && Array.isArray(saved.tasks)) tasks = cloneTasks(saved.tasks);
-    }
-
     refreshTaskUI(ctx);
   };
 
   pi.on("session_start", async (_event, ctx) => reconstructState(ctx));
   pi.on("session_tree", async (_event, ctx) => reconstructState(ctx));
 
+  pi.on("turn_start", async () => { turnsSinceTask++; });
+  pi.on("session_compact", async () => {
+    if (store.read().some((task) => task.status !== "completed")) reminderDue = true;
+  });
+  pi.on("context", async (event) => {
+    const open = store.read().filter((task) => task.status !== "completed");
+    if (open.length === 0 || (!reminderDue && turnsSinceTask < 2)) return;
+    reminderDue = false;
+    turnsSinceTask = 0;
+    const shown = open.slice(0, 7).map((task) => `${task.status}: ${task.text.replace(/\s+/g, " ").slice(0, 120)}`);
+    const more = open.length > shown.length ? `; ${open.length - shown.length} more open` : "";
+    return {
+      messages: [
+        ...event.messages,
+        {
+          role: "user" as const,
+          content: [{ type: "text" as const, text: `Task status reminder (${open.length} open): ${shown.join("; ")}${more}. Check actual work, then call task to update the complete list before reporting completion. Do not mark unfinished work complete.` }],
+          timestamp: Date.now(),
+        },
+      ],
+    };
+  });
+  pi.on("agent_settled", async (_event, ctx) => {
+    const tasks = store.read();
+    const open = tasks.filter((task) => task.status !== "completed").length;
+    if (open > 0 && ctx.mode === "tui") {
+      ctx.ui.notify(`${open} task(s) still open. If work is complete, update the task list.`, "warning");
+    }
+  });
+
   pi.registerTool({
     name: "task",
     label: "Task",
@@ -117,36 +128,23 @@ export default function (pi: ExtensionAPI) {
     promptGuidelines: [
       "Use task for work with multiple meaningful steps; skip it for simple one-step requests.",
       "Keep task items short and outcome-oriented, and submit the complete ordered list whenever it changes.",
-      "Keep exactly one task in_progress while actively working, and complete all tasks before finishing.",
+      "Keep exactly one task in_progress while actively working. Mark each task completed after verifying it.",
+      "Before reporting work complete, call task with the full updated list; do not leave finished work in_progress.",
     ],
     parameters: TaskParams,
 
-    async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
+    async execute(_toolCallId, params, signal, _onUpdate, ctx) {
-      const wasComplete = tasks.length > 0 && tasks.every((task) => task.status === "completed");
+      const { tasks, wasComplete } = await store.call(params.tasks, signal);
-
+      turnsSinceTask = 0;
-      if (params.tasks !== undefined) {
+      reminderDue = false;
-        const normalized = params.tasks.map((task) => ({
-          text: task.text.trim(),
-          status: task.status,
-        }));
-
-        if (normalized.some((task) => task.text.length === 0)) {
-          throw new Error("Task text cannot be empty");
-        }
-
-        const active = normalized.filter((task) => task.status === "in_progress").length;
-        if (active > 1) {
-          throw new Error("Only one task may be in_progress");
-        }
-
-        tasks = normalized;
-      }
-
       const isComplete = tasks.length > 0 && tasks.every((task) => task.status === "completed");
-      refreshTaskUI(ctx);
+      // Rendering must not turn a committed task result into an error result.
-      if (ctx.hasUI && isComplete && !wasComplete) {
+      try {
-        ctx.ui.notify(`✓ All ${tasks.length} tasks completed`, "info");
+        refreshTaskUI(ctx);
-      }
+        if (ctx.hasUI && isComplete && !wasComplete) {
+          ctx.ui.notify(`✓ All ${tasks.length} tasks completed`, "info");
+        }
+      } catch { /* Keep the saved result even if the UI is unavailable. */ }
 
       return {
         content: [
@@ -155,7 +153,7 @@ export default function (pi: ExtensionAPI) {
             text: tasks.length > 0 ? tasks.map(formatTask).join("\n") : "No tasks",
           },
         ],
-        details: { tasks: cloneTasks(tasks) } satisfies TaskDetails,
+        details: { tasks } satisfies TaskDetails,
       };
     },
   });
@@ -165,6 +163,7 @@ export default function (pi: ExtensionAPI) {
     handler: async (_args, ctx) => {
       refreshTaskUI(ctx);
       if (ctx.hasUI) {
+        const tasks = store.read();
         const remaining = tasks.filter((task) => task.status !== "completed").length;
         const message =
           tasks.length === 0

pi/extensions/pi-task/state.test.ts

@@ -0,0 +1,47 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import { TaskStore, restoreTasks, type Task } from "./state.ts";
+
+const active: Task[] = [{ text: "Build feature", status: "in_progress" }];
+const done: Task[] = [{ text: "Build feature", status: "completed" }];
+const result = (tasks: Task[], isError = false) => ({
+  type: "message", message: { role: "toolResult", toolName: "task", isError, details: { tasks } },
+});
+
+test("restores the last valid snapshot on the active branch, including a clear", () => {
+  assert.deepEqual(restoreTasks([result(active), result(done, true), result(done)]), done);
+  assert.deepEqual(restoreTasks([result(active), result([])]), []);
+  assert.deepEqual(restoreTasks([result(active), {
+    type: "message", message: { role: "toolResult", toolName: "task", details: { tasks: [{ text: "bad", status: "invalid" }] } },
+  }]), active);
+});
+
+test("serializes concurrent updates and returns independent snapshots", async () => {
+  const store = new TaskStore();
+  const first = store.call(active);
+  const second = store.call(done);
+  assert.deepEqual(await first, { tasks: active, wasComplete: false });
+  assert.deepEqual(await second, { tasks: done, wasComplete: false });
+  assert.deepEqual(store.read(), done);
+  const snapshot = store.read();
+  snapshot[0].text = "changed";
+  assert.deepEqual(store.read(), done);
+});
+
+test("rejects invalid and cancelled calls without changing state", async () => {
+  const store = new TaskStore();
+  await store.call(active);
+  await assert.rejects(store.call([{ text: "  ", status: "pending" }]), /empty/);
+  const controller = new AbortController();
+  controller.abort();
+  await assert.rejects(store.call(done, controller.signal), /cancelled/);
+  assert.deepEqual(store.read(), active);
+});
+
+test("rejects queued calls after a branch change", async () => {
+  const store = new TaskStore();
+  const queued = store.call(done);
+  store.restore([result(active)]);
+  await assert.rejects(queued, /session changed/);
+  assert.deepEqual(store.read(), active);
+});

pi/extensions/pi-task/state.ts

@@ -0,0 +1,83 @@
+export type TaskStatus = "pending" | "in_progress" | "completed";
+
+export interface Task {
+  text: string;
+  status: TaskStatus;
+}
+
+export interface TaskDetails {
+  tasks: Task[];
+}
+
+export function cloneTasks(tasks: Task[]): Task[] {
+  return tasks.map((task) => ({ ...task }));
+}
+
+export function normalizeTasks(tasks: Task[]): Task[] {
+  const normalized = tasks.map((task) => ({ text: task.text.trim(), status: task.status }));
+  if (normalized.some((task) => task.text.length === 0)) {
+    throw new Error("Task text cannot be empty");
+  }
+  if (normalized.filter((task) => task.status === "in_progress").length > 1) {
+    throw new Error("Only one task may be in_progress");
+  }
+  return normalized;
+}
+
+export function restoreTasks(
+  entries: Iterable<{ type: string; message?: { role: string; toolName?: string; isError?: boolean; details?: unknown } }>,
+): Task[] {
+  let tasks: Task[] = [];
+  for (const entry of entries) {
+    if (entry.type !== "message") continue;
+    const message = entry.message;
+    if (message?.role !== "toolResult" || message.toolName !== "task" || message.isError) continue;
+    const saved = message.details as TaskDetails | undefined;
+    if (!saved || !Array.isArray(saved.tasks)) continue;
+    try {
+      if (saved.tasks.some((task) =>
+        !task || typeof task.text !== "string" ||
+        !["pending", "in_progress", "completed"].includes(task.status)
+      )) continue;
+      tasks = normalizeTasks(saved.tasks);
+    } catch {
+      // Ignore malformed snapshots; preserve the last valid state on this branch.
+    }
+  }
+  return tasks;
+}
+
+/** Serializes calls and rejects calls queued before a branch or session change. */
+export class TaskStore {
+  private tasks: Task[] = [];
+  private generation = 0;
+  private tail: Promise<void> = Promise.resolve();
+
+  read(): Task[] {
+    return cloneTasks(this.tasks);
+  }
+
+  restore(entries: Parameters<typeof restoreTasks>[0]): void {
+    this.generation++;
+    this.tasks = restoreTasks(entries);
+  }
+
+  async call(next: Task[] | undefined, signal?: AbortSignal): Promise<{ tasks: Task[]; wasComplete: boolean }> {
+    const generation = this.generation;
+    const previous = this.tail;
+    let release!: () => void;
+    this.tail = new Promise<void>((resolve) => { release = resolve; });
+    await previous;
+    try {
+      if (generation !== this.generation || signal?.aborted) {
+        throw new Error("Task call cancelled or session changed");
+      }
+      const normalized = next === undefined ? undefined : normalizeTasks(next);
+      const wasComplete = this.tasks.length > 0 && this.tasks.every((task) => task.status === "completed");
+      if (normalized) this.tasks = normalized;
+      return { tasks: this.read(), wasComplete };
+    } finally {
+      release();
+    }
+  }
+}