fix: add cycle detection and atomic writes to task dependency updates

Author: cs50victor

.gitignore

@@ -13,3 +13,4 @@ venv/
 .ruff_cache/
 *.so
 .DS_Store
+.mcp.json

src/claude_teams/server.py

@@ -285,7 +285,10 @@ def task_update(
 @mcp.tool
 def task_list(team_name: str) -> list[dict]:
     """List all tasks for a team with their current status and assignments."""
-    result = tasks.list_tasks(team_name)
+    try:
+        result = tasks.list_tasks(team_name)
+    except ValueError as e:
+        raise ToolError(str(e))
     return [t.model_dump(by_alias=True, exclude_none=True) for t in result]
 
 

src/claude_teams/tasks.py

@@ -2,6 +2,7 @@
 
 import fcntl
 import json
+from collections import deque
 from contextlib import contextmanager
 from pathlib import Path
 from typing import Any
@@ -30,13 +31,34 @@ def _tasks_dir(base_dir: Path | None = None) -> Path:
 _STATUS_ORDER = {"pending": 0, "in_progress": 1, "completed": 2}
 
 
-def _sync_dependency(team_dir: Path, target_id: str, field: str, value: str) -> None:
-    fpath = team_dir / f"{target_id}.json"
-    other = TaskFile(**json.loads(fpath.read_text()))
-    lst = getattr(other, field)
-    if value not in lst:
-        lst.append(value)
-        fpath.write_text(json.dumps(other.model_dump(by_alias=True, exclude_none=True)))
+def _flush_pending_writes(pending_writes: dict[Path, TaskFile]) -> None:
+    for path, task_obj in pending_writes.items():
+        path.write_text(json.dumps(task_obj.model_dump(by_alias=True, exclude_none=True)))
+
+
+def _would_create_cycle(
+    team_dir: Path, from_id: str, to_id: str, pending_edges: dict[str, set[str]]
+) -> bool:
+    """True if making from_id blocked_by to_id creates a cycle.
+
+    BFS from to_id through blocked_by chains (on-disk + pending);
+    cycle if it reaches from_id.
+    """
+    visited: set[str] = set()
+    queue = deque([to_id])
+    while queue:
+        current = queue.popleft()
+        if current == from_id:
+            return True
+        if current in visited:
+            continue
+        visited.add(current)
+        fpath = team_dir / f"{current}.json"
+        if fpath.exists():
+            task = TaskFile(**json.loads(fpath.read_text()))
+            queue.extend(d for d in task.blocked_by if d not in visited)
+        queue.extend(d for d in pending_edges.get(current, set()) if d not in visited)
+    return False
 
 
 def next_task_id(team_name: str, base_dir: Path | None = None) -> str:
@@ -110,51 +132,43 @@ def update_task(
     fpath = team_dir / f"{task_id}.json"
 
     with file_lock(lock_path):
+        # --- Phase 1: Read ---
         task = TaskFile(**json.loads(fpath.read_text()))
 
-        if subject is not None:
-            task.subject = subject
-        if description is not None:
-            task.description = description
-        if active_form is not None:
-            task.active_form = active_form
-        if owner is not None:
-            task.owner = owner
+        # --- Phase 2: Validate (no disk writes) ---
+        pending_edges: dict[str, set[str]] = {}
 
         if add_blocks:
             for b in add_blocks:
                 if b == task_id:
                     raise ValueError(f"Task {task_id} cannot block itself")
                 if not (team_dir / f"{b}.json").exists():
                     raise ValueError(f"Referenced task {b!r} does not exist")
-            existing = set(task.blocks)
             for b in add_blocks:
-                if b not in existing:
-                    task.blocks.append(b)
-                    existing.add(b)
-                _sync_dependency(team_dir, b, "blocked_by", task_id)
+                pending_edges.setdefault(b, set()).add(task_id)
 
         if add_blocked_by:
             for b in add_blocked_by:
                 if b == task_id:
                     raise ValueError(f"Task {task_id} cannot be blocked by itself")
                 if not (team_dir / f"{b}.json").exists():
                     raise ValueError(f"Referenced task {b!r} does not exist")
-            existing = set(task.blocked_by)
             for b in add_blocked_by:
-                if b not in existing:
-                    task.blocked_by.append(b)
-                    existing.add(b)
-                _sync_dependency(team_dir, b, "blocks", task_id)
+                pending_edges.setdefault(task_id, set()).add(b)
 
-        if metadata is not None:
-            current = task.metadata or {}
-            for k, v in metadata.items():
-                if v is None:
-                    current.pop(k, None)
-                else:
-                    current[k] = v
-            task.metadata = current if current else None
+        if add_blocks:
+            for b in add_blocks:
+                if _would_create_cycle(team_dir, b, task_id, pending_edges):
+                    raise ValueError(
+                        f"Adding block {task_id} -> {b} would create a circular dependency"
+                    )
+
+        if add_blocked_by:
+            for b in add_blocked_by:
+                if _would_create_cycle(team_dir, task_id, b, pending_edges):
+                    raise ValueError(
+                        f"Adding dependency {task_id} blocked_by {b} would create a circular dependency"
+                    )
 
         if status is not None and status != "deleted":
             cur_order = _STATUS_ORDER[task.status]
@@ -165,8 +179,11 @@ def update_task(
                 raise ValueError(
                     f"Cannot transition from {task.status!r} to {status!r}"
                 )
-            if status in ("in_progress", "completed") and task.blocked_by:
-                for blocker_id in task.blocked_by:
+            effective_blocked_by = set(task.blocked_by)
+            if add_blocked_by:
+                effective_blocked_by.update(add_blocked_by)
+            if status in ("in_progress", "completed") and effective_blocked_by:
+                for blocker_id in effective_blocked_by:
                     blocker_path = team_dir / f"{blocker_id}.json"
                     if blocker_path.exists():
                         blocker = TaskFile(**json.loads(blocker_path.read_text()))
@@ -175,8 +192,60 @@ def update_task(
                                 f"Cannot set status to {status!r}: "
                                 f"blocked by task {blocker_id} (status: {blocker.status!r})"
                             )
-            task.status = status
 
+        # --- Phase 3: Mutate (in-memory only) ---
+        pending_writes: dict[Path, TaskFile] = {}
+
+        if subject is not None:
+            task.subject = subject
+        if description is not None:
+            task.description = description
+        if active_form is not None:
+            task.active_form = active_form
+        if owner is not None:
+            task.owner = owner
+
+        if add_blocks:
+            existing = set(task.blocks)
+            for b in add_blocks:
+                if b not in existing:
+                    task.blocks.append(b)
+                    existing.add(b)
+                b_path = team_dir / f"{b}.json"
+                if b_path in pending_writes:
+                    other = pending_writes[b_path]
+                else:
+                    other = TaskFile(**json.loads(b_path.read_text()))
+                if task_id not in other.blocked_by:
+                    other.blocked_by.append(task_id)
+                pending_writes[b_path] = other
+
+        if add_blocked_by:
+            existing = set(task.blocked_by)
+            for b in add_blocked_by:
+                if b not in existing:
+                    task.blocked_by.append(b)
+                    existing.add(b)
+                b_path = team_dir / f"{b}.json"
+                if b_path in pending_writes:
+                    other = pending_writes[b_path]
+                else:
+                    other = TaskFile(**json.loads(b_path.read_text()))
+                if task_id not in other.blocks:
+                    other.blocks.append(task_id)
+                pending_writes[b_path] = other
+
+        if metadata is not None:
+            current = task.metadata or {}
+            for k, v in metadata.items():
+                if v is None:
+                    current.pop(k, None)
+                else:
+                    current[k] = v
+            task.metadata = current if current else None
+
+        if status is not None and status != "deleted":
+            task.status = status
             if status == "completed":
                 for f in team_dir.glob("*.json"):
                     try:
@@ -185,22 +254,27 @@ def update_task(
                         continue
                     if f.stem == task_id:
                         continue
-                    other = TaskFile(**json.loads(f.read_text()))
+                    if f in pending_writes:
+                        other = pending_writes[f]
+                    else:
+                        other = TaskFile(**json.loads(f.read_text()))
                     if task_id in other.blocked_by:
                         other.blocked_by.remove(task_id)
-                        f.write_text(
-                            json.dumps(other.model_dump(by_alias=True, exclude_none=True))
-                        )
+                        pending_writes[f] = other
 
         if status == "deleted":
             task.status = "deleted"
-            fpath.unlink()
             for f in team_dir.glob("*.json"):
                 try:
                     int(f.stem)
                 except ValueError:
                     continue
-                other = TaskFile(**json.loads(f.read_text()))
+                if f.stem == task_id:
+                    continue
+                if f in pending_writes:
+                    other = pending_writes[f]
+                else:
+                    other = TaskFile(**json.loads(f.read_text()))
                 changed = False
                 if task_id in other.blocked_by:
                     other.blocked_by.remove(task_id)
@@ -209,21 +283,26 @@ def update_task(
                     other.blocks.remove(task_id)
                     changed = True
                 if changed:
-                    f.write_text(
-                        json.dumps(other.model_dump(by_alias=True, exclude_none=True))
-                    )
-            return task
+                    pending_writes[f] = other
 
-        fpath.write_text(
-            json.dumps(task.model_dump(by_alias=True, exclude_none=True))
-        )
+        # --- Phase 4: Write ---
+        if status == "deleted":
+            _flush_pending_writes(pending_writes)
+            fpath.unlink()
+        else:
+            fpath.write_text(
+                json.dumps(task.model_dump(by_alias=True, exclude_none=True))
+            )
+            _flush_pending_writes(pending_writes)
 
     return task
 
 
 def list_tasks(
     team_name: str, base_dir: Path | None = None
 ) -> list[TaskFile]:
+    if not team_exists(team_name, base_dir):
+        raise ValueError(f"Team {team_name!r} does not exist")
     team_dir = _tasks_dir(base_dir) / team_name
     tasks: list[TaskFile] = []
     for f in team_dir.glob("*.json"):
@@ -250,7 +329,8 @@ def reset_owner_tasks(
                 continue
             task = TaskFile(**json.loads(f.read_text()))
             if task.owner == agent_name:
-                task.status = "pending"
+                if task.status != "completed":
+                    task.status = "pending"
                 task.owner = None
                 f.write_text(
                     json.dumps(task.model_dump(by_alias=True, exclude_none=True))

tests/test_server.py

@@ -425,6 +425,13 @@ async def test_task_update_wraps_validation_error(self, client: Client):
         assert result.is_error is True
         assert "cannot transition" in result.content[0].text.lower()
 
+    async def test_task_list_wraps_nonexistent_team(self, client: Client):
+        result = await client.call_tool(
+            "task_list", {"team_name": "ghost-team"}, raise_on_error=False,
+        )
+        assert result.is_error is True
+        assert "does not exist" in result.content[0].text.lower()
+
 
 class TestPollInbox:
     async def test_should_return_empty_on_timeout(self, client: Client):

tests/test_tasks.py

@@ -339,3 +339,93 @@ def test_delete_task_cleans_up_stale_refs(tmp_claude_dir, team_tasks_dir):
     t2_after = get_task("test-team", t2.id, base_dir=tmp_claude_dir)
     assert t1.id not in t2_after.blocked_by
     assert t1.id not in t2_after.blocks
+
+
+def test_no_partial_write_when_status_validation_fails(tmp_claude_dir, team_tasks_dir):
+    blocker = create_task("test-team", "Blocker", "b", base_dir=tmp_claude_dir)
+    task = create_task("test-team", "Task", "t", base_dir=tmp_claude_dir)
+    blocker_before = get_task("test-team", blocker.id, base_dir=tmp_claude_dir)
+    task_before = get_task("test-team", task.id, base_dir=tmp_claude_dir)
+    with pytest.raises(ValueError, match="blocked by task"):
+        update_task(
+            "test-team", task.id,
+            add_blocked_by=[blocker.id], status="in_progress",
+            base_dir=tmp_claude_dir,
+        )
+    blocker_after = get_task("test-team", blocker.id, base_dir=tmp_claude_dir)
+    task_after = get_task("test-team", task.id, base_dir=tmp_claude_dir)
+    assert blocker_after.blocks == blocker_before.blocks
+    assert task_after.blocked_by == task_before.blocked_by
+
+
+def test_no_partial_write_on_add_blocks_with_failed_status(tmp_claude_dir, team_tasks_dir):
+    task = create_task("test-team", "Task", "t", base_dir=tmp_claude_dir)
+    other = create_task("test-team", "Other", "o", base_dir=tmp_claude_dir)
+    update_task("test-team", task.id, status="in_progress", base_dir=tmp_claude_dir)
+    with pytest.raises(ValueError, match="Cannot transition"):
+        update_task(
+            "test-team", task.id,
+            add_blocks=[other.id], status="pending",
+            base_dir=tmp_claude_dir,
+        )
+    other_after = get_task("test-team", other.id, base_dir=tmp_claude_dir)
+    task_after = get_task("test-team", task.id, base_dir=tmp_claude_dir)
+    assert task_after.blocks == []
+    assert other_after.blocked_by == []
+
+
+def test_rejects_simple_circular_dependency(tmp_claude_dir, team_tasks_dir):
+    a = create_task("test-team", "A", "d", base_dir=tmp_claude_dir)
+    b = create_task("test-team", "B", "d", base_dir=tmp_claude_dir)
+    update_task("test-team", a.id, add_blocked_by=[b.id], base_dir=tmp_claude_dir)
+    with pytest.raises(ValueError, match="circular dependency"):
+        update_task("test-team", b.id, add_blocked_by=[a.id], base_dir=tmp_claude_dir)
+
+
+def test_rejects_transitive_circular_dependency(tmp_claude_dir, team_tasks_dir):
+    a = create_task("test-team", "A", "d", base_dir=tmp_claude_dir)
+    b = create_task("test-team", "B", "d", base_dir=tmp_claude_dir)
+    c = create_task("test-team", "C", "d", base_dir=tmp_claude_dir)
+    update_task("test-team", a.id, add_blocked_by=[b.id], base_dir=tmp_claude_dir)
+    update_task("test-team", b.id, add_blocked_by=[c.id], base_dir=tmp_claude_dir)
+    with pytest.raises(ValueError, match="circular dependency"):
+        update_task("test-team", c.id, add_blocked_by=[a.id], base_dir=tmp_claude_dir)
+
+
+def test_rejects_circular_via_add_blocks(tmp_claude_dir, team_tasks_dir):
+    a = create_task("test-team", "A", "d", base_dir=tmp_claude_dir)
+    b = create_task("test-team", "B", "d", base_dir=tmp_claude_dir)
+    update_task("test-team", a.id, add_blocked_by=[b.id], base_dir=tmp_claude_dir)
+    with pytest.raises(ValueError, match="circular dependency"):
+        update_task("test-team", a.id, add_blocks=[b.id], base_dir=tmp_claude_dir)
+
+
+def test_allows_non_cyclic_diamond_dependency(tmp_claude_dir, team_tasks_dir):
+    a = create_task("test-team", "A", "d", base_dir=tmp_claude_dir)
+    b = create_task("test-team", "B", "d", base_dir=tmp_claude_dir)
+    c = create_task("test-team", "C", "d", base_dir=tmp_claude_dir)
+    d = create_task("test-team", "D", "d", base_dir=tmp_claude_dir)
+    update_task("test-team", d.id, add_blocked_by=[b.id, c.id], base_dir=tmp_claude_dir)
+    update_task("test-team", b.id, add_blocked_by=[a.id], base_dir=tmp_claude_dir)
+    update_task("test-team", c.id, add_blocked_by=[a.id], base_dir=tmp_claude_dir)
+    d_after = get_task("test-team", d.id, base_dir=tmp_claude_dir)
+    assert set(d_after.blocked_by) == {b.id, c.id}
+
+
+def test_list_tasks_rejects_nonexistent_team(tmp_claude_dir):
+    with pytest.raises(ValueError, match="does not exist"):
+        list_tasks("no-such-team", base_dir=tmp_claude_dir)
+
+
+def test_reset_owner_tasks_preserves_completed_status(tmp_claude_dir, team_tasks_dir):
+    task = create_task("test-team", "Sub", "desc", base_dir=tmp_claude_dir)
+    update_task(
+        "test-team", task.id,
+        owner="w", status="in_progress",
+        base_dir=tmp_claude_dir,
+    )
+    update_task("test-team", task.id, status="completed", base_dir=tmp_claude_dir)
+    reset_owner_tasks("test-team", "w", base_dir=tmp_claude_dir)
+    after = get_task("test-team", task.id, base_dir=tmp_claude_dir)
+    assert after.status == "completed"
+    assert after.owner is None