686aaaa421
冻结地图1二十分钟纵切版的平衡、证据与金标登记。 新增 ReferenceAsset/2 清单、策略、release、受信快照及 CLI/Service/acceptance provenance /4 消费链；保持 survivor live、R1 签名与部署关闭。
516 lines
22 KiB
Python
516 lines
22 KiB
Python
"""staged 产物可信快照：验收与发布共用同一套路径、资源和竞态边界。"""
|
||
|
||
import hashlib
|
||
import json
|
||
import errno
|
||
import os
|
||
import stat
|
||
import struct
|
||
import sys
|
||
import unicodedata
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
from types import MappingProxyType
|
||
from collections.abc import Mapping as MappingABC
|
||
from typing import Mapping
|
||
|
||
|
||
# 与 Node playtest-v3 runner 的 MAX_ARTIFACT_FILES / MAX_ARTIFACT_BYTES 完全同口径。
|
||
MAX_ARTIFACT_FILES = 4096
|
||
MAX_ARTIFACT_BYTES = 128 * 1024 * 1024
|
||
_ARTIFACT_HASH_DOMAIN = b"artifact-snapshot/1\0"
|
||
_SNAPSHOT_STREAM_MAGIC = b"artifact-snapshot-stream/1\n"
|
||
_REFERENCE_SNAPSHOT_HASH_DOMAIN = b"reference-asset-consumption-snapshot/1\n"
|
||
|
||
# 参照资产消费门的单条记录边界；全树快照继续使用上面的旧版本上限。
|
||
MAX_SELECTED_FILES = 512
|
||
MAX_SELECTED_FILE_BYTES = 16 * 1024 * 1024
|
||
MAX_SELECTED_RECORD_BYTES = 64 * 1024 * 1024
|
||
MAX_SELECTED_TOTAL_BYTES = 128 * 1024 * 1024
|
||
|
||
|
||
class ArtifactSnapshotError(ValueError):
|
||
"""选择性可信快照的稳定错误，不把机器路径或文件内容放入异常。"""
|
||
|
||
def __init__(self, code: str, logical_path: str) -> None:
|
||
self.code = code
|
||
self.logical_path = logical_path
|
||
super().__init__(f"{code} path={logical_path}")
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ArtifactSnapshot:
|
||
"""一次捕获的不可变文件映射及其 canonical 指纹。"""
|
||
|
||
files: Mapping[str, bytes]
|
||
artifact_hash: str
|
||
file_count: int
|
||
total_bytes: int
|
||
|
||
@property
|
||
def snapshot_hash(self) -> str:
|
||
"""按参照资产消费契约返回本次文件映射的 canonical snapshot hash。"""
|
||
return consumption_snapshot_hash(self.files)
|
||
|
||
|
||
def _artifact_hash(files: Mapping[str, bytes]) -> str:
|
||
"""按 artifact-snapshot/1 计算无结构歧义的 canonical artifactHash。"""
|
||
digest = hashlib.sha256()
|
||
digest.update(_ARTIFACT_HASH_DOMAIN)
|
||
for relative in sorted(files, key=lambda value: value.encode("utf-8")):
|
||
path_bytes = relative.encode("utf-8")
|
||
content = files[relative]
|
||
digest.update(struct.pack(">Q", len(path_bytes)))
|
||
digest.update(path_bytes)
|
||
digest.update(struct.pack(">Q", len(content)))
|
||
digest.update(content)
|
||
return digest.hexdigest()
|
||
|
||
|
||
def consumption_snapshot_hash(files: Mapping[str, bytes]) -> str:
|
||
"""按消费 snapshot/1 域标签和长度向量计算原始文件快照 hash。"""
|
||
digest = hashlib.sha256()
|
||
digest.update(_REFERENCE_SNAPSHOT_HASH_DOMAIN)
|
||
for relative in sorted(files, key=lambda value: value.encode("utf-8")):
|
||
path_bytes = relative.encode("utf-8")
|
||
content = files[relative]
|
||
digest.update(struct.pack(">Q", len(path_bytes)))
|
||
digest.update(path_bytes)
|
||
digest.update(struct.pack(">Q", len(content)))
|
||
digest.update(content)
|
||
return digest.hexdigest()
|
||
|
||
|
||
def _selected_path(value: str | Path) -> str:
|
||
"""校验仓根相对 NFC POSIX 路径，并区分语法错误与越界意图。"""
|
||
if isinstance(value, Path):
|
||
value = value.as_posix()
|
||
if not isinstance(value, str) or not value or len(value) > 1024 or "\x00" in value:
|
||
raise ArtifactSnapshotError("reference_path_invalid", "<path>")
|
||
if value.startswith("/"):
|
||
raise ArtifactSnapshotError("reference_path_escape", "<absolute>")
|
||
if "\\" in value:
|
||
raise ArtifactSnapshotError("reference_path_invalid", "<path>")
|
||
if unicodedata.normalize("NFC", value) != value:
|
||
raise ArtifactSnapshotError("reference_path_invalid", "<path>")
|
||
parts = value.split("/")
|
||
if any(part == ".." for part in parts):
|
||
raise ArtifactSnapshotError("reference_path_escape", "<path>")
|
||
if any(part in ("", ".") for part in parts):
|
||
raise ArtifactSnapshotError("reference_path_invalid", "<path>")
|
||
if any(ord(char) < 0x20 or ord(char) == 0x7F for char in value):
|
||
raise ArtifactSnapshotError("reference_path_invalid", "<path>")
|
||
return value
|
||
|
||
|
||
def _selected_limit(limits, names: tuple[str, ...], default: int) -> int:
|
||
"""读取可收紧的调用方上限，并始终受消费门硬帽约束。"""
|
||
if limits is None:
|
||
value = default
|
||
elif isinstance(limits, MappingABC):
|
||
value = next((limits[name] for name in names if name in limits), default)
|
||
else:
|
||
value = next((getattr(limits, name) for name in names if hasattr(limits, name)), default)
|
||
try:
|
||
value = int(value)
|
||
except (TypeError, ValueError) as exc:
|
||
raise ArtifactSnapshotError("reference_oversize", "<limits>") from exc
|
||
if value < 0:
|
||
raise ArtifactSnapshotError("reference_oversize", "<limits>")
|
||
return min(value, default)
|
||
|
||
|
||
def _selected_error_from_oserror(exc: OSError, logical_path: str, *, directory: bool = False) -> ArtifactSnapshotError:
|
||
"""把受信 fd 边界上的系统错误收敛为批准的稳定错误码。"""
|
||
if exc.errno == errno.ELOOP:
|
||
code = "reference_symlink"
|
||
elif exc.errno == errno.ENOENT:
|
||
code = "reference_missing"
|
||
elif exc.errno == errno.ENOTDIR:
|
||
code = "reference_not_regular"
|
||
else:
|
||
code = "reference_unreadable"
|
||
return ArtifactSnapshotError(code, logical_path)
|
||
|
||
|
||
def _selected_root_fd(root) -> tuple[int, bool]:
|
||
"""打开或复制可信根 fd；返回 fd 与是否需要由本函数关闭的标志。"""
|
||
flags_dir = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
|
||
if isinstance(root, int):
|
||
try:
|
||
root_fd = os.dup(root)
|
||
if not stat.S_ISDIR(os.fstat(root_fd).st_mode):
|
||
os.close(root_fd)
|
||
raise ArtifactSnapshotError("reference_not_regular", "<root>")
|
||
return root_fd, True
|
||
except ArtifactSnapshotError:
|
||
raise
|
||
except OSError as exc:
|
||
raise _selected_error_from_oserror(exc, "<root>", directory=True) from exc
|
||
# 不直接把完整字符串交给 open：绝对/相对路径都从一个锚点目录 fd 开始，
|
||
# 这样 trusted_root 自身及其祖先分量也不会被隐式跟随 symlink。
|
||
root_path = os.fspath(root)
|
||
path_obj = Path(root_path)
|
||
if path_obj.is_absolute():
|
||
try:
|
||
current_fd = os.open(os.path.sep, flags_dir)
|
||
except OSError as exc:
|
||
raise _selected_error_from_oserror(exc, "<root>", directory=True) from exc
|
||
components = list(path_obj.parts[1:])
|
||
else:
|
||
try:
|
||
current_fd = os.open(".", flags_dir)
|
||
except OSError as exc:
|
||
raise _selected_error_from_oserror(exc, "<root>", directory=True) from exc
|
||
components = list(path_obj.parts)
|
||
try:
|
||
if not components:
|
||
if not stat.S_ISDIR(os.fstat(current_fd).st_mode):
|
||
raise ArtifactSnapshotError("reference_not_regular", "<root>")
|
||
return current_fd, True
|
||
for component in components:
|
||
try:
|
||
before = os.stat(component, dir_fd=current_fd, follow_symlinks=False)
|
||
except OSError as exc:
|
||
raise _selected_error_from_oserror(exc, "<root>", directory=True) from exc
|
||
if stat.S_ISLNK(before.st_mode):
|
||
raise ArtifactSnapshotError("reference_symlink", "<root>")
|
||
if not stat.S_ISDIR(before.st_mode):
|
||
raise ArtifactSnapshotError("reference_not_regular", "<root>")
|
||
try:
|
||
child_fd = os.open(component, flags_dir, dir_fd=current_fd)
|
||
except OSError as exc:
|
||
raise _selected_error_from_oserror(exc, "<root>", directory=True) from exc
|
||
opened = os.fstat(child_fd)
|
||
if (opened.st_dev, opened.st_ino) != (before.st_dev, before.st_ino):
|
||
os.close(child_fd)
|
||
raise ArtifactSnapshotError("reference_changed_during_read", "<root>")
|
||
os.close(current_fd)
|
||
current_fd = child_fd
|
||
return current_fd, True
|
||
except Exception:
|
||
try:
|
||
os.close(current_fd)
|
||
except OSError:
|
||
pass
|
||
raise
|
||
|
||
|
||
def _selected_stat(parent_fd: int, name: str, logical_path: str):
|
||
"""从锚定目录 fd 读取目录项 stat，不跟随符号链接。"""
|
||
try:
|
||
return os.stat(name, dir_fd=parent_fd, follow_symlinks=False)
|
||
except OSError as exc:
|
||
raise _selected_error_from_oserror(exc, logical_path) from exc
|
||
|
||
|
||
def _selected_open_and_read(
|
||
parent_fd: int,
|
||
name: str,
|
||
logical_path: str,
|
||
*,
|
||
max_file_bytes: int,
|
||
remaining_record_bytes: int,
|
||
) -> tuple[bytes, int]:
|
||
"""以同一个 fd 完成普通文件确认、流式读取和前后竞态复核。"""
|
||
before = _selected_stat(parent_fd, name, logical_path)
|
||
if stat.S_ISLNK(before.st_mode):
|
||
raise ArtifactSnapshotError("reference_symlink", logical_path)
|
||
if not stat.S_ISREG(before.st_mode):
|
||
raise ArtifactSnapshotError("reference_not_regular", logical_path)
|
||
if before.st_size > max_file_bytes or before.st_size > remaining_record_bytes:
|
||
raise ArtifactSnapshotError("reference_oversize", logical_path)
|
||
|
||
flags_file = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
|
||
try:
|
||
file_fd = os.open(name, flags_file, dir_fd=parent_fd)
|
||
except OSError as exc:
|
||
raise _selected_error_from_oserror(exc, logical_path) from exc
|
||
|
||
try:
|
||
opened = os.fstat(file_fd)
|
||
identity_fields = ("st_dev", "st_ino", "st_mode", "st_size", "st_mtime_ns", "st_ctime_ns")
|
||
if any(getattr(opened, field) != getattr(before, field) for field in identity_fields):
|
||
raise ArtifactSnapshotError("reference_changed_during_read", logical_path)
|
||
if not stat.S_ISREG(opened.st_mode):
|
||
raise ArtifactSnapshotError("reference_not_regular", logical_path)
|
||
|
||
chunks: list[bytes] = []
|
||
read_bytes = 0
|
||
while True:
|
||
try:
|
||
chunk = os.read(file_fd, 1024 * 1024)
|
||
except OSError as exc:
|
||
raise ArtifactSnapshotError("reference_unreadable", logical_path) from exc
|
||
if not chunk:
|
||
break
|
||
read_bytes += len(chunk)
|
||
if read_bytes > max_file_bytes or read_bytes > remaining_record_bytes:
|
||
raise ArtifactSnapshotError("reference_oversize", logical_path)
|
||
chunks.append(chunk)
|
||
|
||
after = os.fstat(file_fd)
|
||
if any(getattr(opened, field) != getattr(after, field) for field in identity_fields):
|
||
raise ArtifactSnapshotError("reference_changed_during_read", logical_path)
|
||
data = b"".join(chunks)
|
||
if len(data) != after.st_size:
|
||
raise ArtifactSnapshotError("reference_changed_during_read", logical_path)
|
||
|
||
# 再查一次父目录项，捕获“检查后替换”为另一个 inode 或 symlink 的 TOCTOU。
|
||
current = _selected_stat(parent_fd, name, logical_path)
|
||
if stat.S_ISLNK(current.st_mode):
|
||
raise ArtifactSnapshotError("reference_symlink", logical_path)
|
||
if any(getattr(current, field) != getattr(before, field) for field in identity_fields):
|
||
raise ArtifactSnapshotError("reference_changed_during_read", logical_path)
|
||
return data, read_bytes
|
||
finally:
|
||
os.close(file_fd)
|
||
|
||
|
||
def _selected_open_parent(root_fd: int, components: list[str], logical_path: str) -> tuple[int, list[int]]:
|
||
"""沿可信根逐级打开目录；每级均禁止 symlink 并复核 inode。"""
|
||
current_fd = os.dup(root_fd)
|
||
opened_fds = [current_fd]
|
||
try:
|
||
for component in components:
|
||
before = _selected_stat(current_fd, component, logical_path)
|
||
if stat.S_ISLNK(before.st_mode):
|
||
raise ArtifactSnapshotError("reference_symlink", logical_path)
|
||
if not stat.S_ISDIR(before.st_mode):
|
||
raise ArtifactSnapshotError("reference_not_regular", logical_path)
|
||
flags_dir = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
|
||
try:
|
||
child_fd = os.open(component, flags_dir, dir_fd=current_fd)
|
||
except OSError as exc:
|
||
raise _selected_error_from_oserror(exc, logical_path, directory=True) from exc
|
||
opened = os.fstat(child_fd)
|
||
identity_fields = ("st_dev", "st_ino", "st_mode")
|
||
if any(getattr(opened, field) != getattr(before, field) for field in identity_fields):
|
||
os.close(child_fd)
|
||
raise ArtifactSnapshotError("reference_changed_during_read", logical_path)
|
||
opened_fds.append(child_fd)
|
||
current_fd = child_fd
|
||
return current_fd, opened_fds
|
||
except Exception:
|
||
for fd in reversed(opened_fds):
|
||
try:
|
||
os.close(fd)
|
||
except OSError:
|
||
pass
|
||
raise
|
||
|
||
|
||
def capture_selected_files(root, paths, limits=None) -> ArtifactSnapshot:
|
||
"""在可信目录 fd 下捕获指定文件，返回与全树快照相同类型的不可变结果。
|
||
|
||
选择性捕获只接收仓根相对 NFC POSIX 路径。每个目录分量和最终文件都通过锚定
|
||
fd 与 ``O_NOFOLLOW`` 打开，读取前后的身份字段和父目录项都会复核，任何异常都
|
||
在构造结果前抛出，避免调用方看到半成品映射。
|
||
"""
|
||
normalized: list[str] = []
|
||
seen: set[str] = set()
|
||
for value in paths:
|
||
logical_path = _selected_path(value)
|
||
if logical_path in seen:
|
||
raise ArtifactSnapshotError("reference_path_invalid", logical_path)
|
||
seen.add(logical_path)
|
||
normalized.append(logical_path)
|
||
normalized.sort(key=lambda value: value.encode("utf-8"))
|
||
|
||
max_files = _selected_limit(limits, ("max_files", "file_limit"), MAX_SELECTED_FILES)
|
||
max_record_bytes = _selected_limit(
|
||
limits,
|
||
("max_record_bytes", "max_bytes", "record_bytes"),
|
||
MAX_SELECTED_RECORD_BYTES,
|
||
)
|
||
max_total_bytes = _selected_limit(
|
||
limits,
|
||
("max_total_bytes", "total_bytes"),
|
||
MAX_SELECTED_TOTAL_BYTES,
|
||
)
|
||
# 单条记录的 64 MiB 地板不能被一次调用的总预算放宽；总预算只可进一步收紧。
|
||
max_record_bytes = min(max_record_bytes, max_total_bytes)
|
||
max_file_bytes = _selected_limit(
|
||
limits,
|
||
("max_file_bytes", "file_bytes"),
|
||
MAX_SELECTED_FILE_BYTES,
|
||
)
|
||
if len(normalized) > max_files:
|
||
raise ArtifactSnapshotError("reference_oversize", "<files>")
|
||
|
||
root_fd, close_root = _selected_root_fd(root)
|
||
files: dict[str, bytes] = {}
|
||
total_bytes = 0
|
||
try:
|
||
for logical_path in normalized:
|
||
components = logical_path.split("/")
|
||
parent_fd, opened_fds = _selected_open_parent(root_fd, components[:-1], logical_path)
|
||
try:
|
||
remaining = max_record_bytes - total_bytes
|
||
if remaining < 0:
|
||
raise ArtifactSnapshotError("reference_oversize", logical_path)
|
||
data, consumed = _selected_open_and_read(
|
||
parent_fd,
|
||
components[-1],
|
||
logical_path,
|
||
max_file_bytes=max_file_bytes,
|
||
remaining_record_bytes=remaining,
|
||
)
|
||
total_bytes += consumed
|
||
files[logical_path] = data
|
||
finally:
|
||
for fd in reversed(opened_fds):
|
||
try:
|
||
os.close(fd)
|
||
except OSError:
|
||
pass
|
||
finally:
|
||
if close_root:
|
||
os.close(root_fd)
|
||
|
||
immutable_files = MappingProxyType(dict(files))
|
||
return ArtifactSnapshot(
|
||
files=immutable_files,
|
||
artifact_hash=_artifact_hash(immutable_files),
|
||
file_count=len(immutable_files),
|
||
total_bytes=total_bytes,
|
||
)
|
||
|
||
|
||
def capture_artifact_snapshot(root: Path, *, max_files=None, max_bytes=None) -> ArtifactSnapshot:
|
||
"""用目录文件描述符一次捕获全树，拒绝越界、symlink、资源超限和读取竞态。
|
||
|
||
所有路径分量都通过 ``openat + O_NOFOLLOW`` 打开；文件内容、artifactHash 和后续发布字节均来自
|
||
同一份内存快照。资源上限在读取前和读取中双重检查，避免为了判断超限先把异常产物读进内存。
|
||
"""
|
||
root = Path(root)
|
||
max_files = MAX_ARTIFACT_FILES if max_files is None else int(max_files)
|
||
max_bytes = MAX_ARTIFACT_BYTES if max_bytes is None else int(max_bytes)
|
||
if max_files < 0 or max_bytes < 0:
|
||
raise ValueError("staged artifact 资源上限不得为负数")
|
||
|
||
flags_dir = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
|
||
flags_file = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
|
||
try:
|
||
root_fd = os.open(root, flags_dir)
|
||
except OSError as exc:
|
||
try:
|
||
root_mode = root.lstat().st_mode
|
||
except OSError:
|
||
raise exc
|
||
if stat.S_ISLNK(root_mode):
|
||
raise ValueError(f"staged artifact 根目录不得是 symlink:{root}") from exc
|
||
if not stat.S_ISDIR(root_mode):
|
||
raise ValueError(f"staged artifact 根路径不是目录:{root}") from exc
|
||
raise
|
||
|
||
files: dict[str, bytes] = {}
|
||
total_bytes = 0
|
||
|
||
def capture_dir(dir_fd: int, prefix: str) -> None:
|
||
nonlocal total_bytes
|
||
for name in sorted(os.listdir(dir_fd)):
|
||
if not name or name in (".", "..") or "/" in name or "\\" in name:
|
||
raise ValueError("staged artifact 含越界路径分量")
|
||
relative = f"{prefix}/{name}" if prefix else name
|
||
before = os.stat(name, dir_fd=dir_fd, follow_symlinks=False)
|
||
if stat.S_ISLNK(before.st_mode):
|
||
raise ValueError(f"staged artifact 含 symlink:{relative}")
|
||
if stat.S_ISDIR(before.st_mode):
|
||
child_fd = os.open(name, flags_dir, dir_fd=dir_fd)
|
||
try:
|
||
opened = os.fstat(child_fd)
|
||
if (opened.st_dev, opened.st_ino) != (before.st_dev, before.st_ino):
|
||
raise ValueError(f"staged artifact 目录读取时被替换:{relative}")
|
||
capture_dir(child_fd, relative)
|
||
finally:
|
||
os.close(child_fd)
|
||
continue
|
||
if not stat.S_ISREG(before.st_mode):
|
||
raise ValueError(f"staged artifact 含非普通文件:{relative}")
|
||
if len(files) >= max_files:
|
||
raise ValueError(f"staged artifact 文件数超过 {max_files}")
|
||
|
||
file_fd = os.open(name, flags_file, dir_fd=dir_fd)
|
||
try:
|
||
opened = os.fstat(file_fd)
|
||
if (not stat.S_ISREG(opened.st_mode)
|
||
or (opened.st_dev, opened.st_ino) != (before.st_dev, before.st_ino)):
|
||
raise ValueError(f"staged artifact 文件读取时被替换:{relative}")
|
||
if total_bytes + opened.st_size > max_bytes:
|
||
raise ValueError(f"staged artifact 总字节超过 {max_bytes}")
|
||
|
||
chunks = []
|
||
while True:
|
||
chunk = os.read(file_fd, 1024 * 1024)
|
||
if not chunk:
|
||
break
|
||
total_bytes += len(chunk)
|
||
if total_bytes > max_bytes:
|
||
raise ValueError(f"staged artifact 总字节超过 {max_bytes}")
|
||
chunks.append(chunk)
|
||
data = b"".join(chunks)
|
||
after = os.fstat(file_fd)
|
||
stable_fields = ("st_dev", "st_ino", "st_size", "st_mtime_ns", "st_ctime_ns")
|
||
if (any(getattr(opened, field) != getattr(after, field) for field in stable_fields)
|
||
or len(data) != after.st_size):
|
||
raise ValueError(f"staged artifact 文件读取中发生改写:{relative}")
|
||
files[relative] = data
|
||
finally:
|
||
os.close(file_fd)
|
||
|
||
try:
|
||
if not stat.S_ISDIR(os.fstat(root_fd).st_mode):
|
||
raise ValueError("staged artifact 根不是目录")
|
||
capture_dir(root_fd, "")
|
||
finally:
|
||
os.close(root_fd)
|
||
|
||
immutable_files = MappingProxyType(dict(files))
|
||
return ArtifactSnapshot(
|
||
files=immutable_files,
|
||
artifact_hash=_artifact_hash(immutable_files),
|
||
file_count=len(immutable_files),
|
||
total_bytes=total_bytes,
|
||
)
|
||
|
||
|
||
def write_snapshot_stream(root: Path, output) -> None:
|
||
"""把可信快照写成一行 manifest 加连续文件字节，供 Node runner 无损读取。"""
|
||
snapshot = capture_artifact_snapshot(root)
|
||
entries = [
|
||
{"path": relative, "size": len(snapshot.files[relative])}
|
||
for relative in sorted(snapshot.files, key=lambda value: value.encode("utf-8"))
|
||
]
|
||
manifest = {
|
||
"schemaVersion": "artifact-snapshot-stream/1",
|
||
"artifactHash": snapshot.artifact_hash,
|
||
"fileCount": snapshot.file_count,
|
||
"totalBytes": snapshot.total_bytes,
|
||
"entries": entries,
|
||
}
|
||
manifest_bytes = json.dumps(
|
||
manifest, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False,
|
||
).encode("utf-8")
|
||
output.write(_SNAPSHOT_STREAM_MAGIC)
|
||
output.write(manifest_bytes + b"\n")
|
||
for entry in entries:
|
||
output.write(snapshot.files[entry["path"]])
|
||
|
||
|
||
def _main(argv: list[str]) -> int:
|
||
"""仅暴露只读 stream 子命令；错误写 stderr，stdout 永远不混入诊断文本。"""
|
||
if len(argv) != 3 or argv[1] != "--stream":
|
||
print("用法: artifact_snapshot.py --stream <staged-root>", file=sys.stderr)
|
||
return 2
|
||
try:
|
||
write_snapshot_stream(Path(argv[2]), sys.stdout.buffer)
|
||
except Exception as exc: # noqa: BLE001 —— 子进程边界需把原始类别留给 Node 映射为 tester_error
|
||
print(f"{type(exc).__name__}: {exc}", file=sys.stderr)
|
||
return 2
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(_main(sys.argv))
|