recover concatenated artifact writes

HARNESS_VERSION

@@ -1 +1 @@
-0.20.11
+0.20.12

src/clean_agent_preview.py

@@ -624,11 +624,22 @@ def explicit_text_artifact_target(user_text):
 def malformed_write_handoff_target(arguments, required_artifacts=(), user_text=''):
     """Recover one textual write target even when no prompt path was parsed."""
     candidates = tuple(required_artifacts or ())
-    target = candidates[0] if len(candidates) == 1 else (
+    recovered = _partial_json_string_field(arguments, 'path')
-        _partial_json_string_field(arguments, 'path')
+    sole_required = str(candidates[0] or '').strip() if len(candidates) == 1 else ''
-        or explicit_text_artifact_target(user_text)
+    # A runner may require a directory containing several outputs.  It is not
+    # itself a writable file target, so prefer the concrete descendant path in
+    # the malformed call.  Exact required files remain authoritative.
+    target = (
+        sole_required
+        if sole_required and Path(sole_required).suffix
+        else recovered or explicit_text_artifact_target(user_text)
     )
     target = str(target or '').strip()
+    if sole_required and not Path(sole_required).suffix:
+        required_root = os.path.normpath(sole_required)
+        normalized_target = os.path.normpath(target) if target else ''
+        if not normalized_target.startswith(required_root + os.sep):
+            return ''
     if not target or Path(target).suffix.lower() in _NON_TEXT_ARTIFACT_SUFFIXES:
         return ''
     return target
@@ -1637,6 +1648,72 @@ def protocol_safe_tool_calls(calls):
     return safe_calls
 
 
+def expand_concatenated_write_calls(calls, *, max_calls=16):
+    """Split one unambiguous stream of write_file JSON objects.
+
+    Some OpenAI-compatible providers serialize a requested multi-call batch as
+    adjacent JSON objects inside one ``arguments`` string.  Recover only the
+    narrow write-file shape; all expanded calls still pass the ordinary schema,
+    policy, workspace, repeat, and execution checks later in the loop.
+    """
+    expanded = []
+    recovered_count = 0
+    decoder = json.JSONDecoder()
+    for call in calls or ():
+        function = call.get('function') or {}
+        raw = function.get('arguments') or ''
+        if canonical(function.get('name', '')) != 'write_file':
+            expanded.append(call)
+            continue
+        try:
+            json.loads(raw)
+        except (TypeError, ValueError, json.JSONDecodeError):
+            pass
+        else:
+            expanded.append(call)
+            continue
+        if not isinstance(raw, str):
+            expanded.append(call)
+            continue
+        items = []
+        cursor = 0
+        try:
+            while cursor < len(raw):
+                while cursor < len(raw) and raw[cursor].isspace():
+                    cursor += 1
+                if cursor >= len(raw):
+                    break
+                item, cursor = decoder.raw_decode(raw, cursor)
+                items.append(item)
+                if len(items) > max_calls:
+                    raise ValueError('too many concatenated write calls')
+        except (TypeError, ValueError, json.JSONDecodeError):
+            expanded.append(call)
+            continue
+        if not (2 <= len(items) <= max_calls) or not all(
+            isinstance(item, dict)
+            and set(item) == {'path', 'content'}
+            and isinstance(item['path'], str)
+            and bool(item['path'].strip())
+            and isinstance(item['content'], str)
+            for item in items
+        ):
+            expanded.append(call)
+            continue
+        base_id = str(call.get('id') or 'call_write')
+        for index, item in enumerate(items):
+            expanded.append({
+                'id': f'{base_id}_{index}',
+                'type': 'function',
+                'function': {
+                    'name': function.get('name', 'write_file'),
+                    'arguments': json.dumps(item, ensure_ascii=False),
+                },
+            })
+        recovered_count += len(items)
+    return expanded, recovered_count
+
+
 def artifact_body_from_handoff(response):
     """Extract a complete textual artifact body from a no-tools recovery turn."""
     raw = str(response or '').strip()
@@ -4746,6 +4823,14 @@ async def stream_preview(*, endpoint_url, model, messages, headers, turn_contrac
                             })
                         else:
                             unexecutable_dsml_completion = True
+                proposed, recovered_write_calls = expand_concatenated_write_calls(proposed)
+                if recovered_write_calls:
+                    yield event({
+                        'type': 'tool_argument_recovery',
+                        'format': 'concatenated_json_objects',
+                        'round': round_number,
+                        'calls': recovered_write_calls,
+                    })
                 proposed = serialize_required_email_attachment_chain(
                     proposed, contract_required_tools, executions,
                 )

tests/test_clean_agent_preview.py

@@ -926,6 +926,128 @@ async def test_malformed_writer_body_handoff_reopens_writer_after_two_tool_viola
     )
 
 
+@pytest.mark.asyncio
+async def test_concatenated_writer_arguments_execute_as_bounded_ordered_calls(monkeypatch):
+    import src.clean_agent_preview as module
+
+    concatenated = ''.join(json.dumps(item) for item in (
+        {'path': '/workspace/results/1.tex', 'content': '1.tex'},
+        {'path': '/workspace/results/1.tex', 'content': '\\begin{table}One\\end{table}'},
+        {'path': '/workspace/results/2.tex', 'content': '2.tex'},
+        {'path': '/workspace/results/2.tex', 'content': '\\begin{table}Two\\end{table}'},
+    ))
+    responses = iter([
+        {'choices': [{'delta': {'tool_calls': [{
+            'index': 0, 'id': 'concatenated-writes',
+            'function': {'name': 'write_file', 'arguments': concatenated},
+        }]}}]},
+        {'choices': [{'delta': {'content': 'Created the requested TeX files.'}}]},
+    ])
+
+    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))
+
+    executed = []
+
+    async def execute(block, **kwargs):
+        path, content = block.content.split('\n', 1)
+        executed.append({'path': path, 'content': content})
+        return block.tool_type, {'output': 'written', 'exit_code': 0}
+
+    monkeypatch.setattr(module, 'NATIVE_ARTIFACT_RESEARCH_LIMIT', 0)
+    monkeypatch.setattr(module.httpx, 'AsyncClient', Client)
+    monkeypatch.setattr(module, 'execute_tool_block', execute)
+    schema = next(
+        item for item in FUNCTION_TOOL_SCHEMAS
+        if item['function']['name'] == 'write_file'
+    )
+    contract = resolve_full_inventory_contract(schemas=[schema], policy=ToolPolicy())
+    raw = [chunk async for chunk in stream_preview(
+        endpoint_url='http://test', model='test',
+        messages=[{'role': 'user', 'content': 'Extract every table into the results directory.'}],
+        headers={}, turn_contract=contract, session_id='test', owner='test',
+        disabled_tools=set(), tool_policy=ToolPolicy(), workspace='/workspace',
+        client_runtime_context={
+            'surface': 'odysseus-native', 'terminal_agent': True,
+            'unattended_mode': True,
+            'completion_requirements': {
+                'required_artifacts': ['/workspace/results'],
+            },
+        }, max_tokens=8192, max_rounds=2,
+    )]
+
+    assert len(requests) == 2
+    assert executed == [
+        {'path': '/workspace/results/1.tex', 'content': '1.tex'},
+        {'path': '/workspace/results/1.tex', 'content': '\\begin{table}One\\end{table}'},
+        {'path': '/workspace/results/2.tex', 'content': '2.tex'},
+        {'path': '/workspace/results/2.tex', 'content': '\\begin{table}Two\\end{table}'},
+    ]
+    events = [json.loads(chunk[6:]) for chunk in raw if '[DONE]' not in chunk]
+    assert any(
+        event.get('type') == 'tool_argument_recovery'
+        and event.get('format') == 'concatenated_json_objects'
+        and event.get('calls') == 4
+        for event in events
+    )
+    assert not any(
+        event.get('reason') == 'malformed_write_body_handoff'
+        for event in events
+    )
+
+
+def test_concatenated_writer_recovery_fails_closed_for_ambiguous_or_large_batches():
+    from src.clean_agent_preview import expand_concatenated_write_calls
+
+    ambiguous = {
+        'id': 'ambiguous', 'type': 'function', 'function': {
+            'name': 'write_file',
+            'arguments': '{"path":"/workspace/a","content":"a"}{"path":"/workspace/b","mode":"x"}',
+        },
+    }
+    too_large = {
+        'id': 'large', 'type': 'function', 'function': {
+            'name': 'write_file',
+            'arguments': ''.join(
+                json.dumps({'path': f'/workspace/{index}', 'content': 'x'})
+                for index in range(17)
+            ),
+        },
+    }
+
+    assert expand_concatenated_write_calls([ambiguous]) == ([ambiguous], 0)
+    assert expand_concatenated_write_calls([too_large]) == ([too_large], 0)
+
+
+def test_malformed_writer_handoff_uses_descendant_file_not_required_directory():
+    from src.clean_agent_preview import malformed_write_handoff_target
+
+    assert malformed_write_handoff_target(
+        '{"path":"/workspace/results/1.tex"}{"content":"body"}',
+        ['/workspace/results'],
+    ) == '/workspace/results/1.tex'
+    assert malformed_write_handoff_target(
+        '{"path":"/workspace/outside.tex"}{"content":"body"}',
+        ['/workspace/results'],
+    ) == ''
+
+
 @pytest.mark.asyncio
 async def test_binary_artifact_completion_requests_python_not_text_writer(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.11"
+    assert version == "0.20.12"
     assert re.fullmatch(r"\d+\.\d+\.\d+", version)