fix(ego-source-reader): preserve native control stops when CLI output is discarded

docs/integrations/ego-source-reader.md

@@ -95,7 +95,14 @@ navigating or capturing it. Once Ego returns Agent control, the next call uses
 the same space automatically; no extra chat confirmation is needed. This tool
 does not force control, poll in the background or grant approval for protected
 actions. Browser failures and rendering timeouts alone are not requests for
-human handoff. A source's actual login or verification requirement remains a
+human handoff. If control changes after inventory, Ego's native boundary still
+stops the operation. Its hard-stop output sink can discard console sentinels;
+the adapter preserves only the stable user-control/inactive error code in a
+private temporary file and returns `source_reader_not_agent_owned`. It discards
+partial output, makes no replacement or takeover attempt, and removes that
+file after the call. Ordinary failures remain ordinary failures. This error
+transport does not grant control or prove a live human handoff was tested.
+A source's actual login or verification requirement remains a
 separate boundary; follow the installed Ego skill and continue independent work.
 
 On normal MCP shutdown or SIGTERM, it finishes only its own created,

loopx/extensions/ego_source_reader.py

@@ -187,11 +187,30 @@ def _run(executable: str, script: str, *, deadline: float | None = None) -> subp
     timeout = TIMEOUT_SECONDS if deadline is None else deadline - time.monotonic()
     if timeout <= 0:
         raise subprocess.TimeoutExpired(executable, TIMEOUT_SECONDS)
-    return subprocess.run(
-        [executable, "nodejs", "-e", script],
-        stdin=subprocess.DEVNULL, capture_output=True, text=True, encoding="utf-8",
-        timeout=timeout, check=False,
-    )
+    # Ego discards console output on a native control stop, including our
+    # sentinel. Preserve only its stable code outside that buffered channel.
+    with tempfile.TemporaryDirectory(prefix="loopx-source-control-") as directory:
+        stop = Path(directory) / "stop.txt"
+        wrapped = (
+            "try{" + script + "}catch(e){"
+            "if(['EGO_TASK_SPACE_USER_IN_CONTROL','EGO_TASK_SPACE_INACTIVE']"
+            ".includes(e?.error_code)){"
+            f"(await import('node:fs')).writeFileSync({json.dumps(str(stop))},e.error_code);"
+            "}throw e;}"
+        )
+        result = subprocess.run(
+            [executable, "nodejs", "-e", wrapped],
+            stdin=subprocess.DEVNULL, capture_output=True, text=True, encoding="utf-8",
+            timeout=timeout, check=False,
+        )
+        if stop.exists() and stop.read_text(encoding="utf-8") in {
+            "EGO_TASK_SPACE_USER_IN_CONTROL", "EGO_TASK_SPACE_INACTIVE",
+        }:
+            return subprocess.CompletedProcess(
+                result.args, 1,
+                MARKER + '{"error":"source_reader_not_agent_owned"}', "",
+            )
+        return result
 
 
 def _navigation(config: ReaderConfig, url: str, *, image: bool = False) -> str:

tests/test_ego_source_reader.py

@@ -404,6 +404,44 @@ def run(_executable, script, **_kwargs):
     assert observations == [["resolve", "finish"] if ownership == "agent" else []]
 
 
+@pytest.mark.parametrize("code,expected", [
+    ("EGO_TASK_SPACE_USER_IN_CONTROL", "source_reader_not_agent_owned"),
+    ("EGO_TASK_SPACE_INACTIVE", "source_reader_not_agent_owned"),
+    ("EGO_OPERATION_FAILED", "browser_read_failed"),
+])
+@pytest.mark.parametrize("stage", ["resolve", "navigate", "extract"])
+@pytest.mark.parametrize("image", [False, True])
+def test_native_control_race_survives_discarded_cli_output(
+    configured, monkeypatch, code, expected, stage, image,
+):
+    # Inventory still says Agent, but a later SDK boundary denies control.
+    # Run the production script in a real child. Model Ego's hard-stop sink,
+    # which discards all result markers and does not print error_code.
+    real_run = subprocess.run
+    trace = []
+    def run(args, **kwargs):
+        harness = (
+            "const code=" + json.dumps(code) + ";const stage=" + json.dumps(stage) + ";"
+            "function denied(){const e=Error('native wording changed');e.error_code=code;throw e;}"
+            "async function listTaskSpaces(){return [{id:7,ownership:'agent'}];}"
+            "async function taskSpace(){if(stage==='resolve')denied();return {page(){return {"
+            "async goto(){if(stage==='navigate')denied();},async waitForFunction(){},"
+            "async evaluate(){denied();},async screenshot(){throw Error('capture forbidden');}};}};}"
+            "(async()=>{" + args[3] + "})().catch(()=>{process.exitCode=1;});"
+        )
+        child = real_run(["node", "-e", harness], **kwargs)
+        trace.append(child.returncode)
+        # Neither code nor an extraction sentinel survives the real SDK sink.
+        return subprocess.CompletedProcess(args, child.returncode, "", "SDK hard stop guidance")
+    monkeypatch.setattr(reader.subprocess, "run", run)
+    if image:
+        result = json.loads(reader.read_public_image(URL, 0)[0].text)
+    else:
+        result = reader.read_public_url(URL)
+    assert result["ok"] is False and result["error"] == expected
+    assert trace == [1]  # No retry, new space, claim, takeover, or capture.
+
+
 def test_creation_and_read_share_one_timeout_budget(configured, monkeypatch):
     auto_config(monkeypatch)
     clock = iter([0, 5, 29])