|
| 1 | +"""The host-adapter settlement plan step order has one registered vocabulary. |
| 2 | +
|
| 3 | +`loopx/control_plane/turn_driver/host_todo_completion.ts` authors the ordered |
| 4 | +provider plan and `loopx/control_plane/host_adapter_settlement.py` validates the |
| 5 | +plan it receives against the same sequence. Until now the two sequences were |
| 6 | +private, hand-kept copies: the TypeScript one was a module-local `const`, the |
| 7 | +Python one a module-local `_STEP_KINDS`. Neither was exported, so the |
| 8 | +repository's semantic inventory could not see either carrier, and neither |
| 9 | +`loopx/semantics/vocabulary_v0.json` nor the drift smoke could tie them |
| 10 | +together. A divergence between them would have surfaced at runtime as |
| 11 | +`TypeScript host provider step order mismatch`, after the plan had already been |
| 12 | +accepted and partially executed. |
| 13 | +
|
| 14 | +Both sequences are now registered owners of `host_adapter_settlement_step_kind`, |
| 15 | +so the drift smoke checks each of them against the registry on every run, and |
| 16 | +the tests below check the two against each other directly. |
| 17 | +""" |
| 18 | + |
| 19 | +from __future__ import annotations |
| 20 | + |
| 21 | +import json |
| 22 | +import re |
| 23 | +from pathlib import Path |
| 24 | + |
| 25 | +import pytest |
| 26 | + |
| 27 | +from loopx.control_plane.host_adapter_settlement import ( |
| 28 | + HOST_ADAPTER_SETTLEMENT_STEP_SEQUENCE, |
| 29 | + _decode_provider_steps, |
| 30 | +) |
| 31 | + |
| 32 | +REPOSITORY_ROOT = Path(__file__).resolve().parents[2] |
| 33 | +REGISTRY_PATH = REPOSITORY_ROOT / "loopx" / "semantics" / "vocabulary_v0.json" |
| 34 | +TYPESCRIPT_OWNER = REPOSITORY_ROOT / "loopx" / "control_plane" / "turn_driver" / "host_todo_completion.ts" |
| 35 | +VOCABULARY = "host_adapter_settlement_step_kind" |
| 36 | + |
| 37 | + |
| 38 | +def _typescript_step_kinds() -> tuple[str, ...]: |
| 39 | + """Read the TypeScript owner the way the semantic inventory reads it.""" |
| 40 | + source = TYPESCRIPT_OWNER.read_text(encoding="utf-8") |
| 41 | + match = re.search( |
| 42 | + r"export const HOST_ADAPTER_SETTLEMENT_STEP_KINDS = \[([^\]]*)\] as const;", |
| 43 | + source, |
| 44 | + ) |
| 45 | + assert match, "the TypeScript owner must export its step kinds as an as-const array" |
| 46 | + return tuple(re.findall(r'"([^"]+)"', match.group(1))) |
| 47 | + |
| 48 | + |
| 49 | +def test_both_runtimes_carry_the_same_step_sequence() -> None: |
| 50 | + """The plan the TypeScript owner writes must satisfy the Python transport.""" |
| 51 | + assert _typescript_step_kinds() == tuple(HOST_ADAPTER_SETTLEMENT_STEP_SEQUENCE) |
| 52 | + |
| 53 | + |
| 54 | +def test_the_registry_entry_names_both_owners_and_the_same_values() -> None: |
| 55 | + """The registered vocabulary is what the drift smoke checks both owners against.""" |
| 56 | + registry = json.loads(REGISTRY_PATH.read_text(encoding="utf-8")) |
| 57 | + entry = registry["vocabularies"][VOCABULARY] |
| 58 | + assert entry["owners"]["python"] == ( |
| 59 | + "loopx/control_plane/host_adapter_settlement.py::HOST_ADAPTER_SETTLEMENT_STEP_SEQUENCE" |
| 60 | + ) |
| 61 | + assert entry["owners"]["typescript"] == ( |
| 62 | + "loopx/control_plane/turn_driver/host_todo_completion.ts::HOST_ADAPTER_SETTLEMENT_STEP_KINDS" |
| 63 | + ) |
| 64 | + assert entry["values"] == list(HOST_ADAPTER_SETTLEMENT_STEP_SEQUENCE) |
| 65 | + |
| 66 | + |
| 67 | +def _plan(step_kinds: list[str]) -> dict: |
| 68 | + """One minimal provider plan: the last step carries no continuation condition.""" |
| 69 | + steps = [] |
| 70 | + for index, step_kind in enumerate(step_kinds): |
| 71 | + steps.append({ |
| 72 | + "step_kind": step_kind, |
| 73 | + "args": ["todo", "complete"], |
| 74 | + "legacy_args": None, |
| 75 | + "continue_when": {"all": []} if index < len(step_kinds) - 1 else None, |
| 76 | + }) |
| 77 | + return {"provider_id": "loopx_cli", "kind": "ordered_cli_sequence", "steps": steps} |
| 78 | + |
| 79 | + |
| 80 | +def test_the_transport_accepts_the_registered_order() -> None: |
| 81 | + steps = _decode_provider_steps(_plan(list(HOST_ADAPTER_SETTLEMENT_STEP_SEQUENCE))) |
| 82 | + assert [step.step_kind for step in steps] == list(HOST_ADAPTER_SETTLEMENT_STEP_SEQUENCE) |
| 83 | + |
| 84 | + |
| 85 | +def test_the_transport_rejects_a_plan_whose_step_order_drifts() -> None: |
| 86 | + """The tripwire the registry now backs: a reordered plan is refused, not run.""" |
| 87 | + drifted = list(HOST_ADAPTER_SETTLEMENT_STEP_SEQUENCE) |
| 88 | + drifted[0], drifted[1] = drifted[1], drifted[0] |
| 89 | + with pytest.raises(RuntimeError, match="step order mismatch"): |
| 90 | + _decode_provider_steps(_plan(drifted)) |
0 commit comments