"""
Tests for File Coordination Governance system in amicus-mcp.
These tests validate the file intent coordination workflow including:
- Declaring file intents
- Checking for file conflicts
- Releasing file intents
- Listing file intents
- Auto-expiration of stale intents
- Integration with read_state
GitHub Issue #7: Agents must coordinate files they intend to change.
"""
import pytest
import time
from pathlib import Path
from unittest.mock import patch
from amicus.server import (
declare_file_intent as _declare_file_intent,
check_file_conflicts as _check_file_conflicts,
release_file_intent as _release_file_intent,
get_file_intents as _get_file_intents,
read_state as _read_state,
FILE_INTENT_TTL_SECONDS
)
from amicus.core import read_with_lock, write_with_lock, get_state_file
# Access the underlying functions from FunctionTool wrappers
declare_file_intent = _declare_file_intent.fn
check_file_conflicts = _check_file_conflicts.fn
release_file_intent = _release_file_intent.fn
get_file_intents = _get_file_intents.fn
read_state = _read_state.fn
class TestDeclareFileIntent:
"""Test suite for declaring file intents."""
def test_declare_intent_creates_intent_with_correct_structure(self, temp_context_dir):
"""Test that declare_file_intent creates an intent with correct structure."""
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
initial_state = {
"summary": "Test state",
"next_steps": [],
"active_files": [],
"timestamp": time.time()
}
write_with_lock(state_file, initial_state)
result = declare_file_intent(
agent_id="Agent-001",
files=["src/main.py", "src/utils.py"],
intent_type="write",
description="Refactoring utilities"
)
assert "File intent declared" in result
assert "Agent-001" in result
assert "src/main.py" in result
assert "WRITE" in result
# Verify state structure
state = read_with_lock(state_file)
assert "file_intents" in state
assert len(state["file_intents"]) == 1
intent = list(state["file_intents"].values())[0]
assert intent["agent_id"] == "Agent-001"
assert intent["files"] == ["src/main.py", "src/utils.py"]
assert intent["intent_type"] == "write"
assert intent["description"] == "Refactoring utilities"
assert "expires_at" in intent
assert "created_at" in intent
def test_declare_intent_detects_conflicts_with_existing_write_intent(self, temp_context_dir):
"""Test that declaring intent detects conflicts with existing write intents."""
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
# First agent declares intent
declare_file_intent(
agent_id="Agent-001",
files=["src/main.py"],
intent_type="write"
)
# Second agent declares overlapping intent
result = declare_file_intent(
agent_id="Agent-002",
files=["src/main.py", "src/other.py"],
intent_type="write"
)
assert "CONFLICTS DETECTED" in result
assert "src/main.py" in result
assert "Agent-001" in result
def test_declare_read_intent_no_conflict_with_read(self, temp_context_dir):
"""Test that read intents don't conflict with other read intents."""
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
# First agent declares read intent
declare_file_intent(
agent_id="Agent-001",
files=["src/main.py"],
intent_type="read"
)
# Second agent declares read intent on same file
result = declare_file_intent(
agent_id="Agent-002",
files=["src/main.py"],
intent_type="read"
)
# No conflict for read-read
assert "CONFLICTS" not in result
def test_declare_intent_validates_path_safety(self, temp_context_dir):
"""Test that declare_file_intent rejects paths outside project root."""
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
write_with_lock(state_file, {"summary": "Test", "next_steps": [], "active_files": []})
result = declare_file_intent(
agent_id="Agent-001",
files=["/etc/passwd", "src/main.py"],
intent_type="write"
)
assert "ERROR" in result
assert "outside the project root" in result
def test_declare_intent_with_custom_ttl(self, temp_context_dir):
"""Test that custom TTL is respected."""
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
write_with_lock(state_file, {"summary": "Test", "next_steps": [], "active_files": []})
custom_ttl = 60 # 1 minute
result = declare_file_intent(
agent_id="Agent-001",
files=["src/main.py"],
ttl_seconds=custom_ttl
)
assert f"{custom_ttl}s" in result
state = read_with_lock(state_file)
intent = list(state["file_intents"].values())[0]
expected_ttl = intent["expires_at"] - intent["created_at"]
assert abs(expected_ttl - custom_ttl) < 1 # Allow 1 second tolerance
class TestCheckFileConflicts:
"""Test suite for checking file conflicts."""
def test_check_conflicts_returns_clear_when_no_intents(self, temp_context_dir):
"""Test that check returns clear when no intents exist."""
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
write_with_lock(state_file, {"summary": "Test", "next_steps": [], "active_files": []})
result = check_file_conflicts(files=["src/main.py"])
assert "No conflicts" in result
def test_check_conflicts_detects_write_intent(self, temp_context_dir):
"""Test that check detects existing write intent."""
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
# Create existing intent
declare_file_intent(
agent_id="Agent-001",
files=["src/main.py"],
intent_type="write"
)
# Check for conflicts
result = check_file_conflicts(files=["src/main.py"])
assert "CONFLICTS DETECTED" in result
assert "Agent-001" in result
assert "src/main.py" in result
def test_check_conflicts_excludes_own_intents(self, temp_context_dir):
"""Test that agent's own intents are excluded from conflict check."""
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
# Create own intent
declare_file_intent(
agent_id="Agent-001",
files=["src/main.py"],
intent_type="write"
)
# Check for conflicts as same agent
result = check_file_conflicts(
files=["src/main.py"],
agent_id="Agent-001"
)
assert "No conflicts" in result
def test_check_conflicts_shows_ttl_remaining(self, temp_context_dir):
"""Test that conflict check shows TTL remaining."""
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
declare_file_intent(
agent_id="Agent-001",
files=["src/main.py"],
ttl_seconds=120
)
result = check_file_conflicts(files=["src/main.py"])
assert "Expires in:" in result
class TestReleaseFileIntent:
"""Test suite for releasing file intents."""
def test_release_intent_by_id(self, temp_context_dir):
"""Test releasing intent by specific ID."""
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
# Create intent
result = declare_file_intent(
agent_id="Agent-001",
files=["src/main.py"]
)
# Extract intent ID from result
state = read_with_lock(state_file)
intent_id = list(state["file_intents"].keys())[0]
# Release by ID
result = release_file_intent(
agent_id="Agent-001",
intent_id=intent_id
)
assert "Released" in result
assert intent_id in result
# Verify removed from state
state = read_with_lock(state_file)
assert len(state["file_intents"]) == 0
def test_release_intent_by_files(self, temp_context_dir):
"""Test releasing intent by file paths."""
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
declare_file_intent(
agent_id="Agent-001",
files=["src/main.py", "src/utils.py"]
)
result = release_file_intent(
agent_id="Agent-001",
files=["src/main.py"]
)
assert "Released" in result
def test_release_intent_rejects_other_agent(self, temp_context_dir):
"""Test that agent cannot release another agent's intent."""
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
declare_file_intent(
agent_id="Agent-001",
files=["src/main.py"]
)
state = read_with_lock(state_file)
intent_id = list(state["file_intents"].keys())[0]
result = release_file_intent(
agent_id="Agent-002",
intent_id=intent_id
)
assert "ERROR" in result
assert "belongs to Agent-001" in result
def test_release_requires_id_or_files(self, temp_context_dir):
"""Test that release requires either intent_id or files."""
result = release_file_intent(agent_id="Agent-001")
assert "ERROR" in result
assert "Must specify" in result
class TestGetFileIntents:
"""Test suite for listing file intents."""
def test_get_intents_returns_all_active(self, temp_context_dir):
"""Test getting all active intents."""
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
declare_file_intent(agent_id="Agent-001", files=["src/main.py"])
declare_file_intent(agent_id="Agent-002", files=["src/utils.py"])
result = get_file_intents()
assert "Agent-001" in result
assert "Agent-002" in result
assert "src/main.py" in result
assert "src/utils.py" in result
assert "Total active: 2" in result
def test_get_intents_filters_by_agent(self, temp_context_dir):
"""Test filtering intents by agent ID."""
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
declare_file_intent(agent_id="Agent-001", files=["src/main.py"])
declare_file_intent(agent_id="Agent-002", files=["src/utils.py"])
result = get_file_intents(agent_id="Agent-001")
assert "Agent-001" in result
assert "Agent-002" not in result
assert "Total active: 1" in result
def test_get_intents_filters_by_file(self, temp_context_dir):
"""Test filtering intents by file path."""
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
declare_file_intent(agent_id="Agent-001", files=["src/main.py"])
declare_file_intent(agent_id="Agent-002", files=["src/utils.py"])
result = get_file_intents(file_path="src/main.py")
assert "Agent-001" in result
assert "Agent-002" not in result
class TestIntentExpiration:
"""Test suite for intent TTL and expiration."""
def test_expired_intents_are_cleaned_up(self, temp_context_dir):
"""Test that expired intents are removed during operations."""
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
# Create intent with very short TTL
declare_file_intent(
agent_id="Agent-001",
files=["src/main.py"],
ttl_seconds=1
)
# Wait for expiration
time.sleep(1.5)
# Any operation should clean up expired intents
result = get_file_intents()
assert "No active file intents" in result or "Cleaned up" in result
def test_expired_intents_dont_cause_conflicts(self, temp_context_dir):
"""Test that expired intents don't block new declarations."""
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
# Create intent with very short TTL
declare_file_intent(
agent_id="Agent-001",
files=["src/main.py"],
ttl_seconds=1
)
# Wait for expiration
time.sleep(1.5)
# New declaration should not see conflict
result = declare_file_intent(
agent_id="Agent-002",
files=["src/main.py"]
)
assert "CONFLICTS" not in result
class TestReadStateIntegration:
"""Test suite for read_state integration with file intents."""
def test_read_state_shows_file_intents_summary(self, temp_context_dir):
"""Test that read_state displays file intents."""
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
declare_file_intent(
agent_id="Agent-001",
files=["src/main.py"],
intent_type="write"
)
result = read_state()
assert "File Intents" in result
assert "Agent-001" in result
assert "src/main.py" in result
def test_read_state_shows_conflicts_warning(self, temp_context_dir):
"""Test that read_state shows conflict warnings."""
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
# Create conflicting intents
declare_file_intent(agent_id="Agent-001", files=["src/main.py"])
declare_file_intent(agent_id="Agent-002", files=["src/main.py"])
result = read_state()
assert "FILE CONFLICTS" in result
assert "src/main.py" in result
class TestMultiAgentCoordination:
"""Integration tests for multi-agent file coordination scenarios."""
def test_typical_workflow_declare_work_release(self, temp_context_dir):
"""Test typical workflow: declare intent, do work, release."""
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
# Agent 1 declares intent
declare_file_intent(
agent_id="Agent-001",
files=["src/main.py"],
description="Adding new feature"
)
# Agent 2 checks before working
check_result = check_file_conflicts(
files=["src/main.py"],
agent_id="Agent-002"
)
assert "CONFLICTS" in check_result
# Agent 1 completes work and releases
release_file_intent(
agent_id="Agent-001",
files=["src/main.py"]
)
# Agent 2 can now proceed
check_result = check_file_conflicts(
files=["src/main.py"],
agent_id="Agent-002"
)
assert "No conflicts" in check_result
def test_parallel_work_on_different_files(self, temp_context_dir):
"""Test that agents can work on different files without conflict."""
state_file = get_state_file()
state_file.parent.mkdir(parents=True, exist_ok=True)
# Agent 1 works on main.py
result1 = declare_file_intent(
agent_id="Agent-001",
files=["src/main.py"]
)
assert "CONFLICTS" not in result1
# Agent 2 works on utils.py
result2 = declare_file_intent(
agent_id="Agent-002",
files=["src/utils.py"]
)
assert "CONFLICTS" not in result2
# Both should have clear paths
check1 = check_file_conflicts(["src/main.py"], "Agent-001")
check2 = check_file_conflicts(["src/utils.py"], "Agent-002")
assert "No conflicts" in check1
assert "No conflicts" in check2
