"""Claude Code's task tools: which models get them, the tools' results, the list as a fold of the log (another runtime, a fork), and the reminder after ten model turns without them.""" from __future__ import annotations from pathlib import Path from typing import Any import pytest from noeta.backends.stores.sqlite import SqliteStore from noeta.core import snapshots from noeta.core.agent import Agent from noeta.core.client import Client from noeta.core.extension import TaskView from noeta.core.loop.state import TaskState from noeta.core.runner import Runtime from noeta.core.tool import ToolContext, ToolOutput, text_output, tool_flags from noeta.core.types import Message, ModelRequest, Text, Thinking, ToolResult, ToolUse from noeta.plugins.todos import ( PROMPT_LINE, REMINDER, Todos, canonical, gets_task_tools, turns_since, ) from noeta.plugins.todos.state import Item, TodoState from noeta.plugins.todos.tools import ( AGENT_STEER, LIST_STEER, TaskCreateTool, TaskUpdateTool, repair_create, ) from noeta.testing import ScriptedProvider, reply OLD = "claude-sonnet-4-5" async def _no_sleep(_: float) -> None: return None class Probe: name = "Probe" description = "probe" input_schema: dict[str, Any] = {"type": "object"} read_only = True clearable = False async def run(self, input: dict[str, Any], ctx: ToolContext) -> ToolOutput: return text_output("ok") def _runtime(store: SqliteStore, provider: ScriptedProvider, model: str = OLD, todos: Todos | None = None) -> Runtime: agent = Agent(name="a", model=model, tools=[Probe()], extensions=[todos or Todos()]) return Runtime(store=store, agents={"a": agent}, provider_for=lambda _m: provider, sleep=_no_sleep) def _results(request: ModelRequest) -> list[str]: return [b.content[0].text for m in request.messages for b in m.content # type: ignore[union-attr] if isinstance(b, ToolResult)] def _view(model: str, parent: str | None = None) -> TaskView: return TaskView(state=TaskState(task_id="t", config={"model": model}, parent_id=parent), model=model) @pytest.mark.parametrize(("model", "gets"), [ ("claude-sonnet-4-5", True), ("claude-sonnet-4-5-20250929", True), ("claude-opus-4-6", True), ("claude-opus-4-7", True), ("claude-haiku-4-5", True), ("claude-3-7-sonnet-latest", True), ("claude-opus-4-1-20250805", True), ("claude-opus-4-20250514", True), ("claude-sonnet-4-20250514", True), ("claude-opus-4", True), ("claude-sonnet-4", True), ("claude-opus-4-0", True), ("anthropic.claude-opus-4-20250514-v1:0", True), ("us.anthropic.claude-sonnet-4-6", True), ("eu.anthropic.claude-haiku-4-5-20251001-v1:0", True), ("arn:aws:bedrock:us-east-1:1:application-inference-profile/x", True), ("gateway/deepseek-v4.1-flash", False), ("gpt-5.5", False), ("deepseek-chat", False), ("claude-opus-5-5", False), ("opus", False), ("sonnet", False), ("claude-fable-5-1", False), ("claude-haiku-5-5", False), ("claude-opus-4-8", False), ("claude-sonnet-5", False)]) def test_the_models_claude_code_gives_the_tools(model: str, gets: bool) -> None: assert gets_task_tools(model) is gets @pytest.mark.parametrize(("model", "name"), [ ("claude-opus-4-20250514", "claude-opus-4-0"), ("claude-sonnet-4", "claude-sonnet-4-0"), ("anthropic.claude-opus-4-20250514-v1:0", "claude-opus-4-0"), ("claude-opus-4-1-20250805", "claude-opus-4-1"), ("claude-3-5-haiku-20241022", "claude-3-5-haiku"), ("deepseek-v4-20260101", "deepseek-v4"), # Claude Code's order, quirks included. ("claude-opus-4-10", "claude-opus-4-1")]) def test_claude_codes_canonical_names(model: str, name: str) -> None: assert canonical(model) == name def test_switches_and_sub_agents() -> None: names = [t.name for t in Todos().tools(_view(OLD))] assert names == ["TaskCreate", "TaskGet", "TaskUpdate", "TaskList"] assert all(tool_flags(t).deferred for t in Todos().tools(_view(OLD))) assert Todos().tools(_view("claude-opus-5-5")) == () assert len(Todos(all_models=True).tools(_view("claude-opus-5-5"))) == 4 # A sub-agent runs in the background: no task tools, as Claude Code's # background agents have none. assert Todos(all_models=True).tools(_view(OLD, parent="p")) == () [part] = Todos().context(_view(OLD)) assert part.text == PROMPT_LINE assert Todos().context(_view("claude-opus-5-5")) == () async def test_the_tools_results_and_the_list_folded_on_another_runtime(tmp_path: Path) -> None: store = SqliteStore(str(tmp_path / "t.db")) provider = ScriptedProvider([ reply("", ("TaskCreate", {"subject": "A", "description": "do a"}), ("TaskCreate", {"title": "B", "content": "do b"})), reply("", ("TaskUpdate", {"taskId": "2", "addBlockedBy": ["1"]}), ("TaskUpdate", {"id": "1", "status": "in_progress"})), reply("", ("TaskList", {}), ("TaskGet", {"taskId": "2"}), ("TaskGet", {"taskId": "9"})), reply("", ("TaskUpdate", {"taskId": "1", "status": "completed", "metadata": {"k": 1}}), ("TaskList", {})), reply("", ("TaskUpdate", {"taskId": "2", "status": "deleted"}), ("TaskCreate", {"subject": "C", "description": "c"}), ("TaskUpdate", {"taskId": "2", "subject": "x"})), reply("done"), ]) client = Client(store) tid = await client.create("a", "go") await _runtime(store, provider).run_until_idle() assert _results(provider.requests[-1]) == [ "Task #1 created successfully: A", "Task #2 created successfully: B", "Updated task #2 blockedBy", "Updated task #1 status", "#1 [in_progress] A\n#2 [pending] B [blocked by #1]", "Task #2: B\nStatus: pending\nDescription: do b\nBlocked by: #1", "Task not found", "Updated task #1 metadata, status", "#1 [completed] A\n#2 [pending] B", # A deleted id is never given out again. "Updated task #2 deleted", "Task #3 created successfully: C", "Task not found"] # Another worker folds the same list from the log. again = ScriptedProvider([reply("", ("TaskList", {})), reply("ok")]) await client.send(tid, "more") await _runtime(store, again).run_until_idle() assert _results(again.requests[-1])[-1] == "#1 [completed] A\n#3 [pending] C" # A task that continues this conversation starts with the list. forked = ScriptedProvider([reply("", ("TaskList", {})), reply("ok")]) fork = await client.create("a", "fork", history_from=tid) await _runtime(store, forked).run_until_idle() assert _results(forked.requests[-1])[-1] == "#1 [completed] A\n#3 [pending] C" row = await client.get(fork) assert row is not None await store.close() async def test_claude_5_models_get_neither_tools_nor_reminder(tmp_path: Path) -> None: store = SqliteStore(str(tmp_path / "t.db")) provider = ScriptedProvider([*(reply("", ("Probe", {})) for _ in range(12)), reply("done")]) await Client(store).create("a", "go") await _runtime(store, provider, model="claude-opus-5-5").run_until_idle() await store.close() assert not any(t.name.startswith("Task") for r in provider.requests for t in r.tools) assert not any(REMINDER.split("\n")[0] in m.text for r in provider.requests for m in r.messages) assert not any(PROMPT_LINE in p.text for r in provider.requests for p in r.system) async def test_reminded_after_ten_turns_without_the_tools(tmp_path: Path) -> None: store = SqliteStore(str(tmp_path / "t.db")) provider = ScriptedProvider([ reply("", ("TaskCreate", {"subject": "A", "description": "a"})), *(reply("", ("Probe", {})) for _ in range(21)), reply("done")]) await Client(store).create("a", "go") await _runtime(store, provider).run_until_idle() await store.close() def reminders(i: int) -> list[str]: return [m.text for m in provider.requests[i].messages if m.role == "system" and m.text.startswith("The task tools")] # Request k follows reply k: ten turns after the TaskCreate (reply 1) # the result of reply 11 carries the reminder, with the list; ten turns # later, again. counts = [len(reminders(i)) for i in range(len(provider.requests))] assert counts == [0] * 11 + [1] * 10 + [2] * 2 assert reminders(11)[0] == REMINDER + "\nHere are the existing tasks:\n#1. [pending] A" assert any(PROMPT_LINE in p.text for p in provider.requests[0].system) def test_turns_count_content_blocks_and_skip_thinking() -> None: def step(*blocks: Any) -> list[Message]: uses = [b for b in blocks if isinstance(b, ToolUse)] return [Message(role="assistant", content=list(blocks)), Message(role="user", content=[ToolResult(call_id=u.id, content=[Text(text="ok")]) for u in uses])] messages = [*step(ToolUse(id="1", name="TaskUpdate", input={})), *step(Thinking(text="hm"), Text(text="a"), ToolUse(id="2", name="Probe", input={})), *step(Text(text="b"))] seqs = [1, 2, 3, 4, 5, 6] # Never reminded: every turn shown counts. assert turns_since(messages, seqs, -1) == (3, 4) # Reminded after seq 4 (the second step's results): one turn since. assert turns_since(messages, seqs, 4) == (3, 1) # A compaction left two turns: no count reaches back past it. assert turns_since(messages[2:4], seqs[2:4], -1) == (2, 2) assert turns_since(messages[2:4], [0, 0], -1) == (2, 2) assert turns_since(messages[2:4], [0, 0], 0) == (2, 0) # A /compact kept the last two turns (at seq 0); the reminder (after # seq 4) went with what it folded (through seq 6): both count. assert turns_since(messages[2:], [0, 0, 0, 0], 4, 6) == (3, 3) # A reminder after the compaction counts from itself. assert turns_since(messages[2:], [0, 0, 0, 0], 7, 6) == (3, 0) # A continued task's carried turns (seq 0, reminder reset to 0) do not # count, whatever compaction it carried. assert turns_since(messages[2:], [0, 0, 0, 0], 0, 6) == (3, 0) async def test_input_repairs_and_hints() -> None: records: list[tuple[str, str, Any]] = [] ctx = ToolContext(task_id="t", call_id="c", env=None, # type: ignore[arg-type] state={}, record=lambda e, k, d: records.append((e, k, d))) out = await TaskCreateTool().run({"tasks": [{"subject": "a"}]}, ctx) assert out.is_error and out.content[0].text.startswith( # type: ignore[union-attr] f"{LIST_STEER} InputValidationError: TaskCreate failed") out = await TaskCreateTool().run({"prompt": "p", "subagent_type": "x"}, ctx) assert AGENT_STEER in out.content[0].text # type: ignore[union-attr] assert records == [] out = await TaskCreateTool().run({"task": "Write the tests for the parser"}, ctx) assert out.content[0].text == ( # type: ignore[union-attr] "Task #1 created successfully: Write the tests for the parser") assert repair_create({"description": "x" * 50 + " " + "y" * 50}) == { "description": "x" * 50 + " " + "y" * 50, "subject": "x" * 50} assert repair_create({"subject": "s", "description": "d"}) is None out = await TaskUpdateTool().run({"taskId": "1", "status": "done"}, ctx) assert out.is_error and "`status`" in out.content[0].text # type: ignore[union-attr] def test_the_folded_list_snapshots() -> None: state = TaskState(task_id="t", ext={"todos": TodoState(items=( Item(id="1", subject="a", description="b", blocks=("2",), metadata={"k": [1]}), Item(id="2", subject="c", description="d", blocked_by=("1",))), highest=3)}) data = snapshots.encode(state) assert data is not None and snapshots.decode(data) == state def test_a_continued_task_counts_the_reminder_from_its_own_turns() -> None: """The reminder's place is a seq of the source's log; a continued task's carried messages sit at seq 0, so its count starts with its own turns.""" ext = Todos() reminded = ext.reduce(TodoState(), "task_reminder", {"seq": 40}) assert reminded.reminded_at == 40 assert ext.continue_state(reminded).reminded_at == 0 assert ext.continue_state(TodoState()).reminded_at == -1