fix: harden input validation, race-safe inbox reads, and error wrapping

Author: cs50victor

src/claude_teams/messaging.py

@@ -62,22 +62,32 @@ def read_inbox(
     if not path.exists():
         return []
 
-    raw_list = json.loads(path.read_text())
-    all_msgs = [InboxMessage.model_validate(entry) for entry in raw_list]
-
-    if unread_only:
-        result = [m for m in all_msgs if not m.read]
+    if mark_as_read:
+        lock_path = path.parent / ".lock"
+        with file_lock(lock_path):
+            raw_list = json.loads(path.read_text())
+            all_msgs = [InboxMessage.model_validate(entry) for entry in raw_list]
+
+            if unread_only:
+                result = [m for m in all_msgs if not m.read]
+            else:
+                result = list(all_msgs)
+
+            if result:
+                for m in all_msgs:
+                    if m in result:
+                        m.read = True
+                serialized = [m.model_dump(by_alias=True, exclude_none=True) for m in all_msgs]
+                path.write_text(json.dumps(serialized))
+
+            return result
     else:
-        result = list(all_msgs)
-
-    if mark_as_read and result:
-        for m in all_msgs:
-            if m in result:
-                m.read = True
-        serialized = [m.model_dump(by_alias=True, exclude_none=True) for m in all_msgs]
-        path.write_text(json.dumps(serialized))
+        raw_list = json.loads(path.read_text())
+        all_msgs = [InboxMessage.model_validate(entry) for entry in raw_list]
 
-    return result
+        if unread_only:
+            return [m for m in all_msgs if not m.read]
+        return list(all_msgs)
 
 
 def append_message(

src/claude_teams/server.py

@@ -61,7 +61,10 @@ def team_create(
 def team_delete(team_name: str, ctx: Context) -> dict:
     """Delete a team and all its data. Fails if any teammates are still active.
     Removes both team config and task directories."""
-    result = teams.delete_team(team_name)
+    try:
+        result = teams.delete_team(team_name)
+    except (RuntimeError, FileNotFoundError) as e:
+        raise ToolError(str(e))
     _get_lifespan(ctx)["active_team"] = None
     return result.model_dump()
 
@@ -213,6 +216,12 @@ def send_message(
             ).model_dump(exclude_none=True)
 
     elif type == "plan_approval_response":
+        if not recipient:
+            raise ToolError("Plan approval recipient must not be empty")
+        config = teams.read_config(team_name)
+        member_names = {m.name for m in config.members}
+        if recipient not in member_names:
+            raise ToolError(f"Recipient {recipient!r} is not a member of team {team_name!r}")
         if approve:
             messaging.send_plain_message(
                 team_name, sender, recipient,
@@ -354,12 +363,15 @@ async def poll_inbox(
     """Poll an agent's inbox for new unread messages, waiting up to timeout_ms.
     Returns unread messages and marks them as read. Convenience tool for MCP
     clients that cannot watch the filesystem."""
+    msgs = messaging.read_inbox(team_name, agent_name, unread_only=True, mark_as_read=True)
+    if msgs:
+        return [m.model_dump(by_alias=True, exclude_none=True) for m in msgs]
     deadline = time.time() + timeout_ms / 1000.0
     while time.time() < deadline:
+        await asyncio.sleep(0.5)
         msgs = messaging.read_inbox(team_name, agent_name, unread_only=True, mark_as_read=True)
         if msgs:
             return [m.model_dump(by_alias=True, exclude_none=True) for m in msgs]
-        await asyncio.sleep(0.5)
     return []
 
 

src/claude_teams/spawner.py

@@ -8,6 +8,7 @@
 
 from claude_teams import messaging, teams
 from claude_teams.models import COLOR_PALETTE, InboxMessage, TeammateMember
+from claude_teams.teams import _VALID_NAME_RE
 
 
 def discover_claude_binary() -> str:
@@ -62,6 +63,13 @@ def spawn_teammate(
     plan_mode_required: bool = False,
     base_dir: Path | None = None,
 ) -> TeammateMember:
+    if not _VALID_NAME_RE.match(name):
+        raise ValueError(f"Invalid agent name: {name!r}. Use only letters, numbers, hyphens, underscores.")
+    if len(name) > 64:
+        raise ValueError(f"Agent name too long ({len(name)} chars, max 64)")
+    if name == "team-lead":
+        raise ValueError("Agent name 'team-lead' is reserved")
+
     color = assign_color(team_name, base_dir)
     now_ms = int(time.time() * 1000)
 

src/claude_teams/teams.py

@@ -45,6 +45,8 @@ def create_team(
 ) -> TeamCreateResult:
     if not _VALID_NAME_RE.match(name):
         raise ValueError(f"Invalid team name: {name!r}. Use only letters, numbers, hyphens, underscores.")
+    if len(name) > 64:
+        raise ValueError(f"Team name too long ({len(name)} chars, max 64): {name[:20]!r}...")
 
     teams_dir = _teams_dir(base_dir)
     tasks_dir = _tasks_dir(base_dir)
@@ -134,6 +136,9 @@ def delete_team(name: str, base_dir: Path | None = None) -> TeamDeleteResult:
 
 def add_member(name: str, member: TeammateMember, base_dir: Path | None = None) -> None:
     config = read_config(name, base_dir=base_dir)
+    existing_names = {m.name for m in config.members}
+    if member.name in existing_names:
+        raise ValueError(f"Member {member.name!r} already exists in team {name!r}")
     config.members.append(member)
     write_config(name, config, base_dir=base_dir)
 

stress_test_lifecycle.py

@@ -0,0 +1,163 @@
+"""Stress test: Team Lifecycle Edge Cases.
+
+Calls the same functions the MCP server tools delegate to,
+exercising identical validation and I/O code paths.
+"""
+import json
+import tempfile
+import traceback
+from pathlib import Path
+
+from claude_teams import teams
+
+SESSION_ID = "stress-test-session-001"
+
+# Use an isolated temp directory so we don't touch ~/.claude
+tmp_base = Path(tempfile.mkdtemp(prefix="stress_test_"))
+(tmp_base / "teams").mkdir()
+(tmp_base / "tasks").mkdir()
+
+results = []
+
+def run_test(num, name, fn, expected):
+    actual = ""
+    passed = False
+    try:
+        ret = fn()
+        actual = repr(ret) if not isinstance(ret, str) else ret
+        # If we got here without exception, result is "success"
+        if "success" in expected.lower() or "created" in expected.lower() or "returns" in expected.lower() or "valid" in expected.lower():
+            passed = True
+        else:
+            passed = False
+    except Exception as e:
+        actual = f"{type(e).__name__}: {e}"
+        if "error" in expected.lower() or "fail" in expected.lower() or "reject" in expected.lower():
+            passed = True
+        else:
+            passed = False
+    results.append((num, name, expected, actual, "PASS" if passed else "FAIL"))
+    print(f"  {'PASS' if passed else 'FAIL'} | Test {num}: {name}")
+    print(f"       Expected: {expected}")
+    print(f"       Actual:   {actual}")
+    print()
+
+
+# ── Test 1: Create team with empty name ──
+run_test(1, "Create team with empty name",
+    lambda: teams.create_team(name="", session_id=SESSION_ID, base_dir=tmp_base),
+    "Error/reject: invalid team name")
+
+# ── Test 2: Create team with special characters ──
+run_test(2, "Create team with special characters",
+    lambda: teams.create_team(name="test!@#$%^&*()", session_id=SESSION_ID, base_dir=tmp_base),
+    "Error/reject: invalid team name")
+
+# ── Test 3: Create team with spaces ──
+run_test(3, "Create team with spaces",
+    lambda: teams.create_team(name="my team name", session_id=SESSION_ID, base_dir=tmp_base),
+    "Error/reject: invalid team name")
+
+# ── Test 4: Create team with very long name (500 chars) ──
+run_test(4, "Create team with very long name (500 chars)",
+    lambda: teams.create_team(name="a" * 500, session_id=SESSION_ID, base_dir=tmp_base),
+    "Success or error depending on filesystem limits")
+
+# ── Test 5: Create valid team ──
+run_test(5, "Create valid team stress-test-lifecycle-1",
+    lambda: teams.create_team(name="stress-test-lifecycle-1", session_id=SESSION_ID, base_dir=tmp_base),
+    "Success: team created")
+
+# ── Test 6: Create duplicate team ──
+run_test(6, "Create duplicate team stress-test-lifecycle-1",
+    lambda: teams.create_team(name="stress-test-lifecycle-1", session_id=SESSION_ID, base_dir=tmp_base),
+    "Success or silent overwrite (no uniqueness check in create_team)")
+
+# ── Test 7: Read config of non-existent team ──
+run_test(7, "Read config of non-existent team",
+    lambda: teams.read_config(name="nonexistent-team-xyz", base_dir=tmp_base),
+    "Error/fail: team not found or FileNotFoundError")
+
+# ── Test 8: Read config of valid team ──
+run_test(8, "Read config of valid team stress-test-lifecycle-1",
+    lambda: teams.read_config(name="stress-test-lifecycle-1", base_dir=tmp_base),
+    "Success: returns TeamConfig")
+
+# ── Test 9: Delete non-existent team ──
+run_test(9, "Delete non-existent team",
+    lambda: teams.delete_team(name="nonexistent-team-xyz", base_dir=tmp_base),
+    "Error/fail: team not found")
+
+# ── Test 10: Delete valid team ──
+run_test(10, "Delete valid team stress-test-lifecycle-1",
+    lambda: teams.delete_team(name="stress-test-lifecycle-1", base_dir=tmp_base),
+    "Success: team deleted")
+
+# ── Test 11: Double delete ──
+run_test(11, "Double delete stress-test-lifecycle-1",
+    lambda: teams.delete_team(name="stress-test-lifecycle-1", base_dir=tmp_base),
+    "Error/fail: team already deleted")
+
+# ── Test 12: Create team with unicode ──
+run_test(12, "Create team with unicode emoji",
+    lambda: teams.create_team(name="test-unicode-\U0001f680", session_id=SESSION_ID, base_dir=tmp_base),
+    "Error/reject: invalid team name (regex rejects unicode)")
+
+# ── Test 13: Create team with dots ──
+run_test(13, "Create team with dots",
+    lambda: teams.create_team(name="test.dotted.name", session_id=SESSION_ID, base_dir=tmp_base),
+    "Error/reject: dots not in allowed charset")
+
+# ── Test 14: Create team with leading hyphen ──
+run_test(14, "Create team with leading hyphen",
+    lambda: teams.create_team(name="-leading-hyphen", session_id=SESSION_ID, base_dir=tmp_base),
+    "Success: hyphens are allowed by regex ^[A-Za-z0-9_-]+$")
+
+# ── Test 15: Create team with only numbers ──
+run_test(15, "Create team with only numbers",
+    lambda: teams.create_team(name="12345", session_id=SESSION_ID, base_dir=tmp_base),
+    "Success: digits are allowed by regex")
+
+
+# ── Cleanup ──
+print("=" * 60)
+print("CLEANUP: removing teams created during tests")
+cleanup_teams = []
+# Test 4 may have created a long-name team
+long_name = "a" * 500
+for tname in [long_name, "-leading-hyphen", "12345"]:
+    try:
+        teams.delete_team(name=tname, base_dir=tmp_base)
+        cleanup_teams.append(tname[:40])
+    except Exception:
+        pass
+if cleanup_teams:
+    print(f"  Deleted: {cleanup_teams}")
+else:
+    print("  Nothing to clean up.")
+
+import shutil
+shutil.rmtree(tmp_base, ignore_errors=True)
+print(f"  Removed temp dir: {tmp_base}")
+
+# ── Summary Table ──
+print()
+print("=" * 120)
+print(f"| {'#':>2} | {'Test':<50} | {'Expected':<55} | {'Pass/Fail':<9} |")
+print(f"|{'-'*4}|{'-'*52}|{'-'*57}|{'-'*11}|")
+for num, tname, expected, actual, verdict in results:
+    print(f"| {num:>2} | {tname:<50} | {expected:<55} | {verdict:<9} |")
+print("=" * 120)
+
+total = len(results)
+passed = sum(1 for r in results if r[4] == "PASS")
+print(f"\nTotal: {total}  Passed: {passed}  Failed: {total - passed}")
+
+# ── Detailed Actual Results ──
+print()
+print("DETAILED ACTUAL RESULTS:")
+print("-" * 120)
+for num, tname, expected, actual, verdict in results:
+    # Truncate long actual results for readability
+    actual_display = actual if len(actual) < 200 else actual[:200] + "..."
+    print(f"  Test {num:>2} [{verdict}]: {actual_display}")

tests/test_messaging.py

@@ -2,6 +2,7 @@
 
 import json
 import re
+import threading
 from pathlib import Path
 
 import pytest
@@ -163,6 +164,36 @@ def test_send_shutdown_request_with_reason(tmp_claude_dir):
     assert parsed["reason"] == "Done"
 
 
+def test_should_not_lose_message_appended_during_mark_as_read(tmp_claude_dir):
+    import fcntl
+
+    msg_a = InboxMessage(from_="lead", text="A", timestamp=now_iso(), read=False, summary="a")
+    append_message("test-team", "race", msg_a, base_dir=tmp_claude_dir)
+
+    path = inbox_path("test-team", "race", base_dir=tmp_claude_dir)
+    lock_path = path.parent / ".lock"
+    lock_path.touch(exist_ok=True)
+
+    completed = threading.Event()
+
+    def do_read():
+        read_inbox("test-team", "race", mark_as_read=True, base_dir=tmp_claude_dir)
+        completed.set()
+
+    with open(lock_path) as f:
+        fcntl.flock(f.fileno(), fcntl.LOCK_EX)
+        reader = threading.Thread(target=do_read)
+        reader.start()
+        completed_without_lock = completed.wait(timeout=1.0)
+        fcntl.flock(f.fileno(), fcntl.LOCK_UN)
+
+    reader.join(timeout=5)
+
+    assert not completed_without_lock, (
+        "read_inbox(mark_as_read=True) completed without acquiring the inbox lock"
+    )
+
+
 def test_now_iso_format():
     ts = now_iso()
     assert re.match(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$", ts)