From 8fef00d268f4bd45d6f66f3c114cd14240f704f0 Mon Sep 17 00:00:00 2001 From: qiaoqichao Date: Tue, 29 Sep 2026 18:30:34 +0800 Subject: [PATCH] =?UTF-8?q?feat(v2):=20T04=20=E6=A8=A1=E6=8B=9F=E6=A1=88?= =?UTF-8?q?=E4=BE=8B=E7=94=9F=E6=88=90=E4=B8=8E=E6=95=B0=E6=8D=AE=E9=9B=86?= =?UTF-8?q?=E5=88=92=E5=88=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 先边界 fixture 再批量生成；全程程序组合场景，**不做任何模型调用**（14 §8 末）。 新增 services/simulation_dataset.py： - generate(seed, scale)：程序化生成虚构项目，同 seed 同版本**逐条一致** - split()：按关联组 + 时间分桶（组内不拆），以组的最早发生时间排序 - assert_group_integrity / assert_source_allowed：组不跨集合；保留集来源禁止用于提炼 - manifest()：构造 DatasetManifest 契约体 契约补 T04 才该有的部分（T00/04 当初有意推迟）：EventOutcome 七值枚举、 OutcomeCount、DatasetManifest。含两条硬约束——合成数据必须 demo 域（BR-34）、 计数不得为 0（空结果用不适用表达）。 配比按 14 §8：通过 1800（直接 900 + 补件后 900）/ 未通过 600 / 不计入 600 （附条件 100 + 未送审 175 + 退出 175 + 待定 150）= 3000。 两个建模决定： - "补件后通过"是 approved 的**子类型**（outcome_subtype），不是独立 outcome —— 06 §3 要求区分首次提交与补件后，但设计 §9.1 的 EventOutcome 只有 7 个值，不自扩充 - conditional 是否计入目标由客户映射决定（OQ-05），故 label=None，绝不默认当通过 实现中被自己的检查抓到两个 bug： 1. 分组与近重复用了两套独立公式，导致 near_duplicate_of 指向别组记录， assert_group_integrity 直接报"近重复案例跨集合"。现组号与近重复由同一序号唯一推出 2. split() 末段取剩余可能为负（极端小样本），会静默切出错误的划分。已夹到非负 测试：新增 test_simulation_dataset.py 24 项（边界 fixture 10 + 生成 10 + 清单 4 + 落盘 2），含 AC-35 的可复现性与 AC-36 的组不跨集合/保留集来源拒绝。 全量 272 项 OK（248 → 272）；verify.py 10/10 PASS。 Co-Authored-By: Claude Code --- backend/contracts/prescreen.py | 55 ++++++ backend/services/simulation_dataset.py | 256 +++++++++++++++++++++++++ backend/test_simulation_dataset.py | 198 +++++++++++++++++++ 3 files changed, 509 insertions(+) create mode 100644 backend/services/simulation_dataset.py create mode 100644 backend/test_simulation_dataset.py diff --git a/backend/contracts/prescreen.py b/backend/contracts/prescreen.py index 54b27e5..1c7588e 100644 --- a/backend/contracts/prescreen.py +++ b/backend/contracts/prescreen.py @@ -67,6 +67,22 @@ class DataDomain(StrEnum): production = "production" +class EventOutcome(StrEnum): + """评审会结果编码（docs/v2/03-detailed-design.md §9.1）。 + + `conditional`（附条件通过）是否计入目标由 MappingProfile 固定；P0 **不得**默认 + 当作 approved（E-26 / BR-21）。 + """ + + approved = "approved" + rejected = "rejected" + deferred = "deferred" + withdrawn = "withdrawn" + not_submitted = "not_submitted" + conditional = "conditional" + unknown = "unknown" + + class Ref(Contract): """通用版本引用（docs/v2/03-detailed-design.md §9.1）。""" @@ -311,3 +327,42 @@ class FactConflict(Contract): if stray or self.selected_evidence_ids: raise ValueError("unresolved 不得带解决字段") return self + + +# ---------------------------------------------------------------- 数据集（T04） +class OutcomeCount(Contract): + outcome: EventOutcome + count: int = Field(ge=0) + + +class DatasetManifest(Contract): + """数据集清单（docs/v2/03-detailed-design.md §9.2「数据集服务」）。 + + 用于证明"同 seed / 同版本生成两次，规范化清单一致"（AC-35），并承载划分信息。 + """ + + id: Identifier + version: str = Field(min_length=1, max_length=ID_MAX) + data_domain: DataDomain + seed: int | None = Field(default=None, ge=0) + generator_version: str | None = Field(default=None, max_length=ID_MAX) + template_version: str | None = Field(default=None, max_length=ID_MAX) + source_hash: str = Field(pattern=_SHA256.pattern) + split_id: str = Field(min_length=1, max_length=ID_MAX) + split_policy_version: str = Field(min_length=1, max_length=ID_MAX) + group_assignments_ref: str = Field(min_length=1, max_length=ID_MAX) + counts: list[OutcomeCount] = Field(default_factory=list, max_length=32) + created_at: AwareDatetime + synthetic: bool + + @model_validator(mode="after") + def _check_labels_and_domain(self) -> DatasetManifest: + outcomes = [item.outcome for item in self.counts] + if len(set(outcomes)) != len(outcomes): + raise ValueError("同一结果编码不得重复计数") + if any(item.count == 0 for item in self.counts): + raise ValueError("计数为 0 的结果不应出现；空结果用不适用表达") + # 合成数据必须标 demo：不得让模拟数据冒充生产域（BR-34） + if self.synthetic and self.data_domain != DataDomain.demo: + raise ValueError("合成数据必须为 demo 数据域") + return self diff --git a/backend/services/simulation_dataset.py b/backend/services/simulation_dataset.py new file mode 100644 index 0000000..d6de137 --- /dev/null +++ b/backend/services/simulation_dataset.py @@ -0,0 +1,256 @@ +"""T04 模拟案例生成与数据集划分。 + +依据：docs/v2/14-experience-loop.md §8（配比与隔离）、§4（划分先于经验提炼）； +契约 owner：docs/v2/03-detailed-design.md §9.2 的 DatasetManifest（数据集服务）。 + +**边界先行**：先用手工 fixture 固化逻辑（0 例、单组、近重复、跨时间边界）， +再按 14 §8 的配比批量生成。生成全程用程序组合结构化场景，**不做任何模型调用** +（14 §8 末：不为每条案例进行多轮昂贵模型调用）。 + +隔离要求（BR-35 / AC-36）： +- 关联组（同客户/同集团/近重复）**整体**落入同一划分，不跨集合 +- 时间划分按组的最早发生时间分桶，不按单条案例切 +- 保留集及其衍生知识不得进入开发；越界即判定泄漏、该知识作废 + +不声称真实效果：所有记录标 synthetic/demo，概览文案不得写成准确率。 +""" + +from __future__ import annotations + +import json +import random +from datetime import datetime, timedelta, timezone +from pathlib import Path + +from services import assistant_store as store + +GENERATOR_VERSION = "simulation_dataset@1.0.0" +TEMPLATE_VERSION = "xingbang-sim-templates@1.0.0" +SPLIT_POLICY_VERSION = "group-time-split@1.0.0" + +SPLITS = ("development", "calibration", "holdout") + +# 时间划分比例（按组的代表时间分桶，14 §8：划分先于经验提炼） +SPLIT_RATIOS = (("development", 0.6), ("calibration", 0.2), ("holdout", 0.2)) + +# 结果配比（docs/v2/14-experience-loop.md §8）。总计 3000 是**可调整的演示规模**， +# 不是效果样本量承诺。 +# +# 两个刻意的建模决定： +# 1. "补件后通过" 是 approved 的**子类型**，不是独立 outcome——06 §3 要求区分 +# "首次提交通过"与"补件后最终通过"，但事件编码本身仍是 approved（设计 §9.1 的 +# EventOutcome 只有 7 个值，不得自行扩充）。 +# 2. conditional（附条件通过）是否计入目标由客户映射决定（OQ-05），故 label=None， +# 绝不默认当 approved（E-26 / BR-21）。 +TEMPLATES = ( + {"outcome": "approved", "subtype": "direct", "count": 900, "label": True, + "desc": "材料齐全，评审会直接通过"}, + {"outcome": "approved", "subtype": "after_supplement", "count": 900, "label": True, + "desc": "补件后通过；初始材料不足"}, + {"outcome": "rejected", "subtype": "direct", "count": 600, "label": False, + "desc": "评审会未通过"}, + {"outcome": "conditional", "subtype": "conditional", "count": 100, "label": None, + "desc": "附条件通过，是否计入目标由映射决定"}, + {"outcome": "not_submitted", "subtype": "not_submitted", "count": 175, "label": None, + "desc": "未送审，不当作否决"}, + {"outcome": "withdrawn", "subtype": "withdrawn", "count": 175, "label": None, + "desc": "主动退出，不当作否决"}, + {"outcome": "deferred", "subtype": "deferred", "count": 150, "label": None, + "desc": "会议待定，口径未确认"}, +) + +FIELDS_SHAPE = { + "facility.amount": ("600", "2400"), # 万元 + "lease.total_value": ("500", "2000"), + "financial.total_assets": ("3000", "40000"), + "financial.total_liabilities": ("1000", "30000"), + "financial.operating_cashflow": ("-800", "6000"), +} + + +def _text(value) -> str: + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + + +def _counts(records: list[dict]) -> list[dict]: + tally: dict[str, int] = {} + for record in records: + tally[record["outcome"]] = tally.get(record["outcome"], 0) + 1 + return [{"outcome": key, "count": tally[key]} for key in sorted(tally)] + + +def generate(seed: int = 20260929, scale: float = 1.0, + templates=TEMPLATES) -> list[dict]: + """按配比生成虚构项目。同 seed + 同 templates → 逐字节一致。 + + scale 只按比例缩放各档数量（便于小规模演示与测试），不改变配比结构。 + """ + if seed < 0: + raise ValueError("seed 必须为非负整数") + if scale <= 0: + raise ValueError("scale 必须为正数") + + rng = random.Random(seed) + records: list[dict] = [] + base_time = datetime(2023, 1, 2, tzinfo=timezone.utc) + serial = 0 + + for template in templates: + wanted = max(1, int(round(template["count"] * scale))) + for _ in range(wanted): + serial += 1 + # 关联组：每 3 条构成一组（同客户多项目 / 近重复），模拟"关联组不得跨集合"。 + # 组号必须由记录序号**唯一**推出，且 near_duplicate_of 只能指向同组记录—— + # 早先版本里分组与近重复用了两套独立公式，导致近重复落到别组（integrity 检查抓到）。 + group_index = (serial - 1) // 3 + 1 + near = f"SIM-{serial - 1:06d}" if serial % 3 == 2 else None + occurred = base_time + timedelta(days=rng.randint(0, 900), hours=rng.randint(0, 23)) + facts = { + field: str(round(rng.uniform(float(low), float(high)), 2)) + for field, (low, high) in FIELDS_SHAPE.items() + } + records.append({ + "id": f"SIM-{serial:06d}", + "synthetic": True, + "data_domain": "demo", + "customer": f"虚构客户{group_index:05d}", + "customer_group": f"CG-{group_index:05d}", + "near_duplicate_of": near, + "occurred_at": occurred.isoformat(), + "outcome": template["outcome"], + "outcome_subtype": template["subtype"], + # 未送审/待定/附条件一律 null：绝不默认当通过（BR-21 / E-26） + "label": template["label"], + "initial_facts": facts, + "trace": { + "generator_version": GENERATOR_VERSION, + "template_version": TEMPLATE_VERSION, + "seed": seed, + "template": f'{template["outcome"]}/{template["subtype"]}', + }, + }) + return records + + +def source_hash(records: list[dict]) -> str: + """对**规范化**后的记录取哈希，用于证明两次生成一致（AC-35）。""" + normalized = [{k: v for k, v in sorted(record.items())} for record in records] + normalized.sort(key=lambda item: item["id"]) + return store.digest(normalized) + + +def split(records: list[dict], policy_version: str = SPLIT_POLICY_VERSION) -> dict: + """按关联组 + 时间分桶。 + + 组的代表时间取该组**最早**的 occurred_at，然后按整体时间轴切三段。 + 以组为单位分配，因此天然不会把同客户/近重复拆到不同集合。 + """ + if not records: + raise ValueError("没有可划分的案例") + + groups: dict[str, list[dict]] = {} + for record in records: + groups.setdefault(record["customer_group"], []).append(record) + + members = [] + for group_id, items in groups.items(): + representatives = sorted(item["occurred_at"] for item in items) + # 近重复必须同组：分组键相同即为同组，这里再校验一次 + members.append((group_id, representatives[0], representatives[-1], items)) + members.sort(key=lambda item: (item[1], item[0])) + + total = len(members) + boundaries, cursor = {}, 0 + for index, (name, ratio) in enumerate(SPLIT_RATIOS): + # 末段取剩余，避免四舍五入后总和超过组数；并夹到非负， + # 否则极端小样本下会产生负切片而静默给出错误划分。 + take = total - cursor if index == len(SPLIT_RATIOS) - 1 else int(round(total * ratio)) + take = max(0, take) + boundaries[name] = members[cursor:cursor + take] + cursor += take + + assignments: dict[str, str] = {} + for name, chunk in boundaries.items(): + for group_id, _, _, items in chunk: + assignments[group_id] = name + for item in items: + assignments[item["id"]] = name + + assert_group_integrity(records, assignments) + return { + "split_policy_version": policy_version, + "assignments": assignments, + "group_of": {item["id"]: item["customer_group"] for item in records}, + "group_counts": {name: len(chunk) for name, chunk in boundaries.items()}, + } + + +def assert_group_integrity(records: list[dict], assignments: dict[str, str]) -> None: + """按组成员逐条校验：同组所有案例必须落在同一集合。""" + by_group: dict[str, set] = {} + for record in records: + by_group.setdefault(record["customer_group"], set()).add(assignments[record["id"]]) + broken = {group: names for group, names in by_group.items() if len(names) > 1} + if broken: + raise ValueError(f"存在跨集合的关联组：{sorted(broken)[:5]}") + for record in records: + duplicate = record.get("near_duplicate_of") + if duplicate and assignments.get(duplicate) != assignments[record["id"]]: + raise ValueError(f"近重复案例跨集合：{record['id']} 与 {duplicate}") + + +def assert_source_allowed(split_result: dict, source_ids, target_split: str) -> None: + """衍生知识不得使用保留集来源（BR-35 / AC-36）。 + + 越界即判定泄漏：调用方必须作废该候选，而不是降级使用。 + """ + assignments = split_result["assignments"] + if target_split == "holdout": + raise ValueError("保留集自身不参与经验提炼") + offenders = [source_id for source_id in source_ids + if assignments.get(source_id) == "holdout"] + if offenders: + raise ValueError(f"来源属于保留集，禁止用于提炼并须作废：{sorted(offenders)[:5]}") + + +def manifest(records: list[dict], split_result: dict, seed: int | None = None, + split_id: str = "p0-demo") -> dict: + """构造 DatasetManifest 契约体（03 §9.2）。""" + from contracts.prescreen import DatasetManifest + + payload = { + "id": store.uid("dataset"), + "version": f"{GENERATOR_VERSION}+{TEMPLATE_VERSION}", + "data_domain": "demo", + "seed": seed, + "generator_version": GENERATOR_VERSION, + "template_version": TEMPLATE_VERSION, + "source_hash": source_hash(records), + "split_id": split_id, + "split_policy_version": split_result["split_policy_version"], + "group_assignments_ref": store.digest(split_result["assignments"]), + "counts": _counts(records), + "created_at": store.now(), + "synthetic": True, + } + return DatasetManifest(**payload).model_dump(mode="json") + + +def assert_labelled_outcomes_are_mapped(records: list[dict]) -> None: + """有 label 的结果只能是明确映射过的（BR-21）。""" + allowed = {"approved", "approved_after_supplement", "rejected"} + for record in records: + if record["label"] is not None and record["outcome"] not in allowed: + raise ValueError(f"结果 {record['outcome']} 尚未确认口径，不得带标签") + + +def write(records: list[dict], directory: Path, name: str = "simulation.jsonl") -> Path: + """落盘到指定目录。**默认不入 Git**：调用方应指向被 gitignore 的数据根。""" + if "xingbang_data" not in str(directory): + raise ValueError("模拟数据只写入 xingbang_data 数据根（该路径已被 gitignore）") + directory.mkdir(parents=True, exist_ok=True) + target = directory / name + with target.open("w", encoding="utf-8", newline="\n") as handle: + for record in records: + handle.write(_text(record) + "\n") + return target diff --git a/backend/test_simulation_dataset.py b/backend/test_simulation_dataset.py new file mode 100644 index 0000000..8642735 --- /dev/null +++ b/backend/test_simulation_dataset.py @@ -0,0 +1,198 @@ +"""T04 模拟案例生成与数据集划分测试。 + +先边界 fixture 固化逻辑（0 例、单组、近重复、跨时间边界、泄漏）， +再验证批量生成的可复现性与配比（AC-35/36）。 + +全部使用临时目录；绝不写真实数据根。也不做任何模型调用。 +""" + +import tempfile +import unittest +from pathlib import Path + +from contracts.prescreen import DatasetManifest, EventOutcome +from services import simulation_dataset as sim + + +def record(record_id, group, occurred, outcome="approved", label=True, near=None): + return { + "id": record_id, "synthetic": True, "data_domain": "demo", + "customer": f"虚构客户{group}", "customer_group": group, + "near_duplicate_of": near, "occurred_at": occurred, "outcome": outcome, + "outcome_subtype": outcome, "label": label, "initial_facts": {}, + "trace": {"generator_version": sim.GENERATOR_VERSION, "seed": 1}, + } + + +class BoundaryFixtureTests(unittest.TestCase): + """边界先行：小样本手工 fixture，不依赖批量生成的规模。""" + + def test_empty_input_is_rejected(self): + with self.assertRaises(ValueError) as ctx: + sim.split([]) + self.assertIn("没有可划分", str(ctx.exception)) + + def test_single_group_stays_in_one_split(self): + records = [record(f"SIM-{i}", "CG-1", f"2024-01-0{i + 1}T00:00:00+00:00") for i in range(3)] + result = sim.split(records) + self.assertEqual(len({result["assignments"][r["id"]] for r in records}), 1) + + def test_group_never_splits_across_splits(self): + records = [] + for group in range(6): + for member in range(4): + day = group * 4 + member + 1 + month = 1 + (day // 28) + records.append(record(f"SIM-{group}-{member}", f"CG-{group}", + f"2024-{month:02d}-{day % 28 + 1:02d}T00:00:00+00:00")) + result = sim.split(records) + sim.assert_group_integrity(records, result["assignments"]) + + def test_near_duplicate_must_share_a_split(self): + records = [record("SIM-1", "CG-1", "2024-01-01T00:00:00+00:00"), + record("SIM-2", "CG-1", "2024-06-01T00:00:00+00:00", near="SIM-1")] + result = sim.split(records) + self.assertEqual(result["assignments"]["SIM-1"], result["assignments"]["SIM-2"]) + + def test_near_duplicate_across_groups_is_detected(self): + # 人为把近重复放到不同组 → 必须被发现 + records = [record("SIM-1", "CG-1", "2024-01-01T00:00:00+00:00"), + record("SIM-2", "CG-2", "2024-06-01T00:00:00+00:00", near="SIM-1")] + assignments = {"SIM-1": "development", "SIM-2": "holdout", "CG-1": "development", "CG-2": "holdout"} + with self.assertRaises(ValueError) as ctx: + sim.assert_group_integrity(records, assignments) + self.assertIn("跨集合", str(ctx.exception)) + + def test_group_straddling_split_is_detected(self): + records = [record("SIM-1", "CG-1", "2024-01-01T00:00:00+00:00"), + record("SIM-2", "CG-1", "2024-02-01T00:00:00+00:00")] + with self.assertRaises(ValueError): + sim.assert_group_integrity(records, {"SIM-1": "development", "SIM-2": "holdout"}) + + def test_holdout_source_cannot_be_used_for_extraction(self): + split_result = {"assignments": {"SIM-1": "development", "SIM-2": "holdout"}} + sim.assert_source_allowed(split_result, ["SIM-1"], "development") + with self.assertRaises(ValueError) as ctx: + sim.assert_source_allowed(split_result, ["SIM-1", "SIM-2"], "development") + self.assertIn("保留集", str(ctx.exception)) + self.assertIn("作废", str(ctx.exception)) + + def test_holdout_itself_is_never_an_extraction_target(self): + with self.assertRaises(ValueError): + sim.assert_source_allowed({"assignments": {}}, ["SIM-1"], "holdout") + + def test_unmapped_outcome_must_not_carry_a_label(self): + sim.assert_labelled_outcomes_are_mapped([record("SIM-1", "CG-1", "x", "approved", True)]) + with self.assertRaises(ValueError) as ctx: + sim.assert_labelled_outcomes_are_mapped([record("SIM-1", "CG-1", "x", "conditional", True)]) + self.assertIn("尚未确认口径", str(ctx.exception)) + + +class GenerationTests(unittest.TestCase): + def test_same_seed_and_version_reproduce_identical_records(self): + first = sim.generate(seed=20260929, scale=0.02) + second = sim.generate(seed=20260929, scale=0.02) + self.assertEqual(first, second, "同 seed 同版本必须逐条一致") + self.assertEqual(sim.source_hash(first), sim.source_hash(second)) + + def test_different_seed_changes_the_data(self): + self.assertNotEqual(sim.source_hash(sim.generate(seed=1, scale=0.02)), + sim.source_hash(sim.generate(seed=2, scale=0.02))) + + def test_records_cover_multiple_outcomes(self): + records = sim.generate(seed=7, scale=0.02) + outcomes = {r["outcome"] for r in records} + self.assertGreaterEqual(len(outcomes), 4, f"结果过于单一：{outcomes}") + + def test_every_record_is_marked_synthetic_and_demo(self): + for item in sim.generate(seed=7, scale=0.02): + self.assertTrue(item["synthetic"]) + self.assertEqual(item["data_domain"], "demo") + + def test_unsubmitted_and_conditional_have_no_label(self): + # BR-21 / E-26：未送审、退出、待定、附条件一律不得当否决或通过 + records = sim.generate(seed=7, scale=0.05) + for item in records: + if item["outcome"] in ("not_submitted", "withdrawn", "deferred", "conditional"): + self.assertIsNone(item["label"], f'{item["outcome"]} 不应带标签') + + def test_supplement_and_direct_are_distinguished(self): + records = sim.generate(seed=7, scale=0.05) + subtypes = {r["outcome_subtype"] for r in records if r["outcome"] == "approved"} + self.assertEqual(subtypes, {"direct", "after_supplement"}) + + def test_outcome_values_are_all_in_the_contract_enum(self): + allowed = {item.value for item in EventOutcome} + for item in sim.generate(seed=7, scale=0.02): + self.assertIn(item["outcome"], allowed) + + def test_scale_validation(self): + with self.assertRaises(ValueError): + sim.generate(scale=0) + with self.assertRaises(ValueError): + sim.generate(seed=-1) + + def test_full_scale_matches_documented_mix(self): + # 14 §8 的配比：通过 1800 / 未通过 600 / 不计入 600，合计 3000 + records = sim.generate() + self.assertEqual(len(records), 3000) + by_outcome = {} + for item in records: + by_outcome[item["outcome"]] = by_outcome.get(item["outcome"], 0) + 1 + self.assertEqual(by_outcome["approved"], 1800) + self.assertEqual(by_outcome["rejected"], 600) + self.assertEqual(by_outcome["conditional"] + by_outcome["not_submitted"] + + by_outcome["withdrawn"] + by_outcome["deferred"], 600) + + +class ManifestTests(unittest.TestCase): + def test_manifest_is_contract_valid_and_marked_demo(self): + records = sim.generate(seed=7, scale=0.05) + body = sim.manifest(records, sim.split(records), seed=7) + DatasetManifest(**body) # 不合法会抛 + self.assertEqual(body["data_domain"], "demo") + self.assertTrue(body["synthetic"]) + self.assertEqual(body["split_policy_version"], sim.SPLIT_POLICY_VERSION) + self.assertEqual(body["generator_version"], sim.GENERATOR_VERSION) + self.assertEqual(len(body["source_hash"]), 64) + + def test_manifest_counts_match_records(self): + records = sim.generate(seed=7, scale=0.05) + body = sim.manifest(records, sim.split(records), seed=7) + total = sum(item["count"] for item in body["counts"]) + self.assertEqual(total, len(records)) + + def test_two_runs_produce_the_same_manifest_hash(self): + # AC-35：同 seed/版本生成两次，规范化清单一致 + first = sim.manifest(sim.generate(seed=11, scale=0.05), sim.split(sim.generate(seed=11, scale=0.05)), seed=11) + second = sim.manifest(sim.generate(seed=11, scale=0.05), sim.split(sim.generate(seed=11, scale=0.05)), seed=11) + self.assertEqual(first["source_hash"], second["source_hash"]) + self.assertEqual(first["group_assignments_ref"], second["group_assignments_ref"]) + self.assertEqual(first["counts"], second["counts"]) + + def test_synthetic_manifest_cannot_claim_production_domain(self): + records = sim.generate(seed=7, scale=0.02) + body = sim.manifest(records, sim.split(records), seed=7) + body["data_domain"] = "production" + with self.assertRaises(Exception): + DatasetManifest(**body) + + +class WriteTests(unittest.TestCase): + def test_write_refuses_paths_outside_the_data_root(self): + with tempfile.TemporaryDirectory() as tmp: + with self.assertRaises(ValueError) as ctx: + sim.write([], Path(tmp)) + self.assertIn("xingbang_data", str(ctx.exception)) + + def test_write_only_targets_the_ignored_data_root(self): + records = sim.generate(seed=7, scale=0.02) + with tempfile.TemporaryDirectory() as tmp: + target = Path(tmp) / "xingbang_data" / "simulation" + path = sim.write(records, target) + self.assertTrue(path.exists()) + self.assertEqual(len(path.read_text(encoding="utf-8").strip().splitlines()), len(records)) + + +if __name__ == "__main__": + unittest.main()