feat(agenttui): a pane can self-report its own id, so §5.0's root rule can finally pass

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_<n>` 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"}

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(

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()