use native python for directory artifact completion

HARNESS_VERSION

@@ -1 +1 @@
-0.20.13
+0.20.14

src/clean_agent_preview.py

@@ -1603,16 +1603,21 @@ def required_artifact_completion_tool_choice(required_artifacts, offered_schemas
     targets = [str(path or '').strip().rstrip('/') for path in required_artifacts]
     if not targets:
         return None
-    # A required directory commonly contains several files and may need a
-    # programmatic extractor.  Requiring *a* tool call preserves forward
-    # progress without trapping the model in one enormous write_file payload.
-    if len(targets) == 1 and not Path(targets[0]).suffix:
-        return 'required'
     offered = {
         canonical((schema.get('function') or {}).get('name')):
         (schema.get('function') or {}).get('name')
         for schema in offered_schemas
     }
+    # A required directory commonly contains several files and needs a
+    # programmatic extractor.  Native Python avoids shell-quoting failures and
+    # one enormous multi-file write_file payload.  If Python is unavailable,
+    # require any offered tool rather than forcing the text writer.
+    if len(targets) == 1 and not Path(targets[0]).suffix:
+        name = offered.get('python')
+        return (
+            {'type': 'function', 'function': {'name': name}}
+            if name else 'required'
+        )
     preferred = (
         'python'
         if len(targets) == 1
@@ -4689,6 +4694,16 @@ async def stream_preview(*, endpoint_url, model, messages, headers, turn_contrac
                         )
                         if completion_choice is not None:
                             request['tool_choice'] = completion_choice
+                            if isinstance(completion_choice, dict):
+                                selected_name = (
+                                    completion_choice.get('function') or {}
+                                ).get('name')
+                                selected = [
+                                    schema for schema in request.get('tools') or []
+                                    if (schema.get('function') or {}).get('name') == selected_name
+                                ]
+                                if selected:
+                                    request['tools'] = selected
                     # Whole rewrites and inline feedback each have one typed
                     # editor output owner. Bind that sole channel at protocol
                     # level so prose cannot masquerade as an applied edit or

tests/test_clean_agent_preview.py

@@ -693,6 +693,10 @@ def test_required_directory_artifact_allows_any_offered_mutation_tool():
     assert required_artifact_completion_tool_choice(
         ('/tmp_workspace/results',), offered,
     ) == 'required'
+    offered.append({'type': 'function', 'function': {'name': 'python'}})
+    assert required_artifact_completion_tool_choice(
+        ('/tmp_workspace/results',), offered,
+    ) == {'type': 'function', 'function': {'name': 'python'}}
 
 
 def test_action_promise_response_rejects_future_work_but_not_real_answers():
@@ -5513,15 +5517,14 @@ async def test_native_stream_reserves_remaining_budget_for_required_artifact(mon
     ])
     payloads.extend([
         {"choices": [{"delta": {"tool_calls": [{
-            "index": 0, "id": "hallucinated-bash",
-            "function": {"name": "bash", "arguments": json.dumps({
-                "command": "echo should-not-run",
-            })},
-        }]}}]},
-        {"choices": [{"delta": {"tool_calls": [{
-            "index": 0, "id": "write",
-            "function": {"name": "write_file", "arguments": json.dumps({
-                "path": "/tmp_workspace/results/out.md", "content": "evidence",
+            "index": 0, "id": "programmatic-directory-write",
+            "function": {"name": "python", "arguments": json.dumps({
+                "code": (
+                    "from pathlib import Path\n"
+                    "p = Path('/tmp_workspace/results/out.md')\n"
+                    "p.parent.mkdir(parents=True, exist_ok=True)\n"
+                    "p.write_text('evidence')"
+                ),
             })},
         }]}}]},
         {"choices": [{"delta": {"content": "Saved."}}]},
@@ -5592,20 +5595,21 @@ async def test_native_stream_reserves_remaining_budget_for_required_artifact(mon
         and event.get("stage") == "provider_request"
         and event.get("artifact_write_phase") is True
     )
-    assert "write_file" in request_contract["offered_tools"]
+    assert request_contract["offered_tools"] == ["python"]
     assert "web_search" not in request_contract["offered_tools"]
-    assert request_contract["tool_choice"] == "required"
+    assert request_contract["tool_choice"] == {
+        "type": "function", "function": {"name": "python"},
+    }
     reserved_names = [tool["function"]["name"] for tool in requests[12]["tools"]]
-    assert reserved_names == ["bash", "python", "write_file"]
-    assert requests[12]["tool_choice"] == "required"
-    directory_path = requests[12]["tools"][2]["function"]["parameters"]["properties"]["path"]
-    assert re.search(directory_path["pattern"], "/tmp_workspace/results/out.md")
-    assert not re.search(directory_path["pattern"], "/tmp_workspace/results")
+    assert reserved_names == ["python"]
+    assert requests[12]["tool_choice"] == {
+        "type": "function", "function": {"name": "python"},
+    }
     assert "create one or more files inside" in requests[12]["messages"][-1]["content"]
     assert "Do not pass the directory itself as a file path" in requests[12]["messages"][-1]["content"]
-    assert "bash" in executed
+    assert executed[-1] == "python"
     assert any(
-        event.get("type") == "tool_output" and event.get("tool") == "write_file"
+        event.get("type") == "tool_output" and event.get("tool") == "python"
         and not event.get("error") for event in events
     )
 

tests/test_harness_version.py

@@ -8,7 +8,7 @@ ROOT = Path(__file__).resolve().parents[1]
 def test_original_harness_version_is_canonical_semver() -> None:
     version = (ROOT / "HARNESS_VERSION").read_text(encoding="utf-8").strip()
 
-    assert version == "0.20.13"
+    assert version == "0.20.14"
     assert re.fullmatch(r"\d+\.\d+\.\d+", version)