435 lines
20 KiB
Python
435 lines
20 KiB
Python
#!/usr/bin/env python3
|
||
"""HEL-860 负例复现脚本：总管独立复核《HEL-859-独立复核.md》四条新增反例。
|
||
|
||
零真实模型、零网络（socket 硬拒绝 + HF 离线）、零本地嵌入：判据负例为纯函数
|
||
代入；来源锚点负例用假后端 payload 注入 + 复制的真实 history 只读（总管
|
||
audit859 store 快照，不触真实资料、不新建嵌入）；问答出口负例用真实 A.run
|
||
编排 + 假提取/假搜索/假问答全流程捕获。同一脚本在修复前（9f3e6a3）与修复后
|
||
代码上均可运行：
|
||
|
||
修复前（--phase prefix）→ 四条反例全部复现（放行/判通过/毒条目进入出口）
|
||
修复后（默认 postfix） → 同一真实出口全部拒绝/判失败，正例保留
|
||
|
||
四条反例对应独立复核 1-4：
|
||
1. 实际问答出口未经来源门：unknown-origin 毒条目修前进入 rr.ask 实际发送
|
||
上下文（s7_gated_assembly 此前调用 0）；修后全部 8 个问答出口的 mem0
|
||
侧上下文必须出自带门装配（S2/S3/S4/S6 一并核对），毒条目不出现在任何
|
||
发送上下文。
|
||
2. 同组错误角色/矛盾轮次锚点：真实 t1 伴侣原句 + (t1,user) 锚点（t1 told=0）
|
||
绕过分享门、turn_ids=[t2] 与 turn_roles=[(t1,user)] 互矛——修前
|
||
resolved=true 且原句进最终块；修后拒绝，同轮真实用户正例保留。
|
||
3. S6 以字面替代状态：已结侧答"现在还没有告诉我"、两侧"不知道"、未结侧
|
||
同义宣称"结果已收到"——修前 pass=true；修后判失败；正确答案正例通过。
|
||
4. S2 上下文层未纳入总判：两侧实际发送上下文均无细节、答案碰上细节词
|
||
——修前 pass=true；修后判失败；含细节正例通过。
|
||
|
||
用法：python -m spike.hel857.audit860 [--phase prefix|postfix] [输出json路径]
|
||
退出码（HEL-859 教训）：prefix=0 当且仅当四条反例全部复现；postfix=0 当且仅当
|
||
全部反例被拒/判失败且正例全部保留。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import inspect
|
||
import json
|
||
import os
|
||
import shutil
|
||
import socket
|
||
import sqlite3
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
|
||
REPO = Path(__file__).resolve().parents[2]
|
||
DATA = Path(os.environ.get("HEL857_DATA", REPO.parent / "hel857-data"))
|
||
os.environ.setdefault("HF_HUB_OFFLINE", "1")
|
||
os.environ.setdefault("TRANSFORMERS_OFFLINE", "1")
|
||
|
||
|
||
def deny_network(*a, **k):
|
||
raise RuntimeError("audit860 blocks network")
|
||
|
||
|
||
socket.socket.connect = deny_network
|
||
socket.socket.connect_ex = deny_network
|
||
|
||
from . import fixtures as FX
|
||
from . import ab_scenarios as A
|
||
from .ab_scenarios import CompanionShareGate, group_input_sha
|
||
|
||
OUT = {"real_network_allowed": False, "real_model_requests": 0, "local_embeddings": 0}
|
||
PHASE = "postfix"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
def _kw(fn, **kw):
|
||
"""按当前函数签名过滤参数（prefix 代码签名无新参时自动省略）。"""
|
||
params = inspect.signature(fn).parameters
|
||
return {k: v for k, v in kw.items() if k in params}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 反例3/4：总判据负例（纯函数，零嵌入）
|
||
def s6_answer(s):
|
||
return {"answer": s, "ok": True}
|
||
|
||
|
||
def section_criteria() -> dict:
|
||
out: dict = {}
|
||
orch = {"pass": True, "status_annotation_in_block": True,
|
||
"unresolved_contrast": {"built": True, "self_consistent": True}}
|
||
out["S6_resolved_answer_says_open"] = A.s6_result(
|
||
orch, s6_answer("你答应周五晚告诉我面试结果，现在还没有告诉我。"),
|
||
s6_answer("你答应周五晚告诉我面试结果，现在还没有告诉我。"))
|
||
out["S6_both_answers_no_knowledge"] = A.s6_result(
|
||
orch, s6_answer("不知道你答应了什么。"), s6_answer("不知道。"))
|
||
out["S6_unresolved_claims_result_received"] = A.s6_result(
|
||
orch, s6_answer("你答应告诉我面试结果，结果已收到。"),
|
||
s6_answer("你答应告诉我面试结果，结果已收到。"))
|
||
out["S6_unresolved_same_synonym_received"] = A.s6_result(
|
||
orch, s6_answer("你答应周五晚告诉我面试结果，现在已经告诉我了。"),
|
||
s6_answer("你答应周五晚告诉我面试结果，结果已经拿到了。"))
|
||
out["S6_positive"] = A.s6_result(
|
||
orch, s6_answer("你答应过周五晚告诉我面试结果，现在已经告诉我了。"),
|
||
s6_answer("你答应过周五晚告诉我面试结果，还没有告诉我。"))
|
||
|
||
detail_sent = "今天在旧书店淘到一本蓝色封面的诗集，纸页有点受潮的味道，可是我特别喜欢"
|
||
no_detail_sent = "用户买到一本诗集，很喜欢"
|
||
detail_answers = {
|
||
"a_base_reread": s6_answer("纸页有受潮的味道"),
|
||
"a_base_gist": s6_answer("不知道"),
|
||
"a_mem0": s6_answer("纸页有受潮的味道"),
|
||
}
|
||
unknown_answers = {
|
||
"a_base_reread": s6_answer("不知道"),
|
||
"a_base_gist": s6_answer("不知道"),
|
||
"a_mem0": s6_answer("不知道"),
|
||
}
|
||
if "b_reread_sent" in inspect.signature(A.s2_result).parameters:
|
||
out["S2_details_absent_from_both_contexts"] = A.s2_result(
|
||
b_hit=True, m_hit=True, b_reread_sent=no_detail_sent,
|
||
b_gist_sent=no_detail_sent, m_sent=no_detail_sent, **detail_answers)
|
||
out["S2_unknown_answers"] = A.s2_result(
|
||
b_hit=True, m_hit=True, b_reread_sent=detail_sent,
|
||
b_gist_sent=no_detail_sent, m_sent=detail_sent, **unknown_answers)
|
||
out["S2_positive"] = A.s2_result(
|
||
b_hit=True, m_hit=True, b_reread_sent=detail_sent,
|
||
b_gist_sent=no_detail_sent, m_sent=detail_sent,
|
||
a_base_reread=s6_answer("纸页有受潮的味道"),
|
||
a_base_gist=s6_answer("不知道"),
|
||
a_mem0=s6_answer("有潮湿的气味"))
|
||
else: # prefix 签名：布尔上下文层
|
||
out["S2_details_absent_from_both_contexts"] = A.s2_result(
|
||
b_hit=True, m_hit=True, b_detail_ctx=False, b_detail_reread=False,
|
||
m_detail_ctx=False, **detail_answers)
|
||
out["S2_unknown_answers"] = A.s2_result(
|
||
b_hit=True, m_hit=True, b_detail_ctx=False, b_detail_reread=True,
|
||
m_detail_ctx=True, **unknown_answers)
|
||
out["S2_positive"] = A.s2_result(
|
||
b_hit=True, m_hit=True, b_detail_ctx=False, b_detail_reread=True,
|
||
m_detail_ctx=True,
|
||
a_base_reread=s6_answer("纸页有受潮的味道"),
|
||
a_base_gist=s6_answer("不知道"),
|
||
a_mem0=s6_answer("有潮湿的气味"))
|
||
return out
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 反例2：来源锚点负例（假后端 payload 注入 + 复制的真实 history 只读，零嵌入）
|
||
class PayloadClient:
|
||
def __init__(self, payload):
|
||
self.payload = payload
|
||
|
||
def retrieve(self, **kw):
|
||
return [SimpleNamespace(payload=self.payload)]
|
||
|
||
|
||
def fake_memory(payload):
|
||
return SimpleNamespace(collection_name="synthetic",
|
||
vector_store=SimpleNamespace(client=PayloadClient(payload)))
|
||
|
||
|
||
def section_anchors(hist: Path, ctrl: Path) -> dict:
|
||
with sqlite3.connect(str(hist)) as c:
|
||
comp_id = c.execute("SELECT memory_id FROM history WHERE new_memory=?",
|
||
(FX.PUBLIC_CORPUS[1][2],)).fetchone()[0]
|
||
user_id = c.execute("SELECT memory_id FROM history WHERE new_memory=?",
|
||
(FX.PUBLIC_CORPUS[0][2],)).fetchone()[0]
|
||
gate = CompanionShareGate(ctrl) # 真实控制库 t1 told=0
|
||
|
||
# 2a. 真实 t1 伴侣原句 + 同组错挂 (t1,user) 锚点：told=0 却以用户行身份放行
|
||
payload = {"data": FX.PUBLIC_CORPUS[1][2], "src_group": "base",
|
||
"turn_ids": ["t1"], "turn_roles": [["t1", "user"]],
|
||
"input_sha256": group_input_sha("base")}
|
||
hit = {"id": comp_id, "memory": payload["data"]}
|
||
res = A.s7_chain_resolve(fake_memory(payload), hit, gate, **_kw(
|
||
A.s7_chain_resolve, hist_db=hist))
|
||
asm = A.s7_gated_assembly(fake_memory(payload), [hit], gate, **_kw(
|
||
A.s7_gated_assembly, hist_db=hist))
|
||
out = {"companion_raw_with_wrong_same_group_user_anchor": {
|
||
"resolver": {k: res.get(k) for k in ("resolved", "reject_reason", "provenance")},
|
||
"companion_raw_in_block": FX.PUBLIC_CORPUS[1][2] in (asm.get("block") or ""),
|
||
"assembly_pass": bool(asm.get("pass")),
|
||
}}
|
||
|
||
# 2b. turn_ids=[t2] 与 turn_roles=[(t1,user)] 互相矛盾
|
||
payload2 = dict(payload, data=FX.PUBLIC_CORPUS[0][2], turn_ids=["t2"])
|
||
res2 = A.s7_chain_resolve(fake_memory(payload2),
|
||
{"id": user_id, "memory": payload2["data"]}, gate, **_kw(
|
||
A.s7_chain_resolve, hist_db=hist))
|
||
out["turn_ids_turn_roles_disagree"] = {
|
||
"resolver": {k: res2.get(k) for k in ("resolved", "reject_reason", "provenance")},
|
||
}
|
||
|
||
# 正例：同轮真实用户原句 + 正确锚点必须保留
|
||
payload3 = dict(payload, data=FX.PUBLIC_CORPUS[0][2], turn_roles=[["t1", "user"]])
|
||
user_hit = {"id": user_id, "memory": FX.PUBLIC_CORPUS[0][2]}
|
||
res3 = A.s7_chain_resolve(fake_memory(payload3), user_hit, gate, **_kw(
|
||
A.s7_chain_resolve, hist_db=hist))
|
||
asm3 = A.s7_gated_assembly(fake_memory(payload3), [hit, user_hit], gate, **_kw(
|
||
A.s7_gated_assembly, hist_db=hist))
|
||
out["same_turn_user_positive_kept"] = {
|
||
"resolver": {k: res3.get(k) for k in ("resolved", "reject_reason", "provenance")},
|
||
"user_raw_in_block": FX.PUBLIC_CORPUS[0][2] in (asm3.get("block") or ""),
|
||
}
|
||
return out
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 反例1：真实 A.run 编排 + 假外部操作，捕获全部问答出口（零嵌入、零出站）
|
||
POISON = "来源不明且尚未分享的伴侣资料-MANAGER859"
|
||
PROMISE_TEXT = ("User said they would tell the assistant the interview result "
|
||
"on Friday evening.")
|
||
RESULT_TEXT = ("User passed the important interview and received the "
|
||
"notification.")
|
||
ANCHORED_PAYLOADS = {
|
||
"fake-base#0": {"data": FX.PUBLIC_CORPUS[0][2], "src_group": "base",
|
||
"turn_ids": ["t1"], "turn_roles": [["t1", "user"]],
|
||
"input_sha256": group_input_sha("base")},
|
||
"fake-base#1": {"data": FX.PUBLIC_CORPUS[1][2], "src_group": "base",
|
||
"turn_ids": ["t1"], "turn_roles": [["t1", "companion"]],
|
||
"input_sha256": group_input_sha("base")},
|
||
"fake-base#2": {"data": FX.PUBLIC_CORPUS[2][2], "src_group": "base",
|
||
"turn_ids": ["t2"], "turn_roles": [["t2", "user"]],
|
||
"input_sha256": group_input_sha("base")},
|
||
"fake-promise": {"data": PROMISE_TEXT, "src_group": "promise",
|
||
"turn_ids": ["p1"], "turn_roles": [["p1", "user"]],
|
||
"input_sha256": group_input_sha("promise")},
|
||
"fake-result": {"data": RESULT_TEXT, "src_group": "promise",
|
||
"turn_ids": ["p2"], "turn_roles": [["p2", "user"]],
|
||
"input_sha256": group_input_sha("promise")},
|
||
"unknown-origin": {"data": POISON},
|
||
}
|
||
FAKE_HITS = [{"id": "fake-base#0", "memory": ANCHORED_PAYLOADS["fake-base#0"]["data"]},
|
||
{"id": "unknown-origin", "memory": POISON},
|
||
{"id": "fake-base#1", "memory": ANCHORED_PAYLOADS["fake-base#1"]["data"]},
|
||
{"id": "fake-promise", "memory": PROMISE_TEXT},
|
||
{"id": "fake-result", "memory": RESULT_TEXT},
|
||
{"id": "fake-base#2", "memory": ANCHORED_PAYLOADS["fake-base#2"]["data"]}]
|
||
|
||
|
||
class FakeVecClient:
|
||
def retrieve(self, **kw):
|
||
ids = kw.get("ids") or []
|
||
return [SimpleNamespace(payload=ANCHORED_PAYLOADS.get(i))
|
||
for i in ids if i in ANCHORED_PAYLOADS]
|
||
|
||
def close(self):
|
||
pass
|
||
|
||
|
||
class FakeRunner:
|
||
"""替换 A.RealRunner：仅伪造外部操作（提取 add / 问答 ask / update），
|
||
编排、检索装配与门全部走真实代码路径。"""
|
||
|
||
def __init__(self, root: Path, ledger):
|
||
self.ledger = ledger
|
||
self.updates: list[str] = []
|
||
self.pub = SimpleNamespace(collection_name="cg_public",
|
||
vector_store=SimpleNamespace(client=FakeVecClient()))
|
||
root.mkdir(parents=True, exist_ok=True)
|
||
with sqlite3.connect(str(root / "mem0_public_history.db")) as c:
|
||
c.execute("CREATE TABLE IF NOT EXISTS messages(content TEXT)")
|
||
c.execute("INSERT INTO messages VALUES(?)", (FX.PUBLIC_CORPUS[0][2],))
|
||
c.execute("CREATE TABLE IF NOT EXISTS history("
|
||
"memory_id TEXT, old_memory TEXT, new_memory TEXT, "
|
||
"event TEXT, created_at TEXT)")
|
||
for mid, p in ANCHORED_PAYLOADS.items():
|
||
if mid != "unknown-origin":
|
||
c.execute("INSERT INTO history VALUES(?,?,?,?,?)",
|
||
(mid, "", p["data"], "ADD", "2026-10-26T00:00:00Z"))
|
||
|
||
def add(self, messages, metadata=None):
|
||
key = (metadata or {}).get("src_group", "x")
|
||
if key == "promise":
|
||
items = [{"id": "fake-promise", "memory": PROMISE_TEXT},
|
||
{"id": "fake-result", "memory": RESULT_TEXT}]
|
||
else:
|
||
items = [{"id": f"fake-{key}", "memory": "synthetic"}]
|
||
return {"results": items}
|
||
|
||
def update(self, memory_id, text=None, metadata=None):
|
||
self.updates.append(memory_id)
|
||
return {"updated": True}
|
||
|
||
def ask(self, question, context_block):
|
||
CAPTURED.append({"question": question, "context": context_block})
|
||
return {"answer": "不知道", "ok": True}
|
||
|
||
|
||
CAPTURED: list[dict] = []
|
||
|
||
|
||
def section_real_run_exits(fake_root: Path) -> dict:
|
||
global CAPTURED
|
||
CAPTURED = []
|
||
gate_calls: list[bool] = []
|
||
real_gate = A.s7_gated_assembly
|
||
|
||
def gate_spy(*a, **kw):
|
||
gate_calls.append(True)
|
||
return real_gate(*a, **kw)
|
||
|
||
def fake_search(m, query, limit=5):
|
||
return list(FAKE_HITS)
|
||
|
||
class FakeLedger:
|
||
def used(self, purpose="real_llm"):
|
||
return 0
|
||
|
||
def summary(self):
|
||
return {"debited": 0, "rows": []}
|
||
|
||
saved = {k: getattr(A, k) for k in (
|
||
"RealRunner", "mem0_search_texts", "AB", "Ledger", "make_real_memory")}
|
||
saved_sub_run, saved_copytree, saved_copy2 = (
|
||
subprocess.run, shutil.copytree, shutil.copy2)
|
||
A.RealRunner = FakeRunner
|
||
A.mem0_search_texts = fake_search
|
||
A.AB = fake_root
|
||
A.Ledger = FakeLedger
|
||
A.make_real_memory = lambda role, root, ledger: SimpleNamespace(
|
||
vector_store=SimpleNamespace(client=FakeVecClient()))
|
||
A.s7_gated_assembly = gate_spy
|
||
subprocess.run = lambda *a, **kw: SimpleNamespace(
|
||
stdout='{"ok": true}', returncode=0)
|
||
shutil.copytree = lambda *a, **kw: None
|
||
shutil.copy2 = lambda *a, **kw: None
|
||
run_error = None
|
||
try:
|
||
A.run()
|
||
except Exception as e: # 编排意外中断也如实记录（不掩盖）
|
||
run_error = f"{type(e).__name__}: {e}"
|
||
finally:
|
||
for k, v in saved.items():
|
||
setattr(A, k, v)
|
||
A.s7_gated_assembly = real_gate
|
||
subprocess.run, shutil.copytree, shutil.copy2 = (
|
||
saved_sub_run, saved_copytree, saved_copy2)
|
||
|
||
mem0_side = [c for c in CAPTURED if "她记得的更早的事" in (c["context"] or "")]
|
||
return {
|
||
"run_completed": run_error is None,
|
||
"run_error": run_error,
|
||
"captured_qa_exits": len(CAPTURED),
|
||
"mem0_side_gated_exits": len(mem0_side),
|
||
"poison_in_any_sent_context": any(POISON in (c["context"] or "") for c in CAPTURED),
|
||
"poison_in_any_gated_block": any(POISON in (c["context"] or "") for c in mem0_side),
|
||
"source_gate_calls": len(gate_calls),
|
||
"captured_qa": [{"question": c["question"],
|
||
"context_head": (c["context"] or "")[:60]} for c in CAPTURED],
|
||
"note": "真实 A.run 编排；仅外部操作（提取/搜索/问答/子进程/复制）为假实现；"
|
||
"socket 硬拒绝网络，本地嵌入 0",
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
def main(argv: list[str]) -> dict:
|
||
global PHASE
|
||
argv = list(argv)
|
||
rest: list[str] = []
|
||
PHASE = "postfix"
|
||
it = iter(argv[1:])
|
||
for tok in it:
|
||
if tok == "--phase":
|
||
PHASE = next(it, "postfix")
|
||
elif tok.startswith("--phase="):
|
||
PHASE = tok.split("=", 1)[1]
|
||
elif not tok.startswith("--"):
|
||
rest.append(tok)
|
||
args = rest
|
||
OUT["phase"] = PHASE
|
||
|
||
commit = subprocess.run(["git", "rev-parse", "--short", "HEAD"], capture_output=True,
|
||
text=True, cwd=str(REPO)).stdout.strip()
|
||
OUT["commit"] = commit
|
||
|
||
OUT["criteria_negatives"] = section_criteria()
|
||
hist = DATA / "anchor-store" / "mem0_public_history.db"
|
||
ctrl = DATA / "anchor-store" / "share_gate.sqlite3"
|
||
if not hist.exists(): # 从总管 audit859 store 快照复制（只读使用，零嵌入）
|
||
src = Path("/home/xiaomo/multica_workspaces/hello-af23464d222e/"
|
||
"hel-859-1f079e50f551/workdir/audit-data/audit859")
|
||
hist.parent.mkdir(parents=True, exist_ok=True)
|
||
shutil.copy2(src / "store" / "mem0_public_history.db", hist)
|
||
shutil.copy2(src / "share_gate.sqlite3", ctrl)
|
||
OUT["anchor_negatives"] = section_anchors(hist, ctrl)
|
||
|
||
fake_root = DATA / "fake-end-to-end"
|
||
if fake_root.exists():
|
||
shutil.rmtree(fake_root)
|
||
OUT["real_run_exit_capture"] = section_real_run_exits(fake_root)
|
||
OUT["resource_after"] = {"embed_local": 0, "note": "全部假后端/纯函数，零嵌入"}
|
||
|
||
c = OUT["criteria_negatives"]
|
||
a = OUT["anchor_negatives"]
|
||
r = OUT["real_run_exit_capture"]
|
||
negatives_reproduced = (
|
||
c["S6_resolved_answer_says_open"]["pass"]
|
||
and c["S6_both_answers_no_knowledge"]["pass"]
|
||
and c["S6_unresolved_claims_result_received"]["pass"]
|
||
and c["S6_unresolved_same_synonym_received"]["pass"]
|
||
and c["S2_details_absent_from_both_contexts"]["pass"]
|
||
and a["companion_raw_with_wrong_same_group_user_anchor"]["resolver"]["resolved"]
|
||
and a["companion_raw_with_wrong_same_group_user_anchor"]["companion_raw_in_block"]
|
||
and a["turn_ids_turn_roles_disagree"]["resolver"]["resolved"]
|
||
and r["poison_in_any_sent_context"])
|
||
positives_kept = (
|
||
c["S6_positive"]["pass"] and c["S2_positive"]["pass"]
|
||
and c["S2_unknown_answers"]["pass"] is False
|
||
and a["same_turn_user_positive_kept"]["resolver"]["resolved"]
|
||
and a["same_turn_user_positive_kept"]["user_raw_in_block"])
|
||
negatives_rejected = (
|
||
not c["S6_resolved_answer_says_open"]["pass"]
|
||
and not c["S6_both_answers_no_knowledge"]["pass"]
|
||
and not c["S6_unresolved_claims_result_received"]["pass"]
|
||
and not c["S6_unresolved_same_synonym_received"]["pass"]
|
||
and not c["S2_details_absent_from_both_contexts"]["pass"]
|
||
and not a["companion_raw_with_wrong_same_group_user_anchor"]["resolver"]["resolved"]
|
||
and not a["companion_raw_with_wrong_same_group_user_anchor"]["companion_raw_in_block"]
|
||
and not a["turn_ids_turn_roles_disagree"]["resolver"]["resolved"]
|
||
and not r["poison_in_any_sent_context"]
|
||
and r["captured_qa_exits"] >= 8
|
||
and r["mem0_side_gated_exits"] >= 5
|
||
and r["source_gate_calls"] >= 5
|
||
and r["run_completed"])
|
||
OUT["summary"] = {
|
||
"phase": PHASE,
|
||
"negatives_reproduced": negatives_reproduced,
|
||
"negatives_rejected": negatives_rejected,
|
||
"positives_kept": positives_kept,
|
||
"pass": (negatives_reproduced and PHASE == "prefix")
|
||
or (negatives_rejected and positives_kept and PHASE == "postfix"),
|
||
}
|
||
out_path = Path(args[0]) if args else (
|
||
REPO / "spike/hel857/results/"
|
||
f"audit860_{'prefix' if PHASE == 'prefix' else 'postfix'}.json")
|
||
out_path.parent.mkdir(parents=True, exist_ok=True)
|
||
out_path.write_text(json.dumps(OUT, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
print(json.dumps(OUT["summary"] | {"out": str(out_path)}, ensure_ascii=False))
|
||
return OUT
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(0 if main(sys.argv)["summary"]["pass"] else 1)
|