宜博 / lemongraph

#!/usr/bin/env python3
"""Mechanical Benchmark check for "Skill 详情页展示真实内容 (`v_skill_audit_log` 契约收口)".

Validates the target contract described in the approved Benchmark for this
goal: `src/skill_query.py::get_skill_content()`, the `GET /api/skills/<id>`
route in `src/backend_server.py`, and `frontend/app.js::mountSkillAuditLog`
must together show a Skill's real on-disk `SKILL.md` content next to its
existing iteration-audit panel, reject path traversal, and drop the old
`SkillContentMissingError` / `support_files` / `.../content` route names
without a compatibility shim -- and the change must actually be committed,
not just sitting in an uncommitted working-tree diff.

This script is intentionally self-contained (not wired into `python3 -m
unittest discover -s src`, so it can be dropped anywhere in the checkout and
still find the repo root on its own) and is expected to FAIL against the
pre-fix code -- that is its job: it is the acceptance test the Coder's
change must turn green, not a regression test for something already done.

Run:
    python3 scripts/benchmark_skill_content_contract.py -v
or, since it's a normal unittest module:
    python3 -m unittest -v scripts.benchmark_skill_content_contract
"""

import http.client
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import threading
import unittest
from unittest import mock


def _find_repo_root(start):
    d = start
    while True:
        if os.path.isfile(os.path.join(d, "src", "skill_query.py")):
            return d
        parent = os.path.dirname(d)
        if parent == d:
            raise RuntimeError("could not locate graphlemon repo root (no src/skill_query.py found above %s)" % start)
        d = parent


REPO_ROOT = _find_repo_root(os.path.dirname(os.path.abspath(__file__)))
if REPO_ROOT not in sys.path:
    sys.path.insert(0, REPO_ROOT)

from src import backend_server, schema, skill_query, skill_update  # noqa: E402

SAMPLE_CONTENT = "real skill markdown body\nwith <two> lines & a quote \"here\"\n"


def _init_git_repo(path):
    subprocess.run(["git", "init"], cwd=path, check=True, capture_output=True)


# ---------------------------------------------------------------------------
# 1. src/skill_query.py::get_skill_content() -- T1.1-T1.7
# ---------------------------------------------------------------------------
class SkillContentUnitTest(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        _init_git_repo(self.tmp.name)
        self.conn = schema.connect(":memory:")
        schema.create_schema(self.conn)
        self.conn.commit()
        self.addCleanup(self.conn.close)

    def _insert(self, skill_id, file_path, owner_org_node="coder", source="migrated_from_hermes",
                migrated_from_profile="code", current_git_commit_sha="abc123", status="active"):
        self.conn.execute(
            "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, "
            "migrated_from_profile, status) VALUES (?, ?, ?, ?, ?, ?, ?)",
            (skill_id, owner_org_node, file_path, current_git_commit_sha, source, migrated_from_profile, status),
        )
        self.conn.commit()

    def _write_skill_file(self, rel_dir, content=None, extra_files=None):
        skill_dir = os.path.join(self.tmp.name, rel_dir)
        os.makedirs(skill_dir, exist_ok=True)
        if content is not None:
            with open(os.path.join(skill_dir, "SKILL.md"), "w", encoding="utf-8") as f:
                f.write(content)
        for rel in extra_files or []:
            full = os.path.join(skill_dir, rel)
            os.makedirs(os.path.dirname(full), exist_ok=True)
            with open(full, "w", encoding="utf-8") as f:
                f.write("support file body")

    def test_t1_1_normal_content_exact_shape(self):
        self._insert("skill-1", "skills/skill-1")
        self._write_skill_file("skills/skill-1", content=SAMPLE_CONTENT)
        result = skill_query.get_skill_content(self.conn, "skill-1", repo_path=self.tmp.name)
        self.assertEqual(
            set(result.keys()),
            {"skill_id", "owner_org_node", "file_path", "current_git_commit_sha", "source",
             "migrated_from_profile", "status", "content", "supporting_files"},
            "get_skill_content must return exactly these 9 keys, field name is `supporting_files`",
        )
        self.assertEqual(result["skill_id"], "skill-1")
        self.assertEqual(result["owner_org_node"], "coder")
        self.assertEqual(result["file_path"], "skills/skill-1")
        self.assertEqual(result["current_git_commit_sha"], "abc123")
        self.assertEqual(result["source"], "migrated_from_hermes")
        self.assertEqual(result["migrated_from_profile"], "code")
        self.assertEqual(result["status"], "active")
        self.assertEqual(result["content"], SAMPLE_CONTENT)
        self.assertEqual(result["supporting_files"], [])

    def test_t1_2_missing_file_returns_none_content_no_exception(self):
        self._insert("skill-missing-file", "skills/skill-missing-file")
        result = skill_query.get_skill_content(self.conn, "skill-missing-file", repo_path=self.tmp.name)
        self.assertIsNone(result["content"])
        self.assertEqual(result["supporting_files"], [])
        self.assertEqual(result["skill_id"], "skill-missing-file")

    def test_t1_3_nonexistent_skill_raises_not_found(self):
        with self.assertRaises(skill_query.SkillNotFoundError):
            skill_query.get_skill_content(self.conn, "does-not-exist", repo_path=self.tmp.name)

    def test_t1_4_path_traversal_whole_path_escape_rejected_before_open(self):
        self._insert("skill-evil", "../../etc/passwd")
        with mock.patch("builtins.open") as mock_open:
            with self.assertRaises(skill_query.SkillPathTraversalError):
                skill_query.get_skill_content(self.conn, "skill-evil", repo_path=self.tmp.name)
            mock_open.assert_not_called()

    def test_t1_5_path_traversal_dotdot_segment_rejected_before_open(self):
        self._insert("skill-evil-2", "skills/../../../tmp/evil")
        with mock.patch("builtins.open") as mock_open:
            with self.assertRaises(skill_query.SkillPathTraversalError):
                skill_query.get_skill_content(self.conn, "skill-evil-2", repo_path=self.tmp.name)
            mock_open.assert_not_called()

    def test_t1_6_supporting_files_sorted_relative_excludes_main_file(self):
        self._insert("skill-with-support", "skills/skill-with-support")
        self._write_skill_file(
            "skills/skill-with-support", content="body\n",
            extra_files=["scripts/bar.py", "references/foo.md"],
        )
        result = skill_query.get_skill_content(self.conn, "skill-with-support", repo_path=self.tmp.name)
        self.assertEqual(result["supporting_files"], ["references/foo.md", "scripts/bar.py"])

    def test_t1_7_skill_content_missing_error_class_removed(self):
        for root, _dirs, files in os.walk(os.path.join(REPO_ROOT, "src")):
            for name in files:
                if not name.endswith(".py"):
                    continue
                full = os.path.join(root, name)
                with open(full, "r", encoding="utf-8", errors="replace") as f:
                    text = f.read()
                self.assertNotIn(
                    "SkillContentMissingError", text,
                    "SkillContentMissingError must be fully removed, still referenced in " + full,
                )


# ---------------------------------------------------------------------------
# 2. src/backend_server.py route layer -- T2.1-T2.7
# ---------------------------------------------------------------------------
class SkillContentRouteTest(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        _init_git_repo(self.tmp.name)
        self.server = backend_server.create_server(db_path=":memory:", skill_repo_path=self.tmp.name)
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        self.addCleanup(self._stop_server)

        self.server.conn.execute(
            "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, "
            "migrated_from_profile, status) VALUES ('skill-1', 'coder', 'skills/skill-1', 'abc123', "
            "'migrated_from_hermes', 'code', 'active')"
        )
        self.server.conn.execute(
            "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
            "VALUES ('skill-missing-file', 'coder', 'skills/skill-missing-file', 'unknown', 'new', 'active')"
        )
        self.server.conn.execute(
            "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
            "VALUES ('skill-evil', 'coder', '../../etc/passwd', 'unknown', 'new', 'active')"
        )
        self.server.conn.execute(
            "INSERT INTO bug_record (bug_id, source, description, root_cause_category, root_cause_detail, "
            "suggested_route, status) VALUES ('bug-1', 'validator', 'x', 'skill_defect', 'd', 'fast_lane', 'routed')"
        )
        self.server.conn.commit()

        skill_dir = os.path.join(self.tmp.name, "skills", "skill-1")
        os.makedirs(skill_dir, exist_ok=True)
        with open(os.path.join(skill_dir, "SKILL.md"), "w", encoding="utf-8") as f:
            f.write(SAMPLE_CONTENT)

    def _stop_server(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=5)

    def _request(self, method, path, payload=None):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        body = json.dumps(payload).encode("utf-8") if payload is not None else None
        conn.request(method, path, body=body)
        resp = conn.getresponse()
        raw = resp.read()
        conn.close()
        parsed = json.loads(raw) if raw else None
        return resp.status, parsed

    def _get(self, path):
        return self._request("GET", path)

    def test_t2_1_get_skill_detail_200_exact_shape(self):
        status, body = self._get("/api/skills/skill-1")
        self.assertEqual(status, 200)
        self.assertEqual(
            set(body.keys()),
            {"skill_id", "owner_org_node", "file_path", "current_git_commit_sha", "source",
             "migrated_from_profile", "status", "content", "supporting_files"},
        )
        self.assertEqual(body["content"], SAMPLE_CONTENT)
        self.assertEqual(body["supporting_files"], [])

    def test_t2_2_nonexistent_skill_404(self):
        status, body = self._get("/api/skills/does-not-exist")
        self.assertEqual(status, 404)
        self.assertEqual(body, {"error": "NOT_FOUND"})

    def test_t2_3_missing_file_still_200_null_content(self):
        status, body = self._get("/api/skills/skill-missing-file")
        self.assertEqual(status, 200)
        self.assertIsNone(body["content"])
        self.assertNotIn("error", body)

    def test_t2_4_path_traversal_400_invalid_skill_id(self):
        status, body = self._get("/api/skills/skill-evil")
        self.assertEqual(status, 400)
        self.assertEqual(body, {"error": "INVALID_SKILL_ID"})

    def test_t2_5_audit_and_diff_routes_not_swallowed_by_bare_skill_id_branch(self):
        status, body = self._get("/api/skills/skill-1/audit")
        self.assertEqual(status, 200)
        self.assertIsInstance(body, list, "/audit must still return the iteration-list shape, not skill detail JSON")

        status, body = self._get("/api/skills/diff")
        self.assertEqual(status, 400)
        self.assertEqual(body["error"], "INVALID_INPUT")

    def test_t2_6_rollback_route_unaffected(self):
        skill_update._write_and_commit(self.tmp.name, "skills/skill-1/SKILL.md", "v1 content", "seed skill")
        with self.server.conn_lock:
            result = skill_update.update_skill(
                self.server.conn, "skill-1", "bug-1", "v2 content", "improve it", self.tmp.name
            )
        status, body = self._request("POST", "/api/skills/skill-1/rollback", {"iteration_id": result["iteration_id"]})
        self.assertEqual(status, 200)
        self.assertIn("new_iteration_id", body)

    def test_t2_7_old_content_suffixed_route_now_404(self):
        status, _body = self._get("/api/skills/skill-1/content")
        self.assertEqual(status, 404)


# ---------------------------------------------------------------------------
# 4. Defense in depth -- T4.1
# ---------------------------------------------------------------------------
class DefenseInDepthTest(unittest.TestCase):
    def test_t4_1_sentinel_marker_never_leaks(self):
        sentinel_dir = tempfile.TemporaryDirectory()
        self.addCleanup(sentinel_dir.cleanup)
        with open(os.path.join(sentinel_dir.name, "SKILL.md"), "w", encoding="utf-8") as f:
            f.write("SENTINEL_MARKER_xyz")

        repo_tmp = tempfile.TemporaryDirectory()
        self.addCleanup(repo_tmp.cleanup)
        _init_git_repo(repo_tmp.name)
        conn = schema.connect(":memory:")
        schema.create_schema(conn)
        self.addCleanup(conn.close)

        skills_root = os.path.join(repo_tmp.name, "skills")
        rel = os.path.relpath(sentinel_dir.name, skills_root)  # dir with no trailing .md -> "<rel>/SKILL.md"
        conn.execute(
            "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
            "VALUES ('skill-evil', 'coder', ?, 'unknown', 'new', 'active')",
            (rel,),
        )
        conn.commit()

        leaked = False
        try:
            result = skill_query.get_skill_content(conn, "skill-evil", repo_path=repo_tmp.name)
            leaked = "SENTINEL_MARKER_xyz" in json.dumps(result, default=str)
        except skill_query.SkillPathTraversalError as e:
            leaked = "SENTINEL_MARKER_xyz" in str(e) or "SENTINEL_MARKER_xyz" in repr(e.args)
        self.assertFalse(leaked, "the sentinel marker must never appear in the return value or the raised exception")


# ---------------------------------------------------------------------------
# 3. frontend/app.js::mountSkillAuditLog -- T3.1-T3.4 (Node vm smoke test,
# same "no npm dependency" convention as frontend/test_graph_console_smoke.js)
# ---------------------------------------------------------------------------
_FRONTEND_SMOKE_JS = r"""
"use strict";
const fs = require("fs");
const vm = require("vm");
const appJsPath = process.argv[2];
const src = fs.readFileSync(appJsPath, "utf8");

function makeFakeDiv() {
  let text = "";
  return {
    set textContent(v) { text = v == null ? "" : String(v); },
    get textContent() { return text; },
    get innerHTML() {
      return text.replace(/&/g, "&amp;").replace(/</g, "&lt;").replace(/>/g, "&gt;").replace(/"/g, "&quot;");
    },
  };
}

function unescapeHtml(html) {
  return html
    .replace(/&quot;/g, '"')
    .replace(/&gt;/g, ">")
    .replace(/&lt;/g, "<")
    .replace(/&amp;/g, "&");
}

function makePanel() {
  return {
    _innerHTML: "",
    _textContent: "",
    style: { display: "" },
    get innerHTML() { return this._innerHTML; },
    set innerHTML(v) { this._innerHTML = v == null ? "" : String(v); },
    get textContent() { return this._textContent; },
    set textContent(v) { this._textContent = v == null ? "" : String(v); },
  };
}

function extractPreText(panel) {
  const m = panel.innerHTML.match(/<pre[^>]*>([\s\S]*?)<\/pre>/);
  return m ? unescapeHtml(m[1]) : null;
}

let elements = {};
global.document = {
  createElement: () => makeFakeDiv(),
  getElementById: (id) => elements[id],
  querySelectorAll: () => [],
  addEventListener: () => {},
};
global.window = { location: { protocol: "http:", host: "localhost", hash: "" }, addEventListener: () => {} };
global.localStorage = { getItem: () => null, setItem: () => {} };

// vm.runInThisContext (NOT vm.createContext) runs app.js in this process's
// own realm, so top-level `function ...` declarations attach straight to
// globalThis and fetch()'s real Promise machinery is the same Promise our
// assertions below use -- no cross-realm bridging needed for the async
// mountSkillAuditLog() flow.
vm.runInThisContext(src, { filename: "app.js" });

let failures = 0;
function check(cond, msg) {
  if (cond) { console.log("ok: " + msg); }
  else { failures++; console.error("FAIL: " + msg); }
}

function mockFetch(responses) {
  return function fetch(url) {
    const r = responses.find((entry) => entry.match(url));
    if (!r) throw new Error("unexpected fetch url in test: " + url);
    return new Promise((resolve) => {
      setTimeout(() => resolve({ status: r.status, json: async () => r.json }), r.delayMs || 0);
    });
  };
}

async function settle(ticks) {
  for (let i = 0; i < (ticks || 6); i++) {
    await new Promise((resolve) => setTimeout(resolve, 5));
  }
}

(async () => {
  // ---- T3.1: normal content renders ----
  {
    elements = { skill_content: makePanel(), skill_audit_list: makePanel(), skill_audit_error: makePanel() };
    global.fetch = mockFetch([
      { match: (u) => u.endsWith("/audit"), status: 200, json: [] },
      { match: (u) => !u.endsWith("/audit"), status: 200, json: {
        skill_id: "skill-1", owner_org_node: "coder", file_path: "skills/skill-1",
        current_git_commit_sha: "abc123", source: "migrated_from_hermes", migrated_from_profile: "code",
        status: "active", content: SAMPLE_CONTENT_PLACEHOLDER, supporting_files: [],
      } },
    ]);
    mountSkillAuditLog("skill-1");
    await settle();
    const preText = extractPreText(elements.skill_content);
    check(preText === SAMPLE_CONTENT_PLACEHOLDER, "T3.1 <pre> textContent equals raw content exactly, got: " + JSON.stringify(preText));
    check(elements.skill_content.innerHTML.includes("无支持文件"), "T3.1 empty supporting_files renders a '无支持文件' hint");
  }

  // ---- T3.2: content null ----
  {
    elements = { skill_content: makePanel(), skill_audit_list: makePanel(), skill_audit_error: makePanel() };
    global.fetch = mockFetch([
      { match: (u) => u.endsWith("/audit"), status: 200, json: [] },
      { match: (u) => !u.endsWith("/audit"), status: 200, json: {
        skill_id: "skill-missing-file", owner_org_node: "coder", file_path: "skills/skill-missing-file",
        current_git_commit_sha: "unknown", source: "new", migrated_from_profile: null,
        status: "active", content: null, supporting_files: [],
      } },
    ]);
    let threw = false;
    try {
      mountSkillAuditLog("skill-missing-file");
      await settle();
    } catch (e) { threw = true; console.error(e); }
    check(!threw, "T3.2 no uncaught exception when content is null");
    const preText = extractPreText(elements.skill_content);
    check(preText === "（内容文件不可用）", "T3.2 <pre> textContent is the exact placeholder for null content, got: " + JSON.stringify(preText));
  }

  // ---- T3.3: skill not found, shared error banner, order-independent ----
  async function runNotFoundOrdering(detailDelay, auditDelay, label) {
    elements = { skill_content: makePanel(), skill_audit_list: makePanel(), skill_audit_error: makePanel() };
    global.fetch = mockFetch([
      { match: (u) => u.endsWith("/audit"), status: 404, json: { error: "NOT_FOUND" }, delayMs: auditDelay },
      { match: (u) => !u.endsWith("/audit"), status: 404, json: { error: "NOT_FOUND" }, delayMs: detailDelay },
    ]);
    mountSkillAuditLog("skill-x");
    await settle(12);
    check(elements.skill_content.innerHTML === "", label + ": skill_content cleared");
    check(elements.skill_audit_list.innerHTML === "", label + ": skill_audit_list cleared");
    check(elements.skill_audit_error.style.display === "block", label + ": error banner visible");
    check(!!elements.skill_audit_error.textContent, label + ": error banner has non-empty text");
  }
  await runNotFoundOrdering(0, 25, "T3.3 detail-first");
  await runNotFoundOrdering(25, 0, "T3.3 audit-first");

  // ---- T3.4: supporting files listed ----
  {
    elements = { skill_content: makePanel(), skill_audit_list: makePanel(), skill_audit_error: makePanel() };
    global.fetch = mockFetch([
      { match: (u) => u.endsWith("/audit"), status: 200, json: [] },
      { match: (u) => !u.endsWith("/audit"), status: 200, json: {
        skill_id: "skill-with-support", owner_org_node: "coder", file_path: "skills/skill-with-support",
        current_git_commit_sha: "abc123", source: "migrated_from_hermes", migrated_from_profile: "code",
        status: "active", content: "body\n", supporting_files: ["references/foo.md", "scripts/bar.py"],
      } },
    ]);
    mountSkillAuditLog("skill-with-support");
    await settle();
    const html = elements.skill_content.innerHTML;
    check(html.includes("references/foo.md"), "T3.4 references/foo.md listed in the hint");
    check(html.includes("scripts/bar.py"), "T3.4 scripts/bar.py listed in the hint");
  }

  if (failures > 0) {
    console.error(failures + " assertion(s) failed");
    process.exit(1);
  }
  console.log("frontend/app.js skill-content smoke: all assertions passed");
})();
"""


class FrontendSkillContentSmokeTest(unittest.TestCase):
    def test_t3_1_to_t3_4_mount_skill_audit_log_contract(self):
        node = shutil.which("node")
        if node is None:
            self.skipTest("node not on PATH -- required by this repo's own frontend/test_graph_console_smoke.js too")

        app_js_path = os.path.join(REPO_ROOT, "frontend", "app.js")
        self.assertTrue(os.path.isfile(app_js_path), "frontend/app.js not found at " + app_js_path)

        workdir = tempfile.mkdtemp()
        self.addCleanup(shutil.rmtree, workdir, True)
        script_path = os.path.join(workdir, "skill_content_smoke.js")
        js_content = _FRONTEND_SMOKE_JS.replace(
            "SAMPLE_CONTENT_PLACEHOLDER", json.dumps(SAMPLE_CONTENT)
        )
        with open(script_path, "w", encoding="utf-8") as f:
            f.write(js_content)

        result = subprocess.run([node, script_path, app_js_path], capture_output=True, text=True, timeout=60)
        sys.stderr.write(result.stdout)
        sys.stderr.write(result.stderr)
        self.assertEqual(result.returncode, 0, "frontend smoke test failed, see stdout/stderr above")


# ---------------------------------------------------------------------------
# T1.7 (Python side, already covered above) + T3.5: static scan for
# residual old symbol/field/route names
# ---------------------------------------------------------------------------
class StaticScanTest(unittest.TestCase):
    def test_t3_5_no_residual_field_name_or_content_route_in_app_js(self):
        app_js = os.path.join(REPO_ROOT, "frontend", "app.js")
        with open(app_js, "r", encoding="utf-8") as f:
            text = f.read()
        bad_matches = re.findall(r"(?<!ing_)support_files\b", text)
        self.assertEqual(bad_matches, [], "app.js still references the old `support_files` field name")
        self.assertNotIn('/content"', text, 'app.js still builds a "/content"-suffixed fetch URL')


# ---------------------------------------------------------------------------
# 5. Full regression -- T5.1/T5.2 (exact commands from the Benchmark)
# ---------------------------------------------------------------------------
class FullRegressionTest(unittest.TestCase):
    def _run(self, args):
        return subprocess.run([sys.executable] + args, cwd=REPO_ROOT, capture_output=True, text=True)

    def test_t5_1_named_existing_test_targets_still_pass(self):
        commands = [
            ["-m", "unittest", "src.test_skill_query.ListSkillsTest", "-v"],
            ["-m", "unittest", "src.test_skill_query.GetSkillAuditLogTest", "-v"],
            ["-m", "unittest", "src.test_skill_query.GetDiffTest", "-v"],
            ["-m", "unittest", "src.test_backend_server.SkillRouteTest.test_list_returns_seeded_skill", "-v"],
            ["-m", "unittest", "src.test_backend_server.SkillRouteTest", "-k", "audit_log", "-v"],
            ["-m", "unittest", "src.test_backend_server.SkillRouteTest", "-k", "diff", "-v"],
            ["-m", "unittest", "src.test_backend_server.SkillRouteTest", "-k", "rollback", "-v"],
            ["-m", "unittest", "src.test_backend_server.SkillRouteTest.test_full_update_diff_rollback_cycle", "-v"],
        ]
        for cmd in commands:
            result = self._run(cmd)
            self.assertEqual(result.returncode, 0, "command failed: %s\n%s" % (cmd, result.stderr))

    def test_t5_2_full_src_suite_is_green(self):
        result = self._run(["-m", "unittest", "discover", "-s", "src"])
        self.assertEqual(result.returncode, 0, "src suite is not green:\n" + result.stderr[-6000:])


# ---------------------------------------------------------------------------
# 6. Delivery must be committed, not left in an uncommitted/worktree diff --
# T6.1-T6.3
# ---------------------------------------------------------------------------
class GitDeliveryStateTest(unittest.TestCase):
    TARGET_FILES = [
        "src/skill_query.py", "src/backend_server.py", "frontend/app.js",
        "src/test_skill_query.py", "src/test_backend_server.py",
    ]

    def _git(self, args, cwd):
        return subprocess.run(["git"] + args, cwd=cwd, capture_output=True, text=True)

    def test_t6_1_working_tree_clean_for_target_files(self):
        out = self._git(["status", "--porcelain", "--"] + self.TARGET_FILES, REPO_ROOT)
        self.assertEqual(out.returncode, 0, out.stderr)
        self.assertEqual(
            out.stdout.strip(), "",
            "target files must have zero uncommitted changes once the Unit is done:\n" + out.stdout,
        )

    def test_t6_2_commits_exist_and_touch_the_real_contract(self):
        newest_shas = []
        for f in ("src/skill_query.py", "src/backend_server.py", "frontend/app.js"):
            out = self._git(["log", "--oneline", "-1", "--", f], REPO_ROOT)
            self.assertEqual(out.returncode, 0, out.stderr)
            line = out.stdout.strip()
            self.assertTrue(line, "expected at least one commit touching " + f)
            newest_shas.append(line.split()[0])

        found_contract_evidence = False
        for sha in newest_shas:
            show = self._git(["show", sha], REPO_ROOT)
            if "SkillPathTraversalError" in show.stdout or "supporting_files" in show.stdout:
                found_contract_evidence = True
                break
        self.assertTrue(
            found_contract_evidence,
            "none of the newest commits touching the 3 core files mention SkillPathTraversalError/"
            "supporting_files -- looks like an unrelated commit merely touched these files, not the real fix",
        )

    def test_t6_3_changes_are_visible_from_the_main_checkout(self):
        common_dir = self._git(["rev-parse", "--path-format=absolute", "--git-common-dir"], REPO_ROOT)
        if common_dir.returncode != 0:
            self.skipTest("this git version doesn't support --git-common-dir, cannot resolve the main checkout")
        main_root = os.path.dirname(common_dir.stdout.strip())

        if os.path.normpath(main_root) == os.path.normpath(REPO_ROOT):
            self.skipTest("already running from the main checkout -- T6.1/T6.2 already covered this case")

        out = self._git(["status", "--porcelain", "--"] + self.TARGET_FILES, main_root)
        self.assertEqual(out.returncode, 0, out.stderr)
        self.assertEqual(out.stdout.strip(), "", "the main checkout must also be clean for the target files")

        probe = subprocess.run(
            [sys.executable, "-c",
             "import sys; sys.path.insert(0, sys.argv[1]); from src import skill_query; "
             "print(hasattr(skill_query, 'SkillPathTraversalError'))", main_root],
            capture_output=True, text=True,
        )
        self.assertEqual(
            probe.stdout.strip(), "True",
            "main checkout's src/skill_query.py does not define SkillPathTraversalError -- "
            "the fix may only exist in a worktree, not merged back to main:\n" + probe.stderr,
        )


if __name__ == "__main__":
    unittest.main(verbosity=2)