From 89bebe561027fd61521ed93f4f5a836a0fcf697e Mon Sep 17 00:00:00 2001 From: 0xrsydn Date: Sun, 20 Sep 2026 21:26:48 +0700 Subject: [PATCH] fix(pi-task): serialize state updates and remind about open tasks --- pi/extensions/pi-task/README.md | 13 +++- pi/extensions/pi-task/index.ts | 109 ++++++++++++++-------------- pi/extensions/pi-task/state.test.ts | 47 ++++++++++++ pi/extensions/pi-task/state.ts | 83 +++++++++++++++++++++ 4 files changed, 195 insertions(+), 57 deletions(-) create mode 100644 pi/extensions/pi-task/state.test.ts create mode 100644 pi/extensions/pi-task/state.ts diff --git a/pi/extensions/pi-task/README.md b/pi/extensions/pi-task/README.md index da3d85f..8167c1f 100644 --- a/pi/extensions/pi-task/README.md +++ b/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. diff --git a/pi/extensions/pi-task/index.ts b/pi/extensions/pi-task/index.ts index af79f34..0e98919 100644 --- a/pi/extensions/pi-task/index.ts +++ b/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"; - -type TaskStatus = "pending" | "in_progress" | "completed"; - -interface Task { - text: string; - status: TaskStatus; -} - -interface TaskDetails { - tasks: Task[]; -} +import { TaskStore, type Task, type TaskDetails } from "./state.js"; 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 = []; - - for (const entry of ctx.sessionManager.getBranch()) { - 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); - } - + store.restore(ctx.sessionManager.getBranch()); + turnsSinceTask = 0; + reminderDue = store.read().some((task) => task.status !== "completed"); 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) { - const wasComplete = tasks.length > 0 && tasks.every((task) => task.status === "completed"); - - if (params.tasks !== undefined) { - 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; - } - + async execute(_toolCallId, params, signal, _onUpdate, ctx) { + const { tasks, wasComplete } = await store.call(params.tasks, signal); + turnsSinceTask = 0; + reminderDue = false; const isComplete = tasks.length > 0 && tasks.every((task) => task.status === "completed"); - refreshTaskUI(ctx); - if (ctx.hasUI && isComplete && !wasComplete) { - ctx.ui.notify(`✓ All ${tasks.length} tasks completed`, "info"); - } + // Rendering must not turn a committed task result into an error result. + try { + 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 diff --git a/pi/extensions/pi-task/state.test.ts b/pi/extensions/pi-task/state.test.ts new file mode 100644 index 0000000..d4e4f1f --- /dev/null +++ b/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); +}); diff --git a/pi/extensions/pi-task/state.ts b/pi/extensions/pi-task/state.ts new file mode 100644 index 0000000..1d5ffbb --- /dev/null +++ b/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 = Promise.resolve(); + + read(): Task[] { + return cloneTasks(this.tasks); + } + + restore(entries: Parameters[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((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(); + } + } +}