fix: herdr 0.7.5 compatibility — send-target resolution + event-stream model

README.md

@@ -425,6 +425,16 @@ resubscribe to the same official event set.
 
 The daemon owns ongoing turn ingestion. It scans eligible durable Codex, OMP,
 and pane bindings immediately at startup and every two seconds by default.
+For pane bindings, a configured wrapper that still implements
+`pane turn PANE_ID --last --format json` remains authoritative. Standalone
+Herdr 0.7.5 removed that CLI subcommand, so Tendwire recognizes only the
+explicit missing-subcommand error and reconstructs a bounded turn through the
+private socket API (`pane.list`, `agent.list`/`agent.get`, and a
+`recent_unwrapped` `pane.read`). Other adapter errors still fail closed. The
+fallback hashes pane identity, revision, and scrollback into the private source
+identifier; raw pane and terminal identifiers are never published. Installations
+that require agent-specific prompt/final parsing can continue to select their
+wrapper with `TENDWIRE_HERDR_BIN`.
 Persisted
 `pane.created`, `pane.focused`, `pane.moved`, `pane.closed`, `pane.exited`,
 `pane.agent_detected`, `pane.agent_status_changed`, and

src/tendwire/backends/herdr_events.py

@@ -82,18 +82,34 @@
 _AGENT_PAYLOAD_KEYS = ("agents", "workers", "data", "items", "results", "result")
 _PANE_PAYLOAD_KEYS = ("panes", "items", "data", "results", "result")
 _SUPPORTED_EVENT_NAMES = HERDR_OFFICIAL_EVENT_NAMES
-_PANE_SCOPED_REPLAY_EVENT_NAMES = frozenset(
+_HERDR_074_EVENT_NAMES = tuple(
+    event_name for event_name in _SUPPORTED_EVENT_NAMES if event_name != "pane.updated"
+)
+_HERDR_074_PANE_SCOPED_REPLAY_EVENT_NAMES = frozenset(
     {
         "workspace.focused",
         "pane.focused",
         "pane.agent_detected",
         "pane.output_matched",
     }
 )
-_PANE_SCOPED_FALLBACK_EVENT_NAMES = tuple(
+_HERDR_074_PANE_SCOPED_FALLBACK_EVENT_NAMES = tuple(
+    event_name
+    for event_name in _HERDR_074_EVENT_NAMES
+    if event_name not in _HERDR_074_PANE_SCOPED_REPLAY_EVENT_NAMES
+)
+_PARAMETERIZED_EVENT_NAMES = frozenset(
+    {"pane.agent_status_changed", "pane.output_matched"}
+)
+_GLOBAL_EVENT_NAMES = tuple(
     event_name
     for event_name in _SUPPORTED_EVENT_NAMES
-    if event_name not in _PANE_SCOPED_REPLAY_EVENT_NAMES
+    if event_name not in _PARAMETERIZED_EVENT_NAMES
+)
+_HERDR_074_GLOBAL_EVENT_NAMES = tuple(
+    event_name
+    for event_name in _HERDR_074_EVENT_NAMES
+    if event_name not in _PARAMETERIZED_EVENT_NAMES
 )
 _CLOSED_EVENT_NAMES = frozenset({"pane.closed", "pane.exited"})
 _SPACE_EVENT_NAMES = frozenset(
@@ -106,10 +122,15 @@
     }
 )
 _WORKTREE_EVENT_NAMES = frozenset({"worktree.created", "worktree.opened", "worktree.removed"})
+# ``pane.updated`` is a refresh notification, not an authoritative PaneInfo
+# observation. Herdr 0.7.5 may emit it with partial or differently shaped pane
+# metadata, so routing it through worker projection can replace the identity
+# inputs established by pane.list. Keep the 474e9ce identity path unchanged.
 _PANE_WORKER_EVENT_NAMES = frozenset({"pane.created", "pane.focused"})
 _TURN_REFRESH_EVENT_NAMES = frozenset(
     {
         "pane.created",
+        "pane.updated",
         "pane.focused",
         "pane.moved",
         "pane.closed",
@@ -304,6 +325,7 @@ def _canonical_event_name(raw_name: Any) -> str | None:
             "agent_observed": "pane.agent_detected",
             "agent_status_changed": "pane.agent_status_changed",
             "agent_status_updated": "pane.agent_status_changed",
+            "pane_output_changed": "pane.updated",
             "pane_observed": "pane.created",
             "pane_detected": "pane.created",
             "workspace_observed": "workspace.updated",
@@ -1040,24 +1062,44 @@ def _call_list_method(self, client: Any, method_name: str) -> Any:
                 client.close()
 
     def _subscribe_event_stream(self, client: Any) -> Any:
-        if self.subscribe_method == HERDR_EVENTS_SUBSCRIBE_METHOD and hasattr(client, "events_subscribe"):
+        # Herdr 0.7.5 strictly validates pane-scoped status subscriptions and
+        # added the general pane.updated event.  Use one bounded mixed
+        # subscription: one global entry per lifecycle/update type and one
+        # status entry per pane.
+        # pane.output_matched is intentionally absent because 0.7.5 requires a
+        # caller-provided match expression; pane.updated is the generic turn
+        # and stream refresh signal.
+        subscriptions = [{"type": event_name} for event_name in _GLOBAL_EVENT_NAMES]
+        subscriptions.extend(
+            {"type": "pane.agent_status_changed", "pane_id": pane_id}
+            for pane_id in self._subscription_pane_ids
+        )
+        params = {"subscriptions": subscriptions}
+        if hasattr(client, "subscribe"):
             try:
-                return client.events_subscribe(
-                    _SUPPORTED_EVENT_NAMES,
+                return client.subscribe(
+                    self.subscribe_method,
+                    params,
                     timeout=self.config.herdr_timeout_seconds,
                     event_timeout=self.config.herdr_timeout_seconds,
                 )
             except TypeError:
-                return client.events_subscribe(_SUPPORTED_EVENT_NAMES)
+                return client.subscribe(self.subscribe_method, params)
             except (HerdrEnvelopeError, HerdrErrorResponse):
-                if not self._subscription_pane_ids:
-                    raise
                 if hasattr(client, "close"):
                     client.close()
                 if hasattr(client, "connect"):
                     client.connect()
-                return self._subscribe_pane_scoped_event_stream(client)
-        params = build_events_subscribe_params(_SUPPORTED_EVENT_NAMES)
+                return self._subscribe_legacy_event_stream(client)
+        if hasattr(client, "events_subscribe"):
+            try:
+                return client.events_subscribe(
+                    _HERDR_074_EVENT_NAMES,
+                    timeout=self.config.herdr_timeout_seconds,
+                    event_timeout=self.config.herdr_timeout_seconds,
+                )
+            except TypeError:
+                return client.events_subscribe(_HERDR_074_EVENT_NAMES)
         try:
             return client.subscribe(
                 self.subscribe_method,
@@ -1068,12 +1110,26 @@ def _subscribe_event_stream(self, client: Any) -> Any:
         except TypeError:
             return client.subscribe(self.subscribe_method, params)
 
-    def _subscribe_pane_scoped_event_stream(self, client: Any) -> Any:
-        subscriptions = [
-            {"pane_id": pane_id, "type": event_name}
-            for pane_id in self._subscription_pane_ids
-            for event_name in _PANE_SCOPED_FALLBACK_EVENT_NAMES
-        ]
+    def _subscribe_legacy_event_stream(self, client: Any) -> Any:
+        # Preserve 0.7.4's proven compatibility request: its status event is
+        # parameterized by pane, while unrelated global event variants tolerate
+        # the same pane_id field. Replay-only events remain excluded exactly as
+        # before so reconnecting does not synthesize focus/detection activity.
+        if self._subscription_pane_ids:
+            subscriptions = [
+                {"pane_id": pane_id, "type": event_name}
+                for pane_id in self._subscription_pane_ids
+                for event_name in _HERDR_074_PANE_SCOPED_FALLBACK_EVENT_NAMES
+            ]
+        else:
+            # An empty installation still needs a live stream for future
+            # workspace/pane creation.  A zero-entry legacy fallback merely
+            # repeats the negotiation failure/reconnect loop, so retain every
+            # 0.7.4 event that is valid without a pane parameter.
+            subscriptions = [
+                {"type": event_name}
+                for event_name in _HERDR_074_GLOBAL_EVENT_NAMES
+            ]
         params = {"subscriptions": subscriptions}
         try:
             return client.subscribe(

src/tendwire/backends/herdr_protocol.py

@@ -31,6 +31,7 @@
     "workspace.focused",
     "pane.created",
     "pane.closed",
+    "pane.updated",
     "pane.focused",
     "pane.moved",
     "pane.exited",
@@ -269,11 +270,25 @@ def is_event(envelope: Mapping[str, Any]) -> bool:
     return "event" in envelope and "result" not in envelope and "error" not in envelope
 
 
-def validate_response(envelope: Mapping[str, Any]) -> dict[str, Any]:
+def validate_response(
+    envelope: Mapping[str, Any],
+    *,
+    allow_uncorrelated_error: bool = False,
+) -> dict[str, Any]:
     """Validate a response envelope while tolerating unknown fields."""
-    _validated_id(envelope)
     if not is_response(envelope):
         raise HerdrEnvelopeError("Herdr response must contain exactly one of result or error")
+    # Herdr 0.7.5 emits ``{"id":"", "error":...}`` when subscription
+    # parameters fail schema validation. Only the subscription negotiation
+    # path may opt into that compatibility exception; ordinary requests remain
+    # strictly correlated.
+    uncorrelated_subscription_error = (
+        allow_uncorrelated_error
+        and is_error_response(envelope)
+        and envelope.get("id") == ""
+    )
+    if not uncorrelated_subscription_error:
+        _validated_id(envelope)
     return dict(envelope)
 
 
@@ -296,10 +311,17 @@ def validate_event(envelope: Mapping[str, Any]) -> dict[str, Any]:
     return dict(envelope)
 
 
-def validate_server_envelope(envelope: Mapping[str, Any]) -> dict[str, Any]:
+def validate_server_envelope(
+    envelope: Mapping[str, Any],
+    *,
+    allow_uncorrelated_error: bool = False,
+) -> dict[str, Any]:
     """Validate a decoded server response or event envelope."""
     if is_response(envelope):
-        return validate_response(envelope)
+        return validate_response(
+            envelope,
+            allow_uncorrelated_error=allow_uncorrelated_error,
+        )
     if is_event(envelope):
         return validate_event(envelope)
     _validated_id(envelope)

src/tendwire/backends/herdr_socket.py

@@ -155,7 +155,11 @@ def request(
     ) -> Any:
         """Send one request and return its raw result payload."""
         request_id, deadline = self._send_request(method, params, timeout=timeout)
-        response = self._read_response(request_id, deadline=deadline)
+        response = self._read_response(
+            request_id,
+            deadline=deadline,
+            allow_uncorrelated_error=False,
+        )
         if is_error_response(response):
             raise HerdrErrorResponse(error_payload(response), request_id)
         return result_payload(response)
@@ -170,7 +174,11 @@ def subscribe(
     ) -> HerdrEventStream:
         """Send a subscription request and return an iterator over its events."""
         request_id, deadline = self._send_request(method, params, timeout=timeout)
-        response = self._read_response(request_id, deadline=deadline)
+        response = self._read_response(
+            request_id,
+            deadline=deadline,
+            allow_uncorrelated_error=True,
+        )
         if is_error_response(response):
             raise HerdrErrorResponse(error_payload(response), request_id)
         return HerdrEventStream(
@@ -324,25 +332,55 @@ def _write(self, payload: bytes, *, deadline: float) -> None:
             self.close()
             raise HerdrSocketDisconnectedError("Herdr socket disconnected during write") from exc
 
-    def _read_response(self, request_id: str, *, deadline: float) -> dict[str, Any]:
+    def _read_response(
+        self,
+        request_id: str,
+        *,
+        deadline: float,
+        allow_uncorrelated_error: bool,
+    ) -> dict[str, Any]:
         while True:
-            envelope = self._read_server_envelope(deadline=deadline)
+            envelope = self._read_server_envelope(
+                deadline=deadline,
+                allow_uncorrelated_error=allow_uncorrelated_error,
+            )
             if is_event(envelope):
                 if len(self._pending_events) >= _MAX_PENDING_EVENTS:
                     raise HerdrEnvelopeError(
                         "too many Herdr events arrived before the response"
                     )
                 self._pending_events.append(envelope)
                 continue
+            if (
+                allow_uncorrelated_error
+                and is_error_response(envelope)
+                and envelope.get("id") == ""
+            ):
+                # Herdr 0.7.5 does not correlate request-schema errors.  They
+                # still belong to the only in-flight synchronous request and
+                # must reach the caller as a server error, not tear down the
+                # event loop as a malformed envelope.
+                payload = envelope.get("error")
+                if not isinstance(payload, Mapping):
+                    raise HerdrEnvelopeError("Herdr error payload must be an object")
+                raise HerdrErrorResponse(dict(payload), request_id)
             ensure_response_id(envelope, request_id)
             if not (is_result_response(envelope) or is_error_response(envelope)):
                 raise HerdrEnvelopeError("expected Herdr response envelope")
             return envelope
 
-    def _read_server_envelope(self, *, deadline: float) -> dict[str, Any]:
+    def _read_server_envelope(
+        self,
+        *,
+        deadline: float,
+        allow_uncorrelated_error: bool = False,
+    ) -> dict[str, Any]:
         line = self._read_line(deadline=deadline)
         envelope = parse_json_line(line)
-        return validate_server_envelope(envelope)
+        return validate_server_envelope(
+            envelope,
+            allow_uncorrelated_error=allow_uncorrelated_error,
+        )
 
     def _read_line(self, *, deadline: float) -> bytes:
         while True: