diff --git a/.gitignore b/.gitignore index 31488cc..4242d1e 100644 --- a/.gitignore +++ b/.gitignore @@ -69,4 +69,5 @@ Desktop.ini # Локальные настройки и секреты .env *.local +.local/ diff --git a/docs/handoff/NEXT_RX_TASK.md b/docs/handoff/NEXT_RX_TASK.md index ff9f368..217be27 100644 --- a/docs/handoff/NEXT_RX_TASK.md +++ b/docs/handoff/NEXT_RX_TASK.md @@ -1,3 +1,15 @@ +TASK_ID = RF001A_CLKIN_001 +TASK_TITLE = Verify RF-001A slave CLKIN +SOURCE_BRANCH = video-development +EXPECTED_HEAD = b28befc69df8e1493db3ed2b122696b44ec27221 +REQUIRES_APPROVAL = YES +HARDWARE_ACCESS = YES +RF_RX = YES +RF_TX = NO +ALLOW_COMMIT = NO +ALLOW_PUSH = RX_RESULT_ONLY +RISK_LEVEL = YELLOW + This file is the authoritative current task for the RX laptop. The RX assistant must read RX_NODE_STATE.md first, then execute this file. diff --git a/docs/handoff/RX_AGENT_PROTOCOL.md b/docs/handoff/RX_AGENT_PROTOCOL.md new file mode 100644 index 0000000..3d822a4 --- /dev/null +++ b/docs/handoff/RX_AGENT_PROTOCOL.md @@ -0,0 +1,85 @@ +# RX agent handoff protocol + +## Purpose + +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 never executed until the user approves it. The first implementation uses only `ManualExecutor`: it prepares a local task outbox and does not start a model itself. + +## Task header + +`docs/handoff/NEXT_RX_TASK.md` starts with these plain key-value fields: + +```text +TASK_ID = RF001A_CLKIN_001 +TASK_TITLE = Verify RF-001A slave CLKIN +SOURCE_BRANCH = video-development +EXPECTED_HEAD = +REQUIRES_APPROVAL = YES +HARDWARE_ACCESS = YES +RF_RX = YES +RF_TX = NO +ALLOW_COMMIT = NO +ALLOW_PUSH = RX_RESULT_ONLY +RISK_LEVEL = YELLOW +``` + +`TASK_ID` is immutable and unique. A revised task receives a new ID. `EXPECTED_HEAD` names the known base commit against which the task was written; the watcher requires it to be an ancestor of the fetched task commit. This avoids an impossible self-reference to the commit that contains the header. `ALLOW_PUSH = RX_RESULT_ONLY` permits only a small result publication to `rx-node`; it never permits a push to `video-development`. `ALLOW_COMMIT` governs task changes in the source checkout; a separately authorized result-only commit on `rx-node` follows `ALLOW_PUSH`. `RISK_LEVEL` is an optional plain field used to declare RED actions other than RF TX. + +The watcher rejects missing, duplicate or invalid fields. It reads the file with `git show origin/video-development:docs/handoff/NEXT_RX_TASK.md`, so it does not need to change the local checkout. + +## Safety levels and approval + +| Level | Operations | Approval rule | +|---|---|---| +| GREEN | Software-only checks, parsing and local result preparation | The task's normal approval applies | +| YELLOW | Hardware inventory, clock, gain or bounded RF RX | The user must choose `[A] Approve` | +| RED | RF TX, firmware, CPLD, Zadig, USB driver changes, destructive Git or filesystem operations | A normal approval is insufficient; an executor must require the exact separate action `AUTHORIZE RF TX` or an equivalent task-specific explicit confirmation | + +The watcher derives RED whenever `RF_TX = YES` and rejects a lower declared level. Firmware, CPLD, Zadig, USB driver changes and destructive actions must declare `RISK_LEVEL = RED`. A non-RF RED approval requires `AUTHORIZE `; RF TX requires exactly `AUTHORIZE RF TX`. The watcher cannot authorize RED itself, and the executor must reject a body/header conflict. + +At the prompt the minimum choices are: + +```text +[A] Approve +[D] Defer +[S] Stop watcher +``` + +The watcher records the task ID before leaving the prompt. Deferred and already seen IDs are not shown again on every polling cycle. A changed task must have a new `TASK_ID`. + +## Local state and executor boundary + +Runtime state is stored in `.local/rx_agent_state.json`; prepared prompts and result skeletons are stored in `.local/rx_agent_outbox//`. `.local/` is ignored by Git. State includes `last_seen_task_id`, `last_completed_task_id`, `deferred_task_id`, `running` and the watcher PID. Writes use a temporary file followed by an atomic replace. + +`TaskExecutor` is the backend interface. Version 1 provides `ManualExecutor`, which writes `TASK.md`, `RX_RESULT.md` and `RX_RESULT.json` after approval. It refuses to overwrite an existing task outbox. It does not invoke an agent, access SDR hardware, commit or push. A `CodexExecutor` may be added only after the Codex CLI is found and tested on the RX laptop. Availability on the main PC is not sufficient evidence. + +After a human or separately authorized executor finishes the task, mark the local ID complete: + +```powershell +python -B tools/rx_agent_watcher.py --mark-completed +``` + +## Result and `rx-node` branch + +The result follows [RX_RESULT_TEMPLATE.md](RX_RESULT_TEMPLATE.md) and has matching Markdown and JSON. The proposed publication branch is `rx-node`. + +The RX laptop must never commit or push to `video-development`. A separately approved publication step may place only these items on `rx-node`: + +- handoff result Markdown; +- result JSON; +- small log metadata needed to interpret the result. + +Raw IQ, video, waveform files, large logs, secrets and machine-local state remain outside Git. The watcher does not create `rx-node`, switch branches, commit or push. This keeps branch creation and publication subject to the repository's current instructions and a separate explicit authorization. + +The main PC later uses `git fetch origin rx-node` and reads the result without merging it blindly. + +## Token and model behavior + +- Git polling itself does not call an LLM. +- When no new task exists, no model runs. +- A detected but unapproved or deferred task does not run a model. +- After approval, `ManualExecutor` only prepares an outbox. A separately selected executor may invoke one agent. +- The executor stops after task completion. The watcher returns to Git polling and consumes no model tokens. + +## Dry-run contract + +Dry-run requires a local `--task-file`; it does not fetch Git, access SDR, invoke an agent, perform RF RX/TX, commit or push. The tracked fake task is `tests/fixtures/watcher_test_task.md` with `TASK_ID = WATCHER_TEST_001` and `RF_TX = NO`. diff --git a/docs/handoff/RX_AGENT_USAGE.md b/docs/handoff/RX_AGENT_USAGE.md new file mode 100644 index 0000000..b6d73c0 --- /dev/null +++ b/docs/handoff/RX_AGENT_USAGE.md @@ -0,0 +1,76 @@ +# RX watcher usage + +## START + +On the RX laptop, pull `video-development`, install the project's normal Python dependencies, then double-click: + +```text +tools\start_rx_watcher.cmd +``` + +The watcher stays in a visible console and polls `origin/video-development` every 30 seconds. The interval is configurable with `--poll-interval`. It does not start ChatGPT, Codex or another agent. + +## 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. + +## NEW TASK + +The console shows: + +```text +NEW TASK DETECTED + +TASK ID = +TITLE = +SAFETY LEVEL = +HARDWARE ACCESS = +RF RX = +RF TX = +``` + +Choose `[A] Approve`, `[D] Defer` or `[S] Stop watcher`. An RF TX task also requires the exact separate phrase `AUTHORIZE RF TX`; another RED task requires `AUTHORIZE `. This only opens the executor gate; the first version's `ManualExecutor` still performs no RF action and starts no model. + +## APPROVE + +After approval, open: + +```text +.local\rx_agent_outbox\\TASK.md +``` + +Transfer that prepared task to the chosen agent manually. Fill the adjacent `RX_RESULT.md` and `RX_RESULT.json` with measured results. When work is actually complete, run: + +```powershell +python -B tools/rx_agent_watcher.py --mark-completed +``` + +Do not claim completion merely because the prompt was prepared. + +## DONE and result publication + +If the task says `ALLOW_PUSH = RX_RESULT_ONLY` and the user separately authorizes publication, publish only small result Markdown/JSON and log metadata to `rx-node`. Never commit or push from the RX laptop to `video-development`. Do not put raw IQ, video or waveform files in Git. + +## STOP + +Double-click: + +```text +tools\stop_rx_watcher.cmd +``` + +The script creates a local stop request and waits for the watcher PID to clear. The approval prompt is interruptible on Windows, and the polling delay checks the stop request every 0.1 seconds. A successful stop prints `WATCHER STOPPED`; no polling process remains. + +## DRY RUN + +From the repository root: + +```powershell +python -B tools/rx_agent_watcher.py --dry-run --once --approval approve --task-file tests/fixtures/watcher_test_task.md +``` + +Expected behavior: the fake task is detected, approved and written under `.local/rx_agent_outbox/WATCHER_TEST_001/`. No Git fetch, SDR access, agent call, RF RX/TX, commit or push occurs. Delete `.local/` at any time if the local state and outbox are no longer needed. + +## Executor availability + +Version 1 always provides `ManualExecutor`. Before adding or selecting a Codex backend on RX, run `Get-Command codex` on that laptop and verify a harmless local invocation. Do not infer RX availability from the main PC. diff --git a/docs/handoff/RX_RESULT_TEMPLATE.md b/docs/handoff/RX_RESULT_TEMPLATE.md new file mode 100644 index 0000000..af2ad00 --- /dev/null +++ b/docs/handoff/RX_RESULT_TEMPLATE.md @@ -0,0 +1,23 @@ +TASK_ID = +TASK_STATUS = +SOURCE_HEAD = +STARTED_AT = +FINISHED_AT = + +SUMMARY = + +CHECKS = + +HARDWARE_ACTIONS = + +RF_TX = +YES / NO + +FILES_CREATED = + +ERRORS = + +NEXT_ACTION = + +The result publisher must also create `RX_RESULT.json` with the same fields. +Raw IQ, video and waveform files must stay outside Git. diff --git a/pyrightconfig.json b/pyrightconfig.json index e45e140..74190de 100644 --- a/pyrightconfig.json +++ b/pyrightconfig.json @@ -23,11 +23,13 @@ "tools/qpsk_rf_tx.py", "tools/qpsk_rf_rx.py", "tools/rf001a_prepare.py", + "tools/rx_agent_watcher.py", "tools/clock001.py", "tests/test_video005_iq.py", "tests/test_video005_video.py", "tests/test_rf001_preparation.py", "tests/test_rf001a_prepare.py", + "tests/test_rx_agent_watcher.py", "tests/test_clock001.py" ], "typeCheckingMode": "basic", diff --git a/tests/fixtures/watcher_test_task.md b/tests/fixtures/watcher_test_task.md new file mode 100644 index 0000000..e3654d5 --- /dev/null +++ b/tests/fixtures/watcher_test_task.md @@ -0,0 +1,15 @@ +TASK_ID = WATCHER_TEST_001 +TASK_TITLE = Watcher dry-run test +SOURCE_BRANCH = video-development +EXPECTED_HEAD = b28befc69df8e1493db3ed2b122696b44ec27221 +REQUIRES_APPROVAL = YES +HARDWARE_ACCESS = NO +RF_RX = NO +RF_TX = NO +ALLOW_COMMIT = NO +ALLOW_PUSH = NO +RISK_LEVEL = GREEN + +# Fake task + +Create a dry-run result only. Do not access hardware, start an agent, or change Git. diff --git a/tests/test_rx_agent_watcher.py b/tests/test_rx_agent_watcher.py new file mode 100644 index 0000000..6245ff7 --- /dev/null +++ b/tests/test_rx_agent_watcher.py @@ -0,0 +1,189 @@ +"""Unit tests for the model-free RX task watcher.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from tools.rx_agent_watcher import ( + ApprovalDecision, + ManualExecutor, + WatcherState, + handle_task, + is_new_task, + load_state, + mark_completed, + parse_task, + request_approval, + request_stop, + save_state, +) + + +HEAD = "1" * 40 + + +def task_text(*, task_id: str = "WATCHER_TEST_001", rf_tx: str = "NO") -> str: + hardware = "YES" if rf_tx == "YES" else "NO" + risk_level = "RED" if rf_tx == "YES" else "GREEN" + return f"""TASK_ID = {task_id} +TASK_TITLE = Watcher test +SOURCE_BRANCH = video-development +EXPECTED_HEAD = {HEAD} +REQUIRES_APPROVAL = YES +HARDWARE_ACCESS = {hardware} +RF_RX = NO +RF_TX = {rf_tx} +ALLOW_COMMIT = NO +ALLOW_PUSH = RX_RESULT_ONLY +RISK_LEVEL = {risk_level} + +# Body +Do nothing. +""" + + +def responses(*values: str): + iterator = iter(values) + return lambda _prompt: next(iterator) + + +def test_task_parsing_and_validation() -> None: + task = parse_task(task_text()) + assert task.task_id == "WATCHER_TEST_001" + assert task.source_branch == "video-development" + assert task.safety_level == "GREEN" + assert not task.rf_tx + with pytest.raises(ValueError, match="missing task header"): + parse_task("TASK_ID = INCOMPLETE") + with pytest.raises(ValueError, match="duplicate"): + parse_task(task_text().replace( + "TASK_TITLE = Watcher test", + "TASK_TITLE = Watcher test\nTASK_TITLE = Duplicate", + )) + + +def test_state_persistence_and_completion(tmp_path: Path) -> None: + path = tmp_path / "state.json" + original = WatcherState(last_seen_task_id="WATCHER_TEST_001", running=True, pid=123) + save_state(path, original) + assert load_state(path) == original + mark_completed(path, "WATCHER_TEST_001") + completed = load_state(path) + assert completed.last_completed_task_id == "WATCHER_TEST_001" + assert completed.deferred_task_id is None + with pytest.raises(ValueError, match="last seen"): + mark_completed(path, "WATCHER_TEST_002") + + +def test_approval_defer_stop_and_yellow_approve() -> None: + green = parse_task(task_text()) + assert request_approval(green, responses("D"), lambda _text: None) == ApprovalDecision.DEFER + assert request_approval(green, responses("S"), lambda _text: None) == ApprovalDecision.STOP + yellow_text = task_text().replace( + "HARDWARE_ACCESS = NO", "HARDWARE_ACCESS = YES" + ).replace("RISK_LEVEL = GREEN", "RISK_LEVEL = YELLOW") + yellow = parse_task(yellow_text) + assert yellow.safety_level == "YELLOW" + assert request_approval(yellow, responses("A"), lambda _text: None) == ApprovalDecision.APPROVE + + +def test_rf_tx_red_gate_requires_separate_phrase() -> None: + red = parse_task(task_text(rf_tx="YES")) + assert red.safety_level == "RED" + assert request_approval( + red, responses("A", "yes"), lambda _text: None + ) == ApprovalDecision.BLOCKED + assert request_approval( + red, responses("A", "AUTHORIZE RF TX"), lambda _text: None + ) == ApprovalDecision.APPROVE + + +def test_non_rf_red_task_also_requires_separate_phrase() -> None: + text = task_text(task_id="FIRMWARE_TEST_001").replace( + "RISK_LEVEL = GREEN", "RISK_LEVEL = RED" + ) + red = parse_task(text) + assert not red.rf_tx + assert request_approval( + red, responses("A", "AUTHORIZE RF TX"), lambda _text: None + ) == ApprovalDecision.BLOCKED + assert request_approval( + red, responses("A", "AUTHORIZE FIRMWARE_TEST_001"), lambda _text: None + ) == ApprovalDecision.APPROVE + + +def test_duplicate_task_is_suppressed_without_prompt(tmp_path: Path) -> None: + text = task_text() + task = parse_task(text) + state = WatcherState(last_seen_task_id=task.task_id) + assert not is_new_task(task, state) + + def forbidden_input(_prompt: str) -> str: + raise AssertionError("duplicate task prompted the user") + + status, returned = handle_task( + text, + HEAD, + state, + ManualExecutor(tmp_path), + dry_run=True, + input_function=forbidden_input, + output_function=lambda _text: None, + ) + assert status == "DUPLICATE_SUPPRESSED" + assert returned == state + assert list(tmp_path.iterdir()) == [] + + +def test_defer_persists_seen_id_and_creates_no_result(tmp_path: Path) -> None: + state = WatcherState() + status, state = handle_task( + task_text(), + HEAD, + state, + ManualExecutor(tmp_path), + dry_run=True, + input_function=responses("D"), + output_function=lambda _text: None, + ) + assert status == "DEFERRED" + assert state.last_seen_task_id == "WATCHER_TEST_001" + assert state.deferred_task_id == "WATCHER_TEST_001" + assert list(tmp_path.iterdir()) == [] + + +def test_dry_run_approval_writes_prompt_and_matching_results(tmp_path: Path) -> None: + status, state = handle_task( + task_text(), + HEAD, + WatcherState(), + ManualExecutor(tmp_path), + dry_run=True, + input_function=responses("A"), + output_function=lambda _text: None, + ) + directory = tmp_path / "WATCHER_TEST_001" + assert status == "PREPARED" + assert state.last_seen_task_id == "WATCHER_TEST_001" + assert state.last_completed_task_id is None + assert (directory / "TASK.md").exists() + markdown = (directory / "RX_RESULT.md").read_text(encoding="utf-8") + result = json.loads((directory / "RX_RESULT.json").read_text(encoding="utf-8")) + assert "TASK_STATUS = DRY_RUN_PREPARED" in markdown + assert result["task_status"] == "DRY_RUN_PREPARED" + assert result["rf_tx"] == "NO" + assert result["hardware_actions"] == [] + with pytest.raises(FileExistsError): + ManualExecutor(tmp_path).prepare( + parse_task(task_text()), task_text(), HEAD, dry_run=True + ) + + +def test_stop_request_for_idle_watcher(tmp_path: Path) -> None: + state_path = tmp_path / "state.json" + save_state(state_path, WatcherState(running=False, pid=None)) + assert request_stop(state_path, timeout_s=0.1) + assert state_path.with_suffix(".json.stop").read_text(encoding="ascii") == "STOP\n" diff --git a/tools/rx_agent_watcher.py b/tools/rx_agent_watcher.py new file mode 100644 index 0000000..16cb72a --- /dev/null +++ b/tools/rx_agent_watcher.py @@ -0,0 +1,590 @@ +"""Локальный Git-watcher заданий RX без LLM и доступа к SDR.""" + +from __future__ import annotations + +import argparse +from dataclasses import asdict, dataclass +from datetime import datetime, timezone +from enum import Enum +import json +import os +from pathlib import Path +import re +import subprocess +import sys +import time +from typing import Callable, Protocol + + +ROOT = Path(__file__).resolve().parents[1] +DEFAULT_TASK_PATH = "docs/handoff/NEXT_RX_TASK.md" +DEFAULT_STATE_PATH = ROOT / ".local/rx_agent_state.json" +DEFAULT_OUTBOX = ROOT / ".local/rx_agent_outbox" +REQUIRED_FIELDS = ( + "TASK_ID", + "TASK_TITLE", + "SOURCE_BRANCH", + "EXPECTED_HEAD", + "REQUIRES_APPROVAL", + "HARDWARE_ACCESS", + "RF_RX", + "RF_TX", + "ALLOW_COMMIT", + "ALLOW_PUSH", +) +OPTIONAL_FIELDS = ("RISK_LEVEL",) +HEADER_FIELDS = REQUIRED_FIELDS + OPTIONAL_FIELDS +TASK_ID_PATTERN = re.compile(r"^[A-Z0-9][A-Z0-9_-]{2,63}$") +BRANCH_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,127}$") +REMOTE_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$") +HEAD_PATTERN = re.compile(r"^[0-9a-f]{40}$") + + +@dataclass(frozen=True) +class TaskDefinition: + """Проверенный машинно читаемый заголовок задания.""" + + task_id: str + task_title: str + source_branch: str + expected_head: str + requires_approval: bool + hardware_access: bool + rf_rx: bool + rf_tx: bool + allow_commit: bool + allow_push: str + risk_level: str + + @property + def safety_level(self) -> str: + if self.rf_tx or self.risk_level == "RED": + return "RED" + if self.hardware_access or self.rf_rx or self.risk_level == "YELLOW": + return "YELLOW" + return "GREEN" + + +@dataclass +class WatcherState: + """Локальное состояние машины, которое не должно попадать в Git.""" + + last_seen_task_id: str | None = None + last_completed_task_id: str | None = None + deferred_task_id: str | None = None + running: bool = False + pid: int | None = None + + +@dataclass(frozen=True) +class PreparedResult: + task_id: str + task_status: str + source_head: str + started_at: str + finished_at: str + summary: str + checks: list[str] + hardware_actions: list[str] + rf_tx: str + files_created: list[str] + errors: list[str] + next_action: str + + +class ApprovalDecision(Enum): + APPROVE = "APPROVE" + DEFER = "DEFER" + STOP = "STOP" + BLOCKED = "BLOCKED" + + +class TaskExecutor(Protocol): + """Интерфейс исполнителя, не зависящий от конкретного LLM-клиента.""" + + def prepare( + self, + task: TaskDefinition, + task_text: str, + source_head: str, + *, + dry_run: bool, + ) -> tuple[Path, Path, Path]: + """Подготовить локальный outbox без запуска агента.""" + ... + + +def _yes_no(value: str, field: str) -> bool: + normalized = value.upper() + if normalized not in {"YES", "NO"}: + raise ValueError(f"{field} must be YES or NO") + return normalized == "YES" + + +def parse_task(text: str) -> TaskDefinition: + """Разобрать простой key-value заголовок и отклонить неоднозначность.""" + values: dict[str, str] = {} + for raw_line in text.splitlines(): + line = raw_line.strip() + if not line or line == "---": + continue + if line.startswith("#") or line.startswith("```"): + if values: + break + continue + if "=" not in line: + if values: + break + continue + key, value = (part.strip() for part in line.split("=", 1)) + if key not in HEADER_FIELDS: + if values: + break + continue + if key in values: + raise ValueError(f"duplicate task header field: {key}") + if not value: + raise ValueError(f"empty task header field: {key}") + values[key] = value + + missing = [field for field in REQUIRED_FIELDS if field not in values] + if missing: + raise ValueError(f"missing task header fields: {', '.join(missing)}") + task_id = values["TASK_ID"] + branch = values["SOURCE_BRANCH"] + expected_head = values["EXPECTED_HEAD"].lower() + if not TASK_ID_PATTERN.fullmatch(task_id): + raise ValueError("invalid TASK_ID") + if not BRANCH_PATTERN.fullmatch(branch) or ".." in branch or branch.endswith("/"): + raise ValueError("invalid SOURCE_BRANCH") + if not HEAD_PATTERN.fullmatch(expected_head): + raise ValueError("EXPECTED_HEAD must be a full lowercase Git hash") + allow_push = values["ALLOW_PUSH"].upper() + if allow_push not in {"NO", "RX_RESULT_ONLY"}: + raise ValueError("ALLOW_PUSH must be NO or RX_RESULT_ONLY") + risk_level = values.get("RISK_LEVEL", "").upper() + if risk_level and risk_level not in {"GREEN", "YELLOW", "RED"}: + raise ValueError("RISK_LEVEL must be GREEN, YELLOW or RED") + if not risk_level: + risk_level = ( + "RED" if _yes_no(values["RF_TX"], "RF_TX") else + "YELLOW" if ( + _yes_no(values["HARDWARE_ACCESS"], "HARDWARE_ACCESS") + or _yes_no(values["RF_RX"], "RF_RX") + ) else "GREEN" + ) + task = TaskDefinition( + task_id=task_id, + task_title=values["TASK_TITLE"], + source_branch=branch, + expected_head=expected_head, + requires_approval=_yes_no(values["REQUIRES_APPROVAL"], "REQUIRES_APPROVAL"), + hardware_access=_yes_no(values["HARDWARE_ACCESS"], "HARDWARE_ACCESS"), + rf_rx=_yes_no(values["RF_RX"], "RF_RX"), + rf_tx=_yes_no(values["RF_TX"], "RF_TX"), + allow_commit=_yes_no(values["ALLOW_COMMIT"], "ALLOW_COMMIT"), + allow_push=allow_push, + risk_level=risk_level, + ) + if task.rf_tx and not task.hardware_access: + raise ValueError("RF_TX=YES requires HARDWARE_ACCESS=YES") + if task.rf_tx and task.risk_level != "RED": + raise ValueError("RF_TX=YES requires RISK_LEVEL=RED") + if (task.hardware_access or task.rf_rx) and task.risk_level == "GREEN": + raise ValueError("hardware access cannot use RISK_LEVEL=GREEN") + return task + + +def load_state(path: Path) -> WatcherState: + """Прочитать локальное состояние; отсутствие файла означает первый запуск.""" + if not path.exists(): + return WatcherState() + raw = json.loads(path.read_text(encoding="utf-8")) + allowed = set(WatcherState.__dataclass_fields__) + if not isinstance(raw, dict) or any(key not in allowed for key in raw): + raise ValueError("invalid watcher state") + return WatcherState(**raw) + + +def save_state(path: Path, state: WatcherState) -> None: + """Атомарно сохранить состояние рядом с целевым файлом.""" + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(path.suffix + ".tmp") + temporary.write_text(json.dumps(asdict(state), indent=2) + "\n", encoding="utf-8") + os.replace(temporary, path) + + +def is_new_task(task: TaskDefinition, state: WatcherState) -> bool: + """Подавить задание, которое уже показывалось или было завершено.""" + return task.task_id not in {state.last_seen_task_id, state.last_completed_task_id} + + +def task_summary(task: TaskDefinition) -> str: + return ( + "NEW TASK DETECTED\n\n" + f"TASK ID = {task.task_id}\n" + f"TITLE = {task.task_title}\n" + f"SAFETY LEVEL = {task.safety_level}\n" + f"HARDWARE ACCESS = {'YES' if task.hardware_access else 'NO'}\n" + f"RF RX = {'YES' if task.rf_rx else 'NO'}\n" + f"RF TX = {'YES' if task.rf_tx else 'NO'}" + ) + + +def request_approval( + task: TaskDefinition, + input_function: Callable[[str], str] = input, + output_function: Callable[[str], None] = print, +) -> ApprovalDecision: + """Запросить A/D/S; для RED нужна вторая точная фраза.""" + output_function(task_summary(task)) + choice = input_function("[A] Approve [D] Defer [S] Stop watcher: ").strip().upper() + if choice == "D": + return ApprovalDecision.DEFER + if choice == "S": + return ApprovalDecision.STOP + if choice != "A": + return ApprovalDecision.BLOCKED + if task.safety_level == "RED": + required_phrase = "AUTHORIZE RF TX" if task.rf_tx else f"AUTHORIZE {task.task_id}" + phrase = input_function(f"Type {required_phrase} to continue: ").strip() + if phrase != required_phrase: + return ApprovalDecision.BLOCKED + return ApprovalDecision.APPROVE + + +def utc_now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def write_result(output_directory: Path, result: PreparedResult) -> tuple[Path, Path]: + """Записать согласованные Markdown и JSON без публикации.""" + output_directory.mkdir(parents=True, exist_ok=True) + markdown_path = output_directory / "RX_RESULT.md" + json_path = output_directory / "RX_RESULT.json" + markdown = ( + f"TASK_ID = {result.task_id}\n" + f"TASK_STATUS = {result.task_status}\n" + f"SOURCE_HEAD = {result.source_head}\n" + f"STARTED_AT = {result.started_at}\n" + f"FINISHED_AT = {result.finished_at}\n\n" + f"SUMMARY =\n{result.summary}\n\n" + f"CHECKS =\n{json.dumps(result.checks)}\n\n" + f"HARDWARE_ACTIONS =\n{json.dumps(result.hardware_actions)}\n\n" + f"RF_TX =\n{result.rf_tx}\n\n" + f"FILES_CREATED =\n{json.dumps(result.files_created)}\n\n" + f"ERRORS =\n{json.dumps(result.errors)}\n\n" + f"NEXT_ACTION =\n{result.next_action}\n" + ) + markdown_path.write_text(markdown, encoding="utf-8") + json_path.write_text(json.dumps(asdict(result), indent=2) + "\n", encoding="utf-8") + return markdown_path, json_path + + +class ManualExecutor: + """Подготовить локальный prompt и каркас результата без запуска агента.""" + + def __init__(self, outbox: Path = DEFAULT_OUTBOX) -> None: + self.outbox = outbox + + def prepare( + self, + task: TaskDefinition, + task_text: str, + source_head: str, + *, + dry_run: bool, + ) -> tuple[Path, Path, Path]: + directory = self.outbox / task.task_id + directory.mkdir(parents=True, exist_ok=False) + prompt_path = directory / "TASK.md" + prompt_path.write_text(task_text, encoding="utf-8") + result_markdown_path = directory / "RX_RESULT.md" + result_json_path = directory / "RX_RESULT.json" + timestamp = utc_now() + status = "DRY_RUN_PREPARED" if dry_run else "AWAITING_MANUAL_EXECUTION" + result = PreparedResult( + task_id=task.task_id, + task_status=status, + source_head=source_head, + started_at=timestamp, + finished_at=timestamp, + summary="Task prepared for manual transfer; no agent was invoked.", + checks=["TASK_HEADER_VALID", "APPROVAL_GATE_PASS"], + hardware_actions=[], + rf_tx="NO", + files_created=[ + str(prompt_path), + str(result_markdown_path), + str(result_json_path), + ], + errors=[], + next_action="Open TASK.md in the chosen executor after explicit user approval.", + ) + markdown_path, json_path = write_result(directory, result) + return prompt_path, markdown_path, json_path + + +class GitTaskSource: + """Прочитать задание непосредственно из remote-tracking reference.""" + + def __init__( + self, + repository: Path, + *, + remote: str = "origin", + branch: str = "video-development", + task_path: str = DEFAULT_TASK_PATH, + ) -> None: + if not BRANCH_PATTERN.fullmatch(branch) or ".." in branch: + raise ValueError("invalid branch") + if not REMOTE_PATTERN.fullmatch(remote): + raise ValueError("invalid remote") + self.repository = repository + self.remote = remote + self.branch = branch + self.task_path = task_path + + def _run( + self, + arguments: list[str], + *, + check: bool = True, + ) -> subprocess.CompletedProcess[str]: + return subprocess.run( + ["git", *arguments], + cwd=self.repository, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + check=check, + ) + + def fetch(self) -> None: + self._run(["fetch", "--quiet", self.remote, self.branch]) + + def read(self) -> tuple[str, str]: + remote_ref = f"{self.remote}/{self.branch}" + head = self._run(["rev-parse", remote_ref]).stdout.strip() + text = self._run(["show", f"{remote_ref}:{self.task_path}"]).stdout + task = parse_task(text) + if task.source_branch != self.branch: + raise ValueError("task SOURCE_BRANCH disagrees with watched branch") + ancestor = self._run( + ["merge-base", "--is-ancestor", task.expected_head, head], + check=False, + ) + if ancestor.returncode != 0: + raise ValueError("EXPECTED_HEAD is not an ancestor of remote task HEAD") + return text, head + + +def handle_task( + task_text: str, + source_head: str, + state: WatcherState, + executor: TaskExecutor, + *, + dry_run: bool, + input_function: Callable[[str], str] = input, + output_function: Callable[[str], None] = print, +) -> tuple[str, WatcherState]: + """Обработать одно задание без изменений Git и запуска агента.""" + task = parse_task(task_text) + if not is_new_task(task, state): + return "DUPLICATE_SUPPRESSED", state + decision = request_approval(task, input_function, output_function) + state.last_seen_task_id = task.task_id + if decision == ApprovalDecision.DEFER: + state.deferred_task_id = task.task_id + return "DEFERRED", state + if decision == ApprovalDecision.STOP: + return "STOP", state + if decision != ApprovalDecision.APPROVE: + return "APPROVAL_BLOCKED", state + state.deferred_task_id = None + paths = executor.prepare(task, task_text, source_head, dry_run=dry_run) + output_function(f"TASK PREPARED = {paths[0]}") + return "PREPARED", state + + +def stop_path_for(state_path: Path) -> Path: + return state_path.with_suffix(state_path.suffix + ".stop") + + +def process_exists(pid: int | None) -> bool: + if pid is None or pid <= 0: + return False + try: + os.kill(pid, 0) + except OSError: + return False + return True + + +def request_stop(state_path: Path, timeout_s: float) -> bool: + """Запросить остановку watcher и дождаться снятия running state.""" + stop_path = stop_path_for(state_path) + stop_path.parent.mkdir(parents=True, exist_ok=True) + stop_path.write_text("STOP\n", encoding="ascii") + deadline = time.monotonic() + timeout_s + while time.monotonic() < deadline: + state = load_state(state_path) + if not state.running or not process_exists(state.pid): + return True + time.sleep(0.2) + return False + + +def interruptible_input(prompt: str, stop_path: Path) -> str: + """Читать консоль Windows, позволяя stop script прервать ожидание.""" + if os.name != "nt": + return input(prompt) + import msvcrt + + print(prompt, end="", flush=True) + characters: list[str] = [] + while True: + if stop_path.exists(): + print() + return "S" + if not msvcrt.kbhit(): + time.sleep(0.1) + continue + character = msvcrt.getwch() + if character in {"\r", "\n"}: + print() + return "".join(characters) + if character == "\b": + if characters: + characters.pop() + print("\b \b", end="", flush=True) + continue + characters.append(character) + print(character, end="", flush=True) + + +def mark_completed(state_path: Path, task_id: str) -> None: + if not TASK_ID_PATTERN.fullmatch(task_id): + raise ValueError("invalid TASK_ID") + state = load_state(state_path) + if state.last_seen_task_id != task_id: + raise ValueError("only the last seen task may be marked completed") + state.last_completed_task_id = task_id + state.deferred_task_id = None + save_state(state_path, state) + + +def run_watcher(arguments: argparse.Namespace) -> int: + state_path = arguments.state_file.resolve() + if arguments.request_stop: + stopped = request_stop(state_path, arguments.stop_timeout) + print("WATCHER STOPPED" if stopped else "WATCHER STOP TIMEOUT") + return 0 if stopped else 1 + if arguments.mark_completed is not None: + mark_completed(state_path, arguments.mark_completed) + print(f"COMPLETED TASK = {arguments.mark_completed}") + return 0 + if arguments.dry_run and arguments.task_file is None: + raise ValueError("--dry-run requires --task-file and never fetches Git") + + state = load_state(state_path) + if state.running and process_exists(state.pid): + raise RuntimeError(f"watcher already running with PID {state.pid}") + state.running = True + state.pid = os.getpid() + save_state(state_path, state) + stop_path = stop_path_for(state_path) + stop_path.unlink(missing_ok=True) + executor = ManualExecutor(arguments.outbox.resolve()) + source = GitTaskSource( + arguments.repository.resolve(), + remote=arguments.remote, + branch=arguments.branch, + ) + if arguments.approval is not None: + replies = [ + "A" if arguments.approval == "approve" else + "D" if arguments.approval == "defer" else "S" + ] + iterator = iter(replies) + + def automated_input(_prompt: str) -> str: + return next(iterator, "") + + input_function = automated_input + else: + def console_input(prompt: str) -> str: + return interruptible_input(prompt, stop_path) + + input_function = console_input + + try: + while not stop_path.exists(): + try: + if arguments.task_file is not None: + task_text = arguments.task_file.read_text(encoding="utf-8") + source_head = parse_task(task_text).expected_head + else: + source.fetch() + task_text, source_head = source.read() + status, state = handle_task( + task_text, + source_head, + state, + executor, + dry_run=arguments.dry_run, + input_function=input_function, + ) + save_state(state_path, state) + if status != "DUPLICATE_SUPPRESSED": + print(f"WATCHER STATUS = {status}") + if status == "STOP": + break + except (OSError, subprocess.SubprocessError, ValueError) as error: + print(f"WATCHER ERROR = {error}", file=sys.stderr) + if arguments.once or arguments.dry_run: + return 1 + if arguments.once or arguments.dry_run or arguments.task_file is not None: + break + stop_path.parent.mkdir(parents=True, exist_ok=True) + for _ in range(max(1, int(arguments.poll_interval * 10))): + if stop_path.exists(): + break + time.sleep(0.1) + finally: + state.running = False + state.pid = None + save_state(state_path, state) + stop_path.unlink(missing_ok=True) + return 0 + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--repository", type=Path, default=ROOT) + parser.add_argument("--remote", default="origin") + parser.add_argument("--branch", default="video-development") + parser.add_argument("--poll-interval", type=float, default=30.0) + parser.add_argument("--state-file", type=Path, default=DEFAULT_STATE_PATH) + parser.add_argument("--outbox", type=Path, default=DEFAULT_OUTBOX) + parser.add_argument("--once", action="store_true") + parser.add_argument("--dry-run", action="store_true") + parser.add_argument("--task-file", type=Path) + parser.add_argument("--approval", choices=("approve", "defer", "stop")) + parser.add_argument("--request-stop", action="store_true") + parser.add_argument("--stop-timeout", type=float, default=35.0) + parser.add_argument("--mark-completed") + return parser + + +def main() -> None: + arguments = build_parser().parse_args() + if arguments.poll_interval <= 0 or arguments.stop_timeout <= 0: + raise SystemExit("intervals must be positive") + raise SystemExit(run_watcher(arguments)) + + +if __name__ == "__main__": + main() diff --git a/tools/start_rx_watcher.cmd b/tools/start_rx_watcher.cmd new file mode 100644 index 0000000..84d0ce8 --- /dev/null +++ b/tools/start_rx_watcher.cmd @@ -0,0 +1,9 @@ +@echo off +setlocal +cd /d "%~dp0.." +python -B tools\rx_agent_watcher.py +set "RX_WATCHER_EXIT=%ERRORLEVEL%" +echo. +echo RX watcher exited with code %RX_WATCHER_EXIT%. +pause +exit /b %RX_WATCHER_EXIT% diff --git a/tools/stop_rx_watcher.cmd b/tools/stop_rx_watcher.cmd new file mode 100644 index 0000000..1c4e112 --- /dev/null +++ b/tools/stop_rx_watcher.cmd @@ -0,0 +1,8 @@ +@echo off +setlocal +cd /d "%~dp0.." +python -B tools\rx_agent_watcher.py --request-stop --stop-timeout 35 +set "RX_WATCHER_EXIT=%ERRORLEVEL%" +echo. +pause +exit /b %RX_WATCHER_EXIT%