recover required artifacts after tool suppression

HARNESS_VERSION

@@ -1 +1 @@
-0.20.9
+0.20.10

src/clean_agent_preview.py

@@ -6357,6 +6357,44 @@ async def stream_preview(*, endpoint_url, model, messages, headers, turn_contrac
                     break
                 if terminal_suppression_violation:
                     missing_artifacts = missing_workspace_artifacts(latest_user, workspace)
+                    remaining_artifact_tools = artifact_completion_tool_schemas(
+                        [
+                            schema for schema in offered
+                            if canonical(schema['function']['name'])
+                            not in permanently_suppressed_tools
+                        ],
+                        required_artifacts,
+                    )
+                    if (
+                        missing_artifacts
+                        and remaining_artifact_tools
+                        and not suppression_completion_attempted
+                        and round_number < round_limit
+                    ):
+                        suppression_completion_attempted = True
+                        artifact_write_phase = True
+                        force_no_tools_next_round = False
+                        recovery = (
+                            'Completion recovery: the repeated evidence tool is disabled. '
+                            'Do not call it again. Use the evidence already returned and an '
+                            'available workspace writer to create and verify the missing '
+                            'artifact(s): ' + ', '.join(missing_artifacts) + '. '
+                            'Do not perform more research before writing.'
+                        )
+                        if history and history[-1].get('_harness_control'):
+                            history[-1]['content'] = (
+                                str(history[-1].get('content') or '') + ' ' + recovery
+                            )
+                        else:
+                            history.append({
+                                'role': 'user', '_harness_control': True, 'content': recovery,
+                            })
+                        yield event({
+                            'type': 'completion_recovery',
+                            'reason': 'suppressed_tool_artifact_recovery',
+                            'missing_artifacts': list(missing_artifacts),
+                        })
+                        continue
                     if (
                         not missing_artifacts
                         and not suppression_completion_attempted

tests/test_clean_agent_preview.py

@@ -5964,6 +5964,108 @@ async def test_native_stream_terminates_after_calling_a_permanently_suppressed_t
     assert metrics['clean_v3_turn'][-1]['content'] == 'Final answer from existing evidence.'
 
 
+@pytest.mark.asyncio
+async def test_suppressed_evidence_tool_recovers_missing_artifact_with_writer(
+    monkeypatch, tmp_path,
+):
+    import src.clean_agent_preview as module
+
+    output = tmp_path / 'results' / 'report.md'
+    output_alias = '/workspace/results/report.md'
+    search_arguments = json.dumps({'query': 'arxiv cs.CV 2026-02-25'})
+    responses = iter([
+        {'choices': [{'delta': {'tool_calls': [{
+                'index': 0, 'id': f'search-{index}', 'function': {
+                'name': 'web_search', 'arguments': search_arguments,
+                },
+            }]}}]}
+        for index in range(1, 4)
+    ] + [
+        {'choices': [{'delta': {'tool_calls': [{
+            'index': 0, 'id': 'write-1', 'function': {
+                'name': 'write_file', 'arguments': json.dumps({
+                    'path': output_alias, 'content': '# Verified report\nEvidence retained.\n',
+                }),
+            },
+        }]}}]},
+        {'choices': [{'delta': {'content': 'Created and verified the report.'}}]},
+    ])
+
+    class Response:
+        def __init__(self, payload): self.payload = payload
+        async def __aenter__(self): return self
+        async def __aexit__(self, *args): pass
+        def raise_for_status(self): pass
+        async def aiter_lines(self):
+            yield 'data: ' + json.dumps(self.payload)
+            yield 'data: [DONE]'
+
+    requests = []
+
+    class Client:
+        def __init__(self, **kwargs): pass
+        async def __aenter__(self): return self
+        async def __aexit__(self, *args): pass
+        def stream(self, *args, **kwargs):
+            requests.append(kwargs['json'])
+            return Response(next(responses))
+
+    executions = []
+
+    async def execute(block, **kwargs):
+        executions.append(block.tool_type)
+        if block.tool_type == 'write_file':
+            output.parent.mkdir(parents=True, exist_ok=True)
+            output.write_text('# Verified report\nEvidence retained.\n')
+            return 'write_file', {'output': f'wrote {output}', 'exit_code': 0}
+        return 'web_search', {'output': 'useful arxiv evidence', 'exit_code': 0}
+
+    monkeypatch.setattr(module.httpx, 'AsyncClient', Client)
+    monkeypatch.setattr(module, 'execute_tool_block', execute)
+    monkeypatch.setattr(module, 'NATIVE_ARTIFACT_RESEARCH_LIMIT', 100)
+    schemas = [
+        next(s for s in FUNCTION_TOOL_SCHEMAS if s['function']['name'] == name)
+        for name in ('web_search', 'write_file')
+    ]
+    contract = resolve_full_inventory_contract(schemas=schemas, policy=ToolPolicy())
+    raw = [chunk async for chunk in stream_preview(
+        endpoint_url='http://test', model='test', headers={},
+        messages=[{'role': 'user', 'content': (
+            'Find cs.CV papers submitted on 2026-02-25 and create '
+            f'{output_alias} from the returned evidence.'
+        )}],
+        turn_contract=contract, session_id='test', owner='test',
+        disabled_tools=set(), tool_policy=ToolPolicy(), workspace=str(tmp_path),
+        client_runtime_context={
+            'surface': 'odysseus-native', 'terminal_agent': True,
+            'unattended_mode': True,
+            'completion_requirements': {'required_artifacts': [output_alias]},
+        }, max_rounds=8,
+    )]
+
+    events = [json.loads(chunk[6:]) for chunk in raw if '[DONE]' not in chunk]
+    assert executions == ['web_search', 'write_file'], [
+        (event.get('type'), event.get('reason'), event.get('tool')) for event in events
+    ]
+    recovery_events = [
+        event for event in events
+        if event.get('reason') == 'suppressed_tool_artifact_recovery'
+    ]
+    assert recovery_events, [
+        (event.get('type'), event.get('reason'), event.get('tool'), event.get('output'))
+        for event in events
+    ]
+    recovery = recovery_events[0]
+    assert recovery['missing_artifacts'] == [output_alias]
+    assert any(
+        tool['function']['name'] == 'write_file'
+        for tool in requests[4].get('tools', [])
+    )
+    final = [event for event in events if event.get('type') == 'final_response']
+    assert all('repeated a tool call' not in event.get('content', '') for event in final)
+    assert output.read_text().startswith('# Verified report')
+
+
 @pytest.mark.asyncio
 async def test_native_stream_stops_reexecuting_an_identical_failed_call(monkeypatch):
     import src.clean_agent_preview as module

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.9"
+    assert version == "0.20.10"
     assert re.fullmatch(r"\d+\.\d+\.\d+", version)