diff --git a/mini_coding_agent.py b/mini_coding_agent.py index 1bf2403..b815b1d 100644 --- a/mini_coding_agent.py +++ b/mini_coding_agent.py @@ -238,6 +238,7 @@ def __init__( ): self.model_client = model_client self.workspace = workspace + self.workspace_dirty = False self.root = Path(workspace.repo_root) self.session_store = session_store self.approval_policy = approval_policy @@ -370,7 +371,6 @@ def build_prefix(self): "Rules:\n" + rules, "Tools:\n" + tool_text, "Valid response examples:\n" + examples, - self.workspace.text(), ]) def memory_text(self): @@ -420,8 +420,12 @@ def history_text(self): #### 2) Prompt Shape And Cache Reuse (Continued) ####### ######################################################## def prompt(self, user_message): + if self.workspace_dirty: + self.workspace = WorkspaceContext.build(self.workspace.cwd) + self.workspace_dirty = False return "\n\n".join([ self.prefix, + self.workspace.text(), self.memory_text(), "Transcript:\n" + self.history_text(), "Current user request:\n" + user_message, @@ -513,6 +517,10 @@ def run_tool(self, name, args): return clip(tool["run"](args)) except Exception as exc: return f"error: tool {name} failed: {exc}" + finally: + # A tool can change files before failing (for example, a shell timeout). + if name in {"write_file", "patch_file", "run_shell"}: + self.workspace_dirty = True def repeated_tool_call(self, name, args): tool_events = [item for item in self.session["history"] if item["role"] == "tool"] diff --git a/tests/test_mini_coding_agent.py b/tests/test_mini_coding_agent.py index 5c62693..b938d1b 100644 --- a/tests/test_mini_coding_agent.py +++ b/tests/test_mini_coding_agent.py @@ -1,4 +1,5 @@ import json +import subprocess import pytest from unittest.mock import patch @@ -30,6 +31,94 @@ def build_agent(tmp_path, outputs, **kwargs): ) +@pytest.fixture +def git_workspace(tmp_path): + (tmp_path / "README.md").write_text("demo\n", encoding="utf-8") + (tmp_path / ".gitignore").write_text(".mini-coding-agent/\n", encoding="utf-8") + for args in ( + ["init"], + ["add", "."], + ["-c", "user.name=Test", "-c", "user.email=test@example.invalid", + "-c", "commit.gpgsign=false", "commit", "-m", "initial"], + ): + subprocess.run(["git", *args], cwd=tmp_path, check=True, capture_output=True) + return tmp_path + + +def workspace_in_prompt(prompt): + return prompt.split("Workspace:\n", 1)[1].split("\n\nMemory:", 1)[0] + + +@pytest.mark.parametrize("name,args", [ + ("write_file", {"path": "README.md", "content": "updated\n"}), + ("patch_file", {"path": "README.md", "old_text": "demo", "new_text": "updated"}), + ("run_shell", {"command": "echo updated>README.md"}), +]) +def test_workspace_refresh_after_mutating_tool(git_workspace, name, args): + agent = build_agent(git_workspace, [ + "" + json.dumps({"name": name, "args": args}) + "", + "Done.", + ]) + prefix = agent.prefix + with patch.object(agent.model_client, "complete", wraps=agent.model_client.complete) as complete: + assert agent.ask("Update the documentation") == "Done." + + before, after = [workspace_in_prompt(call.args[0]) for call in complete.call_args_list] + assert "- status:\nclean\n" in before + assert "- README.md\ndemo\n" in before + assert "M README.md" in after + assert "- README.md\nupdated" in after + assert "- README.md\ndemo\n" not in after + assert agent.prefix == prefix + + +def test_workspace_refresh_after_shell_changes_git_state(git_workspace): + command = ( + 'git checkout -b refreshed-branch && git -c user.name=Test ' + '-c user.email=test@example.invalid -c commit.gpgsign=false ' + 'commit --allow-empty -m refreshed-commit' + ) + agent = build_agent(git_workspace, [ + "" + json.dumps({"name": "run_shell", "args": {"command": command}}) + "", + "Done.", + ]) + with patch.object(agent.model_client, "complete", wraps=agent.model_client.complete) as complete: + agent.ask("Update the git state") + after = workspace_in_prompt(complete.call_args_list[-1].args[0]) + assert "- branch: refreshed-branch\n" in after + assert "refreshed-commit" in after + + +def test_workspace_refresh_after_partial_tool_failure(tmp_path): + agent = build_agent(tmp_path, []) + + def partial_write(args): + (tmp_path / "README.md").write_text("partial update\n", encoding="utf-8") + raise RuntimeError("synthetic failure after mutation") + + agent.tools["write_file"]["run"] = partial_write + result = agent.run_tool("write_file", {"path": "README.md", "content": "unused"}) + assert "synthetic failure after mutation" in result + assert "- README.md\npartial update\n" in workspace_in_prompt(agent.prompt("Continue")) + + +@pytest.mark.parametrize("name,args,policy", [ + ("read_file", {"path": "README.md"}, "auto"), + ("write_file", {"path": "README.md", "content": "denied"}, "never"), + ("write_file", {"path": "README.md"}, "auto"), +]) +def test_workspace_not_refreshed_without_executed_mutation(tmp_path, name, args, policy): + agent = build_agent(tmp_path, [], approval_policy=policy) + before = workspace_in_prompt(agent.prompt("Inspect")) + with patch.object(WorkspaceContext, "build", wraps=WorkspaceContext.build) as build: + result = agent.run_tool(name, args) + assert workspace_in_prompt(agent.prompt("Continue")) == before + build.assert_not_called() + if policy == "never": + assert "approval denied" in result + assert (tmp_path / "README.md").read_text(encoding="utf-8") == "demo\n" + + def test_agent_runs_tool_then_final(tmp_path): (tmp_path / "hello.txt").write_text("alpha\nbeta\n", encoding="utf-8") agent = build_agent(