宜博 / lemongraph

"""Benchmark acceptance script for "Skill 详情页展示真实内容
(`v_skill_audit_log` 契约收口)" -- mechanically checks the scenarios from
Benchmark §1-§6 against src/skill_query.py::get_skill_content(),
src/backend_server.py's skill routes, frontend/app.js::mountSkillAuditLog,
and the git delivery state of the fix.

This is an acceptance script, not a permanent addition to the existing
src/test_skill_query.py / src/test_backend_server.py suites -- it is
expected to FAIL (via AssertionError or AttributeError on names like
skill_query.SkillPathTraversalError that don't exist yet) against the
pre-fix codebase, and to fully PASS once the fix described in the Benchmark
is implemented and committed.

Run (from repo root, with this file placed at src/test_skill_content_benchmark.py,
same `python3 -m unittest src.<module>` convention every other src/test_*.py file
in this repo uses):
    python3 -m unittest src.test_skill_content_benchmark -v

T5.2 (the full `discover -s src` regression run) is expensive (560+ tests)
and is gated behind RUN_FULL_REGRESSION=1, mirroring this repo's own
RUN_REAL_AGENT_E2E precedent (src/test_support.py:skip_unless_env) for
opt-in expensive checks:
    RUN_FULL_REGRESSION=1 python3 -m unittest src.test_skill_content_benchmark.RegressionSuiteTest.test_t5_2_full_src_suite_green -v
"""

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

from src import schema, skill_query, skill_update, backend_server

_THIS_DIR = os.path.dirname(os.path.abspath(__file__))
REPO_ROOT = os.path.dirname(_THIS_DIR)
APP_JS_PATH = os.path.join(REPO_ROOT, "frontend", "app.js")

SAMPLE_CONTENT = "# skill-1\n\nThis is the real migrated SKILL.md content for skill-1.\n"


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


def _write_file(path, content):
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        f.write(content)


def _seed_skill_fixtures(conn, repo_dir):
    """The four (+1) fixture skills described in the Benchmark's own
    "以下测试夹具" preamble: skill-1 (normal), skill-missing-file (DB row,
    no file on disk), skill-evil (whole-path traversal), skill-evil-2 (a
    second, differently-constructed traversal -- T1.5 requires two distinct
    construction styles), skill-with-support (main file + supporting
    files)."""
    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"),
    )
    _write_file(os.path.join(repo_dir, "skills", "skill-1", "SKILL.md"), SAMPLE_CONTENT)

    conn.execute(
        "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, "
        "migrated_from_profile, status) VALUES (?,?,?,?,?,?,?)",
        ("skill-missing-file", "coder", "skills/skill-missing-file", "unknown", "new", None, "active"),
    )
    # deliberately no file/directory written on disk for skill-missing-file

    conn.execute(
        "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, "
        "migrated_from_profile, status) VALUES (?,?,?,?,?,?,?)",
        ("skill-evil", "coder", "../../etc/passwd", "unknown", "new", None, "active"),
    )

    conn.execute(
        "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, "
        "migrated_from_profile, status) VALUES (?,?,?,?,?,?,?)",
        ("skill-evil-2", "coder", "skills/../../../tmp/evil", "unknown", "new", None, "active"),
    )

    conn.execute(
        "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, "
        "migrated_from_profile, status) VALUES (?,?,?,?,?,?,?)",
        ("skill-with-support", "coder", "skills/skill-with-support", "unknown", "new", None, "active"),
    )
    _write_file(os.path.join(repo_dir, "skills", "skill-with-support", "SKILL.md"), "support demo\n")
    _write_file(os.path.join(repo_dir, "skills", "skill-with-support", "references", "foo.md"), "foo\n")
    _write_file(os.path.join(repo_dir, "skills", "skill-with-support", "scripts", "bar.py"), "# bar\n")

    conn.commit()


# ---------------------------------------------------------------------------
# Section 1 (Benchmark §1): src/skill_query.py::get_skill_content() unit
# layer -- T1.1-T1.7 + T4.1 (sentinel non-leakage).
# ---------------------------------------------------------------------------

class GetSkillContentUnitTest(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()
        _seed_skill_fixtures(self.conn, self.tmp.name)

    def tearDown(self):
        self.conn.close()

    def test_t1_1_normal_content_all_nine_fields(self):
        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",
            },
        )
        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_content_none_no_exception(self):
        result = skill_query.get_skill_content(self.conn, "skill-missing-file", repo_path=self.tmp.name)
        self.assertIsNone(result["content"])
        self.assertIsInstance(result["supporting_files"], list)
        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_whole_path_traversal_rejected_before_any_open(self):
        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)
            self.assertEqual(mock_open.call_count, 0, "path-traversal must be rejected before open() is ever called")

    def test_t1_5_dotdot_segment_traversal_rejected_before_any_open(self):
        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)
            self.assertEqual(mock_open.call_count, 0, "path-traversal must be rejected before open() is ever called")

    def test_t1_6_supporting_files_sorted_relative_excludes_main_file(self):
        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_t4_1_sentinel_content_never_leaks_on_traversal(self):
        sentinel_dir = tempfile.mkdtemp()
        self.addCleanup(shutil.rmtree, sentinel_dir, ignore_errors=True)
        sentinel_path = os.path.join(sentinel_dir, "secret.txt")
        with open(sentinel_path, "w", encoding="utf-8") as f:
            f.write("SENTINEL_MARKER_xyz")

        rel = os.path.relpath(sentinel_path, os.path.join(self.tmp.name, "skills"))
        self.conn.execute("UPDATE skill SET file_path = ? WHERE skill_id = 'skill-evil'", (rel,))
        self.conn.commit()

        try:
            result = skill_query.get_skill_content(self.conn, "skill-evil", repo_path=self.tmp.name)
        except skill_query.SkillPathTraversalError as e:
            self.assertNotIn(
                "SENTINEL_MARKER_xyz", str(e),
                "sentinel content must never appear in the raised exception's message",
            )
        else:
            self.assertNotIn(
                "SENTINEL_MARKER_xyz", json.dumps(result, default=str),
                "sentinel content leaked into a non-exception return value -- traversal was not actually rejected",
            )


# ---------------------------------------------------------------------------
# Section 2 (Benchmark §2): src/backend_server.py HTTP 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)
        _seed_skill_fixtures(self.server.conn, self.tmp.name)

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

    def _get(self, path):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("GET", path)
        resp = conn.getresponse()
        raw = resp.read()
        conn.close()
        try:
            body = json.loads(raw)
        except (json.JSONDecodeError, UnicodeDecodeError):
            body = raw
        return resp.status, body

    def _post(self, path, payload):
        host, port = self.server.server_address
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("POST", path, body=json.dumps(payload).encode("utf-8"))
        resp = conn.getresponse()
        body = json.loads(resp.read())
        conn.close()
        return resp.status, body

    def test_t2_1_normal_detail_response(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_returns_404_not_found(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_returns_200_content_null(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_returns_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 format, not the detail object")

        status, body = self._get("/api/skills/diff")
        self.assertEqual(status, 400)
        self.assertEqual(
            body.get("error"), "INVALID_INPUT",
            "must hit _handle_skill_diff's own validation (INVALID_INPUT), not be misrouted as skill_id='diff' (which would 404 NOT_FOUND)",
        )

    def test_t2_6_rollback_regression_unaffected(self):
        skill_update._write_and_commit(self.tmp.name, "skills/skill-1/SKILL.md", SAMPLE_CONTENT, "seed skill")
        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()
        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._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_route_removed(self):
        # real file present on disk -- under the OLD (pre-fix) /content route
        # this would come back 200 with the real content; the fix must not
        # keep that route alive as a compatibility alias.
        skill_update._write_and_commit(self.tmp.name, "skills/skill-1/SKILL.md", SAMPLE_CONTENT, "seed skill")
        status, _body = self._get("/api/skills/skill-1/content")
        self.assertEqual(status, 404, "the old GET /api/skills/:id/content route must be fully removed (falls through to the generic 404)")


# ---------------------------------------------------------------------------
# Section 3 (Benchmark §3): frontend/app.js::mountSkillAuditLog -- T3.1-T3.4,
# via a bare Node vm.Context sandbox in the style of
# frontend/test_graph_console_smoke.js (this repo's only JS test precedent;
# no npm/DOM-testing framework exists here by design).
# ---------------------------------------------------------------------------

def _build_frontend_smoke_js():
    return textwrap.dedent(r"""
    "use strict";
    const fs = require("fs");
    const vm = require("vm");
    const assert = require("assert");

    const APP_JS_PATH = process.argv[2];
    const SAMPLE_CONTENT = "# skill-1\n\nThis is the real migrated SKILL.md content for skill-1.\n";

    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 makeElement() {
      return { innerHTML: "", textContent: "", style: { display: "" } };
    }

    function mockResp(status, body) {
      return Promise.resolve({ status: status, json: async () => body });
    }

    function sleep(ms) {
      return new Promise((resolve) => setTimeout(resolve, ms));
    }

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

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

    async function run() {
      const elements = {
        skill_content: makeElement(),
        skill_audit_list: makeElement(),
        skill_audit_error: makeElement(),
      };
      const sandbox = {
        document: {
          createElement: () => makeFakeDiv(),
          getElementById: (id) => elements[id] || null,
          querySelectorAll: () => [],
          addEventListener: () => {},
        },
        window: { location: { protocol: "http:", host: "localhost", hash: "" }, addEventListener: () => {} },
        localStorage: { getItem: () => null, setItem: () => {} },
        setTimeout: setTimeout,
        clearTimeout: clearTimeout,
        console: console,
        fetch: null,
      };
      vm.createContext(sandbox);
      const src = fs.readFileSync(APP_JS_PATH, "utf8");
      new vm.Script(src, { filename: "app.js" }).runInContext(sandbox);

      function resetElements() {
        elements.skill_content.innerHTML = "";
        elements.skill_audit_list.innerHTML = "";
        elements.skill_audit_error.innerHTML = "";
        elements.skill_audit_error.textContent = "";
        elements.skill_audit_error.style.display = "";
      }

      // ---- T3.1: normal content rendering ----
      resetElements();
      sandbox.fetch = (url) => {
        if (url === "/api/skills/skill-1") {
          return mockResp(200, {
            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, supporting_files: [],
          });
        }
        if (url === "/api/skills/skill-1/audit") return mockResp(200, []);
        throw new Error("unexpected fetch url: " + url);
      };
      sandbox.mountSkillAuditLog("skill-1");
      await sleep(50);
      assert.strictEqual(preTextContent(elements.skill_content.innerHTML), SAMPLE_CONTENT, "T3.1: pre textContent must equal the real file content exactly");
      assert.ok(elements.skill_content.innerHTML.includes('class="hint"'), "T3.1: a hint element must be present");
      assert.ok(elements.skill_content.innerHTML.includes("无支持文件"), "T3.1: empty supporting_files must render the '无支持文件' hint text");
      console.log("T3.1 OK");

      // ---- T3.2: content null ----
      resetElements();
      sandbox.fetch = (url) => {
        if (url === "/api/skills/skill-2") {
          return mockResp(200, {
            skill_id: "skill-2", owner_org_node: "coder", file_path: "skills/skill-2",
            current_git_commit_sha: "unknown", source: "new",
            migrated_from_profile: null, status: "active",
            content: null, supporting_files: [],
          });
        }
        if (url === "/api/skills/skill-2/audit") return mockResp(200, []);
        throw new Error("unexpected fetch url: " + url);
      };
      sandbox.mountSkillAuditLog("skill-2");
      await sleep(50);
      assert.strictEqual(preTextContent(elements.skill_content.innerHTML), "（内容文件不可用）", "T3.2: null content must render the placeholder text verbatim, with no thrown exception");
      console.log("T3.2 OK");

      // ---- T3.3: skill not found -- shared error banner, both panels cleared ----
      resetElements();
      sandbox.fetch = (url) => {
        if (url === "/api/skills/skill-x") return mockResp(404, { error: "NOT_FOUND" });
        if (url === "/api/skills/skill-x/audit") return mockResp(404, { error: "NOT_FOUND" });
        throw new Error("unexpected fetch url: " + url);
      };
      sandbox.mountSkillAuditLog("skill-x");
      await sleep(50);
      assert.strictEqual(elements.skill_content.innerHTML, "", "T3.3: content panel must be cleared (empty string) when the skill doesn't exist");
      assert.strictEqual(elements.skill_audit_list.innerHTML, "", "T3.3: audit panel must be cleared (empty string) when the skill doesn't exist");
      assert.strictEqual(elements.skill_audit_error.style.display, "block", "T3.3: shared error banner must be visible");
      assert.ok(elements.skill_audit_error.textContent && elements.skill_audit_error.textContent.length > 0, "T3.3: error banner must carry non-empty text");
      console.log("T3.3 OK");

      // ---- T3.4: supporting files listed, in order, no old field-name reliance ----
      resetElements();
      sandbox.fetch = (url) => {
        if (url === "/api/skills/skill-3") {
          return mockResp(200, {
            skill_id: "skill-3", owner_org_node: "coder", file_path: "skills/skill-3",
            current_git_commit_sha: "unknown", source: "new",
            migrated_from_profile: null, status: "active",
            content: "content\n", supporting_files: ["references/foo.md", "scripts/bar.py"],
          });
        }
        if (url === "/api/skills/skill-3/audit") return mockResp(200, []);
        throw new Error("unexpected fetch url: " + url);
      };
      sandbox.mountSkillAuditLog("skill-3");
      await sleep(50);
      assert.ok(elements.skill_content.innerHTML.includes("references/foo.md"), "T3.4: hint must list the first supporting file");
      assert.ok(elements.skill_content.innerHTML.includes("scripts/bar.py"), "T3.4: hint must list the second supporting file");
      console.log("T3.4 OK");

      console.log("ALL_FRONTEND_SMOKE_CHECKS_PASSED");
    }

    run().catch((err) => {
      console.error(String((err && err.stack) || err));
      process.exit(1);
    });
    """)


class FrontendSkillDetailSmokeTest(unittest.TestCase):
    def test_mount_skill_audit_log_smoke(self):
        if shutil.which("node") is None:
            self.skipTest("node is not on PATH -- cannot run the frontend vm-sandbox smoke test")
        fd, script_path = tempfile.mkstemp(suffix=".js")
        os.close(fd)
        with open(script_path, "w", encoding="utf-8") as f:
            f.write(_build_frontend_smoke_js())
        self.addCleanup(os.remove, script_path)

        proc = subprocess.run(
            ["node", script_path, APP_JS_PATH],
            capture_output=True, text=True, timeout=30,
        )
        self.assertEqual(
            proc.returncode, 0,
            "frontend/app.js::mountSkillAuditLog smoke test failed:\nstdout:\n" + proc.stdout + "\nstderr:\n" + proc.stderr,
        )
        self.assertIn("ALL_FRONTEND_SMOKE_CHECKS_PASSED", proc.stdout)


# ---------------------------------------------------------------------------
# T1.7 / T3.5: static structural scans -- old symbols/field names/routes
# must not remain anywhere in the codebase after the fix. Pure-Python
# recursive search (not a shelled-out `grep -P`) so this runs identically on
# BSD grep (macOS default) and GNU grep (Linux) dev machines.
# ---------------------------------------------------------------------------

_THIS_FILE_NAME = os.path.basename(os.path.abspath(__file__))


def _grep_py_tree(subdir, pattern):
    """Excludes this benchmark script's own file: it necessarily *names*
    SkillContentMissingError in its docstrings/messages for documentation
    purposes, which would otherwise make T1.7 permanently unsatisfiable even
    after every real production/test reference is gone."""
    regex = re.compile(pattern)
    matches = []
    root = os.path.join(REPO_ROOT, subdir)
    for dirpath, dirnames, filenames in os.walk(root):
        dirnames[:] = [d for d in dirnames if d != ".git"]
        for name in filenames:
            if not name.endswith(".py") or name == _THIS_FILE_NAME:
                continue
            full = os.path.join(dirpath, name)
            with open(full, "r", encoding="utf-8", errors="replace") as f:
                for lineno, line in enumerate(f, start=1):
                    if regex.search(line):
                        matches.append("%s:%d: %s" % (full, lineno, line.rstrip("\n")))
    return matches


def _grep_file(path, pattern):
    regex = re.compile(pattern)
    matches = []
    with open(path, "r", encoding="utf-8", errors="replace") as f:
        for lineno, line in enumerate(f, start=1):
            if regex.search(line):
                matches.append("%s:%d: %s" % (path, lineno, line.rstrip("\n")))
    return matches


class StaticContractScanTest(unittest.TestCase):
    def test_t1_7_skill_content_missing_error_removed_from_src(self):
        matches = _grep_py_tree("src", r"SkillContentMissingError")
        self.assertEqual(matches, [], "SkillContentMissingError must no longer be defined or referenced anywhere under src/:\n" + "\n".join(matches))

    def test_t3_5_no_legacy_support_files_field_name_in_app_js(self):
        matches = _grep_file(APP_JS_PATH, r"(?<!ing_)support_files\b")
        self.assertEqual(matches, [], "legacy 'support_files' field name must not remain in frontend/app.js (must be 'supporting_files'):\n" + "\n".join(matches))

    def test_t3_5_no_legacy_content_route_suffix_in_app_js(self):
        matches = _grep_file(APP_JS_PATH, r'/content"')
        self.assertEqual(matches, [], "frontend/app.js must no longer fetch the removed .../content route suffix:\n" + "\n".join(matches))


# ---------------------------------------------------------------------------
# Section 5 (Benchmark §5): full regression -- T5.1 (targeted pre-existing
# tests, run every time) / T5.2 (full `discover -s src`, gated -- expensive).
# ---------------------------------------------------------------------------

class RegressionSuiteTest(unittest.TestCase):
    _T5_1_COMMANDS = [
        ["src.test_skill_query.ListSkillsTest"],
        ["src.test_skill_query.GetSkillAuditLogTest"],
        ["src.test_skill_query.GetDiffTest"],
        ["src.test_backend_server.SkillRouteTest.test_list_returns_seeded_skill"],
        ["src.test_backend_server.SkillRouteTest", "-k", "audit_log"],
        ["src.test_backend_server.SkillRouteTest", "-k", "diff"],
        ["src.test_backend_server.SkillRouteTest", "-k", "rollback"],
        ["src.test_backend_server.SkillRouteTest.test_full_update_diff_rollback_cycle"],
    ]

    def test_t5_1_existing_skill_tests_still_pass(self):
        for extra_args in self._T5_1_COMMANDS:
            with self.subTest(args=extra_args):
                proc = subprocess.run(
                    [sys.executable, "-m", "unittest"] + extra_args,
                    cwd=REPO_ROOT, capture_output=True, text=True, timeout=120,
                )
                self.assertEqual(
                    proc.returncode, 0,
                    "regression command failed: unittest " + " ".join(extra_args) + "\n" + proc.stdout + proc.stderr,
                )

    @unittest.skipUnless(
        os.environ.get("RUN_FULL_REGRESSION") == "1",
        "full `discover -s src` (560+ tests) is expensive -- set RUN_FULL_REGRESSION=1 to actually run T5.2",
    )
    def test_t5_2_full_src_suite_green(self):
        proc = subprocess.run(
            [sys.executable, "-m", "unittest", "discover", "-s", "src"],
            cwd=REPO_ROOT, capture_output=True, text=True, timeout=1800,
        )
        tail = (proc.stdout[-4000:] + proc.stderr[-4000:])
        self.assertEqual(proc.returncode, 0, "python3 -m unittest discover -s src failed:\n" + tail)


# ---------------------------------------------------------------------------
# Section 6 (Benchmark §6): delivery must be committed, not left as
# uncommitted worktree diff -- T6.1/T6.2. (T6.3's "re-run everything from the
# main checkout, not just a worktree" step is inherently a manual re-run of
# this same script from a different cwd -- REPO_ROOT here is always whatever
# checkout actually invoked the script.)
# ---------------------------------------------------------------------------

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",
    ]
    _CORE_FILES = ["src/skill_query.py", "src/backend_server.py", "frontend/app.js"]
    _INVESTIGATION_DATE = "2026-08-13"

    def test_t6_1_no_uncommitted_changes_in_target_files(self):
        proc = subprocess.run(
            ["git", "status", "--porcelain", "--"] + self._TARGET_FILES,
            cwd=REPO_ROOT, capture_output=True, text=True, check=True,
        )
        self.assertEqual(
            proc.stdout.strip(), "",
            "these files must be fully committed, not left as uncommitted working-tree changes "
            "(running tests green is not enough -- see Benchmark T6.1):\n" + proc.stdout,
        )

    def test_t6_2_recent_commits_touch_core_contract_files(self):
        newest_shas = []
        for path in self._CORE_FILES:
            proc = subprocess.run(
                ["git", "log", "--format=%H %ci", "-1", "--", path],
                cwd=REPO_ROOT, capture_output=True, text=True, check=True,
            )
            line = proc.stdout.strip()
            self.assertTrue(line, path + " has no commit history at all")
            sha, _, date_part = line.partition(" ")
            self.assertGreater(
                date_part[:10], self._INVESTIGATION_DATE,
                path + "'s latest commit (" + date_part[:10] + ") must be newer than the task's investigation date -- looks like no new commit landed",
            )
            newest_shas.append(sha)

        found_marker = False
        for sha in newest_shas:
            proc = subprocess.run(
                ["git", "show", sha], cwd=REPO_ROOT, capture_output=True, text=True, check=True,
            )
            if "SkillPathTraversalError" in proc.stdout or "supporting_files" in proc.stdout:
                found_marker = True
                break
        self.assertTrue(
            found_marker,
            "none of the latest commits touching skill_query.py/backend_server.py/app.js contain the "
            "new contract markers (SkillPathTraversalError / supporting_files) -- looks like an unrelated "
            "commit merely happened to touch these files, not the real fix",
        )


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