fix(agents): the running-process marker is matched only in the result header, never in command output
Some checks are pending
CI / python (push) Waiting to run
CI / workflows (push) Waiting to run
CI / ci-passed (push) Blocked by required conditions
CI / tui (push) Waiting to run
CI / viewer (push) Waiting to run
CI / package (push) Waiting to run

This commit is contained in:
Ahmed Allam 2026-10-17 23:41:20 +00:00 • committed by Ahmed Allam
parent 2b98deb3d2
commit ee22e2b8ed
2 changed files with 18 additions and 3 deletions

View file

@ -464,8 +464,9 @@ _OUTPUT_HEADER_RE = re.compile(r"^Output:\n", re.MULTILINE)
def _annotate_running_non_tty(parsed: dict[str, Any], result: Any) -> Any:
if parsed.get("tty") or not isinstance(result, str):
return result
match = _PROCESS_RUNNING_RE.search(result)
if match is None:
header = _OUTPUT_HEADER_RE.search(result)
head_end = len(result) if header is None else header.start()
if _PROCESS_RUNNING_RE.search(result, 0, head_end) is None:
return result
notice = _RUNNING_NON_TTY_NOTICE
requested = parsed.get("yield_time_ms")
@ -474,7 +475,6 @@ def _annotate_running_non_tty(parsed: dict[str, Any], result: Any) -> Any:
f"Note: yield_time_ms={int(requested)} was capped at {PTY_YIELD_TIME_MS_MAX} "
f"({_EXEC_YIELD_CAP_S} s). " + notice.removeprefix("Note: ")
)
header = _OUTPUT_HEADER_RE.search(result, match.end())
if header is None:
return f"{result}\n{notice}"
return f"{result[: header.end()]}{notice}\n{result[header.end() :]}"

View file

@ -153,6 +153,21 @@ async def test_wrap_exec_command_leaves_an_exited_process_alone() -> None:
assert result == exited
@pytest.mark.asyncio
async def test_wrap_exec_command_ignores_the_running_marker_inside_command_output() -> None:
exited = (
"Chunk ID: abc123\nWall time: 0.2 seconds\nProcess exited with code 0\nOutput:\n"
"Process running with session ID 7\nOutput:\n"
)
wrapped = factory._wrap_exec_command(_running_exec_tool(exited))
result = await wrapped.on_invoke_tool(
cast("Any", None), json.dumps({"cmd": "cat transcript.txt"})
)
assert result == exited
@pytest.mark.asyncio
async def test_responses_filesystem_custom_tool_output_is_bounded() -> None:
async def invoke(_ctx: Any, _inp: str) -> str: