From 168e03d9e778ab2ec63502552d385ec7be3ef677 Mon Sep 17 00:00:00 2001 From: LittleSam129 <1qwe432@gmail.com> Date: Tue, 22 Sep 2026 15:09:37 +0300 Subject: [PATCH] Add MAIN-to-RX acknowledgement channel --- docs/handoff/NEXT_RX_TASK.md | 23 +- docs/handoff/RF_HANDOFF_CHANNEL_TEST.md | 6 +- docs/handoff/RX_AGENT_PROTOCOL.md | 20 +- docs/handoff/RX_AGENT_USAGE.md | 19 +- docs/handoff/TX_AGENT_USAGE.md | 24 +- tests/test_tx_handoff_acknowledgement.py | 489 +++++++++++++++++++ tools/rx_agent_watcher.py | 93 +++- tools/tx_agent_watcher.py | 92 +++- tools/tx_handoff_channel.py | 594 +++++++++++++++++++++++ 9 files changed, 1330 insertions(+), 30 deletions(-) create mode 100644 tests/test_tx_handoff_acknowledgement.py create mode 100644 tools/tx_handoff_channel.py diff --git a/docs/handoff/NEXT_RX_TASK.md b/docs/handoff/NEXT_RX_TASK.md index f7102de..4183aab 100644 --- a/docs/handoff/NEXT_RX_TASK.md +++ b/docs/handoff/NEXT_RX_TASK.md @@ -21,19 +21,30 @@ START COORDINATED RF-001A COMMON-CLOCK OTA CAPTURE This is a new YELLOW task. Do not reuse approval or completion state from RF001A_OTA_CAPTURE_001, RF001A_OTA_CAPTURE_002 or any earlier task. -One approval for this task covers the model-free channel test, read-only checks, one bounded RF RX capture and publication of the two small allowlisted result files to the existing rx-node branch. It never authorizes RF TX. +One approval for this task covers the read-only bidirectional channel check, other read-only checks, one bounded RF RX capture and publication of the two small allowlisted result files to the existing rx-node branch. It never authorizes RF TX. The preceding RF001A_OTA_CAPTURE_002 produced a real 120.1413067 s, 1920000000-byte capture with CLKIN detected before and after, zero reported overruns and SHA256 99a3e7cb51d3ad4e3b914b63da0695ed6658aa169110c32659aba995b542fa0c. Its READY publication failed, its exact child exit code was not recorded and no synchronized truth metadata was available. Treat that as historical evidence only. -## 1. Model-free handoff channel gate +## 1. Bidirectional handoff channel gate -After the user chooses [A] Approve for this exact TASK_ID, run before any hardware access: +Before offering this task for YELLOW approval, the RX watcher must fetch origin/rx-node and origin/tx-node and scan the acknowledgement history from oldest to newest. It must confirm both exact records: - python -B tools/rf_handoff_smoke.py --hold-seconds 7 + STATE = CHANNEL_READY_OBSERVED + SOURCE_RX_COMMIT = 4cb6c8cad1ab5194755c2740c2a5a83933d099d8 + SOURCE_RX_DIGEST = 23fc54f1f788f068fd4a842daf0a445bd4ec192866b4d89406c74f1797a9952b -This subtask uses TASK_ID RF_HANDOFF_CHANNEL_TEST_001. It accesses no SDR and keeps RF TX = 0. It publishes fake READY and COMPLETE results through the real origin of the main RX checkout. Require the MAIN watcher to observe both states, require confirmed remote commit hashes, and require COMPLETE to remain as the final origin/rx-node state. Require the MAIN watcher state to contain both channel_test_ready_digest and channel_test_complete_digest for this run. +and: -If any channel check or publication fails, do not access HackRF, do not start capture and STOP with RF TX = 0. + STATE = CHANNEL_COMPLETE_OBSERVED + SOURCE_RX_COMMIT = 83d91ff5309777ab639a24f774e9d0b8825a11e9 + SOURCE_RX_DIGEST = 7d98ed2e70e98a087b968c0d65fb1db4277b81f251e075d20583de2e5002b2d6 + +Both acknowledgements must use TASK_ID RF_HANDOFF_CHANNEL_TEST_001, the original shared source HEAD and RF_TX = 0. Required watcher output: + + MAIN READY RECEIVED = CONFIRMED + MAIN COMPLETE RECEIVED = CONFIRMED + +Do not rerun the channel test. If origin/tx-node is missing or either exact acknowledgement is absent or invalid, do not access HackRF, do not start capture and STOP with RF TX = 0. ## 2. Preconditions diff --git a/docs/handoff/RF_HANDOFF_CHANNEL_TEST.md b/docs/handoff/RF_HANDOFF_CHANNEL_TEST.md index ec751a2..2e9a3c3 100644 --- a/docs/handoff/RF_HANDOFF_CHANNEL_TEST.md +++ b/docs/handoff/RF_HANDOFF_CHANNEL_TEST.md @@ -12,7 +12,7 @@ RISK_LEVEL = GREEN # Model-free handoff channel test -Run this gate before RF001A_OTA_CAPTURE_003 and while the MAIN watcher is active: +The real-origin gate was completed before RF001A_OTA_CAPTURE_003 with: python -B tools/rf_handoff_smoke.py --hold-seconds 7 @@ -21,7 +21,9 @@ The script uses the real origin configured in the main RX checkout. It publishes docs/handoff/results/RX_RESULT.json docs/handoff/results/RX_RESULT.md -The MAIN watcher must observe both states and must not request RED approval. Its state file must record both channel_test_ready_digest and channel_test_complete_digest. The final remote state must be CAPTURE_COMPLETE, with rx_ready_for_tx false. A synthetic READY state must not remain active. +The verified READY commit is `4cb6c8cad1ab5194755c2740c2a5a83933d099d8`; its canonical digest is `23fc54f1f788f068fd4a842daf0a445bd4ec192866b4d89406c74f1797a9952b`. The verified COMPLETE commit is `83d91ff5309777ab639a24f774e9d0b8825a11e9`; its canonical digest is `7d98ed2e70e98a087b968c0d65fb1db4277b81f251e075d20583de2e5002b2d6`. + +The MAIN watcher observed both states without requesting RED approval. The RX watcher must now verify corresponding CHANNEL_READY_OBSERVED and CHANNEL_COMPLETE_OBSERVED records in the full origin/tx-node history. Do not rerun this channel test only to recreate MAIN confirmation. The script and watcher do not access SDR hardware, do not run hackrf_transfer and do not authorize any RF transmission. diff --git a/docs/handoff/RX_AGENT_PROTOCOL.md b/docs/handoff/RX_AGENT_PROTOCOL.md index 3d61236..57391b3 100644 --- a/docs/handoff/RX_AGENT_PROTOCOL.md +++ b/docs/handoff/RX_AGENT_PROTOCOL.md @@ -4,7 +4,9 @@ The RX laptop watches origin/video-development with ordinary local Python code. Git polling does not call an LLM, ChatGPT or an agent. A detected task is not prepared until the user approves it. -The reverse result channel uses the separate origin/rx-node branch. The RX laptop publishes small machine-readable readiness and completion results. The MAIN PC reads those results without merging rx-node into video-development. +The RX-to-MAIN result channel uses the separate origin/rx-node branch. The RX laptop publishes small machine-readable readiness and completion results. The MAIN PC reads those results without merging rx-node into video-development. + +The MAIN-to-RX acknowledgement channel uses the independent origin/tx-node branch. The MAIN watcher publishes evidence that it observed one exact rx-node commit and digest. The RX laptop scans the acknowledgement history without merging tx-node into video-development. ## Task header @@ -57,7 +59,9 @@ State includes: - last_approved_task_id; - deferred_task_id; - running; -- watcher PID. +- watcher PID; +- confirmed channel-test READY and COMPLETE digests; +- matching tx-node acknowledgement commits. ManualExecutor writes TASK.md, RX_RESULT.md and RX_RESULT.json after approval. It does not invoke an agent, access SDR hardware, commit or push. @@ -122,7 +126,11 @@ For RF_HANDOFF_CHANNEL_TEST_001 the watcher records synthetic READY and COMPLETE After the phrase, it fetches rx-node again and repeats every check. A changed result or commit, stale publication or fewer than 10 seconds remaining blocks authorization. Historical READY is still observed in order, but an expired capture window never authorizes TX. -The history cursor advances only after successful schema validation and handling. Duplicate content advances the commit cursor without repeating an action. A malformed intermediate result blocks later commits, and a non-fast-forward branch rewrite is rejected. +The history cursor advances only after successful schema validation, handling and required acknowledgement publication. Duplicate content advances the commit cursor without repeating an action. A malformed intermediate result blocks later commits, and a non-fast-forward branch rewrite is rejected. + +For each successfully handled channel-test event, the MAIN watcher publishes a strict TX_RESULT to origin/tx-node. The object contains task ID, acknowledgement state, exact source RX commit and digest, source HEAD, actual observation timestamp, watcher PID and RF safety fields. A channel-test acknowledgement always has RF_TX_AUTHORIZED false, RF_TX_STARTED false and RF_TX 0. + +The tx-node publisher stages exactly `docs/handoff/results/TX_RESULT.json` and `docs/handoff/results/TX_RESULT.md`, checks the staged path allowlist and enforces 64 KiB per file. A push failure prevents the rx-node cursor from advancing. Duplicate acknowledgements are suppressed by their source binding. No local state, code, IQ, video, waveform or large log is published. A successful gate writes one local authorization record under .local/tx_authorizations/. The record has rf_tx_started = false and rf_tx = 0. The separately controlled TX executor must re-fetch and compare the result commit and digest immediately before opening hardware. @@ -137,8 +145,8 @@ Duplicate result digests are not shown repeatedly. - Neither watcher starts RF TX. - The TX executor remains a separate RED-controlled action. -## Model-free channel and dry-run contracts +## Bidirectional channel and dry-run contracts -The preflight channel command publishes synthetic READY and COMPLETE results with TASK_ID RF_HANDOFF_CHANNEL_TEST_001 through the real origin. The MAIN watcher must observe both without a RED prompt, and COMPLETE must be the final remote state. It does not access SDR hardware or run hackrf_transfer. +The completed preflight channel command published synthetic READY and COMPLETE results with TASK_ID RF_HANDOFF_CHANNEL_TEST_001 through origin/rx-node. The MAIN watcher observed both without a RED prompt. The RX watcher now requires exact matching acknowledgements in origin/tx-node before RF001A_OTA_CAPTURE_003 can pass its software gate. -Automated coordination tests use a temporary local bare Git repository and branch named rx-node. They cover worktree publication, READY, COMPLETE, push failure and watcher reception without accessing the real remote or SDR hardware and without transmitting RF. +Automated coordination tests use a temporary local bare Git repository and independent rx-node and tx-node branches. They cover exact commit and digest binding, duplicate suppression, two commits between polls, stale acknowledgement rejection, missing tx-node, push failure, non-fast-forward protection and a full fake two-way flow without accessing the real remote or SDR hardware and without transmitting RF. diff --git a/docs/handoff/RX_AGENT_USAGE.md b/docs/handoff/RX_AGENT_USAGE.md index 1deb215..a025da5 100644 --- a/docs/handoff/RX_AGENT_USAGE.md +++ b/docs/handoff/RX_AGENT_USAGE.md @@ -10,7 +10,7 @@ The watcher stays in a visible console and polls origin/video-development every ## Idle -While no new TASK_ID exists, the watcher performs only git fetch, reads the remote task file and compares IDs with .local/rx_agent_state.json. There is no model execution or token use. +While no new TASK_ID exists, the watcher performs only git fetch, reads the remote task file and compares IDs with .local/rx_agent_state.json. There is no model execution or token use. For RF001A_OTA_CAPTURE_003 it also verifies the required historical acknowledgements on origin/tx-node before offering the task for approval. ## New task @@ -41,7 +41,14 @@ When the task is actually complete, run: ## Publish READY -For RF001A_OTA_CAPTURE_003, first complete RF_HANDOFF_CHANNEL_TEST_001. Then create the strict READY JSON outside Git only after CLKIN is detected, the capture child is alive and the file is growing. Run: +For RF001A_OTA_CAPTURE_003, first require the watcher to print: + + MAIN READY RECEIVED = CONFIRMED + MAIN COMPLETE RECEIVED = CONFIRMED + +These confirmations come from the full origin/tx-node history. They must bind the known channel-test RX commits and canonical digests. The RX watcher stores the two digests and the matching tx-node commit hashes in its local state. A missing branch, wrong digest, wrong RX commit, wrong source HEAD, malformed intermediate acknowledgement or rewritten history blocks the task. + +Do not rerun RF_HANDOFF_CHANNEL_TEST_001 merely to recreate these confirmations. After this gate and the task's YELLOW approval, create the strict READY JSON outside Git only after CLKIN is detected, the capture child is alive and the file is growing. Run: python -B tools/rx_agent_watcher.py --publish-result --result-branch rx-node @@ -58,13 +65,9 @@ The command fails if: A failed publication is not READY evidence and cannot authorize TX. -## Model-free channel test +## Verified model-free channel evidence -Before hardware access, while the MAIN watcher is running, execute: - - python -B tools/rf_handoff_smoke.py --hold-seconds 7 - -It publishes synthetic READY and COMPLETE results through the real origin. Require the MAIN watcher to observe both and require COMPLETE as the final remote state. Failure blocks the hardware capture. REAL RX remains NOT_RUN and RF TX remains 0. +The completed RF_HANDOFF_CHANNEL_TEST_001 evidence is bound to READY commit `4cb6c8cad1ab5194755c2740c2a5a83933d099d8` and COMPLETE commit `83d91ff5309777ab639a24f774e9d0b8825a11e9`. The RX watcher accepts the gate only after origin/tx-node contains the matching MAIN acknowledgements. This read-only check performs no capture and keeps RF TX at 0. ## Publish COMPLETE diff --git a/docs/handoff/TX_AGENT_USAGE.md b/docs/handoff/TX_AGENT_USAGE.md index 273ef7f..64475e4 100644 --- a/docs/handoff/TX_AGENT_USAGE.md +++ b/docs/handoff/TX_AGENT_USAGE.md @@ -2,7 +2,7 @@ ## Purpose -tools/tx_agent_watcher.py is an ordinary local Python process. It polls the small RX result on origin/rx-node. It does not call an LLM and does not open HackRF hardware. +tools/tx_agent_watcher.py is an ordinary local Python process. It polls the small RX result on origin/rx-node and publishes small acknowledgements to origin/tx-node. It does not call an LLM and does not open HackRF hardware. ## Start @@ -10,7 +10,7 @@ From the repository root, double-click: tools\start_tx_watcher.cmd -The visible console polls every 5 seconds. A missing rx-node branch or invalid result is reported. No model runs while polling. +The visible console polls every 5 seconds. A missing rx-node or tx-node branch, an invalid result or an acknowledgement push failure is reported. No model runs while polling or publishing. ## Result history cursor @@ -20,13 +20,29 @@ The watcher stores `last_processed_rx_node_commit` in `.local\tx_agent_watcher_s It reads `RX_RESULT.json` from each exact commit and processes the commits from oldest to newest. READY therefore remains observable when READY and COMPLETE are both published between two polling cycles. -The cursor advances only after that commit has been read, validated and handled successfully. A malformed intermediate commit blocks later commits until the defect is corrected. A non-fast-forward rewrite of `rx-node` is rejected. Duplicate result content is suppressed, but its commit is still marked as processed. +The cursor advances only after that commit has been read, validated, handled and, when required, acknowledged successfully. A failed acknowledgement push leaves the RX commit unprocessed so that the next poll retries it. A malformed intermediate commit blocks later commits until the defect is corrected. A non-fast-forward rewrite of `rx-node` is rejected. Duplicate result content is suppressed, but its commit is still marked as processed. State format version 2 retains the existing digest and authorization fields and adds `last_seen_commit` and `last_processed_rx_node_commit`. A version-1 digest-only state is mapped to the newest matching commit in the existing result history. Back up the local state before manual recovery; do not delete it blindly. +## MAIN acknowledgements + +After it handles a channel-test READY or COMPLETE event, the watcher publishes a strict acknowledgement to origin/tx-node. Each acknowledgement binds the exact `SOURCE_RX_COMMIT`, canonical `SOURCE_RX_DIGEST` and RX result `SOURCE_HEAD`. The channel-test states are: + + CHANNEL_READY_OBSERVED + CHANNEL_COMPLETE_OBSERVED + +The tx-node branch is independent of video-development. Every commit tree may contain only: + + docs/handoff/results/TX_RESULT.json + docs/handoff/results/TX_RESULT.md + +Each file is limited to 64 KiB. Before committing, the publisher compares the complete staged path set with this allowlist. It suppresses an acknowledgement that already exists for the same task, state, RX commit, digest, source HEAD and RF safety values. The RX laptop scans the full tx-node history from oldest to newest, so two acknowledgements published between polls remain observable. + +The acknowledgement operation is ordinary Python and Git. It does not invoke ChatGPT, Codex or another model. It stores no IQ, video, waveform, source code, large logs or local state on tx-node. + ## READY and RED approval -RF_HANDOFF_CHANNEL_TEST_001 READY and COMPLETE states are recorded without a RED prompt and never open the production gate. +RF_HANDOFF_CHANNEL_TEST_001 READY and COMPLETE states are recorded and acknowledged without a RED prompt and never open the production gate. For a fresh valid RF001A_OTA_CAPTURE_003 READY result, the watcher checks the expected source HEAD, current origin/rx-node commit, common clock and capture window. It then prints the task ID, capture start and expected end and asks for: diff --git a/tests/test_tx_handoff_acknowledgement.py b/tests/test_tx_handoff_acknowledgement.py new file mode 100644 index 0000000..d26b84b --- /dev/null +++ b/tests/test_tx_handoff_acknowledgement.py @@ -0,0 +1,489 @@ +"""Model-free tests for the MAIN-to-RX acknowledgement channel.""" + +from __future__ import annotations + +from dataclasses import replace +from datetime import datetime, timedelta, timezone +from pathlib import Path +import subprocess + +import pytest + +import tools.rx_agent_watcher as rx_agent_watcher + +from tools.rf_handoff_channel import ( + CHANNEL_TEST_TASK_ID, + COMPLETE_STATE, + GitResultChannel, + READY_STATE, + ResultBranchMissing, + ResultChannelError, + ResultHistoryRewritten, + RxHandoffResult, + result_digest, +) +from tools.tx_agent_watcher import TxWatcherState, process_history_cycle +from tools.tx_handoff_channel import ( + CHANNEL_COMPLETE_OBSERVED, + CHANNEL_READY_OBSERVED, + TX_RESULT_PATHS, + GitAcknowledgementChannel, + TxHandoffAcknowledgement, + acknowledgement_from_mapping, + channel_test_acknowledgements, +) + + +NOW = datetime(2026, 9, 22, 12, 0, tzinfo=timezone.utc) +SOURCE_HEAD = "a" * 40 + + +def _git(arguments: list[str], cwd: Path) -> str: + completed = subprocess.run( + ["git", *arguments], + cwd=cwd, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + check=True, + ) + return completed.stdout.strip() + + +def repository_fixture( + tmp_path: Path, + *, + with_tx_node: bool = True, +) -> tuple[Path, Path, str | None]: + remote = tmp_path / "remote.git" + work = tmp_path / "work" + _git(["init", "--bare", str(remote)], tmp_path) + work.mkdir() + _git(["init"], work) + _git(["config", "user.name", "Test"], work) + _git(["config", "user.email", "test@example.invalid"], work) + (work / "README.md").write_text("fixture\n", encoding="utf-8") + _git(["add", "README.md"], work) + _git(["commit", "-m", "Seed RX result branch"], work) + initial_branch = _git(["branch", "--show-current"], work) + _git(["remote", "add", "origin", str(remote)], work) + _git(["push", "origin", "HEAD:refs/heads/rx-node"], work) + tx_root: str | None = None + if with_tx_node: + _git(["switch", "--orphan", "tx-node"], work) + _git(["rm", "-rf", "--ignore-unmatch", "."], work) + _git(["commit", "--allow-empty", "-m", "Initialize tx-node"], work) + tx_root = _git(["rev-parse", "HEAD"], work) + _git(["push", "origin", "HEAD:refs/heads/tx-node"], work) + _git(["switch", initial_branch], work) + return remote, work, tx_root + + +def channel_ready(**changes: object) -> RxHandoffResult: + result = RxHandoffResult( + schema_version=1, + task_id=CHANNEL_TEST_TASK_ID, + state=READY_STATE, + rx_ready_for_tx=True, + process_alive=True, + file_growing=True, + capture_pid=1234, + published_at=NOW.isoformat(), + capture_started_at=NOW.isoformat(), + expected_capture_end=(NOW + timedelta(seconds=120)).isoformat(), + capture_path="MODEL_FREE_FAKE_CAPTURE", + common_clock_verified=False, + source_head=SOURCE_HEAD, + rf_tx=0, + exit_code=None, + file_size=None, + duration_seconds=None, + overruns=None, + usb_errors=None, + file_errors=None, + sha256=None, + clkin_after="NOT_RUN", + rf_tx_observed="UNKNOWN", + ) + return replace(result, **changes) + + +def channel_complete( + ready: RxHandoffResult, + **changes: object, +) -> RxHandoffResult: + result = replace( + ready, + state=COMPLETE_STATE, + rx_ready_for_tx=False, + process_alive=False, + file_growing=False, + published_at=(NOW + timedelta(seconds=1)).isoformat(), + exit_code=0, + file_size=4, + duration_seconds=1.0, + overruns=0, + usb_errors=0, + file_errors=0, + sha256="b" * 64, + ) + return replace(result, **changes) + + +def acknowledgement( + state: str, + source_commit: str, + source_digest: str, + **changes: object, +) -> TxHandoffAcknowledgement: + result = TxHandoffAcknowledgement( + schema_version=1, + task_id=CHANNEL_TEST_TASK_ID, + state=state, + source_rx_commit=source_commit, + source_rx_digest=source_digest, + source_head=SOURCE_HEAD, + observed_at=NOW.isoformat(), + main_watcher_pid=4321, + rf_tx_authorized=False, + rf_tx_started=False, + rf_tx=0, + ) + return replace(result, **changes) + + +class FakeExecutor: + def __init__(self) -> None: + self.calls = 0 + + def prepare(self, *_args: object, **_kwargs: object) -> Path: + self.calls += 1 + raise AssertionError("channel test must not prepare RF TX") + + +def test_acknowledgement_schema_binds_digest_and_source_commit() -> None: + valid = acknowledgement(CHANNEL_READY_OBSERVED, "1" * 40, "2" * 64) + raw = valid.__dict__.copy() + assert acknowledgement_from_mapping(raw) == valid + + with pytest.raises(ValueError, match="source_rx_commit"): + acknowledgement_from_mapping({**raw, "source_rx_commit": "1" * 39}) + with pytest.raises(ValueError, match="source_rx_digest"): + acknowledgement_from_mapping({**raw, "source_rx_digest": "2" * 63}) + with pytest.raises(ValueError, match="cannot claim RF TX"): + acknowledgement_from_mapping({**raw, "rf_tx_authorized": True}) + + +def test_main_publish_uses_exact_allowlist_and_suppresses_duplicate( + tmp_path: Path, +) -> None: + remote, work, tx_root = repository_fixture(tmp_path) + channel = GitAcknowledgementChannel(work) + item = acknowledgement(CHANNEL_READY_OBSERVED, "1" * 40, "2" * 64) + + first = channel.publish(item) + second = channel.publish(item) + + assert first.status == "PUBLISHED" + assert second.status == "UNCHANGED" + assert second.commit == first.commit + assert tx_root is not None + assert _git( + ["--git-dir", str(remote), "rev-list", "--count", "refs/heads/tx-node"], + tmp_path, + ) == "2" + tree = set( + _git( + [ + "--git-dir", + str(remote), + "ls-tree", + "-r", + "--name-only", + "refs/heads/tx-node", + ], + tmp_path, + ).splitlines() + ) + assert tree == TX_RESULT_PATHS + + +def test_two_acknowledgements_between_polls_are_both_confirmed( + tmp_path: Path, +) -> None: + _remote, work, _tx_root = repository_fixture(tmp_path) + channel = GitAcknowledgementChannel(work) + ready_commit = "1" * 40 + ready_digest = "2" * 64 + complete_commit = "3" * 40 + complete_digest = "4" * 64 + ready_outcome = channel.publish( + acknowledgement(CHANNEL_READY_OBSERVED, ready_commit, ready_digest) + ) + complete_outcome = channel.publish( + acknowledgement( + CHANNEL_COMPLETE_OBSERVED, + complete_commit, + complete_digest, + observed_at=(NOW + timedelta(seconds=1)).isoformat(), + ) + ) + + status = channel_test_acknowledgements( + channel, + expected_source_head=SOURCE_HEAD, + ready_source_commit=ready_commit, + ready_source_digest=ready_digest, + complete_source_commit=complete_commit, + complete_source_digest=complete_digest, + ) + + assert status.ready_confirmed + assert status.complete_confirmed + assert status.ready_ack_commit == ready_outcome.commit + assert status.complete_ack_commit == complete_outcome.commit + + +@pytest.mark.parametrize("wrong_field", ["digest", "commit"]) +def test_wrong_binding_does_not_confirm_ready( + tmp_path: Path, + wrong_field: str, +) -> None: + _remote, work, _tx_root = repository_fixture(tmp_path) + channel = GitAcknowledgementChannel(work) + expected_commit = "1" * 40 + expected_digest = "2" * 64 + channel.publish( + acknowledgement( + CHANNEL_READY_OBSERVED, + "9" * 40 if wrong_field == "commit" else expected_commit, + "8" * 64 if wrong_field == "digest" else expected_digest, + ) + ) + + status = channel_test_acknowledgements( + channel, + expected_source_head=SOURCE_HEAD, + ready_source_commit=expected_commit, + ready_source_digest=expected_digest, + complete_source_commit="3" * 40, + complete_source_digest="4" * 64, + ) + assert not status.ready_confirmed + assert not status.complete_confirmed + + +def test_stale_acknowledgement_with_old_source_head_is_rejected( + tmp_path: Path, +) -> None: + _remote, work, _tx_root = repository_fixture(tmp_path) + channel = GitAcknowledgementChannel(work) + channel.publish( + acknowledgement( + CHANNEL_READY_OBSERVED, + "1" * 40, + "2" * 64, + source_head="f" * 40, + ) + ) + status = channel_test_acknowledgements( + channel, + expected_source_head=SOURCE_HEAD, + ready_source_commit="1" * 40, + ready_source_digest="2" * 64, + complete_source_commit="3" * 40, + complete_source_digest="4" * 64, + ) + assert not status.ready_confirmed + + +def test_missing_tx_node_is_blocked(tmp_path: Path) -> None: + _remote, work, _tx_root = repository_fixture(tmp_path, with_tx_node=False) + channel = GitAcknowledgementChannel(work) + with pytest.raises(ResultBranchMissing): + channel_test_acknowledgements( + channel, + expected_source_head=SOURCE_HEAD, + ) + + +def test_acknowledgement_push_failure_preserves_remote_head( + tmp_path: Path, +) -> None: + _remote, work, _tx_root = repository_fixture(tmp_path) + + class FailingChannel(GitAcknowledgementChannel): + def _run( + self, + arguments: list[str], + *, + cwd: Path | None = None, + check: bool = True, + ) -> subprocess.CompletedProcess[str]: + if arguments and arguments[0] == "push": + return subprocess.CompletedProcess( + ["git", *arguments], + 1, + "", + "simulated push failure", + ) + return super()._run(arguments, cwd=cwd, check=check) + + channel = FailingChannel(work) + before = channel.remote_head() + with pytest.raises(ResultChannelError, match="simulated push failure"): + channel.publish( + acknowledgement(CHANNEL_READY_OBSERVED, "1" * 40, "2" * 64) + ) + assert channel.remote_head() == before + + +def test_tx_node_history_rewrite_is_rejected(tmp_path: Path) -> None: + _remote, work, tx_root = repository_fixture(tmp_path) + channel = GitAcknowledgementChannel(work) + outcome = channel.publish( + acknowledgement(CHANNEL_READY_OBSERVED, "1" * 40, "2" * 64) + ) + assert tx_root is not None + _git( + ["push", "--force", "origin", f"{tx_root}:refs/heads/tx-node"], + work, + ) + with pytest.raises(ResultHistoryRewritten): + channel.commits_since(outcome.commit) + + +def test_fake_two_way_flow_is_event_safe_and_uses_no_tx( + tmp_path: Path, +) -> None: + _remote, work, _tx_root = repository_fixture(tmp_path) + rx_channel = GitResultChannel(work) + tx_channel = GitAcknowledgementChannel(work) + ready = channel_ready() + ready_outcome = rx_channel.publish(ready) + complete = channel_complete(ready) + complete_outcome = rx_channel.publish(complete) + state = TxWatcherState() + executor = FakeExecutor() + + outcomes = process_history_cycle( + rx_channel, + state, + executor, + now_function=lambda: NOW + timedelta(seconds=2), + input_function=lambda _prompt: "AUTHORIZE RF TX", + save_function=lambda _state: None, + output_function=lambda _text: None, + expected_source_head=SOURCE_HEAD, + acknowledgement_publisher=tx_channel, + main_watcher_pid=4321, + ) + + assert outcomes == [ + (ready_outcome.commit, "CHANNEL_TEST_READY"), + (complete_outcome.commit, "CHANNEL_TEST_COMPLETE"), + ] + confirmations = channel_test_acknowledgements( + tx_channel, + expected_source_head=SOURCE_HEAD, + ready_source_commit=ready_outcome.commit, + ready_source_digest=result_digest(ready), + complete_source_commit=complete_outcome.commit, + complete_source_digest=result_digest(complete), + ) + assert confirmations.ready_confirmed + assert confirmations.complete_confirmed + assert executor.calls == 0 + assert state.last_authorized_digest is None + + +def test_rx_watcher_confirms_both_main_acknowledgements( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + _remote, work, _tx_root = repository_fixture(tmp_path) + rx_channel = GitResultChannel(work) + tx_channel = GitAcknowledgementChannel(work) + ready = channel_ready() + ready_outcome = rx_channel.publish(ready) + complete = channel_complete(ready) + complete_outcome = rx_channel.publish(complete) + tx_channel.publish( + acknowledgement( + CHANNEL_READY_OBSERVED, + ready_outcome.commit, + result_digest(ready), + ) + ) + tx_channel.publish( + acknowledgement( + CHANNEL_COMPLETE_OBSERVED, + complete_outcome.commit, + result_digest(complete), + observed_at=(NOW + timedelta(seconds=1)).isoformat(), + ) + ) + monkeypatch.setattr( + rx_agent_watcher, + "CHANNEL_TEST_READY_COMMIT", + ready_outcome.commit, + ) + monkeypatch.setattr( + rx_agent_watcher, + "CHANNEL_TEST_READY_DIGEST", + result_digest(ready), + ) + monkeypatch.setattr( + rx_agent_watcher, + "CHANNEL_TEST_COMPLETE_COMMIT", + complete_outcome.commit, + ) + monkeypatch.setattr( + rx_agent_watcher, + "CHANNEL_TEST_COMPLETE_DIGEST", + result_digest(complete), + ) + state = rx_agent_watcher.WatcherState() + output: list[str] = [] + + rx_agent_watcher.confirm_main_channel_test( + work, + state, + output_function=output.append, + ) + + assert state.channel_test_ready_digest == result_digest(ready) + assert state.channel_test_complete_digest == result_digest(complete) + assert state.channel_test_ready_ack_commit is not None + assert state.channel_test_complete_ack_commit is not None + assert "MAIN READY RECEIVED = CONFIRMED" in output + assert "MAIN COMPLETE RECEIVED = CONFIRMED" in output + + +def test_ack_push_failure_does_not_advance_rx_cursor(tmp_path: Path) -> None: + _remote, work, _tx_root = repository_fixture(tmp_path) + rx_channel = GitResultChannel(work) + ready_outcome = rx_channel.publish(channel_ready()) + + class FailingPublisher: + def publish(self, acknowledgement: TxHandoffAcknowledgement) -> object: + del acknowledgement + raise ResultChannelError("simulated acknowledgement push failure") + + state = TxWatcherState() + with pytest.raises(ResultChannelError, match="simulated acknowledgement"): + process_history_cycle( + rx_channel, + state, + FakeExecutor(), + now_function=lambda: NOW, + input_function=lambda _prompt: "DENY", + save_function=lambda _state: None, + output_function=lambda _text: None, + expected_source_head=SOURCE_HEAD, + acknowledgement_publisher=FailingPublisher(), + ) + assert state.last_processed_rx_node_commit is None + assert state.channel_test_ready_digest is None + assert ready_outcome.commit != state.last_seen_commit diff --git a/tools/rx_agent_watcher.py b/tools/rx_agent_watcher.py index 272fe18..2ed6205 100644 --- a/tools/rx_agent_watcher.py +++ b/tools/rx_agent_watcher.py @@ -20,7 +20,19 @@ ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) -from tools.rf_handoff_channel import GitResultChannel, load_result # noqa: E402 +from tools.rf_handoff_channel import ( # noqa: E402 + EXPECTED_RX_TASK_ID, + GitResultChannel, + load_result, +) +from tools.tx_handoff_channel import ( # noqa: E402 + CHANNEL_TEST_COMPLETE_COMMIT, + CHANNEL_TEST_COMPLETE_DIGEST, + CHANNEL_TEST_READY_COMMIT, + CHANNEL_TEST_READY_DIGEST, + GitAcknowledgementChannel, + channel_test_acknowledgements, +) DEFAULT_TASK_PATH = "docs/handoff/NEXT_RX_TASK.md" @@ -81,6 +93,10 @@ class WatcherState: deferred_task_id: str | None = None running: bool = False pid: int | None = None + channel_test_ready_digest: str | None = None + channel_test_complete_digest: str | None = None + channel_test_ready_ack_commit: str | None = None + channel_test_complete_ack_commit: str | None = None @dataclass(frozen=True) @@ -387,6 +403,70 @@ class GitTaskSource: return text, head +def confirm_main_channel_test( + repository: Path, + state: WatcherState, + *, + remote: str = "origin", + result_branch: str = "rx-node", + acknowledgement_branch: str = "tx-node", + output_function: Callable[[str], None] = print, +) -> WatcherState: + """Confirm that MAIN acknowledged both exact channel-test events.""" + + rx_channel = GitResultChannel( + repository, + remote=remote, + branch=result_branch, + ) + rx_channel.fetch() + ready_result, ready_digest, _ready_commit = rx_channel.read_commit( + CHANNEL_TEST_READY_COMMIT + ) + complete_result, complete_digest, _complete_commit = rx_channel.read_commit( + CHANNEL_TEST_COMPLETE_COMMIT + ) + if ready_digest != CHANNEL_TEST_READY_DIGEST: + raise ValueError("historical channel-test READY digest is wrong") + if complete_digest != CHANNEL_TEST_COMPLETE_DIGEST: + raise ValueError("historical channel-test COMPLETE digest is wrong") + if ready_result.source_head != complete_result.source_head: + raise ValueError("channel-test source HEAD changed between READY and COMPLETE") + + acknowledgement_channel = GitAcknowledgementChannel( + repository, + remote=remote, + branch=acknowledgement_branch, + ) + confirmations = channel_test_acknowledgements( + acknowledgement_channel, + expected_source_head=ready_result.source_head, + ready_source_commit=CHANNEL_TEST_READY_COMMIT, + ready_source_digest=CHANNEL_TEST_READY_DIGEST, + complete_source_commit=CHANNEL_TEST_COMPLETE_COMMIT, + complete_source_digest=CHANNEL_TEST_COMPLETE_DIGEST, + ) + if not confirmations.ready_confirmed: + raise ValueError("MAIN READY acknowledgement is not confirmed") + if not confirmations.complete_confirmed: + raise ValueError("MAIN COMPLETE acknowledgement is not confirmed") + + state.channel_test_ready_digest = CHANNEL_TEST_READY_DIGEST + state.channel_test_complete_digest = CHANNEL_TEST_COMPLETE_DIGEST + state.channel_test_ready_ack_commit = confirmations.ready_ack_commit + state.channel_test_complete_ack_commit = confirmations.complete_ack_commit + output_function("MAIN READY RECEIVED = CONFIRMED") + output_function("MAIN COMPLETE RECEIVED = CONFIRMED") + output_function( + f"MAIN READY ACK COMMIT = {confirmations.ready_ack_commit}" + ) + output_function( + f"MAIN COMPLETE ACK COMMIT = {confirmations.complete_ack_commit}" + ) + output_function("RF TX = 0") + return state + + def handle_task( task_text: str, source_head: str, @@ -588,6 +668,16 @@ def run_watcher(arguments: argparse.Namespace) -> int: else: source.fetch() task_text, source_head = source.read() + task = parse_task(task_text) + if task.task_id == EXPECTED_RX_TASK_ID and not arguments.dry_run: + confirm_main_channel_test( + arguments.repository.resolve(), + state, + remote=arguments.remote, + result_branch=arguments.result_branch, + acknowledgement_branch=arguments.acknowledgement_branch, + ) + save_state(state_path, state) status, state = handle_task( task_text, source_head, @@ -637,6 +727,7 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument("--mark-completed") parser.add_argument("--publish-result", type=Path) parser.add_argument("--result-branch", default="rx-node") + parser.add_argument("--acknowledgement-branch", default="tx-node") return parser diff --git a/tools/tx_agent_watcher.py b/tools/tx_agent_watcher.py index c0941f8..ec9558e 100644 --- a/tools/tx_agent_watcher.py +++ b/tools/tx_agent_watcher.py @@ -3,7 +3,7 @@ from __future__ import annotations import argparse -from dataclasses import asdict, dataclass +from dataclasses import asdict, dataclass, replace from datetime import datetime, timezone import json import os @@ -33,6 +33,14 @@ from tools.rf_handoff_channel import ( # noqa: E402 load_result, result_digest, ) +from tools.tx_handoff_channel import ( # noqa: E402 + CHANNEL_COMPLETE_OBSERVED, + CHANNEL_READY_OBSERVED, + RX_READY_OBSERVED, + TX_AUTHORIZATION_CREATED, + GitAcknowledgementChannel, + TxHandoffAcknowledgement, +) DEFAULT_STATE_PATH = ROOT / ".local" / "tx_agent_watcher_state.json" @@ -67,6 +75,13 @@ class TxExecutor(Protocol): ) -> Path: ... +class AcknowledgementPublisher(Protocol): + def publish( + self, + acknowledgement: TxHandoffAcknowledgement, + ) -> object: ... + + @dataclass class TxWatcherState: state_version: int = STATE_VERSION @@ -410,6 +425,45 @@ def migrate_state_cursor( return True +def acknowledgement_for_result( + result: RxHandoffResult, + digest: str, + result_head: str, + status: str, + *, + observed_at: datetime, + main_watcher_pid: int, +) -> TxHandoffAcknowledgement | None: + """Build an acknowledgement only for an event handled successfully.""" + + acknowledgement_state: str | None = None + rf_tx_authorized = False + if status == "CHANNEL_TEST_READY": + acknowledgement_state = CHANNEL_READY_OBSERVED + elif status == "CHANNEL_TEST_COMPLETE": + acknowledgement_state = CHANNEL_COMPLETE_OBSERVED + elif status == "RED_APPROVAL_BLOCKED": + acknowledgement_state = RX_READY_OBSERVED + elif status == "TX_AUTHORIZATION_PREPARED": + acknowledgement_state = TX_AUTHORIZATION_CREATED + rf_tx_authorized = True + if acknowledgement_state is None: + return None + return TxHandoffAcknowledgement( + schema_version=1, + task_id=result.task_id, + state=acknowledgement_state, + source_rx_commit=result_head, + source_rx_digest=digest, + source_head=result.source_head, + observed_at=observed_at.astimezone(timezone.utc).isoformat(), + main_watcher_pid=main_watcher_pid, + rf_tx_authorized=rf_tx_authorized, + rf_tx_started=False, + rf_tx=0, + ) + + def process_history_cycle( source: ResultSource, state: TxWatcherState, @@ -423,6 +477,8 @@ def process_history_cycle( expected_source_head: str | None = None, max_age_seconds: float = 30.0, min_remaining_seconds: float = 10.0, + acknowledgement_publisher: AcknowledgementPublisher | None = None, + main_watcher_pid: int | None = None, ) -> list[tuple[str, str]]: """Process every new result commit in chronological order.""" @@ -432,11 +488,12 @@ def process_history_cycle( outcomes: list[tuple[str, str]] = [] for commit in commits: result, digest, result_head = source.read_commit(commit) - status, state = handle_result( + candidate_state = replace(state) + status, candidate_state = handle_result( result, digest, result_head, - state, + candidate_state, source, executor, now_function=now_function, @@ -447,6 +504,26 @@ def process_history_cycle( max_age_seconds=max_age_seconds, min_remaining_seconds=min_remaining_seconds, ) + acknowledgement = acknowledgement_for_result( + result, + digest, + result_head, + status, + observed_at=now_function(), + main_watcher_pid=main_watcher_pid or os.getpid(), + ) + if acknowledgement is not None and acknowledgement_publisher is not None: + publication = acknowledgement_publisher.publish(acknowledgement) + output_function( + "TX ACKNOWLEDGEMENT PUBLISH = " + f"{getattr(publication, 'status', 'PUBLISHED')}" + ) + output_function( + "TX ACKNOWLEDGEMENT HEAD = " + f"{getattr(publication, 'commit', 'UNKNOWN')}" + ) + for field_name in TxWatcherState.__dataclass_fields__: + setattr(state, field_name, getattr(candidate_state, field_name)) state.last_seen_commit = result_head state.last_processed_rx_node_commit = result_head state.state_version = STATE_VERSION @@ -481,6 +558,7 @@ def run_watcher(arguments: argparse.Namespace) -> int: stop_path.unlink(missing_ok=True) executor = ManualTxExecutor(arguments.outbox.resolve()) + acknowledgement_publisher: AcknowledgementPublisher | None = None def input_function(prompt: str) -> str: if arguments.approval == "authorize": @@ -511,6 +589,11 @@ def run_watcher(arguments: argparse.Namespace) -> int: remote=arguments.remote, branch=arguments.result_branch, ) + acknowledgement_publisher = GitAcknowledgementChannel( + repository, + remote=arguments.remote, + branch=arguments.acknowledgement_branch, + ) if not _HEAD_RE.fullmatch(expected_source_head): raise ValueError( "expected source HEAD must be a full lowercase Git hash" @@ -534,6 +617,8 @@ def run_watcher(arguments: argparse.Namespace) -> int: expected_source_head=expected_source_head, max_age_seconds=arguments.max_age, min_remaining_seconds=arguments.min_remaining, + acknowledgement_publisher=acknowledgement_publisher, + main_watcher_pid=os.getpid(), ) if arguments.once or arguments.dry_run: break @@ -569,6 +654,7 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument("--remote", default="origin") parser.add_argument("--source-branch", default="video-development") parser.add_argument("--result-branch", default="rx-node") + parser.add_argument("--acknowledgement-branch", default="tx-node") parser.add_argument("--expected-source-head") parser.add_argument("--poll-interval", type=float, default=5.0) parser.add_argument("--max-age", type=float, default=30.0) diff --git a/tools/tx_handoff_channel.py b/tools/tx_handoff_channel.py new file mode 100644 index 0000000..2996ce3 --- /dev/null +++ b/tools/tx_handoff_channel.py @@ -0,0 +1,594 @@ +"""Strict MAIN-to-RX acknowledgement channel stored on ``tx-node``.""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass, fields +from datetime import datetime, timezone +import hashlib +import json +from pathlib import Path +import re +import tempfile +from typing import Any, Mapping + +from tools.rf_handoff_channel import ( + CHANNEL_TEST_TASK_ID, + MAX_RESULT_BYTES, + PublishOutcome, + ResultBranchMissing, + ResultChannelError, + ResultHistoryRewritten, + GitResultChannel, +) + + +TX_ACK_SCHEMA_VERSION = 1 +TX_RESULT_JSON_PATH = "docs/handoff/results/TX_RESULT.json" +TX_RESULT_MARKDOWN_PATH = "docs/handoff/results/TX_RESULT.md" +TX_RESULT_PATHS = frozenset((TX_RESULT_JSON_PATH, TX_RESULT_MARKDOWN_PATH)) +TX_NODE_BRANCH = "tx-node" + +CHANNEL_READY_OBSERVED = "CHANNEL_READY_OBSERVED" +CHANNEL_COMPLETE_OBSERVED = "CHANNEL_COMPLETE_OBSERVED" +RX_READY_OBSERVED = "RX_READY_OBSERVED" +TX_AUTHORIZATION_CREATED = "TX_AUTHORIZATION_CREATED" +TX_STARTED = "TX_STARTED" +TX_COMPLETE = "TX_COMPLETE" +TX_ACK_STATES = frozenset( + ( + CHANNEL_READY_OBSERVED, + CHANNEL_COMPLETE_OBSERVED, + RX_READY_OBSERVED, + TX_AUTHORIZATION_CREATED, + TX_STARTED, + TX_COMPLETE, + ) +) + +CHANNEL_TEST_READY_COMMIT = "4cb6c8cad1ab5194755c2740c2a5a83933d099d8" +CHANNEL_TEST_READY_DIGEST = ( + "23fc54f1f788f068fd4a842daf0a445bd4ec192866b4d89406c74f1797a9952b" +) +CHANNEL_TEST_COMPLETE_COMMIT = "83d91ff5309777ab639a24f774e9d0b8825a11e9" +CHANNEL_TEST_COMPLETE_DIGEST = ( + "7d98ed2e70e98a087b968c0d65fb1db4277b81f251e075d20583de2e5002b2d6" +) + +_HEAD_RE = re.compile(r"^[0-9a-f]{40}$") +_SHA256_RE = re.compile(r"^[0-9a-f]{64}$") +_TASK_ID_RE = re.compile(r"^[A-Z0-9][A-Z0-9_-]{2,79}$") + + +@dataclass(frozen=True) +class TxHandoffAcknowledgement: + """One strict acknowledgement of an exact event from ``rx-node``.""" + + schema_version: int + task_id: str + state: str + source_rx_commit: str + source_rx_digest: str + source_head: str + observed_at: str + main_watcher_pid: int + rf_tx_authorized: bool + rf_tx_started: bool + rf_tx: int + + +@dataclass(frozen=True) +class ChannelTestAcknowledgements: + ready_confirmed: bool + complete_confirmed: bool + ready_ack_commit: str | None + complete_ack_commit: str | None + + +def _parse_utc(value: str) -> datetime: + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except (TypeError, ValueError) as error: + raise ValueError("observed_at must be an ISO-8601 timestamp") from error + if parsed.tzinfo is None: + raise ValueError("observed_at must contain a timezone") + return parsed.astimezone(timezone.utc) + + +def validate_acknowledgement( + acknowledgement: TxHandoffAcknowledgement, +) -> TxHandoffAcknowledgement: + if type(acknowledgement.schema_version) is not int: + raise ValueError("schema_version must contain an integer") + if acknowledgement.schema_version != TX_ACK_SCHEMA_VERSION: + raise ValueError(f"schema_version must be {TX_ACK_SCHEMA_VERSION}") + string_values = ( + acknowledgement.task_id, + acknowledgement.state, + acknowledgement.source_rx_commit, + acknowledgement.source_rx_digest, + acknowledgement.source_head, + acknowledgement.observed_at, + ) + if any(type(value) is not str for value in string_values): + raise ValueError("acknowledgement string fields must contain strings") + if not _TASK_ID_RE.fullmatch(acknowledgement.task_id): + raise ValueError("task_id is invalid") + if acknowledgement.state not in TX_ACK_STATES: + raise ValueError("state is not supported") + if not _HEAD_RE.fullmatch(acknowledgement.source_rx_commit): + raise ValueError("source_rx_commit must be a lowercase 40-character hash") + if not _SHA256_RE.fullmatch(acknowledgement.source_rx_digest): + raise ValueError("source_rx_digest must be a lowercase SHA256") + if not _HEAD_RE.fullmatch(acknowledgement.source_head): + raise ValueError("source_head must be a lowercase 40-character hash") + _parse_utc(acknowledgement.observed_at) + if ( + type(acknowledgement.main_watcher_pid) is not int + or acknowledgement.main_watcher_pid <= 0 + ): + raise ValueError("main_watcher_pid must be a positive integer") + if type(acknowledgement.rf_tx_authorized) is not bool: + raise ValueError("rf_tx_authorized must contain a boolean") + if type(acknowledgement.rf_tx_started) is not bool: + raise ValueError("rf_tx_started must contain a boolean") + if type(acknowledgement.rf_tx) is not int or acknowledgement.rf_tx not in {0, 1}: + raise ValueError("rf_tx must be 0 or 1") + + if ( + acknowledgement.state + in { + CHANNEL_READY_OBSERVED, + CHANNEL_COMPLETE_OBSERVED, + RX_READY_OBSERVED, + } + and ( + acknowledgement.rf_tx_authorized + or acknowledgement.rf_tx_started + or acknowledgement.rf_tx != 0 + ) + ): + raise ValueError("observation states cannot claim RF TX authorization") + if acknowledgement.state in { + CHANNEL_READY_OBSERVED, + CHANNEL_COMPLETE_OBSERVED, + } and acknowledgement.task_id != CHANNEL_TEST_TASK_ID: + raise ValueError("channel acknowledgement has the wrong task_id") + if ( + acknowledgement.state == TX_AUTHORIZATION_CREATED + and ( + not acknowledgement.rf_tx_authorized + or acknowledgement.rf_tx_started + or acknowledgement.rf_tx != 0 + ) + ): + raise ValueError("TX authorization state is inconsistent") + if acknowledgement.state in {TX_STARTED, TX_COMPLETE} and not ( + acknowledgement.rf_tx_authorized + and acknowledgement.rf_tx_started + and acknowledgement.rf_tx == 1 + ): + raise ValueError("TX execution state is inconsistent") + return acknowledgement + + +def acknowledgement_from_mapping( + raw: Mapping[str, Any], +) -> TxHandoffAcknowledgement: + expected_fields = {field.name for field in fields(TxHandoffAcknowledgement)} + actual_fields = set(raw) + if actual_fields != expected_fields: + missing = sorted(expected_fields - actual_fields) + extra = sorted(actual_fields - expected_fields) + raise ValueError( + f"acknowledgement fields disagree: missing={missing}, extra={extra}" + ) + acknowledgement = TxHandoffAcknowledgement( + **{name: raw[name] for name in expected_fields} + ) + return validate_acknowledgement(acknowledgement) + + +def canonical_acknowledgement_json( + acknowledgement: TxHandoffAcknowledgement, +) -> str: + validate_acknowledgement(acknowledgement) + return json.dumps( + asdict(acknowledgement), + indent=2, + sort_keys=True, + ensure_ascii=False, + ) + "\n" + + +def acknowledgement_digest(acknowledgement: TxHandoffAcknowledgement) -> str: + return hashlib.sha256( + canonical_acknowledgement_json(acknowledgement).encode("utf-8") + ).hexdigest() + + +def render_acknowledgement_markdown( + acknowledgement: TxHandoffAcknowledgement, +) -> str: + validate_acknowledgement(acknowledgement) + return "\n".join( + ( + "# MAIN acknowledgement", + "", + f"SCHEMA_VERSION = {acknowledgement.schema_version}", + f"TASK_ID = {acknowledgement.task_id}", + f"STATE = {acknowledgement.state}", + f"SOURCE_RX_COMMIT = {acknowledgement.source_rx_commit}", + f"SOURCE_RX_DIGEST = {acknowledgement.source_rx_digest}", + f"SOURCE_HEAD = {acknowledgement.source_head}", + f"OBSERVED_AT = {acknowledgement.observed_at}", + f"MAIN_WATCHER_PID = {acknowledgement.main_watcher_pid}", + "RF_TX_AUTHORIZED = " + f"{'YES' if acknowledgement.rf_tx_authorized else 'NO'}", + f"RF_TX_STARTED = {'YES' if acknowledgement.rf_tx_started else 'NO'}", + f"RF_TX = {acknowledgement.rf_tx}", + "", + "Raw IQ, video, waveforms, source code, local state and large logs " + "are not stored in this branch.", + "", + ) + ) + + +class GitAcknowledgementChannel(GitResultChannel): + """Read and publish strict acknowledgements on an independent Git branch.""" + + def __init__( + self, + repository: Path, + *, + remote: str = "origin", + branch: str = TX_NODE_BRANCH, + ) -> None: + super().__init__(repository, remote=remote, branch=branch) + + def read(self) -> tuple[TxHandoffAcknowledgement, str, str]: + remote_head = self.fetch() + return self.read_commit(remote_head) + + def read_commit( + self, + commit: str, + ) -> tuple[TxHandoffAcknowledgement, str, str]: + if not _HEAD_RE.fullmatch(commit): + raise ResultChannelError( + "acknowledgement commit must be a full lowercase Git hash" + ) + commit_check = self._run( + ["cat-file", "-e", f"{commit}^{{commit}}"], + check=False, + ) + if commit_check.returncode != 0: + raise ResultChannelError( + f"acknowledgement commit is unavailable: {commit}" + ) + tree_text = self._run( + ["ls-tree", "-r", "--name-only", commit] + ).stdout + tree_paths = { + line.strip().replace("\\", "/") + for line in tree_text.splitlines() + if line.strip() + } + if tree_paths != TX_RESULT_PATHS: + raise ResultChannelError( + f"tx-node tree contains unexpected paths: {sorted(tree_paths)}" + ) + for path in TX_RESULT_PATHS: + size_text = self._run(["cat-file", "-s", f"{commit}:{path}"]).stdout + if int(size_text.strip()) > MAX_RESULT_BYTES: + raise ResultChannelError(f"acknowledgement file exceeds limit: {path}") + json_result = self._run( + ["show", f"{commit}:{TX_RESULT_JSON_PATH}"], + check=False, + ) + if json_result.returncode != 0: + raise ResultChannelError( + f"TX_RESULT.json is unavailable at commit {commit}: " + f"{json_result.stderr.strip()}" + ) + raw = json.loads(json_result.stdout) + if not isinstance(raw, dict): + raise ResultChannelError( + f"TX_RESULT.json root is not an object at commit {commit}" + ) + acknowledgement = acknowledgement_from_mapping(raw) + return acknowledgement, acknowledgement_digest(acknowledgement), commit + + def commits_since( + self, + last_processed_commit: str | None, + ) -> tuple[list[str], str]: + remote_head = self.fetch() + if last_processed_commit is not None: + if not _HEAD_RE.fullmatch(last_processed_commit): + raise ResultChannelError( + "last processed acknowledgement commit is invalid" + ) + if self._optional_revision(last_processed_commit) is None: + raise ResultHistoryRewritten( + "last processed acknowledgement commit is unavailable" + ) + if not self._is_ancestor(last_processed_commit, remote_head): + raise ResultHistoryRewritten( + "tx-node no longer descends from the last processed commit" + ) + if last_processed_commit == remote_head: + return [], remote_head + history = self._run( + [ + "rev-list", + "--reverse", + "--topo-order", + f"{last_processed_commit}..{remote_head}", + ] + ).stdout + return ( + [line.strip() for line in history.splitlines() if line.strip()], + remote_head, + ) + + history = self._run( + ["rev-list", "--reverse", "--topo-order", remote_head] + ).stdout + all_commits = [line.strip() for line in history.splitlines() if line.strip()] + first_result_index: int | None = None + for index, commit in enumerate(all_commits): + exists = self._run( + ["cat-file", "-e", f"{commit}:{TX_RESULT_JSON_PATH}"], + check=False, + ) + if exists.returncode == 0: + first_result_index = index + break + if first_result_index is None: + return [], remote_head + return all_commits[first_result_index:], remote_head + + def _matching_commit( + self, + acknowledgement: TxHandoffAcknowledgement, + ) -> str | None: + commits, _remote_head = self.commits_since(None) + for commit in reversed(commits): + observed, _digest, _head = self.read_commit(commit) + if ( + observed.task_id == acknowledgement.task_id + and observed.state == acknowledgement.state + and observed.source_rx_commit + == acknowledgement.source_rx_commit + and observed.source_rx_digest + == acknowledgement.source_rx_digest + and observed.source_head == acknowledgement.source_head + and observed.rf_tx_authorized + == acknowledgement.rf_tx_authorized + and observed.rf_tx_started == acknowledgement.rf_tx_started + and observed.rf_tx == acknowledgement.rf_tx + ): + return commit + return None + + def publish( + self, + acknowledgement: TxHandoffAcknowledgement, + ) -> PublishOutcome: + validate_acknowledgement(acknowledgement) + remote_before = self.fetch() + matching_commit = self._matching_commit(acknowledgement) + if matching_commit is not None: + return PublishOutcome( + "UNCHANGED", + matching_commit, + remote_before, + json_path=TX_RESULT_JSON_PATH, + markdown_path=TX_RESULT_MARKDOWN_PATH, + ) + + tracking_reference = f"refs/remotes/{self.remote}/{self.branch}" + with tempfile.TemporaryDirectory(prefix="tx-ack-worktree-") as directory: + checkout = Path(directory) / "tx-node" + worktree_added = False + try: + self._run( + ["worktree", "add", "--detach", str(checkout), tracking_reference] + ) + worktree_added = True + existing_text = self._run( + ["ls-tree", "-r", "--name-only", "HEAD"], + cwd=checkout, + ).stdout + existing_paths = { + line.strip().replace("\\", "/") + for line in existing_text.splitlines() + if line.strip() + } + if not existing_paths.issubset(TX_RESULT_PATHS): + raise ResultChannelError( + "tx-node contains paths outside the acknowledgement allowlist" + ) + + json_path = checkout / TX_RESULT_JSON_PATH + markdown_path = checkout / TX_RESULT_MARKDOWN_PATH + json_path.parent.mkdir(parents=True, exist_ok=True) + json_text = canonical_acknowledgement_json(acknowledgement) + markdown_text = render_acknowledgement_markdown(acknowledgement) + if len(json_text.encode("utf-8")) > MAX_RESULT_BYTES: + raise ResultChannelError("TX_RESULT.json exceeds the size limit") + if len(markdown_text.encode("utf-8")) > MAX_RESULT_BYTES: + raise ResultChannelError("TX_RESULT.md exceeds the size limit") + json_path.write_text(json_text, encoding="utf-8", newline="\n") + markdown_path.write_text( + markdown_text, + encoding="utf-8", + newline="\n", + ) + self._run(["add", "--", *sorted(TX_RESULT_PATHS)], cwd=checkout) + staged_text = self._run( + ["diff", "--cached", "--name-only", "--"], + cwd=checkout, + ).stdout + staged = { + line.strip().replace("\\", "/") + for line in staged_text.splitlines() + if line.strip() + } + if staged != TX_RESULT_PATHS: + raise ResultChannelError( + f"refusing unexpected staged paths: {sorted(staged)}" + ) + self._run( + [ + "-c", + "user.name=MAIN TX Watcher", + "-c", + "user.email=tx-node@local.invalid", + "commit", + "--quiet", + "-m", + f"Acknowledge {acknowledgement.task_id} " + f"{acknowledgement.state}", + ], + cwd=checkout, + ) + commit = self._run( + ["rev-parse", "HEAD"], + cwd=checkout, + ).stdout.strip() + committed_text = self._run( + ["ls-tree", "-r", "--name-only", commit], + cwd=checkout, + ).stdout + committed_paths = { + line.strip().replace("\\", "/") + for line in committed_text.splitlines() + if line.strip() + } + if committed_paths != TX_RESULT_PATHS: + raise ResultChannelError( + "committed tx-node tree violates the path allowlist" + ) + push = self._run( + [ + "push", + "--quiet", + self.remote, + f"HEAD:refs/heads/{self.branch}", + ], + cwd=checkout, + check=False, + ) + if push.returncode != 0: + raise ResultChannelError( + f"acknowledgement push failed: {push.stderr.strip()}" + ) + remote_after = self.remote_head() + if remote_after != commit: + raise ResultChannelError( + "remote tx-node HEAD does not confirm the published commit" + ) + return PublishOutcome( + "PUBLISHED", + commit, + commit, + json_path=TX_RESULT_JSON_PATH, + markdown_path=TX_RESULT_MARKDOWN_PATH, + ) + finally: + if worktree_added: + removal = self._run( + ["worktree", "remove", "--force", str(checkout)], + check=False, + ) + if removal.returncode != 0: + raise ResultChannelError( + "temporary acknowledgement worktree cleanup failed: " + f"{removal.stderr.strip()}" + ) + self._run(["worktree", "prune"], check=False) + + +def channel_test_acknowledgements( + channel: GitAcknowledgementChannel, + *, + expected_source_head: str, + ready_source_commit: str = CHANNEL_TEST_READY_COMMIT, + ready_source_digest: str = CHANNEL_TEST_READY_DIGEST, + complete_source_commit: str = CHANNEL_TEST_COMPLETE_COMMIT, + complete_source_digest: str = CHANNEL_TEST_COMPLETE_DIGEST, +) -> ChannelTestAcknowledgements: + """Find both exact channel-test acknowledgements in chronological history.""" + + if not _HEAD_RE.fullmatch(expected_source_head): + raise ValueError("expected_source_head must be a full lowercase Git hash") + if not _HEAD_RE.fullmatch(ready_source_commit): + raise ValueError("ready_source_commit must be a full lowercase Git hash") + if not _SHA256_RE.fullmatch(ready_source_digest): + raise ValueError("ready_source_digest must be a lowercase SHA256") + if not _HEAD_RE.fullmatch(complete_source_commit): + raise ValueError("complete_source_commit must be a full lowercase Git hash") + if not _SHA256_RE.fullmatch(complete_source_digest): + raise ValueError("complete_source_digest must be a lowercase SHA256") + commits, _remote_head = channel.commits_since(None) + ready_commit: str | None = None + complete_commit: str | None = None + for commit in commits: + acknowledgement, _digest, _head = channel.read_commit(commit) + common = ( + acknowledgement.task_id == CHANNEL_TEST_TASK_ID + and acknowledgement.source_head == expected_source_head + and not acknowledgement.rf_tx_authorized + and not acknowledgement.rf_tx_started + and acknowledgement.rf_tx == 0 + ) + if ( + common + and acknowledgement.state == CHANNEL_READY_OBSERVED + and acknowledgement.source_rx_commit == ready_source_commit + and acknowledgement.source_rx_digest == ready_source_digest + ): + ready_commit = commit + elif ( + common + and ready_commit is not None + and acknowledgement.state == CHANNEL_COMPLETE_OBSERVED + and acknowledgement.source_rx_commit == complete_source_commit + and acknowledgement.source_rx_digest == complete_source_digest + ): + complete_commit = commit + return ChannelTestAcknowledgements( + ready_confirmed=ready_commit is not None, + complete_confirmed=complete_commit is not None, + ready_ack_commit=ready_commit, + complete_ack_commit=complete_commit, + ) + + +__all__ = [ + "CHANNEL_COMPLETE_OBSERVED", + "CHANNEL_READY_OBSERVED", + "CHANNEL_TEST_COMPLETE_COMMIT", + "CHANNEL_TEST_COMPLETE_DIGEST", + "CHANNEL_TEST_READY_COMMIT", + "CHANNEL_TEST_READY_DIGEST", + "ChannelTestAcknowledgements", + "GitAcknowledgementChannel", + "ResultBranchMissing", + "ResultChannelError", + "ResultHistoryRewritten", + "TX_ACK_SCHEMA_VERSION", + "TX_NODE_BRANCH", + "TX_RESULT_JSON_PATH", + "TX_RESULT_MARKDOWN_PATH", + "TX_RESULT_PATHS", + "TxHandoffAcknowledgement", + "acknowledgement_digest", + "acknowledgement_from_mapping", + "canonical_acknowledgement_json", + "channel_test_acknowledgements", + "render_acknowledgement_markdown", + "validate_acknowledgement", +]