test: re-home 3 sequential sys_terminal e2e tests to mock-LLM sessions layer (#594)

Three sys_terminal_* e2e tests in tests/e2e/test_sys_terminal_e2e.py were
quarantined in known_failures.yaml (issue 532, cluster terminal-d6) with a
stale misdiagnosis ("500 / runner availability"). The real reason: they drove
the removed POST /v1/responses route (and poll_until_terminal's
GET /v1/responses/{id}), which no longer exists under omnigent/server/routes/.
They could never go green as written.

Re-home their behavioral intent onto the current runner-bound, mock-LLM
sessions API (the same path the merged D6 re-homes use), then delete the old
e2e tests + their known_failures.yaml entries.

- New file: tests/integration/test_sys_terminal_round_trip.py — 3 tests
  driving the sessions API in mock mode against real tmux.
- sys_terminal_* are server-executed tools: the runner's dispatcher runs
  TerminalRegistry -> real tmux and threads the result back to the model. A
  mock LLM scripted with [launch, send, read, list, close, final_text]
  executes the steps in strict sequential order, giving the same
  launch->send->read->list->close ordering the old real-LLM e2e relied on
  without trusting an LLM to follow a prompt.

Adopt the centralized mock-isolation infra from main (PR #602):
- Module-level pytestmark = pytest.mark.mock_only so the 3 tests skip in the
  real-LLM Integration (*) jobs. The central gate in
  tests/integration/conftest.py keys off the real _is_mock_mode signal
  (absence of --llm-api-key).
- Delete the dead "if mock_llm_server_url is None: pytest.skip(...)" guards:
  the mock server fixture is always started regardless of --llm-api-key, so
  that guard never fired in any job (the cause of the 401 in
  Integration (claude-sdk)).
- Delete the per-file autouse reset_mock_llm fixture: the centralized autouse
  _reset_mock_llm_between_tests in tests/integration/conftest.py now resets
  the shared mock server before/after every integration test.

Removed (tests + their known_failures.yaml entries, issue 532 / terminal-d6):
- test_sys_terminal_basic_round_trip_e2e
- test_sys_terminal_full_workflow_e2e
- test_sys_terminal_send_keys_drives_interactive_e2e

test_sys_terminal_ten_parallel_dispatches_complete_e2e, its known_failures.yaml
entry, and the shared _get_function_call_outputs helper are left intact.

Verified locally:
- pytest --integration --llm-api-key dummy -> all 3 SKIP (the marker gates them
  out of the real-LLM jobs).
- Mixed-order mock shard (round_trip + smoke + multi_turn + sharing) -> 6 passed,
  3x for determinism.

tests/e2e/test_sys_terminal_e2e.py

@@ -66,318 +66,6 @@ def _get_function_call_outputs(
     return outputs
 
 
-def test_sys_terminal_basic_round_trip_e2e(
-    live_server: str,
-    sys_terminal_test_agent: str,
-    http_client: httpx.Client,
-) -> None:
-    """
-    Real LLM drives the full ``sys_terminal_*`` round trip
-    against a real tmux. Asserts on the raw tool outputs (not
-    prose) so flaky LLM wording can't fail the test.
-
-    What this verifies:
-      1. The compat translator threaded ``terminals:`` from
-         omnigent YAML through to ``AgentSpec.terminals``.
-      2. The AP-side ToolManager registered the
-         ``sys_terminal_*`` family from
-         ``ToolManager._register_terminal_tools``.
-      3. The LLM successfully invoked launch + send + read.
-      4. The :class:`TerminalRegistry` spawned a real tmux
-         session; ``send`` reached it; ``read`` saw the marker.
-
-    What breaks if this fails (top suspects):
-      - ``AgentSpec.terminals=None`` after translation → tools
-        not registered → "tool not available" mid-conversation.
-      - Workflow path differs from test ToolManager registration
-        path → tools register in tests but not at runtime.
-      - tmux subprocess spawn fails silently → empty pane reads.
-    """
-    marker = "TERMINAL_E2E_MARKER_AAAA"
-    prompt = (
-        f"Use sys_terminal_launch to start the 'bash' terminal with "
-        f"session 's1'. Then use sys_terminal_send to type "
-        f"'echo {marker}' followed by Enter. Wait briefly for the "
-        f"output, then call sys_terminal_read on session 's1'. "
-        f"Report what you saw. Do this in one go, then reply 'done'."
-    )
-    resp = http_client.post(
-        "/v1/responses",
-        json={
-            "model": sys_terminal_test_agent,
-            "input": prompt,
-            "stream": False,
-        },
-        timeout=180.0,
-    )
-    resp.raise_for_status()
-    body = resp.json()
-    response_id = body["id"]
-    body = poll_until_terminal(http_client, response_id, timeout=180)
-    assert body["status"] == "completed", (
-        f"Workflow failed before completion. Status={body['status']!r}; "
-        f"error={body.get('error')!r}. If 'failed' with an exception "
-        f"about ``sys_terminal_launch``, the tools didn't register on "
-        f"the AP-side ToolManager."
-    )
-
-    conv_id = body["conversation"]["id"]
-
-    # The LLM must have called launch — otherwise the rest of the
-    # test would be testing nothing.
-    launches = _get_function_call_outputs(http_client, conv_id, "sys_terminal_launch")
-    assert len(launches) >= 1, (
-        f"sys_terminal_launch was never called; conv_id={conv_id}. "
-        f"If 0 calls, either the LLM ignored the prompt or the tool "
-        f"wasn't on the schema (registration regression)."
-    )
-    launch_result = json.loads(launches[0])
-    assert launch_result.get("status") == "launched", (
-        f"First launch should report status='launched'; got "
-        f"{launch_result!r}. If the value is 'already_running', the "
-        f"registry reused stale state from a prior test run."
-    )
-
-    # The marker must appear in at least one read output. We
-    # don't constrain the LLM's call ordering (it might read
-    # twice, retry, etc.), only that the data flowed.
-    reads = _get_function_call_outputs(http_client, conv_id, "sys_terminal_read")
-    assert len(reads) >= 1, f"sys_terminal_read was never called; conv_id={conv_id}."
-    combined_screens = " ".join(reads)
-    assert marker in combined_screens, (
-        f"Echo marker {marker!r} not seen in any sys_terminal_read "
-        f"output. Reads: {reads!r}. If empty, the send didn't reach "
-        f"tmux. If reads have a prompt but not the echo, the bash "
-        f"command failed in tmux (e.g. shell-init error)."
-    )
-
-
-def test_sys_terminal_full_workflow_e2e(
-    live_server: str,
-    sys_terminal_test_agent: str,
-    http_client: httpx.Client,
-) -> None:
-    """
-    A coherent task that exercises ALL FIVE ``sys_terminal_*`` tools
-    in one conversation. The LLM is asked to perform a small shell
-    investigation, then verify the cleanup succeeded.
-
-    Flow:
-      1. ``sys_terminal_launch`` — start ``bash:investigate``.
-      2. ``sys_terminal_send`` — write a marker to a tmp file.
-      3. ``sys_terminal_read`` — capture the pane confirming the
-         echo + the file-write completed.
-      4. ``sys_terminal_list`` — confirm the registry shows
-         ``bash:investigate`` as running.
-      5. ``sys_terminal_close`` — kill the session.
-      6. ``sys_terminal_list`` again — confirm the registry no
-         longer reports ``bash:investigate``.
-
-    What this catches that the focused tests don't:
-      - ``sys_terminal_list`` schema/dispatch never gets exercised
-        through the LLM in the focused tests; a malformed list
-        schema or wrong return shape would only fail here.
-      - ``sys_terminal_close`` likewise — the focused tests
-        verify the registry-level close, but not the LLM-driven
-        path through the AP-side ToolManager.
-      - The post-close list is the only e2e check that close
-        actually removed the registry entry (not just killed
-        the process); without it, a leak that only surfaces
-        across multiple turns / closes would be invisible.
-
-    The prompt tells the LLM the sequence explicitly; the
-    assertions check tool names appear in conversation items
-    rather than trusting the LLM's prose summary. LLM ordering
-    flexibility within reason: as long as all 5 tools fire and
-    the markers / list states show up in the right order, the
-    test passes.
-    """
-    marker = "FULL_WORKFLOW_MARKER_BBBB"
-    prompt = (
-        "Perform this exact sequence using sys_terminal_* tools. "
-        "Do NOT skip steps. Reply only after step 6 completes.\n\n"
-        f"  1. sys_terminal_launch terminal='bash' session='investigate'.\n"
-        f"  2. sys_terminal_send terminal='bash' session='investigate' "
-        f"text='echo {marker}' keys='Enter'.\n"
-        "  3. sys_terminal_read terminal='bash' session='investigate'.\n"
-        "  4. sys_terminal_list (no args) — capture the result.\n"
-        "  5. sys_terminal_close terminal='bash' session='investigate'.\n"
-        "  6. sys_terminal_list again (no args) — capture the result.\n\n"
-        "Reply with 'done' once step 6 completes. No prose, no extra "
-        "wording."
-    )
-    resp = http_client.post(
-        "/v1/responses",
-        json={
-            "model": sys_terminal_test_agent,
-            "input": prompt,
-            "stream": False,
-        },
-        timeout=240.0,
-    )
-    resp.raise_for_status()
-    body = poll_until_terminal(http_client, resp.json()["id"], timeout=240)
-    assert body["status"] == "completed", (
-        f"Workflow failed: status={body.get('status')!r}, error={body.get('error')!r}."
-    )
-    conv_id = body["conversation"]["id"]
-
-    # All 5 tool names must appear in the conversation. We pull
-    # the raw call lists so a missing tool can be named in the
-    # failure message.
-    launches = _get_function_call_outputs(http_client, conv_id, "sys_terminal_launch")
-    sends = _get_function_call_outputs(http_client, conv_id, "sys_terminal_send")
-    reads = _get_function_call_outputs(http_client, conv_id, "sys_terminal_read")
-    lists = _get_function_call_outputs(http_client, conv_id, "sys_terminal_list")
-    closes = _get_function_call_outputs(http_client, conv_id, "sys_terminal_close")
-
-    missing = [
-        name
-        for name, calls in [
-            ("sys_terminal_launch", launches),
-            ("sys_terminal_send", sends),
-            ("sys_terminal_read", reads),
-            ("sys_terminal_list", lists),
-            ("sys_terminal_close", closes),
-        ]
-        if not calls
-    ]
-    assert not missing, (
-        f"LLM didn't invoke these tools: {missing!r}. The full-workflow "
-        f"test requires all 5. If sys_terminal_list or sys_terminal_close "
-        f"is missing, those paths have no e2e coverage anywhere else."
-    )
-
-    # The marker must appear in at least one read — proves
-    # send actually reached tmux and read captured the output.
-    combined_reads = " ".join(reads)
-    assert marker in combined_reads, (
-        f"Marker {marker!r} not seen in sys_terminal_read output: "
-        f"{reads!r}. send/read flow broken."
-    )
-
-    # At least one list call must have returned a non-empty list
-    # (the pre-close one in step 4) and at least one must have
-    # returned an empty list (the post-close one in step 6).
-    # We don't pin which is which — LLM may make extra exploratory
-    # list calls — but both states must exist among the calls.
-    saw_running = False
-    saw_empty = False
-    for raw in lists:
-        try:
-            entries = json.loads(raw)
-        except json.JSONDecodeError:
-            continue
-        if isinstance(entries, list):
-            # ``sys_terminal_list`` entries expose ``session`` (the
-            # LLM-facing key), not ``session_key`` (the registry's
-            # internal field). Confirmed via real JSON output.
-            if any(isinstance(e, dict) and e.get("session") == "investigate" for e in entries):
-                saw_running = True
-            if entries == []:
-                saw_empty = True
-    assert saw_running, (
-        f"No sys_terminal_list call ever showed bash:investigate as "
-        f"a registered terminal. Lists: {lists!r}. Either the launch "
-        f"never registered (impossible — launches above non-empty), "
-        f"or list returns the wrong shape."
-    )
-    assert saw_empty, (
-        f"No sys_terminal_list call ever returned an empty list. "
-        f"Lists: {lists!r}. Either close didn't remove the entry "
-        f"(registry leak), or the LLM didn't call list after close."
-    )
-
-    # Close response must have status='closed' (not 'not_found').
-    # This catches the case where the LLM closed a different
-    # session than it launched.
-    close_results = [json.loads(r) for r in closes if r]
-    assert any(c.get("status") == "closed" for c in close_results), (
-        f"No sys_terminal_close returned status='closed'. Got: "
-        f"{close_results!r}. Either the LLM passed the wrong "
-        f"session_key, or close didn't find the registered entry "
-        f"(would be a registry-tooling bug)."
-    )
-
-
-def test_sys_terminal_send_keys_drives_interactive_e2e(
-    live_server: str,
-    sys_terminal_test_agent: str,
-    http_client: httpx.Client,
-) -> None:
-    """
-    Interactive driving — the load-bearing capability that
-    ``sys_terminal_*`` adds over ``sys_os_shell``. Start a Python
-    REPL inside the bash terminal, send ``print(2+2)``, assert
-    ``4`` appears in the pane.
-
-    A naive ``sys_os_shell`` ("python3 -c 'print(2+2)'") would
-    work too, but proves nothing about *interactive* state. The
-    test below requires the REPL to stay running across two
-    separate ``send`` calls, with the second ``send`` interpreted
-    by the live python process from the first.
-
-    What breaks if this fails:
-      - ``send_keys`` parsing regresses (Enter etc. mis-routed).
-      - The 50ms ``asyncio.sleep`` between text and keys collapses
-        and Enter fires before the text lands.
-      - The per-instance lock over-serializes such
-        that the python REPL never gets to read its own input.
-    """
-    prompt = (
-        "Use sys_terminal_launch to start the 'bash' terminal with "
-        "session 'pyrepl'. Then sys_terminal_send 'python3' followed "
-        "by Enter. Wait briefly. Then sys_terminal_send "
-        "'print(2+2)' followed by Enter. Wait briefly. Then "
-        "sys_terminal_read on session 'pyrepl'. Reply 'done' once "
-        "the read completes."
-    )
-    resp = http_client.post(
-        "/v1/responses",
-        json={
-            "model": sys_terminal_test_agent,
-            "input": prompt,
-            "stream": False,
-        },
-        timeout=180.0,
-    )
-    resp.raise_for_status()
-    body = poll_until_terminal(http_client, resp.json()["id"], timeout=180)
-    assert body["status"] == "completed", (
-        f"Workflow failed: {body.get('status')!r}, error={body.get('error')!r}"
-    )
-    conv_id = body["conversation"]["id"]
-
-    # Two distinct sends required — one for the REPL start, one
-    # for the print expression. If only one fired, the LLM
-    # short-circuited to a single sys_os_shell or merged the
-    # commands; the test no longer proves interactive driving.
-    sends = _get_function_call_outputs(http_client, conv_id, "sys_terminal_send")
-    assert len(sends) >= 2, (
-        f"Expected >=2 sys_terminal_send calls (python3 start + "
-        f"print(2+2)), got {len(sends)}. Sends: {sends!r}. The LLM "
-        f"may have collapsed both into a single send — test no "
-        f"longer exercises interactive driving."
-    )
-
-    reads = _get_function_call_outputs(http_client, conv_id, "sys_terminal_read")
-    assert len(reads) >= 1, f"sys_terminal_read never called; conv_id={conv_id}"
-    combined = " ".join(reads)
-
-    # The result of print(2+2) must show in the pane. If a Python
-    # REPL prompt (>>>) shows but no 4, the print was swallowed
-    # by the REPL's input buffer and never executed — points at
-    # the keys=Enter handling regressing.
-    assert "4" in combined, (
-        f"Python REPL output '4' missing from pane after "
-        f"print(2+2). Combined reads:\n{combined!r}\n"
-        f"If the pane shows '>>>' but no 4, the second send's "
-        f"Enter didn't reach python's stdin. If the pane shows "
-        f"nothing useful at all, python3 may not be on PATH in "
-        f"the tmux env."
-    )
-
-
 def test_sys_terminal_ten_parallel_dispatches_complete_e2e(
     live_server: str,
     sys_terminal_test_agent: str,

tests/integration/test_sys_terminal_round_trip.py

@@ -0,0 +1,591 @@
+"""Mock-LLM sessions coverage for re-homed sequential ``sys_terminal_*`` e2e.
+
+Re-homes three suppressed e2e tests that lived in
+``tests/e2e/test_sys_terminal_e2e.py`` and only stayed red because they
+drove the **removed** ``POST /v1/responses`` route (plus
+``poll_until_terminal``'s ``GET /v1/responses/{id}``). That route no
+longer exists under ``omnigent/server/routes/``; the cited
+"500 / runner availability" reason on issue 532 was a stale
+misdiagnosis. These replacements drive the current runner-bound
+sessions API in mock-LLM mode instead — the same path the merged D6
+re-homes (``test_d6_async_cancel_round_trip``,
+``test_d6_parallel_fan_out_round_trip``) use.
+
+Why this faithfully exercises the same surface
+-----------------------------------------------
+``sys_terminal_*`` are AP-side / server-executed tools (the runner's
+tool dispatcher runs ``TerminalRegistry`` → real tmux and threads the
+result back to the model), NOT client-side ``action_required`` tools.
+So a mock LLM scripted to emit ``sys_terminal_launch`` / ``send`` /
+``read`` / ``list`` / ``close`` calls actually drives real tmux on the
+server — no external client has to fulfill anything. The agent loop
+consumes exactly one queued mock response per model call, so a single
+user turn with a queue of ``[launch, send, read, ..., final_text]``
+executes the steps in strict sequential order (each tool result is
+posted back before the next queued response is consumed). That gives
+the same launch→send→read→list→close ordering the old real-LLM e2e
+relied on, without trusting an LLM to follow a prompt.
+
+Harness choice
+--------------
+Runs on the ``openai-agents`` harness — the default in mock mode (see
+``tests/integration/conftest.py::harness_name``). The mock LLM speaks
+the OpenAI ``/v1/responses`` SSE shape that harness consumes. Terminal
+tool execution is harness-agnostic (it is a runner-side registry
+lookup + tmux spawn), so the on-disk ``sys-terminal-test`` claude-sdk
+agent is not usable in mock mode; instead each test registers a minimal
+inline ``openai-agents`` agent carrying the same ``terminals:`` block
+via ``register_inline_agent(extra_config=...)``. This mirrors
+``test_d6_parallel_fan_out_round_trip.py::terminal_mock_agent``.
+
+Coverage deltas vs. the old e2e (honest notes)
+----------------------------------------------
+* The old tests proved a *real LLM* chose to call the tools from a
+  natural-language prompt. The mock layer scripts the calls, so it does
+  NOT prove prompt-following / tool-selection — only that the
+  server-executed terminal plumbing round-trips correctly. The
+  registration→dispatch→tmux→result path, the load-bearing behavior, is
+  exercised identically.
+* Markers are produced by the shell itself (``echo`` output captured by
+  ``sys_terminal_read``), so a passing assertion still proves data flowed
+  send→tmux→read, exactly as before.
+
+Runs in mock mode (no ``--llm-api-key``); the ``tests/integration``
+package gate is lifted in mock mode by
+``tests/integration/conftest.py``. Skipped when tmux is not installed.
+"""
+
+from __future__ import annotations
+
+import json
+import shutil
+import uuid
+from typing import Any
+
+import httpx
+import pytest
+
+from tests.e2e.conftest import (
+    configure_mock_llm,
+    create_runner_bound_session,
+    poll_session_until_terminal,
+    register_inline_agent,
+    send_user_message_to_session,
+)
+
+# The mock LLM is scripted with a fixed tool-call sequence (queues keyed
+# by the agent model name), so these tests cannot run against a real LLM
+# — it would 401 on the mock base URL and could never reproduce the
+# scripted call_ids/markers. The central gate in
+# ``tests/integration/conftest.py`` skips ``mock_only`` tests when a real
+# ``--llm-api-key`` is supplied (the real-LLM Integration jobs). This
+# replaces the dead ``if mock_llm_server_url is None: skip`` guards: that
+# fixture is always started regardless of --llm-api-key, so the guard
+# never fired in any job.
+#
+# Cross-test queue isolation is handled centrally too: the autouse
+# ``_reset_mock_llm_between_tests`` fixture in
+# ``tests/integration/conftest.py`` resets the shared mock-LLM server
+# before and after every integration test, so this file no longer needs
+# its own per-file reset fixture.
+pytestmark = pytest.mark.mock_only
+
+
+def _list_session_items(client: httpx.Client, session_id: str) -> list[dict[str, Any]]:
+    """Return all persisted items for a session in one paginated snapshot.
+
+    :param client: HTTP client pointed at the live server.
+    :param session_id: Session/conversation id.
+    """
+    items: list[dict[str, Any]] = []
+    after: str | None = None
+    while True:
+        params: dict[str, Any] = {"order": "asc", "limit": 1000}
+        if after is not None:
+            params["after"] = after
+        resp = client.get(f"/v1/sessions/{session_id}/items", params=params)
+        resp.raise_for_status()
+        page = resp.json()
+        items.extend(page["data"])
+        if not page.get("has_more"):
+            return items
+        after = page.get("last_id")
+        if after is None:
+            raise AssertionError(f"items page had has_more without last_id: {page}")
+
+
+def _flat_item(item: dict[str, Any]) -> dict[str, Any]:
+    """Flatten a session item's ``data`` block into the Responses-style shape.
+
+    ``GET /v1/sessions/{id}/items`` serializes conversation items with
+    type-specific fields nested under ``data``; the e2e assertions
+    historically read ``name`` / ``call_id`` / ``output`` as top-level
+    fields. Flatten so the same accessors work here.
+
+    :param item: A raw conversation item dict.
+    """
+    data = item.get("data")
+    if not isinstance(data, dict):
+        return item
+    return {
+        "id": item.get("id"),
+        "response_id": item.get("response_id"),
+        "type": item.get("type"),
+        "status": item.get("status"),
+        **data,
+    }
+
+
+def _function_call_outputs_for(
+    client: httpx.Client,
+    *,
+    session_id: str,
+    tool_name: str,
+) -> list[str]:
+    """Return raw outputs of every *tool_name* call in conversation order.
+
+    Walks the persisted ``function_call`` / ``function_call_output``
+    items so assertions land on deterministic tool output strings, not
+    on flaky model prose. Mirrors the old e2e's
+    ``_get_function_call_outputs``.
+
+    :param client: HTTP client pointed at the live server.
+    :param session_id: Session/conversation id.
+    :param tool_name: Only outputs of calls to this tool are returned.
+    :returns: Ordered list of raw output strings.
+    """
+    items = [_flat_item(item) for item in _list_session_items(client, session_id)]
+    call_ids = {
+        item["call_id"]
+        for item in items
+        if item.get("type") == "function_call"
+        and item.get("name") == tool_name
+        and item.get("call_id")
+    }
+    return [
+        str(item.get("output") or "")
+        for item in items
+        if item.get("type") == "function_call_output" and item.get("call_id") in call_ids
+    ]
+
+
+def _register_terminal_agent(
+    http_client: httpx.Client,
+    *,
+    live_runner_id: str,
+    harness_name: str,
+    model_name: str,
+    request: pytest.FixtureRequest,
+    mock_llm_server_url: str,
+) -> str:
+    """Register a minimal inline agent with the ``sys_terminal_*`` tools and bind a session.
+
+    Carries the same ``terminals: {bash: ...}`` block the on-disk
+    ``sys-terminal-test`` agent declares, threaded through the compat
+    translator via ``register_inline_agent(extra_config=...)`` so the
+    five ``sys_terminal_*`` tools register on the AP-side ToolManager.
+
+    :returns: The runner-bound session id.
+    """
+    agent_name = register_inline_agent(
+        http_client,
+        name=f"terminal-seq-{uuid.uuid4().hex[:6]}",
+        harness=harness_name,
+        model=model_name,
+        profile=request.config.getoption("--profile"),
+        prompt=(
+            "You are a terminal test assistant. Follow the scripted mock LLM tool calls exactly."
+        ),
+        mock_llm_base_url=f"{mock_llm_server_url}/v1",
+        extra_config={
+            "os_env": {
+                "type": "caller_process",
+                "cwd": ".",
+                "sandbox": {"type": "none"},
+            },
+            "terminals": {
+                "bash": {
+                    "command": "bash",
+                    "os_env": {
+                        "type": "caller_process",
+                        "sandbox": {"type": "none"},
+                    },
+                },
+            },
+        },
+    )
+    return create_runner_bound_session(
+        http_client,
+        agent_name=agent_name,
+        runner_id=live_runner_id,
+    )
+
+
+@pytest.fixture
+def terminal_session(
+    http_client: httpx.Client,
+    live_runner_id: str,
+    harness_name: str,
+    model_name: str,
+    request: pytest.FixtureRequest,
+    mock_llm_server_url: str | None,
+) -> tuple[str, str]:
+    """A fresh runner-bound session on an inline agent with terminals enabled.
+
+    :returns: ``(session_id, model_name)`` — the model name keys the
+        mock LLM response queue.
+    """
+    if shutil.which("tmux") is None:
+        pytest.skip("tmux not installed; sys_terminal_* tests need tmux on PATH")
+    session_id = _register_terminal_agent(
+        http_client,
+        live_runner_id=live_runner_id,
+        harness_name=harness_name,
+        model_name=model_name,
+        request=request,
+        mock_llm_server_url=mock_llm_server_url,
+    )
+    return session_id, model_name
+
+
+def test_sys_terminal_basic_round_trip(
+    live_server: str,
+    http_client: httpx.Client,
+    terminal_session: tuple[str, str],
+    mock_llm_server_url: str | None,
+) -> None:
+    """Launch → send (echo a marker) → read → the marker comes back.
+
+    Re-homes ``test_sys_terminal_basic_round_trip_e2e``. The mock LLM
+    is scripted with one tool call per loop step; the agent executes
+    each server-side against real tmux and the turn ends on a text
+    response. Asserts:
+
+    * the turn completes (the runner accepted and round-tripped every
+      AP-side terminal result),
+    * the first launch reports ``status="launched"``,
+    * the unique marker echoed by ``echo`` appears in a
+      ``sys_terminal_read`` capture — proving send reached tmux and
+      read saw the output.
+
+    Two reads are scripted purely for timing resilience: ``echo``
+    output lands in the pane asynchronously after Enter, so a single
+    capture can occasionally race ahead of it. The marker need only
+    appear in one.
+    """
+    session_id, model_name = terminal_session
+    marker = f"TERMINAL_RT_MARKER_{uuid.uuid4().hex[:8]}"
+
+    def _call(name: str, args: dict[str, Any]) -> dict[str, Any]:
+        return {
+            "tool_calls": [
+                {
+                    "call_id": f"call_{name}_{uuid.uuid4().hex[:6]}",
+                    "name": name,
+                    "arguments": json.dumps(args),
+                }
+            ]
+        }
+
+    configure_mock_llm(
+        mock_llm_server_url,
+        [
+            _call("sys_terminal_launch", {"terminal": "bash", "session": "s1"}),
+            _call(
+                "sys_terminal_send",
+                {"terminal": "bash", "session": "s1", "text": f"echo {marker}", "keys": "Enter"},
+            ),
+            _call("sys_terminal_read", {"terminal": "bash", "session": "s1"}),
+            _call("sys_terminal_read", {"terminal": "bash", "session": "s1"}),
+            {"text": "done"},
+        ],
+        key=model_name,
+    )
+
+    response_id = send_user_message_to_session(
+        http_client,
+        session_id=session_id,
+        content="Launch bash s1, echo the marker, read it back, then say done.",
+    )
+    result = poll_session_until_terminal(
+        http_client,
+        session_id=session_id,
+        response_id=response_id,
+        timeout=180,
+    )
+    assert result["status"] == "completed", (
+        f"terminal round-trip turn should complete; got {result['status']!r}, "
+        f"error={result.get('error')!r}, output={str(result.get('output'))[:600]!r}"
+    )
+
+    launches = _function_call_outputs_for(
+        http_client, session_id=session_id, tool_name="sys_terminal_launch"
+    )
+    assert len(launches) >= 1, (
+        f"sys_terminal_launch produced no output; session_id={session_id}. "
+        f"If 0, the terminals block never registered the AP-side tools."
+    )
+    launch_result = json.loads(launches[0])
+    assert launch_result.get("status") == "launched", (
+        f"first launch should report status='launched'; got {launch_result!r}"
+    )
+
+    reads = _function_call_outputs_for(
+        http_client, session_id=session_id, tool_name="sys_terminal_read"
+    )
+    assert len(reads) >= 1, f"sys_terminal_read produced no output; session_id={session_id}"
+    combined = " ".join(reads)
+    assert marker in combined, (
+        f"echo marker {marker!r} not seen in any sys_terminal_read output. "
+        f"Reads: {reads!r}. If empty the send didn't reach tmux; if a prompt "
+        f"shows but not the echo, the command failed in tmux."
+    )
+
+
+def test_sys_terminal_full_workflow(
+    live_server: str,
+    http_client: httpx.Client,
+    terminal_session: tuple[str, str],
+    mock_llm_server_url: str | None,
+) -> None:
+    """All five ``sys_terminal_*`` tools, in one ordered sequence.
+
+    Re-homes ``test_sys_terminal_full_workflow_e2e``:
+    launch → send (echo a marker) → read → list → close → list. Asserts
+    state at each step:
+
+    * all five tools produced output (so list/close, which the focused
+      round-trip test never exercises, have coverage here),
+    * the echoed marker is captured by read,
+    * a ``list`` call shows ``bash:investigate`` running (pre-close),
+      and a later ``list`` no longer shows it (post-close) — the only
+      check that ``close`` removed the registry entry, not just killed
+      the process,
+    * ``close`` returned ``status="closed"`` (not ``not_found``).
+
+    Because the agent loop consumes the scripted calls in strict order,
+    the pre-close list provably runs before close and the post-close
+    list after — no reliance on an LLM to order the steps.
+    """
+    session_id, model_name = terminal_session
+    marker = f"FULL_WORKFLOW_MARKER_{uuid.uuid4().hex[:8]}"
+    term = {"terminal": "bash", "session": "investigate"}
+
+    def _call(name: str, args: dict[str, Any]) -> dict[str, Any]:
+        return {
+            "tool_calls": [
+                {
+                    "call_id": f"call_{name}_{uuid.uuid4().hex[:6]}",
+                    "name": name,
+                    "arguments": json.dumps(args),
+                }
+            ]
+        }
+
+    configure_mock_llm(
+        mock_llm_server_url,
+        [
+            _call("sys_terminal_launch", term),
+            _call("sys_terminal_send", {**term, "text": f"echo {marker}", "keys": "Enter"}),
+            _call("sys_terminal_read", term),
+            _call("sys_terminal_read", term),
+            _call("sys_terminal_list", {}),
+            _call("sys_terminal_close", term),
+            _call("sys_terminal_list", {}),
+            {"text": "done"},
+        ],
+        key=model_name,
+    )
+
+    response_id = send_user_message_to_session(
+        http_client,
+        session_id=session_id,
+        content="Run the full terminal workflow, then say done.",
+    )
+    result = poll_session_until_terminal(
+        http_client,
+        session_id=session_id,
+        response_id=response_id,
+        timeout=180,
+    )
+    assert result["status"] == "completed", (
+        f"full-workflow turn should complete; got {result['status']!r}, "
+        f"error={result.get('error')!r}, output={str(result.get('output'))[:600]!r}"
+    )
+
+    launches = _function_call_outputs_for(
+        http_client, session_id=session_id, tool_name="sys_terminal_launch"
+    )
+    sends = _function_call_outputs_for(
+        http_client, session_id=session_id, tool_name="sys_terminal_send"
+    )
+    reads = _function_call_outputs_for(
+        http_client, session_id=session_id, tool_name="sys_terminal_read"
+    )
+    lists = _function_call_outputs_for(
+        http_client, session_id=session_id, tool_name="sys_terminal_list"
+    )
+    closes = _function_call_outputs_for(
+        http_client, session_id=session_id, tool_name="sys_terminal_close"
+    )
+
+    missing = [
+        name
+        for name, calls in [
+            ("sys_terminal_launch", launches),
+            ("sys_terminal_send", sends),
+            ("sys_terminal_read", reads),
+            ("sys_terminal_list", lists),
+            ("sys_terminal_close", closes),
+        ]
+        if not calls
+    ]
+    assert not missing, (
+        f"these terminal tools produced no output: {missing!r}. The "
+        f"full-workflow test requires all five; list/close have no other "
+        f"server-executed coverage at this layer."
+    )
+
+    combined_reads = " ".join(reads)
+    assert marker in combined_reads, (
+        f"marker {marker!r} not seen in sys_terminal_read output: {reads!r}. "
+        f"send/read flow broken."
+    )
+
+    # One list (pre-close) must show bash:investigate running; one
+    # (post-close) must not. The list entries expose ``session`` (the
+    # LLM-facing key), not the registry's internal field.
+    saw_running = False
+    saw_gone = False
+    for raw in lists:
+        try:
+            entries = json.loads(raw)
+        except json.JSONDecodeError:
+            continue
+        if not isinstance(entries, list):
+            continue
+        has_investigate = any(
+            isinstance(e, dict) and e.get("session") == "investigate" for e in entries
+        )
+        if has_investigate:
+            saw_running = True
+        else:
+            saw_gone = True
+    assert saw_running, (
+        f"no sys_terminal_list call showed bash:investigate as running. "
+        f"Lists: {lists!r}. Either launch never registered or list returns "
+        f"the wrong shape."
+    )
+    assert saw_gone, (
+        f"no sys_terminal_list call ran with bash:investigate absent. "
+        f"Lists: {lists!r}. Either close didn't remove the registry entry "
+        f"(leak) or the post-close list never ran."
+    )
+
+    close_results = [json.loads(r) for r in closes if r]
+    assert any(c.get("status") == "closed" for c in close_results), (
+        f"no sys_terminal_close returned status='closed'. Got: {close_results!r}. "
+        f"close didn't find the registered entry."
+    )
+
+
+def test_sys_terminal_send_keys_drives_interactive(
+    live_server: str,
+    http_client: httpx.Client,
+    terminal_session: tuple[str, str],
+    mock_llm_server_url: str | None,
+) -> None:
+    """Two distinct sends drive an interactive Python REPL across calls.
+
+    Re-homes ``test_sys_terminal_send_keys_drives_interactive_e2e`` —
+    the load-bearing capability ``sys_terminal_*`` adds over a one-shot
+    ``sys_os_shell``: a process that stays alive across two separate
+    ``send`` calls, with the second send interpreted by the live process
+    from the first. The script:
+
+    1. launch bash:pyrepl,
+    2. send ``python3`` + Enter (start the REPL),
+    3. send ``print(2+2)`` + Enter (the second send the REPL must
+       interpret),
+    4. read (x3 for timing resilience — see below).
+
+    Asserts ``>= 2`` sends fired (a single merged send would not prove
+    interactive driving) and that ``4`` — the REPL's evaluation of the
+    *second* send — appears in a read capture. The two sends crossing
+    Enter-key handling and the REPL surviving between them is what
+    distinguishes this from a stateless shell command.
+
+    Honest coverage delta vs. the old e2e: identical load-bearing
+    assertions (``>=2`` sends + ``4`` in the pane). The only difference
+    is the calls are scripted rather than chosen by a real LLM, so this
+    does not prove a model would *decide* to drive the REPL with two
+    sends — only that the server-side send/Enter/REPL plumbing works
+    when it does. Three reads (extra model round-trips) give the cold
+    ``python3`` interpreter time to boot and evaluate before capture;
+    the old e2e leaned on real-LLM "wait briefly" pauses for the same
+    reason.
+    """
+    session_id, model_name = terminal_session
+    term = {"terminal": "bash", "session": "pyrepl"}
+
+    def _call(name: str, args: dict[str, Any]) -> dict[str, Any]:
+        return {
+            "tool_calls": [
+                {
+                    "call_id": f"call_{name}_{uuid.uuid4().hex[:6]}",
+                    "name": name,
+                    "arguments": json.dumps(args),
+                }
+            ]
+        }
+
+    configure_mock_llm(
+        mock_llm_server_url,
+        [
+            _call("sys_terminal_launch", term),
+            _call("sys_terminal_send", {**term, "text": "python3", "keys": "Enter"}),
+            _call("sys_terminal_send", {**term, "text": "print(2+2)", "keys": "Enter"}),
+            _call("sys_terminal_read", term),
+            _call("sys_terminal_read", term),
+            _call("sys_terminal_read", term),
+            {"text": "done"},
+        ],
+        key=model_name,
+    )
+
+    response_id = send_user_message_to_session(
+        http_client,
+        session_id=session_id,
+        content="Start python3, print(2+2), read the pane, then say done.",
+    )
+    result = poll_session_until_terminal(
+        http_client,
+        session_id=session_id,
+        response_id=response_id,
+        timeout=180,
+    )
+    assert result["status"] == "completed", (
+        f"interactive turn should complete; got {result['status']!r}, "
+        f"error={result.get('error')!r}, output={str(result.get('output'))[:600]!r}"
+    )
+
+    sends = _function_call_outputs_for(
+        http_client, session_id=session_id, tool_name="sys_terminal_send"
+    )
+    assert len(sends) >= 2, (
+        f"expected >= 2 sys_terminal_send calls (python3 start + print(2+2)), "
+        f"got {len(sends)}: {sends!r}. Fewer means the interactive driving was "
+        f"not exercised."
+    )
+
+    reads = _function_call_outputs_for(
+        http_client, session_id=session_id, tool_name="sys_terminal_read"
+    )
+    assert len(reads) >= 1, f"sys_terminal_read produced no output; session_id={session_id}"
+    combined = " ".join(reads)
+    assert "4" in combined, (
+        f"Python REPL output '4' missing after print(2+2). Combined reads:\n"
+        f"{combined!r}\nIf the pane shows '>>>' but no 4, the second send's "
+        f"Enter didn't reach python's stdin; if nothing useful shows, python3 "
+        f"may not be on PATH in the tmux env."
+    )

tests/known_failures.yaml

@@ -78,12 +78,6 @@ skips:
     cluster: repl-pexpect-cli
     mode: skip
 
-  - id: tests/e2e/test_sys_terminal_e2e.py::test_sys_terminal_send_keys_drives_interactive_e2e
-    reason: "Drives the removed /v1/responses route; re-home to the sessions API in a later batch."
-    issue: 532
-    cluster: terminal-d6
-    mode: skip
-
   - id: tests/e2e/test_async_tools_e2e.py::test_async_tool_real_llm_e2e
     reason: "Shard 0 bulk: async tool real-LLM assertion failure. Triage under #532."
     issue: 532
@@ -260,12 +254,6 @@ skips:
     cluster: run-ap-examples-harness
     mode: skip
 
-  - id: tests/e2e/test_sys_terminal_e2e.py::test_sys_terminal_basic_round_trip_e2e
-    reason: "Shard 3 bulk: sys-terminal basic round-trip. Same sys-terminal family."
-    issue: 532
-    cluster: terminal-d6
-    mode: skip
-
   - id: tests/e2e/omnigent/test_run_omnigent_example_agents.py::test_run_omnigent_example_yaml[agent_with_subagent_session]
     reason: "Shard 3 bulk: run_omnigent example-yaml agent_with_subagent_session."
     issue: 532
@@ -339,12 +327,6 @@ skips:
     cluster: async-dispatch-inbox-sse
     mode: skip
 
-  - id: tests/e2e/test_sys_terminal_e2e.py::test_sys_terminal_full_workflow_e2e
-    reason: "Follow-up: sys-terminal full workflow. sys-terminal cluster."
-    issue: 532
-    cluster: terminal-d6
-    mode: skip
-
   - id: tests/e2e/omnigent/test_repl_session_lifecycle.py::test_repl_effort_command_persists_session_metadata
     reason: "Nightly bulk: REPL session-lifecycle / pexpect cluster."
     issue: 532