fix: handle todo_write string inputs for issue 340

Co-authored-by: gui-yue <yuemeng.gui@gmail.com>

.github/workflows/test.yml

@@ -23,7 +23,7 @@ jobs:
         run: pip install anthropic python-dotenv pytest
 
       - name: Run Python smoke tests
-        run: python -m pytest tests/test_agents_smoke.py -q
+        run: python -m pytest tests -q
 
   web-build:
     runs-on: ubuntu-latest

s05_todo_write/code.py

@@ -28,7 +28,7 @@ Run: python s05_todo_write/code.py
 Needs: pip install anthropic python-dotenv + ANTHROPIC_API_KEY in .env
 """
 
-import os, subprocess
+import ast, json, os, subprocess
 from pathlib import Path
 
 try:
@@ -121,14 +121,31 @@ def run_glob(pattern: str) -> str:
 #  NEW in s05: todo_write tool — plan only, no execution
 # ═══════════════════════════════════════════════════════════
 
+def _normalize_todos(todos):
+    if isinstance(todos, str):
+        try:
+            todos = json.loads(todos)
+        except json.JSONDecodeError:
+            try:
+                todos = ast.literal_eval(todos)
+            except (SyntaxError, ValueError):
+                return None, "Error: todos must be a list or JSON array string"
+    if not isinstance(todos, list):
+        return None, "Error: todos must be a list"
+    for i, t in enumerate(todos):
+        if not isinstance(t, dict):
+            return None, f"Error: todos[{i}] must be an object"
+        if "content" not in t or "status" not in t:
+            return None, f"Error: todos[{i}] missing 'content' or 'status'"
+        if t["status"] not in ("pending", "in_progress", "completed"):
+            return None, f"Error: todos[{i}] has invalid status '{t['status']}'"
+    return todos, None
+
 def run_todo_write(todos: list) -> str:
     global CURRENT_TODOS
-    # validate required fields
+    todos, error = _normalize_todos(todos)
-    for i, t in enumerate(todos):
+    if error:
-        if "content" not in t or "status" not in t:
+        return error
-            return f"Error: todos[{i}] missing 'content' or 'status'"
-        if t["status"] not in ("pending", "in_progress", "completed"):
-            return f"Error: todos[{i}] has invalid status '{t['status']}'"
     CURRENT_TODOS = todos
     lines = ["\n\033[33m## Current Tasks\033[0m"]
     for t in CURRENT_TODOS:

s06_subagent/code.py

@@ -28,7 +28,7 @@ Run: python s06_subagent/code.py
 Needs: pip install anthropic python-dotenv + ANTHROPIC_API_KEY in .env
 """
 
-import os, subprocess
+import ast, json, os, subprocess
 from pathlib import Path
 
 try:
@@ -121,13 +121,31 @@ def run_glob(pattern: str) -> str:
     except Exception as e:
         return f"Error: {e}"
 
+def _normalize_todos(todos):
+    if isinstance(todos, str):
+        try:
+            todos = json.loads(todos)
+        except json.JSONDecodeError:
+            try:
+                todos = ast.literal_eval(todos)
+            except (SyntaxError, ValueError):
+                return None, "Error: todos must be a list or JSON array string"
+    if not isinstance(todos, list):
+        return None, "Error: todos must be a list"
+    for i, t in enumerate(todos):
+        if not isinstance(t, dict):
+            return None, f"Error: todos[{i}] must be an object"
+        if "content" not in t or "status" not in t:
+            return None, f"Error: todos[{i}] missing 'content' or 'status'"
+        if t["status"] not in ("pending", "in_progress", "completed"):
+            return None, f"Error: todos[{i}] has invalid status '{t['status']}'"
+    return todos, None
+
 def run_todo_write(todos: list) -> str:
     global CURRENT_TODOS
-    for i, t in enumerate(todos):
+    todos, error = _normalize_todos(todos)
-        if "content" not in t or "status" not in t:
+    if error:
-            return f"Error: todos[{i}] missing 'content' or 'status'"
+        return error
-        if t["status"] not in ("pending", "in_progress", "completed"):
-            return f"Error: todos[{i}] has invalid status '{t['status']}'"
     CURRENT_TODOS = todos
     lines = ["\n\033[33m## Current Tasks\033[0m"]
     for t in CURRENT_TODOS:

s07_skill_loading/code.py

@@ -26,7 +26,7 @@ Run: python s07_skill_loading/code.py
 Needs: pip install anthropic python-dotenv pyyaml + ANTHROPIC_API_KEY in .env
 """
 
-import os, subprocess
+import ast, json, os, subprocess
 from pathlib import Path
 import yaml
 
@@ -168,13 +168,31 @@ def run_glob(pattern: str) -> str:
     except Exception as e:
         return f"Error: {e}"
 
+def _normalize_todos(todos):
+    if isinstance(todos, str):
+        try:
+            todos = json.loads(todos)
+        except json.JSONDecodeError:
+            try:
+                todos = ast.literal_eval(todos)
+            except (SyntaxError, ValueError):
+                return None, "Error: todos must be a list or JSON array string"
+    if not isinstance(todos, list):
+        return None, "Error: todos must be a list"
+    for i, t in enumerate(todos):
+        if not isinstance(t, dict):
+            return None, f"Error: todos[{i}] must be an object"
+        if "content" not in t or "status" not in t:
+            return None, f"Error: todos[{i}] missing 'content' or 'status'"
+        if t["status"] not in ("pending", "in_progress", "completed"):
+            return None, f"Error: todos[{i}] has invalid status '{t['status']}'"
+    return todos, None
+
 def run_todo_write(todos: list) -> str:
     global CURRENT_TODOS
-    for i, t in enumerate(todos):
+    todos, error = _normalize_todos(todos)
-        if "content" not in t or "status" not in t:
+    if error:
-            return f"Error: todos[{i}] missing 'content' or 'status'"
+        return error
-        if t["status"] not in ("pending", "in_progress", "completed"):
-            return f"Error: todos[{i}] has invalid status '{t['status']}'"
     CURRENT_TODOS = todos
     lines = ["\n\033[33m## Current Tasks\033[0m"]
     for t in CURRENT_TODOS:

s08_context_compact/code.py

@@ -32,7 +32,7 @@ Builds on s07 (skill loading). Usage:
     Needs: pip install anthropic python-dotenv + ANTHROPIC_API_KEY in .env
 """
 
-import os, subprocess, json, time
+import ast, json, os, subprocess, time
 from pathlib import Path
 
 try:
@@ -165,13 +165,31 @@ def run_glob(pattern: str) -> str:
         return "\n".join(results) if results else "(no matches)"
     except Exception as e: return f"Error: {e}"
 
+def _normalize_todos(todos):
+    if isinstance(todos, str):
+        try:
+            todos = json.loads(todos)
+        except json.JSONDecodeError:
+            try:
+                todos = ast.literal_eval(todos)
+            except (SyntaxError, ValueError):
+                return None, "Error: todos must be a list or JSON array string"
+    if not isinstance(todos, list):
+        return None, "Error: todos must be a list"
+    for i, t in enumerate(todos):
+        if not isinstance(t, dict):
+            return None, f"Error: todos[{i}] must be an object"
+        if "content" not in t or "status" not in t:
+            return None, f"Error: todos[{i}] missing 'content' or 'status'"
+        if t["status"] not in ("pending", "in_progress", "completed"):
+            return None, f"Error: todos[{i}] has invalid status '{t['status']}'"
+    return todos, None
+
 def run_todo_write(todos: list) -> str:
     global CURRENT_TODOS
-    for i, t in enumerate(todos):
+    todos, error = _normalize_todos(todos)
-        if "content" not in t or "status" not in t:
+    if error:
-            return f"Error: todos[{i}] missing 'content' or 'status'"
+        return error
-        if t["status"] not in ("pending", "in_progress", "completed"):
-            return f"Error: todos[{i}] has invalid status '{t['status']}'"
     CURRENT_TODOS = todos
     lines = ["\n\033[33m## Current Tasks\033[0m"]
     for t in CURRENT_TODOS:

s20_comprehensive/code.py

@@ -11,7 +11,7 @@ memory, prompt assembly, error recovery, task graph, background tasks, cron,
 teams, protocols, autonomous agents, worktrees, and MCP.
 """
 
-import os, subprocess, json, time, random, threading, re
+import ast, json, os, subprocess, time, random, threading, re
 from pathlib import Path
 from datetime import datetime
 from dataclasses import dataclass, asdict, field
@@ -457,13 +457,31 @@ def call_tool_handler(handler, args: dict, name: str) -> str:
         return f"Error: {e}"
 
 
+def _normalize_todos(todos):
+    if isinstance(todos, str):
+        try:
+            todos = json.loads(todos)
+        except json.JSONDecodeError:
+            try:
+                todos = ast.literal_eval(todos)
+            except (SyntaxError, ValueError):
+                return None, "Error: todos must be a list or JSON array string"
+    if not isinstance(todos, list):
+        return None, "Error: todos must be a list"
+    for i, todo in enumerate(todos):
+        if not isinstance(todo, dict):
+            return None, f"Error: todos[{i}] must be an object"
+        if "content" not in todo or "status" not in todo:
+            return None, f"Error: todos[{i}] missing 'content' or 'status'"
+        if todo["status"] not in ("pending", "in_progress", "completed"):
+            return None, f"Error: todos[{i}] has invalid status '{todo['status']}'"
+    return todos, None
+
 def run_todo_write(todos: list) -> str:
     global CURRENT_TODOS
-    for i, todo in enumerate(todos):
+    todos, error = _normalize_todos(todos)
-        if "content" not in todo or "status" not in todo:
+    if error:
-            return f"Error: todos[{i}] missing 'content' or 'status'"
+        return error
-        if todo["status"] not in ("pending", "in_progress", "completed"):
-            return f"Error: todos[{i}] has invalid status '{todo['status']}'"
     CURRENT_TODOS = todos
     print(f"  \033[33m[todo] updated {len(CURRENT_TODOS)} item(s)\033[0m")
     return f"Updated {len(CURRENT_TODOS)} todos"

tests/test_todo_write_string_input.py

@@ -0,0 +1,115 @@
+import importlib.util
+import os
+import sys
+import tempfile
+import types
+import unittest
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+COURSE_MODULES = [
+    ("s05", REPO_ROOT / "s05_todo_write" / "code.py"),
+    ("s06", REPO_ROOT / "s06_subagent" / "code.py"),
+    ("s07", REPO_ROOT / "s07_skill_loading" / "code.py"),
+    ("s08", REPO_ROOT / "s08_context_compact" / "code.py"),
+    ("s20", REPO_ROOT / "s20_comprehensive" / "code.py"),
+]
+
+
+def load_course_module(module_name: str, module_path: Path, temp_cwd: Path):
+    fake_anthropic = types.ModuleType("anthropic")
+
+    class FakeAnthropic:
+        def __init__(self, *args, **kwargs):
+            self.messages = types.SimpleNamespace(create=None)
+
+    fake_dotenv = types.ModuleType("dotenv")
+    fake_yaml = types.ModuleType("yaml")
+    setattr(fake_anthropic, "Anthropic", FakeAnthropic)
+    setattr(fake_dotenv, "load_dotenv", lambda override=True: None)
+    setattr(fake_yaml, "safe_load", lambda text: {})
+    setattr(fake_yaml, "YAMLError", Exception)
+
+    previous_modules = {
+        "anthropic": sys.modules.get("anthropic"),
+        "dotenv": sys.modules.get("dotenv"),
+        "yaml": sys.modules.get("yaml"),
+    }
+    previous_cwd = Path.cwd()
+    previous_model_id = os.environ.get("MODEL_ID")
+
+    spec = importlib.util.spec_from_file_location(f"{module_name}_todo_test", module_path)
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f"Unable to load {module_path}")
+    module = importlib.util.module_from_spec(spec)
+
+    sys.modules["anthropic"] = fake_anthropic
+    sys.modules["dotenv"] = fake_dotenv
+    sys.modules["yaml"] = fake_yaml
+    try:
+        os.chdir(temp_cwd)
+        os.environ["MODEL_ID"] = "test-model"
+        spec.loader.exec_module(module)
+        return module
+    finally:
+        os.chdir(previous_cwd)
+        if previous_model_id is None:
+            os.environ.pop("MODEL_ID", None)
+        else:
+            os.environ["MODEL_ID"] = previous_model_id
+        for name, previous in previous_modules.items():
+            if previous is None:
+                sys.modules.pop(name, None)
+            else:
+                sys.modules[name] = previous
+
+
+class TodoWriteStringInputTests(unittest.TestCase):
+    def test_issue_340_accepts_json_array_string(self):
+        for module_name, module_path in COURSE_MODULES:
+            with self.subTest(module=module_name), tempfile.TemporaryDirectory() as tmp:
+                module = load_course_module(module_name, module_path, Path(tmp))
+
+                result = module.run_todo_write(
+                    '[{"content": "inspect repo", "status": "pending"}]'
+                )
+
+                self.assertIn("Updated 1", result)
+                self.assertEqual(
+                    module.CURRENT_TODOS,
+                    [{"content": "inspect repo", "status": "pending"}],
+                )
+
+    def test_issue_340_accepts_python_list_repr_string(self):
+        for module_name, module_path in COURSE_MODULES:
+            with self.subTest(module=module_name), tempfile.TemporaryDirectory() as tmp:
+                module = load_course_module(module_name, module_path, Path(tmp))
+
+                result = module.run_todo_write(
+                    "[{'content': 'write tests', 'status': 'in_progress'}]"
+                )
+
+                self.assertIn("Updated 1", result)
+                self.assertEqual(
+                    module.CURRENT_TODOS,
+                    [{"content": "write tests", "status": "in_progress"}],
+                )
+
+    def test_issue_340_does_not_eval_string_inputs(self):
+        for module_name, module_path in COURSE_MODULES:
+            with self.subTest(module=module_name), tempfile.TemporaryDirectory() as tmp:
+                tmp_path = Path(tmp)
+                marker = tmp_path / "eval_was_executed"
+                module = load_course_module(module_name, module_path, tmp_path)
+
+                result = module.run_todo_write(
+                    f"__import__('pathlib').Path({str(marker)!r}).write_text('bad')"
+                )
+
+                self.assertIn("Error:", result)
+                self.assertFalse(marker.exists())
+
+
+if __name__ == "__main__":
+    unittest.main()