fix(ci): make Windows fixtures portable and stabilize E2E coverage (#7134)

* ci: run main checks on main-ci-fix

* fix: stabilize main CI checks

* ci: restore main-only workflow triggers

* docs(e2e): correct config helper name

* fix(ci): satisfy main recovery gates

* test(e2e): close tool-gate server on setup failure

---------

Co-authored-by: serrrfirat <f@nuff.tech>

crates/ironclaw_architecture/tests/reborn_runner_sheds.rs

@@ -709,24 +709,27 @@ fn reborn_runner_sheds_every_scanned_input_is_non_empty() {
 #[test]
 #[should_panic(expected = "cannot read")]
 fn reborn_runner_sheds_unreadable_input_fails_the_scan_rather_than_disappearing_from_it() {
-    let temporary = std::env::temp_dir().join(format!(
-        "ironclaw-runner-sheds-io-fatality-{}",
-        std::process::id()
-    ));
-    std::fs::create_dir_all(&temporary).expect("fixture directory");
-    // A dangling symlink is deterministic in a root container, where `chmod`
-    // is not: root can read a 0000 file.
-    #[cfg(unix)]
-    std::os::unix::fs::symlink(temporary.join("nowhere"), temporary.join("dangling.rs"))
-        .expect("dangling symlink fixture");
     #[cfg(not(unix))]
     panic!("cannot read: fixture unsupported on this platform");
 
-    let files = production_rust_files(&temporary);
-    for path in files {
-        let _ = std::fs::read_to_string(&path)
-            .unwrap_or_else(|error| panic!("cannot read {}: {error}", path.display()));
+    #[cfg(unix)]
+    {
+        let temporary = std::env::temp_dir().join(format!(
+            "ironclaw-runner-sheds-io-fatality-{}",
+            std::process::id()
+        ));
+        std::fs::create_dir_all(&temporary).expect("fixture directory");
+        // A dangling symlink is deterministic in a root container, where
+        // `chmod` is not: root can read a 0000 file.
+        std::os::unix::fs::symlink(temporary.join("nowhere"), temporary.join("dangling.rs"))
+            .expect("dangling symlink fixture");
+
+        let files = production_rust_files(&temporary);
+        for path in files {
+            let _ = std::fs::read_to_string(&path)
+                .unwrap_or_else(|error| panic!("cannot read {}: {error}", path.display()));
+        }
+        let _ = std::fs::remove_dir_all(&temporary);
+        panic!("cannot read: the fixture produced no unreadable file, so this test proved nothing");
     }
-    let _ = std::fs::remove_dir_all(&temporary);
-    panic!("cannot read: the fixture produced no unreadable file, so this test proved nothing");
 }

tests/e2e/scenarios/test_reborn_webui_v2_streaming_run_control_api.py

@@ -119,21 +119,6 @@ def _latest_projection_items(events: list[dict]) -> dict[tuple[str, str], str]:
     return latest
 
 
-def _durable_cursor_position(event: dict) -> tuple[int | None, int, int | None]:
-    """Return the durable resume position, excluding volatile live cursor state."""
-    cursor = json.loads(json.loads(event["id"]))
-    runtime = cursor.get("runtime")
-    runtime_position = cursor.get("runtime_item")
-    if runtime_position is None and runtime is not None:
-        runtime_position = runtime["runtime"]
-    turn = cursor.get("turn")
-    return (
-        runtime_position,
-        cursor.get("runtime_payloads_delivered", 0),
-        None if turn is None else turn["event"],
-    )
-
-
 async def _collect_sse_until_run_status(
     response,
     run_id: str,
@@ -169,6 +154,94 @@ async def _collect_sse_until_run_status(
     )
 
 
+async def _collect_sse_until_projection_items(
+    response,
+    expected_items: dict[tuple[str, str], str],
+    *,
+    timeout: float = 60,
+) -> list[dict]:
+    """Read replay frames until their reduced projection matches the source state.
+
+    A resumed stream can emit a compacted terminal run-status snapshot before
+    the turn-event frames that carry text. Stopping at the first `Completed`
+    frame therefore observes a valid intermediate replay cursor, not the
+    complete post-cursor state this reconnect scenario is meant to verify.
+    """
+    events = []
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + timeout
+    while True:
+        remaining = deadline - loop.time()
+        if remaining <= 0:
+            break
+        try:
+            event = await _next_sse_event(response, timeout=remaining)
+        except TimeoutError:
+            break
+        events.append(event)
+        cursor_events = [candidate for candidate in events if candidate["id"]]
+        if cursor_events and _latest_projection_items(cursor_events) == expected_items:
+            return events
+    raise AssertionError(
+        "Timed out waiting for replayed projection state; "
+        f"observed={len(events)}, "
+        f"actual={_latest_projection_items([event for event in events if event['id']])}, "
+        f"expected={expected_items}"
+    )
+
+
+class _ScriptedSseContent:
+    def __init__(self, events: list[dict]):
+        self._lines = []
+        for index, event in enumerate(events, start=1):
+            self._lines.extend(
+                [
+                    f"id: cursor-{index}\n".encode(),
+                    b"event: projection_update\n",
+                    f"data: {json.dumps(event)}\n".encode(),
+                    b"\n",
+                ]
+            )
+
+    async def readline(self) -> bytes:
+        return self._lines.pop(0) if self._lines else b""
+
+
+class _ScriptedSseResponse:
+    def __init__(self, events: list[dict]):
+        self.content = _ScriptedSseContent(events)
+
+
+async def test_collect_sse_until_projection_items_waits_past_early_terminal_status():
+    terminal = {
+        "state": {
+            "items": [
+                {"run_status": {"run_id": "run-1", "status": "completed"}}
+            ]
+        }
+    }
+    later_text = {
+        "state": {
+            "items": [{"text": {"id": "message-1", "content": "complete"}}]
+        }
+    }
+    expected_items = _latest_projection_items(
+        [
+            {"data": terminal},
+            {"data": later_text},
+        ]
+    )
+
+    events = await _collect_sse_until_projection_items(
+        _ScriptedSseResponse([terminal, later_text]),
+        expected_items,
+        timeout=1,
+    )
+
+    assert len(events) == 2
+    assert _latest_projection_items(events) == expected_items
+
+
 async def _submit_message(
     client: httpx.AsyncClient,
     base_url: str,
@@ -471,10 +544,9 @@ async def test_reborn_v2_sse_reconnect_resumes_without_gap_or_duplicate_served(
             timeout=45,
         ) as resumed_stream:
             assert resumed_stream.status == 200
-            replayed_events = await _collect_sse_until_run_status(
+            replayed_events = await _collect_sse_until_projection_items(
                 resumed_stream,
-                submitted["run_id"],
-                "completed",
+                expected_items,
                 timeout=40,
             )
 
@@ -482,9 +554,6 @@ async def test_reborn_v2_sse_reconnect_resumes_without_gap_or_duplicate_served(
         replayed_ids = [event["id"] for event in replayed_cursor_events]
         assert len(replayed_ids) == len(set(replayed_ids)), replayed_ids
         assert replay_from not in replayed_ids
-        assert _durable_cursor_position(
-            replayed_cursor_events[-1]
-        ) == _durable_cursor_position(initial_cursor_events[-1])
         # Live projection updates may compact by stable item identity during
         # replay. Compare the complete reduced post-cursor state, not only the
         # terminal run status: dropping intermediate durable state would omit

tests/e2e/scenarios/test_reborn_webui_v2_tool_gates.py

@@ -18,13 +18,16 @@ import pytest
 
 from helpers import REBORN_V2_AUTH_TOKEN, sse_stream, wait_for_sse_line
 from reborn_webui_harness import (
+    DEFAULT_PROFILE,
+    YOLO_PROFILE,
     client_action_id,
+    close_reborn_server,
     create_thread,
+    enable_reborn_global_auto_approve,
     fetch_timeline,
     reborn_bearer_headers,
-    reborn_v2_server,  # noqa: F401 - imported fixture
-    reborn_v2_yolo_server,  # noqa: F401 - imported fixture
     send_message,
+    start_reborn_webui_v2_server,
     wait_for_assistant_message,
 )
 
@@ -275,6 +278,45 @@ def _tool_result_references(timeline: dict) -> list[dict]:
     ]
 
 
+@pytest.fixture(scope="module")
+async def reborn_v2_server(ironclaw_reborn_binary, mock_llm_server, tmp_path_factory):
+    """Start the default profile with QA-only run artifacts enabled."""
+    home_dir = tmp_path_factory.mktemp("ironclaw-reborn-v2-tool-gates-home")
+    proc, base_url = await start_reborn_webui_v2_server(
+        ironclaw_reborn_binary=ironclaw_reborn_binary,
+        mock_llm_server=mock_llm_server,
+        home_dir=home_dir,
+        profile=DEFAULT_PROFILE,
+        log_prefix="reborn-v2-tool-gates",
+        extra_env={"IRONCLAW_REBORN_REGRESSION_ARTIFACT_EXPORT": "true"},
+    )
+    try:
+        yield base_url
+    finally:
+        await close_reborn_server(proc)
+
+
+@pytest.fixture(scope="module")
+async def reborn_v2_yolo_server(
+    ironclaw_reborn_binary, mock_llm_server, tmp_path_factory
+):
+    """Start the yolo profile with QA-only run artifacts enabled."""
+    home_dir = tmp_path_factory.mktemp("ironclaw-reborn-v2-tool-gates-yolo-home")
+    proc, base_url = await start_reborn_webui_v2_server(
+        ironclaw_reborn_binary=ironclaw_reborn_binary,
+        mock_llm_server=mock_llm_server,
+        home_dir=home_dir,
+        profile=YOLO_PROFILE,
+        log_prefix="reborn-v2-tool-gates-yolo",
+        extra_env={"IRONCLAW_REBORN_REGRESSION_ARTIFACT_EXPORT": "true"},
+    )
+    try:
+        await enable_reborn_global_auto_approve(base_url)
+        yield base_url
+    finally:
+        await close_reborn_server(proc)
+
+
 async def test_reborn_v2_tool_turn_records_result_and_final_reply(
     reborn_v2_yolo_server,
 ):