798 lines
29 KiB
Python
798 lines
29 KiB
Python
"""Локальный 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]
|
||
if str(ROOT) not in sys.path:
|
||
sys.path.insert(0, str(ROOT))
|
||
|
||
from tools.rf_handoff_channel import ( # noqa: E402
|
||
EXPECTED_RX_TASK_ID,
|
||
GitResultChannel,
|
||
READY_STATE,
|
||
RxHandoffResult,
|
||
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"
|
||
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
|
||
last_approved_task_id: str | None = None
|
||
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)
|
||
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 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,
|
||
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
|
||
state.last_approved_task_id = task.task_id
|
||
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 publish_result(arguments: argparse.Namespace) -> int:
|
||
"""Publish only the approved small RX result to an existing branch."""
|
||
|
||
repository = arguments.repository.resolve()
|
||
task = parse_task((repository / DEFAULT_TASK_PATH).read_text(encoding="utf-8-sig"))
|
||
state = load_state(arguments.state_file.resolve())
|
||
result = load_result(arguments.publish_result.resolve())
|
||
require_live_ready_window(result)
|
||
if task.task_id != result.task_id:
|
||
raise ValueError("result TASK_ID disagrees with current RX task")
|
||
if task.allow_push != "RX_RESULT_ONLY" or task.rf_tx:
|
||
raise ValueError("current task does not allow RX result publication")
|
||
if state.last_approved_task_id != result.task_id:
|
||
raise ValueError("current RX task has not passed the local approval gate")
|
||
|
||
def git_output(arguments: list[str]) -> str:
|
||
return subprocess.run(
|
||
["git", *arguments],
|
||
cwd=repository,
|
||
capture_output=True,
|
||
text=True,
|
||
encoding="utf-8",
|
||
errors="replace",
|
||
check=True,
|
||
).stdout.strip()
|
||
|
||
branch = git_output(["branch", "--show-current"])
|
||
if branch != task.source_branch:
|
||
raise ValueError("current branch disagrees with task SOURCE_BRANCH")
|
||
if git_output(["status", "--porcelain", "--untracked-files=no"]):
|
||
raise ValueError("tracked worktree must be clean before result publication")
|
||
head = git_output(["rev-parse", "HEAD"])
|
||
remote_head = git_output(["rev-parse", f"{arguments.remote}/{task.source_branch}"])
|
||
if head != remote_head:
|
||
raise ValueError("current HEAD disagrees with the remote task branch")
|
||
if result.source_head != head:
|
||
raise ValueError("result SOURCE_HEAD disagrees with current checkout")
|
||
channel = GitResultChannel(
|
||
repository,
|
||
remote=arguments.remote,
|
||
branch=arguments.result_branch,
|
||
)
|
||
outcome = channel.publish(result)
|
||
print(f"RX RESULT PUBLISH = {outcome.status}")
|
||
print(f"RX RESULT HEAD = {outcome.commit}")
|
||
print(f"RX RESULT REMOTE HEAD = {outcome.remote_head}")
|
||
print("RF TX = 0")
|
||
return 0
|
||
|
||
|
||
def require_live_ready_window(
|
||
result: RxHandoffResult,
|
||
*,
|
||
now: datetime | None = None,
|
||
) -> None:
|
||
"""Запретить READY после окончания конечного окна захвата."""
|
||
|
||
if result.state != READY_STATE:
|
||
return
|
||
current = (now or datetime.now(timezone.utc)).astimezone(timezone.utc)
|
||
expected_end = datetime.fromisoformat(
|
||
result.expected_capture_end.replace("Z", "+00:00")
|
||
).astimezone(timezone.utc)
|
||
if current >= expected_end:
|
||
raise ValueError("READY capture window has already ended")
|
||
|
||
|
||
def verify_result_channel_write_access(
|
||
repository: Path,
|
||
*,
|
||
remote: str = "origin",
|
||
result_branch: str = "rx-node",
|
||
output_function: Callable[[str], None] = print,
|
||
) -> str:
|
||
"""Проверить право публикации результата RX без изменения ветки."""
|
||
|
||
channel = GitResultChannel(
|
||
repository,
|
||
remote=remote,
|
||
branch=result_branch,
|
||
)
|
||
remote_head = channel.verify_write_access()
|
||
output_function("RX-NODE WRITE PREFLIGHT = PASS")
|
||
output_function(f"RX-NODE REMOTE HEAD = {remote_head}")
|
||
output_function("RF TX = 0")
|
||
return remote_head
|
||
|
||
|
||
def run_watcher(arguments: argparse.Namespace) -> int:
|
||
state_path = arguments.state_file.resolve()
|
||
if arguments.publish_result is not None:
|
||
return publish_result(arguments)
|
||
if arguments.check_result_write:
|
||
verify_result_channel_write_access(
|
||
arguments.repository.resolve(),
|
||
remote=arguments.remote,
|
||
result_branch=arguments.result_branch,
|
||
)
|
||
return 0
|
||
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()
|
||
task = parse_task(task_text)
|
||
if task.task_id == EXPECTED_RX_TASK_ID and not arguments.dry_run:
|
||
if is_new_task(task, state):
|
||
verify_result_channel_write_access(
|
||
arguments.repository.resolve(),
|
||
remote=arguments.remote,
|
||
result_branch=arguments.result_branch,
|
||
)
|
||
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,
|
||
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")
|
||
parser.add_argument("--publish-result", type=Path)
|
||
parser.add_argument("--check-result-write", action="store_true")
|
||
parser.add_argument("--result-branch", default="rx-node")
|
||
parser.add_argument("--acknowledgement-branch", default="tx-node")
|
||
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()
|