diff --git a/README.md b/README.md index 1131aef..25b08d8 100644 --- a/README.md +++ b/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 diff --git a/src/tendwire/backends/herdr_events.py b/src/tendwire/backends/herdr_events.py index 79164e0..6f28917 100644 --- a/src/tendwire/backends/herdr_events.py +++ b/src/tendwire/backends/herdr_events.py @@ -82,7 +82,10 @@ _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", @@ -90,10 +93,23 @@ "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( diff --git a/src/tendwire/backends/herdr_protocol.py b/src/tendwire/backends/herdr_protocol.py index 82b2604..b444e3f 100644 --- a/src/tendwire/backends/herdr_protocol.py +++ b/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) diff --git a/src/tendwire/backends/herdr_socket.py b/src/tendwire/backends/herdr_socket.py index eddca95..bbbf19b 100644 --- a/src/tendwire/backends/herdr_socket.py +++ b/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,9 +332,18 @@ 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( @@ -334,15 +351,36 @@ def _read_response(self, request_id: str, *, deadline: float) -> dict[str, Any]: ) 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: diff --git a/src/tendwire/backends/herdr_turns.py b/src/tendwire/backends/herdr_turns.py index 7d8af98..9d5a401 100644 --- a/src/tendwire/backends/herdr_turns.py +++ b/src/tendwire/backends/herdr_turns.py @@ -616,6 +616,171 @@ class _TurnReadFailed(Exception): """Fixed internal adapter failure signal; never serialized with raw errors.""" +def _pane_turn_subcommand_missing(completed: subprocess.CompletedProcess[str]) -> bool: + if completed.returncode == 0: + return False + error_text = str(completed.stderr or "").lower() + missing_markers = ( + "unrecognized subcommand", + "unknown subcommand", + "invalid subcommand", + "no such subcommand", + ) + return "turn" in error_text and any( + marker in error_text for marker in missing_markers + ) + + +def _socket_result_mapping(value: Any, key: str) -> Mapping[str, Any] | None: + if not isinstance(value, Mapping): + return None + result = value.get("result") + if isinstance(result, Mapping): + value = result + nested = value.get(key) + if isinstance(nested, Mapping): + return nested + return value + + +def _socket_result_items(value: Any, key: str) -> list[Mapping[str, Any]]: + if not isinstance(value, Mapping): + return [] + result = value.get("result") + if isinstance(result, Mapping): + value = result + items = value.get(key) + if not isinstance(items, list): + return [] + return [item for item in items if isinstance(item, Mapping)] + + +def _new_herdr_socket_client(*, timeout: float) -> Any: + # Preserve the CLI backend's lazy boundary: importing normal Tendwire CLI + # modules must not load the opt-in socket transport unless this exact + # compatibility fallback is needed. + from .herdr_socket import HerdrSocketClient + + return HerdrSocketClient(timeout=timeout) + + +def _read_private_turn_via_socket( + pane_id: str, + *, + timeout_seconds: float, +) -> Mapping[str, Any]: + """Reconstruct a bounded standalone turn after ``pane turn`` removal.""" + from .herdr_protocol import ( + HerdrEnvelopeError, + HerdrErrorResponse, + HerdrProtocolError, + ) + + try: + with _new_herdr_socket_client(timeout=timeout_seconds) as client: + pane_matches = [ + pane + for pane in _socket_result_items( + client.pane_list({}, timeout=timeout_seconds), + "panes", + ) + if str( + pane.get("pane_id") or pane.get("paneId") or "" + ).strip() + == pane_id + ] + if len(pane_matches) != 1: + raise _TurnReadFailed + pane = pane_matches[0] + + agent_matches = [ + agent + for agent in _socket_result_items( + client.agent_list({}, timeout=timeout_seconds), + "agents", + ) + if str( + agent.get("pane_id") or agent.get("paneId") or "" + ).strip() + == pane_id + ] + if len(agent_matches) > 1: + raise _TurnReadFailed + listed_agent = agent_matches[0] if agent_matches else None + try: + agent = _socket_result_mapping( + client.agent_get( + {"target": pane_id}, + timeout=timeout_seconds, + ), + "agent", + ) + except (HerdrEnvelopeError, HerdrErrorResponse): + agent = listed_agent + if agent is None: + agent = pane + agent_pane_id = str( + agent.get("pane_id") or agent.get("paneId") or pane_id + ).strip() + if agent_pane_id != pane_id: + raise _TurnReadFailed + + read = _socket_result_mapping( + client.pane_read( + { + "pane_id": pane_id, + "source": "recent_unwrapped", + "format": "text", + "strip_ansi": True, + }, + timeout=timeout_seconds, + ), + "read", + ) + if read is None: + raise _TurnReadFailed + read_pane_id = str( + read.get("pane_id") or read.get("paneId") or pane_id + ).strip() + text = read.get("text") + if ( + read_pane_id != pane_id + or not isinstance(text, str) + or not text.strip() + ): + raise _TurnReadFailed + + status = str( + agent.get("agent_status") + or agent.get("status") + or pane.get("agent_status") + or pane.get("status") + or "unknown" + ).strip().lower() + settled = status in {"idle", "done", "blocked"} and not bool( + read.get("truncated") + ) + revision = str(read.get("revision") or pane.get("revision") or "") + source_digest = hashlib.sha256( + (pane_id + "\x00" + revision + "\x00" + text).encode("utf-8") + ).hexdigest()[:32] + turn: dict[str, Any] = { + "available": True, + "complete": settled, + "has_open_turn": not settled, + "source_turn_id": f"socket-turn-{source_digest}", + } + if settled: + turn["assistant_final_text"] = text + else: + turn["assistant_stream_text"] = text + return turn + except _TurnReadFailed: + raise + except (OSError, ValueError, HerdrProtocolError): + raise _TurnReadFailed from None + + def _read_private_turn( config: Config, pane_id: str, @@ -686,15 +851,30 @@ def _read_private_turn( raise _TurnReadFailed from None return None if completed.returncode != 0: - if raise_timeout: - raise _TurnReadFailed - return None - try: - payload = json.loads(completed.stdout) - except (json.JSONDecodeError, TypeError, ValueError): - if raise_timeout: - raise _TurnReadFailed from None - return None + if not _pane_turn_subcommand_missing(completed): + if raise_timeout: + raise _TurnReadFailed + return None + try: + payload = _read_private_turn_via_socket( + pane_id, + timeout_seconds=float(timeout), + ) + except _TurnReadFailed: + if raise_timeout: + raise + return None + except (OSError, ValueError): + if raise_timeout: + raise _TurnReadFailed from None + return None + else: + try: + payload = json.loads(completed.stdout) + except (json.JSONDecodeError, TypeError, ValueError): + if raise_timeout: + raise _TurnReadFailed from None + return None turn = _extract_turn_payload(payload) if not isinstance(turn, Mapping): if raise_timeout: diff --git a/src/tendwire/command_submission.py b/src/tendwire/command_submission.py index b1b7405..7dc7a5f 100644 --- a/src/tendwire/command_submission.py +++ b/src/tendwire/command_submission.py @@ -338,12 +338,37 @@ def _pane_id_from_agent_info(value: Any) -> str: def _private_pane_id_for_binding(client: Any, binding: WorkerBinding, *, timeout: float) -> str: if binding.target_kind == "pane_id": return str(binding.target_value or "").strip() - response = _socket_request( - client, - "agent.get", - {"target": binding.target_value}, - timeout=timeout, - ) + try: + response = _socket_request( + client, + "agent.get", + {"target": binding.target_value}, + timeout=timeout, + ) + except Exception as exc: # noqa: BLE001 + from .backends.herdr_protocol import HerdrErrorResponse + + # Herdr 0.7.5 stopped resolving terminal-id targets through agent.get + # while agent.list still publishes the terminal_id -> pane_id mapping, + # so a definite target-lookup error falls back to the listing before + # the caller terminalizes the request. + if not isinstance(exc, HerdrErrorResponse) or binding.target_kind != "terminal_id": + raise + listing = _socket_request(client, "agent.list", {}, timeout=timeout) + agents = listing.get("agents") if isinstance(listing, Mapping) else None + matches: list[str] = [] + for agent in agents or []: + if not isinstance(agent, Mapping): + continue + if str(agent.get("terminal_id") or "") == str(binding.target_value or ""): + pane_id = str(agent.get("pane_id") or "").strip() + if pane_id: + matches.append(pane_id) + if len(matches) > 1: + raise ValueError("ambiguous agent.list terminal_id match") + if matches: + return matches[0] + raise return _pane_id_from_agent_info(response) diff --git a/src/tendwire/store/sqlite.py b/src/tendwire/store/sqlite.py index fa6f506..1f4fe55 100644 --- a/src/tendwire/store/sqlite.py +++ b/src/tendwire/store/sqlite.py @@ -16875,7 +16875,21 @@ def run_store_maintenance( def _turn_merge_match_text(value: Any) -> str: - return "\n".join(" ".join(line.split()) for line in str(value or "").splitlines()).strip() + raw = str(value or "") + start = 0 + end = len(raw) + + def is_framing_control(char: str) -> bool: + code = ord(char) + return (code < 32 and code not in {9, 10}) or 0x7F <= code <= 0x9F + + while start < end and is_framing_control(raw[start]): + start += 1 + while end > start and is_framing_control(raw[end - 1]): + end -= 1 + return "\n".join( + " ".join(line.split()) for line in raw[start:end].splitlines() + ).strip() def _turn_merge_score(payload: Mapping[str, Any], content: Mapping[str, Any]) -> tuple[int, str, str]: diff --git a/tests/test_command_submission.py b/tests/test_command_submission.py index 171aceb..bd0aec8 100644 --- a/tests/test_command_submission.py +++ b/tests/test_command_submission.py @@ -4029,3 +4029,79 @@ def durable_identity() -> tuple[Any, ...]: assert current_revision_count == 1 assert foreign_keys == [] assert raw_source not in json.dumps(after_public, sort_keys=True) + + +def test_terminal_id_binding_falls_back_to_agent_list_when_agent_get_refuses(monkeypatch) -> None: + # Herdr 0.7.5 regression: agent.get no longer resolves terminal-id targets; + # agent.list still publishes terminal_id -> pane_id. A definite error from + # agent.get must fall back to the listing instead of terminalizing. + from tendwire import command_submission as cs + from tendwire.backends.herdr_protocol import HerdrErrorResponse + + calls = [] + + def fake_socket_request(client, method, params, *, timeout): + calls.append(method) + if method == "agent.get": + raise HerdrErrorResponse({"code": "agent_not_found", "message": "agent target term_new not found"}, "req-1") + if method == "agent.list": + return { + "agents": [ + {"terminal_id": "term_other", "pane_id": "w1:p1"}, + {"terminal_id": "term_new", "pane_id": "w1:p9"}, + ] + } + raise AssertionError(f"unexpected method {method}") + + monkeypatch.setattr(cs, "_socket_request", fake_socket_request) + + class _Binding: + target_kind = "terminal_id" + target_value = "term_new" + + pane_id = cs._private_pane_id_for_binding(object(), _Binding(), timeout=5.0) + assert pane_id == "w1:p9" + assert calls == ["agent.get", "agent.list"] + + # A non-terminal binding kind must NOT consult the listing: the original + # error propagates. + class _SessionBinding: + target_kind = "agent_session" + target_value = "sess-1" + + calls.clear() + try: + cs._private_pane_id_for_binding(object(), _SessionBinding(), timeout=5.0) + raise AssertionError("expected HerdrErrorResponse to propagate") + except HerdrErrorResponse: + pass + assert calls == ["agent.get"] + + +def test_terminal_id_agent_list_fallback_rejects_ambiguous_matches(monkeypatch) -> None: + from tendwire import command_submission as cs + from tendwire.backends.herdr_protocol import HerdrErrorResponse + + def fake_socket_request(client, method, params, *, timeout): + if method == "agent.get": + raise HerdrErrorResponse( + {"code": "agent_not_found", "message": "target not found"}, + "req-1", + ) + if method == "agent.list": + return { + "agents": [ + {"terminal_id": "term-shared", "pane_id": "w1:p1"}, + {"terminal_id": "term-shared", "pane_id": "w1:p2"}, + ] + } + raise AssertionError(f"unexpected method {method}") + + monkeypatch.setattr(cs, "_socket_request", fake_socket_request) + + class _Binding: + target_kind = "terminal_id" + target_value = "term-shared" + + with pytest.raises(ValueError, match="ambiguous agent.list terminal_id match"): + cs._private_pane_id_for_binding(object(), _Binding(), timeout=5.0) diff --git a/tests/test_herdr_events.py b/tests/test_herdr_events.py index 0595868..b72df46 100644 --- a/tests/test_herdr_events.py +++ b/tests/test_herdr_events.py @@ -19,7 +19,7 @@ import pytest -from tendwire.backends import herdr_turns +from tendwire.backends import herdr_cli, herdr_turns from tendwire.backends.herdr_events import ( DEFAULT_SUBSCRIBE_METHOD, HerdrEventBackend, @@ -38,7 +38,6 @@ HERDR_EVENTS_SUBSCRIBE_METHOD, HERDR_OFFICIAL_EVENT_NAMES, HerdrEnvelopeError, - build_events_subscribe_params, ) from tendwire.config import Config from tendwire.core.models import BackendHealth, Snapshot, Worker, WorkerBinding @@ -604,7 +603,13 @@ def subscribe( backend.run_forever() - expected_params = build_events_subscribe_params(HERDR_OFFICIAL_EVENT_NAMES) + expected_params = { + "subscriptions": [ + {"type": name} + for name in HERDR_OFFICIAL_EVENT_NAMES + if name not in {"pane.agent_status_changed", "pane.output_matched"} + ] + } assert DEFAULT_SUBSCRIBE_METHOD == HERDR_EVENTS_SUBSCRIBE_METHOD assert client.subscriptions == [(HERDR_EVENTS_SUBSCRIBE_METHOD, expected_params)] subscribed_names = {subscription["type"] for subscription in expected_params["subscriptions"]} @@ -616,7 +621,7 @@ def subscribe( }.isdisjoint(subscribed_names) -def test_backend_falls_back_to_private_pane_scoped_event_subscriptions(tmp_path: Path) -> None: +def test_backend_falls_back_to_herdr_074_pane_scoped_event_subscriptions(tmp_path: Path) -> None: config = _config(tmp_path, "pane-scoped-subscribe") init_store(Path(config.db_path)) backend = HerdrEventBackend(config, debounce_seconds=0, reconnect_delay_seconds=0) @@ -634,7 +639,7 @@ def __init__(self) -> None: } ], ) - self.global_attempts = 0 + self.mixed_attempts = 0 self.subscriptions: list[tuple[str, dict[str, Any]]] = [] self.closed = 0 self.connected = 0 @@ -652,8 +657,7 @@ def events_subscribe( timeout: float | None = None, event_timeout: float | None = None, ) -> Any: - self.global_attempts += 1 - raise HerdrEnvelopeError("Herdr envelope id must be a non-empty string") + raise AssertionError("0.7.4 fallback must retain the pane-scoped request") def subscribe( self, @@ -664,6 +668,12 @@ def subscribe( event_timeout: float | None = None, ) -> Any: self.subscriptions.append((method, dict(params))) + if any( + item.get("type") == "pane.updated" + for item in params.get("subscriptions", []) + ): + self.mixed_attempts += 1 + raise HerdrEnvelopeError("0.7.4 does not know pane.updated") backend.stop_event.set() return SimpleNamespace(subscription_id="pane-scoped-sub") @@ -672,15 +682,24 @@ def subscribe( backend.run_forever() - assert client.global_attempts == 1 + assert client.mixed_attempts == 1 assert client.closed >= 1 assert client.connected >= 1 - assert len(client.subscriptions) == 1 - method, params = client.subscriptions[0] + assert len(client.subscriptions) == 2 + mixed_method, mixed_params = client.subscriptions[0] + assert mixed_method == HERDR_EVENTS_SUBSCRIBE_METHOD + assert {"type": "pane.updated"} in mixed_params["subscriptions"] + assert { + "type": "pane.agent_status_changed", + "pane_id": "pane-private", + } in mixed_params["subscriptions"] + assert {"type": "pane.agent_status_changed"} not in mixed_params["subscriptions"] + method, params = client.subscriptions[1] assert method == HERDR_EVENTS_SUBSCRIBE_METHOD subscriptions = params["subscriptions"] fallback_names = set(HERDR_OFFICIAL_EVENT_NAMES) - { "workspace.focused", + "pane.updated", "pane.focused", "pane.agent_detected", "pane.output_matched", @@ -690,6 +709,152 @@ def subscribe( assert {item["pane_id"] for item in subscriptions} == {"pane-private"} +def test_backend_falls_back_to_herdr_074_global_events_with_zero_panes( + tmp_path: Path, +) -> None: + config = _config(tmp_path, "empty-global-subscribe") + init_store(Path(config.db_path)) + backend = HerdrEventBackend(config, debounce_seconds=0, reconnect_delay_seconds=0) + + class EmptyInstallationClient(_StaticClient): + def __init__(self) -> None: + super().__init__() + self.subscriptions: list[tuple[str, dict[str, Any]]] = [] + self.closed = 0 + self.connected = 0 + + def connect(self) -> None: + self.connected += 1 + + def close(self) -> None: + self.closed += 1 + + def subscribe( + self, + method: str, + params: Mapping[str, Any], + *, + timeout: float | None = None, + event_timeout: float | None = None, + ) -> Any: + copied = { + "subscriptions": [dict(item) for item in params["subscriptions"]] + } + self.subscriptions.append((method, copied)) + if any( + item.get("type") == "pane.updated" + for item in copied["subscriptions"] + ): + raise HerdrEnvelopeError("0.7.4 does not know pane.updated") + backend.stop_event.set() + return SimpleNamespace(subscription_id="global-only-sub") + + client = EmptyInstallationClient() + backend.client_factory = lambda _config: client + + backend.run_forever() + + assert len(client.subscriptions) == 2 + assert client.closed >= 1 + assert client.connected >= 1 + method, params = client.subscriptions[1] + assert method == HERDR_EVENTS_SUBSCRIBE_METHOD + subscriptions = params["subscriptions"] + assert subscriptions + assert all(set(item) == {"type"} for item in subscriptions) + fallback_names = {item["type"] for item in subscriptions} + assert "pane.updated" not in fallback_names + assert "pane.agent_status_changed" not in fallback_names + assert "pane.output_matched" not in fallback_names + + +def test_pane_updated_does_not_reproject_worker_identity_metadata( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + fixed_installation_key = b"tendwire-identity-regression-key" + derivation_inputs: list[tuple[bytes, str, str, str, str]] = [] + real_stable_worker_key = herdr_cli.stable_worker_key + + def capture_stable_worker_key( + installation_key: bytes, + *, + backend: str, + host_id: str, + workspace_id: str, + pane_id: str, + ) -> str: + derivation_inputs.append( + (installation_key, backend, host_id, workspace_id, pane_id) + ) + return real_stable_worker_key( + installation_key, + backend=backend, + host_id=host_id, + workspace_id=workspace_id, + pane_id=pane_id, + ) + + monkeypatch.setattr( + herdr_cli, + "load_or_create_installation_key", + lambda _data_dir: fixed_installation_key, + ) + monkeypatch.setattr(herdr_cli, "stable_worker_key", capture_stable_worker_key) + + backend = _backend(tmp_path, "pane-updated-identity") + backend.reconcile_once( + client=_StaticClient( + workspaces=[{"id": "w123456789abcde", "name": "Build"}], + panes=[ + { + "pane_id": "w123456789abcde:pA", + "terminal_id": "terminal-1", + "agent": "claude", + "workspace_id": "w123456789abcde", + "agent_status": "working", + "label": "identity-pane", + } + ], + ) + ) + before = latest_snapshot(backend.db_path, backend.config.host_id) + assert before is not None + assert len(before.workers) == len(derivation_inputs) == 1 + before_identity = ( + before.workers[0].id, + json.dumps(before.workers[0].meta, sort_keys=True, separators=(",", ":")), + tuple(derivation_inputs), + ) + refreshes: list[None] = [] + backend.set_turn_refresh_callback(lambda: refreshes.append(None)) + + assert backend.queue_event_envelope( + { + "event": "pane.updated", + "data": { + "type": "pane_updated", + "pane": { + "pane_id": "different-pane", + "terminal_id": "terminal-1", + "agent": "claude", + "workspace_id": "different-space", + "agent_status": "blocked", + }, + }, + } + ) + + after = latest_snapshot(backend.db_path, backend.config.host_id) + assert after is not None + assert refreshes == [None] + assert ( + after.workers[0].id, + json.dumps(after.workers[0].meta, sort_keys=True, separators=(",", ":")), + tuple(derivation_inputs), + ) == before_identity + + def test_backend_rejects_non_official_subscribe_method(tmp_path: Path) -> None: config = _config(tmp_path, "custom-subscribe") init_store(Path(config.db_path)) @@ -1817,7 +1982,14 @@ def read_event(self, subscription_id: str, *, timeout: float | None = None) -> d backend.run_forever() - expected = build_events_subscribe_params(HERDR_OFFICIAL_EVENT_NAMES) + expected = { + "subscriptions": [ + {"type": name} + for name in HERDR_OFFICIAL_EVENT_NAMES + if name not in {"pane.agent_status_changed", "pane.output_matched"} + ] + + [{"type": "pane.agent_status_changed", "pane_id": "pane-1"}] + } assert first.subscriptions == [(HERDR_EVENTS_SUBSCRIBE_METHOD, expected)] assert second.subscriptions == [(HERDR_EVENTS_SUBSCRIBE_METHOD, expected)] assert first.read_calls == 2 diff --git a/tests/test_herdr_protocol.py b/tests/test_herdr_protocol.py index ab984b6..09cddad 100644 --- a/tests/test_herdr_protocol.py +++ b/tests/test_herdr_protocol.py @@ -224,6 +224,29 @@ def test_validate_server_envelope_rejects_missing_id() -> None: validate_server_envelope({"result": {"ok": True}}) +def test_validate_server_envelope_scopes_herdr_075_uncorrelated_error() -> None: + envelope = { + "id": "", + "error": { + "code": "invalid_request", + "message": "invalid request: missing field pane_id", + }, + } + + with pytest.raises(HerdrEnvelopeError): + validate_server_envelope(envelope) + + assert validate_server_envelope( + envelope, + allow_uncorrelated_error=True, + ) == envelope + + +def test_validate_server_envelope_rejects_missing_id_error() -> None: + with pytest.raises(HerdrEnvelopeError): + validate_server_envelope({"error": {"message": "uncorrelated"}}) + + @pytest.mark.parametrize( "envelope", [ diff --git a/tests/test_herdr_socket.py b/tests/test_herdr_socket.py index 27142dc..358144f 100644 --- a/tests/test_herdr_socket.py +++ b/tests/test_herdr_socket.py @@ -369,6 +369,46 @@ def handler(conn: _Connection) -> None: client.close() +def test_subscription_surfaces_herdr_075_empty_id_schema_error(tmp_path: Path) -> None: + def handler(conn: _Connection) -> None: + conn.read_request() + conn.send_json( + { + "id": "", + "error": { + "code": "invalid_request", + "message": "invalid request: missing field pane_id", + }, + } + ) + + with _FakeHerdrServer(tmp_path, handler) as server: + client = HerdrSocketClient(str(server.path), timeout=1) + with pytest.raises(HerdrErrorResponse, match="missing field pane_id"): + client.events_subscribe(["pane.agent_status_changed"]) + client.close() + + +def test_ordinary_request_rejects_herdr_075_empty_id_error(tmp_path: Path) -> None: + def handler(conn: _Connection) -> None: + conn.read_request() + conn.send_json( + { + "id": "", + "error": { + "code": "invalid_request", + "message": "uncorrelated ordinary request error", + }, + } + ) + + with _FakeHerdrServer(tmp_path, handler) as server: + client = HerdrSocketClient(str(server.path), timeout=1) + with pytest.raises(HerdrEnvelopeError): + client.pane_list() + client.close() + + def test_events_subscribe_wrapper_sends_official_method_and_params(tmp_path: Path) -> None: event_names = ("workspace.created", "pane.agent_status_changed") diff --git a/tests/test_herdr_turns.py b/tests/test_herdr_turns.py index ba61e64..ec2b1a9 100644 --- a/tests/test_herdr_turns.py +++ b/tests/test_herdr_turns.py @@ -222,6 +222,127 @@ def fake_run(args: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str] assert turn["has_open_turn"] is False +def test_missing_pane_turn_cli_falls_back_to_socket_scrollback( + tmp_path: Path, + monkeypatch, +) -> None: + config = Config( + host_id="turn-host", + db_path=tmp_path / "turns.db", + herdr_bin="herdr", + herdr_timeout_seconds=2, + ) + monkeypatch.setattr( + herdr_turns.subprocess, + "run", + lambda args, **kwargs: subprocess.CompletedProcess( + args=args, + returncode=2, + stdout="", + stderr="error: unrecognized subcommand 'turn'", + ), + ) + calls: list[tuple[str, dict[str, Any]]] = [] + + class FakeSocketClient: + def __init__(self, *, timeout: float) -> None: + assert timeout == 2 + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb) -> None: + return None + + def pane_list(self, params, *, timeout): + calls.append(("pane.list", dict(params))) + return { + "type": "pane_list", + "panes": [ + { + "pane_id": "pane-private", + "terminal_id": "terminal-private", + "agent": "codex", + "agent_status": "done", + } + ], + } + + def agent_list(self, params, *, timeout): + calls.append(("agent.list", dict(params))) + return { + "type": "agent_list", + "agents": [ + { + "pane_id": "pane-private", + "terminal_id": "terminal-private", + "agent": "codex", + "agent_status": "done", + } + ], + } + + def agent_get(self, params, *, timeout): + calls.append(("agent.get", dict(params))) + return { + "type": "agent_info", + "agent": { + "pane_id": "pane-private", + "agent": "codex", + "agent_status": "done", + }, + } + + def pane_read(self, params, *, timeout): + calls.append(("pane.read", dict(params))) + return { + "type": "pane_read", + "read": { + "pane_id": "pane-private", + "source": "recent_unwrapped", + "format": "text", + "text": "standalone socket fallback answer", + "revision": 42, + "truncated": False, + }, + } + + monkeypatch.setattr( + herdr_turns, + "_new_herdr_socket_client", + lambda *, timeout: FakeSocketClient(timeout=timeout), + ) + + turn = herdr_turns._read_private_turn( + config, + "pane-private", + raise_timeout=True, + ) + + assert turn is not None + assert turn["assistant_final_text"] == "standalone socket fallback answer" + assert turn["complete"] is True + assert turn["has_open_turn"] is False + assert turn["source_turn_id"].startswith("socket-turn-") + public_turn = {key: value for key, value in turn.items() if not key.startswith("_")} + assert "pane-private" not in json.dumps(public_turn) + assert "terminal-private" not in json.dumps(public_turn) + assert calls == [ + ("pane.list", {}), + ("agent.list", {}), + ("agent.get", {"target": "pane-private"}), + ( + "pane.read", + { + "pane_id": "pane-private", + "source": "recent_unwrapped", + "format": "text", + "strip_ansi": True, + }, + ), + ] + + def test_refresh_structured_turn_content_reads_codex_session_jsonl( tmp_path: Path, monkeypatch, diff --git a/tests/test_store.py b/tests/test_store.py index 9b0f2a5..2f82596 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -4587,6 +4587,106 @@ def test_source_turn_with_matching_prompt_keeps_command_origin(tmp_path: Path) - assert source_rows[0]["source"] == "command" +@pytest.mark.parametrize("framing", ["\x01", "\x7f", "\x80", "\x9f"]) +def test_source_turn_edge_framing_controls_match_command_origin( + tmp_path: Path, + framing: str, +) -> None: + db_path = tmp_path / "turn-source-framing-command.db" + config = Config(host_id="turn-host", db_path=db_path) + snapshot = project_from_raw( + config, + workers=[ + { + "id": "worker-1", + "name": "claude", + "status": "active", + "space_id": "space-1", + } + ], + ) + init_store(db_path) + save_snapshot(db_path, snapshot) + worker = snapshot.workers[0] + assert store_sqlite.upsert_command_pending_turn( + db_path, + "turn-host", + worker, + request_id="req-1", + instruction_text="hello", + observed_at="2026-01-01T00:00:00+00:00", + ) + + assert merge_turn_content( + db_path, + "turn-host", + "worker-1", + { + "user_text": f"{framing}hello{framing}", + "assistant_final_text": "Done.", + "complete": True, + "has_open_turn": False, + "source_turn_id": f"source-framed-{ord(framing)}", + }, + observed_at="2026-01-01T00:01:00+00:00", + ) == 1 + + payload = turns_payload_from_store(db_path, "turn-host", snapshot=snapshot) + source_row = next( + turn for turn in payload["turns"] if turn.get("assistant_final_text") == "Done." + ) + assert source_row["origin_command_id"] == "req-1" + + +def test_source_turn_interior_control_does_not_match_command_origin(tmp_path: Path) -> None: + db_path = tmp_path / "turn-source-interior-control.db" + config = Config(host_id="turn-host", db_path=db_path) + snapshot = project_from_raw( + config, + workers=[ + { + "id": "worker-1", + "name": "claude", + "status": "active", + "space_id": "space-1", + } + ], + ) + init_store(db_path) + save_snapshot(db_path, snapshot) + worker = snapshot.workers[0] + assert store_sqlite.upsert_command_pending_turn( + db_path, + "turn-host", + worker, + request_id="req-1", + instruction_text="hello", + observed_at="2026-01-01T00:00:00+00:00", + ) + + assert merge_turn_content( + db_path, + "turn-host", + "worker-1", + { + "user_text": "hel\x01lo", + "assistant_final_text": "Separate turn.", + "complete": True, + "has_open_turn": False, + "source_turn_id": "source-interior-control", + }, + observed_at="2026-01-01T00:01:00+00:00", + ) == 1 + + payload = turns_payload_from_store(db_path, "turn-host", snapshot=snapshot) + source_row = next( + turn + for turn in payload["turns"] + if turn.get("assistant_final_text") == "Separate turn." + ) + assert source_row.get("origin_command_id") is None + + def test_store_save_latest_host_scope_and_list_hosts(tmp_path: Path) -> None: db_path = tmp_path / "tendwire.db" config_a = Config(host_id="host-a", db_path=db_path)