core/agent: todo stack is LIFO of task groups, not per-task push

push(titles) creates one group of siblings; pop removes the whole top
group. Matches breakdown: push[T1,T2] → push[T1.1,T1.2] → pop → push[T2.1].
This commit is contained in:
2026-07-18 14:47:38 +08:00
parent 9b38a3b333
commit a7166d863b
5 changed files with 226 additions and 176 deletions
+1 -1
View File
@@ -86,7 +86,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). - 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. - **`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). - **`chat`**: human prompts as tools — `ask_user_line` / `ask_user_choice` / `ask_user_form` (shared `prompting` core).
- **`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. - **`todo`**: LIFO stack of **task groups** (not a queue of tasks) — `todo_push` creates one group of siblings; `todo_pop` removes the whole top group; `todo_update` by id; DFS breakdown 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.
- **`DEFAULT_TOOLS`**: file + process + vcs + chat + todo tool list for a `ToolRegistry`. - **`DEFAULT_TOOLS`**: file + process + vcs + chat + todo tool list for a `ToolRegistry`.
### Prompting (`prompting.py`) ### Prompting (`prompting.py`)
+114 -81
View File
@@ -11,7 +11,7 @@ _OPEN: frozenset[str] = frozenset({"pending", "in_progress"})
class TodoItem(Struct, omit_defaults=True): class TodoItem(Struct, omit_defaults=True):
"""One entry on the todo stack (top of stack = next to work).""" """One task inside a group (siblings share a group; not stacked individually)."""
id: str id: str
title: str title: str
@@ -19,28 +19,34 @@ class TodoItem(Struct, omit_defaults=True):
notes: str = "" notes: str = ""
class TodoStackData(Struct, omit_defaults=True): class TodoGroup(Struct, omit_defaults=True):
"""Serializable stack: ``items[0]`` is bottom, ``items[-1]`` is **top** (LIFO).""" """One stack entry: a group of sibling tasks pushed together."""
items: list[TodoItem] = field(default_factory=list) items: list[TodoItem] = field(default_factory=list)
class TodoStackData(Struct, omit_defaults=True):
"""LIFO of **groups**: ``groups[0]`` bottom, ``groups[-1]`` **TOP** group."""
groups: list[TodoGroup] = field(default_factory=list)
next_id: int = 1 next_id: int = 1
class TodoStack: class TodoStack:
"""True LIFO stack of sub-tasks (not a queue, not a flat checklist). """LIFO stack of **task groups** (not a queue of individual tasks).
- **Top** = last element = next work item. - **Push** always creates **one new group** containing one or more sibling
- ``push`` / multi-push: new items go on the **top**. When pushing several tasks (``push T1, T2`` is one frame, not two stack levels).
titles at once, the **first** title becomes the top (worked first); the - **Pop** removes the entire **top group**.
rest sit under it in order (DFS: children first, then later siblings). - Within a group, update tasks by id (done / in_progress / …).
- ``pop``: remove **only the top** item.
Breakdown pattern:: Breakdown pattern::
push T1, T2 # top=T1, under=T2 push [T1, T2] # top group = {T1, T2}
push T1.1, T1.2 # top=T1.1, then T1.2, then T1, then T2 push [T1.1, T1.2] # top group = {T1.1, T1.2}; under = {T1, T2}
pop / done … # clear T1.1, T1.2, then T1 # finish T1.1 / T1.2 via update…
push T2.1 # top=T2.1 over T2 pop # leave child group; top again = {T1, T2}
push [T2.1] # top group = {T2.1}
""" """
def __init__(self, data: TodoStackData | None = None) -> None: def __init__(self, data: TodoStackData | None = None) -> None:
@@ -48,15 +54,19 @@ class TodoStack:
self._touched_this_turn: bool = False self._touched_this_turn: bool = False
@property @property
def items(self) -> list[TodoItem]: def groups(self) -> list[TodoGroup]:
"""Bottom → top order (last is top).""" """Bottom → top (last is top group)."""
return self._data.items return self._data.groups
@property @property
def top(self) -> TodoItem | None: def top_group(self) -> TodoGroup | None:
if not self._data.items: if not self._data.groups:
return None return None
return self._data.items[-1] return self._data.groups[-1]
@property
def depth(self) -> int:
return len(self._data.groups)
@property @property
def touched_this_turn(self) -> bool: def touched_this_turn(self) -> bool:
@@ -68,11 +78,14 @@ class TodoStack:
def mark_touched(self) -> None: def mark_touched(self) -> None:
self._touched_this_turn = True self._touched_this_turn = True
def all_items(self) -> list[TodoItem]:
return [item for group in self._data.groups for item in group.items]
def open_items(self) -> list[TodoItem]: def open_items(self) -> list[TodoItem]:
return [item for item in self._data.items if item.status in _OPEN] return [item for item in self.all_items() if item.status in _OPEN]
def is_empty(self) -> bool: def is_empty(self) -> bool:
return not self._data.items return not self._data.groups
def needs_review(self) -> bool: def needs_review(self) -> bool:
return bool(self.open_items()) and not self._touched_this_turn return bool(self.open_items()) and not self._touched_this_turn
@@ -81,26 +94,13 @@ class TodoStack:
return self._data return self._data
@classmethod @classmethod
def from_raw(cls, raw: object | None) -> TodoStack: def from_raw(cls, raw: object | None) -> TodoStack: # noqa: C901, PLR0911
if raw is None: if raw is None or not isinstance(raw, dict):
return cls()
# Nested-frame shape (brief intermediate design) → flatten bottom→top.
if isinstance(raw, dict) and "frames" in raw:
try:
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_map = cast("dict[str, object]", frame)
frame_items = msgspec.convert(frame_map.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() return cls()
blob = cast("dict[str, object]", raw)
# Current shape: {groups: [...], next_id}
if "groups" in blob:
try: try:
data = msgspec.convert(raw, type=TodoStackData) data = msgspec.convert(raw, type=TodoStackData)
except msgspec.ValidationError, TypeError, ValueError: except msgspec.ValidationError, TypeError, ValueError:
@@ -109,19 +109,60 @@ class TodoStack:
data = msgspec.structs.replace(data, next_id=1) data = msgspec.structs.replace(data, next_id=1)
return cls(data) return cls(data)
# Intermediate nested frames → groups (same structure).
if "frames" in blob:
try:
frames_raw = blob.get("frames")
next_raw = blob.get("next_id", 1)
next_id = max(1, int(next_raw) if isinstance(next_raw, int | str) else 1)
groups: list[TodoGroup] = []
if isinstance(frames_raw, list):
for frame in cast("list[object]", frames_raw):
if not isinstance(frame, dict):
continue
frame_map = cast("dict[str, object]", frame)
frame_items = msgspec.convert(frame_map.get("items"), type=list[TodoItem])
if frame_items:
groups.append(TodoGroup(items=frame_items))
return cls(TodoStackData(groups=groups, next_id=next_id))
except msgspec.ValidationError, TypeError, ValueError:
return cls()
# Flat items list → one group (legacy).
if "items" in blob:
try:
items = msgspec.convert(blob.get("items"), type=list[TodoItem])
next_raw = blob.get("next_id", 1)
next_id = max(1, int(next_raw) if isinstance(next_raw, int | str) else 1)
except msgspec.ValidationError, TypeError, ValueError:
return cls()
groups = [TodoGroup(items=items)] if items else []
return cls(TodoStackData(groups=groups, next_id=next_id))
return cls()
def to_raw(self) -> dict[str, object]: def to_raw(self) -> dict[str, object]:
out: object = msgspec.to_builtins(self._data) out: object = msgspec.to_builtins(self._data)
if not isinstance(out, dict): if not isinstance(out, dict):
return {"items": [], "next_id": 1} return {"groups": [], "next_id": 1}
raw = cast("dict[object, object]", out) raw = cast("dict[object, object]", out)
return {str(key): value for key, value in raw.items()} return {str(key): value for key, value in raw.items()}
def render(self) -> str: def render(self) -> str:
if not self._data.items: if not self._data.groups:
return "(todo stack empty — top is empty)" return "(todo stack empty — no groups)"
lines: list[str] = ["(LIFO stack: TOP = next to work)"] lines: list[str] = [
# Show top first so the model sees work order. f"(LIFO of groups: depth={self.depth}; TOP group = current breakdown level)",
for offset, item in enumerate(reversed(self._data.items)): ]
# Top group first for the model.
for offset, group in enumerate(reversed(self._data.groups)):
depth = self.depth - 1 - offset
tag = " TOP" if offset == 0 else ""
lines.append(f"group d={depth}{tag}:")
if not group.items:
lines.append(" (empty group)")
continue
for item in group.items:
mark = { mark = {
"pending": "[ ]", "pending": "[ ]",
"in_progress": "[~]", "in_progress": "[~]",
@@ -129,20 +170,17 @@ class TodoStack:
"cancelled": "[-]", "cancelled": "[-]",
}.get(item.status, "[?]") }.get(item.status, "[?]")
note = f"{item.notes}" if item.notes else "" note = f"{item.notes}" if item.notes else ""
tag = " TOP" if offset == 0 else "" lines.append(f" {mark} {item.id}: {item.title}{note}")
lines.append(f"{mark}{tag} {item.id}: {item.title}{note}")
return "\n".join(lines) return "\n".join(lines)
def review_prompt(self) -> str: def review_prompt(self) -> str:
return ( return (
"Todo stack review (internal control — not a human message).\n" "Todo stack review (internal control — not a human message).\n"
"This is a LIFO stack (not a queue): TOP is the only pop target and " "This is a LIFO stack of **task groups** (not a queue of single tasks).\n"
"the next sub-task to execute. Open work remains and you did not call " "push([T1,T2]) creates one group of siblings; push([T1.1,T1.2]) pushes a "
"any todo_* tools this turn.\n" "child group; pop removes the whole top group. Update items by id; "
"Breakdown: push children onto the top, finish/pop them, then the " "pop when a breakdown level is finished. Open work remains and you "
"parent (or next sibling under it) becomes top again. " "did not call any todo_* tools this turn.\n\n"
"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()}" f"Current stack:\n{self.render()}"
) )
@@ -151,53 +189,47 @@ class TodoStack:
self._data.next_id += 1 self._data.next_id += 1
return TodoItem(id=item_id, title=title, status=status, notes=notes) return TodoItem(id=item_id, title=title, status=status, notes=notes)
def push_titles( def push_group(
self, self,
titles: list[str], titles: list[str],
*, *,
notes: str = "", notes: str = "",
status: TodoStatus = "pending", status: TodoStatus = "pending",
) -> list[TodoItem]: ) -> TodoGroup:
"""Push one or more tasks onto the top (LIFO). """Push **one** new group containing all *titles* as siblings."""
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()] cleaned = [t.strip() for t in titles if t and t.strip()]
if not cleaned: if not cleaned:
msg = "at least one non-empty title is required" msg = "at least one non-empty title is required"
raise ValueError(msg) raise ValueError(msg)
note = notes.strip() note = notes.strip()
# Push last→first so first title ends on top. items = [self._alloc(title, notes=note, status=status) for title in cleaned]
created_rev: list[TodoItem] = [] group = TodoGroup(items=items)
for title in reversed(cleaned): self._data.groups.append(group)
item = self._alloc(title, notes=note, status=status)
self._data.items.append(item)
created_rev.append(item)
self.mark_touched() self.mark_touched()
# Return in the same order as *titles* (first = top). return group
return list(reversed(created_rev))
def push(self, title: str, *, notes: str = "", status: TodoStatus = "pending") -> TodoItem: def push(self, title: str, *, notes: str = "", status: TodoStatus = "pending") -> TodoItem:
return self.push_titles([title], notes=notes, status=status)[0] """Push a group with a single task (still one stack level)."""
group = self.push_group([title], notes=notes, status=status)
return group.items[0]
def pop(self) -> TodoItem | None: def pop(self) -> TodoGroup | None:
"""Remove and return the **top** item only (classic stack pop).""" """Pop and return the **top group** (all siblings in that push)."""
if not self._data.items: if not self._data.groups:
return None return None
item = self._data.items.pop() group = self._data.groups.pop()
self.mark_touched() self.mark_touched()
return item return group
def clear(self) -> int: def clear(self) -> int:
n = len(self._data.items) n = sum(len(g.items) for g in self._data.groups)
self._data.items.clear() self._data.groups.clear()
if n: if n:
self.mark_touched() self.mark_touched()
return n return n
def get(self, item_id: str) -> TodoItem | None: def get(self, item_id: str) -> TodoItem | None:
for item in self._data.items: for item in self.all_items():
if item.id == item_id: if item.id == item_id:
return item return item
return None return None
@@ -210,7 +242,8 @@ class TodoStack:
status: TodoStatus | None = None, status: TodoStatus | None = None,
notes: str | None = None, notes: str | None = None,
) -> TodoItem: ) -> TodoItem:
for index, item in enumerate(self._data.items): for group in self._data.groups:
for index, item in enumerate(group.items):
if item.id != item_id: if item.id != item_id:
continue continue
new_title = title.strip() if title is not None else item.title new_title = title.strip() if title is not None else item.title
@@ -220,7 +253,7 @@ class TodoStack:
new_status = status if status is not None else item.status new_status = status if status is not None else item.status
new_notes = notes if notes is not None else item.notes 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) updated = TodoItem(id=item.id, title=new_title, status=new_status, notes=new_notes)
self._data.items[index] = updated group.items[index] = updated
self.mark_touched() self.mark_touched()
return updated return updated
msg = f"unknown todo id {item_id!r}" msg = f"unknown todo id {item_id!r}"
+11 -14
View File
@@ -952,11 +952,11 @@ def history_cmd(
def todos_cmd( # noqa: C901, PLR0911, PLR0912, PLR0915 def todos_cmd( # noqa: C901, PLR0911, PLR0912, PLR0915
state: ReplState, action: str | None, rest: tuple[str, ...] state: ReplState, action: str | None, rest: tuple[str, ...]
) -> None: ) -> None:
"""Show or edit the LIFO todo stack (TOP = next work / only pop target). """Show or edit the LIFO stack of **task groups**.
``/todos`` — list (top first) ``/todos`` — list (top group first)
``/todos push <title>`` or ``T1; T2`` — push (first title becomes TOP) ``/todos push T1; T2`` — one new group of siblings
``/todos pop`` — pop TOP only ``/todos pop`` — pop entire top group
``/todos done <id>`` / ``/todos cancel <id>`` — set status ``/todos done <id>`` / ``/todos cancel <id>`` — set status
``/todos clear`` — wipe stack ``/todos clear`` — wipe stack
""" """
@@ -977,26 +977,23 @@ def todos_cmd( # noqa: C901, PLR0911, PLR0912, PLR0915
click.echo("error: no titles to push") click.echo("error: no titles to push")
return return
try: try:
items = stack.push_titles(titles) group = stack.push_group(titles)
except ValueError as exc: except ValueError as exc:
click.echo(f"error: {exc}") click.echo(f"error: {exc}")
return return
_await(state.persist_todo_stack()) _await(state.persist_todo_stack())
ids = ", ".join(i.id for i in items) ids = ", ".join(i.id for i in group.items)
top = stack.top click.echo(f"pushed group depth={stack.depth} items=[{ids}]")
top_s = f"{top.id}:{top.title}" if top else "?"
click.echo(f"pushed [{ids}] (top now {top_s})")
click.echo(stack.render()) click.echo(stack.render())
return return
if act == "pop": if act == "pop":
item = stack.pop() group = stack.pop()
if item is None: if group is None:
click.echo("todo stack empty") click.echo("todo stack empty")
return return
_await(state.persist_todo_stack()) _await(state.persist_todo_stack())
top = stack.top titles = ", ".join(f"{i.id}:{i.title}" for i in group.items) or "(empty)"
top_s = f"{top.id}:{top.title}" if top else "(empty)" click.echo(f"popped TOP group ({titles})")
click.echo(f"popped TOP {item.id}: {item.title}; new top={top_s}")
click.echo(stack.render()) click.echo(stack.render())
return return
if act in {"done", "cancel", "pending", "in_progress"}: if act in {"done", "cancel", "pending", "in_progress"}:
+19 -20
View File
@@ -41,10 +41,10 @@ def _notify() -> None:
@tool(name="todo_list") @tool(name="todo_list")
def todo_list() -> str: def todo_list() -> str:
"""Show the LIFO todo stack (TOP = next sub-task to work / pop). """Show the LIFO stack of **task groups** (TOP group = current breakdown level).
Not a queue: only the top is popped. Breakdown: push children on top of a Push creates one group of siblings; pop removes the whole top group.
parent, finish and pop them, then the parent (or next sibling) is top again. Pattern: push [T1,T2] → push [T1.1,T1.2] → finish children → pop → push [T2.1]…
""" """
stack = _require_stack() stack = _require_stack()
stack.mark_touched() stack.mark_touched()
@@ -54,38 +54,37 @@ def todo_list() -> str:
@tool(name="todo_push") @tool(name="todo_push")
def todo_push(titles: str, notes: str = "") -> str: def todo_push(titles: str, notes: str = "") -> str:
"""Push task(s) onto the **top** of the LIFO stack. """Push **one task group** (siblings) onto the stack — not one level per title.
``titles``: one title, newlines, ``;``, or JSON array. First title becomes ``titles``: one title, newlines, ``;``, or JSON array. All listed titles
the new TOP (worked first). Example: ``T1\\nT2`` → top=T1, under=T2; then become members of a single new TOP group. Example: ``T1\\nT2`` pushes one
``T1.1\\nT1.2`` → top=T1.1 over T1.2 over T1 over T2. group {T1, T2}; a later ``T1.1\\nT1.2`` pushes a child group above it.
""" """
stack = _require_stack() stack = _require_stack()
parsed = parse_push_titles(titles) parsed = parse_push_titles(titles)
if not parsed: if not parsed:
return "error: titles must contain at least one non-empty title" return "error: titles must contain at least one non-empty title"
try: try:
items = stack.push_titles(parsed, notes=notes) group = stack.push_group(parsed, notes=notes)
except ValueError as exc: except ValueError as exc:
return f"error: {exc}" return f"error: {exc}"
_notify() _notify()
top = stack.top ids = ", ".join(i.id for i in group.items)
top_s = f"{top.id}:{top.title}" if top else "?" return f"pushed group (depth={stack.depth}) items=[{ids}]\n{stack.render()}"
ids = ", ".join(i.id for i in items)
return f"pushed [{ids}] (top now {top_s})\n{stack.render()}"
@tool(name="todo_pop") @tool(name="todo_pop")
def todo_pop() -> str: def todo_pop() -> str:
"""Pop the **top** item only (classic stack). Does not remove items under it.""" """Pop the entire **top group** (all siblings from that push)."""
stack = _require_stack() stack = _require_stack()
item = stack.pop() group = stack.pop()
if item is None: if group is None:
return "todo stack empty" return "todo stack empty"
_notify() _notify()
top = stack.top titles = ", ".join(f"{i.id}:{i.title}" for i in group.items) or "(empty)"
top_s = f"{top.id}:{top.title}" if top else "(empty)" top = stack.top_group
return f"popped TOP {item.id}: {item.title} ({item.status}); new top={top_s}\n{stack.render()}" top_s = "(empty)" if top is None else ", ".join(i.id for i in top.items)
return f"popped TOP group ({titles}); new top group=[{top_s}]\n{stack.render()}"
@tool(name="todo_update") @tool(name="todo_update")
@@ -95,7 +94,7 @@ def todo_update(
title: str = "", title: str = "",
notes: str = "", notes: str = "",
) -> str: ) -> str:
"""Update a todo by id (any depth). Prefer pop when the top task is finished.""" """Update a task by id inside any group. Pop the group when that level is done."""
stack = _require_stack() stack = _require_stack()
status_arg: TodoStatus | None = None status_arg: TodoStatus | None = None
if status.strip(): if status.strip():
@@ -118,7 +117,7 @@ def todo_update(
@tool(name="todo_clear") @tool(name="todo_clear")
def todo_clear() -> str: def todo_clear() -> str:
"""Clear the entire stack.""" """Clear all groups on the stack."""
stack = _require_stack() stack = _require_stack()
n = stack.clear() n = stack.clear()
_notify() _notify()
+67 -46
View File
@@ -30,46 +30,54 @@ def test_parse_push_titles() -> None:
assert parse_push_titles('["A", "B"]') == ["A", "B"] assert parse_push_titles('["A", "B"]') == ["A", "B"]
def test_lifo_not_queue() -> None: def test_push_is_group_not_per_task_stack() -> None:
"""Stack is LIFO: push A then B → top is B; pop is B then A — never FIFO.""" """Multi-title push is one group; pop removes the whole group."""
stack = TodoStack() stack = TodoStack()
_ = stack.push("A") g = stack.push_group(["T1", "T2"])
_ = stack.push("B") assert stack.depth == 1
assert stack.top is not None assert [i.title for i in g.items] == ["T1", "T2"]
assert stack.top.title == "B" assert stack.top_group is g
first = stack.pop() # Not two stack levels of single tasks
second = stack.pop() assert len(stack.groups) == 1
assert first is not None and first.title == "B"
assert second is not None and second.title == "A" g2 = stack.push_group(["T1.1", "T1.2"])
assert stack.pop() is None assert stack.depth == 2
assert [i.title for i in g2.items] == ["T1.1", "T1.2"]
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 stack.top_group is not None
assert [i.title for i in stack.top_group.items] == ["T1", "T2"]
def test_dfs_breakdown_pattern() -> None: def test_dfs_breakdown_with_groups() -> None:
"""push [T1,T2] → top T1; push [T1.1,T1.2] → top T1.1; pop children; then T2.1.""" """push [T1,T2] → push [T1.1,T1.2] → pop → push [T2.1]."""
stack = TodoStack() stack = TodoStack()
root = stack.push_titles(["T1", "T2"]) root = stack.push_group(["T1", "T2"])
assert [i.title for i in root] == ["T1", "T2"] children = stack.push_group(["T1.1", "T1.2"])
assert stack.top is not None assert stack.depth == 2
assert stack.top.title == "T1" stack.update(children.items[0].id, status="done")
# bottom→top: T2, T1 stack.update(children.items[1].id, status="done")
assert [i.title for i in stack.items] == ["T2", "T1"] _ = stack.pop()
assert stack.depth == 1
stack.update(root.items[0].id, status="done")
_ = stack.push_group(["T2.1"])
assert stack.depth == 2
assert stack.top_group is not None
assert stack.top_group.items[0].title == "T2.1"
assert stack.groups[0].items[1].title == "T2"
children = stack.push_titles(["T1.1", "T1.2"])
assert [i.title for i in children] == ["T1.1", "T1.2"]
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"]
p1 = stack.pop() def test_single_title_still_one_group() -> None:
p2 = stack.pop() stack = TodoStack()
assert p1 is not None and p1.title == "T1.1" item = stack.push("only")
assert p2 is not None and p2.title == "T1.2" assert stack.depth == 1
p3 = stack.pop() assert item.title == "only"
assert p3 is not None and p3.title == "T1" g = stack.pop()
assert stack.top is not None and stack.top.title == "T2" assert g is not None and len(g.items) == 1
_ = stack.push_titles(["T2.1"]) assert stack.is_empty()
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: def test_todo_stack_needs_review() -> None:
@@ -82,8 +90,17 @@ def test_todo_stack_needs_review() -> None:
assert not stack.needs_review() assert not stack.needs_review()
def test_legacy_frames_flatten() -> None: def test_legacy_flat_and_frames_migrate() -> None:
stack = TodoStack.from_raw( flat = TodoStack.from_raw(
{
"items": [{"id": "t1", "title": "old", "status": "pending", "notes": ""}],
"next_id": 2,
}
)
assert flat.depth == 1
assert flat.all_items()[0].title == "old"
framed = TodoStack.from_raw(
{ {
"frames": [ "frames": [
{"items": [{"id": "t1", "title": "T1", "status": "pending", "notes": ""}]}, {"items": [{"id": "t1", "title": "T1", "status": "pending", "notes": ""}]},
@@ -92,17 +109,19 @@ def test_legacy_frames_flatten() -> None:
"next_id": 3, "next_id": 3,
} }
) )
assert [i.title for i in stack.items] == ["T1", "T1.1"] assert framed.depth == 2
assert stack.top is not None and stack.top.title == "T1.1" assert framed.top_group is not None
assert framed.top_group.items[0].title == "T1.1"
def test_todo_stack_roundtrip_raw() -> None: def test_todo_stack_roundtrip_raw() -> None:
stack = TodoStack() stack = TodoStack()
_ = stack.push_titles(["x", "y"], notes="n") _ = stack.push_group(["x", "y"], notes="n")
raw = stack.to_raw() raw = stack.to_raw()
assert "groups" in raw
restored = TodoStack.from_raw(raw) restored = TodoStack.from_raw(raw)
assert [i.title for i in restored.items] == ["y", "x"] # bottom y, top x assert restored.depth == 1
assert restored.top is not None and restored.top.title == "x" assert [i.title for i in restored.groups[0].items] == ["x", "y"]
async def test_todo_tools_and_persist(tmp_path: object) -> None: async def test_todo_tools_and_persist(tmp_path: object) -> None:
@@ -114,18 +133,20 @@ async def test_todo_tools_and_persist(tmp_path: object) -> None:
set_todo_stack(stack, on_change=None) set_todo_stack(stack, on_change=None)
registry = ToolRegistry(list(TODO_TOOLS)) registry = ToolRegistry(list(TODO_TOOLS))
out = await registry.execute("todo_push", '{"titles": "T1\\nT2"}') out = await registry.execute("todo_push", '{"titles": "T1\\nT2"}')
assert "top" in out.lower() or "pushed" in out assert "group" in out.lower() or "pushed" in out
assert stack.top is not None and stack.top.title == "T1" assert stack.depth == 1
assert [i.title for i in stack.groups[0].items] == ["T1", "T2"]
_ = await registry.execute("todo_push", '{"titles": "T1.1; T1.2"}') _ = await registry.execute("todo_push", '{"titles": "T1.1; T1.2"}')
assert stack.top is not None and stack.top.title == "T1.1" assert stack.depth == 2
out3 = await registry.execute("todo_pop", "{}") out3 = await registry.execute("todo_pop", "{}")
assert "popped" in out3 assert "popped" in out3
assert stack.top is not None and stack.top.title == "T1.2" assert stack.depth == 1
_ = await memory.update_session_todo_stack(session.sid, stack.to_raw()) _ = await memory.update_session_todo_stack(session.sid, stack.to_raw())
loaded = await memory.get_session_todo_stack(session.sid) loaded = await memory.get_session_todo_stack(session.sid)
assert loaded is not None assert loaded is not None
again = TodoStack.from_raw(loaded) again = TodoStack.from_raw(loaded)
assert again.top is not None and again.top.title == "T1.2" assert again.depth == 1
assert [i.title for i in again.groups[0].items] == ["T1", "T2"]
finally: finally:
set_todo_stack(None) set_todo_stack(None)
await memory.close() await memory.close()