From a9317989a25abd7e5b620fb66510bd2a7bc26b87 Mon Sep 17 00:00:00 2001 From: LittleSam129 <1qwe432@gmail.com> Date: Tue, 22 Sep 2026 11:46:53 +0300 Subject: [PATCH] Process all RX handoff commits in TX watcher --- docs/handoff/NEXT_TX_TASK.md | 2 +- docs/handoff/RX_AGENT_PROTOCOL.md | 6 +- docs/handoff/TX_AGENT_USAGE.md | 14 +- tests/test_rf_handoff_coordination.py | 254 ++++++++++++++++++++++++++ tools/rf_handoff_channel.py | 141 +++++++++++++- tools/tx_agent_watcher.py | 170 ++++++++++++++--- 6 files changed, 544 insertions(+), 43 deletions(-) diff --git a/docs/handoff/NEXT_TX_TASK.md b/docs/handoff/NEXT_TX_TASK.md index 6c6959e..5986cb9 100644 --- a/docs/handoff/NEXT_TX_TASK.md +++ b/docs/handoff/NEXT_TX_TASK.md @@ -26,7 +26,7 @@ Start the local model-free watcher before the RX task begins: tools\start_tx_watcher.cmd -The watcher polls origin/rx-node every 5 seconds and accepts only a strict result for: +The watcher polls origin/rx-node every 5 seconds. It processes every new result-branch commit from `last_processed_rx_node_commit` to the current `origin/rx-node` tip in chronological order. It accepts only a strict result for: RX TASK_ID = RF001A_OTA_CAPTURE_003 STATE = RX_READY_FOR_TX diff --git a/docs/handoff/RX_AGENT_PROTOCOL.md b/docs/handoff/RX_AGENT_PROTOCOL.md index b1c8802..3d61236 100644 --- a/docs/handoff/RX_AGENT_PROTOCOL.md +++ b/docs/handoff/RX_AGENT_PROTOCOL.md @@ -116,11 +116,13 @@ Raw IQ, video, waveform files, large logs, secrets and machine-local state remai ## MAIN PC watcher -tools/tx_agent_watcher.py polls origin/rx-node every 5 seconds by default. It does not call an LLM or open SDR hardware. +tools/tx_agent_watcher.py polls origin/rx-node every 5 seconds by default. It does not call an LLM or open SDR hardware. Its versioned local state stores `last_processed_rx_node_commit`. Each poll processes every commit in `last_processed_rx_node_commit..origin/rx-node` from oldest to newest and reads the result from that exact commit instead of reading only the branch tip. For RF_HANDOFF_CHANNEL_TEST_001 the watcher records synthetic READY and COMPLETE states without requesting RED approval. For a new valid production READY result it checks task ID, source HEAD, current result branch commit, state, live-process flags, common clock, publication age, planned duration and remaining capture time. It then requests the exact phrase AUTHORIZE RF TX. -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. +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. 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. diff --git a/docs/handoff/TX_AGENT_USAGE.md b/docs/handoff/TX_AGENT_USAGE.md index 5a91313..273ef7f 100644 --- a/docs/handoff/TX_AGENT_USAGE.md +++ b/docs/handoff/TX_AGENT_USAGE.md @@ -12,6 +12,18 @@ From the repository root, double-click: The visible console polls every 5 seconds. A missing rx-node branch or invalid result is reported. No model runs while polling. +## Result history cursor + +The watcher stores `last_processed_rx_node_commit` in `.local\tx_agent_watcher_state.json`. After each fetch it enumerates every commit in: + + last_processed_rx_node_commit..origin/rx-node + +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. + +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. + ## 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. @@ -22,7 +34,7 @@ For a fresh valid RF001A_OTA_CAPTURE_003 READY result, the watcher checks the ex Any other input blocks authorization. -After the exact phrase, the watcher fetches rx-node again. It requires the same result digest and commit, the expected source HEAD, a fresh publication, an active 120-second capture window, verified common clock and at least 10 seconds remaining. +After the exact phrase, the watcher fetches rx-node again. It requires the same result digest and commit, the expected source HEAD, a fresh publication, an active 120-second capture window, verified common clock and at least 10 seconds remaining. An expired historical READY is recorded as observed but cannot create an authorization. On success it writes: diff --git a/tests/test_rf_handoff_coordination.py b/tests/test_rf_handoff_coordination.py index 6f415f2..48b2709 100644 --- a/tests/test_rf_handoff_coordination.py +++ b/tests/test_rf_handoff_coordination.py @@ -20,6 +20,7 @@ from tools.rf_handoff_channel import ( RESULT_MARKDOWN_PATH, ResultBranchMissing, ResultChannelError, + ResultHistoryRewritten, RxHandoffResult, canonical_result_json, evaluate_ready, @@ -36,6 +37,8 @@ from tools.tx_agent_watcher import ( TxWatcherState, handle_result, load_state as load_tx_state, + migrate_state_cursor, + process_history_cycle, request_stop, save_state as save_tx_state, ) @@ -140,6 +143,44 @@ def git_fixture(tmp_path: Path) -> tuple[Path, Path]: return remote, work +def commit_on_result_branch( + work: Path, + temporary_root: Path, + message: str, + *, + json_text: str | None = None, + allow_empty: bool = False, +) -> str: + """Create one test-only result branch commit from the main repository.""" + + channel = GitResultChannel(work) + channel.fetch() + checkout = temporary_root / f"result-{message.replace(' ', '-')}" + _git(["worktree", "add", "--detach", str(checkout), "origin/rx-node"], work) + try: + if json_text is not None: + result_path = checkout / RESULT_JSON_PATH + result_path.parent.mkdir(parents=True, exist_ok=True) + result_path.write_text(json_text, encoding="utf-8") + _git(["add", "--", RESULT_JSON_PATH], checkout) + commit_arguments = [ + "-c", + "user.name=Test", + "-c", + "user.email=test@example.invalid", + "commit", + ] + if allow_empty: + commit_arguments.append("--allow-empty") + commit_arguments.extend(["-m", message]) + _git(commit_arguments, checkout) + commit = _git(["rev-parse", "HEAD"], checkout) + _git(["push", "origin", "HEAD:refs/heads/rx-node"], checkout) + return commit + finally: + _git(["worktree", "remove", "--force", str(checkout)], work) + + class FakeSource: def __init__(self, result: RxHandoffResult, digest: str, head: str = HEAD) -> None: self.result = result @@ -154,6 +195,19 @@ class FakeSource: def read(self) -> tuple[RxHandoffResult, str, str]: return self.result, self.digest, self.head + def read_commit(self, commit: str) -> tuple[RxHandoffResult, str, str]: + assert commit == self.head + return self.read() + + def commits_since( + self, + last_processed_commit: str | None, + ) -> tuple[list[str], str]: + return ([] if last_processed_commit == self.head else [self.head], self.head) + + def find_commit_by_digest(self, digest: str) -> str | None: + return self.head if digest == self.digest else None + class FakeExecutor: def __init__(self, path: Path) -> None: @@ -372,6 +426,206 @@ def test_stop_request_clears_stale_watcher_pid(tmp_path: Path) -> None: assert not state_path.with_suffix(".json.stop").exists() +def test_version_one_state_migrates_without_losing_safety_fields( + tmp_path: Path, +) -> None: + state_path = tmp_path / "legacy_state.json" + state_path.write_text( + """{ + "running": false, + "pid": null, + "last_seen_digest": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "last_prompted_digest": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + "last_authorized_digest": null, + "channel_test_ready_digest": null, + "channel_test_complete_digest": "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc" +} +""", + encoding="utf-8", + ) + + state = load_tx_state(state_path) + assert state.state_version == 2 + assert state.last_seen_commit is None + assert state.last_processed_rx_node_commit is None + assert state.last_prompted_digest == "b" * 64 + assert state.channel_test_complete_digest == "c" * 64 + + +def test_history_cycle_processes_ready_duplicate_and_complete_in_order( + tmp_path: Path, +) -> None: + _remote, work = git_fixture(tmp_path) + channel = GitResultChannel(work) + baseline = complete_result( + channel_ready_result(), + duration_seconds=1.0, + file_size=4, + sha256="c" * 64, + ) + baseline_outcome = channel.publish(baseline) + state = TxWatcherState(last_seen_digest=result_digest(baseline)) + assert migrate_state_cursor(channel, state) + assert state.last_processed_rx_node_commit == baseline_outcome.commit + + ready = channel_ready_result(published_at=NOW.isoformat()) + ready_outcome = channel.publish(ready) + duplicate_commit = commit_on_result_branch( + work, + tmp_path, + "Duplicate READY result", + allow_empty=True, + ) + complete = complete_result( + ready, + published_at=(NOW + timedelta(seconds=1)).isoformat(), + duration_seconds=1.0, + file_size=4, + sha256="d" * 64, + ) + complete_outcome = channel.publish(complete) + prompts: list[str] = [] + executor = FakeExecutor(tmp_path / "forbidden.txt") + saved_commits: list[str | None] = [] + + outcomes = process_history_cycle( + channel, + state, + executor, + now_function=lambda: NOW + timedelta(seconds=2), + input_function=lambda prompt: prompts.append(prompt) or "AUTHORIZE RF TX", + save_function=lambda current: saved_commits.append( + current.last_processed_rx_node_commit + ), + output_function=lambda _text: None, + expected_source_head=HEAD, + ) + + assert outcomes == [ + (ready_outcome.commit, "CHANNEL_TEST_READY"), + (duplicate_commit, "DUPLICATE_SUPPRESSED"), + (complete_outcome.commit, "CHANNEL_TEST_COMPLETE"), + ] + assert saved_commits == [ + ready_outcome.commit, + duplicate_commit, + complete_outcome.commit, + ] + assert state.channel_test_ready_digest == result_digest(ready) + assert state.channel_test_complete_digest == result_digest(complete) + assert state.last_seen_commit == complete_outcome.commit + assert state.last_processed_rx_node_commit == complete_outcome.commit + assert prompts == [] + assert executor.calls == 0 + + +def test_malformed_intermediate_commit_blocks_later_history( + tmp_path: Path, +) -> None: + _remote, work = git_fixture(tmp_path) + channel = GitResultChannel(work) + baseline = complete_result( + channel_ready_result(), + duration_seconds=1.0, + file_size=4, + sha256="c" * 64, + ) + baseline_outcome = channel.publish(baseline) + state = TxWatcherState( + last_seen_digest=result_digest(baseline), + last_seen_commit=baseline_outcome.commit, + last_processed_rx_node_commit=baseline_outcome.commit, + ) + ready = channel_ready_result() + ready_outcome = channel.publish(ready) + malformed_commit = commit_on_result_branch( + work, + tmp_path, + "Malformed result", + json_text="{not-json\n", + ) + complete = complete_result( + ready, + duration_seconds=1.0, + file_size=4, + sha256="d" * 64, + ) + complete_outcome = channel.publish(complete) + saved_commits: list[str | None] = [] + + with pytest.raises(ValueError): + process_history_cycle( + channel, + state, + FakeExecutor(tmp_path / "forbidden.txt"), + now_function=lambda: NOW, + input_function=lambda _prompt: "DENY", + save_function=lambda current: saved_commits.append( + current.last_processed_rx_node_commit + ), + output_function=lambda _text: None, + expected_source_head=HEAD, + ) + + assert saved_commits == [ready_outcome.commit] + assert state.last_processed_rx_node_commit == ready_outcome.commit + assert state.last_processed_rx_node_commit != malformed_commit + assert state.last_processed_rx_node_commit != complete_outcome.commit + + +def test_result_history_rewrite_is_rejected(tmp_path: Path) -> None: + _remote, work = git_fixture(tmp_path) + channel = GitResultChannel(work) + ready_outcome = channel.publish(channel_ready_result()) + original_branch_head = _git(["rev-parse", "HEAD"], work) + _git( + [ + "push", + "--force", + "origin", + f"{original_branch_head}:refs/heads/rx-node", + ], + work, + ) + + with pytest.raises(ResultHistoryRewritten): + channel.commits_since(ready_outcome.commit) + + +def test_expired_historical_ready_is_observed_without_authorization( + tmp_path: Path, +) -> None: + _remote, work = git_fixture(tmp_path) + channel = GitResultChannel(work) + started = NOW - timedelta(seconds=200) + expired = ready_result( + capture_started_at=started.isoformat(), + expected_capture_end=(started + timedelta(seconds=120)).isoformat(), + published_at=(NOW - timedelta(seconds=100)).isoformat(), + ) + expired_outcome = channel.publish(expired) + state = TxWatcherState() + prompts: list[str] = [] + executor = FakeExecutor(tmp_path / "forbidden.txt") + + outcomes = process_history_cycle( + channel, + state, + executor, + now_function=lambda: NOW, + input_function=lambda prompt: prompts.append(prompt) or "AUTHORIZE RF TX", + save_function=lambda _state: None, + output_function=lambda _text: None, + dry_run=True, + expected_source_head=HEAD, + ) + + assert outcomes == [(expired_outcome.commit, "STALE_READY")] + assert state.last_processed_rx_node_commit == expired_outcome.commit + assert prompts == [] + assert executor.calls == 0 + + def test_real_origin_detection_and_worktree_publication(tmp_path: Path) -> None: remote, work = git_fixture(tmp_path) branch_before = _git(["branch", "--show-current"], work) diff --git a/tools/rf_handoff_channel.py b/tools/rf_handoff_channel.py index 49d2555..98989aa 100644 --- a/tools/rf_handoff_channel.py +++ b/tools/rf_handoff_channel.py @@ -35,6 +35,10 @@ class ResultBranchMissing(ResultChannelError): """The configured result branch does not exist on the remote.""" +class ResultHistoryRewritten(ResultChannelError): + """The result branch no longer descends from the processed commit.""" + + @dataclass(frozen=True) class RxHandoffResult: schema_version: int @@ -382,6 +386,32 @@ class GitResultChannel: def branch_exists(self) -> bool: return self.remote_head() is not None + def _tracking_reference(self) -> str: + return f"refs/remotes/{self.remote}/{self.branch}" + + def _optional_revision(self, revision: str) -> str | None: + result = self._run(["rev-parse", "--verify", revision], check=False) + if result.returncode != 0: + return None + value = result.stdout.strip() + if not _HEAD_RE.fullmatch(value): + raise ResultChannelError(f"invalid Git revision for {revision}") + return value + + def _is_ancestor(self, ancestor: str, descendant: str) -> bool: + result = self._run( + ["merge-base", "--is-ancestor", ancestor, descendant], + check=False, + ) + if result.returncode == 0: + return True + if result.returncode == 1: + return False + raise ResultChannelError( + f"cannot compare result history {ancestor}..{descendant}: " + f"{result.stderr.strip()}" + ) + def fetch(self) -> str: initial_fetch = self._run( ["fetch", "--quiet", self.remote, self.branch], @@ -399,8 +429,17 @@ class GitResultChannel: ) tracking_reference = ( f"refs/heads/{self.branch}:" - f"refs/remotes/{self.remote}/{self.branch}" + f"{self._tracking_reference()}" ) + previous_tracking_head = self._optional_revision(self._tracking_reference()) + if ( + previous_tracking_head is not None + and not self._is_ancestor(previous_tracking_head, remote_head) + ): + raise ResultHistoryRewritten( + "remote result branch is not a fast-forward of the local " + f"tracking branch: {previous_tracking_head} -> {remote_head}" + ) tracking_fetch = self._run( ["fetch", "--quiet", self.remote, tracking_reference], check=False, @@ -411,7 +450,7 @@ class GitResultChannel: f"{tracking_fetch.stderr.strip()}" ) tracking_head = self._run( - ["rev-parse", f"refs/remotes/{self.remote}/{self.branch}"] + ["rev-parse", self._tracking_reference()] ).stdout.strip() if tracking_head != remote_head: raise ResultChannelError("remote-tracking result branch is stale") @@ -419,16 +458,100 @@ class GitResultChannel: def read(self) -> tuple[RxHandoffResult, str, str]: remote_head = self.fetch() - reference = f"refs/remotes/{self.remote}/{self.branch}" - json_result = self._run(["show", f"{reference}:{RESULT_JSON_PATH}"]) + return self.read_commit(remote_head) + + def read_commit(self, commit: str) -> tuple[RxHandoffResult, str, str]: + """Read and validate the result stored by one exact commit.""" + + if not _HEAD_RE.fullmatch(commit): + raise ResultChannelError("result 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"result commit is unavailable: {commit}") + json_result = self._run( + ["show", f"{commit}:{RESULT_JSON_PATH}"], + check=False, + ) + if json_result.returncode != 0: + raise ResultChannelError( + f"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("remote result JSON root is not an object") + raise ResultChannelError( + f"result JSON root is not an object at commit {commit}" + ) result = result_from_mapping(raw) - observed_head = self._run(["rev-parse", reference]).stdout.strip() - if observed_head != remote_head: - raise ResultChannelError("result commit differs from remote HEAD") - return result, result_digest(result), observed_head + return result, result_digest(result), commit + + def commits_since( + self, + last_processed_commit: str | None, + ) -> tuple[list[str], str]: + """Return every new result-branch commit from oldest to newest.""" + + remote_head = self.fetch() + if last_processed_commit is not None: + if not _HEAD_RE.fullmatch(last_processed_commit): + raise ResultChannelError( + "last processed result commit must be a full lowercase Git hash" + ) + if self._optional_revision(last_processed_commit) is None: + raise ResultHistoryRewritten( + "last processed result commit is unavailable locally" + ) + if not self._is_ancestor(last_processed_commit, remote_head): + raise ResultHistoryRewritten( + "remote result history no longer descends from the last " + f"processed commit: {last_processed_commit} -> {remote_head}" + ) + if last_processed_commit == remote_head: + return [], remote_head + revision_range = f"{last_processed_commit}..{remote_head}" + history = self._run( + ["rev-list", "--reverse", "--topo-order", revision_range] + ).stdout + commits = [line.strip() for line in history.splitlines() if line.strip()] + return commits, 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}:{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 find_commit_by_digest(self, digest: str) -> str | None: + """Find the newest valid result commit with a known canonical digest.""" + + if not _SHA256_RE.fullmatch(digest): + raise ResultChannelError("stored result digest is invalid") + remote_head = self.fetch() + history = self._run(["rev-list", remote_head]).stdout + for commit in ( + line.strip() for line in history.splitlines() if line.strip() + ): + try: + _result, observed_digest, _head = self.read_commit(commit) + except (json.JSONDecodeError, ResultChannelError, ValueError): + continue + if observed_digest == digest: + return commit + return None def publish(self, result: RxHandoffResult) -> PublishOutcome: validate_result(result) diff --git a/tools/tx_agent_watcher.py b/tools/tx_agent_watcher.py index 77e0ca0..c0941f8 100644 --- a/tools/tx_agent_watcher.py +++ b/tools/tx_agent_watcher.py @@ -27,6 +27,7 @@ from tools.rf_handoff_channel import ( # noqa: E402 READY_STATE, ResultBranchMissing, ResultChannelError, + ResultHistoryRewritten, RxHandoffResult, evaluate_ready, load_result, @@ -36,6 +37,7 @@ from tools.rf_handoff_channel import ( # noqa: E402 DEFAULT_STATE_PATH = ROOT / ".local" / "tx_agent_watcher_state.json" DEFAULT_OUTBOX = ROOT / ".local" / "tx_authorizations" +STATE_VERSION = 2 _HEAD_RE = re.compile(r"^[0-9a-f]{40}$") @@ -44,6 +46,15 @@ class ResultSource(Protocol): def read(self) -> tuple[RxHandoffResult, str, str]: ... + def read_commit(self, commit: str) -> tuple[RxHandoffResult, str, str]: ... + + def commits_since( + self, + last_processed_commit: str | None, + ) -> tuple[list[str], str]: ... + + def find_commit_by_digest(self, digest: str) -> str | None: ... + class TxExecutor(Protocol): def prepare( @@ -58,9 +69,12 @@ class TxExecutor(Protocol): @dataclass class TxWatcherState: + state_version: int = STATE_VERSION running: bool = False pid: int | None = None last_seen_digest: str | None = None + last_seen_commit: str | None = None + last_processed_rx_node_commit: str | None = None last_prompted_digest: str | None = None last_authorized_digest: str | None = None channel_test_ready_digest: str | None = None @@ -80,6 +94,21 @@ class StaticResultSource: def read(self) -> tuple[RxHandoffResult, str, str]: return self.result, self.digest, self.result.source_head + def read_commit(self, commit: str) -> tuple[RxHandoffResult, str, str]: + if commit != self.result.source_head: + raise ResultHistoryRewritten("static result commit changed") + return self.read() + + def commits_since( + self, + last_processed_commit: str | None, + ) -> tuple[list[str], str]: + head = self.result.source_head + return ([] if last_processed_commit == head else [head], head) + + def find_commit_by_digest(self, digest: str) -> str | None: + return self.result.source_head if digest == self.digest else None + class ManualTxExecutor: """Write a local authorization record; never access SDR hardware.""" @@ -131,7 +160,12 @@ def load_state(path: Path) -> TxWatcherState: allowed = set(TxWatcherState.__dataclass_fields__) if any(key not in allowed for key in raw): raise ValueError("TX watcher state contains an unknown field") - return TxWatcherState(**raw) + version = raw.get("state_version", 1) + if type(version) is not int or version not in {1, STATE_VERSION}: + raise ValueError("TX watcher state version is not supported") + migrated = dict(raw) + migrated["state_version"] = STATE_VERSION + return TxWatcherState(**migrated) def save_state(path: Path, state: TxWatcherState) -> None: @@ -352,6 +386,77 @@ def _git_output(repository: Path, arguments: list[str]) -> str: ).stdout.strip() +def migrate_state_cursor( + source: ResultSource, + state: TxWatcherState, +) -> bool: + """Map a version-1 digest-only state to its exact result commit.""" + + state.state_version = STATE_VERSION + if state.last_processed_rx_node_commit is not None: + if state.last_seen_commit is None: + state.last_seen_commit = state.last_processed_rx_node_commit + return True + return False + if state.last_seen_digest is None: + return False + matched_commit = source.find_commit_by_digest(state.last_seen_digest) + if matched_commit is None: + raise ResultHistoryRewritten( + "stored result digest cannot be mapped to origin/rx-node history" + ) + state.last_seen_commit = matched_commit + state.last_processed_rx_node_commit = matched_commit + return True + + +def process_history_cycle( + source: ResultSource, + state: TxWatcherState, + executor: TxExecutor, + *, + now_function: Callable[[], datetime], + input_function: Callable[[str], str], + save_function: Callable[[TxWatcherState], None], + output_function: Callable[[str], None] = print, + dry_run: bool = False, + expected_source_head: str | None = None, + max_age_seconds: float = 30.0, + min_remaining_seconds: float = 10.0, +) -> list[tuple[str, str]]: + """Process every new result commit in chronological order.""" + + commits, _remote_head = source.commits_since( + state.last_processed_rx_node_commit + ) + outcomes: list[tuple[str, str]] = [] + for commit in commits: + result, digest, result_head = source.read_commit(commit) + status, state = handle_result( + result, + digest, + result_head, + state, + source, + executor, + now_function=now_function, + input_function=input_function, + output_function=output_function, + dry_run=dry_run, + expected_source_head=expected_source_head, + max_age_seconds=max_age_seconds, + min_remaining_seconds=min_remaining_seconds, + ) + state.last_seen_commit = result_head + state.last_processed_rx_node_commit = result_head + state.state_version = STATE_VERSION + save_function(state) + outcomes.append((result_head, status)) + if status != "DUPLICATE_SUPPRESSED": + output_function(f"TX WATCHER STATUS = {status}") + return outcomes + + def run_watcher(arguments: argparse.Namespace) -> int: state_path = arguments.state_file.resolve() if arguments.request_stop: @@ -375,26 +480,6 @@ def run_watcher(arguments: argparse.Namespace) -> int: stop_path = stop_path_for(state_path) stop_path.unlink(missing_ok=True) - if arguments.result_file is not None: - source: ResultSource = StaticResultSource(load_result(arguments.result_file)) - expected_source_head = ( - arguments.expected_source_head or source.result.source_head - ) - else: - branch = _git_output(repository, ["branch", "--show-current"]) - if branch != arguments.source_branch: - raise ValueError("TX watcher must run on the configured source branch") - expected_source_head = ( - arguments.expected_source_head - or _git_output(repository, ["rev-parse", "HEAD"]) - ) - source = GitResultChannel( - repository, - remote=arguments.remote, - branch=arguments.result_branch, - ) - if not _HEAD_RE.fullmatch(expected_source_head): - raise ValueError("expected source HEAD must be a full lowercase Git hash") executor = ManualTxExecutor(arguments.outbox.resolve()) def input_function(prompt: str) -> str: @@ -405,26 +490,51 @@ def run_watcher(arguments: argparse.Namespace) -> int: return interruptible_input(prompt, stop_path) try: + if arguments.result_file is not None: + static_source = StaticResultSource(load_result(arguments.result_file)) + source: ResultSource = static_source + expected_source_head = ( + arguments.expected_source_head or static_source.result.source_head + ) + else: + branch = _git_output(repository, ["branch", "--show-current"]) + if branch != arguments.source_branch: + raise ValueError( + "TX watcher must run on the configured source branch" + ) + expected_source_head = ( + arguments.expected_source_head + or _git_output(repository, ["rev-parse", "HEAD"]) + ) + source = GitResultChannel( + repository, + remote=arguments.remote, + branch=arguments.result_branch, + ) + if not _HEAD_RE.fullmatch(expected_source_head): + raise ValueError( + "expected source HEAD must be a full lowercase Git hash" + ) + if migrate_state_cursor(source, state): + save_state(state_path, state) + while not stop_path.exists(): try: - result, digest, result_head = source.read() - status, state = handle_result( - result, - digest, - result_head, - state, + process_history_cycle( source, + state, executor, now_function=lambda: datetime.now(timezone.utc), input_function=input_function, + save_function=lambda current: save_state( + state_path, + current, + ), dry_run=arguments.dry_run, expected_source_head=expected_source_head, max_age_seconds=arguments.max_age, min_remaining_seconds=arguments.min_remaining, ) - save_state(state_path, state) - if status != "DUPLICATE_SUPPRESSED": - print(f"TX WATCHER STATUS = {status}") if arguments.once or arguments.dry_run: break except (