diff --git a/CLAUDE.md b/CLAUDE.md index d156d65..8f5e0d3 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -81,7 +81,7 @@ Module-level `@tool` handlers. Call `set_workspace_root()` before use. - Destructive confirms: `classify_danger` + `ToolRegistry(on_confirm=…)`; CLI default deny; config `confirm_destructive` / `path_denylist`. Session YOLO: `/yolo on|off|once` and `--yes` (skip soft confirms; hard denylists unchanged; `once` expires after the next user turn). - **`vcs`**: read-only VCS tools (`vcs_kind` / `vcs_status` / `vcs_diff` / `vcs_log` / `vcs_branch`) via `VcsBackend` protocol; **git** implemented; detectors are pluggable for other systems. - **`chat`**: human prompts as tools — `ask_user_line` / `ask_user_choice` / `ask_user_form` (shared `prompting` core). -- **`todo`**: nested session sub-task stack (frames) — `todo_list` / `todo_push` (multi-title = new frame) / `todo_pop` (leave frame) / `todo_update` / `todo_clear`; pattern push[T1,T2]→push[T1.1…]→pop→push[T2.1…]; stored on session row; agent injects a developer review message if open todos and no todo tool use in the turn. +- **`todo`**: LIFO session sub-task **stack** (not a queue) — `todo_list` / `todo_push` (first title becomes TOP) / `todo_pop` (TOP only) / `todo_update` / `todo_clear`; DFS breakdown push[T1,T2]→push[T1.1…]→pop tops→push[T2.1…]; stored on session row; agent injects a developer review message if open todos and no todo tool use in the turn. - **`DEFAULT_TOOLS`**: file + process + vcs + chat + todo tool list for a `ToolRegistry`. ### Prompting (`prompting.py`) diff --git a/src/plyngent/agent/todo_stack.py b/src/plyngent/agent/todo_stack.py index 5aadf46..84051fc 100644 --- a/src/plyngent/agent/todo_stack.py +++ b/src/plyngent/agent/todo_stack.py @@ -11,7 +11,7 @@ _OPEN: frozenset[str] = frozenset({"pending", "in_progress"}) class TodoItem(Struct, omit_defaults=True): - """One sub-task within a stack frame.""" + """One entry on the todo stack (top of stack = next to work).""" id: str title: str @@ -19,29 +19,28 @@ class TodoItem(Struct, omit_defaults=True): notes: str = "" -class TodoFrame(Struct, omit_defaults=True): - """One nesting level: sibling tasks pushed together.""" +class TodoStackData(Struct, omit_defaults=True): + """Serializable stack: ``items[0]`` is bottom, ``items[-1]`` is **top** (LIFO).""" items: list[TodoItem] = field(default_factory=list) - - -class TodoStackData(Struct, omit_defaults=True): - """Serializable stack body (session storage). - - ``frames`` is bottom→top: root plan first, deepest breakdown last. - Pattern: push [T1,T2] → push [T1.1,T1.2] → pop → push [T2.1] → … - """ - - frames: list[TodoFrame] = field(default_factory=list) next_id: int = 1 class TodoStack: - """Nested todo stack for hierarchical sub-task breakdown. + """True LIFO stack of sub-tasks (not a queue, not a flat checklist). - Each ``push`` of one or more titles creates a **new frame** (deeper level). - ``pop`` removes the **top frame** (leave a breakdown level). - Model tools maintain the stack; humans use ``/todos``. + - **Top** = last element = next work item. + - ``push`` / multi-push: new items go on the **top**. When pushing several + titles at once, the **first** title becomes the top (worked first); the + rest sit under it in order (DFS: children first, then later siblings). + - ``pop``: remove **only the top** item. + + Breakdown pattern:: + + push T1, T2 # top=T1, under=T2 + push T1.1, T1.2 # top=T1.1, then T1.2, then T1, then T2 + pop / done … # clear T1.1, T1.2, then T1 + push T2.1 # top=T2.1 over T2 """ def __init__(self, data: TodoStackData | None = None) -> None: @@ -49,35 +48,33 @@ class TodoStack: self._touched_this_turn: bool = False @property - def frames(self) -> list[TodoFrame]: - return self._data.frames + def items(self) -> list[TodoItem]: + """Bottom → top order (last is top).""" + return self._data.items @property - def depth(self) -> int: - return len(self._data.frames) + def top(self) -> TodoItem | None: + if not self._data.items: + return None + return self._data.items[-1] @property def touched_this_turn(self) -> bool: return self._touched_this_turn def begin_turn(self) -> None: - """Reset the per-turn touch flag (call at start of each user turn).""" self._touched_this_turn = False def mark_touched(self) -> None: self._touched_this_turn = True - def all_items(self) -> list[TodoItem]: - return [item for frame in self._data.frames for item in frame.items] - def open_items(self) -> list[TodoItem]: - return [item for item in self.all_items() if item.status in _OPEN] + return [item for item in self._data.items if item.status in _OPEN] def is_empty(self) -> bool: - return not self._data.frames + return not self._data.items def needs_review(self) -> bool: - """True when open work exists but no todo tool ran this turn.""" return bool(self.open_items()) and not self._touched_this_turn def to_data(self) -> TodoStackData: @@ -87,20 +84,22 @@ class TodoStack: def from_raw(cls, raw: object | None) -> TodoStack: if raw is None: return cls() - # Legacy flat shape: {items: [...], next_id: N} → single root frame. - if isinstance(raw, dict) and "frames" not in raw and "items" in raw: + # Nested-frame shape (brief intermediate design) → flatten bottom→top. + if isinstance(raw, dict) and "frames" in raw: try: - flat = cast("dict[str, object]", raw) - items = msgspec.convert(flat.get("items"), type=list[TodoItem]) - next_raw = flat.get("next_id", 1) + frames_raw = cast("dict[str, object]", raw).get("frames") + next_raw = cast("dict[str, object]", raw).get("next_id", 1) next_id = max(1, int(next_raw) if isinstance(next_raw, int | str) else 1) + items: list[TodoItem] = [] + if isinstance(frames_raw, list): + for frame in cast("list[object]", frames_raw): + if not isinstance(frame, dict): + continue + frame_items = msgspec.convert(frame.get("items"), type=list[TodoItem]) + items.extend(frame_items) + return cls(TodoStackData(items=items, next_id=next_id)) except (msgspec.ValidationError, TypeError, ValueError): return cls() - data = TodoStackData( - frames=[TodoFrame(items=items)] if items else [], - next_id=next_id, - ) - return cls(data) try: data = msgspec.convert(raw, type=TodoStackData) except (msgspec.ValidationError, TypeError, ValueError): @@ -112,42 +111,45 @@ class TodoStack: def to_raw(self) -> dict[str, object]: out: object = msgspec.to_builtins(self._data) if not isinstance(out, dict): - return {"frames": [], "next_id": 1} + return {"items": [], "next_id": 1} raw = cast("dict[object, object]", out) return {str(key): value for key, value in raw.items()} def render(self) -> str: - if not self._data.frames: - return "(todo stack empty)" - lines: list[str] = [f"(depth={self.depth})"] - for depth, frame in enumerate(self._data.frames): - indent = " " * depth - lines.append(f"{indent}frame {depth}:") - if not frame.items: - lines.append(f"{indent} (empty frame)") - continue - for item in frame.items: - mark = { - "pending": "[ ]", - "in_progress": "[~]", - "done": "[x]", - "cancelled": "[-]", - }.get(item.status, "[?]") - note = f" — {item.notes}" if item.notes else "" - lines.append(f"{indent} {mark} {item.id}: {item.title}{note}") + if not self._data.items: + return "(todo stack empty — top is empty)" + lines: list[str] = ["(LIFO stack: TOP = next to work)"] + # Show top first so the model sees work order. + for offset, item in enumerate(reversed(self._data.items)): + mark = { + "pending": "[ ]", + "in_progress": "[~]", + "done": "[x]", + "cancelled": "[-]", + }.get(item.status, "[?]") + note = f" — {item.notes}" if item.notes else "" + tag = " TOP" if offset == 0 else "" + lines.append(f"{mark}{tag} {item.id}: {item.title}{note}") return "\n".join(lines) def review_prompt(self) -> str: - """Control-message body (developer role) when the model finishes without todo ops.""" return ( "Todo stack review (internal control — not a human message).\n" - "Open sub-tasks remain and you did not call any todo_* tools this turn.\n" - "Use nested push/pop for breakdown: push [T1,T2], push [T1.1,T1.2], " - "finish children, pop frame, then push children of T2, etc. " - "Mark done/cancelled, push new frames, or pop a finished level.\n\n" + "This is a LIFO stack (not a queue): TOP is the only pop target and " + "the next sub-task to execute. Open work remains and you did not call " + "any todo_* tools this turn.\n" + "Breakdown: push children onto the top, finish/pop them, then the " + "parent (or next sibling under it) becomes top again. " + "push [T1,T2] then push [T1.1,T1.2] then pop finished tops, then " + "push [T2.1]…\n\n" f"Current stack:\n{self.render()}" ) + def _alloc(self, title: str, *, notes: str, status: TodoStatus) -> TodoItem: + item_id = f"t{self._data.next_id}" + self._data.next_id += 1 + return TodoItem(id=item_id, title=title, status=status, notes=notes) + def push_titles( self, titles: list[str], @@ -155,43 +157,46 @@ class TodoStack: notes: str = "", status: TodoStatus = "pending", ) -> list[TodoItem]: - """Push a new frame containing *titles* (one or more siblings at this depth).""" + """Push one or more tasks onto the top (LIFO). + + Titles are pushed so the **first** listed title becomes the new **top** + (worked first). Example: ``push_titles(["T1","T2"])`` → top=T1, under=T2. + """ cleaned = [t.strip() for t in titles if t and t.strip()] if not cleaned: msg = "at least one non-empty title is required" raise ValueError(msg) note = notes.strip() - created: list[TodoItem] = [] - for title in cleaned: - item_id = f"t{self._data.next_id}" - self._data.next_id += 1 - created.append(TodoItem(id=item_id, title=title, status=status, notes=note)) - self._data.frames.append(TodoFrame(items=created)) + # Push last→first so first title ends on top. + created_rev: list[TodoItem] = [] + for title in reversed(cleaned): + item = self._alloc(title, notes=note, status=status) + self._data.items.append(item) + created_rev.append(item) self.mark_touched() - return created + # Return in the same order as *titles* (first = top). + return list(reversed(created_rev)) def push(self, title: str, *, notes: str = "", status: TodoStatus = "pending") -> TodoItem: - """Push a new frame with a single task (convenience for one title).""" - items = self.push_titles([title], notes=notes, status=status) - return items[0] + return self.push_titles([title], notes=notes, status=status)[0] - def pop(self) -> TodoFrame | None: - """Pop and return the top frame (leave the current breakdown level).""" - if not self._data.frames: + def pop(self) -> TodoItem | None: + """Remove and return the **top** item only (classic stack pop).""" + if not self._data.items: return None - frame = self._data.frames.pop() + item = self._data.items.pop() self.mark_touched() - return frame + return item def clear(self) -> int: - n = sum(len(frame.items) for frame in self._data.frames) - self._data.frames.clear() + n = len(self._data.items) + self._data.items.clear() if n: self.mark_touched() return n def get(self, item_id: str) -> TodoItem | None: - for item in self.all_items(): + for item in self._data.items: if item.id == item_id: return item return None @@ -204,26 +209,25 @@ class TodoStack: status: TodoStatus | None = None, notes: str | None = None, ) -> TodoItem: - for frame in self._data.frames: - for index, item in enumerate(frame.items): - if item.id != item_id: - continue - new_title = title.strip() if title is not None else item.title - if not new_title: - msg = "title must not be empty" - raise ValueError(msg) - new_status = status if status is not None else item.status - new_notes = notes if notes is not None else item.notes - updated = TodoItem(id=item.id, title=new_title, status=new_status, notes=new_notes) - frame.items[index] = updated - self.mark_touched() - return updated + for index, item in enumerate(self._data.items): + if item.id != item_id: + continue + new_title = title.strip() if title is not None else item.title + if not new_title: + msg = "title must not be empty" + raise ValueError(msg) + new_status = status if status is not None else item.status + new_notes = notes if notes is not None else item.notes + updated = TodoItem(id=item.id, title=new_title, status=new_status, notes=new_notes) + self._data.items[index] = updated + self.mark_touched() + return updated msg = f"unknown todo id {item_id!r}" raise KeyError(msg) def parse_push_titles(raw: str) -> list[str]: - """Parse multi-title push input: JSON array, or newline/semicolon-separated.""" + """Parse multi-title push: JSON array, newlines, or ``;``-separated.""" text = raw.strip() if not text: return [] @@ -233,10 +237,13 @@ def parse_push_titles(raw: str) -> list[str]: except (msgspec.DecodeError, UnicodeEncodeError): data = None if isinstance(data, list): - out = [item.strip() for item in cast("list[object]", data) if isinstance(item, str) and item.strip()] + out = [ + item.strip() + for item in cast("list[object]", data) + if isinstance(item, str) and item.strip() + ] if out: return out - # Newlines, then ``;`` as separators. parts: list[str] = [] for line in text.replace(";", "\n").splitlines(): token = line.strip() diff --git a/src/plyngent/cli/slash.py b/src/plyngent/cli/slash.py index a5a9c24..147018e 100644 --- a/src/plyngent/cli/slash.py +++ b/src/plyngent/cli/slash.py @@ -957,11 +957,11 @@ def history_cmd( def todos_cmd( # noqa: C901, PLR0911, PLR0912, PLR0915 state: ReplState, action: str | None, rest: tuple[str, ...] ) -> None: - """Show or edit the nested todo/task stack. + """Show or edit the LIFO todo stack (TOP = next work / only pop target). - ``/todos`` — list frames - ``/todos push `` or ``/todos push T1; T2`` — new frame of siblings - ``/todos pop`` — pop top breakdown frame + ``/todos`` — list (top first) + ``/todos push <title>`` or ``T1; T2`` — push (first title becomes TOP) + ``/todos pop`` — pop TOP only ``/todos done <id>`` / ``/todos cancel <id>`` — set status ``/todos clear`` — wipe stack """ @@ -988,17 +988,20 @@ def todos_cmd( # noqa: C901, PLR0911, PLR0912, PLR0915 return _await(state.persist_todo_stack()) ids = ", ".join(i.id for i in items) - click.echo(f"pushed frame depth={stack.depth} items=[{ids}]") + top = stack.top + top_s = f"{top.id}:{top.title}" if top else "?" + click.echo(f"pushed [{ids}] (top now {top_s})") click.echo(stack.render()) return if act == "pop": - frame = stack.pop() - if frame is None: + item = stack.pop() + if item is None: click.echo("todo stack empty") return _await(state.persist_todo_stack()) - titles = ", ".join(f"{i.id}:{i.title}" for i in frame.items) or "(empty)" - click.echo(f"popped frame ({titles})") + top = stack.top + top_s = f"{top.id}:{top.title}" if top else "(empty)" + click.echo(f"popped TOP {item.id}: {item.title}; new top={top_s}") click.echo(stack.render()) return if act in {"done", "cancel", "pending", "in_progress"}: diff --git a/src/plyngent/tools/todo.py b/src/plyngent/tools/todo.py index c1baf37..5c143f1 100644 --- a/src/plyngent/tools/todo.py +++ b/src/plyngent/tools/todo.py @@ -10,7 +10,6 @@ if TYPE_CHECKING: from plyngent.agent.todo_stack import TodoStack, TodoStatus -# Module-level bind (same pattern as workspace root) for @tool handlers. _stack: TodoStack | None = None _on_change: Callable[[], None] | None = None @@ -42,9 +41,10 @@ def _notify() -> None: @tool(name="todo_list") def todo_list() -> str: - """List the nested todo stack (frames = breakdown levels). + """Show the LIFO todo stack (TOP = next sub-task to work / pop). - Pattern: push [T1,T2] → push [T1.1,T1.2] → finish children → pop → push [T2.1]… + Not a queue: only the top is popped. Breakdown: push children on top of a + parent, finish and pop them, then the parent (or next sibling) is top again. """ stack = _require_stack() stack.mark_touched() @@ -54,11 +54,11 @@ def todo_list() -> str: @tool(name="todo_push") def todo_push(titles: str, notes: str = "") -> str: - """Push a new breakdown frame with one or more sibling tasks. + """Push task(s) onto the **top** of the LIFO stack. - ``titles``: single title, newline-separated list, ``;``-separated, or JSON - string array. Creates a **new nesting level** (does not append to the current - frame). Example sequence: push ``T1\\nT2`` then push ``T1.1\\nT1.2``. + ``titles``: one title, newlines, ``;``, or JSON array. First title becomes + the new TOP (worked first). Example: ``T1\\nT2`` → top=T1, under=T2; then + ``T1.1\\nT1.2`` → top=T1.1 over T1.2 over T1 over T2. """ stack = _require_stack() parsed = parse_push_titles(titles) @@ -69,24 +69,23 @@ def todo_push(titles: str, notes: str = "") -> str: except ValueError as exc: return f"error: {exc}" _notify() + top = stack.top + top_s = f"{top.id}:{top.title}" if top else "?" ids = ", ".join(i.id for i in items) - return f"pushed frame depth={stack.depth} items=[{ids}]\n{stack.render()}" + return f"pushed [{ids}] (top now {top_s})\n{stack.render()}" @tool(name="todo_pop") def todo_pop() -> str: - """Pop the top breakdown frame (leave the current nesting level). - - Use after finishing children of a task (e.g. after T1.1/T1.2, pop back to - the T1/T2 frame). Does not delete sibling frames below. - """ + """Pop the **top** item only (classic stack). Does not remove items under it.""" stack = _require_stack() - frame = stack.pop() - if frame is None: + item = stack.pop() + if item is None: return "todo stack empty" _notify() - titles = ", ".join(f"{i.id}:{i.title}" for i in frame.items) or "(empty)" - return f"popped frame ({len(frame.items)} item(s): {titles})\n{stack.render()}" + top = stack.top + top_s = f"{top.id}:{top.title}" if top else "(empty)" + return f"popped TOP {item.id}: {item.title} ({item.status}); new top={top_s}\n{stack.render()}" @tool(name="todo_update") @@ -96,7 +95,7 @@ def todo_update( title: str = "", notes: str = "", ) -> str: - """Update a todo by id. ``status``: pending|in_progress|done|cancelled.""" + """Update a todo by id (any depth). Prefer pop when the top task is finished.""" stack = _require_stack() status_arg: TodoStatus | None = None if status.strip(): @@ -119,7 +118,7 @@ def todo_update( @tool(name="todo_clear") def todo_clear() -> str: - """Clear the entire todo stack (all frames).""" + """Clear the entire stack.""" stack = _require_stack() n = stack.clear() _notify() diff --git a/tests/test_agent/test_todo_stack.py b/tests/test_agent/test_todo_stack.py index 322a349..7ca5adb 100644 --- a/tests/test_agent/test_todo_stack.py +++ b/tests/test_agent/test_todo_stack.py @@ -30,29 +30,46 @@ def test_parse_push_titles() -> None: assert parse_push_titles('["A", "B"]') == ["A", "B"] -def test_nested_push_pop_pattern() -> None: - """(push)[T1,T2] (push)[T1.1,T1.2] (pop) (push)[T2.1] …""" +def test_lifo_not_queue() -> None: + """Stack is LIFO: push A then B → top is B; pop is B then A — never FIFO.""" + stack = TodoStack() + _ = stack.push("A") + _ = stack.push("B") + assert stack.top is not None + assert stack.top.title == "B" + first = stack.pop() + second = stack.pop() + assert first is not None and first.title == "B" + assert second is not None and second.title == "A" + assert stack.pop() is None + + +def test_dfs_breakdown_pattern() -> None: + """push [T1,T2] → top T1; push [T1.1,T1.2] → top T1.1; pop children; then T2.1.""" stack = TodoStack() root = stack.push_titles(["T1", "T2"]) assert [i.title for i in root] == ["T1", "T2"] - assert stack.depth == 1 + assert stack.top is not None + assert stack.top.title == "T1" + # bottom→top: T2, T1 + assert [i.title for i in stack.items] == ["T2", "T1"] children = stack.push_titles(["T1.1", "T1.2"]) - assert stack.depth == 2 assert [i.title for i in children] == ["T1.1", "T1.2"] - stack.update(children[0].id, status="done") - stack.update(children[1].id, status="done") + assert stack.top is not None and stack.top.title == "T1.1" + # bottom→top: T2, T1, T1.2, T1.1 + assert [i.title for i in stack.items] == ["T2", "T1", "T1.2", "T1.1"] - popped = stack.pop() - assert popped is not None - assert [i.title for i in popped.items] == ["T1.1", "T1.2"] - assert stack.depth == 1 - assert [i.title for i in stack.frames[0].items] == ["T1", "T2"] - - stack.update(root[0].id, status="done") + p1 = stack.pop() + p2 = stack.pop() + assert p1 is not None and p1.title == "T1.1" + assert p2 is not None and p2.title == "T1.2" + p3 = stack.pop() + assert p3 is not None and p3.title == "T1" + assert stack.top is not None and stack.top.title == "T2" _ = stack.push_titles(["T2.1"]) - assert stack.depth == 2 - assert stack.frames[-1].items[0].title == "T2.1" + assert stack.top is not None and stack.top.title == "T2.1" + assert [i.title for i in stack.items] == ["T2", "T2.1"] def test_todo_stack_needs_review() -> None: @@ -65,17 +82,18 @@ def test_todo_stack_needs_review() -> None: assert not stack.needs_review() -def test_todo_stack_legacy_flat_raw() -> None: +def test_legacy_frames_flatten() -> None: stack = TodoStack.from_raw( { - "items": [{"id": "t1", "title": "old", "status": "pending", "notes": ""}], - "next_id": 2, + "frames": [ + {"items": [{"id": "t1", "title": "T1", "status": "pending", "notes": ""}]}, + {"items": [{"id": "t2", "title": "T1.1", "status": "pending", "notes": ""}]}, + ], + "next_id": 3, } ) - assert stack.depth == 1 - assert stack.all_items()[0].title == "old" - raw = stack.to_raw() - assert "frames" in raw + assert [i.title for i in stack.items] == ["T1", "T1.1"] + assert stack.top is not None and stack.top.title == "T1.1" def test_todo_stack_roundtrip_raw() -> None: @@ -83,8 +101,8 @@ def test_todo_stack_roundtrip_raw() -> None: _ = stack.push_titles(["x", "y"], notes="n") raw = stack.to_raw() restored = TodoStack.from_raw(raw) - assert restored.depth == 1 - assert [i.title for i in restored.frames[0].items] == ["x", "y"] + assert [i.title for i in restored.items] == ["y", "x"] # bottom y, top x + assert restored.top is not None and restored.top.title == "x" async def test_todo_tools_and_persist(tmp_path: object) -> None: @@ -96,20 +114,18 @@ async def test_todo_tools_and_persist(tmp_path: object) -> None: set_todo_stack(stack, on_change=None) registry = ToolRegistry(list(TODO_TOOLS)) out = await registry.execute("todo_push", '{"titles": "T1\\nT2"}') - assert "pushed frame" in out - assert stack.depth == 1 - out2 = await registry.execute("todo_push", '{"titles": "T1.1; T1.2"}') - assert stack.depth == 2 - assert "T1.1" in out2 + assert "top" in out.lower() or "pushed" in out + assert stack.top is not None and stack.top.title == "T1" + _ = await registry.execute("todo_push", '{"titles": "T1.1; T1.2"}') + assert stack.top is not None and stack.top.title == "T1.1" out3 = await registry.execute("todo_pop", "{}") - assert "popped frame" in out3 - assert stack.depth == 1 + assert "popped" in out3 + assert stack.top is not None and stack.top.title == "T1.2" _ = await memory.update_session_todo_stack(session.sid, stack.to_raw()) loaded = await memory.get_session_todo_stack(session.sid) assert loaded is not None again = TodoStack.from_raw(loaded) - assert again.depth == 1 - assert [i.title for i in again.frames[0].items] == ["T1", "T2"] + assert again.top is not None and again.top.title == "T1.2" finally: set_todo_stack(None) await memory.close()