737 lines
28 KiB
Python
737 lines
28 KiB
Python
"""Strict Git result channel for coordinated RX and MAIN PC handoff."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import asdict, dataclass, fields
|
||
from datetime import datetime, timezone
|
||
import hashlib
|
||
import json
|
||
from pathlib import Path
|
||
import re
|
||
import subprocess
|
||
import tempfile
|
||
from typing import Any, Mapping
|
||
|
||
|
||
SCHEMA_VERSION = 1
|
||
EXPECTED_RX_TASK_ID = "RF001A_OTA_CAPTURE_004"
|
||
EXPECTED_TX_TASK_ID = "RF001A_OTA_TX_004"
|
||
EXPECTED_CAPTURE_DURATION_SECONDS = 60.0
|
||
CHANNEL_TEST_TASK_ID = "RF_HANDOFF_CHANNEL_TEST_001"
|
||
READY_STATE = "RX_READY_FOR_TX"
|
||
COMPLETE_STATE = "CAPTURE_COMPLETE"
|
||
RESULT_JSON_PATH = "docs/handoff/results/RX_RESULT.json"
|
||
RESULT_MARKDOWN_PATH = "docs/handoff/results/RX_RESULT.md"
|
||
RESULT_PATHS = frozenset((RESULT_JSON_PATH, RESULT_MARKDOWN_PATH))
|
||
MAX_RESULT_BYTES = 64 * 1024
|
||
_SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
|
||
_HEAD_RE = re.compile(r"^[0-9a-f]{40}$")
|
||
_TRANSIENT_PUSH_ERRORS = (
|
||
"authentication failed",
|
||
"failed to authenticate",
|
||
"could not resolve host",
|
||
"connection reset",
|
||
"connection timed out",
|
||
"failed to connect",
|
||
"remote end hung up",
|
||
"rpc failed",
|
||
"schannel",
|
||
"ssl",
|
||
"tls",
|
||
"the requested url returned error: 5",
|
||
"unable to access",
|
||
)
|
||
|
||
|
||
class ResultChannelError(RuntimeError):
|
||
"""The result channel cannot be read or updated safely."""
|
||
|
||
|
||
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
|
||
task_id: str
|
||
state: str
|
||
rx_ready_for_tx: bool
|
||
process_alive: bool
|
||
file_growing: bool
|
||
capture_pid: int | None
|
||
published_at: str
|
||
capture_started_at: str
|
||
expected_capture_end: str
|
||
capture_path: str
|
||
common_clock_verified: bool
|
||
source_head: str
|
||
rf_tx: int
|
||
exit_code: int | None
|
||
file_size: int | None
|
||
duration_seconds: float | None
|
||
overruns: int | None
|
||
usb_errors: int | None
|
||
file_errors: int | None
|
||
sha256: str | None
|
||
clkin_after: str
|
||
rf_tx_observed: str
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ReadyEvaluation:
|
||
ready: bool
|
||
reason: str
|
||
seconds_remaining: float | None = None
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class PublishOutcome:
|
||
status: str
|
||
commit: str
|
||
remote_head: str
|
||
json_path: str = RESULT_JSON_PATH
|
||
markdown_path: str = RESULT_MARKDOWN_PATH
|
||
|
||
|
||
def _parse_utc(value: str, field_name: str) -> datetime:
|
||
try:
|
||
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
||
except (TypeError, ValueError) as error:
|
||
raise ValueError(f"{field_name} must be an ISO-8601 timestamp") from error
|
||
if parsed.tzinfo is None:
|
||
raise ValueError(f"{field_name} must contain a timezone")
|
||
return parsed.astimezone(timezone.utc)
|
||
|
||
|
||
def _validate_types(result: RxHandoffResult) -> None:
|
||
string_fields = (
|
||
result.task_id,
|
||
result.state,
|
||
result.published_at,
|
||
result.capture_started_at,
|
||
result.expected_capture_end,
|
||
result.capture_path,
|
||
result.source_head,
|
||
result.clkin_after,
|
||
result.rf_tx_observed,
|
||
)
|
||
if any(type(value) is not str for value in string_fields):
|
||
raise ValueError("string result fields must contain strings")
|
||
boolean_fields = (
|
||
result.rx_ready_for_tx,
|
||
result.process_alive,
|
||
result.file_growing,
|
||
result.common_clock_verified,
|
||
)
|
||
if any(type(value) is not bool for value in boolean_fields):
|
||
raise ValueError("boolean result fields must contain booleans")
|
||
if type(result.schema_version) is not int or type(result.rf_tx) is not int:
|
||
raise ValueError("schema_version and rf_tx must contain integers")
|
||
optional_integer_fields = (
|
||
result.capture_pid,
|
||
result.exit_code,
|
||
result.file_size,
|
||
result.overruns,
|
||
result.usb_errors,
|
||
result.file_errors,
|
||
)
|
||
if any(
|
||
value is not None and type(value) is not int
|
||
for value in optional_integer_fields
|
||
):
|
||
raise ValueError("integer result fields must contain integers or null")
|
||
if (
|
||
result.duration_seconds is not None
|
||
and (
|
||
isinstance(result.duration_seconds, bool)
|
||
or not isinstance(result.duration_seconds, (int, float))
|
||
)
|
||
):
|
||
raise ValueError("duration_seconds must contain a number or null")
|
||
if result.sha256 is not None and type(result.sha256) is not str:
|
||
raise ValueError("sha256 must contain a string or null")
|
||
|
||
|
||
def _validate_common(result: RxHandoffResult) -> None:
|
||
_validate_types(result)
|
||
if result.schema_version != SCHEMA_VERSION:
|
||
raise ValueError(f"schema_version must be {SCHEMA_VERSION}")
|
||
if result.state not in {READY_STATE, COMPLETE_STATE}:
|
||
raise ValueError("state is not supported")
|
||
if result.rf_tx != 0:
|
||
raise ValueError("rf_tx must remain 0 on the RX node")
|
||
if not result.task_id or any(character in result.task_id for character in "\r\n"):
|
||
raise ValueError("task_id is invalid")
|
||
if not _HEAD_RE.fullmatch(result.source_head):
|
||
raise ValueError("source_head must be a lowercase 40-character Git hash")
|
||
if not result.capture_path or any(
|
||
character in result.capture_path for character in "\r\n"
|
||
):
|
||
raise ValueError("capture_path is invalid")
|
||
published = _parse_utc(result.published_at, "published_at")
|
||
started = _parse_utc(result.capture_started_at, "capture_started_at")
|
||
expected_end = _parse_utc(result.expected_capture_end, "expected_capture_end")
|
||
if published < started:
|
||
raise ValueError("published_at cannot precede capture_started_at")
|
||
if expected_end <= started:
|
||
raise ValueError("expected_capture_end must follow capture_started_at")
|
||
if result.clkin_after not in {"DETECTED", "NOT_DETECTED", "NOT_RUN"}:
|
||
raise ValueError("clkin_after is invalid")
|
||
if result.rf_tx_observed not in {
|
||
"EXPECTED_EXTERNAL_TX",
|
||
"UNKNOWN",
|
||
}:
|
||
raise ValueError("rf_tx_observed is invalid")
|
||
|
||
|
||
def _validate_ready(result: RxHandoffResult) -> None:
|
||
if not (
|
||
result.rx_ready_for_tx
|
||
and result.process_alive
|
||
and result.file_growing
|
||
):
|
||
raise ValueError("READY requires a live process and growing capture")
|
||
if result.task_id == CHANNEL_TEST_TASK_ID:
|
||
if result.common_clock_verified:
|
||
raise ValueError("channel smoke READY must not claim hardware clock proof")
|
||
elif not result.common_clock_verified:
|
||
raise ValueError("production READY requires verified common clock")
|
||
if result.capture_pid is None or result.capture_pid <= 0:
|
||
raise ValueError("READY requires a positive capture_pid")
|
||
final_fields = (
|
||
result.exit_code,
|
||
result.file_size,
|
||
result.duration_seconds,
|
||
result.overruns,
|
||
result.usb_errors,
|
||
result.file_errors,
|
||
result.sha256,
|
||
)
|
||
if any(value is not None for value in final_fields):
|
||
raise ValueError("READY must not contain final capture evidence")
|
||
|
||
|
||
def _validate_complete(result: RxHandoffResult) -> None:
|
||
if result.rx_ready_for_tx or result.process_alive or result.file_growing:
|
||
raise ValueError("COMPLETE cannot advertise a live capture")
|
||
if result.capture_pid is None or result.capture_pid <= 0:
|
||
raise ValueError("COMPLETE requires the capture_pid")
|
||
if result.exit_code is None or result.file_size is None:
|
||
raise ValueError("COMPLETE requires exit_code and file_size")
|
||
if result.file_size < 0:
|
||
raise ValueError("file_size cannot be negative")
|
||
counters = (result.overruns, result.usb_errors, result.file_errors)
|
||
if any(value is not None and value < 0 for value in counters):
|
||
raise ValueError("error counters cannot be negative")
|
||
if result.duration_seconds is None or result.duration_seconds < 0:
|
||
raise ValueError("COMPLETE requires a non-negative duration_seconds")
|
||
if result.sha256 is None or not _SHA256_RE.fullmatch(result.sha256):
|
||
raise ValueError("COMPLETE requires a lowercase SHA256")
|
||
|
||
|
||
def validate_result(result: RxHandoffResult) -> RxHandoffResult:
|
||
_validate_common(result)
|
||
if result.state == READY_STATE:
|
||
_validate_ready(result)
|
||
else:
|
||
_validate_complete(result)
|
||
return result
|
||
|
||
|
||
def result_from_mapping(raw: Mapping[str, Any]) -> RxHandoffResult:
|
||
expected_fields = {field.name for field in fields(RxHandoffResult)}
|
||
actual_fields = set(raw)
|
||
if actual_fields != expected_fields:
|
||
missing = sorted(expected_fields - actual_fields)
|
||
extra = sorted(actual_fields - expected_fields)
|
||
raise ValueError(f"result fields disagree: missing={missing}, extra={extra}")
|
||
result = RxHandoffResult(**{name: raw[name] for name in expected_fields})
|
||
return validate_result(result)
|
||
|
||
|
||
def load_result(path: Path) -> RxHandoffResult:
|
||
raw = json.loads(path.read_text(encoding="utf-8"))
|
||
if not isinstance(raw, dict):
|
||
raise ValueError("result JSON root must be an object")
|
||
return result_from_mapping(raw)
|
||
|
||
|
||
def canonical_result_json(result: RxHandoffResult) -> str:
|
||
validate_result(result)
|
||
return json.dumps(
|
||
asdict(result),
|
||
indent=2,
|
||
sort_keys=True,
|
||
ensure_ascii=False,
|
||
) + "\n"
|
||
|
||
|
||
def result_digest(result: RxHandoffResult) -> str:
|
||
return hashlib.sha256(canonical_result_json(result).encode("utf-8")).hexdigest()
|
||
|
||
|
||
def render_result_markdown(result: RxHandoffResult) -> str:
|
||
validate_result(result)
|
||
rows = [
|
||
"# RX result",
|
||
"",
|
||
f"SCHEMA_VERSION = {result.schema_version}",
|
||
f"TASK_ID = {result.task_id}",
|
||
f"STATE = {result.state}",
|
||
f"RX_READY_FOR_TX = {'YES' if result.rx_ready_for_tx else 'NO'}",
|
||
f"PROCESS_ALIVE = {'YES' if result.process_alive else 'NO'}",
|
||
f"FILE_GROWING = {'YES' if result.file_growing else 'NO'}",
|
||
f"CAPTURE_PID = {result.capture_pid}",
|
||
f"PUBLISHED_AT = {result.published_at}",
|
||
f"CAPTURE_STARTED_AT = {result.capture_started_at}",
|
||
f"EXPECTED_CAPTURE_END = {result.expected_capture_end}",
|
||
f"CAPTURE_PATH = {result.capture_path}",
|
||
f"COMMON_CLOCK_VERIFIED = {'YES' if result.common_clock_verified else 'NO'}",
|
||
f"SOURCE_HEAD = {result.source_head}",
|
||
f"EXIT_CODE = {result.exit_code}",
|
||
f"FILE_SIZE = {result.file_size}",
|
||
f"DURATION_SECONDS = {result.duration_seconds}",
|
||
f"OVERRUNS = {result.overruns}",
|
||
f"USB_ERRORS = {result.usb_errors}",
|
||
f"FILE_ERRORS = {result.file_errors}",
|
||
f"SHA256 = {result.sha256}",
|
||
f"CLKIN_AFTER = {result.clkin_after}",
|
||
f"RF_TX_OBSERVED = {result.rf_tx_observed}",
|
||
f"RF_TX = {result.rf_tx}",
|
||
"",
|
||
"Raw IQ, video, waveforms, and large logs are not stored in this branch.",
|
||
"",
|
||
]
|
||
return "\n".join(rows)
|
||
|
||
|
||
def evaluate_ready(
|
||
result: RxHandoffResult,
|
||
*,
|
||
now: datetime,
|
||
expected_task_id: str = EXPECTED_RX_TASK_ID,
|
||
expected_source_head: str | None = None,
|
||
max_age_seconds: float = 30.0,
|
||
min_remaining_seconds: float = 10.0,
|
||
) -> ReadyEvaluation:
|
||
validate_result(result)
|
||
if now.tzinfo is None:
|
||
raise ValueError("now must contain a timezone")
|
||
now_utc = now.astimezone(timezone.utc)
|
||
if result.task_id != expected_task_id:
|
||
return ReadyEvaluation(False, "WRONG_TASK_ID")
|
||
if expected_source_head is not None and result.source_head != expected_source_head:
|
||
return ReadyEvaluation(False, "WRONG_SOURCE_HEAD")
|
||
if result.state != READY_STATE or not result.rx_ready_for_tx:
|
||
return ReadyEvaluation(False, "MISSING_READY")
|
||
if not result.common_clock_verified:
|
||
return ReadyEvaluation(False, "COMMON_CLOCK_NOT_VERIFIED")
|
||
started = _parse_utc(result.capture_started_at, "capture_started_at")
|
||
expected_end = _parse_utc(result.expected_capture_end, "expected_capture_end")
|
||
published = _parse_utc(result.published_at, "published_at")
|
||
planned_duration = (expected_end - started).total_seconds()
|
||
if not (
|
||
EXPECTED_CAPTURE_DURATION_SECONDS - 1.0
|
||
<= planned_duration
|
||
<= EXPECTED_CAPTURE_DURATION_SECONDS + 1.0
|
||
):
|
||
return ReadyEvaluation(False, "WRONG_CAPTURE_WINDOW")
|
||
age = (now_utc - published).total_seconds()
|
||
if age < -5.0:
|
||
return ReadyEvaluation(False, "FUTURE_READY")
|
||
if age > max_age_seconds:
|
||
return ReadyEvaluation(False, "STALE_READY")
|
||
remaining = (expected_end - now_utc).total_seconds()
|
||
if remaining < min_remaining_seconds:
|
||
return ReadyEvaluation(False, "CAPTURE_WINDOW_EXPIRED", remaining)
|
||
return ReadyEvaluation(True, "READY", remaining)
|
||
|
||
|
||
class GitResultChannel:
|
||
"""Publish through a temporary worktree of the configured main repository."""
|
||
|
||
def __init__(
|
||
self,
|
||
repository: Path,
|
||
*,
|
||
remote: str = "origin",
|
||
branch: str = "rx-node",
|
||
) -> None:
|
||
self.repository = repository
|
||
self.remote = remote
|
||
self.branch = branch
|
||
|
||
def _run(
|
||
self,
|
||
arguments: list[str],
|
||
*,
|
||
cwd: Path | None = None,
|
||
check: bool = True,
|
||
) -> subprocess.CompletedProcess[str]:
|
||
return subprocess.run(
|
||
["git", *arguments],
|
||
cwd=cwd or self.repository,
|
||
capture_output=True,
|
||
text=True,
|
||
encoding="utf-8",
|
||
errors="replace",
|
||
check=check,
|
||
)
|
||
|
||
def remote_head(self) -> str | None:
|
||
result = self._run(
|
||
["ls-remote", "--heads", self.remote, self.branch],
|
||
check=False,
|
||
)
|
||
if result.returncode != 0:
|
||
raise ResultChannelError(
|
||
f"git ls-remote failed for {self.remote}/{self.branch}: "
|
||
f"{result.stderr.strip()}"
|
||
)
|
||
lines = [line for line in result.stdout.splitlines() if line.strip()]
|
||
if not lines:
|
||
return None
|
||
if len(lines) != 1:
|
||
raise ResultChannelError("remote result branch lookup is ambiguous")
|
||
head, reference = lines[0].split(maxsplit=1)
|
||
if reference != f"refs/heads/{self.branch}" or not _HEAD_RE.fullmatch(head):
|
||
raise ResultChannelError("remote result branch returned an invalid HEAD")
|
||
return head
|
||
|
||
def branch_exists(self) -> bool:
|
||
return self.remote_head() is not None
|
||
|
||
def verify_write_access(self) -> str:
|
||
"""Проверить право записи dry-run push без коммита и изменения ветки."""
|
||
|
||
remote_head = self.fetch()
|
||
dry_run = self._run(
|
||
[
|
||
"push",
|
||
"--dry-run",
|
||
"--porcelain",
|
||
self.remote,
|
||
f"{remote_head}:refs/heads/{self.branch}",
|
||
],
|
||
check=False,
|
||
)
|
||
if dry_run.returncode != 0:
|
||
raise ResultChannelError(
|
||
"authenticated result-branch push preflight failed: "
|
||
f"{dry_run.stderr.strip()}"
|
||
)
|
||
confirmed_head = self.remote_head()
|
||
if confirmed_head != remote_head:
|
||
raise ResultChannelError(
|
||
"result branch changed during write preflight"
|
||
)
|
||
return remote_head
|
||
|
||
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],
|
||
check=False,
|
||
)
|
||
remote_head = self.remote_head()
|
||
if remote_head is None:
|
||
raise ResultBranchMissing(
|
||
f"remote branch {self.remote}/{self.branch} does not exist"
|
||
)
|
||
if initial_fetch.returncode != 0:
|
||
raise ResultChannelError(
|
||
f"git fetch failed for {self.remote}/{self.branch}: "
|
||
f"{initial_fetch.stderr.strip()}"
|
||
)
|
||
tracking_reference = (
|
||
f"refs/heads/{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,
|
||
)
|
||
if tracking_fetch.returncode != 0:
|
||
raise ResultChannelError(
|
||
f"tracking fetch failed for {self.remote}/{self.branch}: "
|
||
f"{tracking_fetch.stderr.strip()}"
|
||
)
|
||
tracking_head = self._run(
|
||
["rev-parse", self._tracking_reference()]
|
||
).stdout.strip()
|
||
if tracking_head != remote_head:
|
||
raise ResultChannelError("remote-tracking result branch is stale")
|
||
return remote_head
|
||
|
||
def read(self) -> tuple[RxHandoffResult, str, str]:
|
||
remote_head = self.fetch()
|
||
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(
|
||
f"result JSON root is not an object at commit {commit}"
|
||
)
|
||
result = result_from_mapping(raw)
|
||
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
|
||
|
||
@staticmethod
|
||
def _is_transient_push_error(stderr: str) -> bool:
|
||
normalized = stderr.casefold()
|
||
return any(pattern in normalized for pattern in _TRANSIENT_PUSH_ERRORS)
|
||
|
||
def _push_exact_commit(
|
||
self,
|
||
checkout: Path,
|
||
commit: str,
|
||
remote_before: str,
|
||
) -> str:
|
||
"""Отправить неизменный коммит с одной безопасной повторной попыткой."""
|
||
|
||
push_arguments = [
|
||
"push",
|
||
"--quiet",
|
||
self.remote,
|
||
f"{commit}:refs/heads/{self.branch}",
|
||
]
|
||
first = self._run(push_arguments, cwd=checkout, check=False)
|
||
if first.returncode == 0:
|
||
remote_after = self.remote_head()
|
||
if remote_after != commit:
|
||
raise ResultChannelError(
|
||
"remote result HEAD does not confirm the published commit"
|
||
)
|
||
return commit
|
||
|
||
observed_after_failure = self.remote_head()
|
||
if observed_after_failure == commit:
|
||
return commit
|
||
if observed_after_failure != remote_before:
|
||
raise ResultChannelError(
|
||
"result push failed and remote branch changed to a conflicting "
|
||
f"commit: {remote_before} -> {observed_after_failure}"
|
||
)
|
||
if not self._is_transient_push_error(first.stderr):
|
||
raise ResultChannelError(f"result push failed: {first.stderr.strip()}")
|
||
|
||
fetched_head = self.fetch()
|
||
if fetched_head != remote_before:
|
||
raise ResultChannelError(
|
||
"remote result branch changed before the single retry"
|
||
)
|
||
second = self._run(push_arguments, cwd=checkout, check=False)
|
||
if second.returncode != 0:
|
||
observed_after_retry = self.remote_head()
|
||
if observed_after_retry == commit:
|
||
return commit
|
||
raise ResultChannelError(
|
||
"result push failed after one retry of the same commit: "
|
||
f"{second.stderr.strip()}"
|
||
)
|
||
remote_after = self.remote_head()
|
||
if remote_after != commit:
|
||
raise ResultChannelError(
|
||
"remote result HEAD does not confirm the retried commit"
|
||
)
|
||
return commit
|
||
|
||
def publish(self, result: RxHandoffResult) -> PublishOutcome:
|
||
validate_result(result)
|
||
remote_before = self.fetch()
|
||
tracking_reference = f"refs/remotes/{self.remote}/{self.branch}"
|
||
with tempfile.TemporaryDirectory(prefix="rx-result-worktree-") as directory:
|
||
checkout = Path(directory) / "rx-node"
|
||
worktree_added = False
|
||
try:
|
||
self._run(
|
||
["worktree", "add", "--detach", str(checkout), tracking_reference]
|
||
)
|
||
worktree_added = True
|
||
json_path = checkout / RESULT_JSON_PATH
|
||
markdown_path = checkout / RESULT_MARKDOWN_PATH
|
||
json_path.parent.mkdir(parents=True, exist_ok=True)
|
||
json_text = canonical_result_json(result)
|
||
markdown_text = render_result_markdown(result)
|
||
if len(json_text.encode("utf-8")) > MAX_RESULT_BYTES:
|
||
raise ResultChannelError("result JSON exceeds the size limit")
|
||
if len(markdown_text.encode("utf-8")) > MAX_RESULT_BYTES:
|
||
raise ResultChannelError("result Markdown exceeds the size limit")
|
||
json_path.write_text(json_text, encoding="utf-8", newline="\n")
|
||
markdown_path.write_text(
|
||
markdown_text,
|
||
encoding="utf-8",
|
||
newline="\n",
|
||
)
|
||
self._run(["add", "--", *sorted(RESULT_PATHS)], cwd=checkout)
|
||
staged_text = self._run(
|
||
["diff", "--cached", "--name-only", "--"],
|
||
cwd=checkout,
|
||
).stdout
|
||
staged = {
|
||
line.strip().replace("\\", "/")
|
||
for line in staged_text.splitlines()
|
||
if line.strip()
|
||
}
|
||
if not staged:
|
||
return PublishOutcome("UNCHANGED", remote_before, remote_before)
|
||
if staged != RESULT_PATHS:
|
||
raise ResultChannelError(
|
||
f"refusing unexpected staged paths: {sorted(staged)}"
|
||
)
|
||
self._run(
|
||
[
|
||
"-c",
|
||
"user.name=RX Node",
|
||
"-c",
|
||
"user.email=rx-node@local.invalid",
|
||
"commit",
|
||
"--quiet",
|
||
"-m",
|
||
f"Publish {result.task_id} {result.state}",
|
||
],
|
||
cwd=checkout,
|
||
)
|
||
commit = self._run(["rev-parse", "HEAD"], cwd=checkout).stdout.strip()
|
||
remote_after = self._push_exact_commit(
|
||
checkout,
|
||
commit,
|
||
remote_before,
|
||
)
|
||
return PublishOutcome("PUBLISHED", commit, remote_after)
|
||
finally:
|
||
if worktree_added:
|
||
removal = self._run(
|
||
["worktree", "remove", "--force", str(checkout)],
|
||
check=False,
|
||
)
|
||
if removal.returncode != 0:
|
||
raise ResultChannelError(
|
||
f"temporary worktree cleanup failed: "
|
||
f"{removal.stderr.strip()}"
|
||
)
|
||
self._run(["worktree", "prune"], check=False)
|