#!/usr/bin/env python3
"""判定ファイルと判定の画面のテスト (依頼 183). KIOXIA にも API にも触らない (一時フォルダだけ).

  python3 pi05/verdict_viewer/test_verdicts.py
"""
import json
import os
import sys
import tempfile
import threading
import unittest
import urllib.error
import urllib.request
from pathlib import Path
from unittest import mock

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))
import verdicts as V  # noqa: E402
import seed_verdicts as S  # noqa: E402
import server  # noqa: E402


def row(stem, outcome, labeler, **kw):
    r = {"run_stem": stem, "pick": "unknown", "place": "unknown", "outcome": outcome, "labeler": labeler}
    r.update(kw)
    return r


def make_runs(root):
    """偽の記録: a = json に success, b = json の outcome null, c = frames と log.jsonl だけ, d = 何もない."""
    root = Path(root)
    for stem in ("pi05_20261001_000001", "pi05_20261001_000002", "pi05_20261001_000003", "pi05_20261001_000004"):
        (root / stem).mkdir(parents=True)
    a = root / "pi05_20261001_000001"
    (a / f"{a.name}.json").write_text(json.dumps({"outcome": "success", "ended_at": "2026-10-01T00:00:10",
        "judgments": {"pick": {"verdict": "yes"}, "place": {"verdict": "yes"}}}))
    b = root / "pi05_20261001_000002"
    (b / f"{b.name}.json").write_text(json.dumps({"outcome": None}))
    c = root / "pi05_20261001_000003"
    (c / "frames").mkdir()
    for i in range(3):
        (c / "frames" / f"{i:04d}.jpg").write_bytes(b"\xff\xd8\xff\xd9")
    (c / "frames" / "._0000.jpg").write_bytes(b"x")
    (c / "log.jsonl").write_text("\n".join(json.dumps({"chunk": str(i), "t": str(100 + i), "executed": "True",
        "state": json.dumps([i, 1, 2, 3, 4, -30, 0 if i < 2 else 1])}) for i in range(3)) + "\n")
    return root


class TestAppend(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.csv = Path(self.tmp.name) / "v.csv"

    def tearDown(self):
        self.tmp.cleanup()

    def test_append_only_keeps_existing_bytes(self):
        V.append_rows([row("pi05_1", "success", "run_input"), row("pi05_2", "failure", "ai")], self.csv)
        before = self.csv.read_bytes()
        V.append_rows([row("pi05_1", "failure", "human_only")], self.csv)
        after = self.csv.read_bytes()
        self.assertTrue(after.startswith(before), "既存の行が変わった")
        self.assertEqual(after.count(b"labeled_at,"), 1, "見出しが 2 回書かれた")
        self.assertEqual(len(V.read_rows(self.csv)), 3)

    def test_header_columns(self):
        V.append_rows([row("pi05_1", "success", "ai")], self.csv)
        self.assertEqual(self.csv.read_text().splitlines()[0].split(","), V.COLUMNS)

    def test_latest_row_wins(self):
        V.append_rows([row("pi05_1", "success", "ai"), row("pi05_2", "failure", "ai")], self.csv)
        V.append_rows([row("pi05_1", "failure", "human_fixed_ai")], self.csv)
        V.append_rows([row("pi05_1", "unknown", "human_only")], self.csv)
        lat = V.latest_by_stem(V.read_rows(self.csv))
        self.assertEqual(lat["pi05_1"]["outcome"], "unknown")
        self.assertEqual(lat["pi05_2"]["outcome"], "failure")
        self.assertEqual(V.latest_by_stem(V.read_rows(self.csv), labelers=("ai",))["pi05_1"]["outcome"], "success")

    def test_bad_values_write_nothing(self):
        V.append_rows([row("pi05_1", "success", "ai")], self.csv)
        before = self.csv.read_bytes()
        for bad in (row("pi05_1", "great", "ai"), row("pi05_1", "success", "robot"),
                    row("", "success", "ai"), row("pi05_1", "success", "ai", pick="maybe")):
            with self.assertRaises(ValueError):
                V.append_rows([row("pi05_9", "success", "ai"), bad], self.csv)
        self.assertEqual(self.csv.read_bytes(), before, "まちがった行のときに途中まで書いた")

    def test_place_not_applicable_only_for_place(self):
        # 依頼 193: place は not_applicable (そこまで行かなかった) を書ける. pick には書けない
        V.append_rows([row("pi05_1", "failure", "human_only", pick="no", place="not_applicable")], self.csv)
        self.assertEqual(V.read_rows(self.csv)[-1]["place"], "not_applicable")
        before = self.csv.read_bytes()
        with self.assertRaises(ValueError):
            V.append_rows([row("pi05_2", "failure", "human_only", pick="not_applicable", place="not_applicable")], self.csv)
        self.assertEqual(self.csv.read_bytes(), before)

    def test_comma_and_newline_in_reason(self):
        V.append_rows([row("pi05_1", "failure", "ai", ai_reason='箱の外, "落とした"\n2 行目')], self.csv)
        self.assertEqual(V.read_rows(self.csv)[0]["ai_reason"], '箱の外, "落とした"\n2 行目')

    def test_broken_file_stops(self):
        V.append_rows([row("pi05_1", "success", "ai")], self.csv)
        with open(self.csv, "ab") as f:
            f.write(b"2026,pi05_2,yes")  # 書きかけ (改行なし)
        before = self.csv.read_bytes()
        with self.assertRaises(V.BrokenCsv):
            V.append_rows([row("pi05_3", "success", "ai")], self.csv)
        self.assertEqual(self.csv.read_bytes(), before)
        other = Path(self.tmp.name) / "other.csv"
        other.write_text("a,b\n1,2\n")
        with self.assertRaises(V.BrokenCsv):
            V.append_rows([row("pi05_3", "success", "ai")], other)
        self.assertEqual(other.read_text(), "a,b\n1,2\n")

    def test_unclosed_quote_stops(self):
        V.append_rows([row("pi05_1", "success", "ai")], self.csv)
        with open(self.csv, "ab") as f:
            f.write(b'2026,pi05_2,yes,yes,success,ai,,,,"unfinished\n')
        before = self.csv.read_bytes()
        with self.assertRaises(V.BrokenCsv):
            V.append_rows([row("pi05_3", "success", "ai")], self.csv)
        self.assertEqual(self.csv.read_bytes(), before)

    def test_long_note_is_cut(self):
        V.append_rows([row("pi05_1", "success", "human_only", note="あ" * 200000, ai_reason="い" * 200000)], self.csv)
        r = V.read_rows(self.csv)[0]
        self.assertEqual(len(r["note"]), V.MAX_NOTE + 1)
        self.assertEqual(len(r["ai_reason"]), V.MAX_REASON + 1)
        V.append_rows([row("pi05_2", "success", "ai")], self.csv)

    def test_forbid_dirs(self):
        d = Path(self.tmp.name) / "myruns"
        d.mkdir()
        with self.assertRaises(V.BrokenCsv):
            V.append_rows([row("pi05_1", "success", "ai")], d / "v.csv", forbid_dirs=(d,))
        self.assertFalse((d / "v.csv").exists())

    def test_refuses_symlink_and_record_dirs(self):
        real = Path(self.tmp.name) / "real.csv"
        V.append_rows([row("pi05_1", "success", "ai")], real)
        link = Path(self.tmp.name) / "link.csv"
        link.symlink_to(real)
        with self.assertRaises(V.BrokenCsv):
            V.append_rows([row("pi05_2", "success", "ai")], link)
        runs = Path(self.tmp.name) / "runs"
        (runs / "pi05_1").mkdir(parents=True)
        with mock.patch.object(V, "RUNS_DIR", runs):
            with self.assertRaises(V.BrokenCsv):
                V.append_rows([row("pi05_2", "success", "ai")], runs / "pi05_1" / "x.csv")
        with self.assertRaises(V.BrokenCsv):
            V.append_rows([row("pi05_2", "success", "ai")], Path(self.tmp.name) / "x.json")
        self.assertEqual(len(V.read_rows(real)), 1)

    def test_skip_if_labeled_by(self):
        V.append_rows([row("pi05_1", "failure", "human_only")], self.csv)
        out = V.append_rows([row("pi05_1", "success", "ai"), row("pi05_2", "success", "ai")], self.csv,
                            skip_if_labeled_by=V.HUMAN_LABELERS)
        self.assertEqual([r["run_stem"] for r in out], ["pi05_2"])
        self.assertEqual(V.latest_by_stem(V.read_rows(self.csv))["pi05_1"]["labeler"], "human_only")


class TestMount(unittest.TestCase):
    def test_not_mounted_stops_and_writes_nothing(self):
        target = V.KIOXIA_MOUNT / "r2r2r-data" / "runs" / "pi05_policy_verdicts_v1_TEST_never.csv"
        with mock.patch.object(V.os.path, "ismount", lambda p: False), \
                mock.patch("builtins.open", side_effect=AssertionError("open してはいけない")):
            with self.assertRaises(V.NotMounted):
                V.append_rows([row("pi05_1", "success", "ai")], target)

    def test_mounted_check_only_for_kioxia(self):
        with mock.patch.object(V.os.path, "ismount", lambda p: False):
            V.ensure_ssd_mounted("/tmp/whatever.csv")  # KIOXIA の外は確かめない (テスト用)

    def test_seed_aborts_without_mount(self):
        with mock.patch.object(V.os.path, "ismount", lambda p: False), \
                mock.patch.object(sys, "argv", ["seed", "--run-input"]):
            self.assertEqual(S.main(), 2)

    def test_server_returns_503_without_mount(self):
        with tempfile.TemporaryDirectory() as d:
            runs = make_runs(Path(d) / "runs")
            app = server.App(runs, V.KIOXIA_MOUNT / "r2r2r-data" / "runs" / "never_TEST.csv")
            with mock.patch.object(V.os.path, "ismount", lambda p: False):
                with self.assertRaises(V.NotMounted):
                    app.add_verdict({"run_stem": "pi05_20261001_000003", "outcome": "success"})


class TestLabeler(unittest.TestCase):
    def test_labeler(self):
        self.assertEqual(V.human_labeler_for("success", None), "human_only")
        self.assertEqual(V.human_labeler_for("success", {"outcome": "success"}), "human")
        self.assertEqual(V.human_labeler_for("failure", {"outcome": "success"}), "human_fixed_ai")


class TestSeed(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.runs = make_runs(Path(self.tmp.name) / "runs")
        self.csv = Path(self.tmp.name) / "v.csv"

    def tearDown(self):
        self.tmp.cleanup()

    def test_run_input_copy_once(self):
        c = S.do_run_input(self.csv, self.runs, dry_run=False)
        self.assertEqual(c["success"], 1)
        rows = V.read_rows(self.csv)
        self.assertEqual([(r["run_stem"], r["outcome"], r["labeler"], r["pick"]) for r in rows],
                         [("pi05_20261001_000001", "success", "run_input", "yes")])
        S.do_run_input(self.csv, self.runs, dry_run=False)
        self.assertEqual(len(V.read_rows(self.csv)), 1, "2 回流したら行が増えた")

    def test_ai_targets_and_not_judgeable(self):
        calls = []

        def fake_api(key, imgs, evidence):
            calls.append((len(imgs), evidence))
            return {"pick": "yes", "place": "no", "outcome": "failure", "reason": "箱の外"}, ""
        S.do_run_input(self.csv, self.runs, dry_run=False)
        with mock.patch.object(S, "api_key", lambda: "k"), mock.patch.object(S, "call_api", fake_api):
            c = S.do_ai(self.csv, self.runs, dry_run=False, limit=0, only=None)
        self.assertEqual(calls, [(3, "frames")])  # c だけ API を呼ぶ (frames 3 枚・._ は除く)
        self.assertEqual(c["failure"], 1)
        self.assertEqual(c["not_judgeable"], 2)  # b (null・素材なし) と d (何もない)
        ai = V.latest_by_stem(V.read_rows(self.csv), labelers=("ai",))
        self.assertNotIn("pi05_20261001_000001", ai)
        self.assertEqual(ai["pi05_20261001_000003"]["ai_reason"], "[送ったチャンク 3] 箱の外")
        n = len(V.read_rows(self.csv))
        with mock.patch.object(S, "api_key", lambda: "k"), mock.patch.object(S, "call_api", fake_api):
            S.do_ai(self.csv, self.runs, dry_run=False, limit=0, only=None)
        self.assertEqual(len(V.read_rows(self.csv)), n, "AI の行が 2 回書かれた")

    def test_ai_skips_when_api_unavailable(self):
        def boom(*a):
            raise S.ApiUnavailable("http_401")
        with mock.patch.object(S, "api_key", lambda: "k"), mock.patch.object(S, "call_api", boom):
            c = S.do_ai(self.csv, self.runs, dry_run=False, limit=0, only=None)
        self.assertEqual(c["skipped"], 1)
        self.assertNotIn("pi05_20261001_000003", V.latest_by_stem(V.read_rows(self.csv)))

    def test_seed_never_adds_after_human(self):
        V.append_rows([row("pi05_20261001_000003", "success", "human_only"),
                       row("pi05_20261001_000001", "failure", "human_only")], self.csv)
        with mock.patch.object(S, "api_key", lambda: "k"), \
                mock.patch.object(S, "call_api", lambda *a: ({"outcome": "failure"}, "")):
            S.do_ai(self.csv, self.runs, dry_run=False, limit=0, only=None)
        S.do_run_input(self.csv, self.runs, dry_run=False)
        lat = V.latest_by_stem(V.read_rows(self.csv))
        self.assertEqual(lat["pi05_20261001_000003"]["labeler"], "human_only")
        self.assertEqual(lat["pi05_20261001_000001"]["labeler"], "human_only")

    def test_seed_rechecks_inside_lock(self):
        # API を待つ間に人が確定した → AI の行は足さない
        def api_then_human(*a):
            V.append_rows([row("pi05_20261001_000003", "failure", "human_only")], self.csv)
            return {"outcome": "success"}, ""
        with mock.patch.object(S, "api_key", lambda: "k"), mock.patch.object(S, "call_api", api_then_human):
            c = S.do_ai(self.csv, self.runs, dry_run=False, limit=0, only={"pi05_20261001_000003"})
        self.assertEqual(c["skipped_human"], 1)
        self.assertEqual(V.latest_by_stem(V.read_rows(self.csv))["pi05_20261001_000003"]["labeler"], "human_only")

    def test_symlinked_side_files_are_ignored(self):
        outside = Path(self.tmp.name) / "outside.jsonl"
        outside.write_text(json.dumps({"chunk": "0", "executed": "True"}) + "\n")
        rd = self.runs / "pi05_20261001_000004"
        (rd / "log.jsonl").symlink_to(outside)
        (rd / f"{rd.name}.json").symlink_to(outside)
        self.assertEqual(V.read_log(rd), [])
        self.assertIsNone(V.read_run_json(rd))

    def test_server_refuses_csv_inside_runs_dir(self):
        app = server.App(self.runs, self.runs / "pi05_20261001_000003" / "v.csv")
        with self.assertRaises(V.BrokenCsv):
            app.add_verdict({"run_stem": "pi05_20261001_000003", "outcome": "success"})
        self.assertFalse((self.runs / "pi05_20261001_000003" / "v.csv").exists())

    def test_symlinked_run_and_frames_are_ignored(self):
        outside = Path(self.tmp.name) / "outside"
        (outside / "frames").mkdir(parents=True)
        (outside / "frames" / "0000.jpg").write_bytes(b"secret")
        (self.runs / "pi05_20261001_000099").symlink_to(outside)
        (self.runs / "pi05_20261001_000003" / "frames" / "0009.jpg").symlink_to(outside / "frames" / "0000.jpg")
        self.assertNotIn("pi05_20261001_000099", V.list_run_stems(self.runs))
        self.assertNotIn("0009.jpg", V.frame_files(self.runs / "pi05_20261001_000003"))

    def test_api_key_export_line(self):
        with tempfile.TemporaryDirectory() as d:
            p = Path(d) / ".anthropic_key"
            p.write_text("export ANTHROPIC_API_KEY='sk-test'\n")
            with mock.patch.dict(os.environ, {"HOME": d}, clear=False), \
                    mock.patch.dict(os.environ, {}, clear=False) as env:
                env.pop("ANTHROPIC_API_KEY", None)
                self.assertEqual(S.api_key(), "sk-test")


class TestServer(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.runs = make_runs(Path(self.tmp.name) / "runs")
        self.csv = Path(self.tmp.name) / "v.csv"
        V.append_rows([row("pi05_20261001_000003", "success", "ai", ai_reason="入っている", evidence="frames")], self.csv)
        self.app = server.App(self.runs, self.csv)
        self.srv = server.ThreadingHTTPServer(("127.0.0.1", 0), server.make_handler(self.app))
        threading.Thread(target=self.srv.serve_forever, daemon=True).start()
        self.base = f"http://127.0.0.1:{self.srv.server_address[1]}"

    def tearDown(self):
        self.srv.shutdown()
        self.srv.server_close()
        self.tmp.cleanup()

    def get(self, path):
        with urllib.request.urlopen(self.base + path) as r:
            return json.loads(r.read())

    def post(self, body):
        req = urllib.request.Request(self.base + "/api/verdict", data=json.dumps(body).encode(),
                                     headers={"Content-Type": "application/json"}, method="POST")
        try:
            with urllib.request.urlopen(req) as r:
                return r.status, json.loads(r.read())
        except urllib.error.HTTPError as e:
            return e.code, json.loads(e.read())

    def test_list_and_detail(self):
        runs = {r["run_stem"]: r for r in self.get("/api/runs")["runs"]}
        self.assertEqual(len(runs), 4)
        self.assertEqual(runs["pi05_20261001_000003"]["ai_outcome"], "success")
        self.assertFalse(runs["pi05_20261001_000003"]["confirmed"])
        d = self.get("/api/run/pi05_20261001_000003")
        self.assertEqual(d["frames"], ["0000.jpg", "0001.jpg", "0002.jpg"])
        self.assertEqual(len(d["states"]["joints"]), 3)
        self.assertEqual(d["states"]["grip"][-1][1], 1.0)
        self.assertEqual(d["executed_chunks"], 3)

    def test_post_appends_and_labeler(self):
        before = self.csv.read_bytes()
        st, j = self.post({"run_stem": "pi05_20261001_000003", "outcome": "failure", "place": "no", "note": "箱の縁"})
        self.assertEqual(st, 200)
        self.assertTrue(self.csv.read_bytes().startswith(before))
        last = V.read_rows(self.csv)[-1]
        self.assertEqual((last["labeler"], last["outcome"], last["pick"], last["place"], last["ai_outcome"], last["ai_reason"],
                          last["evidence"], last["note"]),
                         ("human_fixed_ai", "failure", "unknown", "no", "success", "入っている", "frames", "箱の縁"))
        st, j = self.post({"run_stem": "pi05_20261001_000003", "outcome": "success"})
        self.assertEqual((j["row"]["labeler"], j["row"]["pick"], j["row"]["place"]), ("human", "yes", "yes"))
        st, j = self.post({"run_stem": "pi05_20261001_000004", "outcome": "not_judgeable"})
        self.assertEqual(j["row"]["labeler"], "human_only")
        runs = {r["run_stem"]: r for r in self.get("/api/runs")["runs"]}
        self.assertTrue(runs["pi05_20261001_000003"]["confirmed"])
        self.assertEqual(runs["pi05_20261001_000003"]["outcome"], "success")

    def test_post_place_not_applicable(self):
        # 依頼 193: 画面から place=not_applicable を送れる. pick=not_applicable は 400 で書かない
        st, j = self.post({"run_stem": "pi05_20261001_000003", "outcome": "failure", "pick": "no", "place": "not_applicable"})
        self.assertEqual(st, 200)
        self.assertEqual((j["row"]["pick"], j["row"]["place"]), ("no", "not_applicable"))
        before = self.csv.read_bytes()
        st, j = self.post({"run_stem": "pi05_20261001_000003", "outcome": "failure", "pick": "not_applicable"})
        self.assertEqual(st, 400)
        self.assertEqual(self.csv.read_bytes(), before)

    def test_same_op_id_writes_once(self):
        n = len(V.read_rows(self.csv))
        b = {"op_id": "abc", "run_stem": "pi05_20261001_000003", "outcome": "failure"}
        self.assertEqual(self.post(b)[0], 200)
        self.assertEqual(self.post(b)[0], 200)
        self.assertEqual(len(V.read_rows(self.csv)), n + 1)

    def test_media_link_to_other_run_is_refused(self):
        other = self.runs / "pi05_20261001_000004" / "frames"
        other.mkdir()
        (other / "0000.jpg").write_bytes(b"other")
        (self.runs / "pi05_20261001_000003" / "frames" / "0008.jpg").symlink_to(other / "0000.jpg")
        with self.assertRaises(urllib.error.HTTPError) as cm:
            urllib.request.urlopen(self.base + "/media/pi05_20261001_000003/frames/0008.jpg")
        self.assertEqual(cm.exception.code, 404)

    def test_post_rejects_bad(self):
        n = len(V.read_rows(self.csv))
        self.assertEqual(self.post({"run_stem": "pi05_20261001_000003", "outcome": "great"})[0], 400)
        self.assertEqual(self.post({"run_stem": "../etc", "outcome": "success"})[0], 400)
        self.assertEqual(self.post({"run_stem": "pi05_20990101_000000", "outcome": "success"})[0], 400)
        self.assertEqual(len(V.read_rows(self.csv)), n)

    def test_media_and_traversal(self):
        with urllib.request.urlopen(self.base + "/media/pi05_20261001_000003/frames/0001.jpg") as r:
            self.assertEqual(r.read(), b"\xff\xd8\xff\xd9")
        for p in ("/media/pi05_20261001_000003/frames/..%2Flog.jsonl", "/media/..%2F..%2Fetc/top.mp4"):
            with self.assertRaises(urllib.error.HTTPError) as cm:
                urllib.request.urlopen(self.base + p)
            self.assertEqual(cm.exception.code, 404)


class TestConfirm192(unittest.TestCase):
    """依頼 192: y/n/u/x は選ぶだけ・確定 (ボタンか Enter) で 1 行だけ書く. 画面を Chrome で動かす (playwright がなければ飛ばす)."""
    @classmethod
    def setUpClass(cls):
        try:
            from playwright.sync_api import sync_playwright
            cls.pw = sync_playwright().start()
            cls.browser = cls.pw.chromium.launch(channel="chrome")
        except Exception as e:  # noqa: BLE001
            raise unittest.SkipTest(f"playwright / Chrome がない: {e}")

    @classmethod
    def tearDownClass(cls):
        cls.browser.close()
        cls.pw.stop()

    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.runs = make_runs(Path(self.tmp.name) / "runs")
        self.csv = Path(self.tmp.name) / "v.csv"
        V.append_rows([row("pi05_20261001_000003", "success", "ai", ai_reason="入っている", evidence="frames")], self.csv)
        self.srv = server.ThreadingHTTPServer(("127.0.0.1", 0), server.make_handler(server.App(self.runs, self.csv)))
        threading.Thread(target=self.srv.serve_forever, daemon=True).start()
        self.pg = self.browser.new_page()
        self.pg.goto(f"http://127.0.0.1:{self.srv.server_address[1]}/")
        self.wait_loaded()

    def tearDown(self):
        self.pg.close()
        self.srv.shutdown()
        self.srv.server_close()
        self.tmp.cleanup()

    def wait_loaded(self):
        self.pg.wait_for_function("() => { const t = document.querySelector('#title').textContent;"
                                  " return t.startsWith('pi05_') && !t.includes('読み込み中'); }")

    def n_rows(self):
        return len(V.read_rows(self.csv))

    def on_keys(self):
        return self.pg.eval_on_selector_all("button.key.on", "e => e.map(x => x.dataset.oc)")

    def disabled(self):
        return self.pg.eval_on_selector("#confirm", "e => e.disabled")

    def test_choose_only_does_not_write(self):
        n = self.n_rows()
        self.assertTrue(self.disabled())
        self.pg.keyboard.press("n")
        self.assertEqual(self.on_keys(), ["failure"])
        self.assertFalse(self.disabled())
        self.pg.click('button.key[data-oc="success"]')  # 選び直し
        self.assertEqual(self.on_keys(), ["success"])
        self.pg.wait_for_timeout(600)
        self.assertEqual(self.n_rows(), n)

    def test_confirm_before_choose_does_nothing(self):
        n = self.n_rows()
        title = self.pg.text_content("#title")
        self.pg.keyboard.press("Enter")
        self.pg.click("#confirm", force=True)
        self.pg.wait_for_timeout(600)
        self.assertEqual(self.n_rows(), n)
        self.assertEqual(self.pg.text_content("#title"), title)

    def test_enter_writes_one_row(self):
        n = self.n_rows()
        stem = self.pg.text_content("#title")
        self.pg.keyboard.press("n")
        self.pg.keyboard.press("Enter")
        self.pg.keyboard.press("Enter")  # 0.4 秒の間の 2 回目は書かない
        self.pg.wait_for_function("() => document.querySelector('#toast').textContent.includes('確定しました')")
        self.assertIn(stem, self.pg.text_content("#toast"))
        self.pg.wait_for_timeout(800)
        rows = V.read_rows(self.csv)
        self.assertEqual(len(rows), n + 1)
        self.assertEqual((rows[-1]["run_stem"], rows[-1]["outcome"]), (stem, "failure"))

    def test_button_writes_one_row(self):
        n = self.n_rows()
        stem = self.pg.text_content("#title")
        self.pg.keyboard.press("y")
        self.pg.click("#confirm")
        self.pg.click("#confirm", force=True)
        self.pg.wait_for_timeout(1000)
        rows = V.read_rows(self.csv)
        self.assertEqual(len(rows), n + 1)
        self.assertEqual((rows[-1]["run_stem"], rows[-1]["outcome"]), (stem, "success"))

    def test_enter_in_note_confirms_but_keys_do_not_choose(self):
        n = self.n_rows()
        self.pg.keyboard.press("u")
        self.pg.click("#note")
        self.pg.keyboard.type("yn")
        self.assertEqual(self.on_keys(), ["unknown"])
        self.pg.keyboard.press("Enter")
        self.pg.wait_for_timeout(1000)
        rows = V.read_rows(self.csv)
        self.assertEqual(len(rows), n + 1)
        self.assertEqual((rows[-1]["outcome"], rows[-1]["note"]), ("unknown", "yn"))

    def test_moving_discards_choice(self):
        n = self.n_rows()
        stem = self.pg.text_content("#title")
        self.pg.keyboard.press("n")
        self.pg.keyboard.press("ArrowRight")
        self.wait_loaded()
        self.assertNotEqual(self.pg.text_content("#title"), stem)
        self.assertEqual(self.on_keys(), [])
        self.assertTrue(self.disabled())
        self.pg.keyboard.press("ArrowLeft")
        self.wait_loaded()
        self.assertEqual(self.pg.text_content("#title"), stem)
        self.assertEqual(self.on_keys(), [])
        self.pg.keyboard.press("Enter")
        self.pg.wait_for_timeout(600)
        self.assertEqual(self.n_rows(), n)

    # ---- 依頼 193: つかむ / 置くを日本語に・つかめなかったで置くは自動でそこまで行かなかった ----
    def pp_on(self):
        return self.pg.eval_on_selector_all("[data-pp].on", "e => e.map(x => x.dataset.pp + '=' + x.dataset.v)")

    def test_pp_labels_are_japanese(self):
        got = self.pg.eval_on_selector_all("[data-pp]", "e => e.map(x => x.dataset.pp + '=' + x.dataset.v + ':' + x.textContent)")
        self.assertEqual(got, ["pick=yes:つかめた", "pick=no:つかめなかった", "pick=unknown:分からない",
                               "place=yes:置けた", "place=no:置けなかった", "place=not_applicable:そこまで行かなかった",
                               "place=unknown:分からない"])
        self.assertEqual(self.pg.eval_on_selector_all("#panel .row .lbl", "e => e.map(x => x.textContent)"), ["pick", "place"])

    def test_pick_no_selects_place_not_applicable(self):
        n = self.n_rows()
        self.pg.click('[data-pp="pick"][data-v="no"]')
        self.assertEqual(self.pp_on(), ["pick=no", "place=not_applicable"])
        self.pg.click('[data-pp="place"][data-v="unknown"]')  # 選び直せる
        self.assertEqual(self.pp_on(), ["pick=no", "place=unknown"])
        self.pg.click('[data-pp="pick"][data-v="yes"]')  # つかめたに戻すと置くは選んでいない
        self.assertEqual(self.pp_on(), ["pick=yes"])
        self.pg.click('[data-pp="pick"][data-v="no"]')
        self.pg.wait_for_timeout(300)
        self.assertEqual(self.n_rows(), n)  # 選ぶだけでは書かない
        stem = self.pg.text_content("#title")
        self.pg.keyboard.press("n")
        self.pg.keyboard.press("Enter")
        self.pg.wait_for_timeout(1000)
        rows = V.read_rows(self.csv)
        self.assertEqual(len(rows), n + 1)
        self.assertEqual((rows[-1]["run_stem"], rows[-1]["pick"], rows[-1]["place"]), (stem, "no", "not_applicable"))


class TestRule185(unittest.TestCase):
    """エイイチの決め (2026-11-14 依頼 185): run_input も確定・送ったチャンク 0 → not_applicable・素材不足 → not_judgeable."""
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.runs = Path(self.tmp.name) / "runs"
        self.csv = Path(self.tmp.name) / "v.csv"
        # stem: (送ったチャンク, 先に入れる行)
        self.case = {
            "pi05_20261002_000001": (0, [row("pi05_20261002_000001", "not_judgeable", "ai", evidence="frames")]),  # 2 と 3 → 2
            "pi05_20261002_000002": (2, [row("pi05_20261002_000002", "not_judgeable", "ai", evidence="frames")]),  # 3
            "pi05_20261002_000003": (2, [row("pi05_20261002_000003", "not_judgeable", "ai", evidence="")]),        # 3 (素材なし)
            "pi05_20261002_000004": (0, [row("pi05_20261002_000004", "failure", "ai", evidence="top.mp4")]),       # 2
            "pi05_20261002_000005": (0, [row("pi05_20261002_000005", "not_judgeable", "ai", evidence="frames"),
                                         row("pi05_20261002_000005", "failure", "human_fixed_ai")]),             # 人の行 → 足さない
            "pi05_20261002_000006": (2, [row("pi05_20261002_000006", "not_judgeable", "ai", evidence=""),
                                         row("pi05_20261002_000006", "success", "human_only")]),                 # 人の行 → 足さない
            "pi05_20261002_000007": (0, [row("pi05_20261002_000007", "success", "run_input")]),                  # run_input → 足さない
            "pi05_20261002_000008": (2, [row("pi05_20261002_000008", "not_judgeable", "ai", evidence="top.mp4")]),  # 当たらない
            "pi05_20261002_000009": (3, [row("pi05_20261002_000009", "failure", "ai", evidence="frames")]),       # 当たらない
        }
        rows = []
        for stem, (n, rs) in self.case.items():
            d = self.runs / stem
            d.mkdir(parents=True)
            lines = [{"type": "model"}] + [{"chunk": str(i), "executed": "True"} for i in range(n)] + [{"chunk": "9", "executed": "False"}]
            (d / "log.jsonl").write_text("\n".join(json.dumps(x) for x in lines) + "\n")
            rows += rs
        V.append_rows(rows, self.csv)

    def tearDown(self):
        self.tmp.cleanup()

    def rule_rows(self):
        return {r["run_stem"]: r for r in V.read_rows(self.csv) if r["labeler"] == "rule"}

    def test_rule_rows(self):
        c = S.do_rule(self.csv, self.runs, dry_run=False)
        got = {k[-1]: v["outcome"] for k, v in self.rule_rows().items()}
        self.assertEqual(got, {"1": "not_applicable", "2": "not_judgeable", "3": "not_judgeable", "4": "not_applicable"})
        self.assertEqual((c["rule2"], c["rule3"], c["both"]), (2, 2, 1))
        for r in self.rule_rows().values():
            self.assertIn("エイイチの決め 2026-11-14 依頼185", r["note"])
        self.assertIn("決め2", self.rule_rows()["pi05_20261002_000001"]["note"])
        self.assertIn("決め3", self.rule_rows()["pi05_20261002_000002"]["note"])

    def test_rule_never_after_human_or_run_input(self):
        S.do_rule(self.csv, self.runs, dry_run=False)
        for stem in ("pi05_20261002_000005", "pi05_20261002_000006", "pi05_20261002_000007"):
            self.assertNotIn(stem, self.rule_rows())

    def test_rule_twice_adds_nothing(self):
        S.do_rule(self.csv, self.runs, dry_run=False)
        before = self.csv.read_bytes()
        S.do_rule(self.csv, self.runs, dry_run=False)
        self.assertEqual(self.csv.read_bytes(), before)

    def test_rule_rechecks_inside_lock(self):
        # 一覧を読んだあとに人の行が足されたら, rule の行は足さない
        real = V.read_rows
        def late_human(p):
            rows = real(p)
            V.append_rows([row("pi05_20261002_000002", "failure", "human")], self.csv)
            return rows
        with mock.patch.object(V, "read_rows", side_effect=late_human):
            S.do_rule(self.csv, self.runs, dry_run=False)
        self.assertNotIn("pi05_20261002_000002", self.rule_rows())

    def test_confirmed_and_success_rate(self):
        S.do_rule(self.csv, self.runs, dry_run=False)
        app = server.App(self.runs, self.csv)
        conf = {r["run_stem"][-1]: r["confirmed"] for r in app.runs_summary()}
        self.assertEqual(conf, {"1": True, "2": True, "3": True, "4": True, "5": True, "6": True, "7": True,
                                "8": False, "9": False})
        st = app.stats()
        # 確定: success 2 (6・7) / failure 1 (5). not_applicable 2・not_judgeable 2 は分母に入れない. AI の行も入れない
        self.assertEqual((st["success"], st["failure"], st["denominator"], st["confirmed"]), (2, 1, 3, 7))
        self.assertAlmostEqual(st["rate"], 2 / 3)
        self.assertEqual(st["by_outcome"]["not_applicable"], 2)

    def test_success_rate_denominator(self):
        rows = [row("a", "success", "human"), row("b", "failure", "run_input"), row("c", "unknown", "human"),
                row("d", "not_judgeable", "rule"), row("e", "not_applicable", "rule"), row("f", "success", "ai"),
                row("b", "success", "human_fixed_ai")]  # b は後ろの人の行が正
        st = V.success_rate(rows)
        self.assertEqual((st["success"], st["failure"], st["denominator"]), (2, 0, 2))
        self.assertEqual(st["rate"], 1.0)
        self.assertIsNone(V.success_rate([row("x", "not_applicable", "rule")])["rate"])

    def test_ai_after_confirmed_does_not_unconfirm(self):
        rows = [row("a", "not_applicable", "rule"), row("a", "failure", "ai")]
        self.assertEqual(V.current_by_stem(rows)["a"]["labeler"], "rule")

    def test_ai_cannot_write_not_applicable(self):
        r = S.ai_row("pi05_x", {"outcome": "not_applicable", "pick": "no", "place": "no"}, "frames", 0)
        self.assertEqual(r["outcome"], "unknown")


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