diff --git a/loopx/cli_commands/goal_channel.py b/loopx/cli_commands/goal_channel.py index 654eda8b78..3a91547875 100644 --- a/loopx/cli_commands/goal_channel.py +++ b/loopx/cli_commands/goal_channel.py @@ -30,6 +30,13 @@ sync_lark_goal_channel, ) from ..extensions.lark.goal_channel_contracts import binding_for_goal, operation_packet +from ..extensions.lark.goal_channel_message_delivery import ( + GoalChannelDeliveryStageError, +) +from ..extensions.lark.goal_channel_operation import ( + OperationExecutorDriftError, + confirmed_operation_executor, +) from ..extensions.lark.goal_topic_batch import upgrade_lark_goal_topics from ..extensions.runtime import ( default_extension_state_file, @@ -253,8 +260,11 @@ def _error_packet( execute: bool, blocker: str, summary: str, + external_write_performed: bool = False, + failure_stage: str | None = None, + details: dict[str, object] | None = None, ) -> dict[str, object]: - return { + packet: dict[str, object] = { "schema_version": "loopx_goal_channel_operation_v0", "ok": False, "goal_id": goal_id, @@ -262,7 +272,7 @@ def _error_packet( "operation": operation, "execute": execute, "status": "blocked", - "external_write_performed": False, + "external_write_performed": external_write_performed, "readback_verified": False, "idempotency_key": None, "receipt_id": None, @@ -270,6 +280,11 @@ def _error_packet( "private_provider_payload_captured": False, "blocker": blocker, } + if failure_stage: + packet["failure_stage"] = failure_stage + if details: + packet["details"] = dict(details) + return packet def _target_path(args: argparse.Namespace, runtime_root: Path) -> Path: @@ -451,11 +466,20 @@ def _prepare_goal_channel_operation( idempotency_key: str, request_path: Path, execute: bool, + executor_binding_resolver: Callable[[Mapping[str, Any], Path], Mapping[str, Any]] + | None = None, ) -> dict[str, Any]: request = json.loads(request_path.read_text(encoding="utf-8")) if not isinstance(request, dict): raise ValueError("operation request JSON must be an object") parameters = {**request, "goal_id": goal_id, "agent_id": agent_id} + if isinstance(parameters.get("executor"), Mapping): + # Resolve the declared executor and compare the active revision + # before any durable proposal or idempotency entry exists, so a + # stale proposal can never become an unreachable gated record. + confirmed_operation_executor( + parameters, runtime_root, executor_binding_resolver + ) def preview(store_root: Path) -> dict[str, Any]: return ChatActionService( @@ -798,6 +822,37 @@ def handle_goal_channel_command( raise ValueError(f"unknown goal-channel command: {command}") if payload.get("ok"): payload["extension_activation"] = activation + except OperationExecutorDriftError as exc: + payload = _error_packet( + goal_id=goal_id, + operation=command.replace("-", "_"), + execute=execute, + blocker=exc.blocker, + summary=str(exc), + external_write_performed=exc.external_write_performed, + failure_stage=exc.failure_stage, + details=exc.details, + ) + except GoalChannelDeliveryStageError as exc: + outcome = exc.external_write_performed + payload = _error_packet( + goal_id=goal_id, + operation=command.replace("-", "_"), + execute=execute, + blocker=exc.blocker, + summary=str(exc), + # An unknown provider outcome must never be projected as a + # clean not-performed receipt: assume the write happened. + external_write_performed=( + True if outcome is None else outcome + ), + failure_stage=exc.failure_stage, + details=( + {"external_write_outcome": "unknown"} + if outcome is None + else None + ), + ) except ValueError: payload = _error_packet( goal_id=goal_id, diff --git a/loopx/extensions/lark/goal_channel_message_delivery.py b/loopx/extensions/lark/goal_channel_message_delivery.py index 0dc079b951..6bec137da9 100644 --- a/loopx/extensions/lark/goal_channel_message_delivery.py +++ b/loopx/extensions/lark/goal_channel_message_delivery.py @@ -29,6 +29,73 @@ from .presentation.kanban import CommandRunner +class GoalChannelDeliveryStageError(ValueError): + """One typed, public-safe failure stage of a Goal Channel delivery. + + The summary is the only user-visible text and must never carry private + provider or configuration detail. `external_write_performed` is True or + False only when the provider outcome is known; None means the outcome is + unknown and the projected receipt must treat the provider write as + performed instead of claiming a clean run. + """ + + def __init__( + self, + summary: str, + *, + blocker: str, + failure_stage: str, + external_write_performed: bool | None = False, + ) -> None: + super().__init__(summary) + self.blocker = blocker + self.failure_stage = failure_stage + self.external_write_performed = external_write_performed + + +def _has_provider_response_body(result: Mapping[str, Any]) -> bool: + """Whether the provider answered at all, rejection text included.""" + + return any(str(result.get(key) or "").strip() for key in ("stdout", "stderr")) + + +def delivery_send_failure( + result: Mapping[str, Any], +) -> GoalChannelDeliveryStageError: + """Classify a send result that produced no usable message id. + + Only a provider response body is a verdict, and only a non-zero exit with + that body is a rejection. A send that timed out, never started, or answered + without a body leaves the outcome unknown, and the card may already be live + in the chat: reporting a clean no-write there would be the misprojection + this stage's contract forbids. A zero exit without a readable message id is + the same unknown, because the provider accepted a write we cannot name. + """ + + if result.get("spawn_failed") is True: + return GoalChannelDeliveryStageError( + "Goal Channel delivery could not start the Lark CLI", + blocker="provider_unavailable", + failure_stage="send_operation_card", + ) + if ( + result.get("timed_out") is True + or result.get("returncode") == 0 + or not _has_provider_response_body(result) + ): + return GoalChannelDeliveryStageError( + "Goal Channel delivery send outcome is unknown", + blocker="delivery_outcome_unknown", + failure_stage="send_operation_card", + external_write_performed=None, + ) + return GoalChannelDeliveryStageError( + "Goal Channel delivery send failed", + blocker="provider_send_rejected", + failure_stage="send_operation_card", + ) + + def resolve_bound_goal_channel( *, binding_path: Path, @@ -353,7 +420,11 @@ def _existing_message( ) payload = json_payload(result) if result.get("returncode") != 0: - raise ValueError("Goal Channel delivery dedupe readback failed") + raise GoalChannelDeliveryStageError( + "Goal Channel delivery dedupe readback failed", + blocker="dedupe_history_read_failed", + failure_stage="read_dedupe_history", + ) for message in _message_rows(payload): sender_type, sender_app_id = _message_sender(message) if ( @@ -372,7 +443,11 @@ def _existing_message( ): return str(message["message_id"]) if not _history_is_complete(payload): - raise ValueError("Goal Channel delivery dedupe history is incomplete") + raise GoalChannelDeliveryStageError( + "Goal Channel delivery dedupe history is incomplete", + blocker="dedupe_history_incomplete", + failure_stage="read_dedupe_history", + ) return None def resolve(self, requested_goal_id: str) -> Mapping[str, Any]: @@ -434,13 +509,21 @@ def send( ) ) if dict(self.resolve_current_binding()) != self.binding: - raise ValueError("Goal Channel delivery binding drifted") + raise GoalChannelDeliveryStageError( + "Goal Channel delivery binding drifted", + blocker="binding_drifted", + failure_stage="prepare_delivery_transaction", + ) existing_message_id = self._existing_message(card, route) # The history lookup is a provider round trip. Recheck under the same # lock used by binding writers immediately before either accepting # the dedupe result or performing the external write. if dict(self.resolve_current_binding()) != self.binding: - raise ValueError("Goal Channel delivery binding drifted") + raise GoalChannelDeliveryStageError( + "Goal Channel delivery binding drifted", + blocker="binding_drifted", + failure_stage="prepare_delivery_transaction", + ) if existing_message_id is not None: self.expected_cards.setdefault(existing_message_id, []).append( dict(card) @@ -477,7 +560,7 @@ def send( json_payload(result), {"message_id"}, MESSAGE_ID_PATTERN ) if result.get("returncode") != 0 or not message_id: - raise ValueError("Goal Channel delivery send failed") + raise delivery_send_failure(result) self.expected_cards.setdefault(message_id, []).append(dict(card)) return { "message_id": message_id, @@ -541,7 +624,9 @@ def readback(self, message_id: str) -> Mapping[str, Any]: __all__ = [ + "delivery_send_failure", "GoalChannelMessageDeliverySession", + "GoalChannelDeliveryStageError", "normalized_card_text", "goal_channel_delivery_route", "resolve_bound_goal_channel", diff --git a/loopx/extensions/lark/goal_channel_operation.py b/loopx/extensions/lark/goal_channel_operation.py index 6a91061f70..17532d8de2 100644 --- a/loopx/extensions/lark/goal_channel_operation.py +++ b/loopx/extensions/lark/goal_channel_operation.py @@ -24,6 +24,7 @@ goal_channel_delivery_route, ) from .goal_channel_message_delivery import ( + GoalChannelDeliveryStageError, GoalChannelMessageDeliverySession, card_projection_matches, message_card_matches, @@ -430,13 +431,7 @@ def deliver_goal_channel_operation_card( goal_id = str(parameters["goal_id"]) if expected_goal_id is not None and goal_id != expected_goal_id: raise ActionConflictError("operation proposal belongs to another goal") - resolved_executor = dict( - executor_binding_resolver(parameters, runtime_root) - if executor_binding_resolver is not None - else _resolve_operation_executor_binding(parameters, runtime_root=runtime_root) - ) - if resolved_executor.get("revision") != parameters["executor"]["revision"]: - raise ActionConflictError("operation executor revision is not ready") + confirmed_operation_executor(parameters, runtime_root, executor_binding_resolver) agent_id = str(parameters["agent_id"]) binding = resolve_bound_goal_channel( binding_path=binding_path, @@ -485,10 +480,22 @@ def resolve_current() -> Mapping[str, Any]: runner=runner, ) if session.verify(route) is not True: - raise ValueError("Goal Channel sender identity could not be verified") + raise GoalChannelDeliveryStageError( + "Goal Channel sender identity could not be verified", + blocker="sender_identity_unverified", + failure_stage="verify_sender_identity", + ) sent = dict(session.send(card, key, route)) message_id = str(sent.get("message_id") or "") - observed = dict(session.readback(message_id)) + try: + observed = dict(session.readback(message_id)) + except Exception as exc: + raise GoalChannelDeliveryStageError( + "operation card delivery outcome is unknown after the provider write", + blocker="delivery_outcome_unknown", + failure_stage="read_operation_card", + external_write_performed=None, + ) from exc if not ( observed.get("verified") is True and observed.get("message_id") == message_id @@ -508,22 +515,30 @@ def resolve_current() -> Mapping[str, Any]: receipt_id=proposal_id, blocker="readback_unverified", ) - store.record_operation_delivery( - proposal_id, - delivery={ - "provider": "lark", - "message_id": message_id, - "chat_id": route["chat_id"], - "app_id": route["bot_app_id"], - "binding_digest": goal_channel_binding_digest(binding), - "card_digest": card_digest, - "delivered_at": datetime.now(timezone.utc).isoformat(), - }, - ) - store.record_operation_delivery_snapshot( - proposal_id, - submitted_card=card, - ) + try: + store.record_operation_delivery( + proposal_id, + delivery={ + "provider": "lark", + "message_id": message_id, + "chat_id": route["chat_id"], + "app_id": route["bot_app_id"], + "binding_digest": goal_channel_binding_digest(binding), + "card_digest": card_digest, + "delivered_at": datetime.now(timezone.utc).isoformat(), + }, + ) + store.record_operation_delivery_snapshot( + proposal_id, + submitted_card=card, + ) + except Exception as exc: + raise GoalChannelDeliveryStageError( + "operation card was delivered but its receipt could not be recorded", + blocker="delivery_receipt_write_failed", + failure_stage="record_delivery_receipt", + external_write_performed=None, + ) from exc return operation_packet( ok=True, goal_id=goal_id, @@ -830,6 +845,76 @@ def _resolve_operation_executor_binding( ) +class OperationExecutorDriftError(ActionConflictError): + """The requested executor revision does not match the active binding. + + Raised before any durable proposal is written and again at delivery, so + a stale proposal can never reach the provider. `details` carries only + opaque revision identifiers. + """ + + def __init__( + self, + summary: str, + *, + failure_stage: str, + details: Mapping[str, Any] | None = None, + ) -> None: + super().__init__(summary) + self.blocker = "executor_revision_drift" + self.failure_stage = failure_stage + self.details = dict(details or {}) + self.external_write_performed = False + + +def confirmed_operation_executor( + parameters: Mapping[str, Any], + runtime_root: Path, + executor_binding_resolver: ( + Callable[[Mapping[str, Any], Path], Mapping[str, Any]] | None + ), +) -> dict[str, Any]: + """Resolve the declared executor binding against the active revision. + + Shared by prepare (before any durable write) and deliver, so a proposal + whose executor revision no longer matches the installed extension is + rejected with a typed blocker instead of surfacing later as an + unreachable gated proposal. Resolution failures never leak the private + resolver text. ``executor_binding_resolver`` is a test seam with the same + shape as the delivery runner: production callers leave it unset and get the + extension binding resolver, so it is not a supported configuration entry. + """ + + executor = parameters.get("executor") + requested_revision = ( + executor.get("revision") if isinstance(executor, Mapping) else None + ) + try: + resolved = dict( + executor_binding_resolver(parameters, runtime_root) + if executor_binding_resolver is not None + else _resolve_operation_executor_binding( + parameters, runtime_root=runtime_root + ) + ) + except ValueError as exc: + raise GoalChannelDeliveryStageError( + "operation executor binding is unavailable", + blocker="executor_unavailable", + failure_stage="resolve_executor_binding", + ) from exc + if resolved.get("revision") != requested_revision: + raise OperationExecutorDriftError( + "operation executor revision is not ready", + failure_stage="resolve_executor_binding", + details={ + "requested_executor_revision": requested_revision, + "active_executor_revision": resolved.get("revision"), + }, + ) + return resolved + + def _execute_claimed_operation( proposal: Mapping[str, Any], *, runtime_root: Path ) -> dict[str, Any]: diff --git a/loopx/extensions/lark/goal_channel_transport.py b/loopx/extensions/lark/goal_channel_transport.py index 9d9c374639..b4c79a21a1 100644 --- a/loopx/extensions/lark/goal_channel_transport.py +++ b/loopx/extensions/lark/goal_channel_transport.py @@ -144,8 +144,14 @@ def call( ) -> Mapping[str, Any]: try: return runner(args, None, 30) + except subprocess.TimeoutExpired: + # The command ran and was killed, so a provider write it had already + # started is neither proven nor excluded. Carry the fact instead of + # folding it into a bare failure the callers would read as a verdict. + return {"returncode": 1, "stdout": "", "stderr": "", "timed_out": True} except (OSError, subprocess.SubprocessError): - return {"returncode": 1, "stdout": "", "stderr": ""} + # The command never started, so no provider write can have happened. + return {"returncode": 1, "stdout": "", "stderr": "", "spawn_failed": True} def profile_args(profile: str | None) -> list[str]: diff --git a/tests/extensions/test_lark_goal_channel.py b/tests/extensions/test_lark_goal_channel.py index 064ba512d2..638c7d0c2d 100644 --- a/tests/extensions/test_lark_goal_channel.py +++ b/tests/extensions/test_lark_goal_channel.py @@ -20,6 +20,9 @@ setup_lark_goal_channel, sync_lark_goal_channel, ) +from loopx.extensions.lark.goal_channel_message_delivery import ( + GoalChannelDeliveryStageError, +) from loopx.extensions.lark.goal_channel_runtime import ( auto_notify_lark_goal_channel_gate, ) @@ -2312,3 +2315,132 @@ def test_sync_preserves_provider_failure_blocker( assert payload["readback_verified"] is False assert payload["details"]["successful_write_count"] == 1 _assert_public_packet(payload) + + +def test_cli_deliver_operation_projects_typed_stage_blockers( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The CLI keeps the typed blocker, stage, and honest write state.""" + + project = tmp_path / "typed-stage-project" + source_registry_path = project / ".loopx" / "registry.json" + source_registry_path.parent.mkdir(parents=True) + source_registry = _registry(project) + source_registry["goals"][0]["repo"] = str(project) + source_registry_path.write_text(json.dumps(source_registry), encoding="utf-8") + + monkeypatch.setattr( + goal_channel_cli, + "resolve_extension_activation", + lambda *args, **kwargs: {"ok": True}, + ) + monkeypatch.setattr(goal_channel_cli, "_binding_target_name", lambda *args: "") + + def deliver_raises(**kwargs: object) -> object: + raise GoalChannelDeliveryStageError( + "Goal Channel delivery send failed", + blocker="provider_send_rejected", + failure_stage="send_operation_card", + external_write_performed=False, + ) + + monkeypatch.setattr( + goal_channel_cli, "deliver_goal_channel_operation_card", deliver_raises + ) + printed: dict[str, Any] = {} + result = goal_channel_cli.handle_goal_channel_command( + argparse.Namespace( + command="goal-channel", + goal_channel_command="deliver-operation", + goal_id=GOAL_ID, + proposal_id="proposal-typed-stage", + binding_path=None, + target_path=None, + execute=True, + subcommand_format="json", + format=None, + ), + registry_path=source_registry_path, + runtime_root_arg=None, + print_payload=lambda payload, fmt, renderer: printed.update(payload), + output_format=lambda args: "json", + ) + + assert result == 1 + assert printed["ok"] is False + assert printed["blocker"] == "provider_send_rejected" + assert printed["failure_stage"] == "send_operation_card" + assert printed["external_write_performed"] is False + assert "provider rejected" not in json.dumps(printed) + _assert_public_packet(printed) + + +@pytest.mark.parametrize( + ("failure_stage", "blocker"), + [ + # A send with no provider answer may already be live in the chat. + ("send_operation_card", "delivery_outcome_unknown"), + # The readback proved the write; only the local receipt failed. + ("record_delivery_receipt", "delivery_receipt_write_failed"), + ], +) +def test_cli_deliver_operation_treats_unknown_write_as_performed( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + failure_stage: str, + blocker: str, +) -> None: + """An unknown provider outcome is never projected as a clean receipt.""" + + project = tmp_path / "unknown-outcome-project" + source_registry_path = project / ".loopx" / "registry.json" + source_registry_path.parent.mkdir(parents=True) + source_registry = _registry(project) + source_registry["goals"][0]["repo"] = str(project) + source_registry_path.write_text(json.dumps(source_registry), encoding="utf-8") + + monkeypatch.setattr( + goal_channel_cli, + "resolve_extension_activation", + lambda *args, **kwargs: {"ok": True}, + ) + monkeypatch.setattr(goal_channel_cli, "_binding_target_name", lambda *args: "") + + def deliver_unknown(**kwargs: object) -> object: + raise GoalChannelDeliveryStageError( + "Goal Channel delivery outcome is unknown", + blocker=blocker, + failure_stage=failure_stage, + external_write_performed=None, + ) + + monkeypatch.setattr( + goal_channel_cli, "deliver_goal_channel_operation_card", deliver_unknown + ) + printed: dict[str, Any] = {} + result = goal_channel_cli.handle_goal_channel_command( + argparse.Namespace( + command="goal-channel", + goal_channel_command="deliver-operation", + goal_id=GOAL_ID, + proposal_id="proposal-unknown-outcome", + binding_path=None, + target_path=None, + execute=True, + subcommand_format="json", + format=None, + ), + registry_path=source_registry_path, + runtime_root_arg=None, + print_payload=lambda payload, fmt, renderer: printed.update(payload), + output_format=lambda args: "json", + ) + + assert result == 1 + assert printed["ok"] is False + assert printed["blocker"] == blocker + assert printed["failure_stage"] == failure_stage + assert printed["external_write_performed"] is True + assert printed["details"]["external_write_outcome"] == "unknown" + _assert_public_packet(printed) diff --git a/tests/extensions/test_lark_goal_channel_operation.py b/tests/extensions/test_lark_goal_channel_operation.py index 6cdc373389..451d69ffb2 100644 --- a/tests/extensions/test_lark_goal_channel_operation.py +++ b/tests/extensions/test_lark_goal_channel_operation.py @@ -5,6 +5,7 @@ import hashlib import json from pathlib import Path +import subprocess import threading from typing import Any @@ -18,10 +19,12 @@ write_goal_channel_binding, ) from loopx.extensions.lark.goal_channel_message_delivery import ( + GoalChannelDeliveryStageError, message_card_matches, normalized_card_text, ) from loopx.extensions.lark.goal_channel_operation import ( + OperationExecutorDriftError, build_goal_channel_operation_card, build_goal_channel_operation_result_card, deliver_goal_channel_operation_card, @@ -650,6 +653,9 @@ def test_cli_preparation_previews_without_write_then_persists_canonical_proposal idempotency_key="cli-operation-fixture", request_path=request_path, execute=False, + executor_binding_resolver=lambda _parameters, _runtime: { + "revision": "simulator-v0" + }, ) assert preview["status"] == "preview_ready" assert store.list() == [] @@ -663,6 +669,9 @@ def test_cli_preparation_previews_without_write_then_persists_canonical_proposal idempotency_key="cli-operation-fixture", request_path=request_path, execute=True, + executor_binding_resolver=lambda _parameters, _runtime: { + "revision": "simulator-v0" + }, ) assert applied["status"] == "awaiting_confirmation" assert applied["details"]["durable_proposal_written"] is True @@ -1371,3 +1380,274 @@ def test_forwarded_or_unauthorized_card_cannot_claim(tmp_path: Path) -> None: runner=runner, executor=lambda _proposal: {}, ) + + +def test_prepare_rejects_stale_executor_revision_before_any_persistence( + tmp_path: Path, +) -> None: + """A stale revision is blocked in preview and execute without any write.""" + + store, registry, runtime, _binding, _target = _fixture(tmp_path) + request_path = tmp_path / "operation.json" + request_path.write_text( + json.dumps( + { + "schema_version": "loopx_operation_request_v0", + "domain": "finance", + "operation_kind": "finance.order.simulate", + "operation_schema": "finance_order_intent_v0", + "payload_ref": "finance-order:stale-fixture", + "payload": {"asset": "SYNTH"}, + "payload_digest": _digest({"asset": "SYNTH"}), + "projection": { + "schema_version": "loopx_operation_projection_v0", + "title": "Simulated trade request", + "subtitle": "Synthetic fixture", + "focus": "BUY 1 SYNTH @ 10 TEST", + "fields": [{"label": "Order", "value": "Limit · GTC"}], + "warning": "Simulation only.", + "simulated": True, + }, + "destination_account_ref": "account:simulation", + "expires_at": ( + datetime.now(timezone.utc) + timedelta(hours=1) + ).isoformat(), + "authorized_principals": [f"lark:{OPERATOR_ID}"], + "executor": { + "extension_id": "loopx-finance-execution", + "protocol": "finance_operation_executor_v0", + "permission": "finance.operation.simulate", + "revision": "requested-v1", + }, + } + ), + encoding="utf-8", + ) + + for execute in (False, True): + with pytest.raises(OperationExecutorDriftError) as exc_info: + _prepare_goal_channel_operation( + registry_path=registry, + runtime_root=runtime, + goal_id=GOAL_ID, + agent_id=AGENT_ID, + summary="Review one simulated order", + idempotency_key=f"stale-operation-{execute}", + request_path=request_path, + execute=execute, + executor_binding_resolver=lambda _parameters, _runtime: { + "revision": "active-v9" + }, + ) + assert exc_info.value.blocker == "executor_revision_drift" + assert exc_info.value.failure_stage == "resolve_executor_binding" + assert exc_info.value.details == { + "requested_executor_revision": "requested-v1", + "active_executor_revision": "active-v9", + } + assert exc_info.value.external_write_performed is False + + durable = json.loads((store.root / "actions.json").read_text()) + assert durable["proposals"] == {} + assert durable["idempotency"] == {} + + +def test_prepare_rejects_unresolvable_executor_without_private_leak( + tmp_path: Path, +) -> None: + """Resolution failures project a typed blocker without private text.""" + + store, registry, runtime, _binding, _target = _fixture(tmp_path) + request_path = tmp_path / "operation.json" + request_path.write_text( + json.dumps( + { + "schema_version": "loopx_operation_request_v0", + "domain": "finance", + "operation_kind": "finance.order.simulate", + "operation_schema": "finance_order_intent_v0", + "payload_ref": "finance-order:unavailable-fixture", + "payload": {"asset": "SYNTH"}, + "payload_digest": _digest({"asset": "SYNTH"}), + "projection": { + "schema_version": "loopx_operation_projection_v0", + "title": "Simulated trade request", + "subtitle": "Synthetic fixture", + "focus": "BUY 1 SYNTH @ 10 TEST", + "fields": [{"label": "Order", "value": "Limit · GTC"}], + "warning": "Simulation only.", + "simulated": True, + }, + "destination_account_ref": "account:simulation", + "expires_at": ( + datetime.now(timezone.utc) + timedelta(hours=1) + ).isoformat(), + "authorized_principals": [f"lark:{OPERATOR_ID}"], + "executor": { + "extension_id": "loopx-finance-execution", + "protocol": "finance_operation_executor_v0", + "permission": "finance.operation.simulate", + "revision": "simulator-v0", + }, + } + ), + encoding="utf-8", + ) + + def broken_resolver(_parameters: object, _runtime: object) -> object: + raise ValueError("private resolver detail: /home/operator/secret-path") + + with pytest.raises(GoalChannelDeliveryStageError) as exc_info: + _prepare_goal_channel_operation( + registry_path=registry, + runtime_root=runtime, + goal_id=GOAL_ID, + agent_id=AGENT_ID, + summary="Review one simulated order", + idempotency_key="unavailable-operation", + request_path=request_path, + execute=True, + executor_binding_resolver=broken_resolver, + ) + + assert str(exc_info.value) == "operation executor binding is unavailable" + assert exc_info.value.blocker == "executor_unavailable" + assert "secret-path" not in str(exc_info.value) + durable = json.loads((store.root / "actions.json").read_text()) + assert durable["proposals"] == {} + assert durable["idempotency"] == {} + + +def test_delivery_projects_typed_stage_blockers(tmp_path: Path) -> None: + """Stage failures keep their typed blocker and stage at the deliver seam.""" + + def _failed_delivery( + case: str, + fail_args: tuple[str, ...], + *, + outcome: dict[str, Any] | None = None, + raises: Exception | None = None, + ) -> tuple[GoalChannelDeliveryStageError, list[list[str]]]: + case_root = tmp_path / case + case_root.mkdir() + store, registry, runtime, binding, target = _fixture(case_root) + proposal = _prepare(store, registry) + calls: list[list[str]] = [] + base = _runner(calls, {}) + + def runner( + args: list[str], cwd: Path | None, timeout: float | None + ) -> dict[str, Any]: + if all(fragment in args for fragment in fail_args): + calls.append(list(args)) + if raises is not None: + raise raises + if outcome is not None: + return dict(outcome) + return { + "returncode": 1, + "stdout": "", + "stderr": "provider rejected", + } + return base(args, cwd, timeout) + + with pytest.raises(GoalChannelDeliveryStageError) as exc_info: + deliver_goal_channel_operation_card( + proposal_id=proposal["proposal_id"], + action_store_root=store.root, + runtime_root=runtime, + binding_path=binding, + target_path=target, + execute=True, + runner=runner, + executor_binding_resolver=lambda _parameters, _runtime: { + "revision": "simulator-v0" + }, + ) + return exc_info.value, calls + + blocked, identity_calls = _failed_delivery("identity", ("auth", "status")) + assert blocked.blocker == "sender_identity_unverified" + assert blocked.failure_stage == "verify_sender_identity" + assert blocked.external_write_performed is False + assert not any("+messages-send" in call for call in identity_calls) + + blocked, dedupe_calls = _failed_delivery("dedupe", ("+chat-messages-list",)) + assert blocked.blocker == "dedupe_history_read_failed" + assert blocked.failure_stage == "read_dedupe_history" + assert blocked.external_write_performed is False + assert not any("+messages-send" in call for call in dedupe_calls) + + blocked, send_calls = _failed_delivery("send", ("+messages-send",)) + assert blocked.blocker == "provider_send_rejected" + assert blocked.failure_stage == "send_operation_card" + assert blocked.external_write_performed is False + + # A provider answer is the only thing that can be projected as a verdict. + # Without one the card may already be live, so the send stage must report an + # unknown outcome rather than a clean no-write. + for case, outcome, raises in ( + ("send-no-body", {"returncode": 1, "stdout": "", "stderr": ""}, None), + ( + "send-timeout", + None, + subprocess.TimeoutExpired(cmd="lark-cli", timeout=30), + ), + ): + blocked, send_calls = _failed_delivery( + case, ("+messages-send",), outcome=outcome, raises=raises + ) + assert blocked.blocker == "delivery_outcome_unknown", (case, blocked) + assert blocked.failure_stage == "send_operation_card", (case, blocked) + assert blocked.external_write_performed is None, (case, blocked) + assert any("+messages-send" in call for call in send_calls), case + + # A command that never started cannot have written anything, and says so + # with its own blocker instead of being reported as a provider rejection. + blocked, send_calls = _failed_delivery( + "send-unavailable", + ("+messages-send",), + raises=FileNotFoundError("lark-cli"), + ) + assert blocked.blocker == "provider_unavailable", blocked + assert blocked.failure_stage == "send_operation_card", blocked + assert blocked.external_write_performed is False, blocked + + +def test_receipt_treats_post_send_failure_as_unknown_write( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A failure after the provider write must not project a clean receipt.""" + + store, registry, runtime, binding, target = _fixture(tmp_path) + proposal = _prepare(store, registry) + calls: list[list[str]] = [] + + def failing_record( + self: ChatActionStore, *args: object, **kwargs: object + ) -> object: + raise OSError("synthetic receipt write failure") + + monkeypatch.setattr(ChatActionStore, "record_operation_delivery", failing_record) + + with pytest.raises(GoalChannelDeliveryStageError) as exc_info: + deliver_goal_channel_operation_card( + proposal_id=proposal["proposal_id"], + action_store_root=store.root, + runtime_root=runtime, + binding_path=binding, + target_path=target, + execute=True, + runner=_runner(calls, {}), + executor_binding_resolver=lambda _parameters, _runtime: { + "revision": "simulator-v0" + }, + ) + + # The readback already proved the card is live, so this stage keeps its own + # blocker instead of diluting the "provider outcome unknown" signal. + assert exc_info.value.blocker == "delivery_receipt_write_failed" + assert exc_info.value.failure_stage == "record_delivery_receipt" + assert exc_info.value.external_write_performed is None + assert any("+messages-send" in call for call in calls)