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(