Add locked shell environment + centralized shell dialect guard for Bash tool

tests/test_bounded_execution.py

@@ -110,3 +110,89 @@ def test_villani_task_reports_completed_when_done(tmp_path: Path) -> None:
 
     assert result["execution"]["completed"] is True
     assert result["execution"]["terminated_reason"] == "completed"
+
+
+def test_active_solution_failed_validation_blocks_completed_status(tmp_path: Path, monkeypatch) -> None:
+    runner = _runner(
+        tmp_path,
+        [
+            {"role": "assistant", "content": [{"type": "tool_use", "id": "1", "name": "Bash", "input": {"command": "python web_server.py"}}]},
+            {"role": "assistant", "content": [{"type": "text", "text": "all done"}]},
+        ],
+    )
+    def fake_verify(*_args, **_kwargs) -> str:
+        runner._active_solution_file = "web_server.py"
+        runner._active_solution_last_validation_ok = False
+        return ""
+    monkeypatch.setattr(runner, "_run_verification", fake_verify)
+
+    result = runner.run("work")
+
+    assert result["execution"]["completed"] is False
+    assert result["execution"]["terminated_reason"] == "active_solution_validation_failed"
+
+
+def test_active_solution_successful_validation_allows_completed_status(tmp_path: Path, monkeypatch) -> None:
+    runner = _runner(
+        tmp_path,
+        [
+            {"role": "assistant", "content": [{"type": "tool_use", "id": "1", "name": "Bash", "input": {"command": "python web_server.py"}}]},
+            {"role": "assistant", "content": [{"type": "text", "text": "all done"}]},
+        ],
+    )
+    def fake_verify(*_args, **_kwargs) -> str:
+        runner._active_solution_file = "web_server.py"
+        runner._active_solution_last_validation_ok = True
+        return ""
+    monkeypatch.setattr(runner, "_run_verification", fake_verify)
+
+    result = runner.run("work")
+
+    assert result["execution"]["completed"] is True
+    assert result["execution"]["terminated_reason"] == "completed"
+
+
+def test_final_summary_is_honest_when_primary_validation_missing(tmp_path: Path, monkeypatch) -> None:
+    runner = _runner(
+        tmp_path,
+        [
+            {"role": "assistant", "content": [{"type": "tool_use", "id": "1", "name": "Bash", "input": {"command": "python web_server.py"}}]},
+            {"role": "assistant", "content": [{"type": "text", "text": "Fully functional and thoroughly tested."}]},
+        ],
+    )
+    def fake_verify(*_args, **_kwargs) -> str:
+        runner._primary_execution_target = "web_server.py"
+        runner._primary_target_minimally_valid = False
+        runner._active_solution_file = "web_server.py"
+        runner._active_solution_last_validation_ok = False
+        return ""
+    monkeypatch.setattr(runner, "_run_verification", fake_verify)
+    monkeypatch.setattr(runner, "_run_post_execution_validation", lambda _changed: "")
+
+    result = runner.run("work")
+    final_text = result["execution"]["final_text"]
+    assert "Incomplete: primary execution target does not yet have clean direct validation" in final_text
+    assert "Fully functional and thoroughly tested." in final_text
+
+
+def test_final_summary_can_remain_positive_when_primary_validation_is_strong(tmp_path: Path, monkeypatch) -> None:
+    runner = _runner(
+        tmp_path,
+        [
+            {"role": "assistant", "content": [{"type": "tool_use", "id": "1", "name": "Bash", "input": {"command": "python web_server.py"}}]},
+            {"role": "assistant", "content": [{"type": "text", "text": "Fully functional and thoroughly tested."}]},
+        ],
+    )
+    def fake_verify(*_args, **_kwargs) -> str:
+        runner._primary_execution_target = "web_server.py"
+        runner._primary_target_minimally_valid = True
+        runner._active_solution_file = "web_server.py"
+        runner._active_solution_last_validation_ok = True
+        return ""
+    monkeypatch.setattr(runner, "_run_verification", fake_verify)
+    monkeypatch.setattr(runner, "_run_post_execution_validation", lambda _changed: "")
+
+    result = runner.run("work")
+    final_text = result["execution"]["final_text"]
+    assert "Incomplete: primary execution target does not yet have clean direct validation" not in final_text
+    assert "Fully functional and thoroughly tested." in final_text

tests/test_hardening_pass.py

@@ -5,12 +5,14 @@
 
 
 from villani_code.planning import PlanRiskLevel, generate_execution_plan
-from villani_code.project_memory import SessionState, init_project_memory, load_repo_map, load_validation_config
+from villani_code.project_memory import SessionState, ValidationStep, init_project_memory, load_repo_map, load_validation_config
 from villani_code.state import Runner
 from villani_code import state_runtime
 from villani_code.validation_loop import (
+    classify_validation_step_strength,
     infer_targeted_command,
     plan_validation,
+    select_validation_steps,
     summarize_validation_failure,
 )
 
@@ -116,6 +118,34 @@ def test_cost_aware_ordering_and_failure_summary(tmp_path: Path) -> None:
     assert len(failure.compact_output) < 1500
 
 
+def test_validation_step_strength_prefers_direct_over_helper() -> None:
+    direct = ValidationStep(
+        name="direct",
+        command="python service.py",
+        kind="test",
+        cost_level=2,
+        is_mutating=False,
+    )
+    helper = ValidationStep(
+        name="helper",
+        command="python helper_wrapper.py",
+        kind="inspection",
+        cost_level=2,
+        is_mutating=False,
+    )
+    assert classify_validation_step_strength(direct) == "direct_run_or_task_validation"
+    assert classify_validation_step_strength(helper) == "helper_wrapper"
+
+
+def test_select_validation_steps_keeps_stronger_steps_available(tmp_path: Path) -> None:
+    _seed_repo(tmp_path)
+    init_project_memory(tmp_path)
+    cfg = load_validation_config(tmp_path)
+    selected = select_validation_steps(cfg, ["villani_code/mod.py"])
+    kinds = [step.kind for step in selected]
+    assert "test" in kinds
+
+
 def test_dedicated_repair_executor_is_bounded(tmp_path: Path) -> None:
     _seed_repo(tmp_path)
     init_project_memory(tmp_path)

tests/test_mission_state_runtime.py

@@ -34,13 +34,18 @@ def test_mission_state_roundtrip(tmp_path: Path) -> None:
         mode="execution",
         repo_root=str(tmp_path),
         status="active",
+        recovery_mode=True,
+        primary_execution_target="app.py",
+        primary_target_minimally_valid=False,
         verified_facts=[VerifiedFact(kind="k", value="v", source="s")],
         open_hypotheses=[OpenHypothesis(hypothesis_id="h1", statement="maybe", confidence=0.4, status="open")],
     )
     save_mission_state(tmp_path, state)
     loaded = load_mission_state(tmp_path, "m1")
     assert loaded.verified_facts[0].value == "v"
     assert loaded.open_hypotheses[0].hypothesis_id == "h1"
+    assert loaded.recovery_mode is True
+    assert loaded.primary_execution_target == "app.py"
 
 
 def test_mission_directory_and_current_pointer(tmp_path: Path) -> None:

tests/test_shell_command_guard.py

@@ -0,0 +1,152 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from villani_code.shells import classify_and_rewrite_command, detect_shell_environment
+from villani_code.tools import execute_tool
+
+
+def test_env_detection_returns_shell_family(tmp_path: Path) -> None:
+    env = detect_shell_environment(str(tmp_path))
+    assert env.shell_family in {"cmd", "powershell", "bash", "zsh", "unknown"}
+
+
+def test_cmd_rm_rewrites_to_del(tmp_path: Path) -> None:
+    decision = classify_and_rewrite_command("rm foo.txt", "cmd")
+    assert decision.classification == "needs_rewrite"
+    assert decision.command == "del /q foo.txt"
+
+
+def test_cmd_grep_rewrites_to_findstr(tmp_path: Path) -> None:
+    decision = classify_and_rewrite_command("grep hello notes.txt", "cmd")
+    assert decision.classification == "needs_rewrite"
+    assert decision.command == 'findstr /n /c:"hello" notes.txt'
+
+
+def test_cmd_tail_rewrites_to_powershell_get_content(tmp_path: Path) -> None:
+    decision = classify_and_rewrite_command("tail -n 20 app.log", "cmd")
+    assert decision.classification == "needs_rewrite"
+    assert decision.command == 'powershell -NoProfile -Command "Get-Content \'app.log\' -Tail 20"'
+
+
+def test_cmd_heredoc_blocked(tmp_path: Path) -> None:
+    decision = classify_and_rewrite_command("cat <<EOF\nhello\nEOF", "cmd")
+    assert decision.classification == "blocked"
+    assert decision.offending_pattern == "<<EOF"
+
+
+def test_cmd_embedded_head_pipeline_is_blocked(tmp_path: Path) -> None:
+    decision = classify_and_rewrite_command("echo ok && type app.log | head -50", "cmd")
+    assert decision.classification == "blocked"
+    assert decision.offending_token == "head"
+
+
+def test_cmd_head_n_rewrites_to_powershell_totalcount(tmp_path: Path) -> None:
+    decision = classify_and_rewrite_command("head -n 20 app.log", "cmd")
+    assert decision.classification == "needs_rewrite"
+    assert decision.command == 'powershell -NoProfile -Command "Get-Content \'app.log\' -TotalCount 20"'
+
+
+def test_cmd_head_default_rewrites_to_powershell_totalcount_10(tmp_path: Path) -> None:
+    decision = classify_and_rewrite_command("head app.log", "cmd")
+    assert decision.classification == "needs_rewrite"
+    assert decision.command == 'powershell -NoProfile -Command "Get-Content \'app.log\' -TotalCount 10"'
+
+
+def test_cmd_heredoc_anywhere_is_blocked(tmp_path: Path) -> None:
+    decision = classify_and_rewrite_command("echo start && python - <<'EOF'\nprint(1)\nEOF", "cmd")
+    assert decision.classification == "blocked"
+    assert decision.offending_token == "<<"
+
+
+def test_cmd_classifier_scans_full_command_string(tmp_path: Path) -> None:
+    decision = classify_and_rewrite_command("echo ok && dir | head -30", "cmd")
+    assert decision.classification == "blocked"
+
+
+def test_cmd_embedded_tail_or_wc_fragments_are_blocked(tmp_path: Path) -> None:
+    tail_decision = classify_and_rewrite_command("echo ok && type app.log | tail -20", "cmd")
+    wc_decision = classify_and_rewrite_command("echo ok && type app.log | wc -l", "cmd")
+    assert tail_decision.classification == "blocked"
+    assert wc_decision.classification == "blocked"
+
+
+def test_powershell_grep_rewrites_to_select_string(tmp_path: Path) -> None:
+    decision = classify_and_rewrite_command("grep hello notes.txt", "powershell")
+    assert decision.classification == "needs_rewrite"
+    assert decision.command == 'Select-String -Pattern "hello" notes.txt'
+
+
+def test_bash_native_commands_are_unchanged(tmp_path: Path) -> None:
+    decision = classify_and_rewrite_command("tail -n 2 app.log", "bash")
+    assert decision.classification == "allowed"
+    assert decision.command == "tail -n 2 app.log"
+
+
+def test_blocked_commands_produce_compact_structured_feedback(tmp_path: Path) -> None:
+    result = execute_tool(
+        "Bash",
+        {"command": "cat <<EOF\nhello\nEOF", "cwd": ".", "timeout_sec": 5},
+        tmp_path,
+        runtime_state={"shell_environment": {"shell_family": "cmd"}},
+    )
+    assert result["is_error"] is True
+    payload = json.loads(result["content"])
+    assert payload["shell_family"] == "cmd"
+    assert payload["classification"] == "blocked"
+    assert payload["offending_token"] == "<<"
+    assert payload["offending_pattern"] == "<<EOF"
+    assert "short_reason" in payload
+
+
+def test_cmd_invalid_detached_launch_pattern_is_blocked() -> None:
+    decision = classify_and_rewrite_command("python app.py > out.txt 2>&1 & echo STARTED", "cmd")
+    assert decision.classification == "blocked"
+    assert decision.offending_pattern
+    assert "detached" in decision.short_reason
+
+
+def test_cmd_ampersand_detached_assumption_is_blocked() -> None:
+    decision = classify_and_rewrite_command("python app.py & dir", "cmd")
+    assert decision.classification == "blocked"
+
+
+def test_cmd_invalid_detached_launch_feedback_is_compact_structured(tmp_path: Path) -> None:
+    result = execute_tool(
+        "Bash",
+        {"command": "python app.py > out.txt 2>&1 & echo STARTED", "cwd": ".", "timeout_sec": 5},
+        tmp_path,
+        runtime_state={"shell_environment": {"shell_family": "cmd"}},
+    )
+    assert result["is_error"] is True
+    payload = json.loads(result["content"])
+    assert payload["shell_family"] == "cmd"
+    assert payload["classification"] == "blocked"
+    assert payload["short_reason"] == "detached/background launch syntax is not safe in cmd"
+    assert payload["offending_pattern"]
+    assert set(payload) == {"shell_family", "classification", "short_reason", "offending_pattern"}
+
+
+def test_cmd_ordinary_command_unaffected() -> None:
+    decision = classify_and_rewrite_command("echo hello", "cmd")
+    assert decision.classification == "allowed"
+    assert decision.command == "echo hello"
+
+
+def test_cmd_blocked_detached_launch_does_not_execute(tmp_path: Path, monkeypatch) -> None:
+    called = {"value": False}
+
+    def _fail_if_called(*_args, **_kwargs):
+        called["value"] = True
+        raise AssertionError("subprocess.run should not be called for blocked detached cmd launch")
+
+    monkeypatch.setattr("villani_code.tools.subprocess.run", _fail_if_called)
+    result = execute_tool(
+        "Bash",
+        {"command": "python app.py > out.txt 2>&1 & echo STARTED", "cwd": ".", "timeout_sec": 5},
+        tmp_path,
+        runtime_state={"shell_environment": {"shell_family": "cmd"}},
+    )
+    assert result["is_error"] is True
+    assert called["value"] is False

tests/test_state_execution.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+from villani_code.state_execution import collect_validation_artifacts
+
+
+def test_collect_validation_artifacts_ignores_masked_failure_patterns() -> None:
+    transcript = {
+        "tool_results": [
+            {
+                "content": '{"command":"python web_app.py || echo ok","exit_code":0}',
+                "is_error": False,
+            }
+        ]
+    }
+    artifacts = collect_validation_artifacts(transcript)
+    assert artifacts == []
+
+
+def test_collect_validation_artifacts_keeps_clean_command() -> None:
+    transcript = {
+        "tool_results": [
+            {
+                "content": '{"command":"python web_app.py","exit_code":0}',
+                "is_error": False,
+            }
+        ]
+    }
+    artifacts = collect_validation_artifacts(transcript)
+    assert artifacts == ["python web_app.py (exit=0)"]
+
+
+def test_collect_validation_artifacts_skips_error_results_even_with_command_shape() -> None:
+    transcript = {
+        "tool_results": [
+            {
+                "content": '{"command":"python app.py > out.txt 2>&1 & echo STARTED","exit_code":0}',
+                "is_error": True,
+            }
+        ]
+    }
+    artifacts = collect_validation_artifacts(transcript)
+    assert artifacts == []