fix(agent): scope parallel batches from V4A patch headers
This commit is contained in:
Fangliquan
Teknium
parent
003af7f85c
commit
87bc710609
|
|
@ -4,9 +4,10 @@ Pure module-level utilities extracted from ``run_agent.py``:
|
|||
|
||||
* ``_is_destructive_command`` — terminal-command heuristic used to gate
|
||||
parallel batch dispatch.
|
||||
* ``_should_parallelize_tool_batch`` / ``_extract_parallel_scope_path`` /
|
||||
``_paths_overlap`` — the rules engine deciding when a multi-tool batch
|
||||
can run concurrently.
|
||||
* ``_should_parallelize_tool_batch`` / ``_extract_parallel_scope_paths`` /
|
||||
``_extract_parallel_scope_path`` / ``_paths_overlap`` — the rules engine
|
||||
deciding when a multi-tool batch can run concurrently (V4A patch scope
|
||||
uses patch-body file headers, not a decoy ``path=``).
|
||||
* ``_is_multimodal_tool_result`` / ``_multimodal_text_summary`` /
|
||||
``_append_subdir_hint_to_multimodal`` — envelope helpers for the
|
||||
``{"_multimodal": True, "content": [...], "text_summary": ...}`` dict
|
||||
|
|
@ -126,7 +127,7 @@ def _plan_tool_batch_segments(tool_calls, *, execution_cwd: Optional[Path] = Non
|
|||
* ``_NEVER_PARALLEL_TOOLS`` (interactive tools) → barrier.
|
||||
* Unparseable / non-dict arguments → barrier.
|
||||
* Path-scoped tools (``read_file``/``search_files``/``write_file``/
|
||||
``patch``) join a parallel run only when their target path does not
|
||||
``patch``) join a parallel run only when their target path(s) do not
|
||||
CONFLICT with a path already reserved in the same run. Reservations
|
||||
carry a reader/writer role: reader↔reader overlap is harmless (two
|
||||
reads of the same file commute) and stays parallel; any overlap
|
||||
|
|
@ -134,7 +135,9 @@ def _plan_tool_batch_segments(tool_calls, *, execution_cwd: Optional[Path] = Non
|
|||
NEW run after the first completes. ``search_files`` reserves its
|
||||
search root (default ``.``) as a reader — a search batched after a
|
||||
write into the searched subtree is ordered behind that write instead
|
||||
of racing it.
|
||||
of racing it. For V4A ``patch(mode="patch")`` the reserved paths are
|
||||
the file headers in the patch body, not a possibly-stale ``path=``
|
||||
argument.
|
||||
* Anything not in ``_PARALLEL_SAFE_TOOLS`` and not an opted-in MCP
|
||||
tool → barrier.
|
||||
|
||||
|
|
@ -189,14 +192,17 @@ def _plan_tool_batch_segments(tool_calls, *, execution_cwd: Optional[Path] = Non
|
|||
continue
|
||||
|
||||
if tool_name in _PATH_SCOPED_TOOLS:
|
||||
scoped_path = _extract_parallel_scope_path(tool_name, function_args, execution_cwd=execution_cwd)
|
||||
if scoped_path is None:
|
||||
scoped_paths = _extract_parallel_scope_paths(
|
||||
tool_name, function_args, execution_cwd=execution_cwd
|
||||
)
|
||||
if not scoped_paths:
|
||||
_add_sequential(tool_call)
|
||||
continue
|
||||
is_writer = tool_name in _PATH_SCOPED_WRITERS
|
||||
if any(
|
||||
(is_writer or existing_is_writer)
|
||||
and _paths_overlap(scoped_path, existing)
|
||||
for scoped_path in scoped_paths
|
||||
for existing, existing_is_writer in reserved_paths
|
||||
):
|
||||
# Same-subtree conflict inside this run: close it so this
|
||||
|
|
@ -204,7 +210,7 @@ def _plan_tool_batch_segments(tool_calls, *, execution_cwd: Optional[Path] = Non
|
|||
# Reader↔reader overlap never conflicts — concurrent reads
|
||||
# of the same subtree commute.
|
||||
_close_parallel()
|
||||
reserved_paths.append((scoped_path, is_writer))
|
||||
reserved_paths.extend((p, is_writer) for p in scoped_paths)
|
||||
current.append(tool_call)
|
||||
continue
|
||||
|
||||
|
|
@ -256,39 +262,77 @@ def _canonical_path(raw_path: str, execution_cwd: Optional[Path] = None) -> Path
|
|||
return Path(resolved)
|
||||
|
||||
|
||||
def _extract_parallel_scope_path(
|
||||
def _extract_parallel_scope_paths(
|
||||
tool_name: str,
|
||||
function_args: dict,
|
||||
execution_cwd: Optional[Path] = None,
|
||||
) -> Optional[Path]:
|
||||
"""Return the canonical file target for path-scoped tools.
|
||||
) -> List[Path]:
|
||||
"""Return every canonical path this call reserves for overlap checks.
|
||||
|
||||
*execution_cwd* should be the working directory that the tool will
|
||||
actually use at runtime. When omitted the process cwd is used,
|
||||
which may differ from the tool execution environment on some
|
||||
platforms (e.g. WSL, sandboxed sub-processes).
|
||||
|
||||
For ``patch`` in V4A ``mode=patch``, scope comes from patch-body
|
||||
``*** Update/Add/Delete/Move File:`` headers (not a possibly-decoy
|
||||
``path=``). An empty result means the planner cannot determine the
|
||||
scope and must treat the call as a sequential barrier.
|
||||
"""
|
||||
if tool_name not in _PATH_SCOPED_TOOLS:
|
||||
return None
|
||||
return []
|
||||
|
||||
raw_path = function_args.get("path")
|
||||
if not isinstance(raw_path, str) or not raw_path.strip():
|
||||
# ``search_files`` defaults its search root to the cwd when
|
||||
# ``path`` is omitted — reserve that root rather than falling
|
||||
# back to a sequential barrier (returning None here would demote
|
||||
# every bare search to a barrier and destroy read parallelism).
|
||||
if tool_name == "search_files":
|
||||
return _canonical_path(".", execution_cwd)
|
||||
return None
|
||||
raw_paths: List[str] = []
|
||||
if tool_name == "patch" and (function_args.get("mode") or "replace") == "patch":
|
||||
raw_paths.extend(_extract_file_mutation_targets(tool_name, function_args))
|
||||
else:
|
||||
raw_path = function_args.get("path")
|
||||
if isinstance(raw_path, str) and raw_path.strip():
|
||||
raw_paths.append(raw_path)
|
||||
elif tool_name == "search_files":
|
||||
# ``search_files`` defaults its search root to the cwd when
|
||||
# ``path`` is omitted — reserve that root rather than falling
|
||||
# back to a sequential barrier (an empty result here would
|
||||
# demote every bare search to a barrier and destroy read
|
||||
# parallelism).
|
||||
raw_paths.append(".")
|
||||
|
||||
return _canonical_path(raw_path, execution_cwd)
|
||||
scoped: List[Path] = []
|
||||
seen: set[str] = set()
|
||||
for raw in raw_paths:
|
||||
if not isinstance(raw, str) or not raw.strip():
|
||||
continue
|
||||
canonical = _canonical_path(raw, execution_cwd)
|
||||
key = str(canonical)
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
scoped.append(canonical)
|
||||
return scoped
|
||||
|
||||
|
||||
def _extract_parallel_scope_path(
|
||||
tool_name: str,
|
||||
function_args: dict,
|
||||
execution_cwd: Optional[Path] = None,
|
||||
) -> Optional[Path]:
|
||||
"""Return the primary canonical file target for path-scoped tools.
|
||||
|
||||
Thin view over ``_extract_parallel_scope_paths`` kept for callers/tests
|
||||
that only need a single representative path. For multi-file V4A
|
||||
patches this is the first header target.
|
||||
"""
|
||||
scoped = _extract_parallel_scope_paths(
|
||||
tool_name, function_args, execution_cwd=execution_cwd
|
||||
)
|
||||
return scoped[0] if scoped else None
|
||||
|
||||
|
||||
def _paths_overlap(left: Path, right: Path) -> bool:
|
||||
"""Return True when two paths may refer to the same subtree.
|
||||
|
||||
Both *left* and *right* must already be canonical (as returned by
|
||||
``_extract_parallel_scope_path`` / ``_canonical_path``) so that
|
||||
``_extract_parallel_scope_paths`` / ``_canonical_path``) so that
|
||||
symlink aliases and case differences are already normalised.
|
||||
"""
|
||||
left_parts = left.parts
|
||||
|
|
@ -383,8 +427,10 @@ def _extract_file_mutation_targets(tool_name: str, args: Dict[str, Any]) -> List
|
|||
if not isinstance(body, str) or not body:
|
||||
return []
|
||||
paths: List[str] = []
|
||||
# ``\s*`` (not ``\s+``) after ``***`` matches patch_parser / file_tools:
|
||||
# they accept ``***Update File:`` with no space after the asterisks.
|
||||
for _m in re.finditer(
|
||||
r'^\*\*\*\s+(?:Update|Add|Delete)\s+File:\s*(.+)$',
|
||||
r'^\*\*\*\s*(?:Update|Add|Delete)\s+File:\s*(.+)$',
|
||||
body,
|
||||
re.MULTILINE,
|
||||
):
|
||||
|
|
@ -392,7 +438,7 @@ def _extract_file_mutation_targets(tool_name: str, args: Dict[str, Any]) -> List
|
|||
if p:
|
||||
paths.append(p)
|
||||
for _m in re.finditer(
|
||||
r'^\*\*\*\s+Move\s+File:\s*(.+?)\s*->\s*(.+)$',
|
||||
r'^\*\*\*\s*Move\s+File:\s*(.+?)\s*->\s*(.+)$',
|
||||
body,
|
||||
re.MULTILINE,
|
||||
):
|
||||
|
|
@ -672,6 +718,7 @@ __all__ = [
|
|||
"_should_parallelize_tool_batch",
|
||||
"_canonical_path",
|
||||
"_extract_parallel_scope_path",
|
||||
"_extract_parallel_scope_paths",
|
||||
"_paths_overlap",
|
||||
"_is_multimodal_tool_result",
|
||||
"_multimodal_text_summary",
|
||||
|
|
|
|||
|
|
@ -65,6 +65,13 @@ class TestExtractFileMutationTargets:
|
|||
assert paths == ["/tmp/a.md", "/tmp/new.md", "/tmp/old.md"]
|
||||
|
||||
|
||||
def test_patch_v4a_accepts_no_space_after_asterisks(self):
|
||||
"""Match patch_parser / file_tools: ``***Update File:`` (no space)."""
|
||||
body = "***Update File: nospace.py\n"
|
||||
assert _extract_file_mutation_targets(
|
||||
"patch", {"mode": "patch", "patch": body}
|
||||
) == ["nospace.py"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _extract_error_preview
|
||||
|
|
|
|||
|
|
@ -126,6 +126,98 @@ class TestPlanToolBatchSegments:
|
|||
# Order and completeness preserved.
|
||||
assert _flatten_ids(segments) == ["w1", "r1", "w2", "r2"]
|
||||
|
||||
def test_v4a_decoy_path_does_not_parallelize_with_real_target(self, tmp_path):
|
||||
"""mode=patch scopes via V4A headers, not a decoy path= argument.
|
||||
|
||||
A patch that claims path=dummy.txt but updates real.py must not share
|
||||
a parallel segment with write_file/read_file on real.py.
|
||||
"""
|
||||
patch_body = (
|
||||
"*** Begin Patch\n"
|
||||
"*** Update File: real.py\n"
|
||||
"@@\n"
|
||||
"-old\n"
|
||||
"+new\n"
|
||||
"*** End Patch\n"
|
||||
)
|
||||
patch_args = json.dumps({
|
||||
"mode": "patch",
|
||||
"path": "dummy.txt",
|
||||
"patch": patch_body,
|
||||
})
|
||||
calls = [
|
||||
_tc("patch", patch_args, call_id="p1"),
|
||||
_tc("write_file", '{"path":"real.py","content":"x"}', call_id="w1"),
|
||||
]
|
||||
segments = _plan_tool_batch_segments(calls, execution_cwd=tmp_path)
|
||||
assert _flatten_ids(segments) == ["p1", "w1"]
|
||||
# Overlap on real.py must prevent a single parallel segment.
|
||||
assert not (
|
||||
len(segments) == 1
|
||||
and segments[0][0] == "parallel"
|
||||
and [tc.id for tc in segments[0][1]] == ["p1", "w1"]
|
||||
)
|
||||
# Solo runs demote to sequential and may merge; either shape is safe.
|
||||
if len(segments) == 1:
|
||||
assert segments[0][0] == "sequential"
|
||||
else:
|
||||
assert [tc.id for tc in segments[0][1]] == ["p1"]
|
||||
assert [tc.id for tc in segments[1][1]] == ["w1"]
|
||||
|
||||
def test_v4a_multi_file_reserves_all_header_targets(self, tmp_path):
|
||||
"""Multi-file V4A must reserve every Update/Add/Delete/Move target."""
|
||||
patch_body = (
|
||||
"*** Begin Patch\n"
|
||||
"*** Update File: a.py\n"
|
||||
"@@\n-a\n+b\n"
|
||||
"*** Add File: b.py\n"
|
||||
"+fresh\n"
|
||||
"*** End Patch\n"
|
||||
)
|
||||
# Honest path= only names a.py — b.py still must be reserved.
|
||||
patch_args = json.dumps({
|
||||
"mode": "patch",
|
||||
"path": "a.py",
|
||||
"patch": patch_body,
|
||||
})
|
||||
calls = [
|
||||
_tc("patch", patch_args, call_id="p1"),
|
||||
_tc("read_file", '{"path":"b.py"}', call_id="r1"),
|
||||
]
|
||||
segments = _plan_tool_batch_segments(calls, execution_cwd=tmp_path)
|
||||
assert _flatten_ids(segments) == ["p1", "r1"]
|
||||
assert not (
|
||||
len(segments) == 1
|
||||
and segments[0][0] == "parallel"
|
||||
and [tc.id for tc in segments[0][1]] == ["p1", "r1"]
|
||||
)
|
||||
if len(segments) == 1:
|
||||
assert segments[0][0] == "sequential"
|
||||
else:
|
||||
assert [tc.id for tc in segments[0][1]] == ["p1"]
|
||||
assert [tc.id for tc in segments[1][1]] == ["r1"]
|
||||
|
||||
def test_v4a_without_path_arg_still_scopes_from_headers(self, tmp_path):
|
||||
"""mode=patch with no path= must still parallel-scope from V4A headers."""
|
||||
patch_body = (
|
||||
"*** Begin Patch\n"
|
||||
"*** Update File: real.py\n"
|
||||
"@@\n-old\n+new\n"
|
||||
"*** End Patch\n"
|
||||
)
|
||||
patch_args = json.dumps({"mode": "patch", "patch": patch_body})
|
||||
calls = [
|
||||
_tc("patch", patch_args, call_id="p1"),
|
||||
_tc("write_file", '{"path":"other.py","content":"x"}', call_id="w1"),
|
||||
_tc("read_file", '{"path":"real.py"}', call_id="r1"),
|
||||
]
|
||||
segments = _plan_tool_batch_segments(calls, execution_cwd=tmp_path)
|
||||
# p1+w1 are disjoint → can share a parallel run; r1 overlaps real.py → new run.
|
||||
assert _flatten_ids(segments) == ["p1", "w1", "r1"]
|
||||
assert [tc.id for tc in segments[0][1]] == ["p1", "w1"]
|
||||
assert segments[0][0] == "parallel"
|
||||
assert [tc.id for tc in segments[1][1]] == ["r1"]
|
||||
|
||||
def test_path_scoped_tool_without_path_is_a_barrier(self):
|
||||
calls = [
|
||||
_tc("read_file", "{}", call_id="nopath"),
|
||||
|
|
|
|||
Loading…
Reference in New Issue