#!/usr/bin/env python3
|
"""Durable half-hour coordinator for one configured Bilibili creator.
|
|
The coordinator never reads browser credentials and never downloads media. It
|
turns already validated collector/formal records into exact-once role outboxes,
|
tracks non-overlapping half-hour runs, accepts exact downstream receipts, and
|
performs narrowly allowlisted Git delivery. Browser capture and native media
|
work remain in their reviewed components.
|
"""
|
|
from __future__ import annotations
|
|
import argparse
|
import contextlib
|
import hashlib
|
import json
|
import os
|
import re
|
import stat
|
import subprocess
|
import sys
|
import tempfile
|
from dataclasses import dataclass
|
from datetime import datetime, timezone
|
from pathlib import Path
|
from typing import Any, Callable, Iterable, Iterator, Mapping, Sequence
|
from urllib.parse import urlsplit
|
|
|
SCHEMA = 1
|
INTERVAL_MINUTES = 30
|
TASK_ID = "DEV-PROJECT-INFO-BILI-AUTHENTICATED-SESSION-DOWNLOAD-20260805-001"
|
UID = re.compile(r"^[1-9][0-9]{0,19}$")
|
BVID = re.compile(r"^BV1[1-9A-HJ-NP-Za-km-z]{9}$")
|
SHA256 = re.compile(r"^[0-9A-F]{64}$")
|
SHA256_MIXED_ASCII = re.compile(r"^[0-9A-Fa-f]{64}$", re.ASCII)
|
THREAD_ID = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$")
|
SAFE_ID = re.compile(r"^[A-Za-z0-9_.:-]{1,240}$")
|
SECRET_KEY = re.compile(r"(?i)(cookie|sessdata|token|credential|authorization|localstorage|profile|signed_url)")
|
SECRET_VALUE = re.compile(r"(?i)(SYNTHETIC_SECRET|sessdata=|cookie\s*[:=]|authorization\s*[:=]|bearer\s+|token=|signed_url=|localstorage)")
|
FORBIDDEN_GIT_SUFFIXES = (
|
".mkv", ".mp4", ".mov", ".webm", ".download.json", ".flac", ".partial", ".crdownload"
|
)
|
RECEIPT_GIT_KIND_SUFFIX = {
|
"transcript": ".txt",
|
"transcript_txt": ".txt",
|
"transcript_srt": ".srt",
|
"transcript_json": ".json",
|
"minutes": ".md",
|
"minutes_md": ".md",
|
"minutes_pdf": ".pdf",
|
"relocation_manifest": ".json",
|
"documentation": ".md",
|
}
|
CONTENT_TYPES = frozenset({"article", "text", "image"})
|
VIDEO_COMPLETE = "VIDEO_DOWNLOADED_COMPLETE_HANDOFF_SENT"
|
CANONICAL_TITLE_MAX_LENGTH = 64
|
WINDOWS_INVALID_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
|
WINDOWS_RESERVED_NAMES = {
|
"CON", "PRN", "AUX", "NUL",
|
*(f"COM{value}" for value in range(1, 10)),
|
*(f"LPT{value}" for value in range(1, 10)),
|
}
|
RELOCATION_REPORT_TYPE = "VIDEO_ARTIFACT_RELOCATION_BATCH"
|
PUBLIC_VIDEO_KINDS = {
|
"transcript_txt": ".txt",
|
"transcript_srt": ".srt",
|
"transcript_json": ".json",
|
"minutes_md": ".md",
|
"minutes_pdf": ".pdf",
|
}
|
|
|
class PipelineError(RuntimeError):
|
def __init__(self, code: str, message: str) -> None:
|
super().__init__(message)
|
self.code = code
|
|
|
@dataclass(frozen=True)
|
class Config:
|
path: Path
|
project_root: Path
|
creator_uid: str
|
creator_name: str
|
dynamic_url: str
|
archive_root: Path
|
formal_manifest: Path
|
processing_handoffs: Path
|
state_dir: Path
|
video_root: Path
|
video_downloader_thread_id: str
|
media_thread_id: str
|
minutes_thread_id: str
|
reply_thread_id: str
|
git_remote: str
|
git_branch: str
|
git_extensions: frozenset[str]
|
git_doc_paths: frozenset[str]
|
|
@property
|
def state_path(self) -> Path:
|
return self.state_dir / "state.json"
|
|
@property
|
def runs_path(self) -> Path:
|
return self.state_dir / "runs.jsonl"
|
|
@property
|
def outbox_path(self) -> Path:
|
return self.state_dir / "outbox.jsonl"
|
|
@property
|
def terminals_path(self) -> Path:
|
return self.state_dir / "terminals.jsonl"
|
|
@property
|
def lock_path(self) -> Path:
|
return self.state_dir / "coordinator.lock"
|
|
def git_index_guard_path(self, batch_id: str) -> Path:
|
return self.state_dir / f"git-shared-index-guard-{batch_id}.json"
|
|
@property
|
def relocation_root(self) -> Path:
|
return self.archive_root / "artifact-relocations"
|
|
|
def _exact(value: Any, keys: Iterable[str], field: str) -> Mapping[str, Any]:
|
expected = set(keys)
|
if not isinstance(value, dict) or set(value) != expected:
|
raise PipelineError("E_SCHEMA", f"{field} keys differ")
|
return value
|
|
|
def _reject_secrets(value: Any, path: str = "$") -> None:
|
if isinstance(value, dict):
|
for key, item in value.items():
|
if not isinstance(key, str) or SECRET_KEY.search(key):
|
raise PipelineError("E_SECRET_FIELD", f"secret-like field at {path}")
|
_reject_secrets(item, f"{path}.{key}")
|
elif isinstance(value, list):
|
for index, item in enumerate(value):
|
_reject_secrets(item, f"{path}[{index}]")
|
elif isinstance(value, str):
|
if SECRET_VALUE.search(value):
|
raise PipelineError("E_SECRET_FIELD", f"secret-like value at {path}")
|
parsed = urlsplit(value)
|
if parsed.scheme in {"http", "https"} and (parsed.username is not None or parsed.password is not None):
|
raise PipelineError("E_SECRET_FIELD", f"credential-bearing URL at {path}")
|
|
|
def _canonical(value: Mapping[str, Any]) -> bytes:
|
_reject_secrets(value)
|
return (json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":")) + "\n").encode("ascii")
|
|
|
def _strict_json(path: Path, field: str) -> tuple[Any, bytes]:
|
try:
|
payload = path.read_bytes()
|
except OSError as exc:
|
raise PipelineError("E_INPUT", f"{field} is unavailable") from exc
|
if not payload or payload.startswith(b"\xef\xbb\xbf") or b"\r" in payload or not payload.endswith(b"\n"):
|
raise PipelineError("E_INPUT", f"{field} is not strict UTF-8 JSON")
|
try:
|
text = payload.decode("utf-8")
|
value = json.loads(text)
|
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
raise PipelineError("E_INPUT", f"{field} is invalid JSON") from exc
|
_reject_secrets(value)
|
return value, payload
|
|
|
def _absolute(base: Path, value: Any, field: str) -> Path:
|
if not isinstance(value, str) or not value:
|
raise PipelineError("E_CONFIG", f"{field} must be a path")
|
candidate = Path(value)
|
if not candidate.is_absolute():
|
candidate = base / candidate
|
return Path(os.path.abspath(candidate))
|
|
|
def _within(child: Path, parent: Path) -> bool:
|
try:
|
child.relative_to(parent)
|
return True
|
except ValueError:
|
return False
|
|
|
def _canonical_dynamic_url(value: Any, uid: str) -> str:
|
if not isinstance(value, str):
|
raise PipelineError("E_CONFIG", "dynamic_url must be a string")
|
parsed = urlsplit(value)
|
if parsed.scheme != "https" or parsed.hostname != "space.bilibili.com" or parsed.query or parsed.fragment or parsed.path.rstrip("/") != f"/{uid}/dynamic":
|
raise PipelineError("E_CONFIG", "dynamic_url is not canonical")
|
return f"https://space.bilibili.com/{uid}/dynamic"
|
|
|
def _canonical_video_url(value: Any, bvid: str) -> str:
|
if not isinstance(value, str) or not BVID.fullmatch(bvid):
|
raise PipelineError("E_SOURCE_BINDING", "video source identity is incomplete")
|
_reject_secrets(value)
|
parsed = urlsplit(value)
|
if (
|
parsed.scheme != "https"
|
or parsed.hostname != "www.bilibili.com"
|
or parsed.username is not None
|
or parsed.password is not None
|
or parsed.port is not None
|
or parsed.query
|
or parsed.fragment
|
or parsed.path.rstrip("/") != f"/video/{bvid}"
|
):
|
raise PipelineError("E_SOURCE_BINDING", "video source URL is not canonical")
|
return f"https://www.bilibili.com/video/{bvid}"
|
|
|
def load_config(path: Path) -> Config:
|
value, _ = _strict_json(path, "config")
|
root = _exact(value, {"schema_version", "task_id", "interval_minutes", "creator", "paths", "downstream", "git"}, "config")
|
if (
|
type(root["schema_version"]) is not int
|
or root["schema_version"] != SCHEMA
|
or root["task_id"] != TASK_ID
|
or type(root["interval_minutes"]) is not int
|
or root["interval_minutes"] != INTERVAL_MINUTES
|
):
|
raise PipelineError("E_CONFIG", "config identity differs")
|
creator = _exact(root["creator"], {"uid", "name", "dynamic_url"}, "creator")
|
uid = creator["uid"]
|
if not isinstance(uid, str) or not UID.fullmatch(uid) or not isinstance(creator["name"], str) or not creator["name"].strip():
|
raise PipelineError("E_CONFIG", "creator identity differs")
|
paths = _exact(root["paths"], {"project_root", "archive_root", "formal_manifest", "processing_handoffs", "state_dir", "video_root"}, "paths")
|
base = path.parent
|
project_root = _absolute(base, paths["project_root"], "project_root")
|
archive_root = _absolute(project_root, paths["archive_root"], "archive_root")
|
formal = _absolute(project_root, paths["formal_manifest"], "formal_manifest")
|
handoffs = _absolute(project_root, paths["processing_handoffs"], "processing_handoffs")
|
state_dir = _absolute(project_root, paths["state_dir"], "state_dir")
|
video_root = _absolute(project_root, paths["video_root"], "video_root")
|
if not _within(archive_root, project_root) or not _within(formal, archive_root) or not _within(handoffs, archive_root) or not _within(state_dir, project_root):
|
raise PipelineError("E_CONFIG", "project output path escaped its governed root")
|
downstream = _exact(
|
root["downstream"],
|
{"video_downloader_thread_id", "media_processor_thread_id", "minutes_thread_id", "reply_thread_id"},
|
"downstream",
|
)
|
for field, thread_id in downstream.items():
|
if not isinstance(thread_id, str) or not THREAD_ID.fullmatch(thread_id):
|
raise PipelineError("E_CONFIG", f"{field} is not a thread id")
|
git = _exact(root["git"], {"remote", "branch", "allowed_extensions", "allowed_docs"}, "git")
|
if (
|
git["remote"] != "origin"
|
or not isinstance(git["branch"], str)
|
or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._/-]{0,199}", git["branch"])
|
or git["branch"].endswith(("/", "."))
|
or any(marker in git["branch"] for marker in ("..", "//", "@{"))
|
):
|
raise PipelineError("E_CONFIG", "git destination differs")
|
extensions = git["allowed_extensions"]
|
if not isinstance(extensions, list) or not extensions or any(not isinstance(item, str) or not item.startswith(".") or item.lower() in FORBIDDEN_GIT_SUFFIXES for item in extensions):
|
raise PipelineError("E_CONFIG", "git extension allowlist is invalid")
|
docs = git["allowed_docs"]
|
if not isinstance(docs, list) or not docs:
|
raise PipelineError("E_CONFIG", "git document allowlist is invalid")
|
normalized_docs: list[str] = []
|
for value in docs:
|
if not isinstance(value, str) or not value or Path(value).is_absolute():
|
raise PipelineError("E_CONFIG", "git document allowlist is invalid")
|
target = Path(os.path.abspath(project_root / Path(value)))
|
if not _within(target, project_root) or target.suffix.lower() != ".md":
|
raise PipelineError("E_CONFIG", "git document allowlist escaped the project")
|
normalized_docs.append(target.relative_to(project_root).as_posix())
|
if len(set(normalized_docs)) != len(normalized_docs):
|
raise PipelineError("E_CONFIG", "git document allowlist is duplicated")
|
return Config(
|
path=path, project_root=project_root, creator_uid=uid, creator_name=creator["name"].strip(),
|
dynamic_url=_canonical_dynamic_url(creator["dynamic_url"], uid), archive_root=archive_root,
|
formal_manifest=formal, processing_handoffs=handoffs, state_dir=state_dir, video_root=video_root,
|
video_downloader_thread_id=downstream["video_downloader_thread_id"],
|
media_thread_id=downstream["media_processor_thread_id"], minutes_thread_id=downstream["minutes_thread_id"],
|
reply_thread_id=downstream["reply_thread_id"], git_remote=git["remote"], git_branch=git["branch"],
|
git_extensions=frozenset(item.lower() for item in extensions), git_doc_paths=frozenset(normalized_docs),
|
)
|
|
|
def _file_identity(path: Path) -> dict[str, Any]:
|
payload = path.read_bytes() if path.exists() else b""
|
if payload and not payload.endswith(b"\n"):
|
raise PipelineError("E_JOURNAL", f"{path.name} lacks final LF")
|
return {"bytes": len(payload), "lines": payload.count(b"\n"), "sha256": hashlib.sha256(payload).hexdigest().upper()}
|
|
|
def _atomic(path: Path, payload: bytes) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
fd, raw = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".partial", dir=path.parent)
|
partial = Path(raw)
|
try:
|
with os.fdopen(fd, "wb") as stream:
|
stream.write(payload)
|
stream.flush()
|
os.fsync(stream.fileno())
|
os.replace(partial, path)
|
if path.read_bytes() != payload:
|
raise PipelineError("E_DURABILITY", f"{path.name} readback differs")
|
finally:
|
with contextlib.suppress(FileNotFoundError):
|
partial.unlink()
|
|
|
def _append(path: Path, value: Mapping[str, Any]) -> dict[str, Any]:
|
payload = _canonical(value)
|
pre = path.read_bytes() if path.exists() else b""
|
if pre and not pre.endswith(b"\n"):
|
raise PipelineError("E_JOURNAL", f"{path.name} is malformed")
|
_atomic(path, pre + payload)
|
if not path.read_bytes().startswith(pre):
|
raise PipelineError("E_DURABILITY", f"{path.name} prefix changed")
|
return _file_identity(path)
|
|
|
def _create_new(path: Path, payload: bytes) -> None:
|
path.parent.mkdir(parents=True, exist_ok=True)
|
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0)
|
try:
|
descriptor = os.open(path, flags, 0o600)
|
except FileExistsError as exc:
|
raise PipelineError("E_DURABILITY", f"{path.name} already exists") from exc
|
try:
|
with os.fdopen(descriptor, "wb", closefd=False) as stream:
|
stream.write(payload)
|
stream.flush()
|
os.fsync(stream.fileno())
|
finally:
|
os.close(descriptor)
|
if path.read_bytes() != payload:
|
raise PipelineError("E_DURABILITY", f"{path.name} readback differs")
|
|
|
@contextlib.contextmanager
|
def _lock(config: Config) -> Iterator[None]:
|
config.state_dir.mkdir(parents=True, exist_ok=True)
|
stream = config.lock_path.open("a+b")
|
try:
|
if os.name == "nt":
|
import msvcrt
|
stream.seek(0)
|
if stream.tell() == stream.seek(0, os.SEEK_END) == 0:
|
stream.write(b"0")
|
stream.flush()
|
stream.seek(0)
|
try:
|
msvcrt.locking(stream.fileno(), msvcrt.LK_NBLCK, 1)
|
except OSError as exc:
|
raise PipelineError("E_RUN_ACTIVE", "another coordinator owns the run lock") from exc
|
else:
|
import fcntl
|
try:
|
fcntl.flock(stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
except OSError as exc:
|
raise PipelineError("E_RUN_ACTIVE", "another coordinator owns the run lock") from exc
|
yield
|
finally:
|
if os.name == "nt":
|
with contextlib.suppress(OSError):
|
stream.seek(0)
|
msvcrt.locking(stream.fileno(), msvcrt.LK_UNLCK, 1)
|
else:
|
with contextlib.suppress(OSError):
|
fcntl.flock(stream.fileno(), fcntl.LOCK_UN)
|
stream.close()
|
|
|
def _read_lines(
|
path: Path,
|
field: str,
|
*,
|
reject_secrets: bool = True,
|
) -> tuple[list[dict[str, Any]], bytes]:
|
if not path.exists():
|
return [], b""
|
payload = path.read_bytes()
|
if payload and (b"\r" in payload or not payload.endswith(b"\n")):
|
raise PipelineError("E_INPUT", f"{field} is not strict JSONL")
|
rows: list[dict[str, Any]] = []
|
for index, line in enumerate(payload.splitlines(), 1):
|
try:
|
row = json.loads(line.decode("utf-8"))
|
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
raise PipelineError("E_INPUT", f"{field} line {index} is invalid") from exc
|
if not isinstance(row, dict):
|
raise PipelineError("E_INPUT", f"{field} line {index} is not an object")
|
if reject_secrets:
|
_reject_secrets(row)
|
rows.append(row)
|
return rows, payload
|
|
|
def _load_state(config: Config) -> dict[str, Any]:
|
value, _ = _strict_json(config.state_path, "state")
|
state = _exact(value, {"schema_version", "task_id", "creator_uid", "initialized_at", "cursors", "active_run"}, "state")
|
if (
|
type(state["schema_version"]) is not int
|
or state["schema_version"] != SCHEMA
|
or state["task_id"] != TASK_ID
|
or state["creator_uid"] != config.creator_uid
|
):
|
raise PipelineError("E_STATE", "state identity differs")
|
cursors = _exact(state["cursors"], {"formal_lines", "formal_sha256", "handoff_lines", "handoff_sha256"}, "cursors")
|
if (
|
type(cursors["formal_lines"]) is not int
|
or cursors["formal_lines"] < 0
|
or type(cursors["handoff_lines"]) is not int
|
or cursors["handoff_lines"] < 0
|
or not isinstance(cursors["formal_sha256"], str)
|
or not SHA256.fullmatch(cursors["formal_sha256"])
|
or not isinstance(cursors["handoff_sha256"], str)
|
or not SHA256.fullmatch(cursors["handoff_sha256"])
|
):
|
raise PipelineError("E_STATE", "state cursor identity differs")
|
return dict(state)
|
|
|
def _write_state(config: Config, state: Mapping[str, Any]) -> None:
|
_atomic(config.state_path, _canonical(state))
|
|
|
def _now(value: str | None) -> datetime:
|
if value is None:
|
return datetime.now(timezone.utc)
|
try:
|
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
except ValueError as exc:
|
raise PipelineError("E_TIME", "now is invalid") from exc
|
if parsed.tzinfo is None:
|
raise PipelineError("E_TIME", "now must be offset-aware")
|
return parsed.astimezone(timezone.utc)
|
|
|
def _published_at(value: Any, field: str = "published_at") -> datetime:
|
if not isinstance(value, str) or not value:
|
raise PipelineError("E_SOURCE_BINDING", f"{field} differs")
|
try:
|
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
except ValueError as exc:
|
raise PipelineError("E_SOURCE_BINDING", f"{field} differs") from exc
|
if parsed.tzinfo is None:
|
raise PipelineError("E_SOURCE_BINDING", f"{field} differs")
|
return parsed
|
|
|
def _canonical_title(value: Any) -> str:
|
if not isinstance(value, str) or not value or "\x00" in value:
|
raise PipelineError("E_SOURCE_BINDING", "video title differs")
|
cleaned = WINDOWS_INVALID_CHARS.sub("_", value)
|
cleaned = re.sub(r"\s+", " ", cleaned).strip(" .")
|
if not cleaned:
|
cleaned = "untitled"
|
if cleaned.upper().split(".", 1)[0] in WINDOWS_RESERVED_NAMES:
|
cleaned = "_" + cleaned
|
cleaned = cleaned[:CANONICAL_TITLE_MAX_LENGTH].rstrip(" .")
|
return cleaned or "untitled"
|
|
|
def _canonical_video_base(stable_id: Any, title: Any, published_at: Any) -> str:
|
if not isinstance(stable_id, str) or not BVID.fullmatch(stable_id):
|
raise PipelineError("E_SOURCE_BINDING", "video stable identity differs")
|
published = _published_at(published_at)
|
return f"{published:%Y%m%d-%H%M%S}_video_{_canonical_title(title)}_{stable_id}"
|
|
|
def _project_relative(config: Config, path: Path, code: str = "E_ARTIFACT") -> str:
|
target = Path(os.path.abspath(path))
|
if not _within(target, config.project_root):
|
raise PipelineError(code, "path escaped the project root")
|
return target.relative_to(config.project_root).as_posix()
|
|
|
def initialize(config: Config, now: datetime) -> dict[str, Any]:
|
with _lock(config):
|
if config.state_path.exists():
|
state = _load_state(config)
|
return {"status": "ALREADY_INITIALIZED", "state": state, "state_identity": _file_identity(config.state_path)}
|
formal = _file_identity(config.formal_manifest)
|
handoff = _file_identity(config.processing_handoffs)
|
state = {
|
"schema_version": SCHEMA, "task_id": TASK_ID, "creator_uid": config.creator_uid,
|
"initialized_at": now.isoformat(),
|
"cursors": {
|
"formal_lines": formal["lines"], "formal_sha256": formal["sha256"],
|
"handoff_lines": handoff["lines"], "handoff_sha256": handoff["sha256"],
|
},
|
"active_run": None,
|
}
|
_write_state(config, state)
|
return {"status": "INITIALIZED", "baseline": {"formal": formal, "handoff": handoff}, "state_identity": _file_identity(config.state_path)}
|
|
|
def _run_id(config: Config, slot: int) -> str:
|
return hashlib.sha256(f"bili-half-hour-v1\0{config.creator_uid}\0{slot}".encode("ascii")).hexdigest()
|
|
|
def begin(config: Config, now: datetime) -> dict[str, Any]:
|
with _lock(config):
|
state = _load_state(config)
|
slot = int(now.timestamp()) // (INTERVAL_MINUTES * 60)
|
run_id = _run_id(config, slot)
|
active = state["active_run"]
|
if active is not None:
|
if active.get("run_id") == run_id:
|
return {"status": "RUN_RESUMED", "run": active}
|
raise PipelineError("E_RUN_ACTIVE", "a prior half-hour run is still active")
|
event = {
|
"schema_version": SCHEMA, "event": "RUN_STARTED", "run_id": run_id, "slot": slot,
|
"creator_uid": config.creator_uid, "started_at": now.isoformat(),
|
}
|
_append(config.runs_path, event)
|
state["active_run"] = {"run_id": run_id, "slot": slot, "started_at": now.isoformat()}
|
_write_state(config, state)
|
return {"status": "RUN_STARTED", "run": state["active_run"], "dynamic_url": config.dynamic_url}
|
|
|
def _journal_source_row(config: Config, source: Mapping[str, Any]) -> tuple[str, Mapping[str, Any]]:
|
if set(source) != {"journal", "line", "sha256"}:
|
raise PipelineError("E_SOURCE_BINDING", "journal source shape differs")
|
journal = source.get("journal")
|
line = source.get("line")
|
digest = source.get("sha256")
|
if journal not in {"formal", "processing_handoff"} or type(line) is not int or line <= 0 or not isinstance(digest, str) or not SHA256.fullmatch(digest):
|
raise PipelineError("E_SOURCE_BINDING", "journal source identity differs")
|
path = config.formal_manifest if journal == "formal" else config.processing_handoffs
|
# Formal and processing journals contain immutable historical audit fields.
|
# Do not treat a legacy key name (for example a handoff identity containing
|
# "authorization") as a credential. Every consumable row is rebound to a
|
# narrow source-controlled projection below, and the projected outbox is
|
# still subject to the normal key/value secret rejection.
|
rows, raw = _read_lines(path, journal, reject_secrets=False)
|
raw_lines = raw.splitlines()
|
if line > len(rows) or _row_digest(raw_lines[line - 1]) != digest:
|
raise PipelineError("E_SOURCE_BINDING", "journal source bytes differ")
|
row = rows[line - 1]
|
if _creator_uid(row) != config.creator_uid:
|
raise PipelineError("E_SOURCE_BINDING", "journal creator differs")
|
return journal, row
|
|
|
def _safe_text(value: Any, field: str, *, pattern: re.Pattern[str] | None = None) -> str:
|
if not isinstance(value, str) or not value or "\x00" in value or SECRET_KEY.search(value):
|
raise PipelineError("E_SOURCE_BINDING", f"{field} differs")
|
if pattern is not None and not pattern.fullmatch(value):
|
raise PipelineError("E_SOURCE_BINDING", f"{field} differs")
|
return value
|
|
|
def _project_formal_payload(config: Config, kind: str, row: Mapping[str, Any]) -> dict[str, Any]:
|
item_type = row.get("item_type")
|
status_value = row.get("status")
|
stable_id = row.get("stable_id")
|
if kind == "GIT_DELIVERY_READY":
|
if item_type not in CONTENT_TYPES or status_value != "SAVED" or not isinstance(stable_id, str) or not stable_id:
|
raise PipelineError("E_SOURCE_BINDING", "formal content source differs")
|
return {"stable_id": stable_id, "files": _content_artifacts(row, config), "reason": "CONTENT_ARCHIVED"}
|
if kind == "VIDEO_DOWNLOAD_READY":
|
if item_type != "video" or status_value == VIDEO_COMPLETE or not isinstance(stable_id, str) or not BVID.fullmatch(stable_id):
|
raise PipelineError("E_SOURCE_BINDING", "formal video source differs")
|
duration = row.get("expected_duration_seconds")
|
if isinstance(duration, bool) or not isinstance(duration, (int, float)) or not (0 < float(duration) < 86400):
|
raise PipelineError("E_SOURCE_BINDING", "video duration differs")
|
return {
|
"bvid": stable_id,
|
"source_url": _canonical_video_url(row.get("source_url"), stable_id),
|
"title": _safe_text(row.get("title"), "video title"),
|
"published_at": _safe_text(row.get("published_at"), "video publication"),
|
"expected_duration_seconds": duration,
|
}
|
raise PipelineError("E_SOURCE_BINDING", "formal source cannot produce this kind")
|
|
|
def _project_handoff_payload(config: Config, row: Mapping[str, Any]) -> dict[str, Any]:
|
common_keys = {
|
"type", "status", "handoff_id", "queue_job_id", "creator_uid", "bvid",
|
"source_url", "media_path", "mapping_path", "bytes", "sha256", "duration_seconds",
|
"video_codec", "audio_codec", "created_at",
|
}
|
schema_keys = set(row) - common_keys
|
if (
|
schema_keys not in ({"schema"}, {"schema_version"})
|
or set(row) != common_keys | schema_keys
|
or row.get("type") != "media-processing-handoff"
|
or row.get("status") != "READY"
|
):
|
raise PipelineError("E_SOURCE_BINDING", "processing handoff shape differs")
|
schema_value = row[next(iter(schema_keys))]
|
if type(schema_value) is not int or schema_value != SCHEMA or row.get("creator_uid") != config.creator_uid:
|
raise PipelineError("E_SOURCE_BINDING", "processing handoff identity differs")
|
bvid = row.get("bvid")
|
byte_count = row.get("bytes")
|
duration = row.get("duration_seconds")
|
if (
|
not isinstance(bvid, str) or not BVID.fullmatch(bvid)
|
or type(byte_count) is not int or byte_count <= 0
|
or isinstance(duration, bool) or not isinstance(duration, (int, float)) or float(duration) <= 0
|
or not isinstance(row.get("sha256"), str) or not SHA256_MIXED_ASCII.fullmatch(row["sha256"])
|
or not isinstance(row.get("queue_job_id"), str) or not re.fullmatch(r"[0-9a-f]{64}", row["queue_job_id"])
|
):
|
raise PipelineError("E_SOURCE_BINDING", "processing handoff media identity differs")
|
projected = {
|
"type": "media-processing-handoff", "status": "READY",
|
"handoff_id": _safe_text(row.get("handoff_id"), "handoff id", pattern=SAFE_ID),
|
"queue_job_id": row["queue_job_id"], "creator_uid": config.creator_uid, "bvid": bvid,
|
"source_url": _canonical_video_url(row.get("source_url"), bvid),
|
"media_path": _safe_text(row.get("media_path"), "media path"),
|
"mapping_path": _safe_text(row.get("mapping_path"), "mapping path"),
|
"bytes": byte_count, "sha256": row["sha256"].upper(), "duration_seconds": duration,
|
"video_codec": _safe_text(row.get("video_codec"), "video codec", pattern=SAFE_ID),
|
"audio_codec": _safe_text(row.get("audio_codec"), "audio codec", pattern=SAFE_ID),
|
"created_at": _safe_text(row.get("created_at"), "handoff creation time"),
|
}
|
return projected
|
|
|
def _project_relocation_payload(config: Config, source: Mapping[str, Any]) -> dict[str, Any]:
|
if set(source) != {"journal", "path", "bytes", "sha256"} or source.get("journal") != "relocation_report":
|
raise PipelineError("E_SOURCE_BINDING", "relocation source shape differs")
|
relative = source.get("path")
|
size = source.get("bytes")
|
digest = source.get("sha256")
|
if (
|
not isinstance(relative, str)
|
or Path(relative).is_absolute()
|
or type(size) is not int
|
or size <= 0
|
or not isinstance(digest, str)
|
or not SHA256.fullmatch(digest)
|
):
|
raise PipelineError("E_SOURCE_BINDING", "relocation source identity differs")
|
path = Path(os.path.abspath(config.project_root / Path(relative)))
|
if not _within(path, config.relocation_root):
|
raise PipelineError("E_SOURCE_BINDING", "relocation source escaped its root")
|
payload = _stable_artifact_bytes(path, config.relocation_root)
|
if len(payload) != size or hashlib.sha256(payload).hexdigest().upper() != digest:
|
raise PipelineError("E_SOURCE_BINDING", "relocation source bytes differ")
|
report = _read_relocation_report(config, path)
|
files: list[dict[str, Any]] = []
|
for item in report["items"]:
|
files.extend(
|
{
|
"path": alias["new_path"],
|
"bytes": alias["bytes"],
|
"sha256": alias["sha256"],
|
"kind": alias["kind"],
|
}
|
for alias in item["aliases"]
|
)
|
files.extend(report["docs"])
|
files.append({
|
"path": relative,
|
"bytes": size,
|
"sha256": digest,
|
"kind": "relocation_manifest",
|
})
|
return {
|
"stable_id": report["batch_id"],
|
"files": files,
|
"remove_paths": report["remove_paths"],
|
"reason": "CANONICAL_VIDEO_ARTIFACT_MIGRATION",
|
}
|
|
|
def _expected_outbox_payload(config: Config, kind: str, source: Mapping[str, Any]) -> dict[str, Any]:
|
if source.get("journal") == "relocation_report":
|
if kind != "GIT_DELIVERY_READY":
|
raise PipelineError("E_SOURCE_BINDING", "relocation source kind differs")
|
return _project_relocation_payload(config, source)
|
if source.get("journal") in {"formal", "processing_handoff"}:
|
journal, row = _journal_source_row(config, source)
|
if journal == "formal":
|
return _project_formal_payload(config, kind, row)
|
if kind != "VIDEO_TRANSCRIPTION_READY":
|
raise PipelineError("E_SOURCE_BINDING", "handoff source kind differs")
|
return _project_handoff_payload(config, row)
|
allowed = {"journal", "receipt_sha256"}
|
if source.get("projection") is not None:
|
allowed.add("projection")
|
if set(source) != allowed or source.get("journal") != "terminals" or not isinstance(source.get("receipt_sha256"), str) or not SHA256.fullmatch(source["receipt_sha256"]):
|
raise PipelineError("E_SOURCE_BINDING", "terminal source identity differs")
|
terminals = [row for row in _terminal_rows(config) if row.get("receipt_sha256") == source["receipt_sha256"]]
|
if len(terminals) != 1:
|
raise PipelineError("E_SOURCE_BINDING", "terminal source is absent or ambiguous")
|
terminal = terminals[0]
|
files = terminal["files"]
|
stable_id = terminal["stable_id"]
|
projection = source.get("projection")
|
if kind == "MINUTES_READY" and terminal["event"] == "TRANSCRIPTION_COMPLETE" and projection is None:
|
return {"stable_id": stable_id, "transcript_terminal_id": terminal["terminal_id"], "files": files}
|
if kind == "GIT_DELIVERY_READY" and terminal["event"] == "TRANSCRIPTION_COMPLETE" and projection == "transcript":
|
return {"stable_id": stable_id, "files": files, "reason": "TRANSCRIPT_COMPLETE"}
|
if kind == "GIT_DELIVERY_READY" and terminal["event"] == "MINUTES_COMPLETE" and projection == "minutes":
|
return {"stable_id": stable_id, "files": files, "reason": "MINUTES_COMPLETE"}
|
raise PipelineError("E_SOURCE_BINDING", "terminal source projection differs")
|
|
|
def _outbox_id(kind: str, source: Mapping[str, Any], payload: Mapping[str, Any]) -> str:
|
material = _canonical({"kind": kind, "source": dict(source), "payload": dict(payload)})
|
return hashlib.sha256(material).hexdigest()
|
|
|
def _outbox_rows(config: Config) -> list[dict[str, Any]]:
|
rows, _ = _read_lines(config.outbox_path, "outbox")
|
allowed_kinds = {"GIT_DELIVERY_READY", "VIDEO_DOWNLOAD_READY", "VIDEO_TRANSCRIPTION_READY", "MINUTES_READY"}
|
grouped: dict[str, list[dict[str, Any]]] = {}
|
for row in rows:
|
if type(row.get("schema_version")) is not int or row.get("schema_version") != SCHEMA:
|
raise PipelineError("E_OUTBOX", "outbox schema identity differs")
|
outbox_id = row.get("outbox_id")
|
event = row.get("event")
|
if not isinstance(outbox_id, str) or not re.fullmatch(r"[0-9a-f]{64}", outbox_id):
|
raise PipelineError("E_OUTBOX", "outbox identity differs")
|
if event not in {"CREATED", "DISPATCH_INTENT", "OBSERVED", "GIT_COMMIT_INTENT", "COMMIT_CREATED", "COMPLETE"}:
|
raise PipelineError("E_OUTBOX", "outbox event differs")
|
grouped.setdefault(outbox_id, []).append(row)
|
for outbox_id, events in grouped.items():
|
created = [row for row in events if row.get("event") == "CREATED"]
|
if len(created) != 1 or events[0].get("event") != "CREATED":
|
raise PipelineError("E_OUTBOX", "outbox creation history differs")
|
origin = created[0]
|
if set(origin) != {"schema_version", "event", "outbox_id", "kind", "creator_uid", "source", "payload", "created_at"}:
|
raise PipelineError("E_OUTBOX", "outbox creation shape differs")
|
kind = origin.get("kind")
|
source = origin.get("source")
|
if (
|
kind not in allowed_kinds
|
or origin.get("creator_uid") != config.creator_uid
|
or not isinstance(source, dict)
|
or not isinstance(origin.get("payload"), dict)
|
or _outbox_id(kind, source, origin["payload"]) != outbox_id
|
or not isinstance(origin.get("created_at"), str)
|
):
|
raise PipelineError("E_OUTBOX", "outbox creation binding differs")
|
expected_payload = _expected_outbox_payload(config, kind, source)
|
if origin["payload"] != expected_payload:
|
raise PipelineError("E_OUTBOX", "outbox payload differs from its immutable source")
|
counts = {name: sum(row.get("event") == name for row in events) for name in {"DISPATCH_INTENT", "OBSERVED", "GIT_COMMIT_INTENT", "COMMIT_CREATED", "COMPLETE"}}
|
if any(value > 1 for value in counts.values()):
|
raise PipelineError("E_OUTBOX", "outbox event is duplicated")
|
intent = next((row for row in events if row.get("event") == "DISPATCH_INTENT"), None)
|
observed = next((row for row in events if row.get("event") == "OBSERVED"), None)
|
git_intent = next((row for row in events if row.get("event") == "GIT_COMMIT_INTENT"), None)
|
commit = next((row for row in events if row.get("event") == "COMMIT_CREATED"), None)
|
complete = next((row for row in events if row.get("event") == "COMPLETE"), None)
|
if intent is not None:
|
if (
|
kind == "GIT_DELIVERY_READY"
|
or set(intent) != {"schema_version", "event", "outbox_id", "kind", "target_thread_id", "created_at"}
|
or intent.get("kind") != kind
|
or intent.get("target_thread_id") != _dispatch_target(config, kind)
|
or not isinstance(intent.get("created_at"), str)
|
):
|
raise PipelineError("E_OUTBOX", "dispatch intent binding differs")
|
if observed is not None:
|
if (
|
intent is None
|
or set(observed) != {"schema_version", "event", "outbox_id", "delivery_id", "observed_at"}
|
or not isinstance(observed.get("delivery_id"), str)
|
or not SAFE_ID.fullmatch(observed["delivery_id"])
|
or not isinstance(observed.get("observed_at"), str)
|
):
|
raise PipelineError("E_OUTBOX", "dispatch observation binding differs")
|
if git_intent is not None:
|
if (
|
kind != "GIT_DELIVERY_READY"
|
or set(git_intent) != {
|
"schema_version", "event", "outbox_id", "parent_sha", "tree_sha", "files",
|
"message", "author_name", "author_email", "authored_at", "created_at",
|
}
|
or not isinstance(git_intent.get("files"), list)
|
or not isinstance(git_intent.get("message"), str)
|
or not isinstance(git_intent.get("author_name"), str)
|
or not isinstance(git_intent.get("author_email"), str)
|
or not isinstance(git_intent.get("authored_at"), str)
|
or not isinstance(git_intent.get("created_at"), str)
|
or git_intent.get("files") != _git_expected_paths(origin.get("payload", {}))
|
):
|
raise PipelineError("E_OUTBOX", "Git commit intent binding differs")
|
for field in ("parent_sha", "tree_sha"):
|
if not isinstance(git_intent.get(field), str) or not re.fullmatch(r"[0-9a-f]{40,64}", git_intent[field]):
|
raise PipelineError("E_OUTBOX", "Git commit intent identity differs")
|
if commit is not None:
|
if (
|
kind != "GIT_DELIVERY_READY"
|
or git_intent is None
|
or set(commit) != {"schema_version", "event", "outbox_id", "parent_sha", "tree_sha", "commit_sha", "intent_sha256", "files", "created_at"}
|
or not isinstance(commit.get("files"), list)
|
or not isinstance(commit.get("created_at"), str)
|
or commit.get("parent_sha") != git_intent.get("parent_sha")
|
or commit.get("tree_sha") != git_intent.get("tree_sha")
|
or commit.get("files") != git_intent.get("files")
|
or commit.get("intent_sha256") != hashlib.sha256(_canonical(git_intent)).hexdigest().upper()
|
):
|
raise PipelineError("E_OUTBOX", "Git commit binding differs")
|
for field in ("parent_sha", "tree_sha", "commit_sha"):
|
if not isinstance(commit.get(field), str) or not re.fullmatch(r"[0-9a-f]{40,64}", commit[field]):
|
raise PipelineError("E_OUTBOX", "Git commit identity differs")
|
if complete is not None:
|
result = complete.get("result")
|
if not isinstance(complete.get("completed_at"), str):
|
raise PipelineError("E_OUTBOX", "outbox completion time differs")
|
if kind == "GIT_DELIVERY_READY":
|
expected = {"schema_version", "event", "outbox_id", "result", "completed_at"}
|
if result == "PUSHED":
|
expected.add("commit_sha")
|
if set(complete) != expected or result not in {"PUSHED", "NO_CHANGES"}:
|
raise PipelineError("E_OUTBOX", "Git completion binding differs")
|
if result == "PUSHED" and (commit is None or complete.get("commit_sha") != commit.get("commit_sha")):
|
raise PipelineError("E_OUTBOX", "Git completion commit differs")
|
else:
|
if (
|
intent is None or observed is None
|
or set(complete) != {"schema_version", "event", "outbox_id", "result", "terminal_id", "receipt_sha256", "completed_at"}
|
or result != ({"VIDEO_TRANSCRIPTION_READY": "TRANSCRIPTION_COMPLETE", "MINUTES_READY": "MINUTES_COMPLETE"}.get(kind))
|
or not isinstance(complete.get("terminal_id"), str)
|
or not SAFE_ID.fullmatch(complete["terminal_id"])
|
or not isinstance(complete.get("receipt_sha256"), str)
|
or not SHA256.fullmatch(complete["receipt_sha256"])
|
):
|
raise PipelineError("E_OUTBOX", "role completion binding differs")
|
sequence = [row["event"] for row in events]
|
if kind == "GIT_DELIVERY_READY":
|
if intent is not None or observed is not None:
|
raise PipelineError("E_OUTBOX", "Git outbox contains role events")
|
if git_intent is not None and sequence.index("GIT_COMMIT_INTENT") <= 0:
|
raise PipelineError("E_OUTBOX", "Git intent order differs")
|
if commit is not None and sequence.index("COMMIT_CREATED") <= sequence.index("GIT_COMMIT_INTENT"):
|
raise PipelineError("E_OUTBOX", "Git commit order differs")
|
if complete is not None and sequence.index("COMPLETE") != len(sequence) - 1:
|
raise PipelineError("E_OUTBOX", "Git completion order differs")
|
else:
|
if git_intent is not None or commit is not None:
|
raise PipelineError("E_OUTBOX", "role outbox contains Git events")
|
if intent is not None and sequence.index("DISPATCH_INTENT") <= 0:
|
raise PipelineError("E_OUTBOX", "dispatch intent order differs")
|
if observed is not None and (intent is None or sequence.index("OBSERVED") <= sequence.index("DISPATCH_INTENT")):
|
raise PipelineError("E_OUTBOX", "dispatch observation order differs")
|
if complete is not None and (observed is None or sequence.index("COMPLETE") <= sequence.index("OBSERVED") or sequence.index("COMPLETE") != len(sequence) - 1):
|
raise PipelineError("E_OUTBOX", "role completion order differs")
|
return rows
|
|
|
def _append_outbox(config: Config, kind: str, source: Mapping[str, Any], payload: Mapping[str, Any], created_at: str) -> str:
|
expected_payload = _expected_outbox_payload(config, kind, source)
|
if dict(payload) != expected_payload:
|
raise PipelineError("E_OUTBOX", "new outbox payload differs from its immutable source")
|
outbox_id = _outbox_id(kind, source, payload)
|
rows = _outbox_rows(config)
|
same_source = [
|
row for row in rows
|
if row.get("event") == "CREATED" and row.get("kind") == kind and row.get("source") == dict(source)
|
]
|
if same_source and not (
|
len(same_source) == 1
|
and same_source[0].get("outbox_id") == outbox_id
|
and same_source[0].get("payload") == dict(payload)
|
):
|
raise PipelineError("E_OUTBOX", "source identity is already bound to another payload")
|
prior = next((row for row in rows if row.get("event") == "CREATED" and row.get("outbox_id") == outbox_id), None)
|
if prior is not None:
|
if prior.get("kind") != kind or prior.get("source") != dict(source) or prior.get("payload") != dict(payload) or prior.get("creator_uid") != config.creator_uid:
|
raise PipelineError("E_OUTBOX", "existing outbox identity differs")
|
return outbox_id
|
event = {
|
"schema_version": SCHEMA, "event": "CREATED", "outbox_id": outbox_id,
|
"kind": kind, "creator_uid": config.creator_uid, "source": dict(source),
|
"payload": dict(payload), "created_at": created_at,
|
}
|
_append(config.outbox_path, event)
|
return outbox_id
|
|
|
def _row_digest(raw_line: bytes) -> str:
|
return hashlib.sha256(raw_line).hexdigest().upper()
|
|
|
def _prefix_digest(payload: bytes, lines: int, field: str) -> str:
|
if type(lines) is not int or lines < 0:
|
raise PipelineError("E_STATE", f"{field} cursor line count differs")
|
chunks = payload.splitlines(keepends=True)
|
if len(chunks) < lines:
|
raise PipelineError("E_HISTORY_REWRITE", f"{field} lost rows")
|
return hashlib.sha256(b"".join(chunks[:lines])).hexdigest().upper()
|
|
|
def _creator_uid(row: Mapping[str, Any]) -> str | None:
|
value = row.get("creator_uid")
|
if type(value) is int:
|
return str(value)
|
if isinstance(value, str):
|
return value
|
creator = row.get("creator")
|
if isinstance(creator, dict) and isinstance(creator.get("uid"), str):
|
return creator["uid"]
|
return None
|
|
|
def _is_reparse(info: os.stat_result) -> bool:
|
return bool(getattr(info, "st_file_attributes", 0) & getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0x400))
|
|
|
def _stat_identity(info: os.stat_result) -> tuple[int, ...]:
|
return (
|
int(info.st_dev), int(info.st_ino), int(info.st_mode), int(info.st_nlink), int(info.st_size),
|
int(info.st_mtime_ns), int(info.st_ctime_ns), int(getattr(info, "st_file_attributes", 0)),
|
)
|
|
|
def _path_handle_identity(info: os.stat_result) -> tuple[int, ...]:
|
identity = _stat_identity(info)
|
return identity[:6] + identity[7:]
|
|
|
def _chain_identity(info: os.stat_result, *, final_file: bool) -> tuple[int, ...]:
|
if final_file:
|
return _stat_identity(info)
|
return (int(info.st_dev), int(info.st_ino), int(info.st_mode), int(getattr(info, "st_file_attributes", 0)))
|
|
|
def _strict_chain(root: Path, target: Path, *, final_file: bool) -> tuple[tuple[str, tuple[int, ...]], ...]:
|
root = Path(os.path.abspath(root))
|
target = Path(os.path.abspath(target))
|
if not _within(target, root):
|
raise PipelineError("E_ARTIFACT", "path escaped its governed root")
|
root_real = Path(os.path.realpath(root))
|
target_real = Path(os.path.realpath(target))
|
if not _within(target_real, root_real) or os.path.normcase(str(target_real)) != os.path.normcase(str(target)):
|
raise PipelineError("E_ARTIFACT", "path resolution escaped or drifted")
|
anchor = Path(target.anchor)
|
paths: list[Path] = []
|
current = anchor
|
if str(anchor):
|
paths.append(anchor)
|
for part in target.parts[1:] if str(anchor) else target.parts:
|
current = current / part
|
paths.append(current)
|
snapshots: list[tuple[str, tuple[int, ...]]] = []
|
for index, item in enumerate(paths):
|
try:
|
info = os.lstat(item)
|
except OSError as exc:
|
raise PipelineError("E_ARTIFACT", "governed path is unavailable") from exc
|
if stat.S_ISLNK(info.st_mode) or _is_reparse(info):
|
raise PipelineError("E_ARTIFACT", "governed path contains a reparse object")
|
is_final = index == len(paths) - 1
|
if (is_final and final_file and not stat.S_ISREG(info.st_mode)) or ((not is_final or not final_file) and not stat.S_ISDIR(info.st_mode)):
|
raise PipelineError("E_ARTIFACT", "governed path object type differs")
|
snapshots.append((os.path.normcase(str(item)), _chain_identity(info, final_file=is_final and final_file)))
|
return tuple(snapshots)
|
|
|
def _stable_artifact_bytes(target: Path, root: Path) -> bytes:
|
before = _strict_chain(root, target, final_file=True)
|
flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0)
|
try:
|
descriptor = os.open(target, flags)
|
except OSError as exc:
|
raise PipelineError("E_ARTIFACT", "artifact cannot be opened safely") from exc
|
try:
|
opened_before = os.fstat(descriptor)
|
if _path_handle_identity(opened_before) != before[-1][1][:6] + before[-1][1][7:] or not stat.S_ISREG(opened_before.st_mode) or _is_reparse(opened_before):
|
raise PipelineError("E_ARTIFACT", "artifact path and handle differ")
|
with os.fdopen(descriptor, "rb", closefd=False) as stream:
|
payload = stream.read()
|
opened_after = os.fstat(descriptor)
|
after = _strict_chain(root, target, final_file=True)
|
if _stat_identity(opened_before) != _stat_identity(opened_after) or before != after:
|
raise PipelineError("E_ARTIFACT", "artifact identity drifted during read")
|
return payload
|
finally:
|
os.close(descriptor)
|
|
|
def _relocation_batch_id(report: Mapping[str, Any]) -> str:
|
material = {
|
"schema_version": report.get("schema_version"),
|
"type": report.get("type"),
|
"task_id": report.get("task_id"),
|
"creator_uid": report.get("creator_uid"),
|
"baseline_head": report.get("baseline_head"),
|
"items": report.get("items"),
|
"docs": report.get("docs"),
|
"remove_paths": report.get("remove_paths"),
|
}
|
return hashlib.sha256(_canonical(material)).hexdigest().upper()
|
|
|
def _read_relocation_report(config: Config, path: Path) -> dict[str, Any]:
|
_strict_chain(config.relocation_root, path, final_file=True)
|
value, _ = _strict_json(path, "artifact relocation report")
|
report = _exact(
|
value,
|
{
|
"schema_version", "type", "task_id", "creator_uid", "batch_id", "created_at",
|
"baseline_head", "items", "docs", "remove_paths",
|
},
|
"artifact relocation report",
|
)
|
if (
|
type(report["schema_version"]) is not int
|
or report["schema_version"] != SCHEMA
|
or report["type"] != RELOCATION_REPORT_TYPE
|
or report["task_id"] != TASK_ID
|
or report["creator_uid"] != config.creator_uid
|
or not isinstance(report["baseline_head"], str)
|
or not re.fullmatch(r"[0-9a-f]{40,64}", report["baseline_head"])
|
or not isinstance(report["batch_id"], str)
|
or not SHA256.fullmatch(report["batch_id"])
|
or report["batch_id"] != _relocation_batch_id(report)
|
or path.name != f"{report['batch_id']}.json"
|
or not isinstance(report["created_at"], str)
|
or not isinstance(report["items"], list)
|
or not report["items"]
|
or not isinstance(report["docs"], list)
|
or not isinstance(report["remove_paths"], list)
|
):
|
raise PipelineError("E_RELOCATION", "artifact relocation report identity differs")
|
aliases: list[dict[str, Any]] = []
|
stable_ids: set[str] = set()
|
item_ids: list[str] = []
|
for item_value in report["items"]:
|
item = _exact(
|
item_value,
|
{"stable_id", "title", "published_at", "canonical_base", "aliases", "intermediates"},
|
"artifact relocation item",
|
)
|
stable_id = item["stable_id"]
|
if (
|
not isinstance(stable_id, str)
|
or not BVID.fullmatch(stable_id)
|
or stable_id in stable_ids
|
or item["canonical_base"] != _canonical_video_base(stable_id, item["title"], item["published_at"])
|
or not isinstance(item["aliases"], list)
|
or not item["aliases"]
|
or not isinstance(item["intermediates"], list)
|
):
|
raise PipelineError("E_RELOCATION", "artifact relocation item identity differs")
|
stable_ids.add(stable_id)
|
item_ids.append(stable_id)
|
canonical_base = item["canonical_base"]
|
expected_alias_paths = {
|
"transcript_txt": (
|
config.archive_root / f"{stable_id}.transcript" / f"{stable_id}.txt",
|
config.archive_root / f"{canonical_base}.transcript" / f"{canonical_base}.txt",
|
),
|
"transcript_srt": (
|
config.archive_root / f"{stable_id}.transcript" / f"{stable_id}.srt",
|
config.archive_root / f"{canonical_base}.transcript" / f"{canonical_base}.srt",
|
),
|
"transcript_json": (
|
config.archive_root / f"{stable_id}.transcript" / f"{stable_id}.json",
|
config.archive_root / f"{canonical_base}.transcript" / f"{canonical_base}.json",
|
),
|
"minutes_md": (
|
config.archive_root / f"{stable_id}.minutes" / f"{stable_id}.md",
|
config.archive_root / f"{canonical_base}.minutes" / f"{canonical_base}.md",
|
),
|
"minutes_pdf": (
|
config.archive_root / f"{stable_id}.minutes" / f"{stable_id}.pdf",
|
config.archive_root / f"{canonical_base}.minutes" / f"{canonical_base}.pdf",
|
),
|
}
|
item_kinds: set[str] = set()
|
for alias_value in item["aliases"]:
|
alias = _exact(alias_value, {"kind", "old_path", "new_path", "bytes", "sha256"}, "artifact relocation alias")
|
kind = alias["kind"]
|
if (
|
kind not in PUBLIC_VIDEO_KINDS
|
or not isinstance(alias["old_path"], str)
|
or not isinstance(alias["new_path"], str)
|
or alias["old_path"] == alias["new_path"]
|
or type(alias["bytes"]) is not int
|
or alias["bytes"] < 0
|
or not isinstance(alias["sha256"], str)
|
or not SHA256.fullmatch(alias["sha256"])
|
or Path(alias["new_path"]).suffix.lower() != PUBLIC_VIDEO_KINDS[kind]
|
):
|
raise PipelineError("E_RELOCATION", "artifact relocation alias differs")
|
if kind in item_kinds:
|
raise PipelineError("E_RELOCATION", "artifact relocation kind is duplicated")
|
item_kinds.add(kind)
|
expected_old, expected_new = expected_alias_paths[kind]
|
if (
|
alias["old_path"] != _project_relative(config, expected_old)
|
or alias["new_path"] != _project_relative(config, expected_new)
|
):
|
raise PipelineError("E_RELOCATION", "artifact relocation path grammar differs")
|
for field in ("old_path", "new_path"):
|
target = Path(os.path.abspath(config.project_root / Path(alias[field])))
|
if Path(alias[field]).is_absolute() or not _within(target, config.archive_root):
|
raise PipelineError("E_RELOCATION", "artifact relocation path escaped the archive")
|
aliases.append(dict(alias))
|
transcript_kinds = {kind for kind in item_kinds if kind.startswith("transcript_")}
|
minutes_kinds = {kind for kind in item_kinds if kind.startswith("minutes_")}
|
if transcript_kinds != {"transcript_txt", "transcript_srt", "transcript_json"} or minutes_kinds not in (
|
set(), {"minutes_md", "minutes_pdf"},
|
):
|
raise PipelineError("E_RELOCATION", "artifact relocation kind set differs")
|
if len(item["intermediates"]) > 1:
|
raise PipelineError("E_RELOCATION", "artifact relocation intermediate set differs")
|
for intermediate_value in item["intermediates"]:
|
intermediate = _exact(
|
intermediate_value,
|
{"kind", "old_path", "new_path", "bytes", "sha256"},
|
"artifact relocation intermediate",
|
)
|
if (
|
intermediate["kind"] != "audio_flac"
|
or not isinstance(intermediate["old_path"], str)
|
or not isinstance(intermediate["new_path"], str)
|
or type(intermediate["bytes"]) is not int
|
or intermediate["bytes"] <= 0
|
or not isinstance(intermediate["sha256"], str)
|
or not SHA256.fullmatch(intermediate["sha256"])
|
or not str(intermediate["new_path"]).lower().endswith(".audio.flac")
|
):
|
raise PipelineError("E_RELOCATION", "artifact relocation intermediate differs")
|
expected_old = config.archive_root / f"{stable_id}.transcript" / f"{stable_id}.audio.flac"
|
expected_new = config.video_root / "intermediate" / "transcription" / f"{stable_id}.audio.flac"
|
if (
|
intermediate["old_path"] != _project_relative(config, expected_old)
|
or os.path.normcase(intermediate["new_path"]) != os.path.normcase(str(Path(os.path.abspath(expected_new))))
|
):
|
raise PipelineError("E_RELOCATION", "artifact relocation intermediate grammar differs")
|
old_target = Path(os.path.abspath(config.project_root / Path(intermediate["old_path"])))
|
new_target = Path(os.path.abspath(Path(intermediate["new_path"])))
|
if Path(intermediate["old_path"]).is_absolute() or not _within(old_target, config.archive_root) or not _within(new_target, config.video_root):
|
raise PipelineError("E_RELOCATION", "artifact relocation intermediate escaped its boundary")
|
if item_ids != sorted(item_ids):
|
raise PipelineError("E_RELOCATION", "artifact relocation item order differs")
|
old_paths = [alias["old_path"] for alias in aliases]
|
new_paths = [alias["new_path"] for alias in aliases]
|
if len(set(old_paths)) != len(old_paths) or len(set(new_paths)) != len(new_paths):
|
raise PipelineError("E_RELOCATION", "artifact relocation aliases are duplicated")
|
if (
|
report["remove_paths"] != sorted(set(report["remove_paths"]))
|
or any(value not in old_paths for value in report["remove_paths"])
|
):
|
raise PipelineError("E_RELOCATION", "artifact relocation removal scope differs")
|
docs: list[dict[str, Any]] = []
|
for value in report["docs"]:
|
doc = _exact(value, {"path", "bytes", "sha256", "kind"}, "artifact relocation document")
|
if (
|
doc["kind"] != "documentation"
|
or doc["path"] not in config.git_doc_paths
|
or type(doc["bytes"]) is not int
|
or doc["bytes"] <= 0
|
or not isinstance(doc["sha256"], str)
|
or not SHA256.fullmatch(doc["sha256"])
|
):
|
raise PipelineError("E_RELOCATION", "artifact relocation document differs")
|
docs.append(dict(doc))
|
if (
|
len({value["path"] for value in docs}) != len(docs)
|
or [value["path"] for value in docs] != sorted(config.git_doc_paths)
|
):
|
raise PipelineError("E_RELOCATION", "artifact relocation document set differs")
|
return dict(report)
|
|
|
def _relocation_reports(config: Config) -> list[tuple[Path, dict[str, Any]]]:
|
if not config.relocation_root.exists():
|
return []
|
_strict_chain(config.archive_root, config.relocation_root, final_file=False)
|
values: list[tuple[Path, dict[str, Any]]] = []
|
for path in sorted(config.relocation_root.glob("*.json"), key=lambda value: value.name):
|
values.append((path, _read_relocation_report(config, path)))
|
return values
|
|
|
def _relocated_artifact_target(
|
config: Config,
|
old_relative: str,
|
expected_bytes: int,
|
expected_sha256: str,
|
) -> Path | None:
|
matches: list[dict[str, Any]] = []
|
for _, report in _relocation_reports(config):
|
for item in report["items"]:
|
matches.extend(
|
alias for alias in item["aliases"]
|
if alias["old_path"] == old_relative
|
and alias["bytes"] == expected_bytes
|
and alias["sha256"] == expected_sha256.upper()
|
)
|
if not matches:
|
return None
|
if len(matches) != 1:
|
raise PipelineError("E_RELOCATION", "artifact relocation alias is ambiguous")
|
return Path(os.path.abspath(config.project_root / Path(matches[0]["new_path"])))
|
|
|
def _artifact(path_value: Any, bytes_value: Any, sha_value: Any, config: Config) -> dict[str, Any]:
|
if not isinstance(path_value, str) or not path_value or type(bytes_value) is not int or bytes_value < 0 or not isinstance(sha_value, str) or not SHA256.fullmatch(sha_value.upper()):
|
raise PipelineError("E_ARTIFACT", "artifact identity is incomplete")
|
path = Path(path_value)
|
if path.is_absolute():
|
requested = Path(os.path.abspath(path))
|
targets = [requested] if requested.exists() or requested.is_symlink() else []
|
if not targets and _within(requested, config.archive_root):
|
relocated = _relocated_artifact_target(
|
config,
|
requested.relative_to(config.project_root).as_posix(),
|
bytes_value,
|
sha_value.upper(),
|
)
|
if relocated is not None:
|
targets.append(relocated)
|
else:
|
candidates = [Path(os.path.abspath(config.archive_root / path)), Path(os.path.abspath(config.project_root / path))]
|
requested_candidates = [candidate for candidate in candidates if _within(candidate, config.archive_root)]
|
targets = [candidate for candidate in requested_candidates if candidate.exists() or candidate.is_symlink()]
|
if not targets:
|
for candidate in requested_candidates:
|
relocated = _relocated_artifact_target(
|
config,
|
candidate.relative_to(config.project_root).as_posix(),
|
bytes_value,
|
sha_value.upper(),
|
)
|
if relocated is not None:
|
targets.append(relocated)
|
requested = requested_candidates[0] if len(requested_candidates) == 1 else None
|
distinct = {os.path.normcase(str(candidate)): candidate for candidate in targets}
|
if len(distinct) != 1:
|
raise PipelineError("E_ARTIFACT", "artifact path is absent or ambiguous")
|
target = next(iter(distinct.values()))
|
if not _within(target, config.archive_root) or any(str(target).lower().endswith(suffix) for suffix in FORBIDDEN_GIT_SUFFIXES):
|
raise PipelineError("E_ARTIFACT", "artifact path is outside the Git boundary")
|
payload = _stable_artifact_bytes(target, config.archive_root)
|
digest = hashlib.sha256(payload).hexdigest().upper()
|
if len(payload) != bytes_value or digest != sha_value.upper():
|
raise PipelineError("E_ARTIFACT", "artifact readback differs")
|
if path.is_absolute():
|
relative = requested.relative_to(config.project_root).as_posix()
|
else:
|
logical_candidates = [
|
candidate.relative_to(config.project_root).as_posix()
|
for candidate in requested_candidates
|
if _relocated_artifact_target(config, candidate.relative_to(config.project_root).as_posix(), bytes_value, sha_value.upper()) == target
|
or candidate == target
|
]
|
if len(set(logical_candidates)) != 1:
|
raise PipelineError("E_ARTIFACT", "artifact logical path is ambiguous")
|
relative = logical_candidates[0]
|
if target.suffix.lower() not in config.git_extensions:
|
raise PipelineError("E_ARTIFACT", "artifact extension is not allowlisted")
|
return {"path": relative, "bytes": len(payload), "sha256": digest}
|
|
|
def _content_artifacts(row: Mapping[str, Any], config: Config) -> list[dict[str, Any]]:
|
artifacts = [_artifact(row.get("path"), row.get("bytes"), row.get("sha256"), config)]
|
if row.get("image_path") is not None:
|
artifacts.append(_artifact(row.get("image_path"), row.get("image_bytes"), row.get("image_sha256"), config))
|
images = row.get("images")
|
if images is not None:
|
if not isinstance(images, list):
|
raise PipelineError("E_ARTIFACT", "images is not a list")
|
for image in images:
|
if not isinstance(image, dict):
|
raise PipelineError("E_ARTIFACT", "image identity is invalid")
|
candidate = _artifact(image.get("path"), image.get("bytes"), image.get("sha256"), config)
|
if candidate not in artifacts:
|
artifacts.append(candidate)
|
return artifacts
|
|
|
def reconcile(config: Config, now: datetime) -> dict[str, Any]:
|
with _lock(config):
|
state = _load_state(config)
|
active = state["active_run"]
|
if active is None:
|
raise PipelineError("E_NO_ACTIVE_RUN", "begin is required before reconcile")
|
formal_rows, formal_payload = _read_lines(
|
config.formal_manifest, "formal manifest", reject_secrets=False
|
)
|
handoff_rows, handoff_payload = _read_lines(
|
config.processing_handoffs, "processing handoff", reject_secrets=False
|
)
|
cursors = state["cursors"]
|
if (
|
_prefix_digest(formal_payload, cursors["formal_lines"], "formal manifest") != cursors["formal_sha256"]
|
or _prefix_digest(handoff_payload, cursors["handoff_lines"], "processing handoff") != cursors["handoff_sha256"]
|
):
|
raise PipelineError("E_HISTORY_REWRITE", "append-only input prefix changed")
|
created: list[str] = []
|
formal_raw = formal_payload.splitlines()
|
for index in range(cursors["formal_lines"], len(formal_rows)):
|
row = formal_rows[index]
|
if _creator_uid(row) != config.creator_uid:
|
raise PipelineError("E_CREATOR", "new formal row creator differs")
|
source = {"journal": "formal", "line": index + 1, "sha256": _row_digest(formal_raw[index])}
|
item_type = row.get("item_type")
|
if item_type in CONTENT_TYPES and row.get("status") == "SAVED":
|
payload = _expected_outbox_payload(config, "GIT_DELIVERY_READY", source)
|
created.append(_append_outbox(config, "GIT_DELIVERY_READY", source, payload, now.isoformat()))
|
elif item_type == "video" and row.get("status") != VIDEO_COMPLETE:
|
payload = _expected_outbox_payload(config, "VIDEO_DOWNLOAD_READY", source)
|
created.append(_append_outbox(config, "VIDEO_DOWNLOAD_READY", source, payload, now.isoformat()))
|
handoff_raw = handoff_payload.splitlines()
|
for index in range(cursors["handoff_lines"], len(handoff_rows)):
|
row = handoff_rows[index]
|
if _creator_uid(row) != config.creator_uid or row.get("status") != "READY":
|
raise PipelineError("E_HANDOFF", "processing handoff identity differs")
|
stable_id = row.get("bvid")
|
if not isinstance(stable_id, str) or not BVID.fullmatch(stable_id):
|
raise PipelineError("E_HANDOFF", "processing handoff BVID differs")
|
source = {"journal": "processing_handoff", "line": index + 1, "sha256": _row_digest(handoff_raw[index])}
|
payload = _expected_outbox_payload(config, "VIDEO_TRANSCRIPTION_READY", source)
|
created.append(_append_outbox(config, "VIDEO_TRANSCRIPTION_READY", source, payload, now.isoformat()))
|
state["cursors"] = {
|
"formal_lines": len(formal_rows), "formal_sha256": hashlib.sha256(formal_payload).hexdigest().upper(),
|
"handoff_lines": len(handoff_rows), "handoff_sha256": hashlib.sha256(handoff_payload).hexdigest().upper(),
|
}
|
_write_state(config, state)
|
return {"status": "RECONCILED", "run_id": active["run_id"], "created_outbox_ids": sorted(set(created)), "cursors": state["cursors"]}
|
|
|
def pending(config: Config) -> dict[str, Any]:
|
with _lock(config):
|
rows = _outbox_rows(config)
|
latest: dict[str, str] = {}
|
created: dict[str, dict[str, Any]] = {}
|
for row in rows:
|
outbox_id = row.get("outbox_id")
|
if isinstance(outbox_id, str):
|
latest[outbox_id] = str(row.get("event"))
|
if row.get("event") == "CREATED":
|
created[outbox_id] = row
|
values = []
|
for key in sorted(created):
|
state = latest.get(key)
|
if state not in {"CREATED", "DISPATCH_INTENT"}:
|
continue
|
values.append({**created[key], "delivery_state": state})
|
return {"status": "PENDING", "count": len(values), "items": values}
|
|
|
def _dispatch_target(config: Config, kind: str) -> str:
|
targets = {
|
"VIDEO_DOWNLOAD_READY": config.video_downloader_thread_id,
|
"VIDEO_TRANSCRIPTION_READY": config.media_thread_id,
|
"MINUTES_READY": config.minutes_thread_id,
|
}
|
target = targets.get(kind)
|
if target is None:
|
raise PipelineError("E_DISPATCH_KIND", "outbox item is not a role handoff")
|
return target
|
|
|
def _dispatch_envelope(config: Config, created: Mapping[str, Any]) -> dict[str, Any]:
|
return {
|
"schema_version": SCHEMA,
|
"type": created["kind"],
|
"outbox_id": created["outbox_id"],
|
"creator_uid": config.creator_uid,
|
"payload": created["payload"],
|
}
|
|
|
def dispatch_intent(config: Config, outbox_id: str, now: datetime) -> dict[str, Any]:
|
with _lock(config):
|
rows = _outbox_rows(config)
|
matches = [row for row in rows if row.get("outbox_id") == outbox_id]
|
created = next((row for row in matches if row.get("event") == "CREATED"), None)
|
if created is None:
|
raise PipelineError("E_OUTBOX", "outbox item is unknown")
|
kind = created["kind"]
|
target = _dispatch_target(config, kind)
|
envelope = _dispatch_envelope(config, created)
|
observed = next((row for row in matches if row.get("event") == "OBSERVED"), None)
|
completed = next((row for row in matches if row.get("event") == "COMPLETE"), None)
|
if completed is not None:
|
return {
|
"status": "DISPATCH_ALREADY_COMPLETE", "outbox_id": outbox_id,
|
"target_thread_id": target, "envelope": envelope,
|
}
|
if observed is not None:
|
return {
|
"status": "DISPATCH_ALREADY_OBSERVED", "outbox_id": outbox_id,
|
"target_thread_id": target, "delivery_id": observed["delivery_id"], "envelope": envelope,
|
}
|
prior = next((row for row in matches if row.get("event") == "DISPATCH_INTENT"), None)
|
if prior is not None:
|
if prior.get("kind") != kind or prior.get("target_thread_id") != target:
|
raise PipelineError("E_OUTBOX", "dispatch intent identity drifted")
|
return {
|
"status": "DISPATCH_INTENT_RESUMED", "outbox_id": outbox_id,
|
"target_thread_id": target, "envelope": envelope,
|
}
|
event = {"schema_version": SCHEMA, "event": "DISPATCH_INTENT", "outbox_id": outbox_id, "kind": kind, "target_thread_id": target, "created_at": now.isoformat()}
|
_append(config.outbox_path, event)
|
return {
|
"status": "DISPATCH_INTENT_DURABLE", "outbox_id": outbox_id,
|
"target_thread_id": target, "envelope": envelope,
|
}
|
|
|
def observe_dispatch(config: Config, outbox_id: str, delivery_id: str, now: datetime) -> dict[str, Any]:
|
if not isinstance(delivery_id, str) or not SAFE_ID.fullmatch(delivery_id):
|
raise PipelineError("E_DISPATCH_RECEIPT", "delivery identity differs")
|
with _lock(config):
|
rows = _outbox_rows(config)
|
matches = [row for row in rows if row.get("outbox_id") == outbox_id]
|
if not any(row.get("event") == "DISPATCH_INTENT" for row in matches):
|
raise PipelineError("E_DISPATCH_RECEIPT", "dispatch intent is absent")
|
if any(row.get("event") == "COMPLETE" for row in matches):
|
raise PipelineError("E_DISPATCH_RECEIPT", "completed dispatch cannot accept a late observation")
|
observed = [row for row in matches if row.get("event") == "OBSERVED"]
|
if observed:
|
if len(observed) == 1 and observed[0].get("delivery_id") == delivery_id:
|
return {"status": "DISPATCH_ALREADY_OBSERVED", "outbox_id": outbox_id, "delivery_id": delivery_id}
|
raise PipelineError("E_DISPATCH_RECEIPT", "dispatch receipt conflicts")
|
event = {
|
"schema_version": SCHEMA, "event": "OBSERVED", "outbox_id": outbox_id,
|
"delivery_id": delivery_id, "observed_at": now.isoformat(),
|
}
|
_append(config.outbox_path, event)
|
return {"status": "DISPATCH_OBSERVED", "outbox_id": outbox_id, "delivery_id": delivery_id}
|
|
|
def _receipt_files(value: Any, config: Config) -> list[dict[str, Any]]:
|
if not isinstance(value, list) or not value:
|
raise PipelineError("E_RECEIPT", "receipt files are empty")
|
files: list[dict[str, Any]] = []
|
for item in value:
|
item = _exact(item, {"path", "bytes", "sha256", "kind"}, "receipt file")
|
if not isinstance(item["kind"], str) or not SAFE_ID.fullmatch(item["kind"]):
|
raise PipelineError("E_RECEIPT", "receipt file kind differs")
|
files.append({**_artifact(item["path"], item["bytes"], item["sha256"], config), "kind": item["kind"]})
|
if len({item["path"] for item in files}) != len(files):
|
raise PipelineError("E_RECEIPT", "receipt files are duplicated")
|
return files
|
|
|
def _terminal_rows(config: Config) -> list[dict[str, Any]]:
|
rows, _ = _read_lines(config.terminals_path, "terminals")
|
seen_sources: set[str] = set()
|
seen_terminals: set[str] = set()
|
for row in rows:
|
if set(row) != {
|
"schema_version", "event", "source_outbox_id", "stable_id", "terminal_id",
|
"receipt_bytes", "receipt_sha256", "files", "committed_at",
|
}:
|
raise PipelineError("E_RECEIPT_CONFLICT", "terminal shape differs")
|
source = row.get("source_outbox_id")
|
terminal = row.get("terminal_id")
|
if (
|
type(row.get("schema_version")) is not int
|
or row.get("schema_version") != SCHEMA
|
or row.get("event") not in {"TRANSCRIPTION_COMPLETE", "MINUTES_COMPLETE"}
|
or not isinstance(source, str)
|
or not re.fullmatch(r"[0-9a-f]{64}", source)
|
or not isinstance(row.get("stable_id"), str)
|
or not isinstance(terminal, str)
|
or not SAFE_ID.fullmatch(terminal)
|
or type(row.get("receipt_bytes")) is not int
|
or row["receipt_bytes"] <= 0
|
or not isinstance(row.get("receipt_sha256"), str)
|
or not SHA256.fullmatch(row["receipt_sha256"])
|
or not isinstance(row.get("files"), list)
|
or not row["files"]
|
or not isinstance(row.get("committed_at"), str)
|
):
|
raise PipelineError("E_RECEIPT_CONFLICT", "terminal identity differs")
|
if not BVID.fullmatch(row["stable_id"]):
|
raise PipelineError("E_RECEIPT_CONFLICT", "terminal stable identity differs")
|
rebound: list[dict[str, Any]] = []
|
for item in row["files"]:
|
item = _exact(item, {"path", "bytes", "sha256", "kind"}, "terminal file")
|
if not isinstance(item["kind"], str) or not SAFE_ID.fullmatch(item["kind"]):
|
raise PipelineError("E_RECEIPT_CONFLICT", "terminal file kind differs")
|
rebound.append({**_artifact(item["path"], item["bytes"], item["sha256"], config), "kind": item["kind"]})
|
if rebound != row["files"]:
|
raise PipelineError("E_RECEIPT_CONFLICT", "terminal files drifted")
|
if source in seen_sources or terminal in seen_terminals:
|
raise PipelineError("E_RECEIPT_CONFLICT", "terminal identity is duplicated")
|
seen_sources.add(source)
|
seen_terminals.add(terminal)
|
return rows
|
|
|
def ingest_receipt(config: Config, receipt_path: Path, now: datetime) -> dict[str, Any]:
|
with _lock(config):
|
value, raw = _strict_json(receipt_path, "downstream receipt")
|
receipt = _exact(value, {"schema_version", "type", "outbox_id", "stable_id", "terminal_id", "status", "files", "created_at"}, "receipt")
|
if (
|
type(receipt["schema_version"]) is not int
|
or receipt["schema_version"] != SCHEMA
|
or receipt["status"] != "COMPLETE"
|
or not isinstance(receipt["outbox_id"], str)
|
or not isinstance(receipt["stable_id"], str)
|
or not isinstance(receipt["terminal_id"], str)
|
or not SAFE_ID.fullmatch(receipt["terminal_id"])
|
):
|
raise PipelineError("E_RECEIPT", "receipt identity differs")
|
kind = receipt["type"]
|
if kind not in {"TRANSCRIPTION_COMPLETE", "MINUTES_COMPLETE"}:
|
raise PipelineError("E_RECEIPT", "receipt type differs")
|
rows = _outbox_rows(config)
|
source = next((row for row in rows if row.get("event") == "CREATED" and row.get("outbox_id") == receipt["outbox_id"]), None)
|
expected_kind = "VIDEO_TRANSCRIPTION_READY" if kind == "TRANSCRIPTION_COMPLETE" else "MINUTES_READY"
|
if source is None or source.get("kind") != expected_kind:
|
raise PipelineError("E_RECEIPT", "receipt source outbox differs")
|
source_events = [row for row in rows if row.get("outbox_id") == receipt["outbox_id"]]
|
if not any(row.get("event") == "DISPATCH_INTENT" for row in source_events):
|
raise PipelineError("E_RECEIPT", "receipt has no durable dispatch intent")
|
if not any(row.get("event") == "OBSERVED" for row in source_events):
|
raise PipelineError("E_RECEIPT", "receipt arrived before durable dispatch observation")
|
expected_stable_id = source.get("payload", {}).get("bvid" if kind == "TRANSCRIPTION_COMPLETE" else "stable_id")
|
if receipt["stable_id"] != expected_stable_id:
|
raise PipelineError("E_RECEIPT", "receipt stable identity differs")
|
files = _receipt_files(receipt["files"], config)
|
receipt_sha = hashlib.sha256(raw).hexdigest().upper()
|
terminals = _terminal_rows(config)
|
by_source = [item for item in terminals if item.get("source_outbox_id") == receipt["outbox_id"]]
|
if by_source and not (len(by_source) == 1 and by_source[0].get("receipt_sha256") == receipt_sha):
|
raise PipelineError("E_RECEIPT_CONFLICT", "source outbox already has a different terminal")
|
if any(item.get("terminal_id") == receipt["terminal_id"] and item.get("source_outbox_id") != receipt["outbox_id"] for item in terminals):
|
raise PipelineError("E_RECEIPT_CONFLICT", "terminal identity is already bound elsewhere")
|
already_committed = bool(by_source)
|
terminal = {
|
"schema_version": SCHEMA, "event": kind, "source_outbox_id": receipt["outbox_id"],
|
"stable_id": receipt["stable_id"], "terminal_id": receipt["terminal_id"],
|
"receipt_bytes": len(raw), "receipt_sha256": receipt_sha, "files": files,
|
"committed_at": now.isoformat(),
|
}
|
if not already_committed:
|
_append(config.terminals_path, terminal)
|
source_identity = {"journal": "terminals", "receipt_sha256": receipt_sha}
|
created: list[str] = []
|
if kind == "TRANSCRIPTION_COMPLETE":
|
created.append(_append_outbox(config, "MINUTES_READY", source_identity, {
|
"stable_id": receipt["stable_id"], "transcript_terminal_id": receipt["terminal_id"], "files": files
|
}, now.isoformat()))
|
created.append(_append_outbox(config, "GIT_DELIVERY_READY", {**source_identity, "projection": "transcript"}, {
|
"stable_id": receipt["stable_id"], "files": files, "reason": "TRANSCRIPT_COMPLETE"
|
}, now.isoformat()))
|
else:
|
created.append(_append_outbox(config, "GIT_DELIVERY_READY", {**source_identity, "projection": "minutes"}, {
|
"stable_id": receipt["stable_id"], "files": files, "reason": "MINUTES_COMPLETE"
|
}, now.isoformat()))
|
if not any(row.get("event") == "COMPLETE" and row.get("outbox_id") == receipt["outbox_id"] for row in rows):
|
_append(config.outbox_path, {
|
"schema_version": SCHEMA, "event": "COMPLETE", "outbox_id": receipt["outbox_id"],
|
"result": kind, "terminal_id": receipt["terminal_id"],
|
"receipt_sha256": receipt_sha, "completed_at": now.isoformat(),
|
})
|
return {
|
"status": "RECEIPT_ALREADY_COMMITTED" if already_committed else "RECEIPT_COMMITTED",
|
"receipt_sha256": receipt_sha,
|
"created_outbox_ids": [] if already_committed else created,
|
}
|
|
|
def finish(config: Config, status: str, now: datetime) -> dict[str, Any]:
|
if status not in {"COMPLETE", "FAILED"}:
|
raise PipelineError("E_SCHEMA", "finish status differs")
|
with _lock(config):
|
state = _load_state(config)
|
active = state["active_run"]
|
if active is None:
|
raise PipelineError("E_NO_ACTIVE_RUN", "there is no active run")
|
event = {
|
"schema_version": SCHEMA, "event": f"RUN_{status}", "run_id": active["run_id"],
|
"slot": active["slot"], "creator_uid": config.creator_uid, "finished_at": now.isoformat(),
|
}
|
_append(config.runs_path, event)
|
state["active_run"] = None
|
_write_state(config, state)
|
return {"status": f"RUN_{status}", "run_id": active["run_id"]}
|
|
|
def _run(command: Sequence[str], cwd: Path, runner: Callable[..., subprocess.CompletedProcess[str]]) -> subprocess.CompletedProcess[str]:
|
return runner(list(command), cwd=cwd, text=True, capture_output=True, check=False)
|
|
|
def _run_env(
|
command: Sequence[str], cwd: Path, runner: Callable[..., subprocess.CompletedProcess[str]], env: Mapping[str, str],
|
) -> subprocess.CompletedProcess[str]:
|
return runner(list(command), cwd=cwd, text=True, capture_output=True, check=False, env=dict(env))
|
|
|
def _run_env_bytes(
|
command: Sequence[str], cwd: Path, runner: Callable[..., subprocess.CompletedProcess[Any]], env: Mapping[str, str],
|
) -> subprocess.CompletedProcess[bytes]:
|
return runner(list(command), cwd=cwd, text=False, capture_output=True, check=False, env=dict(env))
|
|
|
def _run_env_input(
|
command: Sequence[str], cwd: Path, runner: Callable[..., subprocess.CompletedProcess[str]], env: Mapping[str, str], payload: bytes,
|
) -> subprocess.CompletedProcess[str]:
|
return runner(list(command), cwd=cwd, input=payload, capture_output=True, check=False, env=dict(env))
|
|
|
def _git_artifacts(config: Config, files: Any) -> tuple[list[str], list[dict[str, Any]], list[bytes]]:
|
if not isinstance(files, list) or not files:
|
raise PipelineError("E_GIT_SCOPE", "Git file list is empty")
|
allowed: list[str] = []
|
rebound: list[dict[str, Any]] = []
|
blobs: list[bytes] = []
|
for artifact in files:
|
identity_keys = {"path", "bytes", "sha256"}
|
if not isinstance(artifact, dict) or frozenset(artifact) not in {frozenset(identity_keys), frozenset(identity_keys | {"kind"})}:
|
raise PipelineError("E_GIT_SCOPE", "Git artifact shape differs")
|
path = artifact.get("path")
|
size = artifact.get("bytes")
|
digest = artifact.get("sha256")
|
if not isinstance(path, str) or type(size) is not int or size < 0 or not isinstance(digest, str) or not SHA256.fullmatch(digest):
|
raise PipelineError("E_GIT_SCOPE", "Git artifact identity differs")
|
if Path(path).is_absolute():
|
raise PipelineError("E_GIT_SCOPE", "Git path must be project-relative")
|
requested = Path(os.path.abspath(config.project_root / Path(path)))
|
target = requested
|
if not target.exists() and not target.is_symlink():
|
relocated = _relocated_artifact_target(config, Path(path).as_posix(), size, digest.upper())
|
if relocated is not None:
|
target = relocated
|
relative = target.relative_to(config.project_root).as_posix() if _within(target, config.project_root) else ""
|
if (
|
not (_within(target, config.archive_root) or relative in config.git_doc_paths)
|
or target.suffix.lower() not in config.git_extensions
|
or any(str(target).lower().endswith(suffix) for suffix in FORBIDDEN_GIT_SUFFIXES)
|
):
|
raise PipelineError("E_GIT_SCOPE", "Git path escaped its allowlist")
|
if "kind" in artifact and RECEIPT_GIT_KIND_SUFFIX.get(artifact["kind"]) != target.suffix.lower():
|
raise PipelineError("E_GIT_SCOPE", "Git receipt artifact kind differs")
|
payload = _stable_artifact_bytes(target, config.archive_root if _within(target, config.archive_root) else config.project_root)
|
current = {"path": relative, "bytes": len(payload), "sha256": hashlib.sha256(payload).hexdigest().upper()}
|
expected_identity = {"bytes": artifact["bytes"], "sha256": artifact["sha256"]}
|
if {"bytes": current["bytes"], "sha256": current["sha256"]} != expected_identity:
|
raise PipelineError("E_GIT_SCOPE", "Git artifact identity drifted")
|
allowed.append(current["path"])
|
rebound.append(current)
|
blobs.append(payload)
|
if len(set(allowed)) != len(allowed):
|
raise PipelineError("E_GIT_SCOPE", "Git paths are duplicated")
|
return allowed, rebound, blobs
|
|
|
def _git_remove_paths(config: Config, value: Any) -> list[str]:
|
if value is None:
|
return []
|
if not isinstance(value, list) or not value or any(not isinstance(item, str) or not item for item in value):
|
raise PipelineError("E_GIT_SCOPE", "Git removal scope differs")
|
normalized: list[str] = []
|
for item in value:
|
if Path(item).is_absolute():
|
raise PipelineError("E_GIT_SCOPE", "Git removal path must be project-relative")
|
target = Path(os.path.abspath(config.project_root / Path(item)))
|
if not _within(target, config.archive_root) or any(str(target).lower().endswith(suffix) for suffix in FORBIDDEN_GIT_SUFFIXES):
|
raise PipelineError("E_GIT_SCOPE", "Git removal path escaped the archive")
|
if target.exists() or target.is_symlink():
|
raise PipelineError("E_GIT_SCOPE", "relocated Git source still exists in the worktree")
|
ancestor = target.parent
|
while not ancestor.exists() and ancestor != config.archive_root:
|
ancestor = ancestor.parent
|
_strict_chain(config.archive_root, ancestor, final_file=False)
|
normalized.append(target.relative_to(config.project_root).as_posix())
|
if normalized != sorted(set(normalized)):
|
raise PipelineError("E_GIT_SCOPE", "Git removal paths are not canonical")
|
return normalized
|
|
|
def _git_expected_paths(payload: Mapping[str, Any]) -> list[str]:
|
files = payload.get("files")
|
add_paths = [item.get("path") for item in files] if isinstance(files, list) else []
|
removals = payload.get("remove_paths")
|
if removals is None:
|
return add_paths
|
if not isinstance(removals, list):
|
return []
|
return sorted([*add_paths, *removals])
|
|
|
def _remove_task_index(path: Path, state_dir: Path) -> None:
|
if not path.exists() and not path.is_symlink():
|
return
|
_strict_chain(state_dir, path, final_file=True)
|
path.unlink()
|
|
|
def _git_batch(rows: Sequence[Mapping[str, Any]], item: Mapping[str, Any]) -> tuple[str, list[str]]:
|
created_at = item.get("created_at")
|
if not isinstance(created_at, str):
|
raise PipelineError("E_GIT_INDEX_DIRTY", "Git batch creation identity differs")
|
outbox_ids = sorted(
|
row["outbox_id"] for row in rows
|
if row.get("event") == "CREATED"
|
and row.get("kind") == "GIT_DELIVERY_READY"
|
and row.get("created_at") == created_at
|
)
|
if not outbox_ids or item.get("outbox_id") not in outbox_ids:
|
raise PipelineError("E_GIT_INDEX_DIRTY", "Git batch membership differs")
|
binding = {"created_at": created_at, "outbox_ids": outbox_ids}
|
batch_id = hashlib.sha256(_canonical(binding)).hexdigest().upper()
|
return batch_id, outbox_ids
|
|
|
def _decode_git_paths(result: subprocess.CompletedProcess[Any], error_code: str) -> list[str]:
|
if result.returncode != 0 or not isinstance(result.stdout, bytes):
|
raise PipelineError(error_code, "Git staged paths are unavailable")
|
try:
|
return [part.decode("utf-8").replace("\\", "/") for part in result.stdout.split(b"\0") if part]
|
except UnicodeDecodeError as exc:
|
raise PipelineError(error_code, "Git staged path encoding differs") from exc
|
|
|
def _shared_index_snapshot(
|
config: Config,
|
baseline_head: str,
|
runner: Callable[..., subprocess.CompletedProcess[Any]],
|
) -> dict[str, Any]:
|
if not isinstance(baseline_head, str) or not re.fullmatch(r"[0-9a-f]{40,64}", baseline_head):
|
raise PipelineError("E_GIT_INDEX_DIRTY", "Git baseline HEAD identity differs")
|
index_path = Path(os.path.abspath(config.project_root / ".git" / "index"))
|
payload = _stable_artifact_bytes(index_path, config.project_root)
|
chain = _strict_chain(config.project_root, index_path, final_file=True)
|
staged = _run_env_bytes(
|
["git", "diff", "--cached", "--name-only", "-z", baseline_head, "--"],
|
config.project_root,
|
runner,
|
os.environ,
|
)
|
return {
|
"index_bytes": len(payload),
|
"index_sha256": hashlib.sha256(payload).hexdigest().upper(),
|
"index_identity": list(chain[-1][1]),
|
"staged_paths": _decode_git_paths(staged, "E_GIT_INDEX_DIRTY"),
|
}
|
|
|
def _git_guard_value(
|
config: Config,
|
batch_id: str,
|
outbox_ids: Sequence[str],
|
baseline_head: str,
|
snapshot: Mapping[str, Any],
|
) -> dict[str, Any]:
|
return {
|
"schema_version": SCHEMA,
|
"task_id": TASK_ID,
|
"creator_uid": config.creator_uid,
|
"batch_id": batch_id,
|
"outbox_ids": list(outbox_ids),
|
"baseline_head": baseline_head,
|
"index_bytes": snapshot["index_bytes"],
|
"index_sha256": snapshot["index_sha256"],
|
"index_identity": snapshot["index_identity"],
|
"staged_paths": snapshot["staged_paths"],
|
}
|
|
|
@contextlib.contextmanager
|
def _open_no_write_handle(path: Path, error_code: str) -> Iterator[Any]:
|
descriptor: int | None = None
|
stream = None
|
try:
|
if os.name == "nt":
|
import ctypes # noqa: PLC0415
|
import msvcrt # noqa: PLC0415
|
|
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
|
create_file = kernel32.CreateFileW
|
create_file.argtypes = (
|
ctypes.c_wchar_p, ctypes.c_uint32, ctypes.c_uint32, ctypes.c_void_p,
|
ctypes.c_uint32, ctypes.c_uint32, ctypes.c_void_p,
|
)
|
create_file.restype = ctypes.c_void_p
|
close_handle = kernel32.CloseHandle
|
close_handle.argtypes = (ctypes.c_void_p,)
|
close_handle.restype = ctypes.c_int
|
handle = create_file(
|
str(path), 0x80000000, 0x00000001, None, 3,
|
0x00000080 | 0x00200000 | 0x08000000, None,
|
)
|
if handle in (None, ctypes.c_void_p(-1).value):
|
raise OSError(ctypes.get_last_error(), "CreateFileW failed", str(path))
|
try:
|
descriptor = msvcrt.open_osfhandle(int(handle), os.O_RDONLY | getattr(os, "O_BINARY", 0))
|
handle = None
|
finally:
|
if handle is not None:
|
close_handle(handle)
|
else:
|
descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
|
import fcntl # noqa: PLC0415
|
fcntl.flock(descriptor, fcntl.LOCK_SH | fcntl.LOCK_NB)
|
stream = os.fdopen(descriptor, "rb", closefd=True)
|
descriptor = None
|
yield stream
|
except PipelineError:
|
raise
|
except OSError as exc:
|
raise PipelineError(error_code, "stable file handle is unavailable") from exc
|
finally:
|
if stream is not None:
|
stream.close()
|
elif descriptor is not None:
|
os.close(descriptor)
|
|
|
def _read_exact_held_file(
|
stream: Any,
|
path: Path,
|
root: Path,
|
expected_chain: tuple[tuple[str, tuple[int, ...]], ...],
|
error_code: str,
|
) -> bytes:
|
opened_before = os.fstat(stream.fileno())
|
stream.seek(0)
|
payload = stream.read()
|
opened_after = os.fstat(stream.fileno())
|
chain = _strict_chain(root, path, final_file=True)
|
if (
|
not stat.S_ISREG(opened_before.st_mode)
|
or _is_reparse(opened_before)
|
or _path_handle_identity(opened_before) != expected_chain[-1][1][:6] + expected_chain[-1][1][7:]
|
or _stat_identity(opened_before) != _stat_identity(opened_after)
|
or chain != expected_chain
|
):
|
raise PipelineError(error_code, "held file identity drifted")
|
return payload
|
|
|
@contextlib.contextmanager
|
def _open_no_write_delete_handle(path: Path, error_code: str) -> Iterator[Any]:
|
descriptor: int | None = None
|
stream = None
|
try:
|
if os.name == "nt":
|
import ctypes # noqa: PLC0415
|
import msvcrt # noqa: PLC0415
|
|
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
|
create_file = kernel32.CreateFileW
|
create_file.argtypes = (
|
ctypes.c_wchar_p, ctypes.c_uint32, ctypes.c_uint32, ctypes.c_void_p,
|
ctypes.c_uint32, ctypes.c_uint32, ctypes.c_void_p,
|
)
|
create_file.restype = ctypes.c_void_p
|
close_handle = kernel32.CloseHandle
|
close_handle.argtypes = (ctypes.c_void_p,)
|
close_handle.restype = ctypes.c_int
|
handle = create_file(
|
str(path), 0x80000000 | 0x00010000, 0x00000001, None, 3,
|
0x00000080 | 0x00200000 | 0x08000000, None,
|
)
|
if handle in (None, ctypes.c_void_p(-1).value):
|
raise OSError(ctypes.get_last_error(), "CreateFileW failed", str(path))
|
try:
|
descriptor = msvcrt.open_osfhandle(int(handle), os.O_RDONLY | getattr(os, "O_BINARY", 0))
|
handle = None
|
finally:
|
if handle is not None:
|
close_handle(handle)
|
else:
|
descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
|
import fcntl # noqa: PLC0415
|
fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
stream = os.fdopen(descriptor, "rb", closefd=True)
|
descriptor = None
|
yield stream
|
except PipelineError:
|
raise
|
except OSError as exc:
|
raise PipelineError(error_code, "stable deletable file handle is unavailable") from exc
|
finally:
|
if stream is not None:
|
stream.close()
|
elif descriptor is not None:
|
os.close(descriptor)
|
|
|
def _mark_held_file_for_delete(stream: Any, path: Path) -> None:
|
if os.name == "nt":
|
import ctypes # noqa: PLC0415
|
import msvcrt # noqa: PLC0415
|
|
class FileDispositionInfo(ctypes.Structure):
|
_fields_ = [("DeleteFile", ctypes.c_int)]
|
|
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
|
setter = kernel32.SetFileInformationByHandle
|
setter.argtypes = (ctypes.c_void_p, ctypes.c_int, ctypes.c_void_p, ctypes.c_uint32)
|
setter.restype = ctypes.c_int
|
info = FileDispositionInfo(1)
|
handle = ctypes.c_void_p(msvcrt.get_osfhandle(stream.fileno()))
|
if not setter(handle, 4, ctypes.byref(info), ctypes.sizeof(info)):
|
raise OSError(ctypes.get_last_error(), "SetFileInformationByHandle failed", str(path))
|
return
|
os.unlink(path)
|
|
|
@contextlib.contextmanager
|
def _held_relocation_source(
|
source: Path,
|
source_root: Path,
|
size: int,
|
digest: str,
|
) -> Iterator[tuple[Any, bytes, tuple[tuple[str, tuple[int, ...]], ...]]]:
|
before = _strict_chain(source_root, source, final_file=True)
|
with _open_no_write_delete_handle(source, "E_RELOCATION") as stream:
|
opened_before = os.fstat(stream.fileno())
|
if (
|
_path_handle_identity(opened_before) != before[-1][1][:6] + before[-1][1][7:]
|
or not stat.S_ISREG(opened_before.st_mode)
|
or _is_reparse(opened_before)
|
):
|
raise PipelineError("E_RELOCATION", "migration source path and handle differ")
|
stream.seek(0)
|
payload = stream.read()
|
opened_after = os.fstat(stream.fileno())
|
after = _strict_chain(source_root, source, final_file=True)
|
if (
|
len(payload) != size
|
or hashlib.sha256(payload).hexdigest().upper() != digest
|
or _stat_identity(opened_before) != _stat_identity(opened_after)
|
or before != after
|
):
|
raise PipelineError("E_RELOCATION", "migration source identity differs")
|
yield stream, payload, before
|
|
|
def _delete_held_relocation_source(
|
stream: Any,
|
source: Path,
|
source_root: Path,
|
payload: bytes,
|
before: tuple[tuple[str, tuple[int, ...]], ...],
|
) -> None:
|
stream.seek(0)
|
rebound = stream.read()
|
opened = os.fstat(stream.fileno())
|
after = _strict_chain(source_root, source, final_file=True)
|
if (
|
rebound != payload
|
or before != after
|
or _path_handle_identity(opened) != after[-1][1][:6] + after[-1][1][7:]
|
):
|
raise PipelineError("E_RELOCATION", "migration source drifted before handle-bound delete")
|
try:
|
_mark_held_file_for_delete(stream, source)
|
except OSError as exc:
|
raise PipelineError("E_RELOCATION", "migration source handle delete failed") from exc
|
|
|
@contextlib.contextmanager
|
def _held_exact_git_artifacts(config: Config, allowed: Sequence[str], blobs: Sequence[bytes]) -> Iterator[None]:
|
with contextlib.ExitStack() as stack:
|
for relative, expected in zip(allowed, blobs, strict=True):
|
target = Path(os.path.abspath(config.project_root / relative))
|
governed_root = config.archive_root if _within(target, config.archive_root) else config.project_root
|
before = _strict_chain(governed_root, target, final_file=True)
|
stream = stack.enter_context(_open_no_write_handle(target, "E_GIT_SCOPE"))
|
opened_before = os.fstat(stream.fileno())
|
if (
|
_path_handle_identity(opened_before) != before[-1][1][:6] + before[-1][1][7:]
|
or not stat.S_ISREG(opened_before.st_mode)
|
or _is_reparse(opened_before)
|
):
|
raise PipelineError("E_GIT_SCOPE", "published artifact path and handle differ")
|
stream.seek(0)
|
payload = stream.read()
|
opened_after = os.fstat(stream.fileno())
|
after = _strict_chain(governed_root, target, final_file=True)
|
if payload != expected or _stat_identity(opened_before) != _stat_identity(opened_after) or before != after:
|
raise PipelineError("E_GIT_SCOPE", "published artifact changed before terminal append")
|
yield
|
|
|
@contextlib.contextmanager
|
def _held_git_ref_locks(
|
config: Config,
|
runner: Callable[..., subprocess.CompletedProcess[Any]],
|
) -> Iterator[None]:
|
symbolic = _run(["git", "symbolic-ref", "-q", "HEAD"], config.project_root, runner)
|
expected_local = f"refs/heads/{config.git_branch}"
|
local_ref = symbolic.stdout.strip() if isinstance(symbolic.stdout, str) else ""
|
git_dir_result = _run(["git", "rev-parse", "--absolute-git-dir"], config.project_root, runner)
|
git_dir_text = git_dir_result.stdout.strip() if isinstance(git_dir_result.stdout, str) else ""
|
if symbolic.returncode != 0 or local_ref != expected_local or git_dir_result.returncode != 0 or not git_dir_text:
|
raise PipelineError("E_GIT_PUSH", "published branch ref identity differs")
|
git_dir = Path(os.path.abspath(git_dir_text))
|
_strict_chain(config.project_root, git_dir, final_file=False)
|
lock_paths = sorted({
|
git_dir / "HEAD.lock",
|
git_dir / f"{expected_local}.lock",
|
git_dir / f"refs/remotes/{config.git_remote}/{config.git_branch}.lock",
|
}, key=lambda value: os.path.normcase(str(value)))
|
with contextlib.ExitStack() as stack:
|
for lock_path in lock_paths:
|
_strict_chain(git_dir, lock_path.parent, final_file=False)
|
try:
|
_create_new(lock_path, b"mbx-published-no-changes-lock\n")
|
except PipelineError as exc:
|
raise PipelineError("E_GIT_PUSH", "published Git ref lock is unavailable") from exc
|
|
def cleanup(path: Path = lock_path) -> None:
|
with contextlib.suppress(OSError, PipelineError):
|
_strict_chain(git_dir, path, final_file=True)
|
path.unlink()
|
|
stack.callback(cleanup)
|
stream = stack.enter_context(_open_no_write_handle(lock_path, "E_GIT_PUSH"))
|
if stream.read() != b"mbx-published-no-changes-lock\n":
|
raise PipelineError("E_GIT_PUSH", "published Git ref lock identity differs")
|
yield
|
|
|
@contextlib.contextmanager
|
def _published_head_with_exact_artifacts(
|
config: Config,
|
allowed: Sequence[str],
|
blobs: Sequence[bytes],
|
remove_paths: Sequence[str],
|
runner: Callable[..., subprocess.CompletedProcess[Any]],
|
) -> Iterator[str | None]:
|
head = _run(["git", "rev-parse", "HEAD"], config.project_root, runner)
|
head_sha = head.stdout.strip() if isinstance(head.stdout, str) else ""
|
if head.returncode != 0 or not re.fullmatch(r"[0-9a-f]{40,64}", head_sha):
|
yield None
|
return
|
for relative, payload in zip(allowed, blobs, strict=True):
|
committed = _run_env_bytes(
|
["git", "show", f"{head_sha}:{relative}"],
|
config.project_root,
|
runner,
|
os.environ,
|
)
|
if committed.returncode != 0 or not isinstance(committed.stdout, bytes) or committed.stdout != payload:
|
yield None
|
return
|
for relative in remove_paths:
|
committed = _run_env_bytes(
|
["git", "show", f"{head_sha}:{relative}"],
|
config.project_root,
|
runner,
|
os.environ,
|
)
|
if committed.returncode == 0:
|
yield None
|
return
|
preliminary_remote = _run(
|
["git", "rev-parse", f"refs/remotes/{config.git_remote}/{config.git_branch}"],
|
config.project_root,
|
runner,
|
)
|
preliminary_remote_sha = preliminary_remote.stdout.strip() if isinstance(preliminary_remote.stdout, str) else ""
|
if preliminary_remote.returncode != 0 or preliminary_remote_sha != head_sha:
|
raise PipelineError("E_GIT_PUSH", "exact artifacts are not bound to the published branch")
|
with _held_exact_git_artifacts(config, allowed, blobs), _held_git_ref_locks(config, runner):
|
rebound_head = _run(["git", "rev-parse", "HEAD"], config.project_root, runner)
|
remote = _run(
|
["git", "rev-parse", f"refs/remotes/{config.git_remote}/{config.git_branch}"],
|
config.project_root,
|
runner,
|
)
|
rebound_sha = rebound_head.stdout.strip() if isinstance(rebound_head.stdout, str) else ""
|
remote_sha = remote.stdout.strip() if isinstance(remote.stdout, str) else ""
|
if rebound_head.returncode != 0 or rebound_sha != head_sha or remote.returncode != 0 or remote_sha != head_sha:
|
raise PipelineError("E_GIT_PUSH", "exact artifacts are not bound to the stable published branch")
|
for relative, payload in zip(allowed, blobs, strict=True):
|
committed = _run_env_bytes(
|
["git", "show", f"{head_sha}:{relative}"],
|
config.project_root,
|
runner,
|
os.environ,
|
)
|
if committed.returncode != 0 or not isinstance(committed.stdout, bytes) or committed.stdout != payload:
|
raise PipelineError("E_GIT_SCOPE", "published artifact commit binding drifted")
|
for relative in remove_paths:
|
committed = _run_env_bytes(
|
["git", "show", f"{head_sha}:{relative}"],
|
config.project_root,
|
runner,
|
os.environ,
|
)
|
target = Path(os.path.abspath(config.project_root / relative))
|
if committed.returncode == 0 or target.exists() or target.is_symlink():
|
raise PipelineError("E_GIT_SCOPE", "published relocation removal binding drifted")
|
yield head_sha
|
|
|
def _read_git_guard(config: Config, path: Path) -> dict[str, Any]:
|
value, _ = _strict_json(path, "Git shared-index guard")
|
guard = _exact(value, {
|
"schema_version", "task_id", "creator_uid", "batch_id", "outbox_ids", "baseline_head",
|
"index_bytes", "index_sha256", "index_identity", "staged_paths",
|
}, "Git shared-index guard")
|
if (
|
type(guard["schema_version"]) is not int or guard["schema_version"] != SCHEMA
|
or guard["task_id"] != TASK_ID or guard["creator_uid"] != config.creator_uid
|
or not isinstance(guard["batch_id"], str) or not SHA256.fullmatch(guard["batch_id"])
|
or not isinstance(guard["outbox_ids"], list) or not guard["outbox_ids"]
|
or any(not isinstance(value, str) or not SHA256.fullmatch(value.upper()) for value in guard["outbox_ids"])
|
or not isinstance(guard["baseline_head"], str) or not re.fullmatch(r"[0-9a-f]{40,64}", guard["baseline_head"])
|
or type(guard["index_bytes"]) is not int or guard["index_bytes"] < 0
|
or not isinstance(guard["index_sha256"], str) or not SHA256.fullmatch(guard["index_sha256"])
|
or not isinstance(guard["index_identity"], list) or not guard["index_identity"]
|
or any(type(value) is not int for value in guard["index_identity"])
|
or not isinstance(guard["staged_paths"], list)
|
or any(not isinstance(value, str) for value in guard["staged_paths"])
|
):
|
raise PipelineError("E_GIT_INDEX_DIRTY", "Git shared-index guard identity differs")
|
return dict(guard)
|
|
|
def _ensure_git_index_guard(
|
config: Config,
|
rows: Sequence[Mapping[str, Any]],
|
item: Mapping[str, Any],
|
runner: Callable[..., subprocess.CompletedProcess[Any]],
|
) -> dict[str, Any]:
|
batch_id, outbox_ids = _git_batch(rows, item)
|
guard_path = config.git_index_guard_path(batch_id)
|
if guard_path.exists() or guard_path.is_symlink():
|
_strict_chain(config.state_dir, guard_path, final_file=True)
|
guard = _read_git_guard(config, guard_path)
|
else:
|
batch_events = [
|
row for row in rows
|
if row.get("outbox_id") in outbox_ids and row.get("event") in {"GIT_COMMIT_INTENT", "COMMIT_CREATED", "COMPLETE"}
|
]
|
if batch_events:
|
raise PipelineError("E_GIT_INDEX_DIRTY", "Git baseline guard is absent after batch progress")
|
head = _run(["git", "rev-parse", "HEAD"], config.project_root, runner)
|
baseline_head = head.stdout.strip() if isinstance(head.stdout, str) else ""
|
if head.returncode != 0 or not re.fullmatch(r"[0-9a-f]{40,64}", baseline_head):
|
raise PipelineError("E_GIT_INDEX_DIRTY", "Git baseline HEAD is unavailable")
|
snapshot = _shared_index_snapshot(config, baseline_head, runner)
|
guard = _git_guard_value(config, batch_id, outbox_ids, baseline_head, snapshot)
|
_create_new(guard_path, _canonical(guard))
|
_strict_chain(config.state_dir, guard_path, final_file=True)
|
guard = _read_git_guard(config, guard_path)
|
expected = _git_guard_value(
|
config,
|
batch_id,
|
outbox_ids,
|
guard["baseline_head"],
|
_shared_index_snapshot(config, guard["baseline_head"], runner),
|
)
|
if guard != expected:
|
raise PipelineError("E_GIT_INDEX_DIRTY", "shared Git index changed after the batch baseline")
|
return guard
|
|
|
def recover_git_index_guard(
|
config: Config,
|
outbox_id: str,
|
baseline_head: str,
|
expected_bytes: int,
|
expected_sha256: str,
|
*,
|
runner: Callable[..., subprocess.CompletedProcess[Any]] = subprocess.run,
|
) -> dict[str, Any]:
|
with _lock(config):
|
rows = _outbox_rows(config)
|
item = next((row for row in rows if row.get("event") == "CREATED" and row.get("outbox_id") == outbox_id), None)
|
if item is None or item.get("kind") != "GIT_DELIVERY_READY":
|
raise PipelineError("E_GIT_SCOPE", "Git recovery outbox differs")
|
batch_id, outbox_ids = _git_batch(rows, item)
|
guard_path = config.git_index_guard_path(batch_id)
|
if (
|
not isinstance(baseline_head, str) or not re.fullmatch(r"[0-9a-f]{40,64}", baseline_head)
|
or type(expected_bytes) is not int or expected_bytes < 0
|
or not isinstance(expected_sha256, str) or not SHA256.fullmatch(expected_sha256.upper())
|
):
|
raise PipelineError("E_GIT_INDEX_DIRTY", "expected shared-index identity differs")
|
expected_sha256 = expected_sha256.upper()
|
if guard_path.exists() or guard_path.is_symlink():
|
try:
|
_strict_chain(config.state_dir, guard_path, final_file=True)
|
guard = _read_git_guard(config, guard_path)
|
if (
|
guard["batch_id"] != batch_id
|
or guard["outbox_ids"] != outbox_ids
|
or guard["baseline_head"] != baseline_head
|
or guard["index_bytes"] != expected_bytes
|
or guard["index_sha256"] != expected_sha256
|
):
|
raise PipelineError("E_GIT_INDEX_DIRTY", "durable Git guard recovery binding differs")
|
fresh = _git_guard_value(
|
config,
|
batch_id,
|
outbox_ids,
|
guard["baseline_head"],
|
_shared_index_snapshot(config, guard["baseline_head"], runner),
|
)
|
if guard != fresh:
|
raise PipelineError("E_GIT_INDEX_DIRTY", "shared Git index changed after the durable guard")
|
except PipelineError as exc:
|
if exc.code == "E_GIT_INDEX_DIRTY":
|
raise
|
raise PipelineError("E_GIT_INDEX_DIRTY", "durable Git guard cannot be revalidated") from exc
|
return {"status": "GIT_INDEX_GUARD_ALREADY_DURABLE", "batch_id": batch_id}
|
batch_rows = [row for row in rows if row.get("outbox_id") in outbox_ids]
|
intents = [row for row in batch_rows if row.get("event") == "GIT_COMMIT_INTENT"]
|
commits = [row for row in batch_rows if row.get("event") == "COMMIT_CREATED"]
|
if not intents or len(intents) != len(commits) or intents[0].get("parent_sha") != baseline_head:
|
raise PipelineError("E_GIT_INDEX_DIRTY", "Git recovery history does not bind the baseline")
|
expected_parent = baseline_head
|
for intent, commit in zip(intents, commits, strict=True):
|
if intent.get("parent_sha") != expected_parent or commit.get("parent_sha") != expected_parent:
|
raise PipelineError("E_GIT_INDEX_DIRTY", "Git recovery commit chain differs")
|
expected_parent = commit.get("commit_sha")
|
head = _run(["git", "rev-parse", "HEAD"], config.project_root, runner)
|
if head.returncode != 0 or head.stdout.strip() != expected_parent:
|
raise PipelineError("E_GIT_INDEX_DIRTY", "Git recovery HEAD differs")
|
snapshot = _shared_index_snapshot(config, baseline_head, runner)
|
if snapshot["index_bytes"] != expected_bytes or snapshot["index_sha256"] != expected_sha256:
|
raise PipelineError("E_GIT_INDEX_DIRTY", "shared Git index differs from its frozen first-run bytes")
|
guard = _git_guard_value(config, batch_id, outbox_ids, baseline_head, snapshot)
|
_create_new(guard_path, _canonical(guard))
|
if _read_git_guard(config, guard_path) != guard:
|
raise PipelineError("E_DURABILITY", "Git shared-index guard readback differs")
|
return {"status": "GIT_INDEX_GUARD_RECOVERED", "batch_id": batch_id, "outbox_count": len(outbox_ids)}
|
|
|
def git_preflight(
|
config: Config,
|
*,
|
runner: Callable[..., subprocess.CompletedProcess[Any]] = subprocess.run,
|
) -> dict[str, Any]:
|
"""Read-only shared-index diagnostics; never repairs or rewrites the index."""
|
index_path = Path(os.path.abspath(config.project_root / ".git" / "index"))
|
before_chain = _strict_chain(config.project_root, index_path, final_file=True)
|
with _open_no_write_handle(index_path, "E_GIT_INDEX_DIRTY") as index_stream:
|
before = _read_exact_held_file(
|
index_stream,
|
index_path,
|
config.project_root,
|
before_chain,
|
"E_GIT_INDEX_DIRTY",
|
)
|
index_view = index_path
|
if os.name != "nt":
|
proc_view = Path(f"/proc/self/fd/{index_stream.fileno()}")
|
if proc_view.exists():
|
index_view = proc_view
|
git_env = {
|
**os.environ,
|
"GIT_OPTIONAL_LOCKS": "0",
|
"GIT_INDEX_FILE": str(index_view),
|
}
|
try:
|
head_result = _run_env(["git", "rev-parse", "HEAD"], config.project_root, runner, git_env)
|
head = head_result.stdout.strip() if isinstance(head_result.stdout, str) else ""
|
if head_result.returncode != 0 or not re.fullmatch(r"[0-9a-f]{40,64}", head):
|
raise PipelineError("E_GIT_INDEX_DIRTY", "Git preflight HEAD is unavailable")
|
staged = _run_env_bytes(
|
["git", "diff", "--cached", "--name-only", "-z", head, "--"],
|
config.project_root,
|
runner,
|
git_env,
|
)
|
staged_paths = _decode_git_paths(staged, "E_GIT_INDEX_DIRTY")
|
_read_exact_held_file(index_stream, index_path, config.project_root, before_chain, "E_GIT_INDEX_DIRTY")
|
archive_relative = config.archive_root.relative_to(config.project_root).as_posix()
|
deleted = _run_env_bytes(
|
["git", "diff", "--cached", "--diff-filter=D", "--name-only", "-z", head, "--", archive_relative],
|
config.project_root,
|
runner,
|
git_env,
|
)
|
deleted_paths = _decode_git_paths(deleted, "E_GIT_INDEX_DIRTY")
|
_read_exact_held_file(index_stream, index_path, config.project_root, before_chain, "E_GIT_INDEX_DIRTY")
|
present_deletions: list[str] = []
|
for relative in deleted_paths:
|
target = Path(os.path.abspath(config.project_root / relative))
|
if target.exists() or target.is_symlink():
|
_strict_chain(config.archive_root, target, final_file=True)
|
present_deletions.append(relative)
|
lock_path = Path(os.path.abspath(config.project_root / ".git" / "index.lock"))
|
stale_lock: dict[str, Any] | None = None
|
if lock_path.exists() or lock_path.is_symlink():
|
lock_payload = _stable_artifact_bytes(lock_path, config.project_root)
|
lock_info = os.lstat(lock_path)
|
stale_lock = {
|
"path": ".git/index.lock",
|
"bytes": len(lock_payload),
|
"sha256": hashlib.sha256(lock_payload).hexdigest().upper(),
|
"mtime_ns": lock_info.st_mtime_ns,
|
}
|
return {
|
"status": "GIT_PREFLIGHT",
|
"head": head,
|
"index_bytes": len(before),
|
"index_sha256": hashlib.sha256(before).hexdigest().upper(),
|
"index_identity": list(before_chain[-1][1]),
|
"index_matches_head": not staged_paths,
|
"staged_paths": staged_paths,
|
"archive_staged_delete_present": present_deletions,
|
"stale_index_lock": stale_lock,
|
}
|
finally:
|
after = _read_exact_held_file(
|
index_stream,
|
index_path,
|
config.project_root,
|
before_chain,
|
"E_GIT_INDEX_MUTATION",
|
)
|
if before != after:
|
raise PipelineError("E_GIT_INDEX_MUTATION", "Git preflight changed the shared index")
|
|
|
def _migration_file_identity(config: Config, path: Path, kind: str, root: Path) -> dict[str, Any]:
|
payload = _stable_artifact_bytes(path, root)
|
return {
|
"kind": kind,
|
"bytes": len(payload),
|
"sha256": hashlib.sha256(payload).hexdigest().upper(),
|
}
|
|
|
def _video_metadata(config: Config) -> dict[str, dict[str, Any]]:
|
rows, _ = _read_lines(config.formal_manifest, "formal manifest", reject_secrets=False)
|
selected: dict[str, dict[str, Any]] = {}
|
for row in rows:
|
stable_id = row.get("stable_id")
|
row_creator_uid = _creator_uid(row)
|
creator_matches = row_creator_uid == config.creator_uid or (
|
row_creator_uid is None and row.get("creator") == config.creator_name
|
)
|
if (
|
row.get("item_type") != "video"
|
or not creator_matches
|
or not isinstance(stable_id, str)
|
or not BVID.fullmatch(stable_id)
|
):
|
continue
|
title = row.get("title")
|
published_at = row.get("published_at")
|
if not isinstance(title, str) or not isinstance(published_at, str):
|
continue
|
_canonical_video_base(stable_id, title, published_at)
|
candidate = {
|
"stable_id": stable_id,
|
"title": title,
|
"published_at": published_at,
|
"complete": row.get("status") == VIDEO_COMPLETE,
|
}
|
existing = selected.get(stable_id)
|
if existing is not None and existing["complete"] and candidate["complete"] and (
|
existing["title"] != title or existing["published_at"] != published_at
|
):
|
raise PipelineError("E_RELOCATION", "completed video metadata is ambiguous")
|
if candidate["complete"] or existing is None:
|
selected[stable_id] = candidate
|
return selected
|
|
|
def _video_artifact_ids(config: Config) -> list[str]:
|
values: set[str] = set()
|
for suffix in (".transcript", ".minutes"):
|
for path in config.archive_root.glob(f"*{suffix}"):
|
name = path.name.removesuffix(suffix)
|
stable_id = name if BVID.fullmatch(name) else name.rsplit("_", 1)[-1]
|
if path.is_dir() and BVID.fullmatch(stable_id):
|
_strict_chain(config.archive_root, path, final_file=False)
|
values.add(stable_id)
|
return sorted(values)
|
|
|
def _git_tracked(config: Config, relative: str, runner: Callable[..., subprocess.CompletedProcess[Any]]) -> bool:
|
result = _run(["git", "ls-files", "--error-unmatch", "--", relative], config.project_root, runner)
|
return result.returncode == 0
|
|
|
def _git_head(config: Config, runner: Callable[..., subprocess.CompletedProcess[Any]]) -> str:
|
result = _run(["git", "rev-parse", "HEAD"], config.project_root, runner)
|
head = result.stdout.strip() if isinstance(result.stdout, str) else ""
|
if result.returncode != 0 or not re.fullmatch(r"[0-9a-f]{40,64}", head):
|
raise PipelineError("E_RELOCATION", "migration Git baseline is unavailable")
|
exists = _run(["git", "cat-file", "-e", f"{head}^{{commit}}"], config.project_root, runner)
|
if exists.returncode != 0:
|
raise PipelineError("E_RELOCATION", "migration Git baseline differs")
|
return head
|
|
|
def _git_blob_at(
|
config: Config,
|
baseline_head: str,
|
relative: str,
|
runner: Callable[..., subprocess.CompletedProcess[Any]],
|
) -> bytes | None:
|
result = _run_env_bytes(
|
["git", "show", f"{baseline_head}:{relative}"],
|
config.project_root,
|
runner,
|
{**os.environ, "GIT_OPTIONAL_LOCKS": "0"},
|
)
|
if result.returncode != 0:
|
return None
|
if not isinstance(result.stdout, bytes):
|
raise PipelineError("E_RELOCATION", "migration Git blob encoding differs")
|
return result.stdout
|
|
|
def _migration_current_identity(
|
config: Config,
|
old_path: Path,
|
new_path: Path,
|
kind: str,
|
old_root: Path,
|
new_root: Path,
|
*,
|
allow_recovery_pair: bool,
|
) -> dict[str, Any]:
|
old_exists = old_path.exists() or old_path.is_symlink()
|
new_exists = new_path.exists() or new_path.is_symlink()
|
if not old_exists and not new_exists:
|
raise PipelineError("E_RELOCATION", "migration artifact is absent")
|
if old_exists and new_exists and not allow_recovery_pair:
|
raise PipelineError("E_RELOCATION", "legacy and canonical artifact both exist")
|
identities: list[dict[str, Any]] = []
|
if old_exists:
|
identities.append(_migration_file_identity(config, old_path, kind, old_root))
|
if new_exists:
|
identities.append(_migration_file_identity(config, new_path, kind, new_root))
|
if len(identities) == 2 and identities[0] != identities[1]:
|
raise PipelineError("E_RELOCATION", "migration recovery pair differs")
|
return identities[0]
|
|
|
def _build_video_artifact_migration_report(
|
config: Config,
|
created_at: str,
|
baseline_head: str,
|
stable_ids: Sequence[str],
|
runner: Callable[..., subprocess.CompletedProcess[Any]],
|
) -> dict[str, Any]:
|
_published_at(created_at, "created_at")
|
if (
|
not isinstance(baseline_head, str)
|
or not re.fullmatch(r"[0-9a-f]{40,64}", baseline_head)
|
or list(stable_ids) != sorted(set(stable_ids))
|
or not stable_ids
|
):
|
raise PipelineError("E_RELOCATION", "migration plan identity differs")
|
exists = _run(["git", "cat-file", "-e", f"{baseline_head}^{{commit}}"], config.project_root, runner)
|
if exists.returncode != 0:
|
raise PipelineError("E_RELOCATION", "migration baseline commit is absent")
|
metadata = _video_metadata(config)
|
items: list[dict[str, Any]] = []
|
remove_paths: list[str] = []
|
kind_order = tuple(PUBLIC_VIDEO_KINDS)
|
for stable_id in stable_ids:
|
if not isinstance(stable_id, str) or not BVID.fullmatch(stable_id):
|
raise PipelineError("E_RELOCATION", "migration stable identity differs")
|
source = metadata.get(stable_id)
|
if source is None or not source["complete"]:
|
raise PipelineError("E_RELOCATION", "legacy video lacks a completed formal identity")
|
canonical_base = _canonical_video_base(stable_id, source["title"], source["published_at"])
|
aliases: list[dict[str, Any]] = []
|
intermediates: list[dict[str, Any]] = []
|
locations = {
|
"transcript_txt": (
|
config.archive_root / f"{stable_id}.transcript" / f"{stable_id}.txt",
|
config.archive_root / f"{canonical_base}.transcript" / f"{canonical_base}.txt",
|
),
|
"transcript_srt": (
|
config.archive_root / f"{stable_id}.transcript" / f"{stable_id}.srt",
|
config.archive_root / f"{canonical_base}.transcript" / f"{canonical_base}.srt",
|
),
|
"transcript_json": (
|
config.archive_root / f"{stable_id}.transcript" / f"{stable_id}.json",
|
config.archive_root / f"{canonical_base}.transcript" / f"{canonical_base}.json",
|
),
|
"minutes_md": (
|
config.archive_root / f"{stable_id}.minutes" / f"{stable_id}.md",
|
config.archive_root / f"{canonical_base}.minutes" / f"{canonical_base}.md",
|
),
|
"minutes_pdf": (
|
config.archive_root / f"{stable_id}.minutes" / f"{stable_id}.pdf",
|
config.archive_root / f"{canonical_base}.minutes" / f"{canonical_base}.pdf",
|
),
|
}
|
for kind in kind_order:
|
old_path, new_path = locations[kind]
|
old_exists = old_path.exists() or old_path.is_symlink()
|
new_exists = new_path.exists() or new_path.is_symlink()
|
if not old_exists and not new_exists:
|
continue
|
identity = _migration_current_identity(
|
config,
|
old_path,
|
new_path,
|
kind,
|
config.archive_root,
|
config.archive_root,
|
allow_recovery_pair=True,
|
)
|
old_relative = _project_relative(config, old_path)
|
new_relative = _project_relative(config, new_path)
|
aliases.append({
|
"kind": kind,
|
"old_path": old_relative,
|
"new_path": new_relative,
|
"bytes": identity["bytes"],
|
"sha256": identity["sha256"],
|
})
|
committed = _git_blob_at(config, baseline_head, old_relative, runner)
|
if committed is not None:
|
remove_paths.append(old_relative)
|
transcript_kinds = {alias["kind"] for alias in aliases if alias["kind"].startswith("transcript_")}
|
minutes_kinds = {alias["kind"] for alias in aliases if alias["kind"].startswith("minutes_")}
|
if transcript_kinds != {"transcript_txt", "transcript_srt", "transcript_json"}:
|
raise PipelineError("E_RELOCATION", "legacy video transcript set is incomplete")
|
if minutes_kinds not in (set(), {"minutes_md", "minutes_pdf"}):
|
raise PipelineError("E_RELOCATION", "legacy video minutes set is incomplete")
|
old_flac = config.archive_root / f"{stable_id}.transcript" / f"{stable_id}.audio.flac"
|
new_flac = config.video_root / "intermediate" / "transcription" / f"{stable_id}.audio.flac"
|
if old_flac.exists() or old_flac.is_symlink() or new_flac.exists() or new_flac.is_symlink():
|
identity = _migration_current_identity(
|
config,
|
old_flac,
|
new_flac,
|
"audio_flac",
|
config.archive_root,
|
config.video_root,
|
allow_recovery_pair=True,
|
)
|
intermediates.append({
|
"kind": "audio_flac",
|
"old_path": _project_relative(config, old_flac),
|
"new_path": str(Path(os.path.abspath(new_flac))),
|
"bytes": identity["bytes"],
|
"sha256": identity["sha256"],
|
})
|
items.append({
|
"stable_id": stable_id,
|
"title": source["title"],
|
"published_at": source["published_at"],
|
"canonical_base": canonical_base,
|
"aliases": aliases,
|
"intermediates": intermediates,
|
})
|
docs: list[dict[str, Any]] = []
|
for relative in sorted(config.git_doc_paths):
|
path = Path(os.path.abspath(config.project_root / relative))
|
identity = _migration_file_identity(config, path, "documentation", config.project_root)
|
docs.append({"path": relative, **identity})
|
report: dict[str, Any] = {
|
"schema_version": SCHEMA,
|
"type": RELOCATION_REPORT_TYPE,
|
"task_id": TASK_ID,
|
"creator_uid": config.creator_uid,
|
"batch_id": "",
|
"created_at": created_at,
|
"baseline_head": baseline_head,
|
"items": items,
|
"docs": docs,
|
"remove_paths": sorted(set(remove_paths)),
|
}
|
report["batch_id"] = _relocation_batch_id(report)
|
return report
|
|
|
def _relocation_outbox_exists(config: Config, report_path: Path, report: Mapping[str, Any]) -> bool:
|
if not config.outbox_path.exists():
|
return False
|
payload = _stable_artifact_bytes(report_path, config.relocation_root)
|
source = {
|
"journal": "relocation_report",
|
"path": _project_relative(config, report_path),
|
"bytes": len(payload),
|
"sha256": hashlib.sha256(payload).hexdigest().upper(),
|
}
|
expected_payload = _project_relocation_payload(config, source)
|
expected_id = _outbox_id("GIT_DELIVERY_READY", source, expected_payload)
|
rows = _outbox_rows(config)
|
created = [
|
row for row in rows
|
if row.get("event") == "CREATED" and row.get("outbox_id") == expected_id
|
]
|
if len(created) > 1:
|
raise PipelineError("E_RELOCATION", "relocation outbox is duplicated")
|
return len(created) == 1
|
|
|
def plan_video_artifact_migration(
|
config: Config,
|
now: datetime,
|
*,
|
runner: Callable[..., subprocess.CompletedProcess[Any]] = subprocess.run,
|
) -> dict[str, Any]:
|
reports = _relocation_reports(config)
|
stable_ids = _video_artifact_ids(config)
|
if reports:
|
report_path, existing = reports[-1]
|
report_ids = [item["stable_id"] for item in existing["items"]]
|
outbox_exists = _relocation_outbox_exists(config, report_path, existing)
|
if not outbox_exists and not set(stable_ids).issubset(set(report_ids)):
|
raise PipelineError("E_RELOCATION", "durable relocation report omitted a legacy identity")
|
rebound = _build_video_artifact_migration_report(
|
config,
|
existing["created_at"],
|
existing["baseline_head"],
|
report_ids,
|
runner,
|
)
|
if rebound != existing:
|
raise PipelineError("E_RELOCATION", "durable relocation report differs from the external plan")
|
if not outbox_exists:
|
return existing
|
new_ids = sorted(set(stable_ids) - set(report_ids))
|
if not new_ids:
|
return existing
|
historical_ids = {
|
item["stable_id"]
|
for _, report in reports
|
for item in report["items"]
|
}
|
if any(stable_id in historical_ids for stable_id in new_ids):
|
raise PipelineError("E_RELOCATION", "legacy video identity was already migrated")
|
stable_ids = new_ids
|
if not stable_ids:
|
raise PipelineError("E_RELOCATION", "no legacy video artifacts were found")
|
return _build_video_artifact_migration_report(
|
config,
|
now.isoformat(),
|
_git_head(config, runner),
|
stable_ids,
|
runner,
|
)
|
|
|
def _ensure_directory_chain(root: Path, target: Path) -> None:
|
if not _within(target, root):
|
raise PipelineError("E_RELOCATION", "migration directory escaped its root")
|
missing: list[Path] = []
|
current = target
|
while not current.exists() and current != root:
|
missing.append(current)
|
current = current.parent
|
_strict_chain(root, current, final_file=False)
|
for path in reversed(missing):
|
path.mkdir()
|
_strict_chain(root, path, final_file=False)
|
|
|
def _move_relocation_file(
|
source_root: Path,
|
target_root: Path,
|
source: Path,
|
target: Path,
|
size: int,
|
digest: str,
|
) -> None:
|
source_exists = source.exists() or source.is_symlink()
|
target_exists = target.exists() or target.is_symlink()
|
if not source_exists:
|
if not target_exists:
|
raise PipelineError("E_RELOCATION", "migration source and target are absent")
|
payload = _stable_artifact_bytes(target, target_root)
|
if len(payload) != size or hashlib.sha256(payload).hexdigest().upper() != digest:
|
raise PipelineError("E_RELOCATION", "migration target identity differs")
|
return
|
with _held_relocation_source(source, source_root, size, digest) as (stream, payload, source_chain):
|
if target_exists:
|
rebound = _stable_artifact_bytes(target, target_root)
|
if rebound != payload:
|
raise PipelineError("E_RELOCATION", "migration recovery target differs")
|
else:
|
_ensure_directory_chain(target_root, target.parent)
|
_create_new(target, payload)
|
rebound = _stable_artifact_bytes(target, target_root)
|
if rebound != payload:
|
raise PipelineError("E_RELOCATION", "migration target readback differs")
|
_delete_held_relocation_source(stream, source, source_root, payload, source_chain)
|
if source.exists() or source.is_symlink():
|
raise PipelineError("E_RELOCATION", "migration source remained after handle-bound delete")
|
|
|
def migrate_video_artifacts(
|
config: Config,
|
now: datetime,
|
*,
|
runner: Callable[..., subprocess.CompletedProcess[Any]] = subprocess.run,
|
) -> dict[str, Any]:
|
with _lock(config):
|
state = _load_state(config)
|
if state["active_run"] is not None:
|
raise PipelineError("E_RUN_ACTIVE", "artifact migration requires no active half-hour run")
|
report = plan_video_artifact_migration(config, now, runner=runner)
|
report_path = config.relocation_root / f"{report['batch_id']}.json"
|
if not report_path.exists() and not report_path.is_symlink():
|
_ensure_directory_chain(config.archive_root, config.relocation_root)
|
_create_new(report_path, _canonical(report))
|
rebound = _read_relocation_report(config, report_path)
|
if rebound != report:
|
raise PipelineError("E_RELOCATION", "durable relocation report differs")
|
for item in report["items"]:
|
for alias in item["aliases"]:
|
source = Path(os.path.abspath(config.project_root / alias["old_path"]))
|
target = Path(os.path.abspath(config.project_root / alias["new_path"]))
|
_move_relocation_file(
|
config.archive_root,
|
config.archive_root,
|
source,
|
target,
|
alias["bytes"],
|
alias["sha256"],
|
)
|
for intermediate in item["intermediates"]:
|
source = Path(os.path.abspath(config.project_root / intermediate["old_path"]))
|
target = Path(os.path.abspath(intermediate["new_path"]))
|
_move_relocation_file(
|
config.archive_root,
|
config.video_root,
|
source,
|
target,
|
intermediate["bytes"],
|
intermediate["sha256"],
|
)
|
for suffix in (".transcript", ".minutes"):
|
legacy = config.archive_root / f"{item['stable_id']}{suffix}"
|
if legacy.exists():
|
_strict_chain(config.archive_root, legacy, final_file=False)
|
try:
|
legacy.rmdir()
|
except OSError as exc:
|
raise PipelineError("E_RELOCATION", "legacy artifact directory is not empty") from exc
|
completed = _build_video_artifact_migration_report(
|
config,
|
report["created_at"],
|
report["baseline_head"],
|
[item["stable_id"] for item in report["items"]],
|
runner,
|
)
|
if completed != report:
|
raise PipelineError("E_RELOCATION", "completed migration differs from its durable external plan")
|
report_payload = _stable_artifact_bytes(report_path, config.relocation_root)
|
source = {
|
"journal": "relocation_report",
|
"path": _project_relative(config, report_path),
|
"bytes": len(report_payload),
|
"sha256": hashlib.sha256(report_payload).hexdigest().upper(),
|
}
|
payload = _project_relocation_payload(config, source)
|
outbox_id = _append_outbox(config, "GIT_DELIVERY_READY", source, payload, report["created_at"])
|
return {
|
"status": "VIDEO_ARTIFACTS_MIGRATED",
|
"batch_id": report["batch_id"],
|
"report": source,
|
"item_count": len(report["items"]),
|
"public_file_count": sum(len(item["aliases"]) for item in report["items"]),
|
"intermediate_count": sum(len(item["intermediates"]) for item in report["items"]),
|
"git_outbox_id": outbox_id,
|
}
|
|
|
def git_deliver(config: Config, outbox_id: str, now: datetime, *, runner: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run) -> dict[str, Any]:
|
with _lock(config):
|
rows = _outbox_rows(config)
|
item = next((row for row in rows if row.get("event") == "CREATED" and row.get("outbox_id") == outbox_id), None)
|
if item is None or item.get("kind") != "GIT_DELIVERY_READY":
|
raise PipelineError("E_GIT_SCOPE", "Git outbox differs")
|
if any(row.get("event") == "COMPLETE" and row.get("outbox_id") == outbox_id for row in rows):
|
return {"status": "GIT_ALREADY_COMPLETE", "outbox_id": outbox_id}
|
_ensure_git_index_guard(config, rows, item, runner)
|
payload = item.get("payload", {})
|
files = payload.get("files")
|
allowed, _, blobs = _git_artifacts(config, files)
|
remove_paths = _git_remove_paths(config, payload.get("remove_paths"))
|
expected_paths = _git_expected_paths(payload)
|
if expected_paths != (allowed if not remove_paths else sorted([*allowed, *remove_paths])):
|
raise PipelineError("E_GIT_SCOPE", "Git staged scope differs from its payload")
|
with _published_head_with_exact_artifacts(config, allowed, blobs, remove_paths, runner) as published_head:
|
if published_head is not None:
|
_append(config.outbox_path, {
|
"schema_version": SCHEMA, "event": "COMPLETE", "outbox_id": outbox_id,
|
"result": "NO_CHANGES", "completed_at": now.isoformat(),
|
})
|
return {
|
"status": "GIT_NO_CHANGES", "outbox_id": outbox_id,
|
"commit_sha": published_head, "files": allowed,
|
}
|
git_intents = [row for row in rows if row.get("event") == "GIT_COMMIT_INTENT" and row.get("outbox_id") == outbox_id]
|
commit_events = [row for row in rows if row.get("event") == "COMMIT_CREATED" and row.get("outbox_id") == outbox_id]
|
if len(git_intents) > 1 or len(commit_events) > 1:
|
raise PipelineError("E_GIT_COMMIT", "Git commit journal is ambiguous")
|
if commit_events:
|
event = commit_events[0]
|
commit_sha = event.get("commit_sha")
|
parent_sha = event.get("parent_sha")
|
if (
|
not isinstance(commit_sha, str) or not re.fullmatch(r"[0-9a-f]{40,64}", commit_sha)
|
or not isinstance(parent_sha, str) or not re.fullmatch(r"[0-9a-f]{40,64}", parent_sha)
|
or event.get("files") != expected_paths
|
):
|
raise PipelineError("E_GIT_COMMIT", "Git commit journal identity differs")
|
exists = _run(["git", "cat-file", "-e", f"{commit_sha}^{{commit}}"], config.project_root, runner)
|
head = _run(["git", "rev-parse", "HEAD"], config.project_root, runner)
|
current_head = head.stdout.strip()
|
if exists.returncode != 0 or head.returncode != 0 or current_head not in {parent_sha, commit_sha}:
|
raise PipelineError("E_GIT_COMMIT", "Git recovery identity differs")
|
if current_head == parent_sha:
|
_git_artifacts(config, files)
|
_git_remove_paths(config, payload.get("remove_paths"))
|
updated = _run(["git", "update-ref", "HEAD", commit_sha, parent_sha], config.project_root, runner)
|
if updated.returncode != 0:
|
raise PipelineError("E_GIT_COMMIT", "Git recovery ref update failed")
|
pushed = _run(["git", "push", config.git_remote, f"{commit_sha}:refs/heads/{config.git_branch}"], config.project_root, runner)
|
if pushed.returncode != 0:
|
raise PipelineError("E_GIT_PUSH", "non-force push failed")
|
_append(config.outbox_path, {
|
"schema_version": SCHEMA, "event": "COMPLETE", "outbox_id": outbox_id,
|
"result": "PUSHED", "commit_sha": commit_sha, "completed_at": now.isoformat(),
|
})
|
return {"status": "GIT_PUSHED", "outbox_id": outbox_id, "commit_sha": commit_sha, "files": allowed}
|
task_index = config.state_dir / f"git-index-{outbox_id}"
|
task_lock = Path(str(task_index) + ".lock")
|
task_env = dict(os.environ)
|
task_env["GIT_INDEX_FILE"] = str(task_index)
|
try:
|
git_intent = git_intents[0] if git_intents else None
|
if git_intent is None:
|
_remove_task_index(task_index, config.state_dir)
|
_remove_task_index(task_lock, config.state_dir)
|
read_tree = _run_env(["git", "read-tree", "HEAD"], config.project_root, runner, task_env)
|
added_ok = True
|
for relative, artifact_payload in zip(allowed, blobs, strict=True):
|
blob = _run_env_input(["git", "hash-object", "-w", "--stdin"], config.project_root, runner, task_env, artifact_payload)
|
blob_sha = blob.stdout.decode("ascii").strip() if isinstance(blob.stdout, bytes) else blob.stdout.strip()
|
if blob.returncode != 0 or not re.fullmatch(r"[0-9a-f]{40,64}", blob_sha):
|
added_ok = False
|
break
|
indexed = _run_env(["git", "update-index", "--add", "--cacheinfo", f"100644,{blob_sha},{relative}"], config.project_root, runner, task_env)
|
if indexed.returncode != 0:
|
added_ok = False
|
break
|
for relative in remove_paths:
|
removed = _run_env(
|
["git", "update-index", "--force-remove", "--", relative],
|
config.project_root,
|
runner,
|
task_env,
|
)
|
if removed.returncode != 0:
|
added_ok = False
|
break
|
# `-z` returns repository path bytes without C quoting or console
|
# code-page conversion. Decode Git's UTF-8 path contract directly
|
# before enforcing the exact task-index allowlist.
|
staged_command = ["git", "diff", "--cached", "--name-only", "-z"]
|
if remove_paths:
|
staged_command.append("--no-renames")
|
staged_command.append("--")
|
staged = _run_env_bytes(
|
staged_command,
|
config.project_root,
|
runner,
|
task_env,
|
)
|
try:
|
staged_paths = [part.decode("utf-8").replace("\\", "/") for part in staged.stdout.split(b"\0") if part]
|
except (AttributeError, UnicodeDecodeError) as exc:
|
raise PipelineError("E_GIT_SCOPE", "task index path encoding differs") from exc
|
if read_tree.returncode != 0 or not added_ok:
|
raise PipelineError("E_GIT_ADD", "task-index preparation failed")
|
if staged.returncode != 0 or set(staged_paths) != set(expected_paths):
|
raise PipelineError(
|
"E_GIT_SCOPE",
|
f"task index escaped the exact allowlist ({len(staged_paths)}/{len(expected_paths)})",
|
)
|
if not staged_paths:
|
_append(config.outbox_path, {"schema_version": SCHEMA, "event": "COMPLETE", "outbox_id": outbox_id, "result": "NO_CHANGES", "completed_at": now.isoformat()})
|
return {"status": "GIT_NO_CHANGES", "outbox_id": outbox_id}
|
parent = _run(["git", "rev-parse", "HEAD"], config.project_root, runner)
|
tree = _run_env(["git", "write-tree"], config.project_root, runner, task_env)
|
parent_sha = parent.stdout.strip()
|
tree_sha = tree.stdout.strip()
|
if parent.returncode != 0 or not re.fullmatch(r"[0-9a-f]{40,64}", parent_sha) or tree.returncode != 0 or not re.fullmatch(r"[0-9a-f]{40,64}", tree_sha):
|
raise PipelineError("E_GIT_COMMIT", "Git parent or tree identity is unavailable")
|
git_intent = {
|
"schema_version": SCHEMA, "event": "GIT_COMMIT_INTENT", "outbox_id": outbox_id,
|
"parent_sha": parent_sha, "tree_sha": tree_sha, "files": expected_paths,
|
"message": (
|
f"chore(project-info): migrate Bilibili artifacts {outbox_id[:12]}"
|
if remove_paths else f"chore(project-info): archive Bilibili dynamic {outbox_id[:12]}"
|
),
|
"author_name": "MB-X Bilibili Pipeline", "author_email": "mbx-bili-pipeline@localhost",
|
"authored_at": now.isoformat(), "created_at": now.isoformat(),
|
}
|
_append(config.outbox_path, git_intent)
|
parent_sha = git_intent["parent_sha"]
|
tree_sha = git_intent["tree_sha"]
|
head = _run(["git", "rev-parse", "HEAD"], config.project_root, runner)
|
tree_exists = _run(["git", "cat-file", "-e", f"{tree_sha}^{{tree}}"], config.project_root, runner)
|
if head.returncode != 0 or head.stdout.strip() != parent_sha or tree_exists.returncode != 0 or git_intent["files"] != expected_paths:
|
raise PipelineError("E_GIT_COMMIT", "Git commit intent recovery identity differs")
|
_git_artifacts(config, files)
|
_git_remove_paths(config, payload.get("remove_paths"))
|
commit_env = dict(os.environ)
|
commit_env.update({
|
"GIT_AUTHOR_NAME": git_intent["author_name"], "GIT_COMMITTER_NAME": git_intent["author_name"],
|
"GIT_AUTHOR_EMAIL": git_intent["author_email"], "GIT_COMMITTER_EMAIL": git_intent["author_email"],
|
"GIT_AUTHOR_DATE": git_intent["authored_at"], "GIT_COMMITTER_DATE": git_intent["authored_at"],
|
})
|
committed = _run_env(["git", "commit-tree", tree_sha, "-p", parent_sha, "-m", git_intent["message"]], config.project_root, runner, commit_env)
|
finally:
|
_remove_task_index(task_index, config.state_dir)
|
_remove_task_index(task_lock, config.state_dir)
|
commit_sha = committed.stdout.strip()
|
if committed.returncode != 0 or not re.fullmatch(r"[0-9a-f]{40,64}", commit_sha):
|
raise PipelineError("E_GIT_COMMIT", "commit identity is unavailable")
|
_append(config.outbox_path, {
|
"schema_version": SCHEMA, "event": "COMMIT_CREATED", "outbox_id": outbox_id,
|
"parent_sha": parent_sha, "tree_sha": tree_sha, "commit_sha": commit_sha,
|
"intent_sha256": hashlib.sha256(_canonical(git_intent)).hexdigest().upper(),
|
"files": expected_paths, "created_at": now.isoformat(),
|
})
|
_git_artifacts(config, files)
|
_git_remove_paths(config, payload.get("remove_paths"))
|
updated = _run(["git", "update-ref", "HEAD", commit_sha, parent_sha], config.project_root, runner)
|
if updated.returncode != 0:
|
raise PipelineError("E_GIT_COMMIT", "task-scoped ref update failed")
|
pushed = _run(["git", "push", config.git_remote, f"{commit_sha}:refs/heads/{config.git_branch}"], config.project_root, runner)
|
if pushed.returncode != 0:
|
raise PipelineError("E_GIT_PUSH", "non-force push failed")
|
event = {"schema_version": SCHEMA, "event": "COMPLETE", "outbox_id": outbox_id, "result": "PUSHED", "commit_sha": commit_sha, "completed_at": now.isoformat()}
|
_append(config.outbox_path, event)
|
return {"status": "GIT_PUSHED", "outbox_id": outbox_id, "commit_sha": commit_sha, "files": allowed}
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
parser = argparse.ArgumentParser(description="Durable Bilibili half-hour pipeline coordinator")
|
parser.add_argument("--config", type=Path, required=True)
|
sub = parser.add_subparsers(dest="command", required=True)
|
for name in ("init", "begin", "reconcile", "pending", "git-preflight", "plan-video-artifact-migration", "migrate-video-artifacts"):
|
child = sub.add_parser(name)
|
if name not in {"pending", "git-preflight"}:
|
child.add_argument("--now")
|
dispatch = sub.add_parser("dispatch-intent")
|
dispatch.add_argument("--outbox-id", required=True)
|
dispatch.add_argument("--now")
|
observed = sub.add_parser("observe-dispatch")
|
observed.add_argument("--outbox-id", required=True)
|
observed.add_argument("--delivery-id", required=True)
|
observed.add_argument("--now")
|
receipt = sub.add_parser("ingest-receipt")
|
receipt.add_argument("--receipt", type=Path, required=True)
|
receipt.add_argument("--now")
|
finish_parser = sub.add_parser("finish")
|
finish_parser.add_argument("--status", choices=["COMPLETE", "FAILED"], required=True)
|
finish_parser.add_argument("--now")
|
git_parser = sub.add_parser("git-deliver")
|
git_parser.add_argument("--outbox-id", required=True)
|
git_parser.add_argument("--now")
|
guard_parser = sub.add_parser("recover-git-index-guard")
|
guard_parser.add_argument("--outbox-id", required=True)
|
guard_parser.add_argument("--baseline-head", required=True)
|
guard_parser.add_argument("--expected-bytes", required=True, type=int)
|
guard_parser.add_argument("--expected-sha256", required=True)
|
return parser
|
|
|
def run(argv: Sequence[str] | None = None) -> tuple[int, dict[str, Any]]:
|
args = build_parser().parse_args(argv)
|
try:
|
config = load_config(Path(os.path.abspath(args.config)))
|
if args.command == "init":
|
result = initialize(config, _now(args.now))
|
elif args.command == "begin":
|
result = begin(config, _now(args.now))
|
elif args.command == "reconcile":
|
result = reconcile(config, _now(args.now))
|
elif args.command == "pending":
|
result = pending(config)
|
elif args.command == "git-preflight":
|
result = git_preflight(config)
|
elif args.command == "plan-video-artifact-migration":
|
result = plan_video_artifact_migration(config, _now(args.now))
|
elif args.command == "migrate-video-artifacts":
|
result = migrate_video_artifacts(config, _now(args.now))
|
elif args.command == "dispatch-intent":
|
result = dispatch_intent(config, args.outbox_id, _now(args.now))
|
elif args.command == "observe-dispatch":
|
result = observe_dispatch(config, args.outbox_id, args.delivery_id, _now(args.now))
|
elif args.command == "ingest-receipt":
|
result = ingest_receipt(config, Path(os.path.abspath(args.receipt)), _now(args.now))
|
elif args.command == "finish":
|
result = finish(config, args.status, _now(args.now))
|
elif args.command == "recover-git-index-guard":
|
result = recover_git_index_guard(
|
config,
|
args.outbox_id,
|
args.baseline_head,
|
args.expected_bytes,
|
args.expected_sha256,
|
)
|
else:
|
result = git_deliver(config, args.outbox_id, _now(args.now))
|
return 0, result
|
except PipelineError as exc:
|
return 2, {"status": "FAILED", "error_code": exc.code}
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
code, result = run(argv)
|
sys.stdout.write(json.dumps(result, ensure_ascii=True, sort_keys=True, separators=(",", ":")) + "\n")
|
return code
|
|
|
if __name__ == "__main__":
|
raise SystemExit(main())
|
