From 1946b0baeed9b823f9a904a978bcb44aeb0fc66f Mon Sep 17 00:00:00 2001 From: jerryfane Date: Thu, 16 Jul 2026 21:19:59 +0200 Subject: [PATCH 1/3] Gitmoot implement adhoc-751e1dbc --- README.md | 4 + docs/answer_decision.md | 118 ++++ scripts/sqlite_sidecar_race_benchmark.py | 9 +- src/tendwire/backends/herdr_decision.py | 141 ++++ src/tendwire/backends/herdr_turns.py | 114 ++- src/tendwire/cli.py | 10 +- src/tendwire/command_submission.py | 417 ++++++++++- src/tendwire/core/commands.py | 170 ++++- src/tendwire/core/models.py | 1 + src/tendwire/core/turns.py | 39 ++ src/tendwire/store/sqlite.py | 615 +++++++++++++++- tests/test_answer_decision.py | 856 +++++++++++++++++++++++ tests/test_backend_pending.py | 14 +- tests/test_command_submission.py | 16 +- tests/test_commands.py | 74 +- tests/test_turns.py | 1 + 16 files changed, 2564 insertions(+), 35 deletions(-) create mode 100644 docs/answer_decision.md create mode 100644 src/tendwire/backends/herdr_decision.py create mode 100644 tests/test_answer_decision.py diff --git a/README.md b/README.md index bb20364..5581acd 100644 --- a/README.md +++ b/README.md @@ -1307,6 +1307,10 @@ Tendwire now exposes a minimal, safety-first command interface: echo '{"schema_version": 1, "action": "noop"}' | tendwire command --json ``` +The connector-facing structured Claude decision payload, semantic +`answer_decision` action, validation, and retry behavior are documented in +[docs/answer_decision.md](docs/answer_decision.md). + The `command --json` subcommand reads exactly one schema-v1 JSON request from stdin. A proven result is exactly one schema-v2 command envelope on JSON-only stdout with exit `0`/`1`; unresolved process ambiguity is no stdout envelope diff --git a/docs/answer_decision.md b/docs/answer_decision.md new file mode 100644 index 0000000..77f9c0d --- /dev/null +++ b/docs/answer_decision.md @@ -0,0 +1,118 @@ +# Remote decision contract + +`answer_decision` is Tendwire's semantic connector contract for answering the +current structured Claude decision on one worker. Connectors select public +option references or provide permitted write-in text. They never send pane IDs, +terminal IDs, raw keys, cursor movements, or calibration steps. Tendwire +validates the current prompt and owns the private Herdr calibration. + +## Pending payload + +A structured decision appears on a `pending_interactions` item as +`meta.decision`: + +```json +{ + "decision_ref": "decision-", + "kind": "single", + "prompt": "Choose a database", + "options": [ + {"ref": "1", "label": "Postgres"}, + {"ref": "2", "label": "SQLite"} + ], + "multi_select": false, + "question_count": 1 +} +``` + +`decision_ref` identifies one exact prompt instance and changes whenever the +prompt revision or its private source binding changes. Option `ref` values are +stable 1-based ordinals for that instance. Connectors must treat the whole +object as current-state data and fail closed when `question_count` is greater +than 1. + +Supported shapes are: + +- `single`: exactly one option, or nonempty write-in text; +- `plan`: exactly one option; +- `multi`: one or more options from a single question. + +Tendwire supports at most 11 source option rows. A source decision with more +than 11 rows, including a custom/write-in row beyond that boundary, is not +truncated. Unknown decision kinds, over-bound decisions, and multi-question +decisions do not produce `meta.decision` and cannot be sent. + +## Command request + +```json +{ + "schema_version": 1, + "action": "answer_decision", + "request_id": "connector-request-123", + "dry_run": false, + "target": {"worker_id": "worker-123"}, + "params": { + "decision_ref": "decision-", + "selection": {"option_refs": ["2"]} + } +} +``` + +`selection` contains exactly one of these forms: + +- `{"option_refs":[...]}`: every reference must exist in the current stored + options. `single` and `plan` require exactly one unique reference; `multi` + requires one or more unique references. +- `{"text":"..."}`: a nonempty write-in accepted only for `single`. + +Omitting `dry_run` means `dry_run: true`. A live answer therefore requires the +caller to set `dry_run: false` explicitly. Dry runs perform no pane operation +and write no mutation receipt. + +Before any pane operation, Tendwire freshly proves that the worker exists and +is open, the supplied reference is its current pending decision, the decision +shape is supported, and the selection is valid. A proven live success returns +`ok: true`, `status: "accepted"`, and a `terminal_accepted` disposition. + +The four typed validation failures are terminal for that attempted decision: + +- `unknown_worker`; +- `decision_not_pending`; +- `invalid_selection`; +- `unsupported_decision`. + +All fail before pane input. `unsupported_decision` covers multiple question +groups, more than 11 source rows, and unknown kinds. + +## Concurrency and retries + +Only one request may claim a decision at a time. A competing request receives +`ok: false`, `status: "answer_in_progress"`, with either `no_receipt` (before it +reserved a request) or `in_progress` (when its unsent reservation remains +recoverable). This status is retryable. It never creates a terminal receipt and +never causes a second pane mutation. + +If the winning request fails safely before sending, Tendwire releases its +decision claim so a loser may retry. An abandoned pre-send reservation and +claim can be taken over after their leases expire. Once sending may have +started, Tendwire fails closed with `request_state_uncertain` and does not +automatically resend. + +`request_id` uses the normal command deduplication contract. Repeating the same +canonical request ID replays its authoritative result without sending again; +reusing it for a different canonical mutation is rejected as +`duplicate_request`. + +## Paired adapter requirement + +The Tendwire and Herdres versions must be deployed as a matching pair that both +implement this contract. In particular, the paired Herdres adapter must emit a +single-question `AskUserQuestion` with `multiSelect: true` as a structured +`pending_decision` with `mode: "multi"`, `multi_select: true`, 1-based option +IDs, and no custom row. Older adapters that leave that prompt as an unstructured +`pending_interaction` cannot use semantic multi-select answering. Multi-question +prompts remain unstructured and unsupported. + +The Claude Code multi-select key behavior is a private backend assumption, not +part of this public contract. Its provisional calibration is isolated in +`src/tendwire/backends/herdr_decision.py` for live verification and retuning. diff --git a/scripts/sqlite_sidecar_race_benchmark.py b/scripts/sqlite_sidecar_race_benchmark.py index 226fb89..5d527a6 100755 --- a/scripts/sqlite_sidecar_race_benchmark.py +++ b/scripts/sqlite_sidecar_race_benchmark.py @@ -993,6 +993,13 @@ def _run_herdres_phase( if herdres_root.resolve() not in module_path.parents: raise RuntimeError("herdres_origin_failed") origin_ok = True + # The paired Herdres owns its turn page size (it changed 50 -> 100 in + # luminexord/herdres 31c3152); derive the expected command sequence from + # the paired checkout instead of hardcoding a value that breaks the + # pairing every time Herdres retunes it. + turn_page_limit = getattr(tendwire_client, "TURN_LIST_PAGE_LIMIT", 50) + if type(turn_page_limit) is not int or turn_page_limit < 1: + raise RuntimeError("herdres_turn_page_limit_invalid") runtime = source_sync.SyncRuntime( tendwire=tendwire_client.TendwireClient(timeout=10.0), telegram=telegram_delivery.TelegramClient(token="", dry_run=True), @@ -1043,7 +1050,7 @@ def _run_herdres_phase( "--schema-version", "2", "--limit", - "50", + str(turn_page_limit), "--json", ], ["--socket-path", str(socket_path), "pending", "--json"]) diff --git a/src/tendwire/backends/herdr_decision.py b/src/tendwire/backends/herdr_decision.py new file mode 100644 index 0000000..b9d06f0 --- /dev/null +++ b/src/tendwire/backends/herdr_decision.py @@ -0,0 +1,141 @@ +"""Translate semantic Claude decisions into private Herdr pane input. + +Calibration assumptions are deliberately confined to this backend module: + +* Claude Code displays single-choice and plan rows with 1-based decimal + shortcuts; typing the ordinal and then Enter chooses that row. +* A single-choice write-in row immediately follows the advertised options. + Its ordinal opens/focuses the text field without an intermediate Enter, so + Tendwire types ``N + 1`` and then submits the write-in text with Enter. +* Claude Code multi-select digit toggles are not treated as a supported + contract. Tendwire therefore assumes the cursor starts on row 1, Down/Up move + exactly one option row, Space toggles the current option without moving the + cursor, the Submit row immediately follows the final option, and Enter on + Submit submits. This provisional Claude Code behavior is isolated in + ``MULTI_SELECT_CALIBRATION`` below so live verification can retune it in one + place. +* Herdr's private ``pane.send_keys`` accepts decimal character keys plus + ``Down``, ``Up``, and ``Enter``. A literal multi-select Space is sent through + private ``pane.send_text``; write-in prose uses ``pane.send_input`` so Herdr + owns terminal text encoding and appends the final Enter atomically. + +These steps are internal calibration data. They are never accepted from a +connector and there is intentionally no public raw-key command action. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Literal + + +MULTI_SELECT_CALIBRATION = { + "down": "Down", + "up": "Up", + "toggle_text": " ", + "submit": "Enter", +} + + +@dataclass(frozen=True) +class HerdrDecisionStep: + """One private, already-calibrated Herdr pane operation.""" + + operation: Literal["keys", "text", "input"] + keys: tuple[str, ...] = () + text: str | None = None + + def __post_init__(self) -> None: + if self.operation == "keys": + if not self.keys or self.text is not None: + raise ValueError("key calibration step requires only keys") + elif self.operation == "text": + if self.text != " " or self.keys: + raise ValueError("text calibration step requires one literal space") + elif self.operation == "input": + if not isinstance(self.text, str) or not self.text or self.keys != ("Enter",): + raise ValueError("input calibration step requires text plus Enter") + else: + raise ValueError("unsupported decision calibration operation") + + +def _digit_keys(value: int | str) -> tuple[str, ...]: + text = str(value) + if not text.isdigit() or int(text) < 1: + raise ValueError("decision ordinal must be a positive decimal") + return tuple(text) + + +def calibrate_decision_steps( + *, + kind: Literal["single", "multi", "plan"], + option_count: int, + option_refs: tuple[str, ...] = (), + text: str | None = None, +) -> tuple[HerdrDecisionStep, ...]: + """Return private pane operations for one validated semantic selection.""" + if ( + kind not in {"single", "multi", "plan"} + or not isinstance(option_count, int) + or isinstance(option_count, bool) + or option_count < 1 + ): + raise ValueError("invalid decision calibration context") + if text is not None: + if kind != "single" or option_refs or not isinstance(text, str) or not text: + raise ValueError("invalid decision write-in calibration") + return ( + HerdrDecisionStep("keys", keys=_digit_keys(option_count + 1)), + HerdrDecisionStep("input", keys=("Enter",), text=text), + ) + if not option_refs or len(option_refs) != len(set(option_refs)): + raise ValueError("decision option refs must be nonempty and unique") + ordinals: list[int] = [] + for ref in option_refs: + if not isinstance(ref, str) or not ref.isdigit(): + raise ValueError("decision option ref must be a decimal ordinal") + ordinal = int(ref) + if not 1 <= ordinal <= option_count: + raise ValueError("decision option ref is out of range") + ordinals.append(ordinal) + if kind in {"single", "plan"}: + if len(ordinals) != 1: + raise ValueError("single and plan decisions require one option") + return ( + HerdrDecisionStep( + "keys", + keys=(*_digit_keys(ordinals[0]), "Enter"), + ), + ) + + steps: list[HerdrDecisionStep] = [] + current_row = 1 + for ordinal in ordinals: + delta = ordinal - current_row + if delta: + direction = ( + MULTI_SELECT_CALIBRATION["down"] + if delta > 0 + else MULTI_SELECT_CALIBRATION["up"] + ) + steps.append(HerdrDecisionStep("keys", keys=(direction,) * abs(delta))) + steps.append( + HerdrDecisionStep( + "text", + text=MULTI_SELECT_CALIBRATION["toggle_text"], + ) + ) + current_row = ordinal + submit_navigation = (MULTI_SELECT_CALIBRATION["down"],) * ( + option_count + 1 - current_row + ) + steps.append( + HerdrDecisionStep( + "keys", + keys=( + *submit_navigation, + MULTI_SELECT_CALIBRATION["submit"], + ), + ) + ) + return tuple(steps) diff --git a/src/tendwire/backends/herdr_turns.py b/src/tendwire/backends/herdr_turns.py index 53ea526..b22cc49 100644 --- a/src/tendwire/backends/herdr_turns.py +++ b/src/tendwire/backends/herdr_turns.py @@ -386,8 +386,11 @@ def _extract_turn_payload(value: Any) -> Mapping[str, Any] | None: return value -_PENDING_MAX_CHOICES = 12 +PENDING_DECISION_MAX_OPTIONS = 11 _PENDING_TEXT_MAX = 2000 +_SINGLE_WRITE_IN_OPTION_IDS = frozenset( + {"custom", "other", "writein", "write_in", "write-in"} +) def _private_pending_revision(value: Mapping[str, Any]) -> str: @@ -409,7 +412,7 @@ def _pending_observation_from_turn(turn: Mapping[str, Any]) -> PendingObservatio return PendingObservation("read_succeeded_invalid_prompt") revision = _private_pending_revision(decision) question = redact_private_prompt_text( - decision.get("prompt"), + decision.get("prompt") or decision.get("question"), max_chars=_PENDING_TEXT_MAX, ) options = decision.get("options") @@ -417,8 +420,10 @@ def _pending_observation_from_turn(turn: Mapping[str, Any]) -> PendingObservatio options = [] if not isinstance(options, list): return PendingObservation("read_succeeded_invalid_prompt") + if len(options) > PENDING_DECISION_MAX_OPTIONS: + return PendingObservation("read_succeeded_unsupported_decision") choices: list[PendingObservedChoice] = [] - for ordinal, option in enumerate(options[:_PENDING_MAX_CHOICES], 1): + for ordinal, option in enumerate(options, 1): if not isinstance(option, Mapping): return PendingObservation("read_succeeded_invalid_prompt") label = redact_private_prompt_text( @@ -442,12 +447,95 @@ def _pending_observation_from_turn(turn: Mapping[str, Any]) -> PendingObservatio for option in options if isinstance(option, Mapping) } + raw_kind = str( + decision.get("kind") + or decision.get("tool_name") + or decision.get("name") + or "" + ).strip().lower().replace("-", "_") + compact_kind = raw_kind.replace("_", "") + raw_mode = ( + str(decision.get("mode") or "") + .strip() + .lower() + .replace("-", "_") + ) + compact_mode = raw_mode.replace("_", "") + if compact_kind not in { + "", + "askuserquestion", + "single", + "multi", + "multiselect", + "exitplanmode", + "plan", + } or compact_mode not in { + "", + "buttons", + "single", + "multi", + "multiselect", + "plan", + }: + return PendingObservation("read_succeeded_unsupported_decision") + raw_multi_select = decision.get( + "multi_select", + decision.get("multiSelect", False), + ) + if not isinstance(raw_multi_select, bool): + return PendingObservation("read_succeeded_invalid_prompt") + if ( + compact_mode == "plan" + or compact_kind in {"exitplanmode", "plan"} + or (not compact_kind and "approve" in option_ids) + ): + decision_kind: Literal["single", "multi", "plan"] = "plan" + elif ( + compact_mode in {"multi", "multiselect"} + or raw_multi_select + or compact_kind in {"multi", "multiselect"} + ): + decision_kind = "multi" + else: + decision_kind = "single" + if raw_multi_select is not (decision_kind == "multi"): + return PendingObservation("read_succeeded_unsupported_decision") + raw_question_count = decision.get("question_count") + if raw_question_count is None: + raw_questions = decision.get("questions") + raw_question_count = ( + len(raw_questions) if isinstance(raw_questions, list) else 1 + ) + if ( + not isinstance(raw_question_count, int) + or isinstance(raw_question_count, bool) + or raw_question_count < 1 + ): + return PendingObservation("read_succeeded_invalid_prompt") + decision_option_labels = [choice.label for choice in choices] + if ( + decision_kind == "single" + and options + and isinstance(options[len(decision_option_labels) - 1], Mapping) + and str( + options[len(decision_option_labels) - 1].get("id") or "" + ).strip().lower() + in _SINGLE_WRITE_IN_OPTION_IDS + ): + decision_option_labels.pop() + decision_options = tuple(decision_option_labels) + if not decision_options: + return PendingObservation("read_succeeded_invalid_prompt") return PendingObservation( "open_prompt", question=question, pending_kind="approval" if "approve" in option_ids else "question", choices=tuple(choices), revision_digest=revision, + decision_kind=decision_kind, + decision_options=decision_options, + decision_multi_select=decision_kind == "multi", + decision_question_count=raw_question_count, ) interaction = turn.get("pending_interaction") if interaction is not None: @@ -481,6 +569,24 @@ def _backend_pending_from_turn(turn: Mapping[str, Any]) -> dict[str, Any] | None observation = _pending_observation_from_turn(turn) if observation.kind != "open_prompt": return None + meta: dict[str, Any] = {"source": "backend"} + if observation.decision_kind is not None: + meta["decision"] = { + "decision_ref": ( + "decision-" + + stable_fingerprint( + {"decision_revision": observation.revision_digest} + ) + ), + "kind": observation.decision_kind, + "prompt": observation.question, + "options": [ + {"ref": str(ordinal), "label": label} + for ordinal, label in enumerate(observation.decision_options, 1) + ], + "multi_select": observation.decision_multi_select, + "question_count": observation.decision_question_count, + } return { "question": observation.question, "kind": observation.pending_kind or "question", @@ -488,7 +594,7 @@ def _backend_pending_from_turn(turn: Mapping[str, Any]) -> dict[str, Any] | None {"choice_id": choice.choice_id, "label": choice.label} for choice in observation.choices ], - "meta": {"source": "backend"}, + "meta": meta, } diff --git a/src/tendwire/cli.py b/src/tendwire/cli.py index 63bf6a6..bf542a7 100644 --- a/src/tendwire/cli.py +++ b/src/tendwire/cli.py @@ -1214,7 +1214,7 @@ def command_envelope_from_payload(config: Config, payload: str) -> CommandEnvelo if validation_error is not None: return CommandEnvelope.from_error(request, validation_error) - if request.action in {"send_instruction", "answer_pending"}: + if request.action in {"send_instruction", "answer_pending", "answer_decision"}: from .command_submission import submit_command return submit_command(config, payload) @@ -1267,7 +1267,10 @@ def _requires_daemon_for_mutating_command(config: Config, payload: str) -> Any | validation_error = validate_request(request) if validation_error is not None: return None - if request.action in {"send_instruction", "answer_pending"} and not request.dry_run: + if ( + request.action in {"send_instruction", "answer_pending", "answer_decision"} + and not request.dry_run + ): return request return None @@ -1335,7 +1338,8 @@ def cmd_command( parse_error is None and validation_error is None and parsed_request is not None - and parsed_request.action in {"send_instruction", "answer_pending"} + and parsed_request.action + in {"send_instruction", "answer_pending", "answer_decision"} and parsed_request.dry_run ) daemon_required_request = _requires_daemon_for_mutating_command(config, payload) diff --git a/src/tendwire/command_submission.py b/src/tendwire/command_submission.py index cb0401d..228b362 100644 --- a/src/tendwire/command_submission.py +++ b/src/tendwire/command_submission.py @@ -13,23 +13,29 @@ from .core.commands import ( COMMAND_ENVELOPE_SCHEMA_VERSION, DISPOSITION_IN_PROGRESS, + DISPOSITION_NO_RECEIPT, DISPOSITION_TERMINAL_ACCEPTED, DISPOSITION_TERMINAL_REJECTED, DISPOSITION_TERMINAL_UNCERTAIN, STATUS_ACCEPTED, + STATUS_ANSWER_IN_PROGRESS, STATUS_AMBIGUOUS_BACKEND_TARGET, STATUS_AMBIGUOUS_TARGET, STATUS_BACKEND_FAILED, STATUS_BACKEND_UNAVAILABLE, STATUS_BACKEND_UNSUPPORTED, STATUS_DRY_RUN, + STATUS_DECISION_NOT_PENDING, STATUS_DUPLICATE_REQUEST, + STATUS_INVALID_SELECTION, STATUS_NOT_FOUND, STATUS_PENDING, STATUS_REJECTED, STATUS_REQUEST_STATE_UNCERTAIN, STATUS_RESOLVED, STATUS_STALE_TARGET, + STATUS_UNKNOWN_WORKER, + STATUS_UNSUPPORTED_DECISION, CanonicalMutation, CommandEnvelope, CommandRequest, @@ -44,10 +50,13 @@ ) from .core.models import BackendHealth, Snapshot, Worker, WorkerBinding from .core.projector import project_from_observations +from .backends.herdr_decision import calibrate_decision_steps from .store.sqlite import ( abandon_backend_pending_choice_claim, + abandon_command_request_reservation, backend_pending_choice_terminal_effect, claim_backend_pending_choice, + claim_backend_pending_decision, command_pending_turn_terminal_effect, command_reservation_is_live, envelope_to_receipt_json, @@ -59,11 +68,14 @@ reserve_command_request, reserve_terminal_command_replay, start_backend_pending_choice_send, + start_backend_pending_decision_send, ) HERDR_BACKEND = "herdr" -_MUTATING_ACTIONS = frozenset({"send_instruction", "answer_pending"}) +_MUTATING_ACTIONS = frozenset( + {"send_instruction", "answer_pending", "answer_decision"} +) _LEGACY_V0_REPLAY_WORKER_ID = "legacy-v0-replay-only" _PENDING_CHANGED_MESSAGE = "pending interaction changed or is no longer answerable" _DISALLOWED_SEND_STATUSES = frozenset({"closed", "failed", "unknown"}) @@ -394,6 +406,27 @@ def _request_in_progress(request: CommandRequest) -> CommandEnvelope: ) +def _answer_in_progress( + request: CommandRequest, + *, + receipt_reserved: bool = False, +) -> CommandEnvelope: + return CommandEnvelope.from_result( + request, + ok=False, + status=STATUS_ANSWER_IN_PROGRESS, + disposition=( + DISPOSITION_IN_PROGRESS + if receipt_reserved + else DISPOSITION_NO_RECEIPT + ), + error=error_value( + STATUS_ANSWER_IN_PROGRESS, + "another request is currently answering this decision", + ), + ) + + def _duplicate_request(request: CommandRequest) -> CommandEnvelope: return CommandEnvelope.from_result( request, @@ -474,6 +507,82 @@ def _same_pending_route(left: Any, right: Any) -> bool: ) +def _decision_failure_envelope( + request: CommandRequest, + status: str, +) -> CommandEnvelope: + messages = { + STATUS_ANSWER_IN_PROGRESS: "another request is currently answering this decision", + STATUS_DECISION_NOT_PENDING: "decision is not the worker's current pending decision", + STATUS_UNKNOWN_WORKER: "target worker does not exist or is not open", + STATUS_INVALID_SELECTION: "selection is invalid for the current decision", + STATUS_UNSUPPORTED_DECISION: "multi-question decisions are not supported", + } + return CommandEnvelope.from_result( + request, + ok=False, + status=status, + error=error_value(status, messages[status]), + ) + + +def _decision_claim_has_exact_route(claim: Any) -> bool: + return ( + isinstance(getattr(claim, "worker_id", None), str) + and bool(claim.worker_id) + and isinstance(getattr(claim, "worker_fingerprint", None), str) + and bool(claim.worker_fingerprint) + and isinstance(getattr(claim, "binding_private_fingerprint", None), str) + and bool(claim.binding_private_fingerprint) + and isinstance(getattr(claim, "turn_target_value", None), str) + and bool(claim.turn_target_value.strip()) + and isinstance(getattr(claim, "decision_ref", None), str) + and bool(claim.decision_ref) + and getattr(claim, "decision_kind", None) in {"single", "multi", "plan"} + and isinstance(getattr(claim, "option_count", None), int) + and not isinstance(claim.option_count, bool) + and claim.option_count >= 1 + and isinstance(getattr(claim, "option_refs", None), tuple) + and ( + (claim.text is None and bool(claim.option_refs)) + or ( + isinstance(claim.text, str) + and bool(claim.text) + and not claim.option_refs + ) + ) + ) + + +def _same_decision_route(left: Any, right: Any) -> bool: + return ( + _decision_claim_has_exact_route(left) + and _decision_claim_has_exact_route(right) + and ( + left.worker_id, + left.worker_fingerprint, + left.binding_private_fingerprint, + left.turn_target_value, + left.decision_ref, + left.decision_kind, + left.option_count, + left.option_refs, + left.text, + ) + == ( + right.worker_id, + right.worker_fingerprint, + right.binding_private_fingerprint, + right.turn_target_value, + right.decision_ref, + right.decision_kind, + right.option_count, + right.option_refs, + right.text, + ) + ) + + class PreSendCertainty(Enum): """How a pre-send failure must be classified before any external mutation. @@ -537,6 +646,25 @@ def _abandon_pending_claim(config: Config, claim_token: str | None) -> bool: return False +def _abandon_request_reservation( + config: Config, + request: CommandRequest, + reservation: ReservedCommandMutation, +) -> bool: + if config.db_path is None: + return False + try: + return abandon_command_request_reservation( + config.db_path, + host_id=config.host_id, + request_id=request.request_id or "", + canonical_fingerprint=reservation.canonical.fingerprint, + owner_token=reservation.owner_token, + ) + except Exception: + return False + + def _connect_socket( config: Config, request: CommandRequest, @@ -1452,6 +1580,250 @@ def _answer_pending( ) +def _validate_pending_decision( + config: Config, + request: CommandRequest, +) -> Any | PreSendFailure: + if config.db_path is None: + return _safe_transient_pre_send( + _backend_unavailable(request, "pending state store is unavailable") + ) + params = request.params or {} + target = request.target or {} + try: + validated = claim_backend_pending_decision( + config.db_path, + config.host_id, + str(target.get("worker_id") or ""), + str(params.get("decision_ref") or ""), + params.get("selection") + if isinstance(params.get("selection"), Mapping) + else {}, + claim=False, + ) + except Exception: + return _safe_transient_pre_send( + _backend_unavailable(request, "pending state store is unavailable") + ) + if validated.status == "validated" and _decision_claim_has_exact_route(validated): + return validated + status = { + "already_claimed": STATUS_ANSWER_IN_PROGRESS, + "unknown_worker": STATUS_UNKNOWN_WORKER, + "invalid_selection": STATUS_INVALID_SELECTION, + "unsupported_decision": STATUS_UNSUPPORTED_DECISION, + }.get(validated.status, STATUS_DECISION_NOT_PENDING) + return _permanent_pre_send(_decision_failure_envelope(request, status)) + + +def _claim_pending_decision( + config: Config, + request: CommandRequest, + validated: Any, +) -> Any | CommandEnvelope: + assert config.db_path is not None + params = request.params or {} + target = request.target or {} + try: + claim = claim_backend_pending_decision( + config.db_path, + config.host_id, + str(target.get("worker_id") or ""), + str(params.get("decision_ref") or ""), + params.get("selection") + if isinstance(params.get("selection"), Mapping) + else {}, + claim=True, + ) + except Exception: + return _backend_uncertain(request, "pending decision claim state is uncertain") + if ( + claim.status == "claimed" + and isinstance(claim.claim_token, str) + and claim.claim_token + ): + if _same_decision_route(validated, claim): + return claim + status = { + "already_claimed": STATUS_ANSWER_IN_PROGRESS, + "unknown_worker": STATUS_UNKNOWN_WORKER, + "invalid_selection": STATUS_INVALID_SELECTION, + "unsupported_decision": STATUS_UNSUPPORTED_DECISION, + }.get(claim.status, STATUS_DECISION_NOT_PENDING) + return _decision_failure_envelope(request, status) + + +def _submit_decision_calibration( + client: Any, + pane_id: str, + decision: Any, + *, + timeout: float, +) -> None: + steps = calibrate_decision_steps( + kind=decision.decision_kind, + option_count=decision.option_count, + option_refs=decision.option_refs, + text=decision.text, + ) + for step in steps: + if step.operation == "keys": + _socket_request( + client, + "pane.send_keys", + {"pane_id": pane_id, "keys": list(step.keys)}, + timeout=timeout, + ) + elif step.operation == "text": + _socket_request( + client, + "pane.send_text", + {"pane_id": pane_id, "text": step.text}, + timeout=timeout, + ) + else: + _socket_request( + client, + "pane.send_input", + {"pane_id": pane_id, "text": step.text, "keys": list(step.keys)}, + timeout=timeout, + ) + + +def _decision_public_result( + request: CommandRequest, + claim: Any, +) -> dict[str, Any]: + return { + "target": {"worker_id": claim.worker_id}, + "decision": {"decision_ref": (request.params or {}).get("decision_ref")}, + "delivery_state": "submitted", + "transport_state": "submitted", + "observed_pending_state": "pending_observation", + } + + +def _answer_decision( + config: Config, + request: CommandRequest, + validated: Any, + reservation: ReservedCommandMutation, + client: Any, +) -> CommandEnvelope: + assert config.db_path is not None + claim = _claim_pending_decision(config, request, validated) + if isinstance(claim, CommandEnvelope): + _close_socket_client(client) + if claim.status == STATUS_ANSWER_IN_PROGRESS: + _abandon_request_reservation(config, request, reservation) + return _answer_in_progress(request, receipt_reserved=True) + return _finish_before_send(config, request, reservation, claim) + claim_token = claim.claim_token + + send_start_error = _mark_request_send_started( + config, + request, + reservation, + binding_fingerprint=claim.binding_private_fingerprint, + ) + if send_start_error is not None: + _close_socket_client(client) + claim_released = _abandon_pending_claim(config, claim_token) + if send_start_error.status == STATUS_PENDING and not claim_released: + return _finish_before_send( + config, + request, + reservation, + _backend_uncertain( + request, + "pending decision claim could not be safely released", + ), + ) + return send_start_error + + try: + started = start_backend_pending_decision_send( + config.db_path, + config.host_id, + claim_token, + ) + except Exception: + _close_socket_client(client) + _abandon_pending_claim(config, claim_token) + return _finish_request( + config, + request, + reservation, + _backend_uncertain(request, "pending decision start state is uncertain"), + expected_state="send_started", + terminal_state="uncertain", + terminal_effect=_uncertain_pending_effect(config, claim_token), + ) + if getattr(started, "status", None) != "started" or not _same_decision_route(claim, started): + _close_socket_client(client) + _abandon_pending_claim(config, claim_token) + return _finish_request( + config, + request, + reservation, + _backend_uncertain( + request, + "pending decision state is uncertain after send start", + ), + expected_state="send_started", + terminal_state="uncertain", + terminal_effect=_uncertain_pending_effect(config, claim_token), + ) + + try: + _submit_decision_calibration( + client, + started.turn_target_value.strip(), + started, + timeout=config.herdr_timeout_seconds, + ) + except Exception: # noqa: BLE001 + return _finish_request( + config, + request, + reservation, + _backend_uncertain( + request, + "Herdr decision input state is uncertain after send start", + ), + expected_state="send_started", + terminal_state="uncertain", + terminal_effect=_uncertain_pending_effect(config, claim_token), + ) + finally: + _close_socket_client(client) + + accepted = CommandEnvelope.from_result( + request, + ok=True, + status=STATUS_ACCEPTED, + disposition=DISPOSITION_TERMINAL_ACCEPTED, + result=_decision_public_result(request, started), + ) + try: + effect = backend_pending_choice_terminal_effect( + host_id=config.host_id, + claim_token=claim_token, + accepted=True, + ) + except Exception: + return _recover_request(config, request, reservation.canonical) + return _finish_request( + config, + request, + reservation, + accepted, + expected_state="send_started", + terminal_state="accepted", + terminal_effect=effect, + ) + + def _execute_non_mutating(config: Config, request: CommandRequest) -> CommandEnvelope: if request.action == "noop": return execute_command(request, CommandContext(host_id=config.host_id, workers=[])) @@ -1477,6 +1849,17 @@ def _mutation_dry_run(request: CommandRequest) -> CommandEnvelope: }, ) params = request.params or {} + if request.action == "answer_decision": + return CommandEnvelope.from_result( + request, + ok=True, + status=STATUS_DRY_RUN, + result={ + "target": dict(request.target or {}), + "decision": {"decision_ref": params.get("decision_ref")}, + "delivery_state": "not_submitted", + }, + ) return CommandEnvelope.from_result( request, ok=True, @@ -1569,6 +1952,11 @@ def _proven_replay_worker_id( return stored_worker_id or _LEGACY_V0_REPLAY_WORKER_ID if not stored_worker_id: return _receipt_malformed(request) + if request.action == "answer_decision": + explicit_worker_id = _direct_replay_worker_id(request) + if explicit_worker_id != stored_worker_id: + return _duplicate_request(request) + return stored_worker_id if request.action != "send_instruction": return stored_worker_id @@ -1824,6 +2212,11 @@ def submit_command( ) answer_pre_send: PreSendFailure | None = None + validate_answer = ( + _validate_pending_decision + if request.action == "answer_decision" + else _validate_pending_choice + ) if takeover is not None: # Re-driving an abandoned answer reservation: the receipt already fixed # which worker this request answers, so a pending interaction that now @@ -1833,17 +2226,17 @@ def submit_command( request, public_worker_id=existing_worker_id, ) - validated = _validate_pending_choice(config, request) + validated = validate_answer(config, request) if isinstance(validated, PreSendFailure): answer_pre_send = validated elif validated.worker_id != existing_worker_id: answer_pre_send = _permanent_pre_send(_duplicate_request(request)) else: - validated = _validate_pending_choice(config, request) + validated = validate_answer(config, request) if isinstance(validated, PreSendFailure): # No reservation exists yet, so neither a transient nor a permanent # validation failure writes a receipt here. Return it directly. - if health_error is not None: + if health_error is not None and request.action != "answer_decision": return health_error return validated.envelope canonical = build_canonical_mutation( @@ -1857,6 +2250,14 @@ def submit_command( if takeover is not None: return _request_in_progress(request) return answer_pre_send.envelope + if ( + answer_pre_send is not None + and answer_pre_send.envelope.status == STATUS_ANSWER_IN_PROGRESS + ): + # Another request owns the still-live decision claim. Keep this + # abandoned reservation nonterminal so it can take over after that + # claim is released or expires. + return _answer_in_progress(request, receipt_reserved=True) client_or_error: Any | CommandEnvelope | None = None if answer_pre_send is None and health_error is None: @@ -1888,6 +2289,14 @@ def submit_command( health_error, ) assert client_or_error is not None + if request.action == "answer_decision": + return _answer_decision( + config, + request, + validated, + reservation, + client_or_error, + ) return _answer_pending( config, request, diff --git a/src/tendwire/core/commands.py b/src/tendwire/core/commands.py index 048eba7..440c992 100644 --- a/src/tendwire/core/commands.py +++ b/src/tendwire/core/commands.py @@ -34,7 +34,14 @@ COMMAND_ENVELOPE_SCHEMA_VERSION = 2 ALLOWED_ACTIONS = frozenset( - {"noop", "read_snapshot", "resolve_target", "send_instruction", "answer_pending"} + { + "noop", + "read_snapshot", + "resolve_target", + "send_instruction", + "answer_pending", + "answer_decision", + } ) REQUEST_ALLOWED_FIELDS = frozenset( {"schema_version", "action", "request_id", "dry_run", "target", "instruction", "params"} @@ -58,6 +65,11 @@ STATUS_REQUEST_STATE_UNCERTAIN = "request_state_uncertain" STATUS_INVALID_REQUEST = "invalid_request" STATUS_PENDING = "pending" +STATUS_ANSWER_IN_PROGRESS = "answer_in_progress" +STATUS_DECISION_NOT_PENDING = "decision_not_pending" +STATUS_UNKNOWN_WORKER = "unknown_worker" +STATUS_INVALID_SELECTION = "invalid_selection" +STATUS_UNSUPPORTED_DECISION = "unsupported_decision" CommandDisposition = Literal[ "no_receipt", @@ -101,6 +113,11 @@ STATUS_REQUEST_STATE_UNCERTAIN, STATUS_INVALID_REQUEST, STATUS_PENDING, + STATUS_ANSWER_IN_PROGRESS, + STATUS_DECISION_NOT_PENDING, + STATUS_UNKNOWN_WORKER, + STATUS_INVALID_SELECTION, + STATUS_UNSUPPORTED_DECISION, } ) @@ -116,6 +133,10 @@ STATUS_AMBIGUOUS_BACKEND_TARGET, STATUS_BACKEND_FAILED, STATUS_DUPLICATE_REQUEST, + STATUS_DECISION_NOT_PENDING, + STATUS_UNKNOWN_WORKER, + STATUS_INVALID_SELECTION, + STATUS_UNSUPPORTED_DECISION, } ) @@ -133,6 +154,11 @@ STATUS_BACKEND_UNSUPPORTED, STATUS_AMBIGUOUS_BACKEND_TARGET, STATUS_BACKEND_FAILED, + STATUS_ANSWER_IN_PROGRESS, + STATUS_DECISION_NOT_PENDING, + STATUS_UNKNOWN_WORKER, + STATUS_INVALID_SELECTION, + STATUS_UNSUPPORTED_DECISION, } ) @@ -141,6 +167,7 @@ DRY_RUN_MUTATION_NO_RECEIPT_REJECTION_STATUSES = frozenset( { STATUS_INVALID_REQUEST, + STATUS_INVALID_SELECTION, STATUS_REJECTED, STATUS_NOT_FOUND, STATUS_AMBIGUOUS_TARGET, @@ -162,6 +189,7 @@ ANSWER_PENDING_PARAM_FIELDS = frozenset( {"pending_id", "pending_fingerprint", "choice_id"} ) +ANSWER_DECISION_PARAM_FIELDS = frozenset({"decision_ref", "selection"}) # Connector, low-level terminal, routing, and private fields rejected anywhere in a request. FORBIDDEN_REQUEST_FIELDS = FORBIDDEN_FIELD_NAMES @@ -432,8 +460,10 @@ def build_canonical_mutation( """ if not isinstance(request, CommandRequest): raise TypeError("request must be a CommandRequest") - if request.action not in {"send_instruction", "answer_pending"}: - raise ValueError("canonical mutations require send_instruction or answer_pending") + if request.action not in {"send_instruction", "answer_pending", "answer_decision"}: + raise ValueError( + "canonical mutations require send_instruction, answer_pending, or answer_decision" + ) if request.dry_run is not False: raise ValueError("canonical mutations require a non-dry-run request") request_error = validate_request(request) @@ -451,7 +481,7 @@ def build_canonical_mutation( "instruction": {"text": request.instruction["text"]}, "options": {}, } - else: + elif request.action == "answer_pending": assert request.params is not None canonical_payload = { "canonical_version": CANONICAL_MUTATION_VERSION, @@ -464,6 +494,28 @@ def build_canonical_mutation( }, "options": {}, } + else: + assert request.params is not None + selection = request.params["selection"] + if "option_refs" in selection: + canonical_selection: dict[str, Any] = { + "option_refs": sorted( + selection["option_refs"], + key=lambda ref: int(ref) if str(ref).isdigit() else -1, + ) + } + else: + canonical_selection = {"text": selection["text"]} + canonical_payload = { + "canonical_version": CANONICAL_MUTATION_VERSION, + "action": "answer_decision", + "target": {"worker_id": public_worker_id}, + "decision": { + "decision_ref": request.params["decision_ref"], + "selection": canonical_selection, + }, + "options": {}, + } return CanonicalMutation( canonical_version=CANONICAL_MUTATION_VERSION, @@ -510,8 +562,10 @@ def build_selector_proof(request: CommandRequest) -> str: """ if not isinstance(request, CommandRequest): raise TypeError("request must be a CommandRequest") - if request.action not in {"send_instruction", "answer_pending"}: - raise ValueError("selector proofs require send_instruction or answer_pending") + if request.action not in {"send_instruction", "answer_pending", "answer_decision"}: + raise ValueError( + "selector proofs require send_instruction, answer_pending, or answer_decision" + ) if request.dry_run is not False: raise ValueError("selector proofs require a non-dry-run request") request_error = validate_request(request) @@ -580,7 +634,7 @@ def validate_request(request: CommandRequest) -> dict[str, Any] | None: details={"field": "action", "allowed": sorted(ALLOWED_ACTIONS)}, ) if ( - request.action in {"send_instruction", "answer_pending"} + request.action in {"send_instruction", "answer_pending", "answer_decision"} and request.dry_run is False and not is_valid_request_id(request.request_id) ): @@ -669,6 +723,70 @@ def validate_request(request: CommandRequest) -> dict[str, Any] | None: details={"field": f"params.{field}"}, ) + if request.action == "answer_decision": + if request.target is None or set(request.target) != {"worker_id"}: + return error_value( + STATUS_INVALID_REQUEST, + "answer_decision requires exactly target.worker_id", + details={"field": "target"}, + ) + worker_id = request.target.get("worker_id") + if not isinstance(worker_id, str) or not worker_id.strip(): + return error_value( + STATUS_INVALID_REQUEST, + "answer_decision requires nonblank target.worker_id", + details={"field": "target.worker_id"}, + ) + if request.instruction is not None: + return error_value( + STATUS_INVALID_REQUEST, + "answer_decision does not accept an instruction", + details={"field": "instruction"}, + ) + if not isinstance(request.params, dict) or set(request.params) != ANSWER_DECISION_PARAM_FIELDS: + return error_value( + STATUS_INVALID_REQUEST, + "answer_decision params must contain exactly decision_ref and selection", + details={"field": "params"}, + ) + decision_ref = request.params.get("decision_ref") + if not isinstance(decision_ref, str) or not decision_ref.strip(): + return error_value( + STATUS_INVALID_REQUEST, + "answer_decision requires nonblank params.decision_ref", + details={"field": "params.decision_ref"}, + ) + selection = request.params.get("selection") + if not isinstance(selection, Mapping) or len(selection) != 1: + return error_value( + STATUS_INVALID_SELECTION, + "selection must contain exactly one selection form", + details={"field": "params.selection"}, + ) + if set(selection) == {"option_refs"}: + option_refs = selection.get("option_refs") + if not isinstance(option_refs, list) or not option_refs or any( + not isinstance(ref, str) or not ref.strip() for ref in option_refs + ): + return error_value( + STATUS_INVALID_SELECTION, + "selection.option_refs must be a nonempty array of strings", + details={"field": "params.selection.option_refs"}, + ) + elif set(selection) == {"text"}: + if validate_instruction_text(selection.get("text")) is not None: + return error_value( + STATUS_INVALID_SELECTION, + "selection.text must be nonempty safe text", + details={"field": "params.selection.text"}, + ) + else: + return error_value( + STATUS_INVALID_SELECTION, + "selection must contain option_refs or text", + details={"field": "params.selection"}, + ) + return None @@ -788,7 +906,11 @@ def __post_init__(self) -> None: f"command envelope schema_version must be {COMMAND_ENVELOPE_SCHEMA_VERSION}" ) - mutating = self.action in {"send_instruction", "answer_pending"} + mutating = self.action in { + "send_instruction", + "answer_pending", + "answer_decision", + } live_mutation = mutating and self.dry_run is False if live_mutation and not is_valid_request_id(self.request_id): raise ValueError("non-dry-run mutation requires a valid request_id") @@ -809,12 +931,15 @@ def __post_init__(self) -> None: if self.ok and clean_error is not None: raise ValueError("successful command envelope must not include an error") - if self.disposition == DISPOSITION_NO_RECEIPT: if live_mutation: valid_no_receipt_tuple = ( not self.ok and self.status in LIVE_MUTATION_NO_RECEIPT_REJECTION_STATUSES + and ( + self.status != STATUS_ANSWER_IN_PROGRESS + or self.action == "answer_decision" + ) ) elif mutating: valid_no_receipt_tuple = ( @@ -829,7 +954,16 @@ def __post_init__(self) -> None: if not valid_no_receipt_tuple: raise ValueError("no_receipt disposition has an inconsistent command tuple") elif self.disposition == DISPOSITION_IN_PROGRESS: - if not mutating or self.dry_run or self.ok or self.status != STATUS_PENDING: + if ( + not mutating + or self.dry_run + or self.ok + or self.status not in {STATUS_PENDING, STATUS_ANSWER_IN_PROGRESS} + or ( + self.status == STATUS_ANSWER_IN_PROGRESS + and self.action != "answer_decision" + ) + ): raise ValueError("in_progress disposition has an inconsistent command tuple") elif self.disposition == DISPOSITION_TERMINAL_ACCEPTED: if not mutating or self.dry_run or not self.ok or self.status != STATUS_ACCEPTED: @@ -850,6 +984,16 @@ def __post_init__(self) -> None: or self.status != STATUS_REQUEST_STATE_UNCERTAIN ): raise ValueError("terminal_uncertain disposition has an inconsistent command tuple") + if self.status == STATUS_ANSWER_IN_PROGRESS and not ( + self.action == "answer_decision" + and live_mutation + and not self.ok + and self.disposition + in {DISPOSITION_NO_RECEIPT, DISPOSITION_IN_PROGRESS} + ): + raise ValueError( + "answer_in_progress has an inconsistent command tuple" + ) def to_dict(self) -> dict[str, Any]: return { @@ -939,7 +1083,11 @@ def from_error(cls, request: CommandRequest | None, error: dict[str, Any]) -> "C result=None, error=error, ) - mutating = request.action in {"send_instruction", "answer_pending"} + mutating = request.action in { + "send_instruction", + "answer_pending", + "answer_decision", + } valid_mutation_id = is_valid_request_id(request.request_id) request_id = ( request.request_id diff --git a/src/tendwire/core/models.py b/src/tendwire/core/models.py index 0b85378..33fdc05 100644 --- a/src/tendwire/core/models.py +++ b/src/tendwire/core/models.py @@ -474,6 +474,7 @@ "label", "message", "name", + "prompt", "question", "raw_status", "reason", diff --git a/src/tendwire/core/turns.py b/src/tendwire/core/turns.py index c3b2d3a..089b2f4 100644 --- a/src/tendwire/core/turns.py +++ b/src/tendwire/core/turns.py @@ -1577,6 +1577,7 @@ def _is_pending_routing_meta_key(key: Any) -> bool: "open_prompt", "read_succeeded_no_prompt", "read_succeeded_invalid_prompt", + "read_succeeded_unsupported_decision", "read_failed", "worker_authoritatively_absent", ] @@ -1615,12 +1616,17 @@ class PendingObservation: pending_kind: str | None = None choices: tuple[PendingObservedChoice, ...] = () revision_digest: str | None = None + decision_kind: Literal["single", "multi", "plan"] | None = None + decision_options: tuple[str, ...] = () + decision_multi_select: bool = False + decision_question_count: int = 0 def __post_init__(self) -> None: if self.kind not in { "open_prompt", "read_succeeded_no_prompt", "read_succeeded_invalid_prompt", + "read_succeeded_unsupported_decision", "read_failed", "worker_authoritatively_absent", }: @@ -1648,11 +1654,44 @@ def __post_init__(self) -> None: or not self.pending_kind.strip() ): raise ValueError("invalid pending observation prompt kind") + if self.decision_kind is None: + if ( + self.decision_options + or self.decision_multi_select + or self.decision_question_count != 0 + ): + raise ValueError("non-decision prompt cannot carry decision data") + else: + if self.decision_kind not in {"single", "multi", "plan"}: + raise ValueError("invalid pending observation decision kind") + if ( + not isinstance(self.decision_options, tuple) + or not self.decision_options + or not all( + isinstance(label, str) and label.strip() + for label in self.decision_options + ) + ): + raise ValueError("decision prompt requires options") + if self.decision_multi_select is not ( + self.decision_kind == "multi" + ): + raise ValueError("decision multi-select flag does not match kind") + if ( + not isinstance(self.decision_question_count, int) + or isinstance(self.decision_question_count, bool) + or self.decision_question_count < 1 + ): + raise ValueError("invalid decision question count") elif ( self.question is not None or self.pending_kind is not None or self.choices or self.revision_digest is not None + or self.decision_kind is not None + or self.decision_options + or self.decision_multi_select + or self.decision_question_count != 0 ): raise ValueError("non-open pending observation cannot carry prompt data") diff --git a/src/tendwire/store/sqlite.py b/src/tendwire/store/sqlite.py index 4f3a89e..137df0e 100644 --- a/src/tendwire/store/sqlite.py +++ b/src/tendwire/store/sqlite.py @@ -54,7 +54,7 @@ verify_created_private_sqlite_replacement_at, verify_entry_identity, ) -from ..core.commands import CommandEnvelope, is_selector_proof +from ..core.commands import CommandEnvelope, is_selector_proof, validate_instruction_text from ..core.models import ( FINGERPRINT_HEX_CHARS, SCHEMA_VERSION, @@ -193,6 +193,50 @@ class BackendPendingChoiceSend: picker_ordinal: int | None = None +@dataclass(frozen=True) +class BackendPendingDecisionClaim: + status: Literal[ + "claimed", + "validated", + "unknown_worker", + "decision_not_pending", + "invalid_selection", + "unsupported_decision", + "already_claimed", + ] + claim_token: str | None = None + worker_id: str | None = None + worker_fingerprint: str | None = None + binding_private_fingerprint: str | None = None + turn_target_value: str | None = None + decision_ref: str | None = None + decision_kind: Literal["single", "multi", "plan"] | None = None + option_count: int | None = None + option_refs: tuple[str, ...] = () + text: str | None = None + + +@dataclass(frozen=True) +class BackendPendingDecisionSend: + status: Literal[ + "started", + "not_found", + "stale", + "changed", + "binding_changed", + "already_started", + ] + worker_id: str | None = None + worker_fingerprint: str | None = None + binding_private_fingerprint: str | None = None + turn_target_value: str | None = None + decision_ref: str | None = None + decision_kind: Literal["single", "multi", "plan"] | None = None + option_count: int | None = None + option_refs: tuple[str, ...] = () + text: str | None = None + + @dataclass(frozen=True) class SnapshotRetentionPolicy: """Database-wide snapshot history bounds.""" @@ -15625,6 +15669,57 @@ def _update_persisted_turn_row( return material_changed, item +def _normalize_pending_decision_meta(value: Any) -> dict[str, Any]: + """Validate the connector-facing semantic decision contract.""" + if not isinstance(value, Mapping) or set(value) != { + "decision_ref", + "kind", + "prompt", + "options", + "multi_select", + "question_count", + }: + raise ValueError("invalid backend pending decision") + decision_ref = value.get("decision_ref") + kind = value.get("kind") + prompt = value.get("prompt") + options = value.get("options") + multi_select = value.get("multi_select") + question_count = value.get("question_count") + if ( + type(decision_ref) is not str + or not decision_ref.strip() + or kind not in {"single", "multi", "plan"} + or type(prompt) is not str + or not prompt.strip() + or not isinstance(options, list) + or not options + or not isinstance(multi_select, bool) + or multi_select is not (kind == "multi") + or not isinstance(question_count, int) + or isinstance(question_count, bool) + or question_count < 1 + ): + raise ValueError("invalid backend pending decision") + normalized_options: list[dict[str, str]] = [] + for ordinal, option in enumerate(options, 1): + if not isinstance(option, Mapping) or set(option) != {"ref", "label"}: + raise ValueError("invalid backend pending decision option") + ref = option.get("ref") + label = option.get("label") + if ref != str(ordinal) or type(label) is not str or not label.strip(): + raise ValueError("invalid backend pending decision option") + normalized_options.append({"ref": ref, "label": label}) + return { + "decision_ref": decision_ref, + "kind": kind, + "prompt": prompt, + "options": normalized_options, + "multi_select": multi_select, + "question_count": question_count, + } + + def _normalize_backend_pending_payload( pending: Mapping[str, Any], choice_routes: tuple[tuple[str, int], ...], @@ -15673,11 +15768,16 @@ def _normalize_backend_pending_payload( type(ordinal) is not int or ordinal < 1 for ordinal in route_map.values() ): raise ValueError("backend pending choices do not match private routes") + normalized_meta = sanitize_public_mapping(meta) + if "decision" in normalized_meta: + normalized_meta["decision"] = _normalize_pending_decision_meta( + normalized_meta["decision"] + ) normalized = { "question": question, "kind": kind, "choices": normalized_choices, - "meta": sanitize_public_mapping(meta), + "meta": normalized_meta, } return normalized, _canonical_json(route_map) @@ -15894,6 +15994,60 @@ def _apply_backend_pending_observation_conn( ) return True + if observation.kind == "read_succeeded_unsupported_decision": + unsupported_payload = _canonical_json({"unsupported_decision": True}) + changed = row is None or ( + str(row[0]), + str(row[5]), + str(row[6]), + stored_binding, + stored_target, + ) != ( + unsupported_payload, + "invalid", + "fresh", + effective_binding, + effective_target, + ) + conn.execute( + """ + INSERT INTO backend_pending ( + host_id, worker_id, payload_json, observed_at, revision_digest, + choice_routes_json, binding_private_fingerprint, + observed_turn_target_value, observation_state, freshness, + last_success_at, last_failure_at, grace_deadline, updated_at + ) VALUES (?, ?, ?, ?, '', '{}', ?, ?, 'invalid', 'fresh', + ?, NULL, NULL, ?) + ON CONFLICT(host_id, worker_id) DO UPDATE SET + payload_json = excluded.payload_json, + observed_at = excluded.observed_at, + revision_digest = '', choice_routes_json = '{}', + binding_private_fingerprint = + excluded.binding_private_fingerprint, + observed_turn_target_value = + excluded.observed_turn_target_value, + observation_state = 'invalid', freshness = 'fresh', + last_success_at = excluded.last_success_at, + last_failure_at = NULL, grace_deadline = NULL, + updated_at = excluded.updated_at + """, + ( + *key, + unsupported_payload, + current_time, + effective_binding, + effective_target, + current_time, + current_time, + ), + ) + conn.execute( + "DELETE FROM backend_pending_claims " + "WHERE host_id = ? AND worker_id = ?", + key, + ) + return changed + if observation.kind == "read_succeeded_invalid_prompt": changed = row is None or ( str(row[5]), @@ -16004,6 +16158,19 @@ def _apply_backend_pending_observation_conn( ) for choice in observation.choices ) + public_meta: dict[str, Any] = {"source": "backend"} + if observation.decision_kind is not None: + public_meta["decision"] = { + "decision_ref": f"decision-{revision_digest}", + "kind": observation.decision_kind, + "prompt": observation.question, + "options": [ + {"ref": str(ordinal), "label": label} + for ordinal, label in enumerate(observation.decision_options, 1) + ], + "multi_select": observation.decision_multi_select, + "question_count": observation.decision_question_count, + } public_payload = { "question": observation.question, "kind": observation.pending_kind or "question", @@ -16011,7 +16178,7 @@ def _apply_backend_pending_observation_conn( {"choice_id": choice_id, "label": label} for choice_id, label, _ordinal in persisted_choices ], - "meta": {"source": "backend"}, + "meta": public_meta, } routes = tuple( (choice_id, ordinal) @@ -16116,12 +16283,27 @@ def _merge_backend_pending_conn( for ordinal, choice in enumerate(normalized_choices, 1) ) question = str(clean.get("question") or clean.get("kind") or "Pending action") + decision: dict[str, Any] | None = None + clean_meta = clean.get("meta") + if isinstance(clean_meta, Mapping) and "decision" in clean_meta: + decision = _normalize_pending_decision_meta(clean_meta["decision"]) observation = PendingObservation( "open_prompt", question=question, pending_kind=str(clean.get("kind") or "question"), choices=choices, revision_digest=stable_fingerprint({"legacy_pending_revision": clean}), + decision_kind=(decision or {}).get("kind"), + decision_options=tuple( + str(option["label"]) + for option in (decision or {}).get("options", []) + ), + decision_multi_select=bool( + (decision or {}).get("multi_select", False) + ), + decision_question_count=int( + (decision or {}).get("question_count", 0) + ), ) return _apply_backend_pending_observation_conn( conn, @@ -16453,10 +16635,22 @@ def pending_payload_from_store( updated_at, observation_state, ) in backend_rows: - if str(observation_state) == "none": + state = str(observation_state) + if state == "none": suppressed_worker_ids.add(str(worker_id)) continue - if str(observation_state) != "open": + if state == "invalid": + try: + invalid_payload = json.loads(payload_json) + except (TypeError, ValueError): + invalid_payload = None + if ( + isinstance(invalid_payload, Mapping) + and invalid_payload.get("unsupported_decision") is True + ): + suppressed_worker_ids.add(str(worker_id)) + continue + if state != "open": continue try: pending = json.loads(payload_json) @@ -16719,6 +16913,372 @@ def claim_backend_pending_choice( raise +_DECISION_CLAIM_PREFIX = "decision:" + + +def _decision_from_pending_row( + payload_json: Any, + routes_json: Any, + revision_digest: Any, +) -> dict[str, Any] | None: + try: + payload = json.loads(payload_json) + routes = json.loads(routes_json) + if not isinstance(payload, Mapping) or not isinstance(routes, Mapping): + return None + normalized, _ = _normalize_backend_pending_payload( + payload, + tuple((str(key), int(value)) for key, value in routes.items()), + ) + meta = normalized.get("meta") + decision = meta.get("decision") if isinstance(meta, Mapping) else None + if not isinstance(decision, Mapping): + return None + normalized_decision = _normalize_pending_decision_meta(decision) + if normalized_decision["decision_ref"] != f"decision-{revision_digest}": + return None + return normalized_decision + except (TypeError, ValueError): + return None + + +def _validated_decision_selection( + decision: Mapping[str, Any], + selection: Any, +) -> tuple[tuple[str, ...], str | None] | None: + if not isinstance(selection, Mapping) or len(selection) != 1: + return None + kind = decision.get("kind") + valid_refs = { + str(option.get("ref")) + for option in decision.get("options", []) + if isinstance(option, Mapping) + } + if set(selection) == {"option_refs"}: + raw_refs = selection.get("option_refs") + if not isinstance(raw_refs, list) or not raw_refs or any( + type(ref) is not str for ref in raw_refs + ): + return None + refs = tuple(raw_refs) + if len(refs) != len(set(refs)) or any(ref not in valid_refs for ref in refs): + return None + if kind in {"single", "plan"} and len(refs) != 1: + return None + if kind == "multi" and len(refs) < 1: + return None + return refs, None + if set(selection) == {"text"}: + text = selection.get("text") + if kind != "single" or validate_instruction_text(text) is not None: + return None + return (), str(text) + return None + + +def _encode_decision_claim_selection( + option_refs: tuple[str, ...], + text: str | None, +) -> str: + selection: dict[str, Any] + if text is None: + selection = {"option_refs": list(option_refs)} + else: + selection = {"text": text} + return _DECISION_CLAIM_PREFIX + _canonical_json(selection) + + +def _decode_decision_claim_selection(value: Any) -> Mapping[str, Any] | None: + if not isinstance(value, str) or not value.startswith(_DECISION_CLAIM_PREFIX): + return None + try: + decoded = json.loads(value[len(_DECISION_CLAIM_PREFIX) :]) + except (TypeError, ValueError): + return None + return decoded if isinstance(decoded, Mapping) else None + + +def claim_backend_pending_decision( + db_path: Path | str, + host_id: str, + worker_id: str, + decision_ref: str, + selection: Mapping[str, Any], + *, + claim: bool = True, + observed_at: str | None = None, + claim_lease_seconds: float = BACKEND_PENDING_CLAIM_LEASE_SECONDS, +) -> BackendPendingDecisionClaim: + """Validate and optionally claim one worker's exact current decision.""" + if not _sqlite_store_exists(db_path): + return BackendPendingDecisionClaim("decision_not_pending") + current_time, _ = _pending_observed_time(observed_at) + with _connect(db_path, isolation_level=None) as conn: + _ensure_schema(conn) + conn.execute("BEGIN IMMEDIATE" if claim else "BEGIN") + try: + snapshot_row = conn.execute( + "SELECT payload FROM snapshots WHERE host_id = ? ORDER BY id DESC LIMIT 1", + (str(host_id),), + ).fetchone() + try: + snapshot = ( + Snapshot.from_dict(json.loads(snapshot_row[0])) + if snapshot_row is not None + else None + ) + except Exception: + snapshot = None + worker = next( + ( + item + for item in (snapshot.workers if snapshot is not None else []) + if item.id == str(worker_id) + ), + None, + ) + if worker is None or worker.status in {"closed", "failed", "unknown"}: + conn.rollback() + return BackendPendingDecisionClaim("unknown_worker") + row = conn.execute( + """ + SELECT payload_json, choice_routes_json, revision_digest, + freshness, binding_private_fingerprint, + observed_turn_target_value, observation_state + FROM backend_pending + WHERE host_id = ? AND worker_id = ? + """, + (str(host_id), str(worker_id)), + ).fetchone() + if row is not None and str(row[6]) == "invalid": + try: + unsupported = json.loads(str(row[0])) + except (TypeError, ValueError): + unsupported = None + if ( + isinstance(unsupported, Mapping) + and unsupported.get("unsupported_decision") is True + ): + conn.rollback() + return BackendPendingDecisionClaim("unsupported_decision") + if row is None or str(row[6]) != "open" or str(row[3]) != "fresh": + conn.rollback() + return BackendPendingDecisionClaim("decision_not_pending") + context = _backend_pending_claim_context_conn( + conn, + str(host_id), + str(worker_id), + str(row[4]), + str(row[5]), + observed_at=current_time, + ) + decision = _decision_from_pending_row(row[0], row[1], row[2]) + if ( + context is None + or decision is None + or decision.get("decision_ref") != str(decision_ref) + ): + conn.rollback() + return BackendPendingDecisionClaim("decision_not_pending") + if int(decision["question_count"]) > 1: + conn.rollback() + return BackendPendingDecisionClaim("unsupported_decision") + validated_selection = _validated_decision_selection(decision, selection) + if validated_selection is None: + conn.rollback() + return BackendPendingDecisionClaim("invalid_selection") + option_refs, text = validated_selection + existing = conn.execute( + """ + SELECT state, claimed_at + FROM backend_pending_claims + WHERE host_id = ? AND worker_id = ? + """, + (str(host_id), str(worker_id)), + ).fetchone() + if existing is not None: + reclaimable = ( + str(existing[0]) == "claimed" + and _backend_pending_claim_expired( + existing[1], current_time, claim_lease_seconds + ) + ) + if reclaimable and claim: + conn.execute( + """ + DELETE FROM backend_pending_claims + WHERE host_id = ? AND worker_id = ? AND state = 'claimed' + AND claimed_at = ? + """, + (str(host_id), str(worker_id), str(existing[1])), + ) + elif not reclaimable: + conn.rollback() + return BackendPendingDecisionClaim("already_claimed") + option_count = len(decision["options"]) + picker_ordinal = int(option_refs[0]) if option_refs else option_count + 1 + fields = { + "worker_id": str(worker_id), + "worker_fingerprint": context[2], + "binding_private_fingerprint": context[1], + "turn_target_value": context[3], + "decision_ref": str(decision_ref), + "decision_kind": decision["kind"], + "option_count": option_count, + "option_refs": option_refs, + "text": text, + } + if not claim: + conn.rollback() + return BackendPendingDecisionClaim("validated", **fields) + token = secrets.token_urlsafe(32) + conn.execute( + """ + INSERT INTO backend_pending_claims ( + host_id, worker_id, claim_token, revision_digest, choice_id, + picker_ordinal, worker_fingerprint, + binding_private_fingerprint, turn_target_value, state, + claimed_at, send_started_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'claimed', ?, NULL) + """, + ( + str(host_id), str(worker_id), token, str(row[2]), + _encode_decision_claim_selection(option_refs, text), + picker_ordinal, context[2], context[1], context[3], current_time, + ), + ) + conn.commit() + return BackendPendingDecisionClaim( + "claimed", claim_token=token, **fields + ) + except Exception: + conn.rollback() + raise + + +def start_backend_pending_decision_send( + db_path: Path | str, + host_id: str, + claim_token: str, + *, + observed_at: str | None = None, + claim_lease_seconds: float = BACKEND_PENDING_CLAIM_LEASE_SECONDS, +) -> BackendPendingDecisionSend: + """CAS a decision claim against its current prompt and private binding.""" + if not _sqlite_store_exists(db_path): + return BackendPendingDecisionSend("not_found") + current_time, _ = _pending_observed_time(observed_at) + with _connect(db_path, isolation_level=None) as conn: + _ensure_schema(conn) + conn.execute("BEGIN IMMEDIATE") + try: + row = conn.execute( + """ + SELECT worker_id, revision_digest, choice_id, + worker_fingerprint, binding_private_fingerprint, + turn_target_value, state, claimed_at + FROM backend_pending_claims + WHERE host_id = ? AND claim_token = ? + """, + (str(host_id), str(claim_token)), + ).fetchone() + if row is None: + conn.rollback() + return BackendPendingDecisionSend("not_found") + if ( + str(row[6]) == "claimed" + and _backend_pending_claim_expired( + row[7], current_time, claim_lease_seconds + ) + ): + conn.execute( + """ + DELETE FROM backend_pending_claims + WHERE host_id = ? AND claim_token = ? AND state = 'claimed' + """, + (str(host_id), str(claim_token)), + ) + conn.commit() + return BackendPendingDecisionSend("not_found") + current = conn.execute( + """ + SELECT payload_json, choice_routes_json, revision_digest, + freshness, binding_private_fingerprint, + observed_turn_target_value + FROM backend_pending + WHERE host_id = ? AND worker_id = ? + AND observation_state = 'open' + """, + (str(host_id), str(row[0])), + ).fetchone() + if current is None: + conn.rollback() + return BackendPendingDecisionSend("changed") + if str(current[3]) != "fresh": + conn.rollback() + return BackendPendingDecisionSend("stale") + decision = _decision_from_pending_row(current[0], current[1], current[2]) + selection = _decode_decision_claim_selection(row[2]) + validated_selection = ( + _validated_decision_selection(decision, selection) + if decision is not None and selection is not None + else None + ) + if ( + str(current[2]) != str(row[1]) + or str(current[4]) != str(row[4]) + or str(current[5]) != str(row[5]) + or decision is None + or validated_selection is None + ): + conn.rollback() + return BackendPendingDecisionSend("changed") + option_refs, text = validated_selection + fields = { + "worker_id": str(row[0]), + "worker_fingerprint": str(row[3]), + "binding_private_fingerprint": str(row[4]), + "turn_target_value": str(row[5]), + "decision_ref": str(decision["decision_ref"]), + "decision_kind": decision["kind"], + "option_count": len(decision["options"]), + "option_refs": option_refs, + "text": text, + } + if str(row[6]) == "send_started": + conn.rollback() + return BackendPendingDecisionSend("already_started", **fields) + context = _backend_pending_claim_context_conn( + conn, + str(host_id), + str(row[0]), + str(row[4]), + str(row[5]), + observed_at=current_time, + ) + if context is None or (context[1], context[2], context[3]) != ( + str(row[4]), str(row[3]), str(row[5]) + ): + conn.rollback() + return BackendPendingDecisionSend("binding_changed") + cursor = conn.execute( + """ + UPDATE backend_pending_claims + SET state = 'send_started', send_started_at = ? + WHERE host_id = ? AND claim_token = ? AND state = 'claimed' + """, + (current_time, str(host_id), str(claim_token)), + ) + if cursor.rowcount != 1: + conn.rollback() + return BackendPendingDecisionSend("changed") + conn.commit() + return BackendPendingDecisionSend("started", **fields) + except Exception: + conn.rollback() + raise + + def start_backend_pending_choice_send( db_path: Path | str, host_id: str, @@ -20703,6 +21263,50 @@ def reserve_command_request( conn.close() +def abandon_command_request_reservation( + db_path: Path, + *, + host_id: str, + request_id: str, + canonical_fingerprint: str, + owner_token: str, + now: str | None = None, +) -> bool: + """Release only the caller's unsent reservation for immediate takeover.""" + if not _sqlite_store_exists(db_path): + return False + current = _command_request_now(now) + owner_hash = _owner_token_hash(owner_token) + if not owner_hash: + return False + with _connect(db_path, isolation_level=None) as conn: + _ensure_schema(conn) + conn.execute("BEGIN IMMEDIATE") + try: + updated = conn.execute( + """ + UPDATE command_receipts + SET owner_expires_at = ?, updated_at = ? + WHERE host_id = ? AND request_id = ? + AND canonical_fingerprint = ? + AND state = 'reserved' AND owner_token_hash = ? + """, + ( + current, + current, + str(host_id), + str(request_id), + str(canonical_fingerprint), + owner_hash, + ), + ) + conn.commit() + return int(updated.rowcount or 0) == 1 + except Exception: + conn.rollback() + raise + + def reserve_terminal_command_replay( db_path: Path, *, @@ -20952,6 +21556,7 @@ def finish_command_request( ) if terminal == "rejected" and str(status) in { "pending", + "answer_in_progress", "accepted", "request_state_uncertain", }: diff --git a/tests/test_answer_decision.py b/tests/test_answer_decision.py new file mode 100644 index 0000000..c57499c --- /dev/null +++ b/tests/test_answer_decision.py @@ -0,0 +1,856 @@ +"""Semantic connector answers for current backend-owned Claude decisions.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import pytest +import tendwire.command_submission as command_submission + +from tendwire.backends.herdr_decision import calibrate_decision_steps +from tendwire.backends.herdr_turns import ( + PENDING_DECISION_MAX_OPTIONS, + _pending_observation_from_turn, +) +from tendwire.command_submission import submit_command +from tendwire.core.commands import ( + DISPOSITION_IN_PROGRESS, + DISPOSITION_NO_RECEIPT, + STATUS_ACCEPTED, + STATUS_ANSWER_IN_PROGRESS, + STATUS_DECISION_NOT_PENDING, + STATUS_INVALID_SELECTION, + STATUS_PENDING, + STATUS_REQUEST_STATE_UNCERTAIN, + STATUS_UNKNOWN_WORKER, + STATUS_UNSUPPORTED_DECISION, + CommandRequest, + build_canonical_mutation, +) +from tendwire.core.models import Worker +from tendwire.store.sqlite import ( + abandon_backend_pending_choice_claim, + apply_backend_pending_observation, + claim_backend_pending_decision, + envelope_to_receipt_json, + get_command_request, + pending_payload_from_store, + reserve_command_request, +) + +from tests.test_command_submission import ( + _FakeSocketClient, + _binding, + _config, + _factory, + _seed, +) + + +def _decision_turn( + *, + prompt: str = "Choose a database", + kind: str = "AskUserQuestion", + multi_select: bool = False, + question_count: int = 1, +) -> dict[str, Any]: + return { + "pending_decision": { + "decision_id": "private-tool-use", + "kind": kind, + "question": prompt, + "options": [ + {"id": "postgres", "label": "Postgres"}, + {"id": "sqlite", "label": "SQLite"}, + {"id": "duckdb", "label": "DuckDB"}, + {"id": "mysql", "label": "MySQL"}, + ], + "multi_select": multi_select, + "question_count": question_count, + } + } + + +def _seed_pending_decision( + tmp_path: Path, + *, + turn: dict[str, Any] | None = None, +) -> tuple[Any, Worker, str]: + config = _config(tmp_path) + worker = Worker(id="w-1", name="Alpha", status="active") + binding = _binding( + worker, + private_fingerprint="decision-binding-private", + turn_target_value="decision-pane-private", + ) + _seed(config, [worker], [binding]) + assert config.db_path is not None + observation = _pending_observation_from_turn(turn or _decision_turn()) + assert apply_backend_pending_observation( + config.db_path, + config.host_id, + worker.id, + observation, + binding_private_fingerprint=binding.private_fingerprint, + observed_turn_target_value=binding.turn_target_value, + ) + payload = pending_payload_from_store(config.db_path, config.host_id) + row = next(item for item in payload["pending_interactions"] if item["worker_id"] == worker.id) + return config, worker, row["meta"]["decision"]["decision_ref"] + + +def _answer_request( + decision_ref: str, + *, + request_id: str = "decision-request-1", + worker_id: str = "w-1", + selection: dict[str, Any] | None = None, +) -> dict[str, Any]: + return { + "schema_version": 1, + "action": "answer_decision", + "request_id": request_id, + "dry_run": False, + "target": {"worker_id": worker_id}, + "params": { + "decision_ref": decision_ref, + "selection": selection or {"option_refs": ["2"]}, + }, + } + + +def test_pending_payload_carries_stable_structured_decision_and_rotates_ref( + tmp_path: Path, +) -> None: + config, worker, first_ref = _seed_pending_decision(tmp_path) + assert config.db_path is not None + first = pending_payload_from_store(config.db_path, config.host_id) + first_row = next(item for item in first["pending_interactions"] if item["worker_id"] == worker.id) + assert first_row["meta"]["decision"] == { + "decision_ref": first_ref, + "kind": "single", + "prompt": "Choose a database", + "options": [ + {"ref": "1", "label": "Postgres"}, + {"ref": "2", "label": "SQLite"}, + {"ref": "3", "label": "DuckDB"}, + {"ref": "4", "label": "MySQL"}, + ], + "multi_select": False, + "question_count": 1, + } + + binding = _binding( + worker, + private_fingerprint="decision-binding-private", + turn_target_value="decision-pane-private", + ) + changed = _pending_observation_from_turn( + _decision_turn(prompt="Choose a durable database") + ) + assert apply_backend_pending_observation( + config.db_path, + config.host_id, + worker.id, + changed, + binding_private_fingerprint=binding.private_fingerprint, + observed_turn_target_value=binding.turn_target_value, + ) + second = pending_payload_from_store(config.db_path, config.host_id) + second_row = next(item for item in second["pending_interactions"] if item["worker_id"] == worker.id) + assert second_row["meta"]["decision"]["decision_ref"] != first_ref + + +def test_answer_decision_stale_ref_fails_before_pane_io(tmp_path: Path) -> None: + config, _worker, _decision_ref = _seed_pending_decision(tmp_path) + calls: list[dict[str, Any]] = [] + result = submit_command( + config, + _answer_request("decision-stale"), + socket_client_factory=_factory(calls), + ) + assert result.ok is False + assert result.status == STATUS_DECISION_NOT_PENDING + assert calls == [] + + +def test_answer_decision_omitted_dry_run_is_preview_only(tmp_path: Path) -> None: + config, _worker, decision_ref = _seed_pending_decision(tmp_path) + calls: list[dict[str, Any]] = [] + request = _answer_request(decision_ref) + request.pop("dry_run") + + result = submit_command( + config, + request, + socket_client_factory=_factory(calls), + ) + + assert result.ok is True + assert result.status == "dry_run" + assert result.dry_run is True + assert calls == [] + + +def test_answer_decision_unknown_worker_fails_before_pane_io(tmp_path: Path) -> None: + config = _config(tmp_path) + worker = Worker(id="w-1", name="Alpha", status="active") + _seed(config, [worker], [_binding(worker)]) + calls: list[dict[str, Any]] = [] + result = submit_command( + config, + _answer_request("decision-any", worker_id="worker-missing"), + socket_client_factory=_factory(calls), + ) + assert result.ok is False + assert result.status == STATUS_UNKNOWN_WORKER + assert calls == [] + + +@pytest.mark.parametrize( + "selection", + [ + {"option_refs": ["9"]}, + {"option_refs": ["1", "2"]}, + {"option_refs": ["2", "2"]}, + {"text": ""}, + {"option_refs": ["1"], "text": "both"}, + ], +) +def test_answer_decision_invalid_selection_fails_before_pane_io( + tmp_path: Path, + selection: dict[str, Any], +) -> None: + config, _worker, decision_ref = _seed_pending_decision(tmp_path) + calls: list[dict[str, Any]] = [] + result = submit_command( + config, + _answer_request(decision_ref, selection=selection), + socket_client_factory=_factory(calls), + ) + assert result.ok is False + assert result.status == STATUS_INVALID_SELECTION + assert calls == [] + + +def test_answer_decision_refuses_multi_question_before_pane_io(tmp_path: Path) -> None: + config, _worker, decision_ref = _seed_pending_decision( + tmp_path, + turn=_decision_turn(question_count=2), + ) + calls: list[dict[str, Any]] = [] + result = submit_command( + config, + _answer_request(decision_ref), + socket_client_factory=_factory(calls), + ) + assert result.ok is False + assert result.status == STATUS_UNSUPPORTED_DECISION + assert calls == [] + + +def test_answer_decision_rejects_plan_write_in_before_pane_io(tmp_path: Path) -> None: + config, _worker, decision_ref = _seed_pending_decision( + tmp_path, + turn=_decision_turn(kind="ExitPlanMode"), + ) + calls: list[dict[str, Any]] = [] + result = submit_command( + config, + _answer_request(decision_ref, selection={"text": "Revise this plan"}), + socket_client_factory=_factory(calls), + ) + assert result.ok is False + assert result.status == STATUS_INVALID_SELECTION + assert calls == [] + + +def test_decision_calibration_steps_cover_single_plan_write_in_and_multi() -> None: + single = calibrate_decision_steps( + kind="single", option_count=4, option_refs=("2",) + ) + assert [(item.operation, item.keys, item.text) for item in single] == [ + ("keys", ("2", "Enter"), None) + ] + + plan = calibrate_decision_steps( + kind="plan", option_count=2, option_refs=("1",) + ) + assert [(item.operation, item.keys, item.text) for item in plan] == [ + ("keys", ("1", "Enter"), None) + ] + + write_in = calibrate_decision_steps( + kind="single", option_count=4, text="Use another database" + ) + assert [(item.operation, item.keys, item.text) for item in write_in] == [ + ("keys", ("5",), None), + ("input", ("Enter",), "Use another database"), + ] + + multi = calibrate_decision_steps( + kind="multi", option_count=4, option_refs=("3", "1") + ) + assert [(item.operation, item.keys, item.text) for item in multi] == [ + ("keys", ("Down", "Down"), None), + ("text", (), " "), + ("keys", ("Up", "Up"), None), + ("text", (), " "), + ("keys", ("Down", "Down", "Down", "Down", "Enter"), None), + ] + + +def test_production_shaped_multi_decision_end_to_end(tmp_path: Path) -> None: + tool_input = { + "questions": [ + { + "question": "Which databases should we support?", + "header": "Database support", + "options": [ + {"label": "Postgres", "description": "Primary database"}, + {"label": "SQLite", "description": "Local database"}, + {"label": "DuckDB", "description": "Analytics database"}, + {"label": "MySQL", "description": "Compatibility database"}, + ], + "multiSelect": True, + } + ] + } + question = tool_input["questions"][0] + adapter_pending_decision = { + "decision_id": "toolu_multi_123", + "prompt": f'{question["header"]}: {question["question"]}', + "mode": "multi", + "multi_select": question["multiSelect"], + "options": [ + {"id": str(ordinal), "label": option["label"]} + for ordinal, option in enumerate(question["options"], 1) + ], + } + assert adapter_pending_decision == { + "decision_id": "toolu_multi_123", + "prompt": "Database support: Which databases should we support?", + "mode": "multi", + "multi_select": True, + "options": [ + {"id": "1", "label": "Postgres"}, + {"id": "2", "label": "SQLite"}, + {"id": "3", "label": "DuckDB"}, + {"id": "4", "label": "MySQL"}, + ], + } + + config, worker, decision_ref = _seed_pending_decision( + tmp_path, + turn={"pending_decision": adapter_pending_decision}, + ) + assert config.db_path is not None + payload = pending_payload_from_store(config.db_path, config.host_id) + pending = next( + row for row in payload["pending_interactions"] + if row["worker_id"] == worker.id + ) + assert pending["meta"]["decision"] == { + "decision_ref": decision_ref, + "kind": "multi", + "prompt": "Database support: Which databases should we support?", + "options": [ + {"ref": "1", "label": "Postgres"}, + {"ref": "2", "label": "SQLite"}, + {"ref": "3", "label": "DuckDB"}, + {"ref": "4", "label": "MySQL"}, + ], + "multi_select": True, + "question_count": 1, + } + + calls: list[dict[str, Any]] = [] + result = submit_command( + config, + _answer_request( + decision_ref, + request_id="multi-production-answer", + selection={"option_refs": ["3", "1"]}, + ), + socket_client_factory=_factory(calls), + ) + + assert result.ok is True + assert result.status == STATUS_ACCEPTED + assert calls == [ + { + "method": "pane.send_keys", + "params": { + "pane_id": "decision-pane-private", + "keys": ["Down", "Down"], + }, + }, + { + "method": "pane.send_text", + "params": {"pane_id": "decision-pane-private", "text": " "}, + }, + { + "method": "pane.send_keys", + "params": { + "pane_id": "decision-pane-private", + "keys": ["Up", "Up"], + }, + }, + { + "method": "pane.send_text", + "params": {"pane_id": "decision-pane-private", "text": " "}, + }, + { + "method": "pane.send_keys", + "params": { + "pane_id": "decision-pane-private", + "keys": ["Down", "Down", "Down", "Down", "Enter"], + }, + }, + ] + + +def _bounded_decision_turn(option_count: int, *, custom_last: bool = False) -> dict[str, Any]: + options = [ + {"id": str(ordinal), "label": f"Option {ordinal}"} + for ordinal in range(1, option_count + 1) + ] + if custom_last: + options.append({"id": "custom", "label": "Type something"}) + return { + "pending_decision": { + "decision_id": "toolu_bound", + "kind": "AskUserQuestion", + "prompt": "Choose one", + "multi_select": False, + "options": options, + } + } + + +def _unknown_decision_turn() -> dict[str, Any]: + turn = _bounded_decision_turn(2) + turn["pending_decision"]["kind"] = "FutureDecisionKind" + return turn + + +def test_decision_option_bound_accepts_exactly_eleven(tmp_path: Path) -> None: + config, worker, _decision_ref = _seed_pending_decision( + tmp_path, + turn=_bounded_decision_turn(PENDING_DECISION_MAX_OPTIONS), + ) + assert config.db_path is not None + payload = pending_payload_from_store(config.db_path, config.host_id) + row = next( + item for item in payload["pending_interactions"] + if item["worker_id"] == worker.id + ) + assert len(row["meta"]["decision"]["options"]) == 11 + + +@pytest.mark.parametrize( + "turn", + [ + _bounded_decision_turn(PENDING_DECISION_MAX_OPTIONS + 1), + _bounded_decision_turn(PENDING_DECISION_MAX_OPTIONS, custom_last=True), + _unknown_decision_turn(), + ], + ids=[ + "twelve-real-options", + "custom-row-after-eleven-real-options", + "unknown-kind", + ], +) +def test_over_bound_decision_fails_closed_without_pane_io( + tmp_path: Path, + turn: dict[str, Any], +) -> None: + config = _config(tmp_path) + worker = Worker(id="w-1", name="Alpha", status="active") + binding = _binding( + worker, + private_fingerprint="decision-binding-private", + turn_target_value="decision-pane-private", + ) + _seed(config, [worker], [binding]) + observation = _pending_observation_from_turn(turn) + assert observation.kind == "read_succeeded_unsupported_decision" + assert config.db_path is not None + assert apply_backend_pending_observation( + config.db_path, + config.host_id, + worker.id, + observation, + binding_private_fingerprint=binding.private_fingerprint, + observed_turn_target_value=binding.turn_target_value, + ) + payload = pending_payload_from_store(config.db_path, config.host_id) + assert not any( + item["worker_id"] == worker.id + for item in payload["pending_interactions"] + ) + + calls: list[dict[str, Any]] = [] + result = submit_command( + config, + _answer_request("decision-unsupported-bound"), + socket_client_factory=_factory(calls), + ) + assert result.ok is False + assert result.status == STATUS_UNSUPPORTED_DECISION + assert calls == [] + + +def test_answer_decision_request_id_replay_does_not_resend_keys(tmp_path: Path) -> None: + config, _worker, decision_ref = _seed_pending_decision(tmp_path) + calls: list[dict[str, Any]] = [] + request = _answer_request(decision_ref) + + first = submit_command(config, request, socket_client_factory=_factory(calls)) + replay = submit_command(config, request, socket_client_factory=_factory(calls)) + + assert first.ok is True + assert first.status == STATUS_ACCEPTED + assert replay.to_dict() == first.to_dict() + assert calls == [ + { + "method": "pane.send_keys", + "params": { + "pane_id": "decision-pane-private", + "keys": ["2", "Enter"], + }, + } + ] + + +def test_two_requests_race_one_decision_and_only_first_claimant_sends( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + config, worker, decision_ref = _seed_pending_decision(tmp_path) + assert config.db_path is not None + winner = _answer_request(decision_ref, request_id="decision-winner") + loser = _answer_request(decision_ref, request_id="decision-loser") + calls: list[dict[str, Any]] = [] + raced: dict[str, Any] = {} + real_mark = command_submission._mark_request_send_started + + def race_after_claim(*args: Any, **kwargs: Any) -> Any: + pending = pending_payload_from_store(config.db_path, config.host_id) + assert any( + item["worker_id"] == worker.id + and item["meta"]["decision"]["decision_ref"] == decision_ref + for item in pending["pending_interactions"] + ) + raced["loser"] = submit_command( + config, + loser, + socket_client_factory=_factory(calls), + ) + return real_mark(*args, **kwargs) + + monkeypatch.setattr( + command_submission, + "_mark_request_send_started", + race_after_claim, + ) + first = submit_command( + config, + winner, + socket_client_factory=_factory(calls), + ) + + assert first.ok is True + assert first.status == STATUS_ACCEPTED + assert raced["loser"].ok is False + assert raced["loser"].status == STATUS_ANSWER_IN_PROGRESS + assert raced["loser"].disposition == DISPOSITION_NO_RECEIPT + assert calls == [ + { + "method": "pane.send_keys", + "params": { + "pane_id": "decision-pane-private", + "keys": ["2", "Enter"], + }, + } + ] + + +def test_winner_safe_presend_failure_releases_claim_for_loser_retry( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + config, _worker, decision_ref = _seed_pending_decision(tmp_path) + winner = _answer_request(decision_ref, request_id="decision-safe-failure") + loser = _answer_request(decision_ref, request_id="decision-safe-retry") + calls: list[dict[str, Any]] = [] + raced: dict[str, Any] = {} + real_mark = command_submission._mark_request_send_started + first_mark = True + + def fail_winner_before_send(*args: Any, **kwargs: Any) -> Any: + nonlocal first_mark + if first_mark: + first_mark = False + raced["loser"] = submit_command( + config, + loser, + socket_client_factory=_factory(calls), + ) + return command_submission._request_in_progress(args[1]) + return real_mark(*args, **kwargs) + + monkeypatch.setattr( + command_submission, + "_mark_request_send_started", + fail_winner_before_send, + ) + failed_winner = submit_command( + config, + winner, + socket_client_factory=_factory(calls), + ) + retried_loser = submit_command( + config, + loser, + socket_client_factory=_factory(calls), + ) + + assert failed_winner.status == STATUS_PENDING + assert raced["loser"].status == STATUS_ANSWER_IN_PROGRESS + assert raced["loser"].disposition == DISPOSITION_NO_RECEIPT + assert retried_loser.ok is True + assert retried_loser.status == STATUS_ACCEPTED + assert calls == [ + { + "method": "pane.send_keys", + "params": { + "pane_id": "decision-pane-private", + "keys": ["2", "Enter"], + }, + } + ] + + +def test_claim_time_loser_releases_unsent_reservation_for_takeover( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + config, worker, decision_ref = _seed_pending_decision(tmp_path) + assert config.db_path is not None + request = _answer_request(decision_ref, request_id="claim-time-loser") + calls: list[dict[str, Any]] = [] + real_claim = command_submission._claim_pending_decision + injected_claim: dict[str, Any] = {} + + def lose_during_claim( + claim_config: Any, + claim_request: CommandRequest, + validated: Any, + ) -> Any: + if "value" not in injected_claim: + injected_claim["value"] = claim_backend_pending_decision( + config.db_path, + config.host_id, + worker.id, + decision_ref, + {"option_refs": ["2"]}, + claim=True, + ) + assert injected_claim["value"].status == "claimed" + return real_claim(claim_config, claim_request, validated) + + monkeypatch.setattr( + command_submission, + "_claim_pending_decision", + lose_during_claim, + ) + first = submit_command( + config, + request, + socket_client_factory=_factory(calls), + ) + receipt = get_command_request( + config.db_path, + config.host_id, + "claim-time-loser", + ) + + assert first.status == STATUS_ANSWER_IN_PROGRESS + assert first.disposition == DISPOSITION_IN_PROGRESS + assert receipt is not None + assert receipt["state"] == "reserved" + assert receipt["status"] == STATUS_PENDING + assert calls == [] + + assert abandon_backend_pending_choice_claim( + config.db_path, + config.host_id, + injected_claim["value"].claim_token, + ) + retry = submit_command( + config, + request, + socket_client_factory=_factory(calls), + ) + assert retry.ok is True + assert retry.status == STATUS_ACCEPTED + assert len(calls) == 1 + + +def test_abandoned_reservation_is_not_terminalized_by_live_claim( + tmp_path: Path, +) -> None: + config, worker, decision_ref = _seed_pending_decision(tmp_path) + assert config.db_path is not None + payload = _answer_request(decision_ref, request_id="abandoned-loser") + request = CommandRequest.from_dict(payload) + canonical = build_canonical_mutation(request, public_worker_id=worker.id) + initial = reserve_command_request( + config.db_path, + host_id=config.host_id, + request_id=request.request_id or "", + action=canonical.action, + canonical_version=canonical.canonical_version, + canonical_fingerprint=canonical.fingerprint, + canonical_request_json=canonical.canonical_json, + public_worker_id=canonical.public_worker_id, + pending_result_json=envelope_to_receipt_json( + command_submission._request_in_progress(request) + ), + legacy_raw_payload_fingerprint=request.payload_fingerprint(), + owner_lease_seconds=1, + now="2020-01-01T00:00:00+00:00", + ) + assert initial["status"] == "reserved" + competing = claim_backend_pending_decision( + config.db_path, + config.host_id, + worker.id, + decision_ref, + {"option_refs": ["1"]}, + claim=True, + ) + assert competing.status == "claimed" + + calls: list[dict[str, Any]] = [] + result = submit_command( + config, + payload, + socket_client_factory=_factory(calls), + ) + receipt = get_command_request( + config.db_path, + config.host_id, + request.request_id or "", + ) + + assert result.status == STATUS_ANSWER_IN_PROGRESS + assert result.disposition == DISPOSITION_IN_PROGRESS + assert receipt is not None + assert receipt["state"] == "reserved" + assert receipt["status"] == STATUS_PENDING + assert calls == [] + + +def test_process_loss_before_send_started_is_recovered_after_lease_expiry( + tmp_path: Path, +) -> None: + config, worker, decision_ref = _seed_pending_decision(tmp_path) + assert config.db_path is not None + payload = _answer_request(decision_ref, request_id="crashed-before-send") + request = CommandRequest.from_dict(payload) + canonical = build_canonical_mutation(request, public_worker_id=worker.id) + reservation = reserve_command_request( + config.db_path, + host_id=config.host_id, + request_id=request.request_id or "", + action=canonical.action, + canonical_version=canonical.canonical_version, + canonical_fingerprint=canonical.fingerprint, + canonical_request_json=canonical.canonical_json, + public_worker_id=canonical.public_worker_id, + pending_result_json=envelope_to_receipt_json( + command_submission._request_in_progress(request) + ), + legacy_raw_payload_fingerprint=request.payload_fingerprint(), + owner_lease_seconds=1, + now="2020-01-01T00:00:00+00:00", + ) + assert reservation["status"] == "reserved" + abandoned_claim = claim_backend_pending_decision( + config.db_path, + config.host_id, + worker.id, + decision_ref, + {"option_refs": ["2"]}, + claim=True, + observed_at="2020-01-01T00:00:00+00:00", + claim_lease_seconds=1, + ) + assert abandoned_claim.status == "claimed" + + calls: list[dict[str, Any]] = [] + recovered = submit_command( + config, + payload, + socket_client_factory=_factory(calls), + ) + + assert recovered.ok is True + assert recovered.status == STATUS_ACCEPTED + assert calls == [ + { + "method": "pane.send_keys", + "params": { + "pane_id": "decision-pane-private", + "keys": ["2", "Enter"], + }, + } + ] + + +def test_send_uncertainty_is_durable_and_never_resends( + tmp_path: Path, +) -> None: + config, _worker, decision_ref = _seed_pending_decision(tmp_path) + request = _answer_request(decision_ref, request_id="uncertain-decision") + calls: list[dict[str, Any]] = [] + + class FailingKeyClient(_FakeSocketClient): + def request( + self, + method: str, + params: dict[str, Any], + *, + timeout: float | None = None, + ) -> dict[str, Any]: + self.calls.append({"method": method, "params": dict(params)}) + if method == "pane.send_keys": + raise RuntimeError("response lost after key send") + return {"accepted": True} + + first = submit_command( + config, + request, + socket_client_factory=lambda _config: FailingKeyClient(calls), + ) + replay = submit_command( + config, + request, + socket_client_factory=lambda _config: FailingKeyClient(calls), + ) + + assert first.ok is False + assert first.status == STATUS_REQUEST_STATE_UNCERTAIN + assert replay.status == STATUS_REQUEST_STATE_UNCERTAIN + assert calls == [ + { + "method": "pane.send_keys", + "params": { + "pane_id": "decision-pane-private", + "keys": ["2", "Enter"], + }, + } + ] diff --git a/tests/test_backend_pending.py b/tests/test_backend_pending.py index e39d109..4283a7f 100644 --- a/tests/test_backend_pending.py +++ b/tests/test_backend_pending.py @@ -65,7 +65,19 @@ def test_extract_pending_decision(): assert all("value" not in c for c in pending["choices"]) # decision_id (internal tool_use_id) is not published in public pending. assert "decision_id" not in pending["meta"] - assert pending["meta"] == {"source": "backend"} + decision = pending["meta"]["decision"] + assert decision == { + "decision_ref": decision["decision_ref"], + "kind": "single", + "prompt": "Which database should we use?", + "options": [ + {"ref": "1", "label": "Postgres"}, + {"ref": "2", "label": "SQLite"}, + ], + "multi_select": False, + "question_count": 1, + } + assert decision["decision_ref"].startswith("decision-") def test_extract_plan_approval_kind(): diff --git a/tests/test_command_submission.py b/tests/test_command_submission.py index de474bb..a60c444 100644 --- a/tests/test_command_submission.py +++ b/tests/test_command_submission.py @@ -251,7 +251,9 @@ def _expected_private_clear_calls(pane_id: str = "pane-secret") -> list[dict[str {"method": "pane.send_keys", "params": {"pane_id": pane_id, "keys": ["ctrl+a", "backspace"]}}, ] -@pytest.mark.parametrize("action", ["send_instruction", "answer_pending"]) +@pytest.mark.parametrize( + "action", ["send_instruction", "answer_pending", "answer_decision"] +) @pytest.mark.parametrize( ("request_id", "include_request_id"), [ @@ -312,6 +314,18 @@ def guarded_socket_factory(config: Config) -> _FakeSocketClient: "choice_id": "choice-public", }, } + elif action == "answer_decision": + payload = { + "schema_version": 1, + "action": "answer_decision", + "request_id": request_id, + "dry_run": False, + "target": {"worker_id": "w-1"}, + "params": { + "decision_ref": "decision-public", + "selection": {"option_refs": ["1"]}, + }, + } if include_request_id: payload["request_id"] = request_id else: diff --git a/tests/test_commands.py b/tests/test_commands.py index c52ae3f..f31dace 100644 --- a/tests/test_commands.py +++ b/tests/test_commands.py @@ -21,6 +21,7 @@ TERMINAL_MUTATION_REJECTION_STATUSES, VALID_DISPOSITIONS, STATUS_ACCEPTED, + STATUS_ANSWER_IN_PROGRESS, STATUS_BACKEND_UNAVAILABLE, STATUS_PENDING, STATUS_REQUEST_STATE_UNCERTAIN, @@ -191,6 +192,7 @@ def test_allowed_actions_frozen() -> None: "resolve_target", "send_instruction", "answer_pending", + "answer_decision", } @@ -557,7 +559,9 @@ def test_command_envelope_rejects_inconsistent_receipt_tuples( ) -@pytest.mark.parametrize("action", ["send_instruction", "answer_pending"]) +@pytest.mark.parametrize( + "action", ["send_instruction", "answer_pending", "answer_decision"] +) @pytest.mark.parametrize("request_id", [None, "", "not canonical"]) def test_command_envelope_live_mutations_require_canonical_request_ids( action: str, @@ -567,7 +571,7 @@ def test_command_envelope_live_mutations_require_canonical_request_ids( action=action, request_id=request_id, dry_run=False, - target={"worker_id": "w-1"} if action == "send_instruction" else None, + target={"worker_id": "w-1"} if action != "answer_pending" else None, instruction={"text": "hello"} if action == "send_instruction" else None, params=( { @@ -576,7 +580,14 @@ def test_command_envelope_live_mutations_require_canonical_request_ids( "choice_id": "choice-1", } if action == "answer_pending" - else None + else ( + { + "decision_ref": "decision-1", + "selection": {"option_refs": ["1"]}, + } + if action == "answer_decision" + else None + ) ), ) @@ -638,6 +649,10 @@ def test_mutation_disposition_status_sets_are_explicit_and_fail_closed() -> None "ambiguous_backend_target", "backend_failed", "duplicate_request", + "decision_not_pending", + "unknown_worker", + "invalid_selection", + "unsupported_decision", } assert LIVE_MUTATION_NO_RECEIPT_REJECTION_STATUSES == { "invalid_request", @@ -649,9 +664,15 @@ def test_mutation_disposition_status_sets_are_explicit_and_fail_closed() -> None "backend_unsupported", "ambiguous_backend_target", "backend_failed", + "answer_in_progress", + "decision_not_pending", + "unknown_worker", + "invalid_selection", + "unsupported_decision", } assert DRY_RUN_MUTATION_NO_RECEIPT_REJECTION_STATUSES == { "invalid_request", + "invalid_selection", "rejected", "not_found", "ambiguous_target", @@ -659,7 +680,44 @@ def test_mutation_disposition_status_sets_are_explicit_and_fail_closed() -> None } -@pytest.mark.parametrize("action", ["send_instruction", "answer_pending"]) +def test_answer_in_progress_allows_only_retryable_dispositions() -> None: + request = CommandRequest( + action="answer_decision", + request_id="answer-race", + dry_run=False, + target={"worker_id": "w-1"}, + params={ + "decision_ref": "decision-1", + "selection": {"option_refs": ["1"]}, + }, + ) + for disposition in (DISPOSITION_NO_RECEIPT, DISPOSITION_IN_PROGRESS): + envelope = CommandEnvelope.from_result( + request, + ok=False, + status=STATUS_ANSWER_IN_PROGRESS, + disposition=disposition, + error={"code": STATUS_ANSWER_IN_PROGRESS, "message": "racing"}, + ) + assert envelope.disposition == disposition + for disposition in ( + DISPOSITION_TERMINAL_ACCEPTED, + DISPOSITION_TERMINAL_REJECTED, + DISPOSITION_TERMINAL_UNCERTAIN, + ): + with pytest.raises(ValueError): + CommandEnvelope.from_result( + request, + ok=False, + status=STATUS_ANSWER_IN_PROGRESS, + disposition=disposition, + error={"code": STATUS_ANSWER_IN_PROGRESS, "message": "racing"}, + ) + + +@pytest.mark.parametrize( + "action", ["send_instruction", "answer_pending", "answer_decision"] +) @pytest.mark.parametrize("ok", [False, True]) @pytest.mark.parametrize("status", sorted(VALID_STATUSES)) def test_command_envelope_from_dict_enforces_terminal_rejected_matrix( @@ -688,7 +746,9 @@ def test_command_envelope_from_dict_enforces_terminal_rejected_matrix( CommandEnvelope.from_dict(payload) -@pytest.mark.parametrize("action", ["send_instruction", "answer_pending"]) +@pytest.mark.parametrize( + "action", ["send_instruction", "answer_pending", "answer_decision"] +) @pytest.mark.parametrize("dry_run", [False, True], ids=["live", "dry-run"]) @pytest.mark.parametrize("ok", [False, True]) @pytest.mark.parametrize("status", sorted(VALID_STATUSES)) @@ -721,6 +781,10 @@ def test_command_envelope_from_dict_enforces_mutation_no_receipt_matrix( allowed = ( not ok and status in LIVE_MUTATION_NO_RECEIPT_REJECTION_STATUSES + and ( + status != STATUS_ANSWER_IN_PROGRESS + or action == "answer_decision" + ) ) if allowed: diff --git a/tests/test_turns.py b/tests/test_turns.py index fc0ab08..bf217c9 100644 --- a/tests/test_turns.py +++ b/tests/test_turns.py @@ -491,6 +491,7 @@ def test_pending_observation_models_explicit_outcomes_and_private_picker_routes( for kind in ( "read_succeeded_no_prompt", "read_succeeded_invalid_prompt", + "read_succeeded_unsupported_decision", "read_failed", "worker_authoritatively_absent", ): From bd6082cdf28a89d55b08e99854d117dd117c266e Mon Sep 17 00:00:00 2001 From: jerryfane Date: Thu, 16 Jul 2026 21:39:17 +0200 Subject: [PATCH 2/3] review: replace provisional decision calibration with live-verified key maps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Probed key-by-key on a real Claude Code 2.1.211 pane: - single/plan: the digit alone selects AND submits — the previous trailing Enter would leak into whatever UI appears next (composer or a modal). - write-in: the Type-something row ignores digits; it is reached with Down x N and the focused row is itself the text input. - multi-select: digits toggle rows ABSOLUTELY (cursor-independent), Right reaches the Submit tab, Enter submits — replacing the fragile relative Down/Up + Space navigation entirely. - option bound tightened 11 -> 9: every driven ordinal must stay a single keystroke; larger decisions fail closed to the read-only interaction. Co-Authored-By: Claude Fable 5 --- docs/answer_decision.md | 2 +- src/tendwire/backends/herdr_decision.py | 87 ++++++++++--------------- src/tendwire/backends/herdr_turns.py | 5 +- src/tendwire/command_submission.py | 7 -- tests/test_answer_decision.py | 51 +++++---------- 5 files changed, 56 insertions(+), 96 deletions(-) diff --git a/docs/answer_decision.md b/docs/answer_decision.md index 77f9c0d..1827af9 100644 --- a/docs/answer_decision.md +++ b/docs/answer_decision.md @@ -114,5 +114,5 @@ IDs, and no custom row. Older adapters that leave that prompt as an unstructured prompts remain unstructured and unsupported. The Claude Code multi-select key behavior is a private backend assumption, not -part of this public contract. Its provisional calibration is isolated in +part of this public contract. Its calibration (live-verified against Claude Code 2.1.211) is isolated in `src/tendwire/backends/herdr_decision.py` for live verification and retuning. diff --git a/src/tendwire/backends/herdr_decision.py b/src/tendwire/backends/herdr_decision.py index b9d06f0..c6199bd 100644 --- a/src/tendwire/backends/herdr_decision.py +++ b/src/tendwire/backends/herdr_decision.py @@ -1,23 +1,25 @@ """Translate semantic Claude decisions into private Herdr pane input. -Calibration assumptions are deliberately confined to this backend module: - -* Claude Code displays single-choice and plan rows with 1-based decimal - shortcuts; typing the ordinal and then Enter chooses that row. -* A single-choice write-in row immediately follows the advertised options. - Its ordinal opens/focuses the text field without an intermediate Enter, so - Tendwire types ``N + 1`` and then submits the write-in text with Enter. -* Claude Code multi-select digit toggles are not treated as a supported - contract. Tendwire therefore assumes the cursor starts on row 1, Down/Up move - exactly one option row, Space toggles the current option without moving the - cursor, the Submit row immediately follows the final option, and Enter on - Submit submits. This provisional Claude Code behavior is isolated in - ``MULTI_SELECT_CALIBRATION`` below so live verification can retune it in one - place. +Calibration assumptions are deliberately confined to this backend module, and +every mapping below was LIVE-VERIFIED against Claude Code 2.1.211 on a real +pane (2026-07-16): + +* Single-choice and plan rows carry 1-based decimal shortcuts, and typing the + ordinal alone SELECTS AND SUBMITS the row instantly — no Enter follows. A + trailing Enter would leak into whatever UI appears next, so none is sent. +* The single-choice write-in row ("Type something", at position N + 1) does + NOT respond to its digit. It is reached by pressing Down exactly N times + from the initial cursor on row 1; the focused row is itself a text input, + so Tendwire then sends the write-in prose and submits with Enter. +* Multi-select digits toggle their row ABSOLUTELY without moving the cursor, + Right switches to the Submit tab (a review screen whose default focus is + "Submit answers"), and Enter there submits the selection set. +* Every driven ordinal must stay a single keystroke, so decisions expose at + most 9 real options (PENDING_DECISION_MAX_OPTIONS in herdr_turns). * Herdr's private ``pane.send_keys`` accepts decimal character keys plus - ``Down``, ``Up``, and ``Enter``. A literal multi-select Space is sent through - private ``pane.send_text``; write-in prose uses ``pane.send_input`` so Herdr - owns terminal text encoding and appends the final Enter atomically. + ``Down``, ``Up``, ``Right``, and ``Enter``; write-in prose uses + ``pane.send_input`` so Herdr owns terminal text encoding and appends the + final Enter atomically. These steps are internal calibration data. They are never accepted from a connector and there is intentionally no public raw-key command action. @@ -30,9 +32,7 @@ MULTI_SELECT_CALIBRATION = { - "down": "Down", - "up": "Up", - "toggle_text": " ", + "submit_tab": "Right", "submit": "Enter", } @@ -50,8 +50,7 @@ def __post_init__(self) -> None: if not self.keys or self.text is not None: raise ValueError("key calibration step requires only keys") elif self.operation == "text": - if self.text != " " or self.keys: - raise ValueError("text calibration step requires one literal space") + raise ValueError("text calibration steps are no longer produced") elif self.operation == "input": if not isinstance(self.text, str) or not self.text or self.keys != ("Enter",): raise ValueError("input calibration step requires text plus Enter") @@ -84,8 +83,10 @@ def calibrate_decision_steps( if text is not None: if kind != "single" or option_refs or not isinstance(text, str) or not text: raise ValueError("invalid decision write-in calibration") + # The write-in row ignores digits; reach it with Down x N from row 1, + # where the focused row is itself the text input. return ( - HerdrDecisionStep("keys", keys=_digit_keys(option_count + 1)), + HerdrDecisionStep("keys", keys=("Down",) * option_count), HerdrDecisionStep("input", keys=("Enter",), text=text), ) if not option_refs or len(option_refs) != len(set(option_refs)): @@ -101,39 +102,21 @@ def calibrate_decision_steps( if kind in {"single", "plan"}: if len(ordinals) != 1: raise ValueError("single and plan decisions require one option") - return ( - HerdrDecisionStep( - "keys", - keys=(*_digit_keys(ordinals[0]), "Enter"), - ), - ) - - steps: list[HerdrDecisionStep] = [] - current_row = 1 - for ordinal in ordinals: - delta = ordinal - current_row - if delta: - direction = ( - MULTI_SELECT_CALIBRATION["down"] - if delta > 0 - else MULTI_SELECT_CALIBRATION["up"] - ) - steps.append(HerdrDecisionStep("keys", keys=(direction,) * abs(delta))) - steps.append( - HerdrDecisionStep( - "text", - text=MULTI_SELECT_CALIBRATION["toggle_text"], - ) - ) - current_row = ordinal - submit_navigation = (MULTI_SELECT_CALIBRATION["down"],) * ( - option_count + 1 - current_row - ) + # The digit alone selects AND submits; a trailing Enter would leak into + # the next UI (composer, or worse, a modal). + return (HerdrDecisionStep("keys", keys=_digit_keys(ordinals[0])),) + + # Digits toggle rows absolutely (cursor-independent); Right reaches the + # Submit tab whose default focus is "Submit answers"; Enter submits. + steps: list[HerdrDecisionStep] = [ + HerdrDecisionStep("keys", keys=_digit_keys(ordinal)) + for ordinal in sorted(ordinals) + ] steps.append( HerdrDecisionStep( "keys", keys=( - *submit_navigation, + MULTI_SELECT_CALIBRATION["submit_tab"], MULTI_SELECT_CALIBRATION["submit"], ), ) diff --git a/src/tendwire/backends/herdr_turns.py b/src/tendwire/backends/herdr_turns.py index b22cc49..6cb3d03 100644 --- a/src/tendwire/backends/herdr_turns.py +++ b/src/tendwire/backends/herdr_turns.py @@ -386,7 +386,10 @@ def _extract_turn_payload(value: Any) -> Mapping[str, Any] | None: return value -PENDING_DECISION_MAX_OPTIONS = 11 +# Every driven ordinal must stay a single keystroke (digits select/toggle +# absolutely in the pane, live-verified on Claude Code 2.1.211), so 9 is the +# hard bound; larger decisions fail closed to the read-only interaction. +PENDING_DECISION_MAX_OPTIONS = 9 _PENDING_TEXT_MAX = 2000 _SINGLE_WRITE_IN_OPTION_IDS = frozenset( {"custom", "other", "writein", "write_in", "write-in"} diff --git a/src/tendwire/command_submission.py b/src/tendwire/command_submission.py index 228b362..70476b6 100644 --- a/src/tendwire/command_submission.py +++ b/src/tendwire/command_submission.py @@ -1674,13 +1674,6 @@ def _submit_decision_calibration( {"pane_id": pane_id, "keys": list(step.keys)}, timeout=timeout, ) - elif step.operation == "text": - _socket_request( - client, - "pane.send_text", - {"pane_id": pane_id, "text": step.text}, - timeout=timeout, - ) else: _socket_request( client, diff --git a/tests/test_answer_decision.py b/tests/test_answer_decision.py index c57499c..8e63995 100644 --- a/tests/test_answer_decision.py +++ b/tests/test_answer_decision.py @@ -271,21 +271,21 @@ def test_decision_calibration_steps_cover_single_plan_write_in_and_multi() -> No kind="single", option_count=4, option_refs=("2",) ) assert [(item.operation, item.keys, item.text) for item in single] == [ - ("keys", ("2", "Enter"), None) + ("keys", ("2",), None) # digit alone selects AND submits (live-verified) ] plan = calibrate_decision_steps( kind="plan", option_count=2, option_refs=("1",) ) assert [(item.operation, item.keys, item.text) for item in plan] == [ - ("keys", ("1", "Enter"), None) + ("keys", ("1",), None) ] write_in = calibrate_decision_steps( kind="single", option_count=4, text="Use another database" ) assert [(item.operation, item.keys, item.text) for item in write_in] == [ - ("keys", ("5",), None), + ("keys", ("Down", "Down", "Down", "Down"), None), # write-in row ignores digits ("input", ("Enter",), "Use another database"), ] @@ -293,11 +293,9 @@ def test_decision_calibration_steps_cover_single_plan_write_in_and_multi() -> No kind="multi", option_count=4, option_refs=("3", "1") ) assert [(item.operation, item.keys, item.text) for item in multi] == [ - ("keys", ("Down", "Down"), None), - ("text", (), " "), - ("keys", ("Up", "Up"), None), - ("text", (), " "), - ("keys", ("Down", "Down", "Down", "Down", "Enter"), None), + ("keys", ("1",), None), # digits toggle rows absolutely + ("keys", ("3",), None), + ("keys", ("Right", "Enter"), None), # Submit tab, then submit ] @@ -381,32 +379,15 @@ def test_production_shaped_multi_decision_end_to_end(tmp_path: Path) -> None: assert calls == [ { "method": "pane.send_keys", - "params": { - "pane_id": "decision-pane-private", - "keys": ["Down", "Down"], - }, - }, - { - "method": "pane.send_text", - "params": {"pane_id": "decision-pane-private", "text": " "}, + "params": {"pane_id": "decision-pane-private", "keys": ["1"]}, }, { "method": "pane.send_keys", - "params": { - "pane_id": "decision-pane-private", - "keys": ["Up", "Up"], - }, - }, - { - "method": "pane.send_text", - "params": {"pane_id": "decision-pane-private", "text": " "}, + "params": {"pane_id": "decision-pane-private", "keys": ["3"]}, }, { "method": "pane.send_keys", - "params": { - "pane_id": "decision-pane-private", - "keys": ["Down", "Down", "Down", "Down", "Enter"], - }, + "params": {"pane_id": "decision-pane-private", "keys": ["Right", "Enter"]}, }, ] @@ -435,7 +416,7 @@ def _unknown_decision_turn() -> dict[str, Any]: return turn -def test_decision_option_bound_accepts_exactly_eleven(tmp_path: Path) -> None: +def test_decision_option_bound_accepts_exactly_nine(tmp_path: Path) -> None: config, worker, _decision_ref = _seed_pending_decision( tmp_path, turn=_bounded_decision_turn(PENDING_DECISION_MAX_OPTIONS), @@ -446,7 +427,7 @@ def test_decision_option_bound_accepts_exactly_eleven(tmp_path: Path) -> None: item for item in payload["pending_interactions"] if item["worker_id"] == worker.id ) - assert len(row["meta"]["decision"]["options"]) == 11 + assert len(row["meta"]["decision"]["options"]) == PENDING_DECISION_MAX_OPTIONS @pytest.mark.parametrize( @@ -518,7 +499,7 @@ def test_answer_decision_request_id_replay_does_not_resend_keys(tmp_path: Path) "method": "pane.send_keys", "params": { "pane_id": "decision-pane-private", - "keys": ["2", "Enter"], + "keys": ["2"], }, } ] @@ -571,7 +552,7 @@ def race_after_claim(*args: Any, **kwargs: Any) -> Any: "method": "pane.send_keys", "params": { "pane_id": "decision-pane-private", - "keys": ["2", "Enter"], + "keys": ["2"], }, } ] @@ -627,7 +608,7 @@ def fail_winner_before_send(*args: Any, **kwargs: Any) -> Any: "method": "pane.send_keys", "params": { "pane_id": "decision-pane-private", - "keys": ["2", "Enter"], + "keys": ["2"], }, } ] @@ -805,7 +786,7 @@ def test_process_loss_before_send_started_is_recovered_after_lease_expiry( "method": "pane.send_keys", "params": { "pane_id": "decision-pane-private", - "keys": ["2", "Enter"], + "keys": ["2"], }, } ] @@ -850,7 +831,7 @@ def request( "method": "pane.send_keys", "params": { "pane_id": "decision-pane-private", - "keys": ["2", "Enter"], + "keys": ["2"], }, } ] From acde7571812c3f5f26325978a593249a6996a854 Mon Sep 17 00:00:00 2001 From: plotarmordev Date: Fri, 17 Jul 2026 15:33:33 +0800 Subject: [PATCH 3/3] fix: align remote decision option bounds --- docs/answer_decision.md | 12 +++++++----- src/tendwire/backends/herdr_turns.py | 7 ++++++- tests/test_answer_decision.py | 28 +++++++++++++++++++++++++--- 3 files changed, 38 insertions(+), 9 deletions(-) diff --git a/docs/answer_decision.md b/docs/answer_decision.md index 1827af9..3b6990c 100644 --- a/docs/answer_decision.md +++ b/docs/answer_decision.md @@ -37,10 +37,12 @@ Supported shapes are: - `plan`: exactly one option; - `multi`: one or more options from a single question. -Tendwire supports at most 11 source option rows. A source decision with more -than 11 rows, including a custom/write-in row beyond that boundary, is not -truncated. Unknown decision kinds, over-bound decisions, and multi-question -decisions do not produce `meta.decision` and cannot be sent. +Tendwire supports at most 9 digit-addressable option rows. A single-choice +decision may additionally contain one recognized trailing custom/write-in row; +that row is not exposed as an option reference. A source decision with more +than 9 selectable rows is not truncated. Unknown decision kinds, over-bound +decisions, and multi-question decisions do not produce `meta.decision` and +cannot be sent. ## Command request @@ -82,7 +84,7 @@ The four typed validation failures are terminal for that attempted decision: - `unsupported_decision`. All fail before pane input. `unsupported_decision` covers multiple question -groups, more than 11 source rows, and unknown kinds. +groups, more than 9 selectable rows, and unknown kinds. ## Concurrency and retries diff --git a/src/tendwire/backends/herdr_turns.py b/src/tendwire/backends/herdr_turns.py index 6cb3d03..48bbbbd 100644 --- a/src/tendwire/backends/herdr_turns.py +++ b/src/tendwire/backends/herdr_turns.py @@ -423,7 +423,10 @@ def _pending_observation_from_turn(turn: Mapping[str, Any]) -> PendingObservatio options = [] if not isinstance(options, list): return PendingObservation("read_succeeded_invalid_prompt") - if len(options) > PENDING_DECISION_MAX_OPTIONS: + # A single-choice Claude prompt may have one trailing write-in row in + # addition to the digit-addressable options. Validate the effective + # selectable rows after the decision kind and write-in shape are known. + if len(options) > PENDING_DECISION_MAX_OPTIONS + 1: return PendingObservation("read_succeeded_unsupported_decision") choices: list[PendingObservedChoice] = [] for ordinal, option in enumerate(options, 1): @@ -529,6 +532,8 @@ def _pending_observation_from_turn(turn: Mapping[str, Any]) -> PendingObservatio decision_options = tuple(decision_option_labels) if not decision_options: return PendingObservation("read_succeeded_invalid_prompt") + if len(decision_options) > PENDING_DECISION_MAX_OPTIONS: + return PendingObservation("read_succeeded_unsupported_decision") return PendingObservation( "open_prompt", question=question, diff --git a/tests/test_answer_decision.py b/tests/test_answer_decision.py index 8e63995..0800ab9 100644 --- a/tests/test_answer_decision.py +++ b/tests/test_answer_decision.py @@ -430,16 +430,38 @@ def test_decision_option_bound_accepts_exactly_nine(tmp_path: Path) -> None: assert len(row["meta"]["decision"]["options"]) == PENDING_DECISION_MAX_OPTIONS +def test_decision_option_bound_accepts_nine_plus_trailing_write_in( + tmp_path: Path, +) -> None: + config, worker, _decision_ref = _seed_pending_decision( + tmp_path, + turn=_bounded_decision_turn( + PENDING_DECISION_MAX_OPTIONS, + custom_last=True, + ), + ) + assert config.db_path is not None + payload = pending_payload_from_store(config.db_path, config.host_id) + row = next( + item for item in payload["pending_interactions"] + if item["worker_id"] == worker.id + ) + assert len(row["meta"]["decision"]["options"]) == PENDING_DECISION_MAX_OPTIONS + + @pytest.mark.parametrize( "turn", [ _bounded_decision_turn(PENDING_DECISION_MAX_OPTIONS + 1), - _bounded_decision_turn(PENDING_DECISION_MAX_OPTIONS, custom_last=True), + _bounded_decision_turn( + PENDING_DECISION_MAX_OPTIONS + 1, + custom_last=True, + ), _unknown_decision_turn(), ], ids=[ - "twelve-real-options", - "custom-row-after-eleven-real-options", + "ten-real-options", + "custom-row-after-ten-real-options", "unknown-kind", ], )