13.2 工具链的分层与复用

{
  "version": 1,
  "project": "stardust-inn",
  "note": "13.2 工具链分层与复用账。layers 按 level 从小到大排列，level 越大越靠近「入口」。依赖只允许单向向下：edge.from 的 level 必须严格大于 edge.to 的 level，否则记为违规（含同层耦合与反向跨层）。reuseThreshold = 下沉所需的消费方数下限（≥ 2 才值得进共享层）；exemptLayerIds 里的层是天然单消费方的管线入口，不计「过共享」。reuseGainX100 = (consumers - 1) × localCostX100，即「共享一次替 N 次各写一遍」省下的人天×100。全部为示例参数。",
  "reuseThreshold": 2,
  "exemptLayerIds": ["L2-task"],
  "layers": [
    { "id": "L0-kernel", "name": "纯函数内核", "level": 0, "note": "无 IO、可单测、跨项目复用" },
    { "id": "L1-adapter", "name": "适配层", "level": 1, "note": "读 DCC 导出、写引擎格式" },
    { "id": "L2-task", "name": "任务编排", "level": 2, "note": "管线入口，串起适配层；天然单消费方" },
    { "id": "L3-service", "name": "平台服务", "level": 3, "note": "跨项目共享的调度与看板" }
  ],
  "tools": [
    { "id": "t-naming-kernel", "layer": "L0-kernel", "consumers": 3, "localCostX100": 600 },
    { "id": "t-atlas-pack", "layer": "L0-kernel", "consumers": 2, "localCostX100": 800 },
    { "id": "t-color-kernel", "layer": "L0-kernel", "consumers": 1, "localCostX100": 500 },
    { "id": "t-dcc-export", "layer": "L1-adapter", "consumers": 3, "localCostX100": 700 },
    { "id": "t-engine-import", "layer": "L1-adapter", "consumers": 2, "localCostX100": 700 },
    { "id": "t-pipeline-run", "layer": "L2-task", "consumers": 1, "localCostX100": 400 },
    { "id": "t-vfx-check", "layer": "L2-task", "consumers": 1, "localCostX100": 300 },
    { "id": "t-dashboard", "layer": "L3-service", "consumers": 3, "localCostX100": 900 }
  ],
  "edges": [
    { "from": "t-pipeline-run", "to": "t-dcc-export" },
    { "from": "t-pipeline-run", "to": "t-engine-import" },
    { "from": "t-dcc-export", "to": "t-naming-kernel" },
    { "from": "t-engine-import", "to": "t-atlas-pack" },
    { "from": "t-vfx-check", "to": "t-atlas-pack" },
    { "from": "t-dashboard", "to": "t-pipeline-run" },
    { "from": "t-color-kernel", "to": "t-engine-import" },
    { "from": "t-vfx-check", "to": "t-dashboard" },
    { "from": "t-dcc-export", "to": "t-engine-import" }
  ]
}

#!/usr/bin/env python3
# file: sim/ta13_toolchain.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/ta13_toolchain.py
"""13.2 工具链的分层与复用：依赖方向账 + 复用收益账。

读 config/ta13_toolchain.json。① 依赖方向：edge.from 的 level 必须严格大于 edge.to 的
level，否则违规（含同层耦合）；② 复用收益 reuseGainX100 = (consumers - 1) × localCostX100；
③ 过共享：consumers < reuseThreshold 却摆在共享层（exemptLayerIds 除外）的工具应下沉为本地。
核心观点：**环检测只查得到「转圈」，查不到「方向反了却不转圈」**——t-color-kernel(0) →
t-engine-import(1) 与 t-vfx-check(2) → t-dashboard(3) 都是反向依赖，却都不成环。
产出 out/ta13_toolchain.json，另落 canonical 摘要与 sha256 前 16 位供 P 层对照。"""
import hashlib
import json
import os

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
CONFIG = os.path.join(ROOT, "config/ta13_toolchain.json")
OUT = os.path.join(ROOT, "out")


def analyze(cfg):
    lbi = {l["id"]: l for l in cfg["layers"]}
    tbi = {t["id"]: t for t in cfg["tools"]}
    thr, exempt = cfg["reuseThreshold"], set(cfg["exemptLayerIds"])
    rows = [{"id": t["id"], "layer": t["layer"], "level": lbi[t["layer"]]["level"],
             "consumers": t["consumers"], "reuseGainX100": max((t["consumers"] - 1) * t["localCostX100"], 0),
             "shouldShare": t["consumers"] >= thr,
             "overShared": t["consumers"] < thr and t["layer"] not in exempt} for t in cfg["tools"]]
    # 依赖必须严格向下：to.level < from.level；否则违规（含同层耦合）。
    viol = [{"from": e["from"], "to": e["to"],
             "fromLevel": lbi[tbi[e["from"]]["layer"]]["level"],
             "toLevel": lbi[tbi[e["to"]]["layer"]]["level"]}
            for e in cfg["edges"]
            if lbi[tbi[e["to"]]["layer"]]["level"] >= lbi[tbi[e["from"]]["layer"]]["level"]]
    canon = json.dumps({
        "tools": [{"id": r["id"], "consumers": r["consumers"], "reuseGainX100": r["reuseGainX100"],
                   "shouldShare": r["shouldShare"]} for r in rows],
        "violations": [{"from": v["from"], "to": v["to"]} for v in viol],
    }, separators=(",", ":"), ensure_ascii=False)
    return {
        "version": cfg["version"], "project": cfg["project"], "layerCount": len(cfg["layers"]),
        "toolCount": len(rows), "edgeCount": len(cfg["edges"]), "reuseThreshold": thr, "rows": rows,
        "violations": viol, "violationCount": len(viol),
        "sharedToolCount": sum(1 for r in rows if r["shouldShare"]),
        "overSharedCount": sum(1 for r in rows if r["overShared"]),
        "overSharedIds": [r["id"] for r in rows if r["overShared"]],
        "totalReuseGainX100": sum(r["reuseGainX100"] for r in rows),
        "canonical": canon, "digest": hashlib.sha256(canon.encode("utf-8")).hexdigest()[:16],
    }


def main():
    rep = analyze(json.load(open(CONFIG, encoding="utf-8")))
    os.makedirs(OUT, exist_ok=True)
    with open(os.path.join(OUT, "ta13_toolchain.json"), "w", encoding="utf-8") as f:
        json.dump(rep, f, ensure_ascii=False, indent=2)
        f.write("\n")
    print(f"layers={rep['layerCount']} tools={rep['toolCount']} edges={rep['edgeCount']} "
          f"violations={rep['violationCount']}")
    for v in rep["violations"]:
        print(f"  VIOLATION {v['from']}(L{v['fromLevel']}) -> {v['to']}(L{v['toLevel']})")
    for r in rep["rows"]:
        print(f"  {r['id']:<18} {r['layer']:<11} consumers={r['consumers']} gainX100={r['reuseGainX100']:<5} "
              f"share={r['shouldShare']}{' OVER-SHARED' if r['overShared'] else ''}")
    print(f"shared={rep['sharedToolCount']} overShared={rep['overSharedCount']} "
          f"totalReuseGainX100={rep['totalReuseGainX100']}")
    print(f"digest={rep['digest']}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

layers=4 tools=8 edges=9 violations=3
  VIOLATION t-color-kernel(L0) -> t-engine-import(L1)
  VIOLATION t-vfx-check(L2) -> t-dashboard(L3)
  VIOLATION t-dcc-export(L1) -> t-engine-import(L1)
  t-naming-kernel    L0-kernel   consumers=3 gainX100=1200  share=True
  t-atlas-pack       L0-kernel   consumers=2 gainX100=800   share=True
  t-color-kernel     L0-kernel   consumers=1 gainX100=0     share=False OVER-SHARED
  t-dcc-export       L1-adapter  consumers=3 gainX100=1400  share=True
  t-engine-import    L1-adapter  consumers=2 gainX100=700   share=True
  t-pipeline-run     L2-task     consumers=1 gainX100=0     share=False
  t-vfx-check        L2-task     consumers=1 gainX100=0     share=False
  t-dashboard        L3-service  consumers=3 gainX100=1800  share=True
shared=5 overShared=1 totalReuseGainX100=5900
digest=356f2dfa88c24af1

// file: core/src/ta13/toolchain.ts
// tier: P
// verify: node --test 'core/test/ta13_*.test.ts'
// 13.2 工具链分层内核：依赖方向检测 + 复用收益账。
//
// 关键设计：**分层判据是「单向向下」，不是「无环」**。一条反向依赖（低层指向高层）
// 完全可以不成环，环检测因此放行它 —— 那正是分层被悄悄破坏的方式。这里直接按 level 判：
// to.level 必须严格小于 from.level，同层耦合也算违规。
// 复用收益一律整数：reuseGainX100 = (consumers - 1) × localCostX100，负值截为 0。
// 只允许可擦除 TS 语法（禁 enum / namespace / 参数属性 / 装饰器）。

export type Layer = { id: string; name: string; level: number; note: string };
export type Tool = { id: string; layer: string; consumers: number; localCostX100: number };
export type Edge = { from: string; to: string };
export type Violation = { from: string; to: string; fromLevel: number; toLevel: number };

/** 依赖必须严格向下：to.level < from.level，否则违规（含同层耦合）。 */
export function violations(edges: Edge[], tools: Tool[], layers: Layer[]): Violation[] {
  const levelOf = (toolId: string): number => {
    const t = tools.find((x) => x.id === toolId);
    if (!t) throw new Error(`unknown tool ${toolId}`);
    const l = layers.find((x) => x.id === t.layer);
    if (!l) throw new Error(`unknown layer ${t.layer}`);
    return l.level;
  };
  const out: Violation[] = [];
  for (const e of edges) {
    const lf = levelOf(e.from), lt = levelOf(e.to);
    if (lt >= lf) out.push({ from: e.from, to: e.to, fromLevel: lf, toLevel: lt });
  }
  return out;
}

/** 复用收益（人天×100）：(consumers - 1) × localCostX100，负值截为 0。 */
export function reuseGainX100(t: Tool): number {
  return Math.max((t.consumers - 1) * t.localCostX100, 0);
}

/** 消费方数达到下限才值得下沉到共享层。 */
export function shouldShare(t: Tool, threshold: number): boolean {
  return t.consumers >= threshold;
}

/** 过共享：消费方不足却摆在共享层（天然单消费方的层除外）。 */
export function overShared(t: Tool, threshold: number, exemptLayerIds: string[]): boolean {
  return t.consumers < threshold && !exemptLayerIds.includes(t.layer);
}

/** canonical 摘要：键序与 Python 侧一致，供 digest 对拍。 */
export function canonical(rows: { id: string; consumers: number; reuseGainX100: number; shouldShare: boolean }[],
                          viol: Violation[]): string {
  return JSON.stringify({
    tools: rows.map((r) => ({
      id: r.id, consumers: r.consumers,
      reuseGainX100: r.reuseGainX100, shouldShare: r.shouldShare,
    })),
    violations: viol.map((v) => ({ from: v.from, to: v.to })),
  });
}

// file: core/test/ta13_toolchain.test.ts
// tier: P
// verify: node --test 'core/test/ta13_*.test.ts'
// 工具链分层断言：TS 侧独立复算 config/ta13_toolchain.json，与 Python（sim/ta13_toolchain.py）
// 落盘的 out/ta13_toolchain.json 逐位比对 + digest 对照。负例守住「环检测 ≠ 分层校验」。
import test from "node:test";
import assert from "node:assert/strict";
import { readFileSync } from "node:fs";
import { createHash } from "node:crypto";
import { dirname, resolve } from "node:path";
import { fileURLToPath } from "node:url";
import { violations, reuseGainX100, shouldShare, overShared, canonical } from "../src/ta13/toolchain.ts";
import type { Layer, Tool, Edge, Violation } from "../src/ta13/toolchain.ts";

const ROOT = resolve(dirname(fileURLToPath(import.meta.url)), "../..");
const cfg = JSON.parse(readFileSync(resolve(ROOT, "config/ta13_toolchain.json"), "utf8"));
const py = JSON.parse(readFileSync(resolve(ROOT, "out/ta13_toolchain.json"), "utf8"));
const layers = cfg.layers as Layer[];
const tools = cfg.tools as Tool[];
const edges = cfg.edges as Edge[];
const threshold = cfg.reuseThreshold as number;
const exempt = cfg.exemptLayerIds as string[];
const sha16 = (s: string): string => createHash("sha256").update(s, "utf8").digest("hex").slice(0, 16);
const rows = tools.map((t) => ({
  id: t.id, consumers: t.consumers,
  reuseGainX100: reuseGainX100(t), shouldShare: shouldShare(t, threshold),
}));
const viol = violations(edges, tools, layers);

test("跨语言：复用收益与违规集逐位一致，canonical 与 digest 一致", () => {
  assert.equal(rows.length, py.toolCount);
  for (const r of rows) {
    const t = py.rows.find((x: { id: string }) => x.id === r.id);
    assert.equal(r.reuseGainX100, t.reuseGainX100, r.id);
    assert.equal(r.shouldShare, t.shouldShare, r.id);
  }
  assert.deepEqual(viol.map((v) => [v.from, v.to]),
    (py.violations as Violation[]).map((v) => [v.from, v.to]));
  const canon = canonical(rows, viol);
  assert.equal(canon, py.canonical);
  assert.equal(sha16(canon), py.digest);
  assert.equal(py.digest, "356f2dfa88c24af1");
  assert.equal(py.violationCount, 3);
  assert.equal(py.edgeCount, 9);
  assert.equal(py.totalReuseGainX100, 5900);
  assert.deepEqual(py.overSharedIds, ["t-color-kernel"]);
});

test("负例：反向依赖不成环，环检测会放行，分层校验必须拦下", () => {
  // 朴素检测：只找环。t-color-kernel(0)→t-engine-import(1) 是反向边，但无回边，故不成环
  const adj: Record<string, string[]> = {};
  for (const e of edges) (adj[e.from] ||= []).push(e.to);
  const state: Record<string, number> = {};
  const dfs = (u: string): boolean => {
    state[u] = 1;
    for (const v of adj[u] || []) if (state[v] === 1 || (!state[v] && dfs(v))) return true;
    state[u] = 2;
    return false;
  };
  assert.equal(Object.keys(adj).some((u) => !state[u] && dfs(u)), false, "这套边里没有环");
  const backward = viol.filter((v) => v.toLevel > v.fromLevel);
  assert.equal(backward.length, 2);
  assert.ok(backward.some((v) => v.from === "t-color-kernel" && v.to === "t-engine-import"));
  assert.ok(backward.some((v) => v.from === "t-vfx-check" && v.to === "t-dashboard"));
});

test("负例：同层耦合也是违规，过共享按消费方数判定", () => {
  const same = viol.find((v) => v.fromLevel === v.toLevel);
  assert.ok(same, "同层边应记为违规");
  assert.equal(same?.from, "t-dcc-export");
  const color = tools.find((t) => t.id === "t-color-kernel") as Tool;
  assert.equal(overShared(color, threshold, exempt), true);
  assert.equal(reuseGainX100(color), 0);
});

✔ 跨语言：复用收益与违规集逐位一致，canonical 与 digest 一致
✔ 负例：反向依赖不成环，环检测会放行，分层校验必须拦下
✔ 负例：同层耦合也是违规，过共享按消费方数判定
ℹ tests 3
ℹ pass 3
ℹ fail 0