le-app-design v1.0.0：统一 App 设计工坊——真实集成并调用三上游的编排器。ui-ux-pro-max（MIT，search.py 设计系统/规范检索）+ frontend-design（Apache-2.0，Skill 工具装载的视觉 design lead）+ impeccable（Apache-2.0，CLI 检测器+playbook 审查）；三机制保证真集成可验证：check_deps 硬依赖门（缺即终止）、log_call 证据账本（调用落 calls.jsonl 含装载 sha256）、evidence_report 验收门（三方各 ≥1 成功调用才 PASS）；S0–S7 流水线含两个回合终止停点与轻量任务路由；E2E 已真跑（习惯打卡 App 双屏原型）：7 条真实调用留账、双独立 sub-agent 评审 26/36 + detect 2 findings、修复 3×P1 后字阶清零、交互复测 4/4、验收门 PASS；上游三件补装 ~/.codex/skills，仓库 README 同步
This commit is contained in:
leroy
@@ -0,0 +1,223 @@
|
||||
#!/usr/bin/env python3
|
||||
"""le-app-design 硬依赖门：解析并活性检查三个上游 Skill。
|
||||
|
||||
用法:
|
||||
python3 check_deps.py [--out <RUN>/deps.json] [--quiet]
|
||||
|
||||
解析顺序: $LE_APP_DESIGN_DEP_ROOTS(冒号分隔) > ~/.agents/skills > ~/.zcode/skills
|
||||
> ~/.codex/skills > ~/.claude/skills
|
||||
（优先 ~/.agents/skills：impeccable 的 SKILL.md 自引用该路径。）
|
||||
|
||||
活性检查按依赖形态各自不同——这就是「真实调用形态」的机器定义:
|
||||
ui-ux-pro-max 真跑一次 search.py 检索（stdout 含结果段即活）
|
||||
frontend-design 校验 SKILL.md 原文完整可读（纯文档型 Skill，装载即调用）
|
||||
impeccable 跑其自包含二进制 --version
|
||||
|
||||
全部通过 exit 0；任何缺失/失活 exit 1 并输出明确报告。仅 python3 标准库。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
SKILL_KEY = "le-app-design-dep-check"
|
||||
|
||||
UPSTREAMS = [
|
||||
{
|
||||
"key": "ui-ux-pro-max",
|
||||
"repo": "nextlevelbuilder/ui-ux-pro-max-skill",
|
||||
"license": "MIT",
|
||||
"install": "git clone https://github.com/nextlevelbuilder/ui-ux-pro-max-skill && 拷贝/链接其 .claude/skills/ui-ux-pro-max 到 ~/.agents/skills/ui-ux-pro-max",
|
||||
"kind": "cli-search",
|
||||
},
|
||||
{
|
||||
"key": "frontend-design",
|
||||
"repo": "anthropics/skills",
|
||||
"license": "Apache-2.0",
|
||||
"install": "git clone https://github.com/anthropics/skills && 拷贝/链接 skills/frontend-design 到 ~/.agents/skills/frontend-design",
|
||||
"kind": "doc-guidance",
|
||||
},
|
||||
{
|
||||
"key": "impeccable",
|
||||
"repo": "pbakaus/impeccable",
|
||||
"license": "Apache-2.0",
|
||||
"install": "git clone https://github.com/pbakaus/impeccable && 拷贝/链接到 ~/.agents/skills/impeccable（含 scripts/ 自包含二进制）",
|
||||
"kind": "cli-engine",
|
||||
},
|
||||
]
|
||||
|
||||
FRONTMATTER_MIN_BYTES = 2000 # frontend-design 原文约 10KB；残缺副本在此拦下
|
||||
|
||||
|
||||
def candidate_roots() -> list[Path]:
|
||||
roots: list[Path] = []
|
||||
env = os.environ.get("LE_APP_DESIGN_DEP_ROOTS", "")
|
||||
for part in env.split(os.pathsep):
|
||||
if part.strip():
|
||||
roots.append(Path(part).expanduser())
|
||||
home = Path.home()
|
||||
roots += [
|
||||
home / ".agents" / "skills",
|
||||
home / ".zcode" / "skills",
|
||||
home / ".codex" / "skills",
|
||||
home / ".claude" / "skills",
|
||||
]
|
||||
seen: set[str] = set()
|
||||
unique: list[Path] = []
|
||||
for r in roots:
|
||||
key = str(r)
|
||||
if key not in seen:
|
||||
seen.add(key)
|
||||
unique.append(r)
|
||||
return unique
|
||||
|
||||
|
||||
def read_frontmatter_name(skill_md: Path) -> str | None:
|
||||
try:
|
||||
text = skill_md.read_text(encoding="utf-8", errors="replace")
|
||||
except OSError:
|
||||
return None
|
||||
m = re.match(r"^---\s*\n(.*?\n)?---", text, re.DOTALL)
|
||||
if not m:
|
||||
return None
|
||||
block = m.group(1) or ""
|
||||
nm = re.search(r"^name:\s*['\"]?([A-Za-z0-9_-]+)", block, re.MULTILINE)
|
||||
return nm.group(1) if nm else None
|
||||
|
||||
|
||||
def resolve(key: str) -> tuple[Path | None, str]:
|
||||
"""返回 (解析到的目录, 说明)。"""
|
||||
for root in candidate_roots():
|
||||
dep_dir = root / key
|
||||
skill_md = dep_dir / "SKILL.md"
|
||||
if skill_md.is_file():
|
||||
name = read_frontmatter_name(skill_md)
|
||||
if name == key:
|
||||
return dep_dir, f"resolved at {dep_dir}"
|
||||
return dep_dir, f"SKILL.md frontmatter name={name!r} != {key!r}（目录在但身份不符）"
|
||||
return None, "not found in any candidate root"
|
||||
|
||||
|
||||
def _run(cmd: list[str], timeout: int = 90) -> subprocess.CompletedProcess:
|
||||
env = dict(os.environ, PYTHONDONTWRITEBYTECODE="1")
|
||||
return subprocess.run(
|
||||
cmd, capture_output=True, text=True, timeout=timeout, env=env
|
||||
)
|
||||
|
||||
|
||||
def liveness(spec: dict, dep_dir: Path) -> tuple[bool, str]:
|
||||
key, kind = spec["key"], spec["kind"]
|
||||
try:
|
||||
if kind == "cli-search":
|
||||
script = dep_dir / "scripts" / "search.py"
|
||||
if not script.is_file():
|
||||
return False, f"missing {script}"
|
||||
with tempfile.TemporaryDirectory() as td:
|
||||
proc = subprocess.run(
|
||||
[sys.executable, str(script), "mobile app", "--domain", "style", "-n", "1"],
|
||||
capture_output=True, text=True, timeout=90, cwd=td,
|
||||
env=dict(os.environ, PYTHONDONTWRITEBYTECODE="1"),
|
||||
)
|
||||
if proc.returncode != 0:
|
||||
return False, f"search.py exit={proc.returncode}: {proc.stderr.strip()[:300]}"
|
||||
if "Result" not in proc.stdout:
|
||||
return False, "search.py exit=0 but no result section in stdout"
|
||||
return True, "search.py live query ok"
|
||||
|
||||
if kind == "doc-guidance":
|
||||
skill_md = dep_dir / "SKILL.md"
|
||||
try:
|
||||
data = skill_md.read_bytes()
|
||||
except OSError as e:
|
||||
return False, f"unreadable SKILL.md: {e}"
|
||||
if len(data) < FRONTMATTER_MIN_BYTES:
|
||||
return False, f"SKILL.md only {len(data)} bytes (<{FRONTMATTER_MIN_BYTES}); likely truncated copy"
|
||||
text = data.decode("utf-8", errors="replace")
|
||||
if "name: frontend-design" not in text or "description:" not in text:
|
||||
return False, "SKILL.md frontmatter incomplete"
|
||||
return True, f"original SKILL.md intact ({len(data)} bytes, sha256 {hashlib.sha256(data).hexdigest()[:12]}…)"
|
||||
|
||||
if kind == "cli-engine":
|
||||
launcher = dep_dir / "scripts" / "impeccable"
|
||||
if not launcher.is_file():
|
||||
return False, f"missing launcher {launcher}"
|
||||
proc = _run(["sh", str(launcher), "--version"])
|
||||
if proc.returncode != 0:
|
||||
return False, f"impeccable --version exit={proc.returncode}: {proc.stderr.strip()[:300]}"
|
||||
version = (proc.stdout or "").strip() or (proc.stderr or "").strip()
|
||||
return True, f"impeccable binary ok ({version[:60]})"
|
||||
except subprocess.TimeoutExpired:
|
||||
return False, f"{key} liveness check timed out"
|
||||
except OSError as e:
|
||||
return False, f"liveness check error: {e}"
|
||||
return False, "unknown kind"
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description="le-app-design 依赖门")
|
||||
ap.add_argument("--out", help="JSON 结果写入路径（如 <RUN>/deps.json）")
|
||||
ap.add_argument("--quiet", action="store_true", help="只输出摘要一行")
|
||||
args = ap.parse_args()
|
||||
|
||||
result = {
|
||||
"skill": SKILL_KEY,
|
||||
"ts": datetime.now(timezone.utc).isoformat(timespec="seconds"),
|
||||
"roots_searched": [str(r) for r in candidate_roots()],
|
||||
"deps": {},
|
||||
"ok": False,
|
||||
}
|
||||
|
||||
for spec in UPSTREAMS:
|
||||
dep_dir, note = resolve(spec["key"])
|
||||
entry = {
|
||||
"repo": spec["repo"],
|
||||
"license": spec["license"],
|
||||
"kind": spec["kind"],
|
||||
"path": str(dep_dir) if dep_dir else None,
|
||||
"alive": False,
|
||||
"check": note if dep_dir is None else "",
|
||||
"install": spec["install"],
|
||||
}
|
||||
if dep_dir is not None:
|
||||
alive, check = liveness(spec, dep_dir)
|
||||
entry["alive"] = alive
|
||||
entry["check"] = check
|
||||
result["deps"][spec["key"]] = entry
|
||||
|
||||
result["ok"] = all(d["alive"] for d in result["deps"].values())
|
||||
|
||||
if args.out:
|
||||
out_path = Path(args.out).expanduser()
|
||||
out_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
out_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
|
||||
lines = []
|
||||
for key, d in result["deps"].items():
|
||||
mark = "✅" if d["alive"] else "❌"
|
||||
where = d["path"] or "未找到"
|
||||
lines.append(f"{mark} {key:18} {where}")
|
||||
if not d["alive"]:
|
||||
lines.append(f" 原因: {d['check']}")
|
||||
lines.append(f" 安装: {d['install']}")
|
||||
verdict = "依赖门通过（三方解析且活性正常）" if result["ok"] else "依赖门未通过——禁止继续设计流程"
|
||||
if args.quiet:
|
||||
print(f"{'PASS' if result['ok'] else 'FAIL'}: " + ", ".join(
|
||||
f"{k}={'ok' if v['alive'] else 'MISSING'}" for k, v in result["deps"].items()))
|
||||
else:
|
||||
print("\n".join(lines))
|
||||
print(verdict)
|
||||
if args.out:
|
||||
print(f"JSON: {args.out}")
|
||||
return 0 if result["ok"] else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,140 @@
|
||||
#!/usr/bin/env python3
|
||||
"""le-app-design 验收门：从 calls.jsonl 生成本次运行的集成证据报告。
|
||||
|
||||
用法:
|
||||
python3 evidence_report.py --run-dir <RUN> [--out <RUN>/INTEGRATION-REPORT.md] \
|
||||
[--artifact <路径>]... [--title "任务名"]
|
||||
|
||||
验收规则（硬门）:
|
||||
三个上游 ui-ux-pro-max / frontend-design / impeccable 各需 ≥1 条成功调用
|
||||
（exec 且 exit_code==0，或 load 且 exit_code==0）。
|
||||
--artifact 指定的每个文件必须存在。
|
||||
任一不满足 → 报告标 FAIL，exit 1；全满足 → PASS，exit 0。
|
||||
|
||||
报告内容: 判定横幅、三方调用统计表、逐条调用清单（时间/依赖/命令/退出码）、
|
||||
产物核验清单、失败与缺口说明。这份报告就是「真实集成」的可检查证据。
|
||||
仅 python3 标准库。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
GATE_DEPS = ["ui-ux-pro-max", "frontend-design", "impeccable"]
|
||||
|
||||
|
||||
def load_records(ledger: Path) -> list[dict]:
|
||||
if not ledger.is_file():
|
||||
return []
|
||||
records = []
|
||||
for line in ledger.read_text(encoding="utf-8", errors="replace").splitlines():
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
rec = json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
records.append({"dep": "(unparsable)", "kind": "raw", "exit_code": None, "command": line[:120], "ts": ""})
|
||||
continue
|
||||
records.append(rec)
|
||||
return records
|
||||
|
||||
|
||||
def is_success(rec: dict) -> bool:
|
||||
return rec.get("exit_code") == 0 and rec.get("kind") in ("exec", "load")
|
||||
|
||||
|
||||
def one_line(rec: dict) -> str:
|
||||
cmd = rec.get("command", "")
|
||||
try: # exec 记录的 command 是 JSON 数组字符串
|
||||
parts = json.loads(cmd)
|
||||
if isinstance(parts, list):
|
||||
cmd = " ".join(str(p) for p in parts)
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
pass
|
||||
return cmd if len(cmd) <= 160 else cmd[:157] + "..."
|
||||
|
||||
|
||||
def main() -> int:
|
||||
ap = argparse.ArgumentParser(description="le-app-design 验收门")
|
||||
ap.add_argument("--run-dir", required=True)
|
||||
ap.add_argument("--out", default=None, help="报告写入路径，缺省 <run-dir>/INTEGRATION-REPORT.md")
|
||||
ap.add_argument("--artifact", action="append", default=[], help="须存在的交付物路径（可多次）")
|
||||
ap.add_argument("--title", default="le-app-design 集成证据报告")
|
||||
args = ap.parse_args()
|
||||
|
||||
run_dir = Path(args.run_dir).expanduser()
|
||||
records = load_records(run_dir / "calls.jsonl")
|
||||
out_path = Path(args.out).expanduser() if args.out else run_dir / "INTEGRATION-REPORT.md"
|
||||
|
||||
stats = {dep: {"ok": 0, "fail": 0} for dep in GATE_DEPS}
|
||||
for rec in records:
|
||||
dep = rec.get("dep")
|
||||
if dep in stats:
|
||||
stats[dep]["ok" if is_success(rec) else "fail"] += 1
|
||||
|
||||
gate_pass = all(stats[d]["ok"] >= 1 for d in GATE_DEPS)
|
||||
artifacts_missing = [a for a in args.artifact if not Path(a).expanduser().exists()]
|
||||
overall = gate_pass and not artifacts_missing
|
||||
|
||||
lines: list[str] = []
|
||||
lines.append(f"# {args.title}")
|
||||
lines.append("")
|
||||
lines.append(f"- 运行目录: `{run_dir}`")
|
||||
lines.append(f"- 生成时间: {datetime.now(timezone.utc).isoformat(timespec='seconds')}")
|
||||
lines.append(f"- 调用记录: {len(records)} 条（`calls.jsonl`）")
|
||||
lines.append("")
|
||||
lines.append(f"## 判定: {'✅ PASS' if overall else '❌ FAIL'}")
|
||||
lines.append("")
|
||||
lines.append("| 上游 | 成功调用 | 失败调用 | 门(≥1 成功) |")
|
||||
lines.append("|---|---|---|---|")
|
||||
for dep in GATE_DEPS:
|
||||
s = stats[dep]
|
||||
lines.append(f"| {dep} | {s['ok']} | {s['fail']} | {'✅' if s['ok'] >= 1 else '❌ 未被真实调用'} |")
|
||||
lines.append("")
|
||||
|
||||
if artifacts_missing:
|
||||
lines.append("**缺失产物:**")
|
||||
for a in artifacts_missing:
|
||||
lines.append(f"- ❌ `{a}`")
|
||||
lines.append("")
|
||||
elif args.artifact:
|
||||
lines.append("**产物核验:**")
|
||||
for a in args.artifact:
|
||||
lines.append(f"- ✅ `{a}`")
|
||||
lines.append("")
|
||||
|
||||
lines.append("## 调用清单")
|
||||
lines.append("")
|
||||
lines.append("| # | 时间(UTC) | 依赖 | 类型 | 退出码 | 命令/装载 |")
|
||||
lines.append("|---|---|---|---|---|---|")
|
||||
for i, rec in enumerate(records, 1):
|
||||
lines.append(
|
||||
f"| {i} | {rec.get('ts', '')} | {rec.get('dep', '?')} | {rec.get('kind', '?')} "
|
||||
f"| {rec.get('exit_code', '?')} | `{one_line(rec)}` |"
|
||||
)
|
||||
lines.append("")
|
||||
|
||||
if not overall:
|
||||
lines.append("## 缺口说明")
|
||||
lines.append("")
|
||||
for dep in GATE_DEPS:
|
||||
if stats[dep]["ok"] < 1:
|
||||
lines.append(f"- `{dep}` 无成功调用记录——该上游未被真实使用，本次流程**不得标记为成功**。")
|
||||
for a in artifacts_missing:
|
||||
lines.append(f"- 产物缺失: `{a}`")
|
||||
lines.append("")
|
||||
|
||||
report = "\n".join(lines)
|
||||
out_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
out_path.write_text(report + "\n", encoding="utf-8")
|
||||
print(report)
|
||||
print(f"\n[report] {'PASS' if overall else 'FAIL'} → {out_path}", file=sys.stderr)
|
||||
return 0 if overall else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,140 @@
|
||||
#!/usr/bin/env python3
|
||||
"""le-app-design 证据账本：所有对上游 Skill 的真实调用经此包裹，落入 calls.jsonl。
|
||||
|
||||
用法:
|
||||
执行型调用（CLI 类上游）:
|
||||
python3 log_call.py <dep> --run-dir <RUN> -- python3 <DEP>/scripts/search.py "…" --design-system
|
||||
装载型调用（文档型上游 / playbook）:
|
||||
python3 log_call.py <dep> --run-dir <RUN> --load <DEP>/SKILL.md
|
||||
|
||||
dep ∈ {ui-ux-pro-max, frontend-design, impeccable, le-app-design(自检)}
|
||||
--run-dir 缺省时按序: $LE_APP_DESIGN_RUN_DIR > <cwd>/le-app-design-runs/<UTC 时间戳>-run/
|
||||
|
||||
行为:
|
||||
- exec: 运行命令，透传 stdout/stderr，透传退出码；成功失败都记账。
|
||||
- load: 读文件记 sha256 与字节数（证明用的是原文而非转述），文件缺失记 FAIL 退出 1。
|
||||
- 记录字段: ts / dep / kind / command / exit_code / duration_ms / stdout_tail /
|
||||
stderr_tail / bytes / sha256 / note。stdout/stderr 仅保留尾部 4000 字符防账本膨胀。
|
||||
|
||||
仅 python3 标准库。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
TAIL = 4000
|
||||
KNOWN_DEPS = {"ui-ux-pro-max", "frontend-design", "impeccable", "le-app-design"}
|
||||
|
||||
|
||||
def default_run_dir() -> Path:
|
||||
env = os.environ.get("LE_APP_DESIGN_RUN_DIR")
|
||||
if env:
|
||||
return Path(env).expanduser()
|
||||
stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%SZ")
|
||||
return Path.cwd() / "le-app-design-runs" / f"{stamp}-run"
|
||||
|
||||
|
||||
def append_record(run_dir: Path, record: dict) -> Path:
|
||||
run_dir.mkdir(parents=True, exist_ok=True)
|
||||
ledger = run_dir / "calls.jsonl"
|
||||
with ledger.open("a", encoding="utf-8") as f:
|
||||
f.write(json.dumps(record, ensure_ascii=False) + "\n")
|
||||
return ledger
|
||||
|
||||
|
||||
def confirm(dep: str, kind: str, status: str, ledger: Path, extra: str = "") -> None:
|
||||
print(
|
||||
f"[log_call] {dep} {kind} {status} → {ledger}"
|
||||
+ (f" ({extra})" if extra else ""),
|
||||
file=sys.stderr,
|
||||
)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
# 手工在第一个裸 "--" 处切分：argparse 的 REMAINDER 会把 -- 之前的选项误吞进命令
|
||||
argv = sys.argv[1:]
|
||||
cmd: list[str] = []
|
||||
if "--" in argv:
|
||||
split = argv.index("--")
|
||||
argv, cmd = argv[:split], argv[split + 1:]
|
||||
|
||||
ap = argparse.ArgumentParser(description="le-app-design 调用账本")
|
||||
ap.add_argument("dep", help="上游标识: ui-ux-pro-max | frontend-design | impeccable | le-app-design")
|
||||
ap.add_argument("--run-dir", default=None, help="运行目录（缺省自动创建）")
|
||||
ap.add_argument("--note", default="", help="阶段/用途备注，如 S2-design-system")
|
||||
ap.add_argument("--load", metavar="PATH", default=None, help="装载型调用：记录读取的原文文件")
|
||||
args = ap.parse_args(argv)
|
||||
|
||||
dep = args.dep
|
||||
if dep not in KNOWN_DEPS:
|
||||
print(f"[log_call] 未知依赖标识 {dep!r}；允许: {sorted(KNOWN_DEPS)}", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
run_dir = Path(args.run_dir).expanduser() if args.run_dir else default_run_dir()
|
||||
base = {"ts": datetime.now(timezone.utc).isoformat(timespec="seconds"), "dep": dep, "note": args.note}
|
||||
|
||||
if args.load:
|
||||
path = Path(args.load).expanduser()
|
||||
rec = dict(base, kind="load", command=f"load:{path}")
|
||||
try:
|
||||
data = path.read_bytes()
|
||||
except OSError as e:
|
||||
rec.update(exit_code=1, stderr_tail=f"read failed: {e}"[:TAIL])
|
||||
ledger = append_record(run_dir, rec)
|
||||
confirm(dep, "load", "FAIL", ledger, str(e))
|
||||
return 1
|
||||
rec.update(
|
||||
exit_code=0,
|
||||
bytes=len(data),
|
||||
sha256=hashlib.sha256(data).hexdigest(),
|
||||
)
|
||||
ledger = append_record(run_dir, rec)
|
||||
confirm(dep, "load", "OK", ledger, f"{len(data)} bytes sha256:{rec['sha256'][:12]}…")
|
||||
return 0
|
||||
|
||||
if not cmd:
|
||||
print("[log_call] 需要 --load PATH 或 -- <命令…>", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
rec = dict(base, kind="exec", command=json.dumps(cmd, ensure_ascii=False))
|
||||
t0 = time.monotonic()
|
||||
try:
|
||||
proc = subprocess.run(
|
||||
cmd, capture_output=True, text=True, timeout=900,
|
||||
env=dict(os.environ, PYTHONDONTWRITEBYTECODE="1"),
|
||||
)
|
||||
rc, out, err = proc.returncode, proc.stdout or "", proc.stderr or ""
|
||||
except subprocess.TimeoutExpired as e:
|
||||
rc, out, err = 124, (e.stdout or b"").decode(errors="replace") if isinstance(e.stdout, bytes) else (e.stdout or ""), f"timeout after {e.timeout}s"
|
||||
except FileNotFoundError as e:
|
||||
rc, out, err = 127, "", f"command not found: {e}"
|
||||
|
||||
rec.update(
|
||||
exit_code=rc,
|
||||
duration_ms=int((time.monotonic() - t0) * 1000),
|
||||
stdout_tail=out[-TAIL:],
|
||||
stderr_tail=err[-TAIL:],
|
||||
)
|
||||
ledger = append_record(run_dir, rec)
|
||||
if out:
|
||||
sys.stdout.write(out)
|
||||
if not out.endswith("\n"):
|
||||
sys.stdout.write("\n")
|
||||
if err:
|
||||
sys.stderr.write(err)
|
||||
if not err.endswith("\n"):
|
||||
sys.stderr.write("\n")
|
||||
confirm(dep, "exec", f"exit={rc}", ledger, f"{rec['duration_ms']}ms")
|
||||
return rc
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user