diff --git a/overlay/scripts/agenttui.py b/overlay/scripts/agenttui.py index 6c937b6..1a45da1 100644 --- a/overlay/scripts/agenttui.py +++ b/overlay/scripts/agenttui.py @@ -581,6 +581,31 @@ def available(self) -> Capability: """Is this transport's own tooling usable at all in this process?""" raise NotImplementedError + def self_pane_ref(self) -> dict[str, str] | None: + """This process's **own** pane, as reported by the multiplexer. None = unknown. + + Why this exists at all: §5.0's root rule refuses a write whose + self-identification evidence does not *uniquely determine* the pane, and + until now that rule could only ever refuse -- nothing produced evidence + strong enough to pass it. A multiplexer that injects the pane id into each + pane's process environment produces exactly that evidence, and it is + correct **by construction**: the value is not inferred, not matched against + a display title, and not read out of a listing that could name someone + else's pane. + + The distinction that makes this safe: a *self*-query answers "which pane am + I", which no amount of ambiguity elsewhere can corrupt. Reading a pane + listing answers "which panes exist", and choosing one of them from a title + is the guessing the root rule forbids -- a title is a display name, and the + one time a stale ref pointed at a live pane belonging to somebody else, the + id looked entirely plausible. + + Contract for implementations: **measurement only**. Any problem yields None, + which callers must read as "unknown", never as "not in a pane". Never + synthesise a pane id, and never fall back to a listing. + """ + return None + def exists(self, pane_ref: dict[str, str]) -> Capability: """Existence preflight for the addressed pane (rule 5). @@ -1445,6 +1470,30 @@ def _active_tab(self, session: str) -> str | None: def _own_session(self) -> str | None: return os.environ.get("ZELLIJ_SESSION_NAME") or None + def self_pane_ref(self) -> dict[str, str] | None: + """Own pane from `ZELLIJ_SESSION_NAME` + `ZELLIJ_PANE_ID`. Runs no command. + + `ZELLIJ_PANE_ID` holds a bare integer while `zellij action --pane-id` + documents `terminal_1, plugin_2 or 3 (equivalent to terminal_3)` -- both + forms address the same pane, so a bare id is *accepted* by the write path. + It is nonetheless canonicalised to `terminal_` here, because + `action list-panes` reports the prefixed form and the reachability check + compares the stored value against that listing **as a string**: storing the + bare form would leave delivery working while the *verification* refused, + i.e. a fail-closed refusal on a pane that was fine. + + Only a bare integer is prefixed. A value that already carries a kind + (`terminal_`/`plugin_`) is passed through untouched -- rewriting it would be + this method inventing an id, which the base contract forbids. + """ + session = self._own_session() + pane = (os.environ.get("ZELLIJ_PANE_ID") or "").strip() + if not session or not pane: + return None + if pane.isdigit(): + pane = f"terminal_{pane}" + return {"multiplexer": self.name, "session": session, "pane_id": pane} + def _classify_intrusion(self, outcome: CommandOutcome) -> str | None: if outcome.rejected: # Nothing was delivered and nothing moved; the intrusion question @@ -1853,6 +1902,42 @@ def _own_session(self) -> str | None: return None return outcome.stdout.strip() or None + def self_pane_ref(self) -> dict[str, str] | None: + """Own pane from `TMUX_PANE`, plus the session name via a *self*-query. + + `TMUX_PANE` is the pane's own id, injected by tmux -- the same reading the + existing same-session measurement already relies on. The session name needs + one command, and `display-message -p` **with no `-t`** is the one legitimate + use of it: with no target it is a self-query, so the "silently falls back to + the current pane" behaviour that disqualifies it as an existence probe is + precisely what is wanted here. + + The socket dimension is carried when `$TMUX` reports one, because the + `(multiplexer, session, pane_id)` triple can name two different real panes + on two different servers. + """ + pane = (os.environ.get("TMUX_PANE") or "").strip() + if not pane: + return None + try: + outcome = self._run( + [self.executable, "display-message", "-p", "#{session_name}"], + cwd=None, + timeout=None, + ) + except Exception: # measurement must never synthesise a reading + return None + if outcome.rejected or outcome.returncode != 0: + return None + session = outcome.stdout.strip() + if not session: + return None + ref = {"multiplexer": self.name, "session": session, "pane_id": pane} + socket = os.environ.get("TMUX", "").split(",")[0].strip() + if socket: + ref["socket"] = socket + return ref + @staticmethod def _same_server(pane_ref: dict[str, str]) -> bool | None: """Is the addressed server this process's own server? None = unknown. @@ -1920,6 +2005,37 @@ def send_key( } +def self_reported_pane_ref() -> tuple[dict[str, str] | None, str]: + """This process's own pane, asked of every transport. → (pane_ref | None, reason). + + **Ambiguity refuses.** If more than one transport reports a pane, this process + sits inside nested multiplexers and there is no evidence for which one owns the + pane that a peer would have to address. Picking one would be a guess, and the + guess is silent: the wrong choice yields a pane_ref that resolves to a real, + live pane belonging to someone else. That is the exact harm §5.0's root rule + exists to prevent, so the ambiguous case reports None with a reason rather than + a plausible answer. + """ + found: list[dict[str, str]] = [] + for factory in TRANSPORTS.values(): + try: + ref = factory().self_pane_ref() + except Exception: # measurement must never break the caller + ref = None + if ref is not None: + found.append(ref) + if not found: + return None, "no multiplexer reported a pane for this process" + if len(found) > 1: + names = ", ".join(sorted(str(ref.get("multiplexer")) for ref in found)) + return None, ( + f"nested multiplexers reported a pane ({names}); which one a peer must " + f"address is undetermined, and guessing yields a ref that resolves to " + f"someone else's live pane" + ) + return found[0], "self-reported by the multiplexer to this process" + + def resolve_transport( multiplexer: str, *, @@ -2552,6 +2668,32 @@ def write_runtime_state( runtime = read_json(agent.runtime_path) runtime["state"] = state runtime["last_seen"] = now + + # A heartbeat is written by the owner from inside its own pane, so it is the one + # moment where evidence that *uniquely determines* the pane is available -- and + # §5.0's root rule until now could only refuse, because nothing produced such + # evidence. Take it, and take it unconditionally. + # + # Absence CLEARS rather than preserves, and the direction is not symmetric: a + # stale ref is worse than a missing one. Missing degrades delivery to "no + # operational route" (visible, fail-closed); stale sends keystrokes into a pane + # that is alive and belongs to somebody else. Measured instance: after a machine + # migration, leaves carried a previous host's ids while a same-named session was + # live here, so every one of those refs resolved -- to the wrong panes. + # + # The contract this implies, stated so it is not discovered by surprise: + # **run the heartbeat from inside the agent's own pane.** Running it from + # elsewhere is not a mistake to be tolerated by keeping the old value -- from + # elsewhere the old value is, by definition, no longer self-evidenced. + if state == "active": + self_ref, reason = self_reported_pane_ref() + runtime["pane_ref"] = self_ref + if self_ref is None: + runtime["pane_ref_cleared_reason"] = reason + else: + runtime.pop("pane_ref_cleared_reason", None) + runtime["pane_ref_source"] = reason + atomic_write_json(agent.runtime_path, runtime) if global_index is None: return {"leaf": "written", "summary": "not-requested"} diff --git a/tests/test_agenttui_preflight.py b/tests/test_agenttui_preflight.py index 035b491..4cd789e 100644 --- a/tests/test_agenttui_preflight.py +++ b/tests/test_agenttui_preflight.py @@ -14,6 +14,8 @@ import importlib.util import inspect +import textwrap +import ast import io import json import contextlib @@ -2149,15 +2151,48 @@ def test_no_command_this_transport_can_issue_moves_the_focus(self) -> None: for forbidden in ("select-pane", "select-window", "focus"): self.assertNotIn(forbidden, " ".join(argv)) - def test_the_only_property_read_is_a_self_query_without_a_target(self) -> None: - # Mechanical: the banned command appears exactly once in the transport, in - # the self-query, and that call site passes no target. - source = inspect.getsource(AGENTTUI.TmuxTransport._own_session) - whole = inspect.getsource(AGENTTUI.TmuxTransport) + def test_every_display_message_call_site_is_a_self_query_without_a_target(self) -> None: + """`display-message` is only admissible as a *self*-query (no `-t`). + + Why the shape changed: this check used to assert the string appeared + **exactly once** in the transport. That count was a *proxy* for the real + property, and it was enumerative — adding a second, equally legitimate + self-query (own-pane self-report) broke it while violating nothing. Bumping + the number to 2 would have kept the proxy and merely re-tuned it, i.e. moved + the enumeration rather than removed it. + + So the property is checked directly: parse the transport, find every argv + list that mentions the banned command, and require that none of them carries + a target flag. That admits any number of self-queries and still refuses the + one thing the rule is about — using it with `-t`, where its silent fallback + to the current pane makes a missing pane look present. + """ + tree = ast.parse(textwrap.dedent(inspect.getsource(AGENTTUI.TmuxTransport))) + + call_sites: list[list[str]] = [] + for node in ast.walk(tree): + if not isinstance(node, (ast.List, ast.Tuple)): + continue + literals = [ + element.value + for element in node.elts + if isinstance(element, ast.Constant) and isinstance(element.value, str) + ] + if "display-message" in literals: + call_sites.append(literals) - self.assertIn("display-message", source) - self.assertNotIn('"-t"', source) - self.assertEqual(1, whole.count('"display-message"')) + # Not vacuous: if the parse found nothing, the check proves nothing. + self.assertTrue( + call_sites, + "no display-message argv found — the check would pass vacuously", + ) + for literals in call_sites: + self.assertNotIn( + "-t", literals, + f"display-message used with a target: {literals!r} — with `-t` it " + f"silently falls back to the current pane, which makes a missing " + f"pane look present", + ) def test_a_missing_pane_is_refused_with_zero_injection_commands(self) -> None: transport, runner = self.transport( diff --git a/tests/test_pane_self_report.py b/tests/test_pane_self_report.py new file mode 100644 index 0000000..c600b01 --- /dev/null +++ b/tests/test_pane_self_report.py @@ -0,0 +1,135 @@ +#!/usr/bin/env python3 +"""自报本 pane:`§5.0` 的根规则第一次拿到一个能让它**通过**的证据来源。 + +那条根规则是「凡自识别证据不足以唯一确定该 pane ⇒ 拒绝写入」。在此之前它只能拒 —— +没有任何东西产出足够强的证据,于是「正确」的路径不存在,只剩「被拒」和「猜」。 + +多路复用器把 pane id 注入到每个 pane 的进程环境里,产出的正是那种证据,而且**按构造正确**: +不是推断出来的、不是从显示标题匹配出来的、也不是从一份可能指到别人 pane 的列举里挑出来的。 + +**关键区别**:自查询答的是「我是哪个 pane」,这一问不会因别处的歧义而失真;读列举答的是 +「有哪些 pane」,再按标题挑一个就是根规则禁止的猜 —— 标题是显示名,而真实发生过的那次, +陈旧 ref 指向的是一个活着、且属于别人的 pane,id 看起来完全合理。 +""" + +from __future__ import annotations + +import importlib.util +import os +import unittest +from pathlib import Path +from unittest import mock + +ROOT = Path(__file__).resolve().parents[1] +SPEC = importlib.util.spec_from_file_location( + "agenttui_under_test", ROOT / "overlay" / "scripts" / "agenttui.py" +) +assert SPEC and SPEC.loader +AGENTTUI = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(AGENTTUI) + + +def only(**environ: str) -> mock._patch_dict: + """把环境**整个换掉**,而不是叠加 —— 否则真实运行环境里的 ZELLIJ_* 会污染断言。""" + return mock.patch.dict(os.environ, environ, clear=True) + + +class ZellijSelfReportTest(unittest.TestCase): + def transport(self) -> object: + return AGENTTUI.ZellijTransport() + + def test_it_reports_the_pane_from_the_injected_environment(self) -> None: + with only(ZELLIJ_SESSION_NAME="workspace", ZELLIJ_PANE_ID="7"): + self.assertEqual( + {"multiplexer": "zellij", "session": "workspace", "pane_id": "terminal_7"}, + self.transport().self_pane_ref(), + ) + + def test_a_bare_integer_is_canonicalised_to_the_prefixed_form(self) -> None: + """写路径两种形态都收,但**核验**拿存的值去比 `list-panes` 的输出。 + + 存裸形态会让投递可用而核验拒绝 —— 一个 pane 明明好着,却 fail-closed。 + """ + with only(ZELLIJ_SESSION_NAME="w", ZELLIJ_PANE_ID="0"): + self.assertEqual("terminal_0", self.transport().self_pane_ref()["pane_id"]) + + def test_an_id_that_already_carries_a_kind_is_passed_through_untouched(self) -> None: + """改写它就是本方法在**发明** id,基类契约禁止。""" + for given in ("terminal_3", "plugin_2"): + with only(ZELLIJ_SESSION_NAME="w", ZELLIJ_PANE_ID=given): + self.assertEqual(given, self.transport().self_pane_ref()["pane_id"]) + + def test_a_missing_pane_id_yields_unknown_not_a_synthesised_ref(self) -> None: + with only(ZELLIJ_SESSION_NAME="workspace"): + self.assertIsNone(self.transport().self_pane_ref()) + + def test_a_missing_session_yields_unknown(self) -> None: + with only(ZELLIJ_PANE_ID="7"): + self.assertIsNone(self.transport().self_pane_ref()) + + def test_an_empty_environment_yields_unknown(self) -> None: + with only(): + self.assertIsNone(self.transport().self_pane_ref()) + + def test_it_runs_no_command(self) -> None: + """纯读环境。跑命令会让「测量」有失败模式,而测量不得能拖垮调用方。""" + transport = self.transport() + with mock.patch.object( + transport, "_run", side_effect=AssertionError("self-report ran a command") + ): + with only(ZELLIJ_SESSION_NAME="w", ZELLIJ_PANE_ID="1"): + self.assertIsNotNone(transport.self_pane_ref()) + + +class ResolverRefusesAmbiguityTest(unittest.TestCase): + """开火构造:歧义必须拒,而不是挑一个看起来合理的。""" + + def test_nested_multiplexers_refuse_instead_of_picking_one(self) -> None: + zellij = AGENTTUI.ZellijTransport() + tmux = AGENTTUI.TmuxTransport() + with mock.patch.object( + zellij, "self_pane_ref", + return_value={"multiplexer": "zellij", "session": "a", "pane_id": "terminal_1"}, + ), mock.patch.object( + tmux, "self_pane_ref", + return_value={"multiplexer": "tmux", "session": "b", "pane_id": "%3"}, + ), mock.patch.dict( + AGENTTUI.TRANSPORTS, {"zellij": lambda: zellij, "tmux": lambda: tmux}, clear=True + ): + ref, reason = AGENTTUI.self_reported_pane_ref() + self.assertIsNone(ref) + self.assertIn("nested", reason) + self.assertIn("someone else", reason) + + def test_no_multiplexer_reporting_yields_a_stated_reason(self) -> None: + with only(): + ref, reason = AGENTTUI.self_reported_pane_ref() + self.assertIsNone(ref) + self.assertTrue(reason.strip()) + + def test_a_transport_that_raises_is_treated_as_unknown_not_fatal(self) -> None: + broken = AGENTTUI.ZellijTransport() + with mock.patch.object(broken, "self_pane_ref", side_effect=RuntimeError("boom")), \ + mock.patch.dict(AGENTTUI.TRANSPORTS, {"zellij": lambda: broken}, clear=True): + ref, reason = AGENTTUI.self_reported_pane_ref() + self.assertIsNone(ref) + self.assertTrue(reason.strip()) + + def test_exactly_one_reporter_is_taken(self) -> None: + zellij = AGENTTUI.ZellijTransport() + expected = {"multiplexer": "zellij", "session": "a", "pane_id": "terminal_1"} + with mock.patch.object(zellij, "self_pane_ref", return_value=expected), \ + mock.patch.dict(AGENTTUI.TRANSPORTS, {"zellij": lambda: zellij}, clear=True): + ref, reason = AGENTTUI.self_reported_pane_ref() + self.assertEqual(expected, ref) + self.assertIn("self-reported", reason) + + +class BaseContractTest(unittest.TestCase): + def test_the_abstract_transport_reports_unknown_rather_than_raising(self) -> None: + """新 transport 忘了实现时,方向必须是「未知」,不是崩,也不是编一个。""" + self.assertIsNone(AGENTTUI.PaneTransport().self_pane_ref()) + + +if __name__ == "__main__": + unittest.main()