feat(commands): neutral send_keys action to replay a key/text step list to a pane

src/tendwire/command_submission.py

@@ -387,6 +387,30 @@ def _submit_private_pane_input(client: Any, pane_id: str, instruction_text: str,
     )
 
 
+def _submit_private_pane_keys(client: Any, pane_id: str, steps: list[Mapping[str, Any]], *, timeout: float) -> None:
+    """Replay an ordered send_keys step list to a pane. Each step sends EITHER a run of neutral key
+    tokens (pane.send_keys) or a run of literal text (pane.send_text). herdres builds the sequence
+    that answers the TUI prompt (digit+enter, arrow toggles, or navigate + write-in); tendwire relays
+    it verbatim. A small inter-step delay lets the foreground TUI repaint between groups."""
+    for index, step in enumerate(steps):
+        if index:
+            time.sleep(_SUBMIT_ENTER_DELAY_SECONDS)
+        if "text" in step:
+            _socket_request(
+                client,
+                "pane.send_text",
+                {"pane_id": pane_id, "text": str(step.get("text") or "")},
+                timeout=timeout,
+            )
+        else:
+            _socket_request(
+                client,
+                "pane.send_keys",
+                {"pane_id": pane_id, "keys": [str(token) for token in (step.get("keys") or [])]},
+                timeout=timeout,
+            )
+
+
 def _target_state_at_send(worker: Worker) -> str:
     status = str(worker.status or "").strip().lower().replace("-", "_")
     return status or "unknown"
@@ -577,6 +601,114 @@ def _socket_send_envelope(
     )
 
 
+def _socket_send_keys_envelope(
+    config: Config,
+    request: CommandRequest,
+    resolved: ResolvedCommandTarget,
+    *,
+    socket_client_factory: SocketClientFactory | None = None,
+) -> CommandEnvelope:
+    """Deliver a send_keys request over the Herdr socket: resolve the neutral target to a pane and
+    replay params.steps via pane.send_keys / pane.send_text. Mirrors _socket_send_envelope's
+    connect/resolve/error handling, but replays a key/text step list instead of instruction text."""
+    steps = list((request.params or {}).get("steps") or [])
+    if not steps:
+        return CommandEnvelope.from_result(
+            request,
+            ok=False,
+            status=STATUS_BACKEND_FAILED,
+            error=error_value(STATUS_BACKEND_FAILED, "send_keys steps are missing after validation"),
+        )
+
+    factory = socket_client_factory or _default_socket_client_factory
+    client: Any | None = None
+    try:
+        client = factory(config)
+        if not hasattr(client, "request"):
+            raise TypeError("socket client does not expose generic request")
+        if hasattr(client, "connect"):
+            client.connect()
+    except Exception:  # noqa: BLE001
+        if client is not None and hasattr(client, "close"):
+            try:
+                client.close()
+            except Exception:
+                pass
+        return _backend_unavailable(request, "Herdr socket could not be reached")
+
+    from .backends.herdr_protocol import HerdrErrorResponse, HerdrProtocolError
+    from .backends.herdr_socket import (
+        HerdrSocketConnectionError,
+        HerdrSocketDisconnectedError,
+        HerdrSocketTimeoutError,
+    )
+
+    try:
+        pane_id = _private_pane_id_for_binding(
+            client,
+            resolved.binding,
+            timeout=config.herdr_timeout_seconds,
+        )
+    except Exception as exc:  # noqa: BLE001
+        if hasattr(client, "close"):
+            client.close()
+        if isinstance(exc, HerdrErrorResponse):
+            return _backend_failure(request, "Herdr socket could not resolve the private send target")
+        if isinstance(
+            exc,
+            HerdrSocketConnectionError | HerdrSocketTimeoutError | HerdrSocketDisconnectedError | HerdrProtocolError,
+        ) or isinstance(exc, OSError):
+            return _backend_unavailable(request, "Herdr socket could not resolve the private send target")
+        if isinstance(exc, (TypeError, ValueError)):
+            return _backend_failure(request, "Herdr socket private send target is unsupported")
+        raise
+
+    if not pane_id:
+        if hasattr(client, "close"):
+            client.close()
+        return _backend_failure(request, "Herdr socket private send target has no pane")
+
+    _append_command_event(
+        config,
+        "command.send_started",
+        request,
+        status=STATUS_PENDING,
+        target_worker_id=resolved.worker.id,
+    )
+    try:
+        _submit_private_pane_keys(
+            client,
+            pane_id,
+            steps,
+            timeout=config.herdr_timeout_seconds,
+        )
+    except Exception as exc:  # noqa: BLE001
+        if isinstance(exc, HerdrErrorResponse):
+            return _backend_failure(request, "Herdr socket pane keys returned an error response")
+        if isinstance(
+            exc,
+            HerdrSocketConnectionError | HerdrSocketTimeoutError | HerdrSocketDisconnectedError | HerdrProtocolError,
+        ) or isinstance(exc, OSError):
+            return _backend_uncertain(request, "Herdr socket pane keys state is uncertain after send start")
+        raise
+    finally:
+        if hasattr(client, "close"):
+            client.close()
+
+    return CommandEnvelope.from_result(
+        request,
+        ok=True,
+        status=STATUS_ACCEPTED,
+        result={
+            "target": {"worker_id": resolved.worker.id},
+            "delivery_state": "submitted",
+            "transport_state": "submitted",
+            "target_state_at_send": _target_state_at_send(resolved.worker),
+            "observed_turn_state": "pending_observation",
+        },
+    )
+
+
 def _envelope_from_receipt(request: CommandRequest, receipt: Mapping[str, Any]) -> CommandEnvelope:
     if receipt.get("payload_fingerprint") != request.payload_fingerprint():
         return CommandEnvelope.error(
@@ -615,8 +747,12 @@ def _envelope_from_receipt(request: CommandRequest, receipt: Mapping[str, Any])
     return CommandEnvelope.from_dict(data)
 
 
+# Actions that mutate a pane (real, non-dry-run send) and so need receipts + socket delivery.
+_MUTATING_ACTIONS = frozenset({"send_instruction", "send_keys"})
+
+
 def _reserve_mutating_request(config: Config, request: CommandRequest) -> CommandEnvelope | None:
-    if request.action != "send_instruction" or request.dry_run or not has_nonblank_request_id(request.request_id):
+    if request.action not in _MUTATING_ACTIONS or request.dry_run or not has_nonblank_request_id(request.request_id):
         return None
     if config.db_path is None:
         return _backend_unavailable(request, "command receipt store is unavailable")
@@ -659,7 +795,7 @@ def _save_mutating_result(
     *,
     worker: Worker | None = None,
 ) -> None:
-    if request.action != "send_instruction" or request.dry_run or not has_nonblank_request_id(request.request_id):
+    if request.action not in _MUTATING_ACTIONS or request.dry_run or not has_nonblank_request_id(request.request_id):
         return
     if config.db_path is None:
         return
@@ -673,7 +809,7 @@ def _save_mutating_result(
         result_json=envelope_to_receipt_json(envelope),
         uncertain=envelope.status == STATUS_REQUEST_STATE_UNCERTAIN,
     )
-    if envelope.status == STATUS_ACCEPTED and worker is not None:
+    if request.action == "send_instruction" and envelope.status == STATUS_ACCEPTED and worker is not None:
         upsert_command_pending_turn(
             config.db_path,
             config.host_id,
@@ -725,7 +861,7 @@ def submit_command(
     if validation_error is not None:
         return CommandEnvelope.error(request, validation_error)
 
-    if request.action != "send_instruction" or request.dry_run:
+    if request.action not in _MUTATING_ACTIONS or request.dry_run:
         return _execute_non_mutating(config, request)
 
     receipt_envelope = _reserve_mutating_request(config, request)
@@ -744,14 +880,22 @@ def submit_command(
             envelope = resolved
         else:
             resolved_worker = resolved.worker
-            envelope = _duplicate_instruction_envelope(config, request, resolved.worker)
-            if envelope is None:
-                envelope = _socket_send_envelope(
+            if request.action == "send_keys":
+                envelope = _socket_send_keys_envelope(
                     config,
                     request,
                     resolved,
                     socket_client_factory=socket_client_factory,
                 )
+            else:
+                envelope = _duplicate_instruction_envelope(config, request, resolved.worker)
+                if envelope is None:
+                    envelope = _socket_send_envelope(
+                        config,
+                        request,
+                        resolved,
+                        socket_client_factory=socket_client_factory,
+                    )
 
     _save_mutating_result(config, request, envelope, worker=resolved_worker)
     return envelope

src/tendwire/core/actions.py

@@ -205,6 +205,53 @@ def _send_instruction_result(request: CommandRequest, context: CommandContext) -
     )
 
 
+def _send_keys_result(request: CommandRequest, context: CommandContext) -> CommandEnvelope:
+    """Pure-path handling for send_keys: resolve the target and preview a dry run. The real key
+    replay is delivered through the daemon socket path (command_submission), so a non-dry-run
+    send_keys that reaches this pure path (no socket backend) reports backend_unsupported."""
+    resolved, _candidates, status = resolve_target(
+        request.target,
+        context.workers,
+        allow_disallowed_status=True,
+        include_backend_target=True,
+    )
+    if status != STATUS_RESOLVED:
+        return _resolve_target_result(request, context.workers)
+    resolved_worker = next(
+        (w for w in context.workers if w.id == (resolved or {}).get("worker_id")),
+        None,
+    )
+    if resolved_worker is not None and resolved_worker.status in {"closed", "failed", "unknown"}:
+        return CommandEnvelope.from_result(
+            request,
+            ok=False,
+            status=STATUS_REJECTED,
+            result={"candidates": [worker_candidate(resolved_worker)]},
+            error=error_value(
+                STATUS_REJECTED,
+                f"target worker status does not allow input: {resolved_worker.status!r}",
+            ),
+        )
+    steps = (request.params or {}).get("steps") or []
+    if request.dry_run:
+        public_target = worker_candidate(resolved_worker) if resolved_worker is not None else (resolved or {})
+        return CommandEnvelope.from_result(
+            request,
+            ok=True,
+            status=STATUS_DRY_RUN,
+            result={"target": public_target, "steps": len(steps)},
+        )
+    return CommandEnvelope.from_result(
+        request,
+        ok=False,
+        status=STATUS_BACKEND_UNSUPPORTED,
+        error=error_value(
+            STATUS_BACKEND_UNSUPPORTED,
+            "send_keys is delivered through the daemon socket path",
+        ),
+    )
+
+
 def execute_command(request: CommandRequest, context: CommandContext) -> CommandEnvelope:
     """Execute a validated command request and return a neutral envelope."""
     validation_error = validate_request(request)
@@ -230,6 +277,9 @@ def execute_command(request: CommandRequest, context: CommandContext) -> Command
     if request.action == "send_instruction":
         return _send_instruction_result(request, context)
 
+    if request.action == "send_keys":
+        return _send_keys_result(request, context)
+
     return CommandEnvelope.error(
         request,
         error_value(STATUS_REJECTED, f"unknown action {request.action!r}"),

src/tendwire/core/commands.py

@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import json
+import re
 from collections.abc import Mapping
 from dataclasses import dataclass, field
 from typing import Any
@@ -27,7 +28,7 @@
 
 COMMAND_SCHEMA_VERSION = 1
 
-ALLOWED_ACTIONS = frozenset({"noop", "read_snapshot", "resolve_target", "send_instruction"})
+ALLOWED_ACTIONS = frozenset({"noop", "read_snapshot", "resolve_target", "send_instruction", "send_keys"})
 REQUEST_ALLOWED_FIELDS = frozenset(
     {"schema_version", "action", "request_id", "dry_run", "target", "instruction", "params"}
 )
@@ -79,6 +80,17 @@
 TARGET_ALLOWED_FIELDS = frozenset({"worker_id", "worker_fingerprint", "space_id", "name"})
 INSTRUCTION_ALLOWED_FIELDS = frozenset({"text"})
 
+# send_keys carries an ordered `params.steps` list. Each step replays EITHER a run of neutral key
+# tokens (`pane.send_keys`) OR a run of literal text (`pane.send_text`) — enough to answer a TUI
+# prompt (digit+enter, arrow toggles, or navigate + type a write-in). Key tokens are a fixed neutral
+# whitelist (navigation + a digit + ctrl+<letter>); they never carry routing/identity data.
+STEP_ALLOWED_FIELDS = frozenset({"keys", "text"})
+_ALLOWED_KEY_TOKEN_RE = re.compile(
+    r"^(?:enter|tab|space|backspace|escape|up|down|left|right|home|end|delete|pageup|pagedown|[0-9]|ctrl\+[a-z])$"
+)
+MAX_KEYS_STEPS = 32
+MAX_KEYS_PER_STEP = 64
+
 # Connector, low-level terminal, routing, and private fields rejected anywhere in a request.
 FORBIDDEN_REQUEST_FIELDS = FORBIDDEN_FIELD_NAMES
 
@@ -199,6 +211,45 @@ def validate_instruction_text(text: Any) -> dict[str, Any] | None:
     return None
 
 
+def keys_error(code: str, message: str) -> dict[str, Any]:
+    return error_value(code, message, details={"field": "params.steps"})
+
+
+def validate_pane_steps(steps: Any) -> dict[str, Any] | None:
+    """Validate a send_keys `params.steps` list (ordered key/text replay), or None if valid.
+
+    Each step is a mapping with EXACTLY one of `keys` (a non-empty list of whitelisted neutral key
+    tokens) or `text` (a literal string validated exactly like instruction.text). Text steps reuse
+    validate_instruction_text so a write-in cannot smuggle escape/control sequences."""
+    if not isinstance(steps, list) or not steps:
+        return keys_error(STATUS_INVALID_REQUEST, "params.steps must be a non-empty list")
+    if len(steps) > MAX_KEYS_STEPS:
+        return keys_error(STATUS_INVALID_REQUEST, f"params.steps exceeds {MAX_KEYS_STEPS} steps")
+    for index, step in enumerate(steps):
+        if not isinstance(step, dict):
+            return keys_error(STATUS_INVALID_REQUEST, f"params.steps[{index}] must be a mapping")
+        extra = set(step) - STEP_ALLOWED_FIELDS
+        if extra:
+            return keys_error(STATUS_INVALID_REQUEST, f"params.steps[{index}] has disallowed keys {sorted(extra)}")
+        has_keys, has_text = "keys" in step, "text" in step
+        if has_keys == has_text:
+            return keys_error(STATUS_INVALID_REQUEST, f"params.steps[{index}] needs exactly one of keys/text")
+        if has_keys:
+            keys = step["keys"]
+            if not isinstance(keys, list) or not keys:
+                return keys_error(STATUS_INVALID_REQUEST, f"params.steps[{index}].keys must be a non-empty list")
+            if len(keys) > MAX_KEYS_PER_STEP:
+                return keys_error(STATUS_INVALID_REQUEST, f"params.steps[{index}].keys exceeds {MAX_KEYS_PER_STEP}")
+            for token in keys:
+                if not isinstance(token, str) or not _ALLOWED_KEY_TOKEN_RE.match(token):
+                    return keys_error(STATUS_INVALID_REQUEST, f"params.steps[{index}] has an unknown key token {token!r}")
+        else:
+            text_err = validate_instruction_text(step["text"])
+            if text_err is not None:
+                return keys_error(STATUS_INVALID_REQUEST, f"params.steps[{index}].text is invalid")
+    return None
+
+
 def _validate_target_shape(target: dict[str, Any] | None) -> dict[str, Any] | None:
     if target is None:
         return None
@@ -363,6 +414,23 @@ def validate_request(request: CommandRequest) -> dict[str, Any] | None:
                 details={"field": "request_id"},
             )
 
+    if request.action == "send_keys":
+        if request.target is None or not _target_has_explicit_selector(request.target):
+            return error_value(
+                STATUS_INVALID_REQUEST,
+                "send_keys requires at least one explicit target selector",
+                details={"field": "target", "allowed": sorted(TARGET_ALLOWED_FIELDS)},
+            )
+        steps_err = validate_pane_steps((request.params or {}).get("steps"))
+        if steps_err is not None:
+            return steps_err
+        if not request.dry_run and not has_nonblank_request_id(request.request_id):
+            return error_value(
+                STATUS_INVALID_REQUEST,
+                "non-dry-run send_keys requires request_id",
+                details={"field": "request_id"},
+            )
+
     return None